diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md new file mode 100644 index 0000000..dbe7293 --- /dev/null +++ b/.github/copilot-instructions.md @@ -0,0 +1,52 @@ +# Copilot Instructions + +## Architecture & Tech Stack + +This project is a hybrid Flask + React application for uploading MCAP files to AWS S3. + +- **Backend (`app/`)**: Python 3.11+ Flask API. + - Serves the built React frontend from `frontend/dist`. + - Uses `boto3` for AWS S3 operations. + - Uses `modaq_toolkit` for MCAP file parsing. + - Handles file uploads, duplicate detection, and progress tracking via SSE. +- **Frontend (`frontend/`)**: React 19 SPA built with Vite. + - TypeScript, TailwindCSS, Zustand (state), TanStack Table. + - Communicates with backend via `/api/*` endpoints. + - Proxies to backend port 5000 during development. + +## Development Workflow + +- **Backend**: Run `python app.py` (serves on :5000). +- **Frontend**: Run `cd frontend && npm run dev` (serves on :3000, proxies `/api` to :5000). +- **Production**: Frontend is built to `frontend/dist`, which Flask serves statically. + +## Build, Test, and Lint Commands + +### Python (Backend) +- **Test**: `pytest tests/ -v` (Single file: `pytest tests/test_mcap_service.py -v`) +- **Lint**: `ruff check app/ tests/` +- **Format**: `ruff format app/ tests/` +- **Type Check**: `mypy app/` + +### JavaScript/TypeScript (Frontend) +- **Build**: `cd frontend && npm run build` (outputs to `frontend/dist`) +- **Test**: `cd frontend && npm run test` (Vitest) +- **Lint**: `cd frontend && npm run lint` (Biome) +- **Type Check**: `cd frontend && npm run typecheck` + +## Key Conventions + +### Branding & UI +- **Organization Name**: National Laboratory of the Rockies (**NLR**). **NEVER** use "NREL". +- **CSS Classes**: Use `nlr-` prefix for custom classes (e.g., `nlr-blue-500` in Tailwind). +- **Icons**: Import icons ONLY from `frontend/src/utils/icons.tsx` (abstraction over lucide-react). Do not import directly from icon libraries. + +### Data & S3 +- **S3 Paths**: Use Hive partitioning: `year=YYYY/month=MM/day=DD/hour=HH/minute=M0/filename.mcap`. + - Minutes are bucketed to 10-minute intervals (00, 10, 20...). +- **Timestamps**: Extracted from MCAP files using `MCAPParser` from `modaq_toolkit`. + +### Project Structure +- `app/routes/main.py`: Serves the React app (`frontend/dist/index.html`). +- `app/routes/upload.py`: Handles upload logic and SSE progress streams. +- `app/static`: **Legacy/Unused**. Do not use for new frontend code; work in `frontend/`. diff --git a/README.md b/README.md index a649ab4..0efd477 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ A Python based local web application for uploading MODAQ files to S3 with progress tracking, duplicate detection, and configuration. -![MODAQ Upload front page showing the upload page with drag-and-drop area](./docs/img/upload_screen.png) +![MODAQ Upload: Upload page showing the folder browser with per-folder upload status](./docs/images/index_upload.png) ## Features @@ -39,21 +39,21 @@ python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate ``` -3. Install dependencies: +3. Install Python dependencies: ```bash pip install -r requirements.txt ``` -4. (Optional) Install development dependencies: +4. Install frontend dependencies and build the UI: ```bash -pip install -r requirements-dev.txt +cd frontend && npm install && npm run build && cd .. ``` ## Usage -### Running the Application (Development) +### Running the Application ```bash python app.py @@ -61,6 +61,29 @@ python app.py The application will be available at `http://localhost:5000`. +### Development (Live Reload) + +Run the Flask backend and Vite frontend separately for hot-reload during development: + +Terminal 1: Flask API (port 5000): + +```bash +python app.py +``` + +Terminal 2: Vite dev server (port 3000): + +```bash +cd frontend && npm run dev +``` + +Open `http://localhost:3000`. The Vite server proxies `/api` requests to Flask. +After making frontend changes you're happy with, rebuild for production: + +```bash +cd frontend && npm run build +``` + ### Production Deployment (Linux) For production use on a Linux machine, use the automated installation script which sets up a systemd service with Gunicorn. @@ -175,28 +198,140 @@ Settings are saved to `settings.json` (gitignored). ### Uploading Files -1. Go to the Upload page -2. Drag and drop MCAP files or click to select -3. Review the analysis results showing timestamps and S3 paths -4. Check "Skip duplicates" to avoid re-uploading existing files -5. Click Upload Files to start -6. Monitor progress in real-time +See [Application Guide: Upload](#upload) below for a walkthrough. ### Browsing Files -1. Go to the Browse Files page -2. Navigate through the Hive-partitioned folder structure -3. Use the search box to find specific files +See [Application Guide: Browse Uploaded Files](#browse-uploaded-files) below. ### Updating the Application -1. Go to Settings > Application Updates -2. Click Check for Updates to see if updates are available -3. Click Update Application to: - - Pull latest changes from git - - Reinstall Python dependencies - - Update modaq_toolkit to the latest version -4. Restart the application after updating +See [Application Guide: Updating the Software](#updating-the-software) below. + +## Application Guide + +### Upload + +![Upload page showing the folder browser with per-folder upload status and step indicator](./docs/images/index_upload.png) + +The Upload page walks you through uploading files in four steps: Select → Review → Upload → Complete. + +1. Navigate to the folder containing your MCAP files using the file browser. Quick Links on the left give fast access to common locations. +2. The browser shows each subfolder's upload status: data file count, log file count, and how many have already been uploaded to S3. +3. Check or uncheck folders to include or exclude them. Use Select All / None or search to filter. +4. Click Upload N files to proceed to the Review step, where you can inspect the per-file S3 destination paths before committing. +5. Already-uploaded files are skipped; no duplicates are created. + +For more than 500 files, use [Large Folder Upload](#large-folder-upload) instead. + + +### Large Folder Upload + +![Large Folder Upload page showing the folder sync interface](./docs/images/large_folder_upload_index.png) + +The Large Folder Upload page syncs an entire folder tree to S3 without per-file analysis. Use this when you have 500+ files and don't need to inspect each file's timestamp individually. + +1. Navigate to the root folder you want to sync. +2. Click Select This Folder to confirm. +3. The folder structure is copied to S3 as-is. Already-uploaded files are skipped. + + +### Browse Uploaded Files + +![Browse Uploaded Files page showing the S3 bucket folder list](./docs/images/browse_index.png) + +The Browse Uploaded Files page lets you navigate the contents of your S3 bucket directly from the app. + +1. Click any folder to drill down into it. +2. Use the breadcrumb trail at the top to navigate back up. +3. This is useful for verifying that uploads landed in the correct location. + + +### History + +![History page showing upload sessions with file counts, data sizes, and transfer speeds](./docs/images/logs_index.png) + +The History page keeps a record of every upload session run from this machine. + +- Upload History tab: each session's date, file count, data transferred, transfer speed, and outcome (completed / skipped / failed). Click any row to expand the per-file breakdown. Use CSV to export a session log. +- Event Log tab: application events for troubleshooting. + +The running totals at the top (files uploaded, total data, failed, sessions) summarise all sessions. + + +### Clear Hard Drive + +![Clear Hard Drive page showing folder selection with uploaded/deletable file counts](./docs/images/delete_index.png) + +The Clear Hard Drive page removes local files that have already been uploaded to S3. Files are verified against S3 before any deletion. + +1. Navigate to the folder you want to clean up. +2. The browser shows how many files are uploaded (deletable) vs not yet uploaded. +3. Only uploaded files are deleted; files not yet in S3 are not touched. +4. Click Clear N files and confirm to proceed through the workflow (Select → Review → Confirm → Clear → Complete). + + +### Settings + +![Settings page showing AWS configuration fields](./docs/images/settings_index.png) + +The Settings page controls the AWS connection used for all uploads and browsing. + +- AWS Profile: profile from `~/.aws/credentials` to use. +- AWS Region: region of your S3 bucket. +- S3 Bucket: bucket files are uploaded to. +- Default Upload Folder: pre-populates the file browser on the Upload page. +- Display Name: title shown in the application header. +- Log Directory: where upload history logs are stored. + +Fields marked *Locked: set by environment variable* are controlled by your `.env` file and cannot be changed from the UI (see [Configuration](#configuration)). + + +### Updating the Software + +#### v1.1 and later: in-app update + +From v1.1 onwards, updates can be applied from within the app. + +Option A: via Settings: + +1. Click Settings in the navigation bar. +2. Scroll down to the Software Update section. +3. Click Check for updates. + +![Software Update section in Settings showing current version, commit, and Check for updates button](./docs/images/settings_index_software_update.png) + +4. If updates are available, click Update Application. This pulls the latest code, reinstalls Python dependencies, and updates `modaq_toolkit`. +5. Restart the application after the update completes (`Ctrl+C` then `python app.py`, or restart the systemd service). + +Option B: via the About modal: + +1. Click the version badge (e.g. v1.1.0) in the navigation bar to open the About modal. +2. Expand the Software Update section and follow the same steps. + +#### Before v1.1: manual update + +If you are running a version prior to v1.1, the in-app updater is not available. Update manually from the terminal: + +```bash +cd modaq_upload +git pull +source venv/bin/activate # On Windows: venv\Scripts\activate +pip install -r requirements.txt +cd frontend && npm install && npm run build && cd .. +``` + +Then restart the application: + +```bash +python app.py +``` + +Or, if running as a systemd service: + +```bash +sudo systemctl restart modaq-upload +``` ## S3 Path Format @@ -238,31 +373,37 @@ ruff format app/ tests/ mypy app/ ``` +### Install Development Dependencies + +```bash +pip install -r requirements-dev.txt +``` + ### JavaScript Linting (Biome) ```bash -npm run lint # Check -npm run lint:fix # Auto-fix +cd frontend && npm run lint # Check +cd frontend && npm run lint:fix # Auto-fix ``` ### JavaScript Type Checking ```bash -npm run typecheck # tsc --checkJs via jsconfig.json +cd frontend && npm run typecheck ``` ### JavaScript Testing (Vitest) ```bash -npm run test # Run all JS tests -npm run test:watch # Watch mode -npm run test:coverage # With coverage report +cd frontend && npm run test # Run all JS tests +cd frontend && npm run test:watch # Watch mode +cd frontend && npm run test:coverage # With coverage report ``` ### All JS Checks ```bash -npm run check # Biome + tsc + Vitest +cd frontend && npm run check # Biome + tsc + Vitest ``` ## Architecture diff --git a/app/__init__.py b/app/__init__.py index d418c3a..3a6a505 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -15,11 +15,11 @@ def _sse_cleanup_worker() -> None: """Background worker that periodically cleans up stale SSE queues.""" - from app.routes.upload import _cleanup_old_sse_queues + from app.services.sse_manager import get_sse_manager while not _cleanup_stop_event.wait(timeout=300): # Check every 5 minutes try: - removed = _cleanup_old_sse_queues() + removed = get_sse_manager().cleanup_old_queues() if removed > 0: from app.services.log_service import get_log_service @@ -72,6 +72,7 @@ def inject_display_name() -> dict[str, str]: # Register blueprints from app.routes.delete import delete_bp from app.routes.files import files_bp + from app.routes.large_folder_upload import large_folder_upload_bp from app.routes.logs import logs_bp from app.routes.main import main_bp from app.routes.settings import settings_bp @@ -83,6 +84,7 @@ def inject_display_name() -> dict[str, str]: app.register_blueprint(settings_bp, url_prefix="/api/settings") app.register_blueprint(logs_bp, url_prefix="/api/logs") app.register_blueprint(delete_bp, url_prefix="/api/delete") + app.register_blueprint(large_folder_upload_bp, url_prefix="/api/large-folder-upload") # Start background SSE cleanup thread _start_sse_cleanup() diff --git a/app/config.py b/app/config.py index e79d22e..f4699d2 100644 --- a/app/config.py +++ b/app/config.py @@ -30,6 +30,7 @@ ENV_DEFAULT_UPLOAD_FOLDER = "MODAQ_DEFAULT_UPLOAD_FOLDER" ENV_DISPLAY_NAME = "MODAQ_DISPLAY_NAME" ENV_LOG_DIRECTORY = "MODAQ_LOG_DIRECTORY" +ENV_ALLOWED_EXTENSIONS = "MODAQ_ALLOWED_EXTENSIONS" @functools.cache @@ -84,6 +85,20 @@ def _load_settings(self) -> None: "default_upload_folder": "", "display_name": "MODAQ Uploader", "log_directory": "logs", + "file_categories": [ + { + "name": "data", + "extensions": ["mcap", "tdms", "done", "csv"], + "partition_interval": "10min", + "description": "High-frequency data files", + }, + { + "name": "logs", + "extensions": ["txt", "log", "yaml", "logs"], + "partition_interval": "daily", + "description": "System and operation logs", + }, + ], "batch_processing": { "enabled": True, "batch_size": 100, @@ -206,6 +221,24 @@ def log_directory(self) -> Path: path = BASE_DIR / path return path + @property + def file_categories(self) -> list[dict[str, Any]]: + """Get the list of file categories.""" + return list(self._settings.get("file_categories", [])) + + @property + def allowed_extensions(self) -> list[str]: + """Get the flat list of allowed file extensions (lowercase, no dot). + + Aggregates extensions from all file_categories. + """ + categories = self.file_categories + all_exts = set() + for cat in categories: + for ext in cat.get("extensions", []): + all_exts.add(str(ext).lower().lstrip(".")) + return sorted(list(all_exts)) + @property def batch_processing(self) -> dict[str, Any]: """Get batch processing configuration.""" @@ -227,6 +260,25 @@ class AppUpdater: def __init__(self) -> None: self.base_dir = BASE_DIR + def _get_remote_version(self) -> str | None: + """Read the version from the remote pyproject.toml (FETCH_HEAD).""" + try: + result = subprocess.run( + ["git", "show", "FETCH_HEAD:pyproject.toml"], + cwd=self.base_dir, + capture_output=True, + text=True, + check=True, + ) + for line in result.stdout.splitlines(): + stripped = line.strip() + if stripped.startswith("version") and "=" in stripped: + raw = stripped.split("=", 1)[1].strip().strip('"').strip("'") + return raw + except Exception: + pass + return None + def check_for_updates(self) -> dict[str, Any]: """Check if there are updates available from git remote.""" try: @@ -238,8 +290,22 @@ def check_for_updates(self) -> dict[str, Any]: check=True, ) + # Check how many commits behind we are + behind_result = subprocess.run( + ["git", "rev-list", "--count", "HEAD..@{u}"], + cwd=self.base_dir, + capture_output=True, + text=True, + ) + commits_behind = 0 + if behind_result.returncode == 0: + try: + commits_behind = int(behind_result.stdout.strip()) + except ValueError: + commits_behind = 0 + # Check if we're behind remote - result = subprocess.run( + status_result = subprocess.run( ["git", "status", "-uno"], cwd=self.base_dir, capture_output=True, @@ -247,8 +313,8 @@ def check_for_updates(self) -> dict[str, Any]: check=True, ) - behind = "Your branch is behind" in result.stdout - up_to_date = "Your branch is up to date" in result.stdout + behind = commits_behind > 0 + up_to_date = "Your branch is up to date" in status_result.stdout # Get current commit current = subprocess.run( @@ -259,10 +325,29 @@ def check_for_updates(self) -> dict[str, Any]: check=True, ) + # Get remote commit + remote_commit_result = subprocess.run( + ["git", "rev-parse", "--short", "@{u}"], + cwd=self.base_dir, + capture_output=True, + text=True, + ) + remote_commit = ( + remote_commit_result.stdout.strip() + if remote_commit_result.returncode == 0 + else None + ) + + # Try to read the remote version from pyproject.toml + remote_version = self._get_remote_version() if behind else None + return { "updates_available": behind, "up_to_date": up_to_date, "current_commit": current.stdout.strip(), + "remote_commit": remote_commit, + "commits_behind": commits_behind, + "remote_version": remote_version, "error": None, } except subprocess.CalledProcessError as e: @@ -270,17 +355,44 @@ def check_for_updates(self) -> dict[str, Any]: "updates_available": False, "up_to_date": False, "current_commit": None, + "remote_commit": None, + "commits_behind": 0, + "remote_version": None, "error": str(e), } + # Human-readable labels for each update step + STEP_LABELS: dict[str, str] = { + "git_pull": "Downloading update", + "pip_install": "Installing Python packages", + "modaq_toolkit": "Updating data tools", + "npm_install": "Installing app dependencies", + "frontend_build": "Rebuilding interface", + } + def update_application(self) -> dict[str, Any]: - """Pull latest changes from git and reinstall dependencies.""" + """Pull latest changes from git and reinstall dependencies. + + Saves the pre-update commit so the caller can offer rollback on failure. + """ + # Capture the current commit so we can roll back if needed + pre_update_commit: str | None = None + try: + cp = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=self.base_dir, + capture_output=True, + text=True, + check=True, + ) + pre_update_commit = cp.stdout.strip() + except subprocess.CalledProcessError: + pass + + step_order = ["git_pull", "pip_install", "modaq_toolkit", "npm_install", "frontend_build"] results: dict[str, Any] = { - "git_pull": {"success": False, "output": ""}, - "pip_install": {"success": False, "output": ""}, - "modaq_toolkit": {"success": False, "output": ""}, - "npm_install": {"success": False, "output": ""}, - "frontend_build": {"success": False, "output": ""}, + name: {"success": False, "output": "", "label": self.STEP_LABELS.get(name, name)} + for name in step_order } frontend_dir = str(self.base_dir / "frontend") @@ -309,6 +421,7 @@ def update_application(self) -> dict[str, Any]: ("frontend_build", ["npm", "run", "build"], frontend_dir), ] + failed_at: str | None = None for step_name, cmd, cwd in steps: try: result = subprocess.run( @@ -321,15 +434,137 @@ def update_application(self) -> dict[str, Any]: results[step_name] = { "success": True, "output": result.stdout + result.stderr, + "label": self.STEP_LABELS.get(step_name, step_name), } except subprocess.CalledProcessError as e: results[step_name] = { "success": False, - "output": e.stdout + e.stderr if e.stdout else str(e), + "output": e.stdout + e.stderr if (e.stdout or e.stderr) else str(e), + "label": self.STEP_LABELS.get(step_name, step_name), } + failed_at = step_name break - return results + all_success = failed_at is None + return { + "results": results, + "step_order": step_order, + "success": all_success, + "failed_at": failed_at, + "pre_update_commit": pre_update_commit, + } + + def rollback_update(self, commit: str) -> dict[str, Any]: + """Roll back to a specific git commit and rebuild the frontend.""" + try: + # Hard-reset to the saved commit + subprocess.run( + ["git", "reset", "--hard", commit], + cwd=self.base_dir, + capture_output=True, + text=True, + check=True, + ) + except subprocess.CalledProcessError as e: + return { + "success": False, + "output": e.stdout + e.stderr if (e.stdout or e.stderr) else str(e), + "error": "Failed to reset git repository", + } + + # Rebuild the frontend so the rolled-back version is served correctly + frontend_dir = str(self.base_dir / "frontend") + build_output = "" + try: + result = subprocess.run( + ["npm", "run", "build"], + cwd=frontend_dir, + capture_output=True, + text=True, + check=True, + ) + build_output = result.stdout + result.stderr + except subprocess.CalledProcessError as e: + build_output = e.stdout + e.stderr if (e.stdout or e.stderr) else str(e) + + return { + "success": True, + "commit": commit, + "output": build_output, + "error": None, + } + + def get_branches(self) -> dict[str, Any]: + """Get current branch and list of all local and remote branches.""" + try: + current = subprocess.run( + ["git", "branch", "--show-current"], + cwd=self.base_dir, + capture_output=True, + text=True, + check=True, + ) + + local = subprocess.run( + ["git", "branch", "--format=%(refname:short)"], + cwd=self.base_dir, + capture_output=True, + text=True, + check=True, + ) + + remote = subprocess.run( + ["git", "branch", "-r", "--format=%(refname:short)"], + cwd=self.base_dir, + capture_output=True, + text=True, + check=True, + ) + + local_branches = [b.strip() for b in local.stdout.splitlines() if b.strip()] + # Strip "origin/" prefix and de-duplicate with local branches + remote_branches = [ + b.strip().removeprefix("origin/") + for b in remote.stdout.splitlines() + if b.strip() and "HEAD" not in b and b.strip() != "origin" + ] + all_branches = sorted(set(local_branches + remote_branches)) + + return { + "current": current.stdout.strip(), + "branches": all_branches, + "error": None, + } + except subprocess.CalledProcessError as e: + return { + "current": None, + "branches": [], + "error": str(e), + } + + def switch_branch(self, branch: str) -> dict[str, Any]: + """Switch to the specified git branch.""" + try: + result = subprocess.run( + ["git", "checkout", branch], + cwd=self.base_dir, + capture_output=True, + text=True, + check=True, + ) + return { + "success": True, + "branch": branch, + "output": result.stdout + result.stderr, + "error": None, + } + except subprocess.CalledProcessError as e: + return { + "success": False, + "branch": branch, + "output": e.stdout + e.stderr if e.stdout or e.stderr else "", + "error": str(e), + } def get_version_info(self) -> dict[str, Any]: """Get current version information.""" diff --git a/app/routes/delete.py b/app/routes/delete.py index 7f4b518..b36c59a 100644 --- a/app/routes/delete.py +++ b/app/routes/delete.py @@ -5,7 +5,6 @@ import subprocess import threading import time -from collections import deque from collections.abc import Generator from pathlib import Path from typing import Any @@ -14,21 +13,10 @@ from app.config import get_settings from app.services.delete_manager import DeleteJob, get_delete_manager +from app.services.sse_manager import get_sse_manager delete_bp = Blueprint("delete", __name__) -# SSE client queues for delete jobs -_sse_queues: dict[str, list[deque[dict[str, Any]]]] = {} -_sse_lock = threading.Lock() - - -def _send_sse_event(job_id: str, data: dict[str, Any]) -> None: - """Send an SSE event to all clients listening for a delete job.""" - with _sse_lock: - queues = _sse_queues.get(job_id, []) - for q in queues: - q.append(data) - @delete_bp.route("/scan", methods=["POST"]) def scan_folder() -> tuple[Response, int]: @@ -109,14 +97,15 @@ def start_delete(job_id: str) -> tuple[Response, int]: def progress_callback(job: DeleteJob) -> None: """Send progress updates via SSE.""" + sse_mgr = get_sse_manager() if job.status in ("completed", "failed", "cancelled"): - _send_sse_event(job.job_id, {"type": "delete_complete", **job.to_dict()}) + sse_mgr.send_event(job.job_id, {"type": "delete_complete", **job.to_dict()}) else: - _send_sse_event(job.job_id, {"type": "delete_progress", **job.to_progress_dict()}) + sse_mgr.send_event(job.job_id, {"type": "delete_progress", **job.to_progress_dict()}) def batch_callback(batch_event: dict[str, Any]) -> None: """Send batch-level events via SSE.""" - _send_sse_event(job_id, batch_event) + get_sse_manager().send_event(job_id, batch_event) def run_delete() -> None: manager.start_delete_job( @@ -146,11 +135,9 @@ def get_progress(job_id: str) -> Response: manager = get_delete_manager() def generate() -> Generator[str, None, None]: - queue: deque[dict[str, Any]] = deque() - with _sse_lock: - if job_id not in _sse_queues: - _sse_queues[job_id] = [] - _sse_queues[job_id].append(queue) + sse_mgr = get_sse_manager() + queue, event = sse_mgr.register_client(job_id) + last_heartbeat_time = time.time() try: # Send initial state @@ -162,12 +149,20 @@ def generate() -> Generator[str, None, None]: while queue: data = queue.popleft() yield f"data: {json.dumps(data)}\n\n" + last_heartbeat_time = time.time() - # Terminal events if data.get("type") == "delete_complete": return - time.sleep(0.1) + # Send heartbeat if no activity for a while + now = time.time() + if now - last_heartbeat_time > sse_mgr.heartbeat_interval: + yield ": heartbeat\n\n" + last_heartbeat_time = now + + # Wait for signal (event-driven, no busy polling) + event.wait(timeout=sse_mgr.heartbeat_interval) + event.clear() # Check if job still exists job = manager.get_job(job_id) @@ -181,11 +176,7 @@ def generate() -> Generator[str, None, None]: return finally: - with _sse_lock: - if job_id in _sse_queues and queue in _sse_queues[job_id]: - _sse_queues[job_id].remove(queue) - if not _sse_queues[job_id]: - del _sse_queues[job_id] + sse_mgr.deregister_client(job_id, queue) return Response( generate(), diff --git a/app/routes/files.py b/app/routes/files.py index 3237814..4621bcd 100644 --- a/app/routes/files.py +++ b/app/routes/files.py @@ -143,16 +143,31 @@ def browse_local() -> tuple[Response, int]: if not path.is_dir(): return jsonify({"error": f"Not a directory: {path}"}), 400 - # Build response — single-pass walk for recursive MCAP counts + cache checks. + # Build response — single-pass walk for recursive file counts + cache checks. # os.walk with onerror skips unreadable subdirectories instead of aborting, # which is important on Linux where permission errors are common. cache = get_cache_service() bucket = g.settings.s3_bucket - folder_mcap_counts: dict[str, int] = {} + # Get allowed extensions (add dot prefix) and a per-extension category lookup. + allowed_exts = tuple(f".{ext}" for ext in g.settings.allowed_extensions) + category_names = [str(cat["name"]) for cat in g.settings.file_categories] + ext_to_category: dict[str, str] = {} + for cat in g.settings.file_categories: + cat_name = str(cat.get("name", "other")) + for ext in cat.get("extensions", []): + ext_to_category[str(ext).lower().lstrip(".")] = cat_name + + def _empty_counts() -> dict[str, int]: + return dict.fromkeys(category_names, 0) + + folder_file_counts: dict[str, int] = {} folder_uploaded_counts: dict[str, int] = {} + folder_category_counts: dict[str, dict[str, int]] = {} + direct_category_counts: dict[str, int] = _empty_counts() + total_category_counts: dict[str, int] = _empty_counts() files: list[dict[str, str | int | float | bool]] = [] - mcap_count = 0 + file_count = 0 direct_uploaded = 0 def _walk_error(err: OSError) -> None: @@ -163,47 +178,58 @@ def _walk_error(err: OSError) -> None: dirnames[:] = [d for d in dirnames if not d.startswith(".")] for fname in filenames: - if not fname.endswith(".mcap") or fname.startswith("."): + if not fname.lower().endswith(allowed_exts) or fname.startswith("."): continue - mcap_path = Path(dirpath) / fname - rel = mcap_path.relative_to(path) + file_path = Path(dirpath) / fname + rel = file_path.relative_to(path) parts = rel.parts try: - file_stat = mcap_path.stat() + file_stat = file_path.stat() except OSError: continue uploaded = ( - cache.check_exists_by_filename(bucket, mcap_path.name, file_stat.st_size) is True + cache.check_exists_by_filename(bucket, file_path.name, file_stat.st_size) is True ) + ext = file_path.suffix.lower().lstrip(".") + category = ext_to_category.get(ext, "other") + if category in total_category_counts: + total_category_counts[category] += 1 + if len(parts) == 1: - # Direct child MCAP file - mcap_count += 1 + # Direct child file + file_count += 1 if uploaded: direct_uploaded += 1 + if category in direct_category_counts: + direct_category_counts[category] += 1 files.append( { - "name": mcap_path.name, - "path": str(mcap_path), + "name": file_path.name, + "path": str(file_path), "size": file_stat.st_size, "mtime": file_stat.st_mtime, "already_uploaded": uploaded, + "file_category": category, } ) else: # Nested — attribute to the immediate subfolder folder_name = parts[0] - folder_mcap_counts[folder_name] = folder_mcap_counts.get(folder_name, 0) + 1 + folder_file_counts[folder_name] = folder_file_counts.get(folder_name, 0) + 1 if uploaded: folder_uploaded_counts[folder_name] = ( folder_uploaded_counts.get(folder_name, 0) + 1 ) + cat_counts = folder_category_counts.setdefault(folder_name, _empty_counts()) + if category in cat_counts: + cat_counts[category] += 1 # Build folder list from direct children (non-hidden directories) - folders: list[dict[str, str | int]] = [] + folders: list[dict[str, str | int | dict[str, int]]] = [] try: for entry in sorted(path.iterdir(), key=lambda x: x.name.lower()): if entry.name.startswith("."): @@ -214,8 +240,11 @@ def _walk_error(err: OSError) -> None: { "name": entry.name, "path": str(entry), - "mcap_count": folder_mcap_counts.get(entry.name, 0), + "file_count": folder_file_counts.get(entry.name, 0), "already_uploaded": folder_uploaded_counts.get(entry.name, 0), + "category_counts": folder_category_counts.get( + entry.name, _empty_counts() + ), } ) except PermissionError: @@ -299,9 +328,11 @@ def _walk_error(err: OSError) -> None: "breadcrumbs": breadcrumbs, "quick_links": quick_links, "folders": folders, - "files": files, # Only MCAP files - "mcap_count": mcap_count, - "total_mcap_count": mcap_count + sum(folder_mcap_counts.values()), + "files": files, # All allowed files + "file_count": file_count, + "total_file_count": file_count + sum(folder_file_counts.values()), + "category_counts": direct_category_counts, + "total_category_counts": total_category_counts, "already_uploaded": direct_uploaded + sum(folder_uploaded_counts.values()), } ), 200 diff --git a/app/routes/large_folder_upload.py b/app/routes/large_folder_upload.py new file mode 100644 index 0000000..4199453 --- /dev/null +++ b/app/routes/large_folder_upload.py @@ -0,0 +1,467 @@ +"""Large Folder Upload API routes — streams aws s3 sync output via SSE.""" + +import csv +import io +import json +import os +import subprocess +import threading +import time +import uuid +from collections.abc import Generator +from dataclasses import dataclass, field +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +from flask import Blueprint, Response, jsonify, request + +from app.config import get_settings +from app.services.sse_manager import get_sse_manager + +large_folder_upload_bp = Blueprint("large_folder_upload", __name__) + +# In-memory registry of active sync jobs +_jobs: dict[str, "SyncJob"] = {} +_jobs_lock = threading.Lock() + + +@dataclass +class SyncJob: + job_id: str + folder_path: str + s3_prefix: str + s3_uri: str + cmd_base: list[str] # base command without --dryrun + status: str = "running" # running | completed | failed | cancelled + process: subprocess.Popen[str] | None = field(default=None, repr=False) + return_code: int | None = None + lines: list[str] = field(default_factory=list) + total_files: int = 0 # populated after dry-run + done_files: int = 0 + started_at: datetime = field(default_factory=lambda: datetime.now(UTC)) + upload_started_at: float = field(default_factory=time.monotonic) + created_at: float = field(default_factory=time.time) + + +def _register(job: SyncJob) -> None: + with _jobs_lock: + _jobs[job.job_id] = job + + +def _get(job_id: str) -> SyncJob | None: + with _jobs_lock: + return _jobs.get(job_id) + + +def _save_sync_log(job: SyncJob, completed_at: datetime) -> None: + """Write a summary JSONL log entry, a raw output text file, and an upload-history CSV.""" + try: + from app.services.log_service import get_log_service + from app.services.utils import format_file_size + + log = get_log_service() + settings = get_settings() + duration_s = (completed_at - job.started_at).total_seconds() + log_dir = settings.log_directory + + # ── Raw text log ────────────────────────────────────────────────────── + hive_txt = ( + log_dir + / "sync" + / f"year={completed_at.year:04d}" + / f"month={completed_at.month:02d}" + / f"day={completed_at.day:02d}" + ) + hive_txt.mkdir(parents=True, exist_ok=True) + time_str = completed_at.strftime("%H%M%S") + short_id = job.job_id[:8] + txt_path = hive_txt / f"sync-{time_str}-{short_id}.txt" + with open(txt_path, "w", encoding="utf-8") as fh: + fh.write(f"# Large Folder Upload — {job.job_id}\n") + fh.write(f"# folder: {job.folder_path}\n") + fh.write(f"# dest: {job.s3_uri}\n") + fh.write(f"# started: {job.started_at.isoformat()}\n") + fh.write(f"# ended: {completed_at.isoformat()}\n") + fh.write(f"# status: {job.status}\n\n") + fh.write("\n".join(job.lines)) + + # ── Upload-history CSV (appears in Upload History tab) ──────────────── + _write_history_csv(job, completed_at, log_dir, time_str, short_id, format_file_size) + + # ── JSONL event log entry ───────────────────────────────────────────── + if job.status == "completed": + level = "INFO" + elif job.status == "cancelled": + level = "WARNING" + else: + level = "ERROR" + log.log( + level, + "large_folder_sync", + f"sync_{job.status}", + f"Large folder sync {job.status}: {Path(job.folder_path).name} → {job.s3_uri}", + { + "job_id": job.job_id, + "folder_path": job.folder_path, + "s3_uri": job.s3_uri, + "s3_prefix": job.s3_prefix, + "return_code": job.return_code, + "duration_seconds": round(duration_s, 1), + "output_lines": len(job.lines), + "log_file": str(txt_path.relative_to(log_dir)), + }, + ) + except Exception: + pass # Never crash the streaming thread over logging + + +def _write_history_csv( + job: SyncJob, + completed_at: datetime, + log_dir: Path, + time_str: str, + short_id: str, + format_file_size: Any, +) -> None: + """Write a CSV into logs/csv/ so this sync shows in the Upload History tab.""" + # Parse "upload: /local/path to s3://bucket/key" lines + bucket = get_settings().s3_bucket + upload_lines = [ln for ln in job.lines if ln.startswith("upload:")] + if not upload_lines: + # Nothing was uploaded (all skipped); still write an empty-session CSV + upload_lines = [] + + num_files = len(upload_lines) + total_duration_s = (completed_at - job.started_at).total_seconds() + per_file_duration = total_duration_s / num_files if num_files > 0 else 0.0 + + columns = [ + "job_id", + "filename", + "file_size_bytes", + "file_size_formatted", + "s3_path", + "status", + "data_start_time", + "upload_started_at", + "upload_completed_at", + "upload_duration_seconds", + "upload_speed_mbps", + "is_duplicate", + "is_valid", + "error_message", + ] + + buf = io.StringIO() + writer = csv.writer(buf) + writer.writerow(columns) + + for line in upload_lines: + # "upload: /local/path to s3://bucket/key" + try: + rest = line[len("upload:") :].strip() + local_path, s3_full = rest.split(" to ", 1) + local_path = local_path.strip() + s3_full = s3_full.strip() + filename = Path(local_path).name + # Strip "s3://bucket/" to get the relative key + s3_path = s3_full.replace(f"s3://{bucket}/", "", 1) if bucket else s3_full + except ValueError: + continue + + try: + size_bytes = os.path.getsize(local_path) + except OSError: + size_bytes = 0 + + speed = ( + round(size_bytes / per_file_duration / 1024 / 1024 * 8, 2) + if per_file_duration > 0 and size_bytes > 0 + else "" + ) + + writer.writerow( + [ + job.job_id, + filename, + size_bytes, + format_file_size(size_bytes), + s3_path, + "completed", + "", # data_start_time — not available for sync + job.started_at.isoformat(), + completed_at.isoformat(), + round(per_file_duration, 3), + speed, + False, # is_duplicate — these were NOT skipped + True, # is_valid + "", + ] + ) + + hive_csv = ( + log_dir + / "csv" + / f"year={completed_at.year:04d}" + / f"month={completed_at.month:02d}" + / f"day={completed_at.day:02d}" + ) + hive_csv.mkdir(parents=True, exist_ok=True) + csv_path = hive_csv / f"upload-summary-{time_str}-{short_id}.csv" + with open(csv_path, "w", encoding="utf-8", newline="") as fh: + fh.write(buf.getvalue()) + + +def _stream_process(job: SyncJob) -> None: + """Dry-run to count files, then stream the real upload with progress events.""" + sse = get_sse_manager() + + # ── Phase 1: dry-run to count files that will actually be uploaded ── + try: + dryrun = subprocess.run( + [*job.cmd_base, "--dryrun"], + capture_output=True, + text=True, + timeout=300, + ) + total = sum(1 for ln in dryrun.stdout.splitlines() if "(dryrun) upload:" in ln) + job.total_files = total + sse.send_event(job.job_id, {"type": "plan", "total_files": total}) + except Exception: + # Dry-run failed — proceed without a known total + sse.send_event(job.job_id, {"type": "plan", "total_files": 0}) + + if job.status != "running": + return # Cancelled during dry-run + + # ── Phase 2: real upload ── + try: + proc = subprocess.Popen( + job.cmd_base, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + bufsize=1, + ) + job.process = proc + job.upload_started_at = time.monotonic() + + for raw_line in proc.stdout: # type: ignore[union-attr] + stripped = raw_line.rstrip("\n") + job.lines.append(stripped) + + if stripped.startswith("upload:"): + job.done_files += 1 + elapsed = time.monotonic() - job.upload_started_at + eta_s: int | None = None + if job.total_files > 0 and job.done_files > 0: + remaining = job.total_files - job.done_files + eta_s = int(elapsed / job.done_files * remaining) if remaining > 0 else 0 + sse.send_event( + job.job_id, + { + "type": "file_done", + "line": stripped, + "done": job.done_files, + "total": job.total_files, + "elapsed_s": int(elapsed), + "eta_s": eta_s, + }, + ) + else: + sse.send_event(job.job_id, {"type": "line", "line": stripped}) + + proc.wait() + job.return_code = proc.returncode + + if job.status == "running": + job.status = "completed" if job.return_code == 0 else "failed" + + completed_at = datetime.now(UTC) + _save_sync_log(job, completed_at) + + sse.send_event( + job.job_id, + { + "type": "done", + "status": job.status, + "return_code": job.return_code, + "done": job.done_files, + "total": job.total_files, + }, + ) + except Exception as exc: + job.status = "failed" + completed_at = datetime.now(UTC) + _save_sync_log(job, completed_at) + sse.send_event( + job.job_id, + {"type": "done", "status": "failed", "error": str(exc)}, + ) + + +@large_folder_upload_bp.route("/start", methods=["POST"]) +def start_sync() -> tuple[Response, int]: + """Start an aws s3 sync job. + + Request body: + folder_path: Local folder to sync from + s3_prefix: S3 key prefix (e.g. "user_upload_2025-01-01T12-00-00") + + Returns: + JSON with job_id + """ + if not request.is_json: + return jsonify({"error": "JSON body required"}), 400 + + data: dict[str, Any] = request.get_json() or {} + folder_path: str = data.get("folder_path", "").strip() + s3_prefix: str = data.get("s3_prefix", "").strip() + + if not folder_path: + return jsonify({"error": "folder_path is required"}), 400 + if not os.path.isdir(folder_path): + return jsonify({"error": f"folder_path does not exist: {folder_path}"}), 400 + if not s3_prefix: + return jsonify({"error": "s3_prefix is required"}), 400 + + settings = get_settings() + if not settings.s3_bucket: + return jsonify({"error": "S3 bucket not configured. Check Settings."}), 400 + + s3_uri = f"s3://{settings.s3_bucket}/{s3_prefix.strip('/')}/" + + cmd_base = [ + "aws", + "s3", + "sync", + folder_path, + s3_uri, + "--no-progress", + "--region", + settings.aws_region, + ] + if settings.aws_profile and settings.aws_profile != "default": + cmd_base += ["--profile", settings.aws_profile] + + # Verify aws CLI is available before creating the job + try: + subprocess.run(["aws", "--version"], capture_output=True, check=True, timeout=5) + except (FileNotFoundError, subprocess.CalledProcessError): + return jsonify({"error": "aws CLI not found. Install the AWS CLI and try again."}), 500 + + job_id = str(uuid.uuid4()) + job = SyncJob( + job_id=job_id, + folder_path=folder_path, + s3_prefix=s3_prefix, + s3_uri=s3_uri, + cmd_base=cmd_base, + ) + _register(job) + + # Log the start + try: + from app.services.log_service import get_log_service + + get_log_service().info( + "large_folder_sync", + "sync_started", + f"Large folder sync started: {folder_path} → {s3_uri}", + {"job_id": job_id, "folder_path": folder_path, "s3_uri": s3_uri}, + ) + except Exception: + pass + + thread = threading.Thread(target=_stream_process, args=(job,), daemon=True) + thread.start() + + cmd_display = " ".join(cmd_base) + return jsonify({"job_id": job_id, "s3_uri": s3_uri, "cmd": cmd_display}), 202 + + +@large_folder_upload_bp.route("/progress/", methods=["GET"]) +def stream_progress(job_id: str) -> Response: + """SSE stream of aws s3 sync output lines for a job.""" + + def generate() -> Generator[str, None, None]: + sse_mgr = get_sse_manager() + queue, event = sse_mgr.register_client(job_id) + try: + job = _get(job_id) + if not job: + yield f"data: {json.dumps({'error': 'Job not found'})}\n\n" + return + + # Replay lines already captured before client connected + for line in list(job.lines): + yield f"data: {json.dumps({'type': 'line', 'line': line})}\n\n" + + # If job already finished before the SSE connection opened, send done immediately + if job.status in ("completed", "failed", "cancelled"): + done_payload = { + "type": "done", + "status": job.status, + "return_code": job.return_code, + } + yield f"data: {json.dumps(done_payload)}\n\n" + return + + last_heartbeat = time.time() + while True: + while queue: + data = queue.popleft() + yield f"data: {json.dumps(data)}\n\n" + last_heartbeat = time.time() + if data.get("type") == "done": + return + + now = time.time() + if now - last_heartbeat > sse_mgr.heartbeat_interval: + yield ": heartbeat\n\n" + last_heartbeat = now + + event.wait(timeout=sse_mgr.heartbeat_interval) + event.clear() + + # Re-check job existence + if not _get(job_id): + yield f"data: {json.dumps({'error': 'Job not found'})}\n\n" + return + finally: + sse_mgr.deregister_client(job_id, queue) + + return Response( + generate(), + mimetype="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + "Connection": "keep-alive", + }, + ) + + +@large_folder_upload_bp.route("/cancel/", methods=["POST"]) +def cancel_sync(job_id: str) -> tuple[Response, int]: + """Cancel a running sync job.""" + job = _get(job_id) + if not job: + return jsonify({"error": "Job not found"}), 404 + if job.status != "running": + return jsonify({"message": "Job is not running", "status": job.status}), 200 + + job.status = "cancelled" + if job.process and job.process.poll() is None: + job.process.terminate() + try: + job.process.wait(timeout=5) + except subprocess.TimeoutExpired: + job.process.kill() + + get_sse_manager().send_event( + job_id, + {"type": "done", "status": "cancelled", "return_code": None}, + ) + return jsonify({"job_id": job_id, "status": "cancelled"}), 200 diff --git a/app/routes/main.py b/app/routes/main.py index 76107ba..ae61138 100644 --- a/app/routes/main.py +++ b/app/routes/main.py @@ -17,6 +17,7 @@ @main_bp.route("/") @main_bp.route("/delete") @main_bp.route("/files") +@main_bp.route("/large-folder-upload") @main_bp.route("/settings") @main_bp.route("/logs") def serve_spa() -> Response: diff --git a/app/routes/settings.py b/app/routes/settings.py index f58fb9a..541cfe3 100644 --- a/app/routes/settings.py +++ b/app/routes/settings.py @@ -194,26 +194,120 @@ def check_updates() -> tuple[Response, int]: @settings_bp.route("/update", methods=["POST"]) def run_update() -> tuple[Response, int]: - """Run application update (git pull + pip install). + """Run application update (git pull + pip install + frontend build). Returns: - JSON response with update results + JSON response with update results including pre_update_commit for rollback """ updater = get_updater() result = updater.update_application() - # Determine overall success - all_success = all(step["success"] for step in result.values()) + log = get_log_service() + if result["success"]: + log.info("settings", "app_updated", "Application updated successfully") + else: + log.warning( + "settings", + "app_update_failed", + f"Update failed at step: {result.get('failed_at')}", + {"failed_at": result.get("failed_at")}, + ) - return jsonify( - { - "success": all_success, - "results": result, - "message": "Update completed successfully" - if all_success - else "Update completed with some errors", - } - ), 200 + return jsonify(result), 200 + + +@settings_bp.route("/rollback", methods=["POST"]) +def rollback_update() -> tuple[Response, int]: + """Roll back to a previous commit. + + Request body: + commit: Full git commit hash to roll back to + + Returns: + JSON response with rollback result + """ + if not request.is_json: + return jsonify({"error": "JSON body required"}), 400 + + data = request.get_json() or {} + commit = data.get("commit", "").strip() + + if not commit: + return jsonify({"error": "commit hash required"}), 400 + + updater = get_updater() + result = updater.rollback_update(commit) + + log = get_log_service() + if result["success"]: + log.info( + "settings", + "app_rolled_back", + f"Application rolled back to {commit[:7]}", + {"commit": commit}, + ) + else: + log.error( + "settings", + "rollback_failed", + f"Rollback to {commit[:7]} failed: {result.get('error')}", + {"commit": commit, "error": result.get("error")}, + ) + + return jsonify(result), 200 + + +@settings_bp.route("/branches", methods=["GET"]) +def get_branches() -> tuple[Response, int]: + """Get current branch and list of all available branches. + + Returns: + JSON response with current branch and branch list + """ + updater = get_updater() + result = updater.get_branches() + return jsonify(result), 200 + + +@settings_bp.route("/branches/switch", methods=["POST"]) +def switch_branch() -> tuple[Response, int]: + """Switch to a specified git branch. + + Request body: + branch: Name of the branch to switch to + + Returns: + JSON response with switch result + """ + if not request.is_json: + return jsonify({"error": "JSON body required"}), 400 + + data = request.get_json() or {} + branch = data.get("branch", "").strip() + + if not branch: + return jsonify({"error": "branch name required"}), 400 + + updater = get_updater() + result = updater.switch_branch(branch) + + log = get_log_service() + if result["success"]: + log.info( + "settings", + "branch_switch", + f"Switched to branch '{branch}'", + {"branch": branch}, + ) + else: + log.warning( + "settings", + "branch_switch", + f"Failed to switch to branch '{branch}': {result['error']}", + {"branch": branch, "error": result["error"]}, + ) + + return jsonify(result), 200 @settings_bp.route("/cache/stats", methods=["GET"]) diff --git a/app/routes/upload.py b/app/routes/upload.py index cd4cb74..419c916 100644 --- a/app/routes/upload.py +++ b/app/routes/upload.py @@ -4,7 +4,6 @@ import tempfile import threading import time -from collections import deque from collections.abc import Callable, Generator from pathlib import Path from typing import Any @@ -12,6 +11,7 @@ from flask import Blueprint, Response, jsonify, request from app.config import get_settings +from app.services.sse_manager import get_sse_manager from app.services.upload_manager import ( FileUploadState, UploadJob, @@ -21,57 +21,6 @@ upload_bp = Blueprint("upload", __name__) -# Store for SSE clients per job -_sse_queues: dict[str, list[deque[dict[str, Any]]]] = {} -_sse_events: dict[str, threading.Event] = {} # Events to signal new data (replaces polling) -_sse_timestamps: dict[str, float] = {} # Track last activity for TTL cleanup -_sse_lock = threading.Lock() - -# Configuration -SSE_QUEUE_TTL_SECONDS = 3600 # Clean up queues after 1 hour of inactivity -SSE_HEARTBEAT_INTERVAL_SECONDS = 15 # Send heartbeat every 15 seconds - - -def _cleanup_old_sse_queues() -> int: - """Clean up SSE queues that haven't been accessed recently. - - Returns: - Number of queues removed - """ - now = time.time() - removed = 0 - with _sse_lock: - expired = [ - job_id - for job_id, timestamp in _sse_timestamps.items() - if now - timestamp > SSE_QUEUE_TTL_SECONDS - ] - for job_id in expired: - _sse_queues.pop(job_id, None) - _sse_events.pop(job_id, None) - _sse_timestamps.pop(job_id, None) - removed += 1 - return removed - - -def send_sse_event(job_id: str, data: dict[str, Any]) -> None: - """Send an SSE event to all clients listening for a job. - - This wakes up all waiting SSE generators via the event signal, - avoiding busy-wait polling. - """ - with _sse_lock: - queues = _sse_queues.get(job_id, []) - for q in queues: - q.append(data) - - # Update last activity timestamp - _sse_timestamps[job_id] = time.time() - - # Signal waiting threads that new data is available - if job_id in _sse_events: - _sse_events[job_id].set() - def _make_analysis_callback( job_id: str, @@ -79,7 +28,7 @@ def _make_analysis_callback( """Create an analysis progress callback that sends SSE events.""" def callback(job: UploadJob, file_state: FileUploadState) -> None: - send_sse_event( + get_sse_manager().send_event( job_id, { "type": "analysis_progress", @@ -94,6 +43,64 @@ def callback(job: UploadJob, file_state: FileUploadState) -> None: return callback +# 4 Hz cap on non-terminal progress events. The S3 byte_callback fires per chunk +# (~1 M times across a 1 TB upload); without throttling we would emit ~1 M SSE +# events. Terminal events (COMPLETED / FAILED / CANCELLED) always bypass the +# throttle so the frontend never misses the final state. +SSE_EMIT_INTERVAL_SECONDS = 0.25 +_TERMINAL_JOB_STATUSES = ( + UploadStatus.COMPLETED, + UploadStatus.FAILED, + UploadStatus.CANCELLED, +) + + +def _make_throttled_progress_callback( + large_job_threshold: int | None = None, +) -> Callable[[UploadJob], None]: + """Create a progress callback that coalesces SSE events at 4 Hz. + + Terminal events always emit. Closure-local state means each job gets its + own throttle window — safe to share across threads since callers hold + ``job._progress_lock`` for the check-and-stamp. + """ + + def progress_callback(job: UploadJob) -> None: + is_terminal = job.status in _TERMINAL_JOB_STATUSES + if not is_terminal: + now = time.monotonic() + with job._progress_lock: + if (now - job._last_emit_ts) < SSE_EMIT_INTERVAL_SECONDS: + return + job._last_emit_ts = now + + sse = get_sse_manager() + if is_terminal: + # Large jobs read per-file results from /api/upload/results (SQLite-backed) + # rather than receiving a 10k-row payload here. Small jobs keep the + # legacy behavior — frontend merges the full file array directly. + if ( + large_job_threshold is not None + and len(job.files) >= large_job_threshold + ): + payload = job.to_progress_dict() + payload["terminal"] = True + sse.send_event(job.job_id, payload) + else: + sse.send_event(job.job_id, job.to_dict()) + else: + sse.send_event(job.job_id, job.to_progress_dict()) + + return progress_callback + + +def _large_job_threshold() -> int: + """Read the live setting for the large-job cutoff.""" + return int( + get_settings().batch_processing.get("large_job_threshold", 1000) + ) + + @upload_bp.route("/analyze", methods=["POST"]) def analyze_files() -> tuple[Response, int]: """Analyze uploaded files and prepare for upload. @@ -150,7 +157,7 @@ def run_analysis() -> None: # Send final job state when analysis completes final_job = manager.get_job(job.job_id) if final_job: - send_sse_event( + get_sse_manager().send_event( job.job_id, { "type": "analysis_complete", @@ -198,12 +205,7 @@ def start_upload(job_id: str) -> tuple[Response, int]: if data: skip_duplicates = data.get("skip_duplicates", True) - def progress_callback(job: UploadJob) -> None: - """Send progress updates via SSE — lightweight during upload, full at completion.""" - if job.status in (UploadStatus.COMPLETED, UploadStatus.FAILED, UploadStatus.CANCELLED): - send_sse_event(job.job_id, job.to_dict()) - else: - send_sse_event(job.job_id, job.to_progress_dict()) + progress_callback = _make_throttled_progress_callback(_large_job_threshold()) # Start upload in background thread def run_upload() -> None: @@ -238,19 +240,12 @@ def get_progress(job_id: str) -> Response: manager = get_upload_manager() def generate() -> Generator[str, None, None]: - # Create a queue for this client and an event for signaling - queue: deque[dict[str, Any]] = deque() - event = threading.Event() - - with _sse_lock: - if job_id not in _sse_queues: - _sse_queues[job_id] = [] - _sse_queues[job_id].append(queue) - _sse_events[job_id] = event - _sse_timestamps[job_id] = time.time() + # Register this client with the SSE manager + sse_mgr = get_sse_manager() + queue, event = sse_mgr.register_client(job_id) # Periodic cleanup of old queues - _cleanup_old_sse_queues() + sse_mgr.cleanup_old_queues() try: # Send initial state @@ -281,7 +276,38 @@ def generate() -> Generator[str, None, None]: } yield f"data: {json.dumps(replay)}\n\n" elif scan_job: - yield f"data: {json.dumps({'type': 'scan_initial', 'status': scan_job.status})}\n\n" + if scan_job.status in ("completed", "failed", "cancelled"): + # Fast/cached scan completed before this EventSource connected — + # all SSE events were sent to an empty queue and dropped. + # Replay the full results immediately so the frontend never waits + # for the 15-second heartbeat timeout. + for folder_data in scan_job.scanned_folders: + replay_event = { + "type": "scan_folder_complete", + "folder": folder_data, + "folders_scanned": scan_job.folders_scanned, + "folders_total": scan_job.folders_total, + "running_totals": { + "total_files_found": scan_job.total_files_found, + "total_already_uploaded": scan_job.total_already_uploaded, + "total_size": scan_job.total_size, + }, + } + yield f"data: {json.dumps(replay_event)}\n\n" + terminal_data = { + "type": "scan_complete", + "status": scan_job.status, + "folders_scanned": scan_job.folders_scanned, + "folders_total": scan_job.folders_total, + "total_files_found": scan_job.total_files_found, + "total_already_uploaded": scan_job.total_already_uploaded, + "total_size": scan_job.total_size, + } + yield f"data: {json.dumps(terminal_data)}\n\n" + return + else: + initial = {"type": "scan_initial", "status": scan_job.status} + yield f"data: {json.dumps(initial)}\n\n" last_heartbeat_time = time.time() @@ -304,12 +330,12 @@ def generate() -> Generator[str, None, None]: # Send heartbeat if no activity for a while now = time.time() - if now - last_heartbeat_time > SSE_HEARTBEAT_INTERVAL_SECONDS: + if now - last_heartbeat_time > sse_mgr.heartbeat_interval: yield ": heartbeat\n\n" # Comment line, ignored by EventSource last_heartbeat_time = now # Wait for signal (blocking, no CPU waste) with timeout for heartbeat - event.wait(timeout=SSE_HEARTBEAT_INTERVAL_SECONDS) + event.wait(timeout=sse_mgr.heartbeat_interval) event.clear() # Check if job still exists (upload or scan) @@ -355,15 +381,7 @@ def generate() -> Generator[str, None, None]: return finally: - # Clean up queue and event - with _sse_lock: - if job_id in _sse_queues and queue in _sse_queues[job_id]: - _sse_queues[job_id].remove(queue) - if not _sse_queues[job_id]: - # Last client disconnected, remove event too - _sse_queues.pop(job_id, None) - _sse_events.pop(job_id, None) - # Keep timestamp for TTL cleanup + sse_mgr.deregister_client(job_id, queue) return Response( generate(), @@ -494,13 +512,13 @@ def cleanup_sse_queues() -> tuple[Response, int]: Returns: JSON response with cleanup statistics """ - removed = _cleanup_old_sse_queues() + removed = get_sse_manager().cleanup_old_queues() return jsonify( { "success": True, "queues_removed": removed, - "active_queues": len(_sse_queues), - "ttl_seconds": SSE_QUEUE_TTL_SECONDS, + "active_queues": get_sse_manager().queue_count, + "ttl_seconds": get_sse_manager().ttl_seconds, } ), 200 @@ -628,7 +646,7 @@ def scan_folder_async() -> tuple[Response, int]: ) def scan_progress_callback(job_id: str, event_data: dict[str, Any]) -> None: - send_sse_event(job_id, event_data) + get_sse_manager().send_event(job_id, event_data) def run_scan() -> None: manager.scan_folder_async( @@ -709,12 +727,7 @@ def bulk_analyze() -> tuple[Response, int]: job.pre_filter_stats = pre_filter_stats analysis_progress_callback = _make_analysis_callback(job.job_id) - def upload_progress_callback(job: UploadJob) -> None: - """Send upload progress updates via SSE — lightweight during upload, full at completion.""" - if job.status in (UploadStatus.COMPLETED, UploadStatus.FAILED, UploadStatus.CANCELLED): - send_sse_event(job.job_id, job.to_dict()) - else: - send_sse_event(job.job_id, job.to_progress_dict()) + upload_progress_callback = _make_throttled_progress_callback(_large_job_threshold()) # Start in background thread def run_bulk_job() -> None: @@ -741,7 +754,7 @@ def run_bulk_job() -> None: ) final_job = manager.get_job(job.job_id) if final_job: - send_sse_event( + get_sse_manager().send_event( job.job_id, { "type": "analysis_complete", diff --git a/app/services/delete_manager.py b/app/services/delete_manager.py index cbb8680..52c273c 100644 --- a/app/services/delete_manager.py +++ b/app/services/delete_manager.py @@ -2,8 +2,6 @@ import hashlib import os -import threading -import uuid from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field @@ -13,6 +11,7 @@ from typing import Any from app.services.cache_service import get_cache_service +from app.services.job_models import BaseFileState, BaseJob, BaseJobManager from app.services.log_service import get_log_service from app.services.s3_service import create_s3_client, get_object_metadata @@ -32,28 +31,22 @@ class DeleteStatus(Enum): @dataclass -class FileDeleteState: +class FileDeleteState(BaseFileState): """State for a single file in a delete job.""" - filename: str - local_path: str - file_size: int - s3_path: str - s3_bucket: str + s3_path: str = "" + s3_bucket: str = "" writable: bool = True status: DeleteStatus = DeleteStatus.PENDING local_md5: str = "" s3_etag: str = "" s3_size: int = 0 verification: str = "" - error_message: str = "" def to_dict(self) -> dict[str, Any]: """Serialize to dictionary for API responses.""" return { - "filename": self.filename, - "local_path": self.local_path, - "file_size": self.file_size, + **self._base_dict(), "s3_path": self.s3_path, "s3_bucket": self.s3_bucket, "writable": self.writable, @@ -62,33 +55,25 @@ def to_dict(self) -> dict[str, Any]: "s3_etag": self.s3_etag, "s3_size": self.s3_size, "verification": self.verification, - "error_message": self.error_message, } @dataclass -class DeleteJob: +class DeleteJob(BaseJob): """A delete job tracking multiple files.""" - job_id: str - files: list[FileDeleteState] = field(default_factory=list) status: str = "pending" created_at: str = field(default_factory=lambda: datetime.now(UTC).isoformat()) started_at: str | None = None completed_at: str | None = None - cancelled: bool = False - lock: threading.Lock = field(default_factory=threading.Lock) def to_dict(self) -> dict[str, Any]: """Serialize to dictionary for API responses.""" with self.lock: - status_counts: dict[str, int] = {} - total_deleted_size = 0 - for f in self.files: - status_counts[f.status.value] = status_counts.get(f.status.value, 0) + 1 - if f.status == DeleteStatus.DELETED: - total_deleted_size += f.file_size - + status_counts = self._compute_status_counts() + total_deleted_size = sum( + f.file_size for f in self.files if f.status == DeleteStatus.DELETED + ) return { "job_id": self.job_id, "status": self.status, @@ -105,20 +90,16 @@ def to_dict(self) -> dict[str, Any]: def to_progress_dict(self) -> dict[str, Any]: """Lightweight progress for SSE streaming.""" with self.lock: - status_counts: dict[str, int] = {} - total_deleted_size = 0 - for f in self.files: - status_counts[f.status.value] = status_counts.get(f.status.value, 0) + 1 - if f.status == DeleteStatus.DELETED: - total_deleted_size += f.file_size - + status_counts = self._compute_status_counts() + total_deleted_size = sum( + f.file_size for f in self.files if f.status == DeleteStatus.DELETED + ) files_processed = sum( 1 for f in self.files if f.status not in (DeleteStatus.PENDING, DeleteStatus.VERIFYING, DeleteStatus.DELETING) ) - return { "job_id": self.job_id, "status": self.status, @@ -164,11 +145,11 @@ def is_multipart_etag(etag: str) -> bool: return "-" in etag -class DeleteManager: +class DeleteManager(BaseJobManager): """Manages local file deletion jobs with MD5 verification against S3.""" def __init__(self, batch_config: dict[str, Any] | None = None) -> None: - self.jobs: dict[str, DeleteJob] = {} + super().__init__() # Load batch processing configuration if batch_config is None: @@ -203,7 +184,7 @@ def scan_folder( Returns: A new DeleteJob with files matched against the cache """ - job_id = str(uuid.uuid4()) + job_id = self._new_job_id() job = DeleteJob(job_id=job_id) cache = get_cache_service() folder = Path(folder_path) @@ -243,7 +224,7 @@ def scan_folder( ) job.files.append(file_state) - self.jobs[job_id] = job + self._register_job(job) return job def start_delete_job( @@ -411,7 +392,7 @@ def verify_against_s3(file_state: FileDeleteState) -> None: # Process verification in batches def check_cancelled() -> bool: - return job.cancelled + return bool(job.cancelled) # Create a wrapper that uses _verify_batch def process_batch_fn( @@ -662,21 +643,6 @@ def _finalize_cancelled( if progress_callback: progress_callback(job) - def cancel_job(self, job_id: str) -> bool: - """Cancel a delete job. - - Args: - job_id: The job to cancel - - Returns: - True if job was found and cancelled - """ - job = self.jobs.get(job_id) - if not job: - return False - job.cancelled = True - return True - def get_job(self, job_id: str) -> DeleteJob | None: """Retrieve a delete job by ID.""" return self.jobs.get(job_id) diff --git a/app/services/file_service.py b/app/services/file_service.py new file mode 100644 index 0000000..220b46a --- /dev/null +++ b/app/services/file_service.py @@ -0,0 +1,101 @@ +"""File service for handling generic file operations and path generation.""" + +from datetime import UTC, datetime +from pathlib import Path + +from app.config import get_settings +from app.services import mcap_service + + +def extract_timestamp(file_path: str, skip_validation: bool = False) -> datetime: + """Extract timestamp from a file. + + For .mcap files, attempts to parse the internal timestamp (delegating to mcap_service). + For other files, uses the file modification time (UTC). + + Args: + file_path: Path to the file. + skip_validation: If True, skips expensive MCAP parsing. + + Returns: + datetime: The extracted timestamp (UTC). + """ + path = Path(file_path) + ext = path.suffix.lower() + + if ext == ".mcap": + # For MCAP, try internal timestamp first + try: + return mcap_service.extract_start_time(file_path, skip_validation=skip_validation) + except Exception: + # Fallback to mtime if MCAP extraction fails (optional, but robust) + pass + + # For non-MCAP (or failed MCAP), use filesystem modification time + stat = path.stat() + return datetime.fromtimestamp(stat.st_mtime, tz=UTC) + + +def get_file_category(filename: str) -> str: + """Return the configured file category name for a given filename. + + Looks up the extension in the application's `file_categories` setting. Falls + back to "other" when no category matches (the same fallback used by + `generate_s3_key`). + """ + settings = get_settings() + ext = Path(filename).suffix.lower().lstrip(".") + if not ext: + return "other" + for cat in settings.file_categories: + if ext in [e.lower().lstrip(".") for e in cat.get("extensions", [])]: + return str(cat.get("name", ext)) + return "other" + + +def generate_s3_key(filename: str, timestamp: datetime) -> str: + """Generate Hive-partitioned S3 key based on file category configuration. + + The path structure and partition frequency are determined by the 'file_categories' + setting in the application configuration. + + Args: + filename: Name of the file. + timestamp: Timestamp to use for partitioning. + + Returns: + str: The S3 object key. + """ + settings = get_settings() + categories = settings.file_categories + + # Extract extension (lowercase, no dot) + ext = Path(filename).suffix.lower().lstrip(".") + if not ext: + ext = "other" + + # Find category for this extension + category_name = ext # Default to extension name if no category found + partition_interval = "daily" # Default to daily + + for cat in categories: + if ext in [e.lower().lstrip(".") for e in cat.get("extensions", [])]: + category_name = cat.get("name", ext) + partition_interval = cat.get("partition_interval", "daily") + break + + # Base path: category/year/month/day + base_path = ( + f"{category_name}/" + f"year={timestamp.year:04d}/" + f"month={timestamp.month:02d}/" + f"day={timestamp.day:02d}" + ) + + if partition_interval == "10min": + # 10-minute buckets + minute_bucket = (timestamp.minute // 10) * 10 + return f"{base_path}/hour={timestamp.hour:02d}/minute={minute_bucket:02d}/{filename}" + else: + # Daily buckets (no hour/minute) - default fallback + return f"{base_path}/{filename}" diff --git a/app/services/job_models.py b/app/services/job_models.py new file mode 100644 index 0000000..1dd2895 --- /dev/null +++ b/app/services/job_models.py @@ -0,0 +1,173 @@ +"""Shared base classes for upload and delete job models. + +Provides: +- ``BaseFileState``: common fields and helpers for per-file state objects. +- ``BaseJob``: common fields and helpers for job container dataclasses. +- ``BaseJobManager``: common job registry, ``get_job``, ``cancel_job``, and + ``cleanup_old_jobs`` shared by ``UploadManager`` and ``DeleteManager``. +""" + +import threading +import uuid +from dataclasses import dataclass, field +from datetime import UTC, datetime +from typing import Any + +# --------------------------------------------------------------------------- +# File state base +# --------------------------------------------------------------------------- + + +@dataclass +class BaseFileState: + """Common fields for per-file workflow state (upload or delete). + + Subclasses add workflow-specific fields and call ``_base_dict()`` from + their own ``to_dict()`` implementation. + """ + + filename: str + local_path: str + file_size: int + error_message: str = "" + + def _base_dict(self) -> dict[str, Any]: + """Return the fields shared across all file state types.""" + return { + "filename": self.filename, + "local_path": self.local_path, + "file_size": self.file_size, + "error_message": self.error_message, + } + + +# --------------------------------------------------------------------------- +# Job container base +# --------------------------------------------------------------------------- + + +@dataclass +class BaseJob: + """Common fields and helpers for job container dataclasses. + + Both ``UploadJob`` and ``DeleteJob`` inherit from this. Subclasses keep + their own ``status`` field (UploadJob uses an Enum, DeleteJob uses str) + and override ``to_dict()`` / ``to_progress_dict()`` as needed. + """ + + job_id: str + files: list[Any] = field(default_factory=list) + cancelled: bool = False + lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + + def _compute_status_counts(self) -> dict[str, int]: + """Count files by status value. + + Works for both Enum-based statuses (``f.status.value``) and plain + string statuses. + """ + counts: dict[str, int] = {} + for f in self.files: + val: str = f.status.value if hasattr(f.status, "value") else str(f.status) + counts[val] = counts.get(val, 0) + 1 + return counts + + +# --------------------------------------------------------------------------- +# Job manager base +# --------------------------------------------------------------------------- + + +class BaseJobManager: + """Common job registry for upload and delete managers. + + Provides thread-safe job registration, lookup, cancellation, and + age-based cleanup. Subclasses implement workflow-specific logic by + overriding ``_on_cancel`` and optionally ``_completed_at_datetime``. + """ + + def __init__(self) -> None: + self.jobs: dict[str, Any] = {} + self._lock = threading.Lock() + + # ------------------------------------------------------------------ + # Registry + # ------------------------------------------------------------------ + + def _register_job(self, job: Any) -> None: + """Add a job to the registry.""" + with self._lock: + self.jobs[job.job_id] = job + + def get_job(self, job_id: str) -> Any | None: + """Return a job by ID, or None if not found.""" + return self.jobs.get(job_id) + + # ------------------------------------------------------------------ + # Cancellation + # ------------------------------------------------------------------ + + def cancel_job(self, job_id: str) -> bool: + """Set ``job.cancelled = True`` and call ``_on_cancel``. + + Returns: + True if the job was found (and cancellation flagged). + """ + job = self.get_job(job_id) + if not job: + return False + job.cancelled = True + self._on_cancel(job) + return True + + def _on_cancel(self, job: Any) -> None: + """Hook called after ``cancelled`` is set. Override for cleanup.""" + + # ------------------------------------------------------------------ + # Cleanup + # ------------------------------------------------------------------ + + def _completed_at_datetime(self, job: Any) -> datetime | None: + """Return ``job.completed_at`` as a timezone-aware datetime, or None. + + Handles both ``datetime`` objects (UploadJob) and ISO 8601 strings + (DeleteJob) transparently. + """ + ca = job.completed_at + if ca is None: + return None + if isinstance(ca, datetime): + return ca if ca.tzinfo else ca.replace(tzinfo=UTC) + try: + return datetime.fromisoformat(str(ca)) + except (ValueError, TypeError): + return None + + def cleanup_old_jobs(self, max_age_seconds: int = 3600) -> int: + """Remove completed jobs older than ``max_age_seconds`` from memory. + + Returns: + Number of jobs removed. + """ + now = datetime.now(UTC) + removed = 0 + with self._lock: + to_remove = [ + job_id + for job_id, job in self.jobs.items() + if (completed_at := self._completed_at_datetime(job)) is not None + and (now - completed_at).total_seconds() > max_age_seconds + ] + for job_id in to_remove: + del self.jobs[job_id] + removed += 1 + return removed + + # ------------------------------------------------------------------ + # Convenience + # ------------------------------------------------------------------ + + @staticmethod + def _new_job_id() -> str: + """Generate a new unique job ID.""" + return str(uuid.uuid4()) diff --git a/app/services/job_storage.py b/app/services/job_storage.py index 4d3b7b3..5e8fc41 100644 --- a/app/services/job_storage.py +++ b/app/services/job_storage.py @@ -74,6 +74,10 @@ def _initialize_db(self) -> None: """Initialize the database and create tables if needed.""" try: conn = sqlite3.connect(str(DB_FILE), check_same_thread=False) + # WAL mode lets one writer + many readers proceed concurrently — + # required because per-file status updates fire from upload worker + # threads while the Flask route handlers serve /api/upload/results. + conn.execute("PRAGMA journal_mode=WAL") conn.executescript(SCHEMA_SQL) conn.commit() conn.close() diff --git a/app/services/sse_manager.py b/app/services/sse_manager.py new file mode 100644 index 0000000..2fbbd4a --- /dev/null +++ b/app/services/sse_manager.py @@ -0,0 +1,134 @@ +"""Shared Server-Sent Events (SSE) manager for job progress streaming.""" + +import threading +import time +from collections import deque +from typing import Any + +# Default configuration constants (can be overridden at construction time) +SSE_QUEUE_TTL_SECONDS = 3600 # Remove queues after 1 hour of inactivity +SSE_HEARTBEAT_INTERVAL_SECONDS = 15 # Heartbeat cadence for the /progress endpoints + + +class SSEManager: + """Manages SSE client queues, event signaling, and TTL cleanup for job streams. + + Supports multiple concurrent clients per job. Each client gets its own + ``deque`` that receives a copy of every event broadcast via ``send_event()``. + A shared ``threading.Event`` per job allows the progress generator to block + efficiently (no busy-polling) until new data arrives. + + Usage pattern in a route:: + + manager = get_sse_manager() + + def generate(): + queue, event = manager.register_client(job_id) + try: + # ... yield initial state, then loop reading queue / waiting on event + finally: + manager.deregister_client(job_id, queue) + + # Background thread / callback: + manager.send_event(job_id, {"type": "progress", ...}) + """ + + def __init__( + self, + ttl_seconds: int = SSE_QUEUE_TTL_SECONDS, + heartbeat_interval: int = SSE_HEARTBEAT_INTERVAL_SECONDS, + ) -> None: + self._queues: dict[str, list[deque[dict[str, Any]]]] = {} + self._events: dict[str, threading.Event] = {} + self._timestamps: dict[str, float] = {} + self._lock = threading.Lock() + self.ttl_seconds = ttl_seconds + self.heartbeat_interval = heartbeat_interval + + def send_event(self, job_id: str, data: dict[str, Any]) -> None: + """Broadcast an event to all clients currently listening for ``job_id``. + + Thread-safe. Wakes up any blocked generator via the job's Event. + """ + with self._lock: + for q in self._queues.get(job_id, []): + q.append(data) + self._timestamps[job_id] = time.time() + if job_id in self._events: + self._events[job_id].set() + + def register_client(self, job_id: str) -> tuple[deque[dict[str, Any]], threading.Event]: + """Register a new SSE client for ``job_id``. + + Returns a ``(queue, event)`` tuple. The generator should read from + ``queue`` and call ``event.wait()`` when the queue is empty. + """ + queue: deque[dict[str, Any]] = deque() + with self._lock: + if job_id not in self._queues: + self._queues[job_id] = [] + self._queues[job_id].append(queue) + if job_id not in self._events: + self._events[job_id] = threading.Event() + event = self._events[job_id] + self._timestamps[job_id] = time.time() + return queue, event + + def deregister_client(self, job_id: str, queue: deque[dict[str, Any]]) -> None: + """Remove a client queue when its connection closes. + + If this was the last client for ``job_id``, the Event is also removed + (but the timestamp is kept for TTL cleanup). + """ + with self._lock: + if job_id in self._queues: + try: + self._queues[job_id].remove(queue) + except ValueError: + pass + if not self._queues[job_id]: + del self._queues[job_id] + self._events.pop(job_id, None) + + @property + def queue_count(self) -> int: + """Number of jobs that currently have at least one active client queue.""" + with self._lock: + return len(self._queues) + + def cleanup_old_queues(self) -> int: + """Remove SSE state for jobs idle longer than ``ttl_seconds``. + + Returns: + Number of job entries removed. + """ + now = time.time() + removed = 0 + with self._lock: + expired = [ + job_id for job_id, ts in self._timestamps.items() if now - ts > self.ttl_seconds + ] + for job_id in expired: + self._queues.pop(job_id, None) + self._events.pop(job_id, None) + self._timestamps.pop(job_id, None) + removed += 1 + return removed + + +# --------------------------------------------------------------------------- +# Singleton accessor +# --------------------------------------------------------------------------- + +_sse_manager: SSEManager | None = None +_sse_manager_lock = threading.Lock() + + +def get_sse_manager() -> SSEManager: + """Return the process-wide SSEManager singleton.""" + global _sse_manager + if _sse_manager is None: + with _sse_manager_lock: + if _sse_manager is None: + _sse_manager = SSEManager() + return _sse_manager diff --git a/app/services/upload_manager.py b/app/services/upload_manager.py index 2237018..fbcc0f3 100644 --- a/app/services/upload_manager.py +++ b/app/services/upload_manager.py @@ -4,7 +4,6 @@ import os import shutil import threading -import uuid from collections.abc import Callable from concurrent.futures import ( FIRST_COMPLETED, @@ -19,8 +18,9 @@ from pathlib import Path from typing import Any -from app.services import mcap_service, s3_service +from app.services import file_service, mcap_service, s3_service from app.services.cache_service import get_cache_service +from app.services.job_models import BaseFileState, BaseJob, BaseJobManager from app.services.log_service import get_log_service from app.services.s3_service import UploadCancelledError from app.services.utils import format_file_size @@ -42,7 +42,7 @@ def _extract_start_time_worker(local_path: str, skip_validation: bool = False) - datetime on success, or error message string on failure. """ try: - return mcap_service.extract_start_time(local_path, skip_validation=skip_validation) + return file_service.extract_timestamp(local_path, skip_validation=skip_validation) except Exception as e: return str(e) @@ -61,17 +61,13 @@ class UploadStatus(Enum): @dataclass -class FileUploadState: +class FileUploadState(BaseFileState): """State of a single file in an upload job.""" - filename: str - local_path: str - file_size: int status: UploadStatus = UploadStatus.PENDING s3_path: str = "" start_time: datetime | None = None # MCAP file's data start time bytes_uploaded: int = 0 - error_message: str = "" is_duplicate: bool = False is_valid: bool = True # False if timestamp is invalid (1970/epoch) upload_started_at: datetime | None = None # When upload began @@ -88,12 +84,11 @@ def to_dict(self) -> dict[str, Any]: """Convert to dictionary for JSON serialization.""" duration = self.upload_duration_seconds return { - "filename": self.filename, - "local_path": self.local_path, - "file_size": self.file_size, + **self._base_dict(), "file_size_formatted": format_file_size(self.file_size), "status": self.status.value, "s3_path": self.s3_path, + "file_category": file_service.get_file_category(self.filename), "start_time": self.start_time.isoformat() if self.start_time else None, "bytes_uploaded": self.bytes_uploaded, "progress_percent": round( @@ -118,62 +113,159 @@ def to_dict(self) -> dict[str, Any]: @dataclass -class UploadJob: +class UploadJob(BaseJob): """Represents an upload job containing multiple files.""" - job_id: str - files: list[FileUploadState] = field(default_factory=list) status: UploadStatus = UploadStatus.PENDING created_at: datetime = field(default_factory=lambda: datetime.now(UTC)) started_at: datetime | None = None completed_at: datetime | None = None - cancelled: bool = False auto_upload: bool = False # Auto-start upload when analysis completes temp_dir: str | None = None # Temp directory for cleanup pre_filter_stats: dict[str, Any] = field(default_factory=dict) # Pre-filter statistics - lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + + # Cumulative counters maintained incrementally by set_file_status / + # set_bytes_uploaded so to_progress_dict() is O(1) in len(files). For 10k+ + # file jobs the prior O(N) sums fired per S3 chunk callback (~1M times) + # and dominated CPU. + total_uploaded_bytes: int = 0 + total_files_completed: int = 0 # COMPLETED + SKIPPED (matches legacy files_completed) + total_files_failed: int = 0 + total_files_skipped: int = 0 + total_files_uploaded: int = 0 # COMPLETED only + total_bytes_cached: int = 0 # Set once when files are populated + + # SSE throttle state — accessed under ``_progress_lock`` to coalesce + # per-chunk byte_callback emissions down to ~4 Hz. + _last_emit_ts: float = 0.0 + _progress_lock: threading.Lock = field(default_factory=threading.Lock, repr=False) + + # When True, terminal per-file transitions are mirrored to JobStorage so the + # frontend can read per-file results via /api/upload/results without holding + # 10k rows in browser memory. Set by UploadManager.create_job for large jobs. + _use_db: bool = False @property def total_bytes(self) -> int: - """Total bytes across all files.""" - return sum(f.file_size for f in self.files) + """Total bytes across all files (cached when files are populated).""" + if self.total_bytes_cached: + return self.total_bytes_cached + # Fallback for legacy callers that bypass create_job. + self.total_bytes_cached = sum(f.file_size for f in self.files) + return self.total_bytes_cached @property def uploaded_bytes(self) -> int: - """Total bytes uploaded across all files.""" - return sum(f.bytes_uploaded for f in self.files) + """Total bytes uploaded across all files (cumulative counter).""" + return self.total_uploaded_bytes @property def progress_percent(self) -> float: """Overall progress percentage.""" - if self.total_bytes == 0: + total = self.total_bytes + if total == 0: return 0.0 - return round(self.uploaded_bytes / self.total_bytes * 100, 1) + return round(self.total_uploaded_bytes / total * 100, 1) @property def files_completed(self) -> int: - """Number of files completed.""" - return sum( - 1 for f in self.files if f.status in (UploadStatus.COMPLETED, UploadStatus.SKIPPED) - ) + """Number of files in a terminal-success state (COMPLETED + SKIPPED).""" + return self.total_files_completed @property def files_failed(self) -> int: """Number of files failed.""" - return sum(1 for f in self.files if f.status == UploadStatus.FAILED) + return self.total_files_failed + + # ------------------------------------------------------------------ + # State mutation helpers — keep cumulative counters in sync + # ------------------------------------------------------------------ + + def _adjust_counters_for_status(self, status: UploadStatus, sign: int) -> None: + """Add ``sign`` (+1 or -1) to the counters for ``status``.""" + if status == UploadStatus.COMPLETED: + self.total_files_completed += sign + self.total_files_uploaded += sign + elif status == UploadStatus.SKIPPED: + self.total_files_completed += sign + self.total_files_skipped += sign + elif status == UploadStatus.FAILED: + self.total_files_failed += sign + + _TERMINAL_STATUSES = frozenset( + { + UploadStatus.COMPLETED, + UploadStatus.FAILED, + UploadStatus.SKIPPED, + UploadStatus.CANCELLED, + } + ) + + def set_file_status(self, file_state: "FileUploadState", new_status: UploadStatus) -> None: + """Transition a file's status and maintain cumulative counters. + + Idempotent on identical transitions. Use this instead of writing + ``file_state.status = ...`` directly so counters stay accurate. + + When ``self._use_db`` is True and the transition lands in a terminal + state, the change is mirrored to JobStorage so the frontend's summary + phase can lazy-load per-file results without holding them all in + browser memory. Storage failures are swallowed — bookkeeping must + never break an in-progress upload. + """ + old = file_state.status + if old == new_status: + return + self._adjust_counters_for_status(old, -1) + file_state.status = new_status + self._adjust_counters_for_status(new_status, +1) + + if self._use_db and new_status in UploadJob._TERMINAL_STATUSES: + self._persist_file_state(file_state) + + def _persist_file_state(self, file_state: "FileUploadState") -> None: + """Best-effort mirror of a file's terminal state into JobStorage.""" + try: + from app.services.job_storage import get_job_storage + + get_job_storage().update_file_status( + self.job_id, + file_state.filename, + file_state.status.value, + bytes_uploaded=file_state.bytes_uploaded, + error_message=file_state.error_message or None, + upload_started_at=file_state.upload_started_at, + upload_completed_at=file_state.upload_completed_at, + ) + except Exception: + logger.debug( + "JobStorage.update_file_status failed for %s/%s", + self.job_id, + file_state.filename, + exc_info=True, + ) + + def set_bytes_uploaded(self, file_state: "FileUploadState", new_bytes: int) -> None: + """Set ``file_state.bytes_uploaded`` and bump the cumulative byte counter.""" + delta = new_bytes - file_state.bytes_uploaded + if delta == 0: + return + file_state.bytes_uploaded = new_bytes + self.total_uploaded_bytes += delta @property def eta_seconds(self) -> int | None: """Estimated time remaining in seconds.""" - if not self.started_at or self.uploaded_bytes == 0: + uploaded = self.total_uploaded_bytes + if not self.started_at or uploaded == 0: return None elapsed = (datetime.now(UTC) - self.started_at).total_seconds() if elapsed <= 0: return None - bytes_per_second = self.uploaded_bytes / elapsed - remaining_bytes = self.total_bytes - self.uploaded_bytes + bytes_per_second = uploaded / elapsed + remaining_bytes = self.total_bytes - uploaded if bytes_per_second <= 0: return None @@ -210,26 +302,29 @@ def average_upload_speed_mbps(self) -> float | None: def to_progress_dict(self) -> dict[str, Any]: """Lightweight dict for SSE progress events. - Includes only job-level stats and currently active files (uploading/analyzing), - dropping the full 20K-file array that to_dict() includes. + O(1) in len(files) for all aggregate counters (maintained incrementally + by set_file_status / set_bytes_uploaded). The active_files slice scans + until 8 are found and breaks — typically only ~max_workers files are + active at any moment, so the scan terminates quickly. """ - active_files = [ - f.to_dict() - for f in self.files - if f.status in (UploadStatus.UPLOADING, UploadStatus.ANALYZING) - ] + active_files: list[dict[str, Any]] = [] + for f in self.files: + if f.status in (UploadStatus.UPLOADING, UploadStatus.ANALYZING): + active_files.append(f.to_dict()) + if len(active_files) >= 8: + break return { "job_id": self.job_id, "status": self.status.value, "progress_percent": self.progress_percent, - "files_completed": self.files_completed, + "files_completed": self.total_files_completed, "total_files": len(self.files), - "uploaded_bytes_formatted": format_file_size(self.uploaded_bytes), + "uploaded_bytes_formatted": format_file_size(self.total_uploaded_bytes), "total_bytes_formatted": format_file_size(self.total_bytes), "eta_seconds": self.eta_seconds, - "files_failed": self.files_failed, - "files_skipped": sum(1 for f in self.files if f.status == UploadStatus.SKIPPED), - "files_uploaded": sum(1 for f in self.files if f.status == UploadStatus.COMPLETED), + "files_failed": self.total_files_failed, + "files_skipped": self.total_files_skipped, + "files_uploaded": self.total_files_uploaded, "cancelled": self.cancelled, "files": active_files, } @@ -311,14 +406,13 @@ class ScanJob: lock: threading.Lock = field(default_factory=threading.Lock, repr=False) -class UploadManager: +class UploadManager(BaseJobManager): """Manages upload jobs and their execution.""" def __init__(self, max_workers: int = 4, batch_config: dict[str, Any] | None = None) -> None: - self.jobs: dict[str, UploadJob] = {} + super().__init__() self.scan_jobs: dict[str, ScanJob] = {} self.max_workers = max_workers - self._lock = threading.Lock() # Load batch processing configuration if batch_config is None: @@ -351,21 +445,34 @@ def create_job( Returns: The created UploadJob """ - job_id = str(uuid.uuid4()) + job_id = self._new_job_id() job = UploadJob(job_id=job_id, auto_upload=auto_upload, temp_dir=temp_dir) + total_bytes = 0 for path_str in file_paths: path = Path(path_str) if path.exists(): + size = path.stat().st_size file_state = FileUploadState( filename=path.name, local_path=str(path.absolute()), - file_size=path.stat().st_size, + file_size=size, ) job.files.append(file_state) + total_bytes += size + job.total_bytes_cached = total_bytes - with self._lock: - self.jobs[job_id] = job + # Large jobs: mirror per-file state into SQLite so the summary page can + # lazy-load results via /api/upload/results without sending all 10k rows + # over SSE at terminal. Best-effort: save_job failure must not break + # job creation. + if ( + self.batch_config.use_database_for_large_jobs + and len(job.files) >= self.batch_config.large_job_threshold + ): + self._save_job_to_storage(job, total_bytes) + + self._register_job(job) log = get_log_service() log.info( @@ -382,8 +489,66 @@ def create_job( return job + def _save_job_to_storage(self, job: UploadJob, total_bytes: int) -> None: + """Persist initial job + file rows into SQLite for large jobs. + + Sets ``job._use_db = True`` on success so ``set_file_status`` mirrors + subsequent terminal transitions. Failure is logged and swallowed — + bookkeeping must never block job creation. + """ + try: + from app.services.job_storage import get_job_storage + + file_states = [ + { + "filename": f.filename, + "local_path": f.local_path, + "file_size": f.file_size, + "status": f.status.value, + "s3_path": f.s3_path, + "start_time": (f.start_time.isoformat() if f.start_time else None), + "is_duplicate": f.is_duplicate, + "is_valid": f.is_valid, + } + for f in job.files + ] + get_job_storage().save_job( + job_id=job.job_id, + job_type="upload", + total_files=len(job.files), + file_states=file_states, + metadata={"total_bytes": total_bytes}, + ) + job._use_db = True + except Exception: + logger.warning( + "JobStorage.save_job failed for %s; falling back to in-memory only", + job.job_id, + exc_info=True, + ) + + def _persist_job_terminal(self, job: UploadJob) -> None: + """Mirror the job's terminal state to SQLite. Best-effort.""" + if not job._use_db: + return + try: + from app.services.job_storage import get_job_storage + + get_job_storage().update_job_status( + job_id=job.job_id, + status=job.status.value, + files_processed=job.total_files_completed + job.total_files_failed, + files_uploaded=job.total_files_uploaded, + files_failed=job.total_files_failed, + total_bytes=job.total_uploaded_bytes, + started_at=job.started_at, + completed_at=job.completed_at, + ) + except Exception: + logger.debug("JobStorage.update_job_status failed for %s", job.job_id, exc_info=True) + def get_job(self, job_id: str) -> UploadJob | None: - """Get a job by ID.""" + """Get an upload job by ID.""" return self.jobs.get(job_id) def analyze_job( @@ -424,22 +589,22 @@ def analyze_job( except Exception as e: job.status = UploadStatus.FAILED for file_state in job.files: - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = f"Failed to create S3 client: {e}" return job # Analyze each file for file_state in job.files: - file_state.status = UploadStatus.ANALYZING + job.set_file_status(file_state, UploadStatus.ANALYZING) try: - # Extract timestamp from MCAP - start_time = mcap_service.extract_start_time( + # Extract timestamp (supports MCAP and generic files) + start_time = file_service.extract_timestamp( file_state.local_path, skip_validation=skip_validation ) file_state.start_time = start_time # Generate S3 path - s3_path = mcap_service.generate_s3_path(start_time, file_state.filename) + s3_path = file_service.generate_s3_key(file_state.filename, start_time) file_state.s3_path = s3_path # Check for duplicates @@ -447,10 +612,10 @@ def analyze_job( s3_client, s3_bucket, s3_path ) - file_state.status = UploadStatus.READY + job.set_file_status(file_state, UploadStatus.READY) except Exception as e: - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = str(e) # Update job status @@ -516,22 +681,29 @@ def _analyze_single_file( The updated FileUploadState """ log = get_log_service() - file_state.status = UploadStatus.ANALYZING + if job is not None: + job.set_file_status(file_state, UploadStatus.ANALYZING) + else: + file_state.status = UploadStatus.ANALYZING if progress_callback and job: progress_callback(job, file_state) try: - # Extract timestamp from MCAP - start_time = mcap_service.extract_start_time( + # Extract timestamp (supports MCAP and generic files) + start_time = file_service.extract_timestamp( file_state.local_path, skip_validation=skip_validation ) file_state.start_time = start_time # Check if timestamp is valid (after 1980) + # Use file_service or mcap_service utility? mcap_service has to_naive_utc. + # I'll just keep using mcap_service.to_naive_utc as it's a utility. + from app.services import mcap_service + naive_start = mcap_service.to_naive_utc(start_time) file_state.is_valid = naive_start >= EPOCH_CUTOFF.replace(tzinfo=None) # Generate S3 path - s3_path = mcap_service.generate_s3_path(start_time, file_state.filename) + s3_path = file_service.generate_s3_key(file_state.filename, start_time) file_state.s3_path = s3_path # Check for duplicates - try cache first @@ -559,7 +731,10 @@ def _analyze_single_file( file_state.file_size, ) - file_state.status = UploadStatus.READY + if job is not None: + job.set_file_status(file_state, UploadStatus.READY) + else: + file_state.status = UploadStatus.READY log.info( "analysis", @@ -576,7 +751,10 @@ def _analyze_single_file( ) except Exception as e: - file_state.status = UploadStatus.FAILED + if job is not None: + job.set_file_status(file_state, UploadStatus.FAILED) + else: + file_state.status = UploadStatus.FAILED file_state.error_message = str(e) log.error( @@ -638,7 +816,7 @@ def analyze_job_async( except Exception as e: job.status = UploadStatus.FAILED for file_state in job.files: - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = f"Failed to create S3 client: {e}" if progress_callback: progress_callback(job, file_state) @@ -648,7 +826,7 @@ def analyze_job_async( # parallelism across cores, bypassing the GIL. cpu_workers = max(1, (os.cpu_count() or 4) - 1) for file_state in job.files: - file_state.status = UploadStatus.PENDING + job.set_file_status(file_state, UploadStatus.PENDING) files_iter_async = iter(job.files) active_async: dict[Any, FileUploadState] = {} @@ -657,7 +835,7 @@ def _submit_next_async(proc_executor: ProcessPoolExecutor) -> None: fs = next(files_iter_async, None) if fs is None or job.cancelled: return - fs.status = UploadStatus.ANALYZING + job.set_file_status(fs, UploadStatus.ANALYZING) if progress_callback: progress_callback(job, fs) # "queued → analyzing" event fut = proc_executor.submit(_extract_start_time_worker, fs.local_path, skip_validation) @@ -679,7 +857,7 @@ def _submit_next_async(proc_executor: ProcessPoolExecutor) -> None: result = future.result() if isinstance(result, str): # Error message returned from worker - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = result log.error( "analysis", @@ -689,10 +867,12 @@ def _submit_next_async(proc_executor: ProcessPoolExecutor) -> None: ) else: file_state.start_time = result + from app.services import mcap_service + naive_start = mcap_service.to_naive_utc(result) file_state.is_valid = naive_start >= EPOCH_CUTOFF.replace(tzinfo=None) - file_state.s3_path = mcap_service.generate_s3_path( - result, file_state.filename + file_state.s3_path = file_service.generate_s3_key( + file_state.filename, result ) if progress_callback: progress_callback(job, file_state) @@ -716,7 +896,7 @@ def _submit_next_async(proc_executor: ProcessPoolExecutor) -> None: file_state = dup_futures[fut] try: fut.result() - file_state.status = UploadStatus.READY + job.set_file_status(file_state, UploadStatus.READY) log.info( "analysis", "file_analysis_completed", @@ -732,7 +912,7 @@ def _submit_next_async(proc_executor: ProcessPoolExecutor) -> None: ) except Exception as e: with job.lock: - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = str(e) if progress_callback: @@ -794,7 +974,7 @@ def start_upload( job.status = UploadStatus.FAILED for file_state in job.files: if file_state.status == UploadStatus.READY: - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = f"Failed to create S3 client: {e}" return @@ -807,8 +987,8 @@ def start_upload( continue if skip_duplicates and file_state.is_duplicate: - file_state.status = UploadStatus.SKIPPED - file_state.bytes_uploaded = file_state.file_size + job.set_file_status(file_state, UploadStatus.SKIPPED) + job.set_bytes_uploaded(file_state, file_state.file_size) log.info( "upload", "file_upload_skipped", @@ -819,7 +999,7 @@ def start_upload( # Skip files with invalid timestamps if not file_state.is_valid: - file_state.status = UploadStatus.SKIPPED + job.set_file_status(file_state, UploadStatus.SKIPPED) file_state.error_message = "Invalid timestamp (pre-1980)" log.warning( "upload", @@ -848,12 +1028,12 @@ def make_upload_task( def upload_task() -> Any: if job.cancelled: with job.lock: - fs.status = UploadStatus.CANCELLED + job.set_file_status(fs, UploadStatus.CANCELLED) return None # Mark UPLOADING inside the worker so files stay READY until picked up with job.lock: - fs.status = UploadStatus.UPLOADING + job.set_file_status(fs, UploadStatus.UPLOADING) fs.upload_started_at = datetime.now(UTC) log.info( "upload", @@ -871,7 +1051,7 @@ def upload_task() -> Any: def byte_callback(uploaded: int, total: int) -> None: with job.lock: - fs.bytes_uploaded = uploaded + job.set_bytes_uploaded(fs, uploaded) if progress_callback: progress_callback(job) @@ -899,8 +1079,8 @@ def byte_callback(uploaded: int, total: int) -> None: continue file_state.upload_completed_at = datetime.now(UTC) if result["success"]: - file_state.status = UploadStatus.COMPLETED - file_state.bytes_uploaded = file_state.file_size + job.set_file_status(file_state, UploadStatus.COMPLETED) + job.set_bytes_uploaded(file_state, file_state.file_size) log.info( "upload", "file_upload_completed", @@ -926,7 +1106,7 @@ def byte_callback(uploaded: int, total: int) -> None: except Exception: logger.debug("Cache update failed after upload", exc_info=True) else: - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = result.get("error", "Unknown error") log.error( "upload", @@ -940,11 +1120,11 @@ def byte_callback(uploaded: int, total: int) -> None: ) except UploadCancelledError: with job.lock: - file_state.status = UploadStatus.CANCELLED + job.set_file_status(file_state, UploadStatus.CANCELLED) file_state.upload_completed_at = datetime.now(UTC) except Exception as e: file_state.upload_completed_at = datetime.now(UTC) - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = str(e) log.error( "upload", @@ -970,6 +1150,9 @@ def byte_callback(uploaded: int, total: int) -> None: # Clean up temp directory when upload completes self.cleanup_temp_dir(job_id) + # Mirror terminal job state to SQLite for large jobs (best-effort). + self._persist_job_terminal(job) + # Send terminal event IMMEDIATELY so the frontend unblocks. # Heavy I/O (logging, CSV, S3 sync) follows below. if progress_callback: @@ -1104,7 +1287,7 @@ def analyze_and_upload_pipeline( except Exception as e: job.status = UploadStatus.FAILED for file_state in job.files: - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = f"Failed to create S3 client: {e}" if analysis_callback: analysis_callback(job, file_state) @@ -1117,7 +1300,7 @@ def analyze_and_upload_pipeline( # Mark all files as PENDING (waiting their turn in the analysis pool) for fs in job.files: - fs.status = UploadStatus.PENDING + job.set_file_status(fs, UploadStatus.PENDING) files_iter = iter(job.files) active: dict[Any, FileUploadState] = {} @@ -1126,7 +1309,7 @@ def _submit_next(proc_executor: ProcessPoolExecutor) -> None: fs = next(files_iter, None) if fs is None or job.cancelled: return - fs.status = UploadStatus.ANALYZING + job.set_file_status(fs, UploadStatus.ANALYZING) if analysis_callback: analysis_callback(job, fs) # "queued → analyzing" event fut = proc_executor.submit(_extract_start_time_worker, fs.local_path, skip_validation) @@ -1151,7 +1334,7 @@ def _submit_next(proc_executor: ProcessPoolExecutor) -> None: if isinstance(result, str): # Parse failed - fs.status = UploadStatus.FAILED + job.set_file_status(fs, UploadStatus.FAILED) fs.error_message = result log.error( "analysis", @@ -1172,7 +1355,7 @@ def _submit_next(proc_executor: ProcessPoolExecutor) -> None: # Check duplicate (I/O but fast — cache lookup or S3 HEAD) self._check_duplicate(fs, s3_client, s3_bucket, use_cache) - fs.status = UploadStatus.READY + job.set_file_status(fs, UploadStatus.READY) log.info( "analysis", @@ -1197,7 +1380,7 @@ def _submit_next(proc_executor: ProcessPoolExecutor) -> None: # Decide: skip or upload? if not fs.is_valid: - fs.status = UploadStatus.SKIPPED + job.set_file_status(fs, UploadStatus.SKIPPED) fs.error_message = "Invalid timestamp (pre-1980)" log.warning( "upload", @@ -1214,8 +1397,8 @@ def _submit_next(proc_executor: ProcessPoolExecutor) -> None: continue if skip_duplicates and fs.is_duplicate: - fs.status = UploadStatus.SKIPPED - fs.bytes_uploaded = fs.file_size + job.set_file_status(fs, UploadStatus.SKIPPED) + job.set_bytes_uploaded(fs, fs.file_size) log.info( "upload", "file_upload_skipped", @@ -1237,7 +1420,7 @@ def make_upload_task( def upload_task() -> Any: if job.cancelled: with job.lock: - file_state.status = UploadStatus.CANCELLED + job.set_file_status(file_state, UploadStatus.CANCELLED) if analysis_callback: analysis_callback(job, file_state) if upload_callback: @@ -1246,7 +1429,7 @@ def upload_task() -> Any: try: with job.lock: - file_state.status = UploadStatus.UPLOADING + job.set_file_status(file_state, UploadStatus.UPLOADING) file_state.upload_started_at = datetime.now(UTC) log.info( "upload", @@ -1264,7 +1447,7 @@ def upload_task() -> Any: def byte_callback(uploaded: int, total: int) -> None: with job.lock: - file_state.bytes_uploaded = uploaded + job.set_bytes_uploaded(file_state, uploaded) if upload_callback: upload_callback(job) @@ -1280,8 +1463,8 @@ def byte_callback(uploaded: int, total: int) -> None: # Handle completion inline file_state.upload_completed_at = datetime.now(UTC) if upload_result["success"]: - file_state.status = UploadStatus.COMPLETED - file_state.bytes_uploaded = file_state.file_size + job.set_file_status(file_state, UploadStatus.COMPLETED) + job.set_bytes_uploaded(file_state, file_state.file_size) log.info( "upload", "file_upload_completed", @@ -1311,7 +1494,7 @@ def byte_callback(uploaded: int, total: int) -> None: exc_info=True, ) else: - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = upload_result.get( "error", "Unknown error" ) @@ -1328,11 +1511,11 @@ def byte_callback(uploaded: int, total: int) -> None: ) except UploadCancelledError: with job.lock: - file_state.status = UploadStatus.CANCELLED + job.set_file_status(file_state, UploadStatus.CANCELLED) file_state.upload_completed_at = datetime.now(UTC) except Exception as e: file_state.upload_completed_at = datetime.now(UTC) - file_state.status = UploadStatus.FAILED + job.set_file_status(file_state, UploadStatus.FAILED) file_state.error_message = str(e) log.error( "upload", @@ -1380,7 +1563,7 @@ def byte_callback(uploaded: int, total: int) -> None: UploadStatus.ANALYZING, UploadStatus.UPLOADING, ): - fs.status = UploadStatus.CANCELLED + job.set_file_status(fs, UploadStatus.CANCELLED) # Final job status job.completed_at = datetime.now(UTC) @@ -1396,6 +1579,9 @@ def byte_callback(uploaded: int, total: int) -> None: # Clean up temp directory self.cleanup_temp_dir(job_id) + # Mirror terminal job state to SQLite for large jobs (best-effort). + self._persist_job_terminal(job) + # Send terminal event IMMEDIATELY so the frontend unblocks. # Heavy I/O (logging, CSV, S3 sync) follows below. if upload_callback: @@ -1469,42 +1655,28 @@ def byte_callback(uploaded: int, total: int) -> None: except Exception: logger.debug("Log sync to S3 failed", exc_info=True) - def cancel_job(self, job_id: str) -> bool: - """Cancel an upload job. - - Args: - job_id: The job ID to cancel - - Returns: - True if job was found and cancelled - """ - job = self.get_job(job_id) - if not job: - return False - - job.cancelled = True - with job.lock: - for file_state in job.files: + def _on_cancel(self, job: Any) -> None: + """Upload-specific cancel logic: mark pending files cancelled, clean up temp dir.""" + upload_job = job # type: UploadJob + with upload_job.lock: + for file_state in upload_job.files: if file_state.status in ( UploadStatus.PENDING, UploadStatus.READY, UploadStatus.ANALYZING, ): - file_state.status = UploadStatus.CANCELLED + upload_job.set_file_status(file_state, UploadStatus.CANCELLED) - # Clean up temp directory when job is cancelled - self.cleanup_temp_dir(job_id) + self.cleanup_temp_dir(upload_job.job_id) log = get_log_service() log.warning( "upload", "upload_job_cancelled", - f"Upload job {job_id} cancelled", - {"job_id": job_id}, + f"Upload job {upload_job.job_id} cancelled", + {"job_id": upload_job.job_id}, ) - return True - def cleanup_temp_dir(self, job_id: str) -> bool: """Clean up temp directory for a job. @@ -1591,18 +1763,26 @@ def pre_filter_files( file_statuses.append(file_status) continue - # Try to extract timestamp from filename (fast, no file I/O) + # Try to extract timestamp (fast, no file I/O if in filename) + # Use mcap_service._extract_timestamp_from_filename utility directly? + # Or assume file_service handles it? + # file_service.extract_timestamp does I/O (stat/parse). + # We want FAST pre-filtering. + # We can use the regex utility from mcap_service for filenames. + from app.services import mcap_service + timestamp = mcap_service._extract_timestamp_from_filename(path.name) if timestamp is None: # Can't extract timestamp from filename, need full analysis + # (This is true for generic files without timestamps in names too) stats["no_timestamp"] += 1 files_to_analyze.append(file_path) file_statuses.append(file_status) continue # Generate S3 path from filename timestamp - s3_path = mcap_service.generate_s3_path(timestamp, path.name) + s3_path = file_service.generate_s3_key(path.name, timestamp) file_status["s3_path"] = s3_path # Check cache by S3 path @@ -1676,7 +1856,7 @@ def create_scan_job( Returns: The created ScanJob """ - job_id = str(uuid.uuid4()) + job_id = self._new_job_id() scan_job = ScanJob( job_id=job_id, root_folder=folder_path, @@ -1716,7 +1896,11 @@ def scan_folder_async( progress_callback: Callable[[str, dict[str, Any]], None] | None = None, cache_only: bool = False, ) -> None: - """Scan a folder asynchronously, processing subfolder by subfolder. + """Scan a folder asynchronously, emitting SSE events per subfolder as discovered. + + Uses os.walk so that ``scan_started`` fires immediately and + ``scan_folder_complete`` events stream in as each folder is processed, + giving real-time UI feedback even on first (cold-cache) scans. Args: job_id: The scan job ID @@ -1732,58 +1916,60 @@ def scan_folder_async( root = Path(scan_job.root_folder) log = get_log_service() + from app.config import get_settings + + allowed_extensions = set(get_settings().allowed_extensions) + try: # Build exclusion sets for fast lookup excluded_subs_set = set(scan_job.excluded_subfolders) excluded_files_set = set(scan_job.excluded_files) - # Phase 1: Enumerate subfolders containing .mcap files (fast, metadata only) - folder_map: dict[str, list[Path]] = {} - for mcap_path in root.rglob("*.mcap"): - if scan_job.cancelled: - break - if mcap_path.is_file(): - rel = mcap_path.relative_to(root) - parts = rel.parts - # Skip root-level files in excluded_files list - if len(parts) == 1 and parts[0] in excluded_files_set: - continue - # Skip files under excluded subfolders - if len(parts) > 1 and parts[0] in excluded_subs_set: - continue - parent = str(mcap_path.parent) - if parent not in folder_map: - folder_map[parent] = [] - folder_map[parent].append(mcap_path) - - if scan_job.cancelled: - if progress_callback: - progress_callback( - job_id, - { - "type": "scan_complete", - "status": "cancelled", - }, - ) - return - - scan_job.folders_total = len(folder_map) - + # Emit scan_started immediately so the UI knows scanning has begun. + # folders_total is 0 (unknown) at this point; the modal shows + # live stats without a percentage bar until we're done. if progress_callback: progress_callback( job_id, { "type": "scan_started", - "folders_total": scan_job.folders_total, + "folders_total": 0, "root_folder": scan_job.root_folder, }, ) - # Phase 2: Process each subfolder - for folder_path_str, mcap_paths in sorted(folder_map.items()): + # Walk the directory tree on-the-fly. topdown=True lets us prune + # excluded top-level subfolders before os.walk descends into them, + # which is more efficient than the previous post-filter approach. + for dirpath_str, dirnames, filenames in os.walk(str(root), topdown=True): if scan_job.cancelled: break + dirpath = Path(dirpath_str) + rel_dir = dirpath.relative_to(root) + + # Prune excluded top-level subfolders from traversal + if rel_dir == Path("."): + dirnames[:] = [d for d in dirnames if d not in excluded_subs_set] + + # Collect allowed files in this directory + folder_path_str = dirpath_str + mcap_paths = [] + for fname in sorted(filenames): + # Skip excluded root-level files + if rel_dir == Path(".") and fname in excluded_files_set: + continue + ext = Path(fname).suffix.lower().lstrip(".") + if ext in allowed_extensions: + mcap_paths.append(dirpath / fname) + + # Sort subdirectory traversal order for deterministic results; + # must happen before any `continue` so os.walk enters them in order. + dirnames.sort() + + if not mcap_paths: + continue # No matching files in this directory — skip + try: relative_path = str(Path(folder_path_str).relative_to(root)) if relative_path == ".": @@ -1793,17 +1979,18 @@ def scan_folder_async( file_paths: list[str] = [] files_info: list[dict[str, Any]] = [] folder_size = 0 - for mcap_path in sorted(mcap_paths, key=lambda p: p.name): - stat = mcap_path.stat() - file_paths.append(str(mcap_path)) + for file_path in sorted(mcap_paths, key=lambda p: p.name): + stat = file_path.stat() + file_paths.append(str(file_path)) folder_size += stat.st_size files_info.append( { - "path": str(mcap_path), - "filename": mcap_path.name, + "path": str(file_path), + "filename": file_path.name, "size": stat.st_size, "mtime": stat.st_mtime, - "relative_path": str(mcap_path.relative_to(root)), + "relative_path": str(file_path.relative_to(root)), + "file_category": file_service.get_file_category(file_path.name), } ) @@ -1893,6 +2080,9 @@ def scan_folder_async( scan_job.status = "cancelled" else: scan_job.status = "completed" + # folders_total was 0 at scan_started (unknown); set it to the + # actual count now that the walk is complete. + scan_job.folders_total = scan_job.folders_scanned if progress_callback: progress_callback( @@ -1937,32 +2127,6 @@ def get_active_jobs(self) -> list[UploadJob]: with self._lock: return [j for j in self.jobs.values() if j.status in active_statuses] - def cleanup_old_jobs(self, max_age_seconds: int = 3600) -> int: - """Remove completed jobs older than max_age_seconds. - - Args: - max_age_seconds: Maximum age in seconds for completed jobs - - Returns: - Number of jobs removed - """ - now = datetime.now(UTC) - removed = 0 - - with self._lock: - to_remove = [] - for job_id, job in self.jobs.items(): - if job.completed_at: - age = (now - job.completed_at).total_seconds() - if age > max_age_seconds: - to_remove.append(job_id) - - for job_id in to_remove: - del self.jobs[job_id] - removed += 1 - - return removed - # Global upload manager instance _upload_manager: UploadManager | None = None diff --git a/docs/images/about_modal.png b/docs/images/about_modal.png new file mode 100644 index 0000000..6eec431 Binary files /dev/null and b/docs/images/about_modal.png differ diff --git a/docs/images/browse_index.png b/docs/images/browse_index.png new file mode 100644 index 0000000..f865627 Binary files /dev/null and b/docs/images/browse_index.png differ diff --git a/docs/images/delete_index.png b/docs/images/delete_index.png new file mode 100644 index 0000000..77d2db1 Binary files /dev/null and b/docs/images/delete_index.png differ diff --git a/docs/images/index_upload.png b/docs/images/index_upload.png new file mode 100644 index 0000000..ea86456 Binary files /dev/null and b/docs/images/index_upload.png differ diff --git a/docs/images/large_folder_upload_index.png b/docs/images/large_folder_upload_index.png new file mode 100644 index 0000000..260e921 Binary files /dev/null and b/docs/images/large_folder_upload_index.png differ diff --git a/docs/images/logs_index.png b/docs/images/logs_index.png new file mode 100644 index 0000000..ff2dc11 Binary files /dev/null and b/docs/images/logs_index.png differ diff --git a/docs/images/settings_index.png b/docs/images/settings_index.png new file mode 100644 index 0000000..c350266 Binary files /dev/null and b/docs/images/settings_index.png differ diff --git a/docs/images/settings_index_software_update.png b/docs/images/settings_index_software_update.png new file mode 100644 index 0000000..4d71bae Binary files /dev/null and b/docs/images/settings_index_software_update.png differ diff --git a/frontend/biome.json b/frontend/biome.json new file mode 100644 index 0000000..4ef3a2c --- /dev/null +++ b/frontend/biome.json @@ -0,0 +1,54 @@ +{ + "$schema": "https://biomejs.dev/schemas/2.4.15/schema.json", + "assist": { + "enabled": true, + "actions": { + "source": { + "organizeImports": "on" + } + } + }, + "linter": { + "enabled": true, + "rules": { + "recommended": true + } + }, + "formatter": { + "enabled": true, + "indentStyle": "space", + "indentWidth": 2, + "lineWidth": 100 + }, + "css": { + "parser": { + "tailwindDirectives": true + }, + "linter": { + "enabled": true + }, + "formatter": { + "enabled": true, + "indentStyle": "space", + "indentWidth": 2 + } + }, + "javascript": { + "formatter": { + "quoteStyle": "single", + "trailingCommas": "all" + } + }, + "overrides": [ + { + "includes": ["tests/**"], + "linter": { + "rules": { + "style": { + "noNonNullAssertion": "off" + } + } + } + } + ] +} diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 249a958..98300b4 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -1,12 +1,12 @@ { "name": "frontend", - "version": "0.0.0", + "version": "1.1.0", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "frontend", - "version": "0.0.0", + "version": "1.1.0", "dependencies": { "@tanstack/react-table": "^8.21.3", "@tanstack/react-virtual": "^3.13.18", @@ -17,6 +17,7 @@ "zustand": "^5.0.11" }, "devDependencies": { + "@biomejs/biome": "^2.4.15", "@eslint/js": "^9.39.1", "@tailwindcss/vite": "^4.1.18", "@testing-library/dom": "^10.4.0", @@ -418,6 +419,181 @@ "node": ">=18" } }, + "node_modules/@biomejs/biome": { + "version": "2.4.15", + "resolved": "https://registry.npmjs.org/@biomejs/biome/-/biome-2.4.15.tgz", + "integrity": "sha512-j5VH3a/h/HXTKBM50MDMxRCzkeLv9S2XJcW2WgnZT1+xyisi+0bISrXR82gCX+8S9lvK0skEvHJRN+3Ktr2hlw==", + "dev": true, + "license": "MIT OR Apache-2.0", + "bin": { + "biome": "bin/biome" + }, + "engines": { + "node": ">=14.21.3" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/biome" + }, + "optionalDependencies": { + "@biomejs/cli-darwin-arm64": "2.4.15", + "@biomejs/cli-darwin-x64": "2.4.15", + "@biomejs/cli-linux-arm64": "2.4.15", + "@biomejs/cli-linux-arm64-musl": "2.4.15", + "@biomejs/cli-linux-x64": "2.4.15", + "@biomejs/cli-linux-x64-musl": "2.4.15", + "@biomejs/cli-win32-arm64": "2.4.15", + "@biomejs/cli-win32-x64": "2.4.15" + } + }, + "node_modules/@biomejs/cli-darwin-arm64": { + "version": "2.4.15", + "resolved": "https://registry.npmjs.org/@biomejs/cli-darwin-arm64/-/cli-darwin-arm64-2.4.15.tgz", + "integrity": "sha512-rF3PPqLq1yoST79zaQbDjVJwsuIeci/O+9bgNmC5QpgOqz6aqYuzA4abyAGx+mgyiDXn4A049xAN8gijbuR1Qg==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT OR Apache-2.0", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": ">=14.21.3" + } + }, + "node_modules/@biomejs/cli-darwin-x64": { + "version": "2.4.15", + "resolved": "https://registry.npmjs.org/@biomejs/cli-darwin-x64/-/cli-darwin-x64-2.4.15.tgz", + "integrity": "sha512-/5KHXYMfSJs1fNXiX30xFtI8JcCFV6zaVVLxOa0M2sfqBKHkpQhRTv94yxQWxeTY2lzo2OuTlNvPC+hDQt2wcQ==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT OR Apache-2.0", + "optional": true, + "os": [ + "darwin" + ], + "engines": { + "node": ">=14.21.3" + } + }, + "node_modules/@biomejs/cli-linux-arm64": { + "version": "2.4.15", + "resolved": "https://registry.npmjs.org/@biomejs/cli-linux-arm64/-/cli-linux-arm64-2.4.15.tgz", + "integrity": "sha512-owaAMZD/T4LrD0ELNCk0Km3qrRHuM0X6EAyVE1FSqGY0rbLoiDLrO4Us2tllm6cAeB2Ioa9C2C08NZPdr8+0Ug==", + "cpu": [ + "arm64" + ], + "dev": true, + "libc": [ + "glibc" + ], + "license": "MIT OR Apache-2.0", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=14.21.3" + } + }, + "node_modules/@biomejs/cli-linux-arm64-musl": { + "version": "2.4.15", + "resolved": "https://registry.npmjs.org/@biomejs/cli-linux-arm64-musl/-/cli-linux-arm64-musl-2.4.15.tgz", + "integrity": "sha512-ZPcxznxm0pogHBLZhYntyR3sR+MrZjqJIKEr7ZqVen0Rl+P/4upVmfYXjftizi9RoqZntg33fv/1fbdhbYXpEQ==", + "cpu": [ + "arm64" + ], + "dev": true, + "libc": [ + "musl" + ], + "license": "MIT OR Apache-2.0", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=14.21.3" + } + }, + "node_modules/@biomejs/cli-linux-x64": { + "version": "2.4.15", + "resolved": "https://registry.npmjs.org/@biomejs/cli-linux-x64/-/cli-linux-x64-2.4.15.tgz", + "integrity": "sha512-0jj7THz12GbUOLmMibktK6DZjqz2zV64KFxyBtcFTKPiiOIY0a7vns1elpO1dERvxpsZ5ik0oFfz0oGwFde1+g==", + "cpu": [ + "x64" + ], + "dev": true, + "libc": [ + "glibc" + ], + "license": "MIT OR Apache-2.0", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=14.21.3" + } + }, + "node_modules/@biomejs/cli-linux-x64-musl": { + "version": "2.4.15", + "resolved": "https://registry.npmjs.org/@biomejs/cli-linux-x64-musl/-/cli-linux-x64-musl-2.4.15.tgz", + "integrity": "sha512-CNq/9W38SYSH023lfcQ4KKU8K0YX8T//FZUhcgtMMRABDojx5XsMV7jlweAvGSl389wJQB29Qo6Zb/a+jdvt+w==", + "cpu": [ + "x64" + ], + "dev": true, + "libc": [ + "musl" + ], + "license": "MIT OR Apache-2.0", + "optional": true, + "os": [ + "linux" + ], + "engines": { + "node": ">=14.21.3" + } + }, + "node_modules/@biomejs/cli-win32-arm64": { + "version": "2.4.15", + "resolved": "https://registry.npmjs.org/@biomejs/cli-win32-arm64/-/cli-win32-arm64-2.4.15.tgz", + "integrity": "sha512-ouhkYdlhp/1GghEJPdWwD/Vi3gQ1nFxuSpMolWsbq3Lsq3QUR4jl6UdhhscdCugKU5vOEuMiJhvKj66O0OCq+w==", + "cpu": [ + "arm64" + ], + "dev": true, + "license": "MIT OR Apache-2.0", + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=14.21.3" + } + }, + "node_modules/@biomejs/cli-win32-x64": { + "version": "2.4.15", + "resolved": "https://registry.npmjs.org/@biomejs/cli-win32-x64/-/cli-win32-x64-2.4.15.tgz", + "integrity": "sha512-zBrGq5mx5wwpnow4+2BxUvleDM+GNd4sLbPaMapsSLQLD0NGRCquqPBTgN+7XkUteHvj7M+BstuI8tmnV7+HgQ==", + "cpu": [ + "x64" + ], + "dev": true, + "license": "MIT OR Apache-2.0", + "optional": true, + "os": [ + "win32" + ], + "engines": { + "node": ">=14.21.3" + } + }, "node_modules/@csstools/color-helpers": { "version": "6.0.1", "resolved": "https://registry.npmjs.org/@csstools/color-helpers/-/color-helpers-6.0.1.tgz", @@ -1870,6 +2046,66 @@ "node": ">=14.0.0" } }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/core": { + "version": "1.7.1", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "@emnapi/wasi-threads": "1.1.0", + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/runtime": { + "version": "1.7.1", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@emnapi/wasi-threads": { + "version": "1.1.0", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@napi-rs/wasm-runtime": { + "version": "1.1.0", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "@emnapi/core": "^1.7.1", + "@emnapi/runtime": "^1.7.1", + "@tybys/wasm-util": "^0.10.1" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/@tybys/wasm-util": { + "version": "0.10.1", + "dev": true, + "inBundle": true, + "license": "MIT", + "optional": true, + "dependencies": { + "tslib": "^2.4.0" + } + }, + "node_modules/@tailwindcss/oxide-wasm32-wasi/node_modules/tslib": { + "version": "2.8.1", + "dev": true, + "inBundle": true, + "license": "0BSD", + "optional": true + }, "node_modules/@tailwindcss/oxide-win32-arm64-msvc": { "version": "4.1.18", "resolved": "https://registry.npmjs.org/@tailwindcss/oxide-win32-arm64-msvc/-/oxide-win32-arm64-msvc-4.1.18.tgz", diff --git a/frontend/package.json b/frontend/package.json index 2686499..b3f7ed1 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -1,17 +1,23 @@ { "name": "frontend", "private": true, - "version": "1.0.0", + "version": "1.1.0", "type": "module", "scripts": { "dev": "vite", "build": "tsc -b && vite build", "preview": "vite preview", "typecheck": "tsc -b --noEmit", + "lint": "eslint src tests", + "lint:fix": "eslint src tests --fix", + "format": "biome format src tests", + "format:fix": "biome format --write src tests", + "biome:check": "biome check src tests", + "biome:fix": "biome check --write src tests", "test": "vitest run", "test:watch": "vitest", "test:coverage": "vitest run --coverage", - "check": "tsc -b --noEmit && vitest run" + "check": "tsc -b --noEmit && biome check src tests && eslint src tests && vitest run" }, "dependencies": { "@tanstack/react-table": "^8.21.3", @@ -23,6 +29,7 @@ "zustand": "^5.0.11" }, "devDependencies": { + "@biomejs/biome": "^2.4.15", "@eslint/js": "^9.39.1", "@tailwindcss/vite": "^4.1.18", "@testing-library/dom": "^10.4.0", diff --git a/frontend/public/images/alliance-for-energy-innovation-fy26-logo-black.svg b/frontend/public/images/alliance-for-energy-innovation-fy26-logo-black.svg new file mode 100644 index 0000000..bd213fb --- /dev/null +++ b/frontend/public/images/alliance-for-energy-innovation-fy26-logo-black.svg @@ -0,0 +1,44 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/frontend/public/images/nlr-logo-horizontal.svg b/frontend/public/images/nlr-logo-horizontal.svg new file mode 100755 index 0000000..4017b68 --- /dev/null +++ b/frontend/public/images/nlr-logo-horizontal.svg @@ -0,0 +1,61 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index b7ccc87..8939eb5 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1,31 +1,39 @@ -import { useEffect } from "react"; -import { Route, Routes } from "react-router-dom"; -import Layout from "./components/layout/Layout.tsx"; -import DeletePage from "./pages/DeletePage.tsx"; -import FilesPage from "./pages/FilesPage.tsx"; -import LogsPage from "./pages/LogsPage.tsx"; -import SettingsPage from "./pages/SettingsPage.tsx"; -import UploadPage from "./pages/UploadPage.tsx"; -import { useAppStore } from "./stores/appStore.ts"; +import { useEffect } from 'react'; +import { Route, Routes } from 'react-router-dom'; +import Layout from './components/layout/Layout.tsx'; +import UpdateModal from './components/settings/UpdateModal.tsx'; +import DeletePage from './pages/DeletePage.tsx'; +import FilesPage from './pages/FilesPage.tsx'; +import LargeFolderUploadPage from './pages/LargeFolderUploadPage.tsx'; +import LogsPage from './pages/LogsPage.tsx'; +import SettingsPage from './pages/SettingsPage.tsx'; +import UploadPage from './pages/UploadPage.tsx'; +import { useAppStore } from './stores/appStore.ts'; export default function App() { const loadSettings = useAppStore((s) => s.loadSettings); const loadVersion = useAppStore((s) => s.loadVersion); + const runAutoCheck = useAppStore((s) => s.runAutoCheck); useEffect(() => { loadSettings(); loadVersion(); - }, [loadSettings, loadVersion]); + runAutoCheck(); + }, [loadSettings, loadVersion, runAutoCheck]); return ( - - }> - } /> - } /> - } /> - } /> - } /> - - + <> + + }> + } /> + } /> + } /> + } /> + } /> + } /> + + + + ); } diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index f0ba620..f2a2fcd 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -4,7 +4,7 @@ export class ApiError extends Error { status: number; constructor(status: number, message: string) { super(message); - this.name = "ApiError"; + this.name = 'ApiError'; this.status = status; } } @@ -18,15 +18,15 @@ async function handleResponse(res: Response): Promise { } export async function apiGet(url: string, params?: Record): Promise { - const qs = params ? `?${new URLSearchParams(params)}` : ""; + const qs = params ? `?${new URLSearchParams(params)}` : ''; const res = await fetch(`${url}${qs}`); return handleResponse(res); } export async function apiPost(url: string, body?: unknown): Promise { const res = await fetch(url, { - method: "POST", - headers: body != null ? { "Content-Type": "application/json" } : undefined, + method: 'POST', + headers: body != null ? { 'Content-Type': 'application/json' } : undefined, body: body != null ? JSON.stringify(body) : undefined, }); return handleResponse(res); @@ -34,8 +34,8 @@ export async function apiPost(url: string, body?: unknown): Promise { export async function apiPut(url: string, body: unknown): Promise { const res = await fetch(url, { - method: "PUT", - headers: { "Content-Type": "application/json" }, + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, body: JSON.stringify(body), }); return handleResponse(res); diff --git a/frontend/src/api/largeFolderUpload.ts b/frontend/src/api/largeFolderUpload.ts new file mode 100644 index 0000000..4d41c95 --- /dev/null +++ b/frontend/src/api/largeFolderUpload.ts @@ -0,0 +1,23 @@ +/** API helpers for the Large Folder Upload feature. */ + +import { apiPost } from './client.ts'; + +export interface StartSyncResponse { + job_id: string; + s3_uri: string; + cmd: string; +} + +export async function startLargeFolderSync( + folderPath: string, + s3Prefix: string, +): Promise { + return apiPost('/api/large-folder-upload/start', { + folder_path: folderPath, + s3_prefix: s3Prefix, + }); +} + +export async function cancelLargeFolderSync(jobId: string): Promise { + await apiPost(`/api/large-folder-upload/cancel/${jobId}`); +} diff --git a/frontend/src/components/common/AlertBanner.tsx b/frontend/src/components/common/AlertBanner.tsx index 5b802c4..95573a7 100644 --- a/frontend/src/components/common/AlertBanner.tsx +++ b/frontend/src/components/common/AlertBanner.tsx @@ -2,10 +2,10 @@ * Reusable alert banner for displaying information, warnings, errors, and success messages. */ -import { InfoIcon, WarningIcon, ErrorIcon, SuccessIcon, ShieldIcon } from "../../utils/icons.tsx"; -import type { ReactNode } from "react"; +import type { ReactNode } from 'react'; +import { ErrorIcon, InfoIcon, ShieldIcon, SuccessIcon, WarningIcon } from '../../utils/icons.tsx'; -type AlertType = "info" | "warning" | "error" | "success" | "shield"; +type AlertType = 'info' | 'warning' | 'error' | 'success' | 'shield'; interface AlertBannerProps { type: AlertType; @@ -16,27 +16,27 @@ interface AlertBannerProps { } const alertStyles: Record = { - info: "bg-blue-50 border-blue-200", - warning: "bg-yellow-50 border-yellow-200", - error: "bg-red-50 border-red-200", - success: "bg-green-50 border-green-200", - shield: "bg-green-50 border-green-200", + info: 'bg-blue-50 border-blue-200', + warning: 'bg-yellow-50 border-yellow-200', + error: 'bg-red-50 border-red-200', + success: 'bg-green-50 border-green-200', + shield: 'bg-green-50 border-green-200', }; const titleStyles: Record = { - info: "text-blue-800", - warning: "text-yellow-800", - error: "text-red-800", - success: "text-green-800", - shield: "text-green-800", + info: 'text-blue-800', + warning: 'text-yellow-800', + error: 'text-red-800', + success: 'text-green-800', + shield: 'text-green-800', }; const messageStyles: Record = { - info: "text-blue-700", - warning: "text-yellow-700", - error: "text-red-700", - success: "text-green-700", - shield: "text-green-700", + info: 'text-blue-700', + warning: 'text-yellow-700', + error: 'text-red-700', + success: 'text-green-700', + shield: 'text-green-700', }; const iconMap: Record = { @@ -48,11 +48,11 @@ const iconMap: Record = { }; const iconColorStyles: Record = { - info: "text-blue-500", - warning: "text-yellow-500", - error: "text-red-500", - success: "text-green-700", - shield: "text-green-700", + info: 'text-blue-500', + warning: 'text-yellow-500', + error: 'text-red-500', + success: 'text-green-700', + shield: 'text-green-700', }; export default function AlertBanner({ @@ -60,7 +60,7 @@ export default function AlertBanner({ title, message, icon, - className = "", + className = '', }: AlertBannerProps) { const Icon = iconMap[type]; @@ -70,7 +70,7 @@ export default function AlertBanner({ {icon || }
{title &&

{title}

} -
{message}
+
{message}
diff --git a/frontend/src/components/common/Breadcrumb.tsx b/frontend/src/components/common/Breadcrumb.tsx index 9e408aa..bbe4a42 100644 --- a/frontend/src/components/common/Breadcrumb.tsx +++ b/frontend/src/components/common/Breadcrumb.tsx @@ -1,5 +1,5 @@ -import { useEffect, useRef } from "react"; -import { ChevronRightIcon } from "../../utils/icons.tsx"; +import { useEffect, useRef } from 'react'; +import { ChevronRightIcon } from '../../utils/icons.tsx'; interface BreadcrumbItem { label: string; @@ -19,7 +19,7 @@ export default function Breadcrumb({ items }: BreadcrumbProps) { if (el) { el.scrollLeft = el.scrollWidth; } - }, [items]); + }, []); return (