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

Apply code formatting and fix linting issues

Browse Source 
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-07 17:37:50 +00:00
parent 2db58afedf
commit 7d24b7c668
4 changed files with 288 additions and 300 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/tasks/convert_to_pdf.py
+94 -69
View File
@@ -1,23 +1,25 @@
#!/usr/bin/env python3
import os
import requests
import logging
import mimetypes
import json
import os
import requests
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(bind=True)
def convert_to_pdf(self, file_path, original_filename=None):
"""
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.
Args:
file_path: Path to the file to convert
original_filename: Optional original filename (if different from path basename)
@@ -25,7 +27,7 @@ def convert_to_pdf(self, file_path, original_filename=None):
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(f"[{task_id}] Gotenberg URL is not configured in settings.")
@@ -42,68 +44,89 @@ def convert_to_pdf(self, file_path, original_filename=None):
endpoint = None
form_data = {}
files = {}
# Dictionary mapping file extensions to their handlers
OFFICE_EXTENSIONS = {
'.doc', '.docx', '.docm', '.dot', '.dotx', '.dotm', # Word
'.xls', '.xlsx', '.xlsm', '.xlsb', '.xlt', '.xltx', '.xlw', # Excel
'.ppt', '.pptx', '.pptm', '.pps', '.ppsx', '.pot', '.potx', # PowerPoint
'.odt', '.ods', '.odp', '.odg', '.odf', # OpenOffice/LibreOffice
'.rtf', '.txt', '.csv', # Text formats
'.pdf', # PDF (already in PDF format but can be processed)
".doc",
".docx",
".docm",
".dot",
".dotx",
".dotm", # Word
".xls",
".xlsx",
".xlsm",
".xlsb",
".xlt",
".xltx",
".xlw", # Excel
".ppt",
".pptx",
".pptm",
".pps",
".ppsx",
".pot",
".potx", # PowerPoint
".odt",
".ods",
".odp",
".odg",
".odf", # OpenOffice/LibreOffice
".rtf",
".txt",
".csv", # Text formats
".pdf", # PDF (already in PDF format but can be processed)
}
IMAGE_EXTENSIONS = {
'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.tif', '.webp', '.svg'
}
HTML_EXTENSIONS = {
'.html', '.htm'
}
IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".bmp", ".tiff", ".tif", ".webp", ".svg"}
HTML_EXTENSIONS = {".html", ".htm"}
# Use LibreOffice endpoint for office documents and images
if (mime_type and 'office' in mime_type) or \
(mime_type and 'opendocument' in mime_type) or \
(mime_type and mime_type.startswith('image/')) or \
file_ext in OFFICE_EXTENSIONS or \
file_ext in IMAGE_EXTENSIONS:
if (
(mime_type and "office" in mime_type)
or (mime_type and "opendocument" in mime_type)
or (mime_type and mime_type.startswith("image/"))
or file_ext in OFFICE_EXTENSIONS
or file_ext in IMAGE_EXTENSIONS
):
endpoint = f"{gotenberg_url}/forms/libreoffice/convert"
files = {'files': (os.path.basename(file_path), open(file_path, 'rb'))}
files = {"files": (os.path.basename(file_path), open(file_path, "rb"))}
# Add some quality settings for better PDF output
form_data = {
'landscape': 'false',
'exportBookmarks': 'true',
'exportNotes': 'false',
'losslessImageCompression': 'true', # Use lossless compression for images
'pdfa': 'PDF/A-2b', # Produce PDF/A-2b compatible output
"landscape": "false",
"exportBookmarks": "true",
"exportNotes": "false",
"losslessImageCompression": "true", # Use lossless compression for images
"pdfa": "PDF/A-2b", # Produce PDF/A-2b compatible output
}
# Use Chromium endpoint for HTML documents
elif (mime_type and mime_type == 'text/html') or file_ext in HTML_EXTENSIONS:
elif (mime_type and mime_type == "text/html") or file_ext in HTML_EXTENSIONS:
endpoint = f"{gotenberg_url}/forms/chromium/convert/html"
# Gotenberg requires the form field to be exactly 'index.html'
# The content filename doesn't matter, just the form field key
files = {'index.html': ('index.html', open(file_path, 'rb'))}
files = {"index.html": ("index.html", open(file_path, "rb"))}
# Add options for better HTML to PDF conversion
form_data = {
'paperWidth': '8.27', # A4 width in inches
'paperHeight': '11.7', # A4 height in inches
'marginTop': '0.4',
'marginBottom': '0.4',
'marginLeft': '0.4',
'marginRight': '0.4',
'printBackground': 'true',
'preferCssPageSize': 'false',
'waitDelay': '2s', # Wait for JavaScript to execute
"paperWidth": "8.27", # A4 width in inches
"paperHeight": "11.7", # A4 height in inches
"marginTop": "0.4",
"marginBottom": "0.4",
"marginLeft": "0.4",
"marginRight": "0.4",
"printBackground": "true",
"preferCssPageSize": "false",
"waitDelay": "2s", # Wait for JavaScript to execute
}
# Use Markdown route for markdown files
elif (mime_type and mime_type in ['text/markdown', 'text/x-markdown']) or file_ext in ['.md', '.markdown']:
elif (mime_type and mime_type in ["text/markdown", "text/x-markdown"]) or file_ext in [".md", ".markdown"]:
# For Markdown, we need both the markdown file and an HTML wrapper
endpoint = f"{gotenberg_url}/forms/chromium/convert/markdown"
# Create a simple HTML wrapper for the markdown
# IMPORTANT: The filename in the template must match the key used in the files dictionary
markdown_filename = os.path.basename(file_path)
@@ -125,35 +148,35 @@ def convert_to_pdf(self, file_path, original_filename=None):
{{{{ toHTML "{markdown_filename}" }}}}
</body>
</html>"""
# Create a temporary HTML wrapper file
wrapper_path = os.path.join(os.path.dirname(file_path), "md_wrapper.html")
with open(wrapper_path, 'w') as f:
with open(wrapper_path, "w") as f:
f.write(html_wrapper)
try:
files = {
'index.html': ('index.html', open(wrapper_path, 'rb')),
markdown_filename: (markdown_filename, open(file_path, 'rb'))
"index.html": ("index.html", open(wrapper_path, "rb")),
markdown_filename: (markdown_filename, open(file_path, "rb")),
}
form_data = {
'paperWidth': '8.27', # A4 width in inches
'paperHeight': '11.7', # A4 height in inches
'marginTop': '0.4',
'marginBottom': '0.4',
'marginLeft': '0.4',
'marginRight': '0.4',
"paperWidth": "8.27", # A4 width in inches
"paperHeight": "11.7", # A4 height in inches
"marginTop": "0.4",
"marginBottom": "0.4",
"marginLeft": "0.4",
"marginRight": "0.4",
}
finally:
# Clean up the temporary wrapper file after preparing the request
if os.path.exists(wrapper_path):
os.remove(wrapper_path)
# Fallback to LibreOffice for everything else
else:
endpoint = f"{gotenberg_url}/forms/libreoffice/convert"
files = {'files': (os.path.basename(file_path), open(file_path, 'rb'))}
files = {"files": (os.path.basename(file_path), open(file_path, "rb"))}
logger.warning(f"Using fallback conversion for unknown type: {mime_type} / {file_ext}")
if not endpoint:
@@ -164,20 +187,22 @@ def convert_to_pdf(self, file_path, original_filename=None):
try:
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)
if response.status_code == 200:
# Save the converted PDF
converted_file_path = os.path.splitext(file_path)[0] + ".pdf"
with open(converted_file_path, "wb") as out_file:
out_file.write(response.content)
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)}")
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, preserving original filename if provided
if original_filename:
# Change extension to .pdf for the original filename
@@ -186,7 +211,7 @@ def convert_to_pdf(self, file_path, original_filename=None):
process_document.delay(converted_file_path, original_filename=pdf_original_filename)
else:
process_document.delay(converted_file_path)
return converted_file_path
else:
error_msg = f"Status code: {response.status_code}"
app/tasks/process_document.py
+15 -24
View File
@@ -1,21 +1,22 @@
#!/usr/bin/env python3
import os
import uuid
import shutil
import mimetypes
import logging
import mimetypes
import os
import shutil
import uuid
import PyPDF2 # Replace fitz with PyPDF2
from app.celery_app import celery
from app.config import settings
from app.tasks.retry_config import BaseTaskWithRetry
from app.database import SessionLocal
from app.models import FileRecord
from app.tasks.extract_metadata_with_gpt import extract_metadata_with_gpt
from app.tasks.process_with_azure_document_intelligence import (
process_with_azure_document_intelligence,
)
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.tasks.retry_config import BaseTaskWithRetry
from app.utils import hash_file, log_task_progress
logger = logging.getLogger(__name__)
@@ -63,9 +64,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
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}"
)
logger.info(f"[{task_id}] File hash: {filehash[:10]}..., Size: {file_size} bytes, MIME: {mime_type}")
log_task_progress(
task_id,
"hash_file",
@@ -77,9 +76,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
with SessionLocal() as db:
existing = db.query(FileRecord).filter_by(filehash=filehash).one_or_none()
if existing:
logger.info(
f"[{task_id}] 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",
@@ -95,9 +92,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
# 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"
)
log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record")
new_record = FileRecord(
filehash=filehash,
original_filename=original_filename,
@@ -169,9 +164,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
break
if has_text:
logger.info(
f"[{task_id}] 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",
@@ -221,9 +214,7 @@ def process_document(self, original_local_file: str, original_filename: str = No
}
# 3. If no embedded text, queue Azure Document Intelligence processing
logger.info(
f"[{task_id}] No embedded text found. Queueing Azure Document Intelligence processing"
)
logger.info(f"[{task_id}] No embedded text found. Queueing Azure Document Intelligence processing")
log_task_progress(
task_id,
"check_text",