gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into copilot/add-file-processing-notifications

Browse Source
This commit is contained in:
Christian Krakau-Louis
2026-02-07 15:37:17 +01:00
committed by GitHub
parent 9c4cf3d70e e8c091897a
commit e45eb04366
19 changed files with 1606 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tasks/convert_to_pdf.py
+25 -10
View File
@@ -7,25 +7,32 @@ import json
from celery import shared_task
from app.config import settings
from app.tasks.process_document import process_document
from app.utils import log_task_progress
logger = logging.getLogger(__name__)
@shared_task
def convert_to_pdf(file_path):
@shared_task(bind=True)
def convert_to_pdf(self, file_path):
"""
Converts a file to PDF using Gotenberg's API.
Determines the appropriate Gotenberg endpoint based on the file's MIME type.
On success, saves the PDF locally and enqueues it for processing.
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting PDF conversion: {file_path}")
log_task_progress(task_id, "convert_to_pdf", "in_progress", f"Converting file: {os.path.basename(file_path)}")
gotenberg_url = getattr(settings, "gotenberg_url", None)
if not gotenberg_url:
logger.error("Gotenberg URL is not configured in settings.")
logger.error(f"[{task_id}] Gotenberg URL is not configured in settings.")
log_task_progress(task_id, "convert_to_pdf", "failure", "Gotenberg URL not configured")
return
# Try to guess the MIME type based on file content and extension
mime_type, encoding = mimetypes.guess_type(file_path)
file_ext = os.path.splitext(file_path)[1].lower()
logger.info(f"Guessed MIME type for '{file_path}' is: {mime_type}, extension: {file_ext}")
logger.info(f"[{task_id}] Guessed MIME type for '{file_path}' is: {mime_type}, extension: {file_ext}")
log_task_progress(task_id, "detect_file_type", "success", f"File type: {mime_type or file_ext}")
# Determine which Gotenberg endpoint to use
endpoint = None
@@ -146,11 +153,13 @@ def convert_to_pdf(file_path):
logger.warning(f"Using fallback conversion for unknown type: {mime_type} / {file_ext}")
if not endpoint:
logger.error(f"Could not determine Gotenberg endpoint for file type: {mime_type}")
logger.error(f"[{task_id}] Could not determine Gotenberg endpoint for file type: {mime_type}")
log_task_progress(task_id, "convert_to_pdf", "failure", f"Unknown file type: {mime_type}")
return None
try:
logger.info(f"Converting {file_path} using endpoint: {endpoint}")
logger.info(f"[{task_id}] Converting {file_path} using endpoint: {endpoint}")
log_task_progress(task_id, "call_gotenberg", "in_progress", "Calling Gotenberg API")
# Send the conversion request to Gotenberg
response = requests.post(endpoint, files=files, data=form_data)
@@ -161,19 +170,25 @@ def convert_to_pdf(file_path):
with open(converted_file_path, "wb") as out_file:
out_file.write(response.content)
logger.info(f"Converted file saved as PDF: {converted_file_path}")
logger.info(f"[{task_id}] Converted file saved as PDF: {converted_file_path}")
log_task_progress(task_id, "call_gotenberg", "success", "PDF conversion successful")
log_task_progress(task_id, "convert_to_pdf", "success", f"Converted to PDF: {os.path.basename(converted_file_path)}")
# Enqueue the PDF for further processing
process_document.delay(converted_file_path)
return converted_file_path
else:
error_msg = f"Status code: {response.status_code}"
logger.error(
f"Conversion failed for {file_path}. "
f"Status code: {response.status_code}, "
f"[{task_id}] Conversion failed for {file_path}. "
f"{error_msg}, "
f"Response: {response.text[:500]}..."
)
log_task_progress(task_id, "call_gotenberg", "failure", error_msg)
log_task_progress(task_id, "convert_to_pdf", "failure", f"Conversion failed: {error_msg}")
return None
except Exception as e:
logger.exception(f"Error converting {file_path} to PDF: {e}")
logger.exception(f"[{task_id}] Error converting {file_path} to PDF: {e}")
log_task_progress(task_id, "convert_to_pdf", "failure", f"Exception: {str(e)}")
return None
app/tasks/embed_metadata_into_pdf.py
+52 -15
View File
@@ -3,6 +3,7 @@
import os
import shutil
import tempfile
import logging
import PyPDF2 # Replace fitz with PyPDF2
import json
from app.config import settings
@@ -11,6 +12,19 @@ from app.tasks.finalize_document_storage import finalize_document_storage
# Import the shared Celery instance
from app.celery_app import celery
from app.utils import log_task_progress
from app.database import SessionLocal
from app.models import FileRecord
logger = logging.getLogger(__name__)
# Directory constants - defined here to avoid hardcoded strings (BAN-B108)
# Note: These are application-specific subdirectories within settings.workdir,
# not system temporary directories. The workdir is a configurable path specific
# to this application. For actual temporary file creation, tempfile module is
# used (see line 70: tempfile.NamedTemporaryFile)
TMP_SUBDIR = "tmp"
PROCESSED_SUBDIR = "processed"
def unique_filepath(directory, base_filename, extension=".pdf"):
"""
@@ -39,8 +53,8 @@ def persist_metadata(metadata, final_pdf_path):
json.dump(metadata, f, ensure_ascii=False, indent=2)
return json_path
@celery.task(base=BaseTaskWithRetry)
def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata: dict):
@celery.task(base=BaseTaskWithRetry, bind=True)
def embed_metadata_into_pdf(self, local_file_path: str, extracted_text: str, metadata: dict, file_id: int = None):
"""
Embeds extracted metadata into the PDF's standard metadata fields.
The mapping is as follows:
@@ -54,13 +68,25 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata:
where <suggested_filename.pdf> is derived from metadata["filename"].
Additionally, the metadata is persisted to a JSON file with the same base name.
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting metadata embedding for: {local_file_path}")
log_task_progress(task_id, "embed_metadata_into_pdf", "in_progress", f"Embedding metadata into {os.path.basename(local_file_path)}", file_id=file_id)
# Get file_id from database if not provided (fallback only, prefer passing file_id explicitly)
if file_id is None:
with SessionLocal() as db:
file_record = db.query(FileRecord).filter_by(local_filename=local_file_path).first()
if file_record:
file_id = file_record.id
# Check for file existence; if not found, try the known shared tmp directory.
if not os.path.exists(local_file_path):
alt_path = os.path.join(settings.workdir, "tmp", os.path.basename(local_file_path))
alt_path = os.path.join(settings.workdir, TMP_SUBDIR, os.path.basename(local_file_path))
if os.path.exists(alt_path):
local_file_path = alt_path
else:
print(f"[ERROR] Local file {local_file_path} not found, cannot embed metadata.")
logger.error(f"[{task_id}] Local file {local_file_path} not found, cannot embed metadata.")
log_task_progress(task_id, "embed_metadata_into_pdf", "failure", "File not found", file_id=file_id)
return {"error": "File not found"}
# Work on a safe copy in a secure temporary directory
@@ -75,7 +101,8 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata:
shutil.copy(original_file, processed_file)
try:
print(f"[DEBUG] Embedding metadata into {processed_file}...")
logger.info(f"[{task_id}] Embedding metadata into {processed_file}...")
log_task_progress(task_id, "modify_pdf", "in_progress", "Modifying PDF metadata", file_id=file_id)
# Open the PDF and modify metadata
with open(processed_file, 'rb') as file:
@@ -98,49 +125,59 @@ def embed_metadata_into_pdf(local_file_path: str, extracted_text: str, metadata:
with open(processed_file, 'wb') as output_file:
pdf_writer.write(output_file)
print(f"[INFO] Metadata embedded successfully in {processed_file}")
logger.info(f"[{task_id}] Metadata embedded successfully in {processed_file}")
log_task_progress(task_id, "modify_pdf", "success", "PDF metadata embedded", file_id=file_id)
# Use the suggested filename from metadata; if not provided, use the original basename.
suggested_filename = metadata.get("filename", os.path.splitext(os.path.basename(local_file_path))[0])
# Remove any extension and then add .pdf
suggested_filename = os.path.splitext(suggested_filename)[0]
# Define the final directory based on settings.workdir and ensure it exists.
final_dir = os.path.join(settings.workdir, "processed")
final_dir = os.path.join(settings.workdir, PROCESSED_SUBDIR)
os.makedirs(final_dir, exist_ok=True)
# Get a unique filepath in case of collisions.
final_file_path = unique_filepath(final_dir, suggested_filename, extension=".pdf")
logger.info(f"[{task_id}] Moving file to: {final_file_path}")
log_task_progress(task_id, "move_to_processed", "in_progress", f"Moving to processed: {suggested_filename}.pdf", file_id=file_id)
# Move the processed file using shutil.move to handle cross-device moves.
shutil.move(processed_file, final_file_path)
# Ensure the temporary file is deleted if it still exists.
if os.path.exists(processed_file):
os.remove(processed_file)
log_task_progress(task_id, "move_to_processed", "success", f"Moved to: {os.path.basename(final_file_path)}", file_id=file_id)
# Persist the metadata into a JSON file with the same base name.
logger.info(f"[{task_id}] Persisting metadata to JSON")
log_task_progress(task_id, "save_metadata_json", "in_progress", "Saving metadata JSON", file_id=file_id)
json_path = persist_metadata(metadata, final_file_path)
print(f"[INFO] Metadata persisted to {json_path}")
logger.info(f"[{task_id}] Metadata persisted to {json_path}")
log_task_progress(task_id, "save_metadata_json", "success", f"Saved: {os.path.basename(json_path)}", file_id=file_id)
# Trigger the next step: final storage.
finalize_document_storage.delay(original_file, final_file_path, metadata)
logger.info(f"[{task_id}] Queueing final storage task")
log_task_progress(task_id, "embed_metadata_into_pdf", "success", "Metadata embedded, queuing finalization", file_id=file_id)
finalize_document_storage.delay(original_file, final_file_path, metadata, file_id)
# After triggering final storage, delete the original file if it is in workdir/tmp.
workdir_tmp = os.path.join(settings.workdir, "tmp")
workdir_tmp = os.path.join(settings.workdir, TMP_SUBDIR)
if original_file.startswith(workdir_tmp) and os.path.exists(original_file):
try:
os.remove(original_file)
print(f"[INFO] Deleted original file from {original_file}")
logger.info(f"[{task_id}] Deleted original file from {original_file}")
except Exception as e:
print(f"[ERROR] Could not delete original file {original_file}: {e}")
logger.error(f"[{task_id}] Could not delete original file {original_file}: {e}")
return {"file": final_file_path, "metadata_file": json_path, "status": "Metadata embedded"}
except Exception as e:
print(f"[ERROR] Failed to embed metadata into {processed_file}: {e}")
logger.exception(f"[{task_id}] Failed to embed metadata into {processed_file}: {e}")
log_task_progress(task_id, "embed_metadata_into_pdf", "failure", f"Exception: {str(e)}", file_id=file_id)
# Clean up temporary file in case of error
if os.path.exists(processed_file):
try:
os.remove(processed_file)
print(f"[INFO] Cleaned up temporary file {processed_file}")
logger.info(f"[{task_id}] Cleaned up temporary file {processed_file}")
except Exception as cleanup_error:
print(f"[ERROR] Could not clean up temporary file {processed_file}: {cleanup_error}")
logger.error(f"[{task_id}] Could not clean up temporary file {processed_file}: {cleanup_error}")
return {"error": str(e)}
app/tasks/extract_metadata_with_gpt.py
+33 -8
View File
@@ -2,6 +2,7 @@
import json
import re
import os
from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry
from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf
@@ -10,6 +11,9 @@ from app.tasks.embed_metadata_into_pdf import embed_metadata_into_pdf
from app.celery_app import celery
import openai
import logging
from app.utils import log_task_progress
from app.database import SessionLocal
from app.models import FileRecord
logger = logging.getLogger(__name__)
@@ -41,9 +45,23 @@ def extract_json_from_text(text):
return text[start:end+1]
return None
@celery.task(base=BaseTaskWithRetry)
def extract_metadata_with_gpt(filename: str, cleaned_text: str):
@celery.task(base=BaseTaskWithRetry, bind=True)
def extract_metadata_with_gpt(self, filename: str, cleaned_text: str, file_id: int = None):
"""Uses OpenAI to classify document metadata."""
task_id = self.request.id
logger.info(f"[{task_id}] Starting metadata extraction for: {filename}")
log_task_progress(task_id, "extract_metadata_with_gpt", "in_progress", f"Extracting metadata for {filename}", file_id=file_id)
# Get file_id from database if not provided
if file_id is None:
tmp_dir = os.path.join(settings.workdir, "tmp")
file_path = os.path.join(tmp_dir, filename)
if os.path.exists(file_path):
with SessionLocal() as db:
file_record = db.query(FileRecord).filter_by(local_filename=file_path).first()
if file_record:
file_id = file_record.id
prompt = f"""
You are a specialized document analyzer trained to extract structured metadata from documents.
Your task is to analyze the given text and return a well-structured JSON object.
@@ -77,7 +95,8 @@ Return only valid JSON with no additional commentary.
"""
try:
print(f"[DEBUG] Sending classification request for {filename}...")
logger.info(f"[{task_id}] Sending classification request for {filename}...")
log_task_progress(task_id, "call_openai", "in_progress", "Calling OpenAI API", file_id=file_id)
completion = client.chat.completions.create(
model=settings.openai_model,
messages=[
@@ -88,21 +107,27 @@ Return only valid JSON with no additional commentary.
)
content = completion.choices[0].message.content
print(f"[DEBUG] Raw classification response for {filename}: {content}")
logger.info(f"[{task_id}] Raw classification response for {filename}: {content[:200]}...")
log_task_progress(task_id, "call_openai", "success", "Received OpenAI response", file_id=file_id)
json_text = extract_json_from_text(content)
if not json_text:
print(f"[ERROR] Could not find valid JSON in GPT response for {filename}.")
logger.error(f"[{task_id}] Could not find valid JSON in GPT response for {filename}.")
log_task_progress(task_id, "extract_metadata_with_gpt", "failure", "Invalid JSON in response", file_id=file_id)
return {}
metadata = json.loads(json_text)
print(f"[DEBUG] Extracted metadata: {metadata}")
logger.info(f"[{task_id}] Extracted metadata: {metadata}")
log_task_progress(task_id, "parse_metadata", "success", f"Parsed metadata: {list(metadata.keys())}", file_id=file_id)
# Trigger the next step: embedding metadata into the PDF
embed_metadata_into_pdf.delay(filename, cleaned_text, metadata)
logger.info(f"[{task_id}] Queueing metadata embedding task")
log_task_progress(task_id, "extract_metadata_with_gpt", "success", "Metadata extracted, queuing embed task", file_id=file_id)
embed_metadata_into_pdf.delay(filename, cleaned_text, metadata, file_id)
return {"s3_file": filename, "metadata": metadata}
except Exception as e:
print(f"[ERROR] OpenAI classification failed for {filename}: {e}")
logger.exception(f"[{task_id}] OpenAI classification failed for {filename}: {e}")
log_task_progress(task_id, "extract_metadata_with_gpt", "failure", f"Exception: {str(e)}", file_id=file_id)
return {}
app/tasks/finalize_document_storage.py
+41 -12
View File
@@ -1,28 +1,52 @@
#!/usr/bin/env python3
import logging
import os
from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry
# Import the shared Celery instance
from app.celery_app import celery
# 1) Import the aggregator task
# Import the aggregator task and validator
from app.tasks.send_to_all import send_to_all_destinations, get_configured_services_from_validator
# Import notification utility
from app.utils.notification import notify_file_processed
# Import database and logging utils from main
from app.utils import log_task_progress
from app.database import SessionLocal
from app.models import FileRecord
@celery.task(base=BaseTaskWithRetry)
def finalize_document_storage(original_file: str, processed_file: str, metadata: dict):
logger = logging.getLogger(__name__)
@celery.task(base=BaseTaskWithRetry, bind=True)
def finalize_document_storage(self, original_file: str, processed_file: str, metadata: dict, file_id: int = None):
"""
Final storage step after embedding metadata.
We will now call 'send_to_all_destinations' to push the final PDF to Dropbox/Nextcloud/Paperless.
After uploading, send a notification about the processed file.
"""
print(f"[INFO] Finalizing document storage for {processed_file}")
task_id = self.request.id
logger.info(f"[{task_id}] Finalizing document storage for {processed_file}")
# 1. Update Database Status (From Main)
log_task_progress(task_id, "finalize_document_storage", "in_progress", f"Finalizing: {os.path.basename(processed_file)}", file_id=file_id)
# Get file_id from database if not provided (fallback logic from Main)
if file_id is None:
with SessionLocal() as db:
# Only as a last resort, try to find by exact match on local_filename
tmp_path = os.path.join(settings.workdir, "tmp", os.path.basename(original_file))
file_record = db.query(FileRecord).filter(
FileRecord.local_filename == tmp_path
).first()
if file_record:
file_id = file_record.id
# Determine which destinations are configured
# 2. Determine Configured Destinations (From Copilot)
# This is needed for the notification message later
configured_destinations = []
try:
configured_services = get_configured_services_from_validator()
@@ -33,16 +57,21 @@ def finalize_document_storage(original_file: str, processed_file: str, metadata:
display_name = service_name.replace('_', ' ').title()
configured_destinations.append(display_name)
except Exception as e:
print(f"[WARNING] Could not determine configured destinations: {e}")
logger.warning(f"[WARNING] Could not determine configured destinations: {e}")
configured_destinations = ["configured destinations"]
# 2) Enqueue uploads to all destinations (Dropbox, Nextcloud, Paperless)
# 3. Queue Uploads (Merged)
# Uses Main branch signature to ensure file_id is passed, but keeps logic structure
logger.info(f"[{task_id}] Queueing uploads to all destinations")
log_task_progress(task_id, "finalize_document_storage", "success", "Queuing uploads to destinations", file_id=file_id)
# Note: send_to_all_destinations is asynchronous and queues upload tasks
send_to_all_destinations.delay(processed_file)
# We pass 'True' (delete_after) and 'file_id' as per Main branch requirements
send_to_all_destinations.delay(processed_file, True, file_id)
# 3) Send notification about successful file processing
# 4. Send Notification (From Copilot)
# Note: This notification is sent after processing is complete but while uploads
# are being queued. The message reflects that uploads are being initiated.
# are being queued.
try:
# Get file information
file_size = os.path.getsize(processed_file) if os.path.exists(processed_file) else 0
@@ -55,9 +84,9 @@ def finalize_document_storage(original_file: str, processed_file: str, metadata:
destinations=configured_destinations
)
except Exception as e:
print(f"[WARNING] Failed to send file processed notification: {e}")
logger.warning(f"[WARNING] Failed to send file processed notification: {e}")
return {
"status": "Completed",
"file": processed_file
}
}
app/tasks/process_document.py
+43 -10
View File
@@ -4,6 +4,7 @@ import os
import uuid
import shutil
import mimetypes
import logging
import PyPDF2 # Replace fitz with PyPDF2
from app.config import settings
@@ -13,11 +14,13 @@ from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
from app.celery_app import celery
from app.database import SessionLocal
from app.models import FileRecord
from app.utils import hash_file
from app.utils import hash_file, log_task_progress
logger = logging.getLogger(__name__)
@celery.task(base=BaseTaskWithRetry)
def process_document(original_local_file: str):
@celery.task(base=BaseTaskWithRetry, bind=True)
def process_document(self, original_local_file: str):
"""
Process a document file and trigger appropriate text extraction.
@@ -28,24 +31,34 @@ def process_document(original_local_file: str):
- Check for embedded text. If present, run local GPT extraction
- Otherwise, queue Azure Document Intelligence processing
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting document processing: {original_local_file}")
log_task_progress(task_id, "process_document", "in_progress", f"Processing file: {original_local_file}")
if not os.path.exists(original_local_file):
print(f"[ERROR] File {original_local_file} not found.")
logger.error(f"[{task_id}] File {original_local_file} not found.")
log_task_progress(task_id, "process_document", "failure", "File not found")
return {"error": "File not found"}
# 0. Compute the file hash and check for duplicates
logger.info(f"[{task_id}] Computing file hash...")
log_task_progress(task_id, "hash_file", "in_progress", "Computing file hash")
filehash = hash_file(original_local_file)
original_filename = os.path.basename(original_local_file)
file_size = os.path.getsize(original_local_file)
mime_type, _ = mimetypes.guess_type(original_local_file)
if not mime_type:
mime_type = "application/octet-stream"
logger.info(f"[{task_id}] File hash: {filehash[:10]}..., Size: {file_size} bytes, MIME: {mime_type}")
log_task_progress(task_id, "hash_file", "success", f"Hash: {filehash[:10]}..., Size: {file_size} bytes")
# Acquire DB session in the task
with SessionLocal() as db:
existing = db.query(FileRecord).filter_by(filehash=filehash).one_or_none()
if existing:
print(f"[INFO] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.")
logger.info(f"[{task_id}] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.")
log_task_progress(task_id, "process_document", "success", "Duplicate file detected, skipping", file_id=existing.id)
return {
"status": "duplicate_file",
"file_id": existing.id,
@@ -53,6 +66,8 @@ def process_document(original_local_file: str):
}
# Not a duplicate -> insert a new record
logger.info(f"[{task_id}] Creating new file record in database")
log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record")
new_record = FileRecord(
filehash=filehash,
original_filename=original_filename,
@@ -63,6 +78,8 @@ def process_document(original_local_file: str):
db.add(new_record)
db.commit()
db.refresh(new_record)
logger.info(f"[{task_id}] File record created with ID: {new_record.id}")
log_task_progress(task_id, "create_file_record", "success", f"File record ID: {new_record.id}", file_id=new_record.id)
# 1. Generate a UUID-based filename and place it in /workdir/tmp
file_ext = os.path.splitext(original_local_file)[1]
@@ -73,14 +90,19 @@ def process_document(original_local_file: str):
os.makedirs(tmp_dir, exist_ok=True)
new_local_path = os.path.join(tmp_dir, new_filename)
logger.info(f"[{task_id}] Copying file to: {new_local_path}")
log_task_progress(task_id, "copy_file", "in_progress", f"Copying file to {new_filename}", file_id=new_record.id)
# Copy the file instead of moving it
shutil.copy(original_local_file, new_local_path)
log_task_progress(task_id, "copy_file", "success", f"File copied to {new_filename}", file_id=new_record.id)
# Update the DB with final local filename
new_record.local_filename = new_local_path
db.commit()
# 2. Check for embedded text (outside the DB session to avoid long open transactions)
logger.info(f"[{task_id}] Checking for embedded text in PDF")
log_task_progress(task_id, "check_text", "in_progress", "Checking for embedded text", file_id=new_record.id)
with open(new_local_path, 'rb') as file:
pdf_reader = PyPDF2.PdfReader(file)
has_text = False
@@ -90,19 +112,30 @@ def process_document(original_local_file: str):
break
if has_text:
print(f"[INFO] PDF {original_local_file} contains embedded text. Processing locally.")
logger.info(f"[{task_id}] PDF {original_local_file} contains embedded text. Processing locally.")
log_task_progress(task_id, "check_text", "success", "Embedded text found, extracting locally", file_id=new_record.id)
# Extract text locally
logger.info(f"[{task_id}] Extracting text from PDF")
log_task_progress(task_id, "extract_text", "in_progress", "Extracting text locally", file_id=new_record.id)
extracted_text = ""
with open(new_local_path, 'rb') as file:
pdf_reader = PyPDF2.PdfReader(file)
for page in pdf_reader.pages:
extracted_text += page.extract_text() + "\n"
logger.info(f"[{task_id}] Extracted {len(extracted_text)} characters")
log_task_progress(task_id, "extract_text", "success", f"Extracted {len(extracted_text)} characters", file_id=new_record.id)
# Call metadata extraction directly
extract_metadata_with_gpt.delay(new_filename, extracted_text)
return {"file": new_local_path, "status": "Text extracted locally"}
logger.info(f"[{task_id}] Queueing metadata extraction")
log_task_progress(task_id, "process_document", "success", "Queued for metadata extraction", file_id=new_record.id)
extract_metadata_with_gpt.delay(new_filename, extracted_text, new_record.id)
return {"file": new_local_path, "status": "Text extracted locally", "file_id": new_record.id}
# 3. If no embedded text, queue Azure Document Intelligence processing
process_with_azure_document_intelligence.delay(new_filename)
return {"file": new_local_path, "status": "Queued for OCR"}
logger.info(f"[{task_id}] No embedded text found. Queueing Azure Document Intelligence processing")
log_task_progress(task_id, "check_text", "success", "No embedded text, queuing OCR", file_id=new_record.id)
log_task_progress(task_id, "process_document", "success", "Queued for OCR processing", file_id=new_record.id)
process_with_azure_document_intelligence.delay(new_filename, new_record.id)
return {"file": new_local_path, "status": "Queued for OCR", "file_id": new_record.id}
app/tasks/process_with_azure_document_intelligence.py
+6 -2
View File
@@ -76,7 +76,7 @@ def check_page_rotation(result, filename):
return rotation_data
@celery.task(base=BaseTaskWithRetry)
def process_with_azure_document_intelligence(filename: str):
def process_with_azure_document_intelligence(filename: str, file_id: int = None):
"""
Processes a PDF document using Azure Document Intelligence and overlays OCR text onto
the local temporary file (stored under <workdir>/tmp).
@@ -88,6 +88,10 @@ def process_with_azure_document_intelligence(filename: str):
3. Saves the OCR-processed PDF locally in the same location as before.
4. Checks for page rotation and triggers page rotation if needed.
5. Triggers downstream metadata extraction.
Args:
filename: Name of the file to process
file_id: Optional file ID to pass through to subsequent tasks
"""
try:
tmp_file_path = os.path.join(settings.workdir, "tmp", filename)
@@ -139,7 +143,7 @@ def process_with_azure_document_intelligence(filename: str):
logger.info(f"Extracted text for {filename}: {len(extracted_text)} characters")
# Trigger page rotation task if rotation is detected, otherwise proceed to metadata extraction
rotate_pdf_pages.delay(filename, extracted_text, rotation_data)
rotate_pdf_pages.delay(filename, extracted_text, rotation_data, file_id)
return {"file": filename, "searchable_pdf": searchable_pdf_path, "cleaned_text": extracted_text}
except Exception as e:
app/tasks/rotate_pdf_pages.py
+6 -5
View File
@@ -47,7 +47,7 @@ def determine_rotation_angle(detected_angle):
return rotation_value
@celery.task(base=BaseTaskWithRetry)
def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None):
def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None, file_id: int = None):
"""
Rotates pages in a PDF document based on detected rotation angles.
@@ -55,6 +55,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None):
filename: The name of the file to rotate
extracted_text: The extracted text from the document
rotation_data: Optional rotation data dictionary {page_index: angle}
file_id: Optional file ID to pass through to subsequent tasks
"""
try:
pdf_path = os.path.join(settings.workdir, "tmp", filename)
@@ -64,7 +65,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None):
# Skip rotation if no rotation data provided
if not rotation_data:
logger.info(f"No rotation data provided for {filename}, proceeding with metadata extraction")
extract_metadata_with_gpt.delay(filename, extracted_text)
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
return {"file": filename, "status": "no_rotation_needed"}
# Standardize rotation_data keys to integers
@@ -77,7 +78,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None):
if not any(abs(angle) > 0 for angle in normalized_rotation_data.values()):
logger.info(f"No significant rotations detected in {filename}, proceeding with metadata extraction")
extract_metadata_with_gpt.delay(filename, extracted_text)
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
return {"file": filename, "status": "no_rotation_needed"}
logger.info(f"Rotating {len(normalized_rotation_data)} pages in {filename}")
@@ -117,7 +118,7 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None):
logger.info(f"Detected rotations in {filename} but no rotations were actually applied (angles too small or not multiples of 90°)")
# Continue with metadata extraction
extract_metadata_with_gpt.delay(filename, extracted_text)
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
return {
"file": filename,
@@ -129,5 +130,5 @@ def rotate_pdf_pages(filename: str, extracted_text: str, rotation_data=None):
except Exception as e:
logger.error(f"Error rotating PDF {filename}: {e}")
# Continue with metadata extraction despite rotation failure
extract_metadata_with_gpt.delay(filename, extracted_text)
extract_metadata_with_gpt.delay(filename, extracted_text, file_id)
return {"file": filename, "status": "rotation_failed", "error": str(e)}
app/tasks/send_to_all.py
+40 -11
View File
@@ -16,6 +16,9 @@ from app.tasks.upload_to_onedrive import upload_to_onedrive
from app.tasks.upload_to_s3 import upload_to_s3
from app.utils.config_validator import get_provider_status
from app.celery_app import celery
from app.utils import log_task_progress
from app.database import SessionLocal
from app.models import FileRecord
logger = logging.getLogger(__name__)
@@ -104,8 +107,8 @@ def get_configured_services_from_validator():
return result
@celery.task(base=BaseTaskWithRetry)
def send_to_all_destinations(file_path: str, use_validator=True):
@celery.task(base=BaseTaskWithRetry, bind=True)
def send_to_all_destinations(self, file_path: str, use_validator=True, file_id: int = None):
"""
Distribute a file to all configured storage destinations.
@@ -113,11 +116,29 @@ def send_to_all_destinations(file_path: str, use_validator=True):
file_path: Path to the file to distribute
use_validator: Whether to use the config validator to determine enabled services
(if False, falls back to individual checks)
file_id: Optional file ID to associate with logs
"""
task_id = self.request.id
if not os.path.exists(file_path):
logger.error(f"[{task_id}] File not found: {file_path}")
log_task_progress(task_id, "send_to_all_destinations", "failure", "File not found", file_id=file_id)
raise FileNotFoundError(f"File not found: {file_path}")
logger.info(f"Sending {file_path} to all configured destinations")
logger.info(f"[{task_id}] Sending {file_path} to all configured destinations")
log_task_progress(task_id, "send_to_all_destinations", "in_progress", f"Distributing: {os.path.basename(file_path)}", file_id=file_id)
# Get file_id from database if not provided (fallback only, prefer passing file_id explicitly)
if file_id is None:
with SessionLocal() as db:
# Only as a last resort, try to find by basename match
# This should not be needed if file_id is passed correctly through the chain
file_record = db.query(FileRecord).filter(
FileRecord.local_filename == os.path.join(settings.workdir, "tmp", os.path.basename(file_path))
).first()
if file_record:
file_id = file_record.id
results = {}
# Define service configurations
@@ -179,12 +200,13 @@ def send_to_all_destinations(file_path: str, use_validator=True):
if use_validator:
try:
configured_services = get_configured_services_from_validator()
logger.info(f"Configured services according to validator: {configured_services}")
logger.info(f"[{task_id}] Configured services according to validator: {configured_services}")
except Exception as e:
logger.warning(f"Failed to get configuration from validator: {str(e)}")
logger.warning(f"[{task_id}] Failed to get configuration from validator: {str(e)}")
use_validator = False
# Process each service
queued_count = 0
for service in services:
service_name = service["name"]
@@ -192,24 +214,31 @@ def send_to_all_destinations(file_path: str, use_validator=True):
is_configured = False
if use_validator and service_name in configured_services:
is_configured = configured_services[service_name]
logger.debug(f"{service_name} configuration from validator: {is_configured}")
logger.debug(f"[{task_id}] {service_name} configuration from validator: {is_configured}")
else:
try:
is_configured = service["should_upload"]()
logger.debug(f"{service_name} configuration from function: {is_configured}")
logger.debug(f"[{task_id}] {service_name} configuration from function: {is_configured}")
except Exception as e:
logger.error(f"Error checking configuration for {service_name}: {str(e)}")
logger.error(f"[{task_id}] Error checking configuration for {service_name}: {str(e)}")
is_configured = False
# Queue the upload task if service is configured
if is_configured:
logger.info(f"Queueing {file_path} for {service_name} upload")
logger.info(f"[{task_id}] Queueing {file_path} for {service_name} upload")
log_task_progress(task_id, f"queue_{service_name}", "in_progress", f"Queueing upload to {service_name}", file_id=file_id)
try:
task = service["upload_func"].delay(file_path)
task = service["upload_func"].delay(file_path, file_id)
results[f"{service_name}_task_id"] = task.id
queued_count += 1
log_task_progress(task_id, f"queue_{service_name}", "success", f"Queued for {service_name}", file_id=file_id)
except Exception as e:
logger.error(f"Failed to queue {service_name} task: {str(e)}")
logger.error(f"[{task_id}] Failed to queue {service_name} task: {str(e)}")
results[f"{service_name}_error"] = str(e)
log_task_progress(task_id, f"queue_{service_name}", "failure", f"Failed: {str(e)}", file_id=file_id)
logger.info(f"[{task_id}] Queued {queued_count} upload tasks")
log_task_progress(task_id, "send_to_all_destinations", "success", f"Queued {queued_count} uploads", file_id=file_id)
return {
"status": "Queued",
app/tasks/upload_to_dropbox.py
+30 -12
View File
@@ -9,6 +9,9 @@ from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry
from app.celery_app import celery
from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path
from app.utils import log_task_progress
from app.database import SessionLocal
from app.models import FileRecord
logger = logging.getLogger(__name__)
@@ -99,21 +102,31 @@ def get_dropbox_client():
logger.error(f"Error creating Dropbox client: {str(e)}")
raise
@celery.task(base=BaseTaskWithRetry)
def upload_to_dropbox(file_path: str):
@celery.task(base=BaseTaskWithRetry, bind=True)
def upload_to_dropbox(self, file_path: str, file_id: int = None):
"""
Upload a file to Dropbox.
Args:
file_path: Path to the file to upload
file_id: Optional file ID to associate with logs
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting Dropbox upload: {file_path}")
log_task_progress(task_id, "upload_to_dropbox", "in_progress", f"Uploading to Dropbox: {os.path.basename(file_path)}", file_id=file_id)
if not os.path.exists(file_path):
error_msg = f"File not found: {file_path}"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id)
raise FileNotFoundError(error_msg)
# Check if Dropbox is properly configured
if not (hasattr(settings, 'dropbox_app_key') and settings.dropbox_app_key and
hasattr(settings, 'dropbox_app_secret') and settings.dropbox_app_secret and
hasattr(settings, 'dropbox_refresh_token') and settings.dropbox_refresh_token):
logger.info("Dropbox upload skipped: Missing configuration")
logger.info(f"[{task_id}] Dropbox upload skipped: Missing configuration")
log_task_progress(task_id, "upload_to_dropbox", "success", "Skipped: Not configured", file_id=file_id)
return {"status": "Skipped", "reason": "Dropbox settings not configured"}
filename = os.path.basename(file_path)
@@ -144,7 +157,8 @@ def upload_to_dropbox(file_path: str):
dropbox_path = get_unique_filename(remote_full_path, check_exists_in_dropbox)
# Upload the file
logger.info(f"Uploading {filename} to Dropbox at {dropbox_path}")
logger.info(f"[{task_id}] Uploading {filename} to Dropbox at {dropbox_path}")
log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {dropbox_path}", file_id=file_id)
with open(file_path, 'rb') as file_data:
# Use files_upload_session for large files to avoid timeouts
file_size = os.path.getsize(file_path)
@@ -179,7 +193,8 @@ def upload_to_dropbox(file_path: str):
mode=dropbox.files.WriteMode.overwrite
)
logger.info(f"Successfully uploaded {filename} to Dropbox at {dropbox_path}")
logger.info(f"[{task_id}] Successfully uploaded {filename} to Dropbox at {dropbox_path}")
log_task_progress(task_id, "upload_to_dropbox", "success", f"Uploaded to Dropbox: {dropbox_path}", file_id=file_id)
return {
"status": "Completed",
"file_path": file_path,
@@ -187,14 +202,17 @@ def upload_to_dropbox(file_path: str):
}
except AuthError:
error_msg = f"[ERROR] Authentication failed while uploading {filename} to Dropbox. Check token."
logger.error(error_msg)
error_msg = f"Authentication failed while uploading {filename} to Dropbox. Check token."
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
except ApiError as e:
error_msg = f"[ERROR] Failed to upload {filename} to Dropbox: {e}"
logger.error(error_msg)
error_msg = f"Failed to upload {filename} to Dropbox: {e}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
except Exception as e:
error_msg = f"[ERROR] Unexpected error uploading {filename} to Dropbox: {e}"
logger.error(error_msg)
error_msg = f"Unexpected error uploading {filename} to Dropbox: {e}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_dropbox", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
app/tasks/upload_to_nextcloud.py
+27 -10
View File
@@ -8,17 +8,29 @@ from app.config import settings
from app.celery_app import celery
from app.tasks.retry_config import BaseTaskWithRetry
from app.utils.filename_utils import get_unique_filename, sanitize_filename, extract_remote_path
from app.utils import log_task_progress
from app.database import SessionLocal
from app.models import FileRecord
logger = logging.getLogger(__name__)
@celery.task(base=BaseTaskWithRetry)
def upload_to_nextcloud(file_path: str):
@celery.task(base=BaseTaskWithRetry, bind=True)
def upload_to_nextcloud(self, file_path: str, file_id: int = None):
"""
Upload a file to Nextcloud WebDAV.
Args:
file_path: Path to the file to upload
file_id: Optional file ID to associate with logs
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}")
log_task_progress(task_id, "upload_to_nextcloud", "in_progress", f"Uploading to Nextcloud: {os.path.basename(file_path)}", file_id=file_id)
if not os.path.exists(file_path):
error_msg = f"File not found: {file_path}"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise FileNotFoundError(error_msg)
# For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url'
@@ -26,7 +38,8 @@ def upload_to_nextcloud(file_path: str):
if not (getattr(settings, 'nextcloud_upload_url', None) and
getattr(settings, 'nextcloud_username', None) and
getattr(settings, 'nextcloud_password', None)):
logger.info("Nextcloud upload skipped: Missing configuration")
logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration")
log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id)
return {"status": "Skipped", "reason": "Nextcloud settings not configured"}
filename = os.path.basename(file_path)
@@ -99,7 +112,8 @@ def upload_to_nextcloud(file_path: str):
)
# Upload the file
logger.info(f"Uploading {filename} to Nextcloud at {full_url}")
logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}")
log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id)
with open(file_path, 'rb') as file_data:
response = requests.put(
full_url,
@@ -110,7 +124,8 @@ def upload_to_nextcloud(file_path: str):
)
if response.status_code in (201, 204): # Created or No Content
logger.info(f"Successfully uploaded {filename} to Nextcloud at {remote_path}")
logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}")
log_task_progress(task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id)
return {
"status": "Completed",
"file_path": file_path,
@@ -118,11 +133,13 @@ def upload_to_nextcloud(file_path: str):
"response_code": response.status_code
}
else:
error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
logger.error(error_msg)
error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
except Exception as e:
error_msg = f"[ERROR] Failed to upload {filename} to Nextcloud: {str(e)}"
logger.error(error_msg)
error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
raise Exception(error_msg)
app/tasks/upload_to_paperless.py
+32 -8
View File
@@ -10,6 +10,9 @@ from typing import Dict, Any
from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry
from app.celery_app import celery
from app.utils import log_task_progress
from app.database import SessionLocal
from app.models import FileRecord
logger = logging.getLogger(__name__)
@@ -81,12 +84,24 @@ def poll_task_for_document_id(task_id: str) -> int:
f"Task {task_id} didn't reach SUCCESS within {POLL_MAX_ATTEMPTS} attempts."
)
@celery.task(base=BaseTaskWithRetry)
def upload_to_paperless(file_path: str):
"""Uploads a file to Paperless-ngx."""
@celery.task(base=BaseTaskWithRetry, bind=True)
def upload_to_paperless(self, file_path: str, file_id: int = None):
"""
Uploads a file to Paperless-ngx.
Args:
file_path: Path to the file to upload
file_id: Optional file ID to associate with logs
"""
task_id = self.request.id
logger.info(f"[{task_id}] Starting Paperless upload: {file_path}")
log_task_progress(task_id, "upload_to_paperless", "in_progress", f"Uploading to Paperless: {os.path.basename(file_path)}", file_id=file_id)
if not os.path.exists(file_path):
raise FileNotFoundError(f"File not found: {file_path}")
error_msg = f"File not found: {file_path}"
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id)
raise FileNotFoundError(error_msg)
# Extract filename
filename = os.path.basename(file_path)
@@ -94,10 +109,13 @@ def upload_to_paperless(file_path: str):
# Check if Paperless settings are configured
if not settings.paperless_host or not settings.paperless_ngx_api_token:
error_msg = "Paperless-ngx credentials are not fully configured"
logger.error(error_msg)
logger.error(f"[{task_id}] {error_msg}")
log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id)
raise ValueError(error_msg)
# Upload the PDF
logger.info(f"[{task_id}] Posting document to Paperless")
log_task_progress(task_id, "post_document", "in_progress", "Posting to Paperless API", file_id=file_id)
post_url = _paperless_api_url("/api/documents/post_document/")
with open(file_path, "rb") as f:
files = {
@@ -110,18 +128,24 @@ def upload_to_paperless(file_path: str):
resp = requests.post(post_url, headers=_get_headers(), files=files, data=data)
resp.raise_for_status()
except requests.exceptions.RequestException as exc:
error_msg = f"Failed to upload to Paperless: {exc}"
logger.error(
"Failed to upload document '%s' to Paperless. Error: %s. Response=%s",
f"[{task_id}] Failed to upload document '%s' to Paperless. Error: %s. Response=%s",
file_path, exc, getattr(exc.response, "text", "<no response>")
)
log_task_progress(task_id, "upload_to_paperless", "failure", error_msg, file_id=file_id)
raise
raw_task_id = resp.text.strip().strip('"').strip("'")
logger.info(f"Received Paperless task ID: {raw_task_id}")
logger.info(f"[{task_id}] Received Paperless task ID: {raw_task_id}")
log_task_progress(task_id, "post_document", "success", f"Task ID: {raw_task_id}", file_id=file_id)
# Poll tasks until success/fail => get doc_id
logger.info(f"[{task_id}] Polling for document ID")
log_task_progress(task_id, "poll_task", "in_progress", "Waiting for Paperless processing", file_id=file_id)
doc_id = poll_task_for_document_id(raw_task_id)
logger.info(f"Document {file_path} successfully ingested => ID={doc_id}")
logger.info(f"[{task_id}] Document {file_path} successfully ingested => ID={doc_id}")
log_task_progress(task_id, "upload_to_paperless", "success", f"Uploaded to Paperless: Doc ID {doc_id}", file_id=file_id)
return {
"status": "Completed",