From 9a2c2d20b1b2bcb1d46bb7c96cfd77b64d17c301 Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Thu, 12 Feb 2026 02:37:51 +0100 Subject: [PATCH] feat(duplication): implement duplicate file handling and detection across processing steps --- README.md | 2 + app/database.py | 15 ++++ app/models.py | 3 +- app/tasks/convert_to_pdf.py | 134 ++++++++++++++++++++++++++-- app/tasks/process_document.py | 93 ++++++++++++++++--- app/tasks/upload_to_paperless.py | 42 ++++++++- app/utils/file_queries.py | 6 ++ app/utils/file_status.py | 25 +++++- app/utils/step_manager.py | 15 +++- frontend/templates/file_detail.html | 8 +- frontend/templates/files.html | 8 +- requirements.txt | 2 + 12 files changed, 325 insertions(+), 28 deletions(-) diff --git a/README.md b/README.md index 3b4de2a3..8e0361c8 100644 --- a/README.md +++ b/README.md @@ -207,6 +207,8 @@ The following is a summary of the licenses used by our direct dependencies: | OpenAI | MIT | | PyPDF2 | BSD | | Requests | Apache 2.0 | +| puremagic | MIT | +| filetype | MIT | | Dropbox | MIT | | Azure AI Document Intelligence | MIT | | Authlib | BSD | diff --git a/app/database.py b/app/database.py index c33a5188..5df280a8 100644 --- a/app/database.py +++ b/app/database.py @@ -102,6 +102,21 @@ def _run_schema_migrations(engine): conn.execute(text("ALTER TABLE files ADD COLUMN duplicate_of_id INTEGER")) logger.info("Migration complete: 'duplicate_of_id' column added to files") + # Migration: Drop unique index on filehash to allow duplicate records + try: + indexes = inspector.get_indexes("files") + unique_filehash_indexes = [ + index for index in indexes if index.get("unique") and "filehash" in index.get("column_names", []) + ] + if unique_filehash_indexes: + logger.info("Migrating files: dropping unique index on 'filehash'") + with engine.begin() as conn: + for index in unique_filehash_indexes: + conn.execute(text(f"DROP INDEX IF EXISTS {index['name']}")) + logger.info("Migration complete: unique index on 'filehash' removed") + except Exception as exc: + logger.warning(f"Skipping filehash unique index drop: {exc}") + def get_db(): """ diff --git a/app/models.py b/app/models.py index 90c74d86..7481618e 100644 --- a/app/models.py +++ b/app/models.py @@ -22,7 +22,8 @@ class FileRecord(Base): id = Column(Integer, primary_key=True, index=True) # Hash of the file content (e.g. SHA-256) - filehash = Column(String, unique=True, index=True, nullable=False) + # Note: duplicates are allowed so filehash is not unique + filehash = Column(String, index=True, nullable=False) # The name of the file as it was originally uploaded (if known) original_filename = Column(String) diff --git a/app/tasks/convert_to_pdf.py b/app/tasks/convert_to_pdf.py index 929abf3e..edb43599 100644 --- a/app/tasks/convert_to_pdf.py +++ b/app/tasks/convert_to_pdf.py @@ -2,7 +2,10 @@ import logging import mimetypes import os +from typing import Optional, Tuple +import filetype +import puremagic import requests from celery import shared_task @@ -13,8 +16,115 @@ from app.utils import log_task_progress logger = logging.getLogger(__name__) +def _detect_mime_type_from_magic(file_path: str) -> Optional[str]: + """ + Detect MIME type from file headers using platform-agnostic libraries. + + Args: + file_path: Path to the file on disk. + + Returns: + Detected MIME type or None if unknown. + """ + try: + matches = puremagic.from_file(file_path) + if matches: + return matches[0].mime_type + except puremagic.PureError: + pass + + guess = filetype.guess(file_path) + if guess: + return guess.mime + + return None + + +def _detect_mime_type(file_path: str, original_filename: Optional[str]) -> Tuple[Optional[str], Optional[str]]: + """ + Detect MIME type using extension, original filename, or magic bytes. + + Args: + file_path: Path to the file on disk. + original_filename: Optional original filename provided at upload time. + + Returns: + Tuple of detected MIME type and encoding. + """ + mime_type, encoding = mimetypes.guess_type(file_path) + if mime_type: + return mime_type, encoding + + if original_filename: + mime_type, encoding = mimetypes.guess_type(original_filename) + if mime_type: + return mime_type, encoding + + return _detect_mime_type_from_magic(file_path), encoding + + +def _detect_extension(file_path: str, original_filename: Optional[str], mime_type: Optional[str]) -> str: + """ + Detect file extension from file path, original filename, or MIME type. + + Args: + file_path: Path to the file on disk. + original_filename: Optional original filename provided at upload time. + mime_type: Detected MIME type if available. + + Returns: + File extension with leading dot (e.g., ".pdf") or empty string if unknown. + """ + file_ext = os.path.splitext(file_path)[1].lower() + if file_ext: + return file_ext + + if original_filename: + file_ext = os.path.splitext(original_filename)[1].lower() + if file_ext: + return file_ext + + if mime_type: + return mimetypes.guess_extension(mime_type) or "" + + try: + matches = puremagic.from_file(file_path) + if matches and matches[0].extension: + return f".{matches[0].extension.lstrip('.')}" + except puremagic.PureError: + pass + + guess = filetype.guess(file_path) + if guess and guess.extension: + return f".{guess.extension.lstrip('.')}" + + return "" + + +def _build_filename(file_path: str, original_filename: Optional[str], file_ext: str) -> str: + """ + Build a filename for upload that includes a valid extension when possible. + + Args: + file_path: Path to the file on disk. + original_filename: Optional original filename provided at upload time. + file_ext: Detected file extension. + + Returns: + Filename to send to Gotenberg. + """ + if original_filename and os.path.splitext(original_filename)[1]: + return original_filename + + base_name = os.path.basename(file_path) + if file_ext and not base_name.lower().endswith(file_ext): + return f"{base_name}{file_ext}" + + return base_name + + @shared_task(bind=True) -def convert_to_pdf(self, file_path, original_filename=None): +def convert_to_pdf(self, file_path: str, original_filename: Optional[str] = None) -> Optional[str]: """ Converts a file to PDF using Gotenberg's API. Determines the appropriate Gotenberg endpoint based on the file's MIME type. @@ -35,9 +145,23 @@ def convert_to_pdf(self, file_path, original_filename=None): 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() + mime_type, encoding = _detect_mime_type(file_path, original_filename) + file_ext = _detect_extension(file_path, original_filename, mime_type) logger.info(f"[{task_id}] Guessed MIME type for '{file_path}' is: {mime_type}, extension: {file_ext}") + if not mime_type and not file_ext: + log_task_progress( + task_id, + "detect_file_type", + "failure", + "Unable to determine file type for conversion", + detail=( + f"File: {file_path}\n" + f"Original filename: {original_filename or 'N/A'}\n" + "No extension and no detectable magic header." + ), + ) + logger.error(f"[{task_id}] Unable to determine file type for conversion: {file_path}") + return None log_task_progress(task_id, "detect_file_type", "success", f"File type: {mime_type or file_ext}") # Determine which Gotenberg endpoint to use @@ -91,7 +215,7 @@ def convert_to_pdf(self, file_path, original_filename=None): 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": (_build_filename(file_path, original_filename, file_ext), open(file_path, "rb"))} # Add some quality settings for better PDF output form_data = { @@ -176,7 +300,7 @@ def convert_to_pdf(self, file_path, original_filename=None): # 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": (_build_filename(file_path, original_filename, file_ext), open(file_path, "rb"))} logger.warning(f"Using fallback conversion for unknown type: {mime_type} / {file_ext}") if not endpoint: diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index 7279a9c8..44c0c179 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -7,6 +7,7 @@ import shutil import uuid import PyPDF2 # Replace fitz with PyPDF2 +from PyPDF2.errors import PdfReadError from app.celery_app import celery from app.config import settings @@ -112,25 +113,44 @@ def process_document(self, original_local_file: str, original_filename: str = No else: # Check for duplicate only if this is a new file (not reprocessing) # IMPORTANT: Only consider it a duplicate if it matches a DIFFERENT file - existing = db.query(FileRecord).filter_by(filehash=filehash).one_or_none() - + existing = db.query(FileRecord).filter(FileRecord.filehash == filehash, FileRecord.is_duplicate.is_(False)).order_by(FileRecord.created_at.asc()).first() + if existing is None: + existing = ( + db.query(FileRecord) + .filter(FileRecord.filehash == filehash) + .order_by(FileRecord.id.asc()) + .first() + ) + # A file is only a duplicate if it matches a different file's hash # (not its own hash when reprocessing) if existing and existing.id != file_id and settings.enable_deduplication: logger.info(f"[{task_id}] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.") - # Log the deduplication result without creating a new database record - # This avoids UNIQUE constraint violations on filehash + duplicate_record = FileRecord( + filehash=filehash, + original_filename=original_filename, + local_filename="", + file_size=file_size, + mime_type=mime_type, + is_duplicate=True, + duplicate_of_id=existing.id, + ) + db.add(duplicate_record) + db.commit() + db.refresh(duplicate_record) + if settings.enable_deduplication and settings.show_deduplication_step: log_task_progress( task_id, "check_for_duplicates", "success", f"Duplicate detected - matching file ID {existing.id}", - file_id=existing.id, + file_id=duplicate_record.id, detail=( f"Duplicate file detected.\n" f"File hash: {filehash}\n" f"Original file record ID: {existing.id}\n" + f"This file record ID: {duplicate_record.id}\n" f"Original filename: {original_filename}" ), ) @@ -139,7 +159,7 @@ def process_document(self, original_local_file: str, original_filename: str = No "process_document", "success", "Duplicate file detected, skipping", - file_id=existing.id, + file_id=duplicate_record.id, detail=( f"Duplicate file detected.\n" f"File hash: {filehash}\n" @@ -149,7 +169,7 @@ def process_document(self, original_local_file: str, original_filename: str = No ) return { "status": "duplicate_file", - "file_id": existing.id, + "file_id": duplicate_record.id, "original_file_id": existing.id, "detail": "File already processed.", } @@ -289,6 +309,30 @@ def process_document(self, original_local_file: str, original_filename: str = No process_with_azure_document_intelligence.delay(new_filename, file_id) return {"file": new_local_path, "status": "Queued for forced OCR", "file_id": file_id} + # If the file is not a PDF, skip embedded text check and convert to PDF first + is_pdf = mime_type == "application/pdf" or os.path.splitext(new_local_path)[1].lower() == ".pdf" + if not is_pdf: + logger.info(f"[{task_id}] Non-PDF file detected, queuing PDF conversion before OCR") + log_task_progress( + task_id, + "check_text", + "skipped", + "Non-PDF file detected, converting to PDF", + file_id=file_id, + ) + log_task_progress( + task_id, + "process_document", + "success", + "Queued for PDF conversion", + file_id=file_id, + ) + celery.send_task( + "app.tasks.convert_to_pdf.convert_to_pdf", + args=[new_local_path, original_filename], + ) + return {"file": new_local_path, "status": "Queued for PDF conversion", "file_id": file_id} + logger.info(f"[{task_id}] Checking for embedded text in PDF") log_task_progress( task_id, @@ -297,13 +341,34 @@ def process_document(self, original_local_file: str, original_filename: str = No "Checking for embedded text", file_id=file_id, ) - with open(new_local_path, "rb") as file: - pdf_reader = PyPDF2.PdfReader(file) - has_text = False - for page in pdf_reader.pages: - if page.extract_text().strip(): - has_text = True - break + try: + with open(new_local_path, "rb") as file: + pdf_reader = PyPDF2.PdfReader(file) + has_text = False + for page in pdf_reader.pages: + if page.extract_text().strip(): + has_text = True + break + except PdfReadError as exc: + logger.warning(f"[{task_id}] PDF read error during embedded text check: {exc}") + log_task_progress( + task_id, + "check_text", + "in_progress", + "PDF read error, retrying embedded text check", + file_id=file_id, + detail=str(exc), + ) + raise self.retry( + exc=exc, + countdown=10, + kwargs={ + "original_local_file": original_local_file, + "original_filename": original_filename, + "file_id": file_id, + "force_cloud_ocr": force_cloud_ocr, + }, + ) if has_text: logger.info(f"[{task_id}] PDF {original_local_file} contains embedded text. Processing locally.") diff --git a/app/tasks/upload_to_paperless.py b/app/tasks/upload_to_paperless.py index 73cc6a52..5f16d3fe 100644 --- a/app/tasks/upload_to_paperless.py +++ b/app/tasks/upload_to_paperless.py @@ -4,6 +4,7 @@ import json import logging import os import time +from typing import Optional import requests @@ -55,7 +56,23 @@ def normalize_metadata_value(value) -> str: return str_value -def poll_task_for_document_id(task_id: str) -> int: +def _is_duplicate_error(result_message: str) -> bool: + """ + Determine whether a Paperless task failure indicates a duplicate document. + + Args: + result_message: Failure result string from Paperless. + + Returns: + True if the message indicates a duplicate, otherwise False. + """ + if not result_message: + return False + lowered = result_message.lower() + return "duplicate" in lowered and "not consuming" in lowered + + +def poll_task_for_document_id(task_id: str) -> Optional[int]: """ Polls /api/tasks/?task_id= until we get status=SUCCESS or FAILURE, or until we run out of attempts. @@ -92,7 +109,11 @@ def poll_task_for_document_id(task_id: str) -> int: return int(doc_str) raise RuntimeError(f"Task {task_id} completed but no doc ID found. Task info: {task_info}") elif status == "FAILURE": - raise RuntimeError(f"Task {task_id} failed: {task_info.get('result')}") + result_message = task_info.get("result") + if _is_duplicate_error(result_message): + logger.info("Task %s reported duplicate document: %s", task_id, result_message) + return None + raise RuntimeError(f"Task {task_id} failed: {result_message}") attempts += 1 time.sleep(POLL_INTERVAL_SEC) @@ -266,6 +287,23 @@ def upload_to_paperless(self, file_path: str, file_id: int = None): 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) + if doc_id is None: + logger.info(f"[{task_id}] Paperless reported duplicate for {file_path}; skipping upload") + log_task_progress( + task_id, + "upload_to_paperless", + "skipped", + "Duplicate document detected by Paperless - skipping", + file_id=file_id, + detail="Paperless reported duplicate; document was not consumed.", + ) + return { + "status": "Duplicate", + "paperless_task_id": raw_task_id, + "paperless_document_id": None, + "file_path": file_path, + } + 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 diff --git a/app/utils/file_queries.py b/app/utils/file_queries.py index 1bb5c876..49397b0d 100644 --- a/app/utils/file_queries.py +++ b/app/utils/file_queries.py @@ -100,6 +100,9 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que query = query.filter(FileRecord.id.in_(db.query(subq.c.file_id))) elif status == "completed": # Files where all real steps are either success or skipped (no failures or in_progress) + # Exclude duplicates from completed + query = query.filter(FileRecord.is_duplicate.is_(False)) + # Get files that have real steps files_with_real_steps = real_steps_subq.distinct().subquery() @@ -115,5 +118,8 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que query = query.filter(FileRecord.id.in_(db.query(files_with_real_steps.c.file_id))).filter( ~FileRecord.id.in_(db.query(files_with_issues.c.file_id)) ) + elif status == "duplicate": + # Files marked as duplicates + query = query.filter(FileRecord.is_duplicate.is_(True)) return query diff --git a/app/utils/file_status.py b/app/utils/file_status.py index 88608ab1..a955b9e4 100644 --- a/app/utils/file_status.py +++ b/app/utils/file_status.py @@ -6,7 +6,7 @@ from typing import Dict, List from sqlalchemy.orm import Session -from app.models import FileProcessingStep, ProcessingLog +from app.models import FileProcessingStep, FileRecord, ProcessingLog from app.utils.step_manager import get_file_overall_status, get_step_summary @@ -23,6 +23,15 @@ def get_file_processing_status(db: Session, file_id: int) -> Dict: Returns: dict with status, last_step, and has_errors """ + file_record = db.query(FileRecord).filter(FileRecord.id == file_id).one_or_none() + if file_record and file_record.is_duplicate: + return { + "status": "duplicate", + "last_step": "check_for_duplicates", + "has_errors": False, + "total_steps": 0, + } + # Use the new status table approach overall_status = get_file_overall_status(db, file_id) @@ -61,6 +70,12 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D Returns: dict mapping file_id to status dict """ + # Preload duplicate flags for all files + duplicate_flags = { + record.id: record.is_duplicate + for record in db.query(FileRecord.id, FileRecord.is_duplicate).filter(FileRecord.id.in_(file_ids)).all() + } + # Define which steps are "real" status-determining steps from app.config import settings @@ -107,6 +122,14 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D # Compute status for each file result = {} for file_id in file_ids: + if duplicate_flags.get(file_id): + result[file_id] = { + "status": "duplicate", + "last_step": "check_for_duplicates", + "has_errors": False, + "total_steps": 0, + } + continue file_steps = steps_by_file.get(file_id, []) if not file_steps: result[file_id] = {"status": "pending", "last_step": None, "has_errors": False, "total_steps": 0} diff --git a/app/utils/step_manager.py b/app/utils/step_manager.py index aff9db6d..73a6426f 100644 --- a/app/utils/step_manager.py +++ b/app/utils/step_manager.py @@ -11,7 +11,7 @@ from typing import Dict, List, Optional from sqlalchemy.orm import Session from app.config import settings -from app.models import FileProcessingStep +from app.models import FileProcessingStep, FileRecord # Define the expected processing steps for a standard file workflow # The "check_for_duplicates" step is conditionally included based on enable_deduplication setting @@ -186,6 +186,19 @@ def get_file_overall_status(db: Session, file_id: int) -> Dict: "in_progress_steps": 2 } """ + # If file is marked as duplicate, return duplicate status immediately + file_record = db.query(FileRecord).filter(FileRecord.id == file_id).one_or_none() + if file_record and file_record.is_duplicate: + return { + "status": "duplicate", + "has_errors": False, + "total_steps": 0, + "completed_steps": 0, + "failed_steps": 0, + "in_progress_steps": 0, + "skipped_steps": 0, + } + # Define which steps are "real" status-determining steps # Only high-level logical steps, not implementation sub-steps REAL_MAIN_STEPS = { diff --git a/frontend/templates/file_detail.html b/frontend/templates/file_detail.html index 65008692..12b2cc18 100644 --- a/frontend/templates/file_detail.html +++ b/frontend/templates/file_detail.html @@ -90,6 +90,10 @@ background-color: #FEE2E2; color: #991B1B; } + .status-duplicate { + background-color: #E5E7EB; + color: #374151; + } /* Step summary section */ .step-summary { @@ -928,7 +932,9 @@
{% set main_completed = step_summary.main.success + step_summary.main.skipped %} {% set uploads_completed = step_summary.uploads.success + step_summary.uploads.skipped %} - {% if step_summary.main.failure > 0 or step_summary.uploads.failure > 0 %} + {% if file.is_duplicate %} + Duplicate + {% elif step_summary.main.failure > 0 or step_summary.uploads.failure > 0 %} Failed {% elif step_summary.main["in_progress"] > 0 or step_summary.uploads["in_progress"] > 0 %} Processing diff --git a/frontend/templates/files.html b/frontend/templates/files.html index e27e295b..3de7c881 100644 --- a/frontend/templates/files.html +++ b/frontend/templates/files.html @@ -159,6 +159,10 @@ background-color: #FEE2E2; color: #991B1B; } + .status-duplicate { + background-color: #E5E7EB; + color: #374151; + } /* Action buttons */ .action-btn { @@ -393,6 +397,7 @@ +
@@ -585,9 +590,6 @@ } function viewFileDetail(fileId, event) { - if (event && event.target.closest('.action-btn')) { - return; // Don't navigate if clicking action button - } if (event) event.stopPropagation(); window.location.href = `/files/${fileId}/detail`; } diff --git a/requirements.txt b/requirements.txt index f6786519..2fffd659 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,6 +8,8 @@ cryptography>=41.0.0 # Encryption for sensitive settings in database openai # GPT integration for metadata extraction PyPDF2>=3.0.0 # PDF processing for text extraction, metadata editing and rotation (replaces PyMuPDF) requests # HTTP client +puremagic>=1.25,<2.0 # File type detection (pure Python) +filetype>=1.2.0,<2.0 # File type detection fallback (pure Python) dropbox>=11.36.0 # Dropbox integration azure-ai-documentintelligence # Azure OCR service authlib>=1.6.5 # Authentication - fixed security vulnerabilities (GHSA-xxx)