feat(02-02): add file list scroll navigation and tidEDA export
- Create files.html template with scrollable file list - Add /files route to serve the Flash Memory page - Add POST /api/files/export endpoint for tidEDA format - Include file count display and search/filter functionality - Add export, view, and delete buttons for each file - Security: prevent path traversal in file operations Implements FILE-03 (scroll navigation) and FILE-05 (tidEDA export)
This commit is contained in:
162
src/app.py
162
src/app.py
@@ -222,6 +222,15 @@ def calibration():
|
||||
build_date=BUILD_DATE)
|
||||
|
||||
|
||||
@app.route('/files')
|
||||
def files_page():
|
||||
"""Flash memory / Files page"""
|
||||
return render_template('files.html',
|
||||
page='files',
|
||||
version=SOFTWARE_VERSION,
|
||||
build_date=BUILD_DATE)
|
||||
|
||||
|
||||
# ==================== API ENDPOINTS ====================
|
||||
|
||||
@app.route('/api/status')
|
||||
@@ -309,6 +318,159 @@ def api_calibration_reset():
|
||||
return jsonify({"success": True})
|
||||
|
||||
|
||||
# ==================== FILE MANAGEMENT ====================
|
||||
|
||||
# Logger directory - use /myvscada/logger or fall back to src/data/logger
|
||||
LOGGER_DIR = '/myvscada/logger'
|
||||
FALLBACK_LOGGER_DIR = os.path.join(os.path.dirname(__file__), 'data', 'logger')
|
||||
|
||||
def get_logger_dir():
|
||||
"""Get the logger directory, creating fallback if needed"""
|
||||
if os.path.exists(LOGGER_DIR):
|
||||
return LOGGER_DIR
|
||||
# Use fallback and ensure it exists
|
||||
os.makedirs(FALLBACK_LOGGER_DIR, exist_ok=True)
|
||||
return FALLBACK_LOGGER_DIR
|
||||
|
||||
|
||||
@app.route('/api/files')
|
||||
def api_files_list():
|
||||
"""List all CSV files in the logger directory"""
|
||||
try:
|
||||
logger_dir = get_logger_dir()
|
||||
files = []
|
||||
|
||||
if os.path.exists(logger_dir):
|
||||
for filename in os.listdir(logger_dir):
|
||||
if filename.endswith('.csv'):
|
||||
filepath = os.path.join(logger_dir, filename)
|
||||
stat = os.stat(filepath)
|
||||
files.append({
|
||||
'name': filename,
|
||||
'size': stat.st_size,
|
||||
'modified': datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S'),
|
||||
'created': datetime.fromtimestamp(stat.st_ctime).strftime('%Y-%m-%d %H:%M:%S')
|
||||
})
|
||||
|
||||
# Sort by creation time, newest first
|
||||
files.sort(key=lambda x: x['created'], reverse=True)
|
||||
|
||||
return jsonify({
|
||||
'files': files,
|
||||
'total': len(files)
|
||||
})
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e), 'files': [], 'total': 0}), 500
|
||||
|
||||
|
||||
@app.route('/api/files/<filename>')
|
||||
def api_files_get(filename):
|
||||
"""Get contents of a specific CSV file"""
|
||||
try:
|
||||
# Security: prevent path traversal
|
||||
filename = os.path.basename(filename)
|
||||
if not filename.endswith('.csv'):
|
||||
return jsonify({'error': 'Only CSV files are allowed'}), 400
|
||||
|
||||
logger_dir = get_logger_dir()
|
||||
filepath = os.path.join(logger_dir, filename)
|
||||
|
||||
if not os.path.exists(filepath):
|
||||
return jsonify({'error': 'File not found'}), 404
|
||||
|
||||
with open(filepath, 'r') as f:
|
||||
content = f.read()
|
||||
|
||||
lines = content.split('\n')
|
||||
# Limit preview to first 100 lines
|
||||
preview_lines = lines[:100]
|
||||
preview_content = '\n'.join(preview_lines)
|
||||
|
||||
return jsonify({
|
||||
'name': filename,
|
||||
'content': preview_content,
|
||||
'lines': len(lines),
|
||||
'truncated': len(lines) > 100
|
||||
})
|
||||
except Exception as e:
|
||||
return jsonify({'error': str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/files/<filename>', methods=['DELETE'])
|
||||
def api_files_delete(filename):
|
||||
"""Delete a specific CSV file"""
|
||||
try:
|
||||
# Security: prevent path traversal
|
||||
filename = os.path.basename(filename)
|
||||
if not filename.endswith('.csv'):
|
||||
return jsonify({'success': False, 'error': 'Only CSV files are allowed'}), 400
|
||||
|
||||
logger_dir = get_logger_dir()
|
||||
filepath = os.path.join(logger_dir, filename)
|
||||
|
||||
if not os.path.exists(filepath):
|
||||
return jsonify({'success': False, 'error': 'File not found'}), 404
|
||||
|
||||
os.remove(filepath)
|
||||
return jsonify({'success': True})
|
||||
except Exception as e:
|
||||
return jsonify({'success': False, 'error': str(e)}), 500
|
||||
|
||||
|
||||
@app.route('/api/files/export', methods=['POST'])
|
||||
def api_files_export():
|
||||
"""Generate tidEDA formatted export from CSV file"""
|
||||
try:
|
||||
data = request.get_json() or {}
|
||||
filename = data.get('filename')
|
||||
|
||||
if not filename:
|
||||
return jsonify({"error": "filename required"}), 400
|
||||
|
||||
# Validate filename (security: prevent path traversal)
|
||||
if '..' in filename or '/' in filename or '\\' in filename:
|
||||
return jsonify({"error": "invalid filename"}), 400
|
||||
|
||||
# Ensure .csv extension
|
||||
if not filename.endswith('.csv'):
|
||||
filename = filename + '.csv'
|
||||
|
||||
# Get the logger directory
|
||||
logger_dir = get_logger_dir()
|
||||
file_path = os.path.join(logger_dir, filename)
|
||||
|
||||
if not os.path.exists(file_path):
|
||||
return jsonify({"error": "file not found"}), 404
|
||||
|
||||
# Generate tidEDA format
|
||||
# Structure:
|
||||
# $STATION,{station_id}
|
||||
# $DATETIME,{timestamp}
|
||||
# $DATA
|
||||
# {original_csv_content}
|
||||
# $END
|
||||
|
||||
settings = load_settings()
|
||||
station_id = settings.get("station", {}).get("id", "UNKNOWN")
|
||||
|
||||
with open(file_path, 'r') as f:
|
||||
csv_content = f.read()
|
||||
|
||||
tideda_content = f"$STATION,{station_id}\n"
|
||||
tideda_content += f"$DATETIME,{datetime.now().isoformat()}\n"
|
||||
tideda_content += "$DATA\n"
|
||||
tideda_content += csv_content
|
||||
tideda_content += "$END\n"
|
||||
|
||||
return jsonify({
|
||||
"filename": filename.replace('.csv', '.tideda'),
|
||||
"content": tideda_content,
|
||||
"format": "tidEDA v1.0"
|
||||
})
|
||||
except Exception as e:
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
|
||||
# ==================== MAIN ====================
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
Reference in New Issue
Block a user