diff --git a/app/config.py b/app/config.py index d827066b..e54d160f 100644 --- a/app/config.py +++ b/app/config.py @@ -179,6 +179,16 @@ class Settings(BaseSettings): description="Maximum size for a single file chunk in bytes. If set and file exceeds this, it will be split into smaller chunks for processing. Default: None (no splitting).", ) + # Deduplication settings - prevents processing of duplicate files + enable_deduplication: bool = Field( + default=True, + description="Enable deduplication check before processing. If enabled, files with the same SHA-256 hash as previously processed files will not be processed again. Default: True (enabled).", + ) + show_deduplication_step: bool = Field( + default=True, + description="Show the 'Check for Duplicates' step in processing history. If False, the check is still performed but not displayed. Default: True.", + ) + # Security Headers Configuration (see SECURITY_AUDIT.md and docs/DeploymentGuide.md) # Disabled by default since most deployments use a reverse proxy (Traefik, Nginx, etc.) # that already adds these headers. Enable if deploying directly without a reverse proxy. diff --git a/app/models.py b/app/models.py index 9ca3c859..90c74d86 100644 --- a/app/models.py +++ b/app/models.py @@ -1,6 +1,6 @@ # app/models.py -from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func +from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text, UniqueConstraint, func, Boolean from app.database import Base @@ -43,6 +43,13 @@ class FileRecord(Base): # MIME type or extension (optional) mime_type = Column(String) + + # Deduplication tracking: True if this file is a duplicate of another file + # When a duplicate is detected, this file record is created but marked as duplicate + is_duplicate = Column(Boolean, default=False, nullable=False, index=True) + + # If this is a duplicate, record the ID of the original file for reference + duplicate_of_id = Column(Integer, ForeignKey("files.id"), nullable=True) # Timestamp when we inserted this record created_at = Column(DateTime(timezone=True), server_default=func.now()) diff --git a/app/tasks/process_document.py b/app/tasks/process_document.py index d2d7b966..b43dbbf8 100644 --- a/app/tasks/process_document.py +++ b/app/tasks/process_document.py @@ -62,10 +62,15 @@ def process_document(self, original_local_file: str, original_filename: str = No ) 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) + # 0. Check for duplicate files (if enabled) + if settings.enable_deduplication: + logger.info(f"[{task_id}] Computing file hash for deduplication check...") + log_task_progress(task_id, "check_for_duplicates", "in_progress", "Computing file hash for deduplication") + filehash = hash_file(original_local_file) + else: + logger.info(f"[{task_id}] Computing file hash (deduplication disabled)...") + filehash = hash_file(original_local_file) + # Use provided original_filename or fall back to basename of path if original_filename is None: original_filename = os.path.basename(original_local_file) @@ -75,12 +80,15 @@ def process_document(self, original_local_file: str, original_filename: str = No 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", - ) + + # Log deduplication step result (only if enabled) + if settings.enable_deduplication: + log_task_progress( + task_id, + "check_for_duplicates", + "in_progress", + f"Hash: {filehash[:10]}..., checking for duplicates", + ) # Acquire DB session in the task with SessionLocal() as db: @@ -103,29 +111,66 @@ def process_document(self, original_local_file: str, original_filename: str = No new_record = existing_record else: existing = db.query(FileRecord).filter_by(filehash=filehash).one_or_none() - if existing: + if existing and settings.enable_deduplication: logger.info(f"[{task_id}] Duplicate file detected (hash={filehash[:10]}...) Skipping processing.") + # Create a file record for this duplicate with is_duplicate=True + 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=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}" + ), + ) log_task_progress( task_id, "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" - f"Existing file record ID: {existing.id}\n" + f"Original file record ID: {existing.id}\n" f"Original filename: {original_filename}" ), ) return { "status": "duplicate_file", - "file_id": existing.id, + "file_id": duplicate_record.id, + "original_file_id": existing.id, "detail": "File already processed.", } - # Not a duplicate -> insert a new record + # Not a duplicate (or deduplication disabled) -> insert a new record logger.info(f"[{task_id}] Creating new file record in database") + if settings.enable_deduplication and settings.show_deduplication_step: + log_task_progress( + task_id, + "check_for_duplicates", + "success", + "New file - no duplicates found", + ) log_task_progress(task_id, "create_file_record", "in_progress", "Creating file record") new_record = FileRecord( filehash=filehash, @@ -133,6 +178,7 @@ def process_document(self, original_local_file: str, original_filename: str = No local_filename="", # Will fill in after we move it file_size=file_size, mime_type=mime_type, + is_duplicate=False, ) db.add(new_record) db.commit() diff --git a/app/utils/file_queries.py b/app/utils/file_queries.py index 5c0457e7..1bb5c876 100644 --- a/app/utils/file_queries.py +++ b/app/utils/file_queries.py @@ -52,6 +52,8 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que # Define which steps are "real" status-determining steps # Only high-level logical steps and actual upload destinations (not queue_* steps) + from app.config import settings + REAL_STEPS = { "create_file_record", "check_text", @@ -73,6 +75,10 @@ def apply_status_filter(query: Query, db: Session, status: Optional[str]) -> Que "upload_to_email", "upload_to_s3", } + + # Add check_for_duplicates if deduplication is enabled + if settings.enable_deduplication: + REAL_STEPS.add("check_for_duplicates") # Filter to only real steps real_steps_subq = ( diff --git a/app/utils/file_status.py b/app/utils/file_status.py index bfdca39a..88608ab1 100644 --- a/app/utils/file_status.py +++ b/app/utils/file_status.py @@ -62,6 +62,8 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D dict mapping file_id to status dict """ # Define which steps are "real" status-determining steps + from app.config import settings + REAL_STEPS = { "create_file_record", "check_text", @@ -83,6 +85,10 @@ def get_files_processing_status(db: Session, file_ids: List[int]) -> Dict[int, D "upload_to_email", "upload_to_s3", } + + # Add check_for_duplicates if deduplication is enabled + if settings.enable_deduplication: + REAL_STEPS.add("check_for_duplicates") # Get all REAL steps for these files in one query steps = ( diff --git a/app/utils/step_manager.py b/app/utils/step_manager.py index 621a46bd..aff9db6d 100644 --- a/app/utils/step_manager.py +++ b/app/utils/step_manager.py @@ -10,11 +10,12 @@ from typing import Dict, List, Optional from sqlalchemy.orm import Session +from app.config import settings from app.models import FileProcessingStep # Define the expected processing steps for a standard file workflow -MAIN_PROCESSING_STEPS = [ - "hash_file", +# The "check_for_duplicates" step is conditionally included based on enable_deduplication setting +BASE_MAIN_PROCESSING_STEPS = [ "create_file_record", "check_text", "extract_text", # Or "process_with_azure_document_intelligence" @@ -24,6 +25,16 @@ MAIN_PROCESSING_STEPS = [ "send_to_all_destinations", ] +OPTIONAL_PROCESSING_STEPS = { + "check_for_duplicates": settings.enable_deduplication, # Only if deduplication is enabled +} + +# Combine steps based on configuration +MAIN_PROCESSING_STEPS = [] +if settings.enable_deduplication: + MAIN_PROCESSING_STEPS.append("check_for_duplicates") +MAIN_PROCESSING_STEPS.extend(BASE_MAIN_PROCESSING_STEPS) + def initialize_file_steps(db: Session, file_id: int, include_uploads: bool = False) -> None: """ @@ -188,6 +199,10 @@ def get_file_overall_status(db: Session, file_id: int) -> Dict: "send_to_all_destinations", } + # Add check_for_duplicates if deduplication is enabled + if settings.enable_deduplication: + REAL_MAIN_STEPS.add("check_for_duplicates") + all_steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all() # Filter to only real steps @@ -270,6 +285,10 @@ def get_step_summary(db: Session, file_id: int) -> Dict: "send_to_all_destinations", } + # Add check_for_duplicates if deduplication is enabled + if settings.enable_deduplication: + REAL_MAIN_STEPS.add("check_for_duplicates") + steps = db.query(FileProcessingStep).filter(FileProcessingStep.file_id == file_id).all() main_counts = {"queued": 0, "in_progress": 0, "success": 0, "failure": 0, "skipped": 0} diff --git a/app/views/files.py b/app/views/files.py index 2e5b165c..862bb2e3 100644 --- a/app/views/files.py +++ b/app/views/files.py @@ -217,7 +217,7 @@ def _compute_processing_flow(logs): """ # Define the main processing stages stages = { - "hash_file": {"label": "File Upload & Hash", "next": ["create_file_record"]}, + "check_for_duplicates": {"label": "Check for Duplicates", "next": ["create_file_record"]}, "create_file_record": {"label": "Create File Record", "next": ["check_text"]}, "check_text": { "label": "Check Embedded Text", @@ -233,6 +233,14 @@ def _compute_processing_flow(logs): "finalize_document_storage": {"label": "Finalize & Queue Distribution", "next": ["send_to_all_destinations"]}, "send_to_all_destinations": {"label": "Upload to Destinations", "next": [], "has_branches": True}, } + + # Filter out deduplication step if not enabled or if not showing it + from app.config import settings + if not settings.enable_deduplication or not settings.show_deduplication_step: + stages.pop("check_for_duplicates", None) + # Update the next pointer for create_file_record + if "create_file_record" in stages: + stages["create_file_record"]["next"] = ["check_text"] # Define upload sub-tasks (branches) upload_tasks = { @@ -345,9 +353,13 @@ def _compute_step_summary(logs): Note: This function is order-independent - it selects the latest status per step based on timestamp, regardless of input log ordering. """ + from app.config import settings + # Count statuses for main processing steps (not uploads) - main_steps = [ - "hash_file", + main_steps = [] + if settings.enable_deduplication and settings.show_deduplication_step: + main_steps.append("check_for_duplicates") + main_steps.extend([ "create_file_record", "check_text", "extract_text", @@ -356,7 +368,7 @@ def _compute_step_summary(logs): "embed_metadata_into_pdf", "finalize_document_storage", "send_to_all_destinations", - ] + ]) upload_prefixes = ["upload_to_", "queue_"] diff --git a/migrations/versions/003_add_deduplication_support.py b/migrations/versions/003_add_deduplication_support.py new file mode 100644 index 00000000..4a16906e --- /dev/null +++ b/migrations/versions/003_add_deduplication_support.py @@ -0,0 +1,45 @@ +"""Add deduplication support with is_duplicate and duplicate_of_id fields + +Revision ID: 003_add_deduplication_support +Revises: 002_add_file_paths +Create Date: 2026-02-12 + +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "003_add_deduplication_support" +down_revision: Union[str, None] = "002_add_file_paths" +depends_on: Union[str, None] = None + + +def upgrade() -> None: + """Add is_duplicate and duplicate_of_id columns to files table.""" + # Add is_duplicate column with default False + op.add_column("files", sa.Column("is_duplicate", sa.Boolean(), nullable=False, server_default="0")) + + # Add duplicate_of_id column as foreign key to self + op.add_column("files", sa.Column("duplicate_of_id", sa.Integer(), nullable=True)) + + # Create index on is_duplicate for efficient filtering + op.create_index("ix_files_is_duplicate", "files", ["is_duplicate"]) + + # Create foreign key relationship + op.create_foreign_key("fk_files_duplicate_of_id", "files", "files", ["duplicate_of_id"], ["id"]) + + +def downgrade() -> None: + """Remove is_duplicate and duplicate_of_id columns from files table.""" + # Drop foreign key + op.drop_constraint("fk_files_duplicate_of_id", "files", type_="foreignkey") + + # Drop index + op.drop_index("ix_files_is_duplicate", table_name="files") + + # Drop columns + op.drop_column("files", "duplicate_of_id") + op.drop_column("files", "is_duplicate")