style: Apply Black formatting to modified files
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
+52
-57
@@ -368,43 +368,39 @@ def bulk_reprocess_files(request: Request, file_ids: List[int], db: Session = De
|
||||
def reprocess_single_file(request: Request, file_id: int, db: Session = Depends(get_db)):
|
||||
"""
|
||||
Reprocess a single file by queuing it for processing again.
|
||||
|
||||
|
||||
Args:
|
||||
file_id: ID of the file to reprocess
|
||||
|
||||
|
||||
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 if local file exists
|
||||
if not file_record.local_filename or not os.path.exists(file_record.local_filename):
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Local file not found on disk. Cannot reprocess."
|
||||
)
|
||||
|
||||
raise HTTPException(status_code=400, detail="Local file not found on disk. Cannot reprocess.")
|
||||
|
||||
# Queue the file for processing
|
||||
task = process_document.delay(file_record.local_filename, original_filename=file_record.original_filename)
|
||||
|
||||
|
||||
logger.info(
|
||||
f"Reprocessing file: ID={file_record.id}, "
|
||||
f"Filename={file_record.original_filename}, TaskID={task.id}"
|
||||
f"Reprocessing file: ID={file_record.id}, " f"Filename={file_record.original_filename}, TaskID={task.id}"
|
||||
)
|
||||
|
||||
|
||||
return {
|
||||
"status": "success",
|
||||
"message": "File queued for reprocessing",
|
||||
"file_id": file_record.id,
|
||||
"filename": file_record.original_filename,
|
||||
"task_id": task.id
|
||||
"task_id": task.id,
|
||||
}
|
||||
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
@@ -415,28 +411,28 @@ def reprocess_single_file(request: Request, file_id: int, db: Session = Depends(
|
||||
@router.post("/files/{file_id}/retry-subtask")
|
||||
@require_login
|
||||
def retry_subtask(
|
||||
request: Request,
|
||||
file_id: int,
|
||||
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)
|
||||
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")
|
||||
|
||||
|
||||
# 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
|
||||
@@ -448,7 +444,7 @@ def retry_subtask(
|
||||
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,
|
||||
@@ -459,19 +455,19 @@ def retry_subtask(
|
||||
"upload_to_webdav": upload_to_webdav,
|
||||
"upload_to_ftp": upload_to_ftp,
|
||||
"upload_to_sftp": upload_to_sftp,
|
||||
"upload_to_email": upload_to_email
|
||||
"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())}"
|
||||
detail=f"Invalid subtask name: {subtask_name}. Must be one of: {', '.join(task_map.keys())}",
|
||||
)
|
||||
|
||||
|
||||
# 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 = [
|
||||
@@ -479,36 +475,30 @@ def retry_subtask(
|
||||
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."
|
||||
)
|
||||
|
||||
raise HTTPException(status_code=400, detail="Processed file not found. Cannot retry upload.")
|
||||
|
||||
# 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}"
|
||||
)
|
||||
|
||||
|
||||
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
|
||||
"task_id": task.id,
|
||||
}
|
||||
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
@@ -518,38 +508,43 @@ def retry_subtask(
|
||||
|
||||
@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)):
|
||||
def get_file_preview(
|
||||
request: Request,
|
||||
file_id: int,
|
||||
version: str = Query("original", description="original or processed"),
|
||||
db: Session = Depends(get_db),
|
||||
):
|
||||
"""
|
||||
Get file content for preview (original or processed version).
|
||||
|
||||
|
||||
Args:
|
||||
file_id: ID of the file
|
||||
version: "original" for tmp file, "processed" for processed file
|
||||
|
||||
|
||||
Returns:
|
||||
File content for preview
|
||||
"""
|
||||
from fastapi.responses import FileResponse
|
||||
|
||||
|
||||
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")
|
||||
|
||||
|
||||
if version == "original":
|
||||
# Return the original file from tmp
|
||||
if not file_record.local_filename or not os.path.exists(file_record.local_filename):
|
||||
raise HTTPException(status_code=404, detail="Original file not found on disk")
|
||||
|
||||
|
||||
file_path = file_record.local_filename
|
||||
|
||||
|
||||
elif version == "processed":
|
||||
# Look for processed file in /workdir/processed/
|
||||
workdir = settings.workdir
|
||||
processed_dir = os.path.join(workdir, "processed")
|
||||
|
||||
|
||||
# Try to find the processed file (same hash or UUID-based naming)
|
||||
base_filename = os.path.splitext(file_record.original_filename)[0]
|
||||
potential_paths = [
|
||||
@@ -557,25 +552,25 @@ def get_file_preview(request: Request, file_id: int, version: str = Query("origi
|
||||
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=404, detail="Processed file not found")
|
||||
else:
|
||||
raise HTTPException(status_code=400, detail="Invalid version parameter. Use 'original' or 'processed'")
|
||||
|
||||
|
||||
# Return the file
|
||||
return FileResponse(
|
||||
path=file_path,
|
||||
media_type=file_record.mime_type or "application/pdf",
|
||||
filename=file_record.original_filename
|
||||
filename=file_record.original_filename,
|
||||
)
|
||||
|
||||
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
|
||||
+150
-144
@@ -1,6 +1,7 @@
|
||||
"""
|
||||
File management views for displaying and managing files.
|
||||
"""
|
||||
|
||||
from fastapi import Request, Depends, Query
|
||||
from sqlalchemy.orm import Session
|
||||
from typing import Optional
|
||||
@@ -11,6 +12,7 @@ from app.config import settings
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/files")
|
||||
@require_login
|
||||
def files_page(
|
||||
@@ -22,7 +24,7 @@ def files_page(
|
||||
sort_order: str = Query("desc"),
|
||||
search: Optional[str] = Query(None),
|
||||
mime_type: Optional[str] = Query(None),
|
||||
status: Optional[str] = Query(None)
|
||||
status: Optional[str] = Query(None),
|
||||
):
|
||||
"""
|
||||
Return the 'files.html' template with server-side pagination, sorting, and filtering
|
||||
@@ -31,18 +33,18 @@ def files_page(
|
||||
# Import the model here to avoid circular imports
|
||||
from app.models import FileRecord, ProcessingLog
|
||||
from sqlalchemy import desc, asc, or_
|
||||
|
||||
|
||||
# Start with base query
|
||||
query = db.query(FileRecord)
|
||||
|
||||
|
||||
# Apply search filter
|
||||
if search:
|
||||
query = query.filter(FileRecord.original_filename.ilike(f"%{search}%"))
|
||||
|
||||
|
||||
# Apply MIME type filter
|
||||
if mime_type:
|
||||
query = query.filter(FileRecord.mime_type == mime_type)
|
||||
|
||||
|
||||
# Apply status filter (before pagination for correct counts)
|
||||
if status:
|
||||
# Subquery to get file IDs matching the status
|
||||
@@ -52,106 +54,102 @@ def files_page(
|
||||
query = query.filter(~FileRecord.id.in_(subq))
|
||||
elif status == "processing":
|
||||
# Files with in_progress logs
|
||||
subq = db.query(ProcessingLog.file_id).filter(
|
||||
ProcessingLog.status == "in_progress"
|
||||
).distinct()
|
||||
subq = db.query(ProcessingLog.file_id).filter(ProcessingLog.status == "in_progress").distinct()
|
||||
query = query.filter(FileRecord.id.in_(subq))
|
||||
elif status == "failed":
|
||||
# Files with failure logs
|
||||
subq = db.query(ProcessingLog.file_id).filter(
|
||||
ProcessingLog.status == "failure"
|
||||
).distinct()
|
||||
subq = db.query(ProcessingLog.file_id).filter(ProcessingLog.status == "failure").distinct()
|
||||
query = query.filter(FileRecord.id.in_(subq))
|
||||
elif status == "completed":
|
||||
# Files with success logs but no failures or in_progress
|
||||
success_files = db.query(ProcessingLog.file_id).filter(
|
||||
ProcessingLog.status == "success"
|
||||
).distinct().subquery()
|
||||
|
||||
failed_files = db.query(ProcessingLog.file_id).filter(
|
||||
or_(ProcessingLog.status == "failure", ProcessingLog.status == "in_progress")
|
||||
).distinct().subquery()
|
||||
|
||||
query = query.filter(
|
||||
FileRecord.id.in_(db.query(success_files.c.file_id))
|
||||
).filter(
|
||||
success_files = (
|
||||
db.query(ProcessingLog.file_id).filter(ProcessingLog.status == "success").distinct().subquery()
|
||||
)
|
||||
|
||||
failed_files = (
|
||||
db.query(ProcessingLog.file_id)
|
||||
.filter(or_(ProcessingLog.status == "failure", ProcessingLog.status == "in_progress"))
|
||||
.distinct()
|
||||
.subquery()
|
||||
)
|
||||
|
||||
query = query.filter(FileRecord.id.in_(db.query(success_files.c.file_id))).filter(
|
||||
~FileRecord.id.in_(db.query(failed_files.c.file_id))
|
||||
)
|
||||
|
||||
|
||||
# Get total count before pagination
|
||||
total_items = query.count()
|
||||
|
||||
|
||||
# Apply sorting
|
||||
sort_column = {
|
||||
"id": FileRecord.id,
|
||||
"original_filename": FileRecord.original_filename,
|
||||
"file_size": FileRecord.file_size,
|
||||
"mime_type": FileRecord.mime_type,
|
||||
"created_at": FileRecord.created_at
|
||||
"created_at": FileRecord.created_at,
|
||||
}.get(sort_by, FileRecord.created_at)
|
||||
|
||||
|
||||
if sort_order == "asc":
|
||||
query = query.order_by(asc(sort_column))
|
||||
else:
|
||||
query = query.order_by(desc(sort_column))
|
||||
|
||||
|
||||
# Apply pagination
|
||||
offset = (page - 1) * per_page
|
||||
files = query.offset(offset).limit(per_page).all()
|
||||
|
||||
|
||||
# Get processing status for all files efficiently (avoids N+1)
|
||||
file_ids = [f.id for f in files]
|
||||
statuses = get_files_processing_status(db, file_ids)
|
||||
|
||||
|
||||
# Add status to each file
|
||||
files_with_status = []
|
||||
for file in files:
|
||||
file.processing_status = statuses.get(file.id, {}).get("status", "pending")
|
||||
files_with_status.append(file)
|
||||
|
||||
|
||||
# Calculate pagination info
|
||||
total_pages = (total_items + per_page - 1) // per_page
|
||||
|
||||
|
||||
# Get unique MIME types for filter dropdown
|
||||
mime_types = db.query(FileRecord.mime_type).distinct().filter(
|
||||
FileRecord.mime_type.isnot(None)
|
||||
).all()
|
||||
mime_types = db.query(FileRecord.mime_type).distinct().filter(FileRecord.mime_type.isnot(None)).all()
|
||||
mime_types = [mt[0] for mt in mime_types if mt[0]]
|
||||
|
||||
|
||||
# Debug output
|
||||
logger.info(f"Retrieved {len(files_with_status)} files from database (page {page}/{total_pages})")
|
||||
|
||||
return templates.TemplateResponse("files.html", {
|
||||
"request": request,
|
||||
"files": files_with_status,
|
||||
"pagination": {
|
||||
"page": page,
|
||||
"per_page": per_page,
|
||||
"total_items": total_items,
|
||||
"total_pages": total_pages
|
||||
|
||||
return templates.TemplateResponse(
|
||||
"files.html",
|
||||
{
|
||||
"request": request,
|
||||
"files": files_with_status,
|
||||
"pagination": {
|
||||
"page": page,
|
||||
"per_page": per_page,
|
||||
"total_items": total_items,
|
||||
"total_pages": total_pages,
|
||||
},
|
||||
"sort_by": sort_by,
|
||||
"sort_order": sort_order,
|
||||
"search": search or "",
|
||||
"mime_type": mime_type or "",
|
||||
"status": status or "",
|
||||
"mime_types": mime_types,
|
||||
},
|
||||
"sort_by": sort_by,
|
||||
"sort_order": sort_order,
|
||||
"search": search or "",
|
||||
"mime_type": mime_type or "",
|
||||
"status": status or "",
|
||||
"mime_types": mime_types
|
||||
})
|
||||
)
|
||||
except Exception as e:
|
||||
# Log any errors
|
||||
logger.error(f"Error retrieving files: {str(e)}")
|
||||
# Return error message to template
|
||||
return templates.TemplateResponse("files.html", {
|
||||
"request": request,
|
||||
"files": [],
|
||||
"pagination": {
|
||||
"page": 1,
|
||||
"per_page": per_page,
|
||||
"total_items": 0,
|
||||
"total_pages": 0
|
||||
return templates.TemplateResponse(
|
||||
"files.html",
|
||||
{
|
||||
"request": request,
|
||||
"files": [],
|
||||
"pagination": {"page": 1, "per_page": per_page, "total_items": 0, "total_pages": 0},
|
||||
"error": str(e),
|
||||
},
|
||||
"error": str(e)
|
||||
})
|
||||
)
|
||||
|
||||
|
||||
@router.get("/files/{file_id}/detail")
|
||||
@@ -163,25 +161,26 @@ def file_detail_page(request: Request, file_id: int, db: Session = Depends(get_d
|
||||
try:
|
||||
from app.models import FileRecord, ProcessingLog
|
||||
import os
|
||||
|
||||
|
||||
# Find the file record
|
||||
file_record = db.query(FileRecord).filter(FileRecord.id == file_id).first()
|
||||
|
||||
|
||||
if not file_record:
|
||||
return templates.TemplateResponse("file_detail.html", {
|
||||
"request": request,
|
||||
"file": None,
|
||||
"error": f"File with ID {file_id} not found"
|
||||
})
|
||||
|
||||
return templates.TemplateResponse(
|
||||
"file_detail.html", {"request": request, "file": None, "error": f"File with ID {file_id} not found"}
|
||||
)
|
||||
|
||||
# Get processing logs
|
||||
logs = db.query(ProcessingLog).filter(
|
||||
ProcessingLog.file_id == file_id
|
||||
).order_by(ProcessingLog.timestamp.asc()).all()
|
||||
|
||||
logs = (
|
||||
db.query(ProcessingLog)
|
||||
.filter(ProcessingLog.file_id == file_id)
|
||||
.order_by(ProcessingLog.timestamp.asc())
|
||||
.all()
|
||||
)
|
||||
|
||||
# Check if file exists on disk
|
||||
file_exists = os.path.exists(file_record.local_filename) if file_record.local_filename else False
|
||||
|
||||
|
||||
# Check if processed file exists
|
||||
processed_exists = False
|
||||
workdir = settings.workdir
|
||||
@@ -197,35 +196,34 @@ def file_detail_page(request: Request, file_id: int, db: Session = Depends(get_d
|
||||
if os.path.exists(path):
|
||||
processed_exists = True
|
||||
break
|
||||
|
||||
|
||||
# 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,
|
||||
"step_summary": step_summary
|
||||
})
|
||||
|
||||
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,
|
||||
"step_summary": step_summary,
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error retrieving file details: {str(e)}")
|
||||
return templates.TemplateResponse("file_detail.html", {
|
||||
"request": request,
|
||||
"file": None,
|
||||
"error": str(e)
|
||||
})
|
||||
return templates.TemplateResponse("file_detail.html", {"request": request, "file": None, "error": str(e)})
|
||||
|
||||
|
||||
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.
|
||||
"""
|
||||
@@ -233,19 +231,25 @@ def _compute_processing_flow(logs):
|
||||
stages = {
|
||||
"hash_file": {"label": "File Upload & Hash", "next": ["create_file_record"]},
|
||||
"create_file_record": {"label": "Create File Record", "next": ["check_text"]},
|
||||
"check_text": {"label": "Check Embedded Text", "next": ["extract_text", "process_with_azure_document_intelligence"]},
|
||||
"check_text": {
|
||||
"label": "Check Embedded Text",
|
||||
"next": ["extract_text", "process_with_azure_document_intelligence"],
|
||||
},
|
||||
"extract_text": {"label": "Extract Text (Local)", "next": ["extract_metadata_with_gpt"]},
|
||||
"process_with_azure_document_intelligence": {"label": "OCR Processing (Azure)", "next": ["extract_metadata_with_gpt"]},
|
||||
"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": ["send_to_all_destinations"]},
|
||||
"send_to_all_destinations": {"label": "Upload to Destinations", "next": [], "has_branches": True}
|
||||
"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_nextcloud": "Nextcloud",
|
||||
"upload_to_paperless": "Paperless-ngx",
|
||||
"upload_to_google_drive": "Google Drive",
|
||||
"upload_to_onedrive": "OneDrive",
|
||||
@@ -263,44 +267,38 @@ def _compute_processing_flow(logs):
|
||||
"queue_webdav": "WebDAV",
|
||||
"queue_ftp": "FTP Storage",
|
||||
"queue_sftp": "SFTP Storage",
|
||||
"queue_email": "Email"
|
||||
"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
|
||||
|
||||
|
||||
# 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
|
||||
})
|
||||
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
|
||||
})
|
||||
|
||||
step_map[step_name].append(
|
||||
{"status": log.status, "message": log.message, "timestamp": log.timestamp, "task_id": log.task_id}
|
||||
)
|
||||
|
||||
# Build the flow structure
|
||||
flow = []
|
||||
for stage_key, stage_info in stages.items():
|
||||
stage_logs = step_map.get(stage_key, [])
|
||||
|
||||
|
||||
# Determine overall status for this stage
|
||||
if stage_logs:
|
||||
latest_log = stage_logs[-1]
|
||||
@@ -313,7 +311,7 @@ def _compute_processing_flow(logs):
|
||||
message = None
|
||||
timestamp = None
|
||||
task_id = None
|
||||
|
||||
|
||||
stage_data = {
|
||||
"key": stage_key,
|
||||
"label": stage_info["label"],
|
||||
@@ -322,65 +320,73 @@ def _compute_processing_flow(logs):
|
||||
"timestamp": timestamp,
|
||||
"task_id": task_id,
|
||||
"can_retry": status == "failure",
|
||||
"is_branch_parent": stage_info.get("has_branches", False)
|
||||
"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"
|
||||
})
|
||||
|
||||
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"
|
||||
"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
|
||||
@@ -389,15 +395,15 @@ def _compute_step_summary(logs):
|
||||
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)
|
||||
"total_upload_tasks": len(upload_tasks_seen),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user