diff --git a/app/api/files.py b/app/api/files.py index 94106db5..0b7ce49c 100644 --- a/app/api/files.py +++ b/app/api/files.py @@ -412,6 +412,110 @@ def reprocess_single_file(request: Request, file_id: int, db: Session = Depends( raise HTTPException(status_code=500, detail=f"Error reprocessing file: {str(e)}") +@router.post("/files/{file_id}/retry-subtask") +@require_login +def retry_subtask( + request: Request, + file_id: int, + subtask_name: str = Query(..., description="Name of the upload subtask to retry (e.g., 'upload_to_dropbox')"), + db: Session = Depends(get_db) +): + """ + Retry a specific failed upload subtask for a file. + + Args: + file_id: ID of the file + subtask_name: Name of the upload task (e.g., upload_to_dropbox, upload_to_s3) + + Returns: + Task ID and status information + """ + try: + # Find the file record + file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first() + + if not file_record: + raise HTTPException(status_code=404, detail=f"File with ID {file_id} not found") + + # Check for processed file (upload tasks work with processed files) + workdir = settings.workdir + processed_dir = os.path.join(workdir, "processed") + + # Try to find the processed file + base_filename = os.path.splitext(file_record.original_filename)[0] + potential_paths = [ + os.path.join(processed_dir, f"{file_record.filehash}.pdf"), + os.path.join(processed_dir, f"{base_filename}_processed.pdf"), + os.path.join(processed_dir, file_record.original_filename), + ] + + file_path = None + for path in potential_paths: + if os.path.exists(path): + file_path = path + break + + if not file_path: + raise HTTPException( + status_code=400, + detail="Processed file not found. Cannot retry upload." + ) + + # Map subtask names to their corresponding Celery tasks + from app.tasks.upload_to_dropbox import upload_to_dropbox + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + from app.tasks.upload_to_paperless import upload_to_paperless + from app.tasks.upload_to_google_drive import upload_to_google_drive + from app.tasks.upload_to_onedrive import upload_to_onedrive + from app.tasks.upload_to_s3 import upload_to_s3 + from app.tasks.upload_to_webdav import upload_to_webdav + from app.tasks.upload_to_ftp import upload_to_ftp + from app.tasks.upload_to_sftp import upload_to_sftp + from app.tasks.upload_to_email import upload_to_email + + task_map = { + "upload_to_dropbox": upload_to_dropbox, + "upload_to_nextcloud": upload_to_nextcloud, + "upload_to_paperless": upload_to_paperless, + "upload_to_google_drive": upload_to_google_drive, + "upload_to_onedrive": upload_to_onedrive, + "upload_to_s3": upload_to_s3, + "upload_to_webdav": upload_to_webdav, + "upload_to_ftp": upload_to_ftp, + "upload_to_sftp": upload_to_sftp, + "upload_to_email": upload_to_email + } + + if subtask_name not in task_map: + raise HTTPException( + status_code=400, + detail=f"Invalid subtask name: {subtask_name}. Must be one of: {', '.join(task_map.keys())}" + ) + + # Queue the specific upload task + upload_task = task_map[subtask_name] + task = upload_task.delay(file_path, file_id) + + logger.info( + f"Retrying upload subtask: FileID={file_record.id}, " + f"Subtask={subtask_name}, TaskID={task.id}" + ) + + return { + "status": "success", + "message": f"Upload task {subtask_name} queued for retry", + "file_id": file_record.id, + "subtask_name": subtask_name, + "task_id": task.id + } + + except HTTPException: + raise + except Exception as e: + logger.exception(f"Error retrying subtask {subtask_name} for file {file_id}: {str(e)}") + raise HTTPException(status_code=500, detail=f"Error retrying subtask: {str(e)}") + + @router.get("/files/{file_id}/preview") @require_login def get_file_preview(request: Request, file_id: int, version: str = Query("original", description="original or processed"), db: Session = Depends(get_db)): diff --git a/app/views/files.py b/app/views/files.py index 10298c81..5bc059d0 100644 --- a/app/views/files.py +++ b/app/views/files.py @@ -200,13 +200,17 @@ def file_detail_page(request: Request, file_id: int, db: Session = Depends(get_d # Compute processing flow for visualization flow_data = _compute_processing_flow(logs) + # Compute step-aligned summary + step_summary = _compute_step_summary(logs) + return templates.TemplateResponse("file_detail.html", { "request": request, "file": file_record, "logs": logs, "file_exists": file_exists, "processed_exists": processed_exists, - "flow_data": flow_data + "flow_data": flow_data, + "step_summary": step_summary }) except Exception as e: logger.error(f"Error retrieving file details: {str(e)}") @@ -221,8 +225,9 @@ def _compute_processing_flow(logs): Compute the processing flow structure from logs for visualization. Returns a structured representation of the processing pipeline with branches. + Detects upload sub-tasks and organizes them as branches under the parent upload stage. """ - # Define the processing stages and their relationships + # Define the main processing stages stages = { "hash_file": {"label": "File Upload & Hash", "next": ["create_file_record"]}, "create_file_record": {"label": "Create File Record", "next": ["check_text"]}, @@ -231,22 +236,63 @@ def _compute_processing_flow(logs): "process_with_azure_document_intelligence": {"label": "OCR Processing (Azure)", "next": ["extract_metadata_with_gpt"]}, "extract_metadata_with_gpt": {"label": "Extract Metadata (GPT)", "next": ["embed_metadata_into_pdf"]}, "embed_metadata_into_pdf": {"label": "Embed Metadata into PDF", "next": ["finalize_document_storage"]}, - "finalize_document_storage": {"label": "Finalize & Queue Distribution", "next": ["upload_destinations"]}, - "upload_destinations": {"label": "Upload to Destinations", "next": []} + "finalize_document_storage": {"label": "Finalize & Queue Distribution", "next": ["send_to_all_destinations"]}, + "send_to_all_destinations": {"label": "Upload to Destinations", "next": [], "has_branches": True} + } + + # Define upload sub-tasks (branches) + upload_tasks = { + "upload_to_dropbox": "Dropbox", + "upload_to_nextcloud": "Nextcloud", + "upload_to_paperless": "Paperless-ngx", + "upload_to_google_drive": "Google Drive", + "upload_to_onedrive": "OneDrive", + "upload_to_s3": "S3 Storage", + "upload_to_webdav": "WebDAV", + "upload_to_ftp": "FTP Storage", + "upload_to_sftp": "SFTP Storage", + "upload_to_email": "Email", + "queue_dropbox": "Dropbox", + "queue_nextcloud": "Nextcloud", + "queue_paperless": "Paperless-ngx", + "queue_google_drive": "Google Drive", + "queue_onedrive": "OneDrive", + "queue_s3": "S3 Storage", + "queue_webdav": "WebDAV", + "queue_ftp": "FTP Storage", + "queue_sftp": "SFTP Storage", + "queue_email": "Email" } # Create a map of step names to their log entries step_map = {} + upload_branches = {} + for log in logs: step_name = log.step_name - if step_name not in step_map: - step_map[step_name] = [] - step_map[step_name].append({ - "status": log.status, - "message": log.message, - "timestamp": log.timestamp, - "task_id": log.task_id - }) + + # Check if this is an upload sub-task + if step_name in upload_tasks: + # Extract the actual upload task name (remove queue_ prefix if present) + upload_key = step_name.replace("queue_", "upload_to_") + if upload_key not in upload_branches: + upload_branches[upload_key] = [] + upload_branches[upload_key].append({ + "status": log.status, + "message": log.message, + "timestamp": log.timestamp, + "task_id": log.task_id + }) + else: + # Regular processing step + if step_name not in step_map: + step_map[step_name] = [] + step_map[step_name].append({ + "status": log.status, + "message": log.message, + "timestamp": log.timestamp, + "task_id": log.task_id + }) # Build the flow structure flow = [] @@ -266,14 +312,90 @@ def _compute_processing_flow(logs): timestamp = None task_id = None - flow.append({ + stage_data = { "key": stage_key, "label": stage_info["label"], "status": status, "message": message, "timestamp": timestamp, "task_id": task_id, - "can_retry": status == "failure" - }) + "can_retry": status == "failure", + "is_branch_parent": stage_info.get("has_branches", False) + } + + # If this is the upload stage, add branches + if stage_info.get("has_branches") and upload_branches: + branches = [] + for upload_key, upload_logs in upload_branches.items(): + latest_upload = upload_logs[-1] + upload_name = upload_tasks.get(upload_key, upload_key.replace("upload_to_", "").title()) + + branches.append({ + "key": upload_key, + "label": upload_name, + "status": latest_upload["status"], + "message": latest_upload["message"], + "timestamp": latest_upload["timestamp"], + "task_id": latest_upload["task_id"], + "can_retry": latest_upload["status"] == "failure" + }) + stage_data["branches"] = branches + + flow.append(stage_data) return flow + + +def _compute_step_summary(logs): + """ + Compute a step-aligned summary from logs showing queued, success, and failure counts. + + Returns a dictionary with main step counts and upload branch counts. + """ + # Count statuses for main processing steps (not uploads) + main_steps = [ + "hash_file", "create_file_record", "check_text", "extract_text", + "process_with_azure_document_intelligence", "extract_metadata_with_gpt", + "embed_metadata_into_pdf", "finalize_document_storage", "send_to_all_destinations" + ] + + upload_prefixes = ["upload_to_", "queue_"] + + main_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0} + upload_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0} + + # Track which steps we've seen + main_steps_seen = set() + upload_tasks_seen = {} + + for log in logs: + step_name = log.step_name + status = log.status.lower() + + # Normalize status + if status == "pending": + status = "queued" + + # Check if it's an upload task + is_upload = any(step_name.startswith(prefix) for prefix in upload_prefixes) + + if is_upload: + # Track latest status for each unique upload task + upload_tasks_seen[step_name] = status + elif step_name in main_steps: + # Track latest status for main steps + main_steps_seen.add(step_name) + if status in main_counts: + main_counts[status] += 1 + + # Count upload task statuses + for task_status in upload_tasks_seen.values(): + if task_status in upload_counts: + upload_counts[task_status] += 1 + + return { + "main": main_counts, + "uploads": upload_counts, + "total_main_steps": len(main_steps_seen), + "total_upload_tasks": len(upload_tasks_seen) + } diff --git a/frontend/templates/file_detail.html b/frontend/templates/file_detail.html index 00aadf2f..138f2595 100644 --- a/frontend/templates/file_detail.html +++ b/frontend/templates/file_detail.html @@ -85,7 +85,71 @@ color: #991B1B; } - /* Processing logs */ + /* Step summary section */ + .step-summary { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); + gap: 1rem; + margin-bottom: 1.5rem; + } + .summary-card { + background-color: #f7fafc; + border-radius: 0.5rem; + padding: 1rem; + border-left: 4px solid #e2e8f0; + } + .summary-card.main-steps { + border-left-color: #4299e1; + } + .summary-card.upload-steps { + border-left-color: #48bb78; + } + .summary-title { + font-weight: 600; + color: #2d3748; + margin-bottom: 0.5rem; + font-size: 0.875rem; + text-transform: uppercase; + } + .summary-counts { + display: flex; + gap: 1rem; + flex-wrap: wrap; + } + .summary-count { + display: flex; + align-items: center; + gap: 0.25rem; + font-size: 0.875rem; + } + .count-badge { + display: inline-flex; + align-items: center; + justify-content: center; + width: 1.5rem; + height: 1.5rem; + border-radius: 50%; + font-weight: 600; + font-size: 0.75rem; + } + .count-badge.success { + background-color: #48bb78; + color: white; + } + .count-badge.failure { + background-color: #f56565; + color: white; + } + .count-badge.in-progress { + background-color: #4299e1; + color: white; + } + .count-badge.queued { + background-color: #ecc94b; + color: white; + } + + /* Processing logs - collapsible */ .timeline { position: relative; padding-left: 2rem; @@ -170,6 +234,29 @@ color: #a0aec0; } + .logs-toggle { + cursor: pointer; + user-select: none; + display: inline-flex; + align-items: center; + gap: 0.5rem; + color: #3182ce; + font-weight: 600; + margin-bottom: 1rem; + } + .logs-toggle:hover { + color: #2c5aa0; + } + .logs-content { + max-height: 0; + overflow: hidden; + transition: max-height 0.3s ease-out; + } + .logs-content.expanded { + max-height: 5000px; + transition: max-height 0.5s ease-in; + } + .no-logs { text-align: center; padding: 3rem; @@ -206,6 +293,178 @@ background-color: #FEE2E2; color: #991B1B; } + + /* Flow visualization with branches */ + .flow-stage { + display: flex; + align-items: flex-start; + margin-bottom: 1.5rem; + } + .flow-indicator { + width: 40px; + height: 40px; + border-radius: 50%; + display: flex; + align-items: center; + justify-content: center; + margin-right: 1rem; + flex-shrink: 0; + } + .flow-indicator.success { + background-color: #48bb78; + color: white; + } + .flow-indicator.failure { + background-color: #f56565; + color: white; + } + .flow-indicator.in_progress { + background-color: #4299e1; + color: white; + } + .flow-indicator.pending, .flow-indicator.queued { + background-color: #ecc94b; + color: white; + } + .flow-indicator.not_run { + background-color: #e2e8f0; + color: #718096; + } + .flow-content { + flex: 1; + background-color: #f7fafc; + padding: 1rem; + border-radius: 0.5rem; + border-left: 3px solid #e2e8f0; + } + .flow-content.success { + border-left-color: #48bb78; + background-color: #f0fff4; + } + .flow-content.failure { + border-left-color: #f56565; + background-color: #fff5f5; + } + .flow-content.in_progress { + border-left-color: #4299e1; + background-color: #ebf8ff; + } + .flow-content.pending, .flow-content.queued { + border-left-color: #ecc94b; + background-color: #fffff0; + } + .flow-title { + font-weight: 600; + color: #2d3748; + margin-bottom: 0.25rem; + } + .flow-message { + color: #4a5568; + font-size: 0.875rem; + } + .flow-timestamp { + color: #718096; + font-size: 0.75rem; + margin-top: 0.25rem; + } + .flow-connector { + width: 2px; + height: 20px; + background-color: #e2e8f0; + margin-left: 19px; + margin-bottom: 0.5rem; + } + + /* Branch visualization */ + .flow-branches { + margin-left: 56px; + margin-top: 1rem; + padding-left: 1rem; + border-left: 2px dashed #cbd5e0; + } + .branch-item { + display: flex; + align-items: flex-start; + margin-bottom: 1rem; + position: relative; + } + .branch-item::before { + content: ''; + position: absolute; + left: -1rem; + top: 20px; + width: 1rem; + height: 2px; + background-color: #cbd5e0; + } + .branch-indicator { + width: 30px; + height: 30px; + border-radius: 50%; + display: flex; + align-items: center; + justify-content: center; + margin-right: 0.75rem; + flex-shrink: 0; + font-size: 0.875rem; + } + .branch-indicator.success { + background-color: #48bb78; + color: white; + } + .branch-indicator.failure { + background-color: #f56565; + color: white; + } + .branch-indicator.in_progress { + background-color: #4299e1; + color: white; + } + .branch-indicator.pending, .branch-indicator.queued { + background-color: #ecc94b; + color: white; + } + .branch-content { + flex: 1; + background-color: white; + padding: 0.75rem; + border-radius: 0.5rem; + border: 1px solid #e2e8f0; + } + .branch-content.failure { + border-color: #f56565; + background-color: #fffafa; + } + .branch-title { + font-weight: 600; + color: #2d3748; + font-size: 0.875rem; + margin-bottom: 0.25rem; + display: flex; + justify-content: space-between; + align-items: center; + } + .branch-message { + color: #4a5568; + font-size: 0.75rem; + } + .retry-btn { + background-color: #f56565; + color: white; + border: none; + padding: 0.25rem 0.75rem; + border-radius: 0.25rem; + font-size: 0.75rem; + cursor: pointer; + font-weight: 600; + } + .retry-btn:hover { + background-color: #e53e3e; + } + .retry-btn:disabled { + background-color: #cbd5e0; + cursor: not-allowed; + } {% endblock %} @@ -322,7 +654,77 @@ - + + {% if step_summary %} +