Add branching visualization, per-subtask retry, and step summary to file detail view

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
2026-02-07 22:08:09 +00:00
parent d71cd5034e
commit 910fe86603
3 changed files with 734 additions and 75 deletions
@@ -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)):
@@ -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)
+    }