diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7a60b85 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc diff --git a/automatic-pdf-renamer/app/config.py b/automatic-pdf-renamer/app/config.py new file mode 100644 index 0000000..f33cfb1 --- /dev/null +++ b/automatic-pdf-renamer/app/config.py @@ -0,0 +1,288 @@ +"""Configuration module for Automatic PDF Renamer. + +Provides environment variable parsing and template YAML loading. +""" + +import logging +import os +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + +logger = logging.getLogger(__name__) + + +@dataclass +class S3Config: + """S3 storage configuration.""" + + endpoint_url: str + access_key: str + secret_key: str + bucket_name: str + region: str = "us-east-1" + + @classmethod + def from_env(cls) -> "S3Config": + """Create S3Config from environment variables.""" + return cls( + endpoint_url=os.environ.get("S3_ENDPOINT_URL", ""), + access_key=os.environ.get("S3_ACCESS_KEY_ID", ""), + secret_key=os.environ.get("S3_SECRET_ACCESS_KEY", ""), + bucket_name=os.environ.get("S3_BUCKET_NAME", ""), + region=os.environ.get("S3_REGION", "us-east-1"), + ) + + def validate(self) -> None: + """Validate that all required fields are set.""" + if not self.endpoint_url: + raise ValueError("S3_ENDPOINT_URL environment variable is required") + if not self.access_key: + raise ValueError("S3_ACCESS_KEY_ID environment variable is required") + if not self.secret_key: + raise ValueError("S3_SECRET_ACCESS_KEY environment variable is required") + if not self.bucket_name: + raise ValueError("S3_BUCKET_NAME environment variable is required") + + +@dataclass +class PaperlessConfig: + """Paperless-ngx API configuration.""" + + url: str + token: str + timeout: int = 30 + + @classmethod + def from_env(cls) -> "PaperlessConfig": + """Create PaperlessConfig from environment variables.""" + return cls( + url=os.environ.get("PAPERLESS_URL", ""), + token=os.environ.get("PAPERLESS_TOKEN", ""), + timeout=int(os.environ.get("PAPERLESS_TIMEOUT", "30")), + ) + + def validate(self) -> None: + """Validate that all required fields are set.""" + if not self.url: + raise ValueError("PAPERLESS_URL environment variable is required") + if not self.token: + raise ValueError("PAPERLESS_TOKEN environment variable is required") + + +@dataclass +class PollerConfig: + """Inbox poller configuration.""" + + interval_seconds: int = 60 + inbox_prefix: str = "inbox/" + pending_prefix: str = "pending/" + files_prefix: str = "files/" + + @classmethod + def from_env(cls) -> "PollerConfig": + """Create PollerConfig from environment variables.""" + return cls( + interval_seconds=int(os.environ.get("POLLER_INTERVAL", "60")), + inbox_prefix=os.environ.get("INBOX_PREFIX", "inbox/"), + pending_prefix=os.environ.get("PENDING_PREFIX", "pending/"), + files_prefix=os.environ.get("FILES_PREFIX", "files/"), + ) + + +@dataclass +class ClassifierConfig: + """Classifier configuration.""" + + confidence_threshold: float = 0.8 + model_dir: str = "./models" + + @classmethod + def from_env(cls) -> "ClassifierConfig": + """Create ClassifierConfig from environment variables.""" + return cls( + confidence_threshold=float(os.environ.get("CONFIDENCE_THRESHOLD", "0.8")), + model_dir=os.environ.get("MODEL_DIR", "./models"), + ) + + +@dataclass +class AppConfig: + """Main application configuration.""" + + s3: S3Config = field(default_factory=S3Config) + paperless: PaperlessConfig | None = None + poller: PollerConfig = field(default_factory=PollerConfig) + classifier: ClassifierConfig = field(default_factory=ClassifierConfig) + debug: bool = False + sqlite_path: str = "./renamer.db" + + @classmethod + def from_env(cls) -> "AppConfig": + """Create AppConfig from environment variables.""" + debug = os.environ.get("DEBUG", "").lower() in ("true", "1", "yes") + + config = cls( + s3=S3Config.from_env(), + paperless=None, # Only needed for import script + poller=PollerConfig.from_env(), + classifier=ClassifierConfig.from_env(), + debug=debug, + sqlite_path=os.environ.get("SQLITE_PATH", "./renamer.db"), + ) + + # Only load paperless config if PAPERLESS_URL is set + if os.environ.get("PAPERLESS_URL"): + config.paperless = PaperlessConfig.from_env() + + return config + + def validate(self) -> None: + """Validate the configuration.""" + self.s3.validate() + if self.paperless: + self.paperless.validate() + + +@dataclass +class TemplateConfig: + """Template configuration from YAML.""" + + name: str + document_family: str + document_types: list[str] + period: dict[str, Any] + optional_fields: dict[str, Any] = field(default_factory=dict) + key_pattern: str = "" + target_prefix: str = "" + + @classmethod + def from_dict(cls, name: str, data: dict[str, Any]) -> "TemplateConfig": + """Create a TemplateConfig from a YAML dict.""" + return cls( + name=name, + document_family=data.get("document_family", name), + document_types=data.get("document_types", []), + period=data.get("period", {}), + optional_fields=data.get("optional_fields", {}), + key_pattern=data.get("key_pattern", ""), + target_prefix=data.get("target_prefix", ""), + ) + + +class TemplateRegistry: + """Registry for template configurations.""" + + def __init__(self): + self._templates: dict[str, TemplateConfig] = {} + + def load_from_file(self, filepath: str) -> None: + """Load templates from a YAML file. + + Args: + filepath: Path to the YAML file. + """ + with open(filepath, "r", encoding="utf-8") as f: + data = yaml.safe_load(f) + + if not data or not isinstance(data, dict): + logger.warning(f"No templates found in {filepath}") + return + + for name, config in data.items(): + template = TemplateConfig.from_dict(name, config) + self._templates[name] = template + logger.info(f"Loaded template: {name}") + + def load_from_directory(self, dirpath: str) -> None: + """Load templates from all YAML files in a directory. + + Args: + dirpath: Path to the directory containing YAML files. + """ + path = Path(dirpath) + if not path.exists(): + logger.warning(f"Template directory {dirpath} does not exist") + return + + for yaml_file in path.glob("*.yaml") + path.glob("*.yml"): + self.load_from_file(str(yaml_file)) + + def get(self, name: str) -> TemplateConfig | None: + """Get a template by name. + + Args: + name: The template name. + + Returns: + The TemplateConfig, or None if not found. + """ + return self._templates.get(name) + + def list_all(self) -> list[str]: + """List all template names. + + Returns: + List of template names. + """ + return list(self._templates.keys()) + + def get_period_rule(self, template_name: str) -> "PeriodRule" | None: + """Get the period rule for a template. + + Args: + template_name: The template name. + + Returns: + PeriodRule, or None if not configured. + """ + from .period import PeriodRule, PeriodGranularity, PeriodOffset + + template = self.get(template_name) + if not template or not template.period: + return None + + return PeriodRule( + granularity=PeriodGranularity(template.period.get("granularity", "month")), + offset=PeriodOffset(template.period.get("offset", "current")), + ) + + +def load_config() -> AppConfig: + """Load application configuration from environment variables. + + Returns: + AppConfig instance. + """ + return AppConfig.from_env() + + +def load_templates(template_path: str | None = None) -> TemplateRegistry: + """Load template configurations. + + Args: + template_path: Path to YAML file or directory. If None, uses TEMPLATE_PATH env var. + + Returns: + TemplateRegistry with loaded templates. + """ + registry = TemplateRegistry() + + if template_path is None: + template_path = os.environ.get("TEMPLATE_PATH", "./templates.yaml") + + if not template_path: + logger.warning("No template path configured") + return registry + + path = Path(template_path) + if path.is_file(): + registry.load_from_file(str(path)) + elif path.is_dir(): + registry.load_from_directory(str(path)) + else: + logger.warning(f"Template path {template_path} does not exist") + + return registry diff --git a/automatic-pdf-renamer/app/extractor.py b/automatic-pdf-renamer/app/extractor.py new file mode 100644 index 0000000..76682af --- /dev/null +++ b/automatic-pdf-renamer/app/extractor.py @@ -0,0 +1,175 @@ +"""Text extraction module for Automatic PDF Renamer. + +Provides PDF text extraction using PyMuPDF (PRD FR2.1, FR2.2). +This is the same extraction used by both the import script and the service. +""" + +import io +import logging +from typing import Any + +logger = logging.getLogger(__name__) + +# Minimum text length threshold (PRD FR2.3) +MIN_TEXT_LENGTH = 10 + + +class TextExtractionError(Exception): + """Error raised when text extraction fails.""" + pass + + +def extract_first_page_text(pdf_bytes: bytes) -> str: + """Extract text from the first page of a PDF. + + Uses PyMuPDF for fast text extraction (PRD stack decision). + Extracts only the first page as per FR2.2. + + Args: + pdf_bytes: The PDF file content as bytes. + + Returns: + The extracted text from the first page. + + Raises: + TextExtractionError: If extraction fails or PDF is invalid. + """ + try: + import fitz # PyMuPDF + except ImportError: + raise TextExtractionError( + "PyMuPDF is required for text extraction. Install with: pip install pymupdf" + ) + + try: + # Open PDF from bytes + doc = fitz.open(stream=pdf_bytes) + + if len(doc) == 0: + logger.warning("PDF has no pages") + return "" + + # Extract text from first page only (FR2.2) + first_page = doc[0] + text = first_page.get_text() + + doc.close() + return text + + except Exception as e: + logger.error(f"Failed to extract text from PDF: {e}") + raise TextExtractionError(f"Text extraction failed: {e}") from e + + +def extract_first_page_text_from_file(filepath: str) -> str: + """Extract text from the first page of a PDF file. + + Args: + filepath: Path to the PDF file. + + Returns: + The extracted text from the first page. + + Raises: + TextExtractionError: If extraction fails or PDF is invalid. + FileNotFoundError: If the file does not exist. + """ + try: + with open(filepath, "rb") as f: + pdf_bytes = f.read() + except FileNotFoundError: + raise FileNotFoundError(f"PDF file not found: {filepath}") + + return extract_first_page_text(pdf_bytes) + + +def is_text_empty(text: str, min_length: int = MIN_TEXT_LENGTH) -> bool: + """Check if extracted text is effectively empty. + + Considers text empty if: + - It's an empty string + - It's only whitespace + - It's shorter than min_length characters + + Args: + text: The text to check. + min_length: Minimum length threshold (default: 10). + + Returns: + True if text is considered empty, False otherwise. + """ + if not text: + return True + if not text.strip(): + return True + if len(text.strip()) < min_length: + return True + return False + + +def extract_and_validate(pdf_bytes: bytes, min_length: int = MIN_TEXT_LENGTH) -> tuple[str, bool]: + """Extract first page text and check if it's valid. + + Args: + pdf_bytes: The PDF file content as bytes. + min_length: Minimum text length threshold. + + Returns: + Tuple of (extracted_text, is_valid). + is_valid is False if text is empty or below min_length. + + Raises: + TextExtractionError: If extraction fails. + """ + text = extract_first_page_text(pdf_bytes) + is_valid = not is_text_empty(text, min_length) + return (text, is_valid) + + +class TextExtractor: + """Text extractor class for dependency injection. + + Wraps the extraction functions to allow for mocking in tests + and potential future swapping of extraction libraries. + """ + + def __init__(self, min_length: int = MIN_TEXT_LENGTH): + """Initialize the text extractor. + + Args: + min_length: Minimum text length threshold. + """ + self.min_length = min_length + + def extract(self, pdf_bytes: bytes) -> str: + """Extract first page text from PDF bytes. + + Args: + pdf_bytes: The PDF file content as bytes. + + Returns: + The extracted text. + """ + return extract_first_page_text(pdf_bytes) + + def extract_and_validate(self, pdf_bytes: bytes) -> tuple[str, bool]: + """Extract and validate text. + + Args: + pdf_bytes: The PDF file content as bytes. + + Returns: + Tuple of (text, is_valid). + """ + return extract_and_validate(pdf_bytes, self.min_length) + + def is_empty(self, text: str) -> bool: + """Check if text is empty. + + Args: + text: The text to check. + + Returns: + True if text is considered empty. + """ + return is_text_empty(text, self.min_length) diff --git a/automatic-pdf-renamer/app/period.py b/automatic-pdf-renamer/app/period.py new file mode 100644 index 0000000..0b386b1 --- /dev/null +++ b/automatic-pdf-renamer/app/period.py @@ -0,0 +1,484 @@ +"""Period computation module for Automatic PDF Renamer. + +Provides period parsing from filenames and computation from emission dates +per template rules (PRD section 4.1, FR4.2, FR4.3). +""" + +import re +from dataclasses import dataclass +from datetime import date, datetime +from enum import Enum +from typing import Any + +from dateutil.relativedelta import relativedelta + + +class PeriodGranularity(Enum): + """Granularity of period computation (PRD section 4.1).""" + + DAY = "day" + MONTH = "month" + QUARTER = "quarter" + YEAR = "year" + + +class PeriodOffset(Enum): + """Offset for period computation (PRD section 4.1).""" + + CURRENT = "current" + NEXT = "next" + + +@dataclass +class PeriodRule: + """Template period rule for period computation.""" + + granularity: PeriodGranularity + offset: PeriodOffset = PeriodOffset.CURRENT + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> "PeriodRule": + """Create a PeriodRule from a dict.""" + return cls( + granularity=PeriodGranularity(data.get("granularity", "month")), + offset=PeriodOffset(data.get("offset", "current")), + ) + + +# French month names mapping +_FRENCH_MONTHS = { + "janvier": 1, + "février": 2, + "mars": 3, + "avril": 4, + "mai": 5, + "juin": 6, + "juillet": 7, + "août": 8, + "septembre": 9, + "octobre": 10, + "novembre": 11, + "décembre": 12, + # English fallbacks + "january": 1, + "february": 2, + "march": 3, + "april": 4, + "may": 5, + "june": 6, + "july": 7, + "august": 8, + "september": 9, + "october": 10, + "november": 11, + "december": 12, +} + +# Month name to full French name for output +_FRENCH_MONTH_NAMES = { + 1: "janvier", 2: "février", 3: "mars", 4: "avril", 5: "mai", 6: "juin", + 7: "juillet", 8: "août", 9: "septembre", 10: "octobre", 11: "novembre", 12: "décembre", +} + +# Month abbreviation mapping +_MONTH_ABBR = { + 1: "Jan", 2: "Feb", 3: "Mar", 4: "Apr", 5: "May", 6: "Jun", + 7: "Jul", 8: "Aug", 9: "Sep", 10: "Oct", 11: "Nov", 12: "Dec", +} + +# Quarter pattern - matches Q3 2025, T3 2025, 3 2025, Q3-2025, trimestre 3 2025, 3T2025, etc. +_QUARTER_PATTERN = re.compile( + r"(?:Q|T|trimestre|quarter)?[\s\-]?(\d{1,2})[\s\-]?(?:Q|T|trimestre|quarter)?[\s\-]?(\d{4})\b|\b(\d{1,2})[TQ]?(\d{4})\b", + re.IGNORECASE, +) + +# Year pattern +_YEAR_PATTERN = re.compile(r"\b(\d{4})\b") + +# Month-year patterns (various formats) +_MONTH_YEAR_PATTERNS = [ + # "août 2025", "August 2025" - French/English month name + year + re.compile( + rf"\b([{''.join(_FRENCH_MONTHS.keys())}]+)\s+(\d{{4}})\b", + re.IGNORECASE, + ), + # "2025-08", "2025/08", "08-2025" + re.compile(r"(\d{4})[-/](\d{1,2})\b"), + re.compile(r"\b(\d{1,2})[-/](\d{4})\b"), + # "Aug 2025", "Aug-2025" + re.compile(r"\b([A-Za-z]{3})[-\s]?(\d{4})\b"), + # "2025 08", "08 2025" + re.compile(r"\b(\d{4})\s+(\d{1,2})\b"), + re.compile(r"\b(\d{1,2})\s+(\d{4})\b"), +] + +# Date pattern (YYYY-MM-DD, DD/MM/YYYY, etc.) +_DATE_PATTERN = re.compile( + r"(\d{4})[-/](\d{1,2})[-/](\d{1,2})|(\d{1,2})[-/](\d{1,2})[-/](\d{4})" +) + + +def _parse_french_month(month_str: str) -> int | None: + """Parse a French or English month name to month number.""" + month_lower = month_str.lower().strip() + return _FRENCH_MONTHS.get(month_lower) + + +def _parse_month_abbr(abbr: str) -> int | None: + """Parse a 3-letter month abbreviation to month number.""" + abbr_upper = abbr.upper() + for num, name in _MONTH_ABBR.items(): + if name.upper() == abbr_upper: + return num + return None + + +def parse_period_from_filename(filename: str) -> str | None: + """Parse a period string from a filename. + + Tries to extract period from the filename, looking at various positions. + Supports formats: + - Month-year: "août 2025", "August 2025", "Aug 2025", "2025-08" + - Quarter: "Q3 2025", "T3 2025", "3 2025", "Q3-2025" + - Year: "2025" + - Date: "2025-08-31", "31/08/2025" + + Args: + filename: The filename to parse (without path). + + Returns: + Parsed period string (e.g., "août 2025", "Q3 2025", "2025"), or None. + """ + # Get the filename without path + basename = filename.split("/")[-1].split("\\")[-1] + # Remove extension + name = basename.rsplit(".", 1)[0] + + # First, try to find quarter patterns anywhere in the filename + quarter_match = _QUARTER_PATTERN.search(name) + if quarter_match: + # Handle different group patterns + groups = quarter_match.groups() + quarter_num = None + year = None + for g in groups: + if g and g.isdigit() and len(g) <= 2: + q = int(g) + if 1 <= q <= 4 and quarter_num is None: + quarter_num = q + elif g and g.isdigit() and len(g) == 4: + year = int(g) + if quarter_num and year: + return format_quarter(quarter_num, year) + + # Try to find month-year patterns anywhere in the filename + for pattern in _MONTH_YEAR_PATTERNS: + match = pattern.search(name) + if match: + groups = match.groups() + year = None + month = None + + for g in groups: + if g and g.isdigit() and len(g) == 4: + year = int(g) + elif g and g.isdigit() and len(g) <= 2: + if month is None: + month = int(g) + elif g: + # Try as month name (French or English) + parsed = _parse_french_month(g) + if parsed: + month = parsed + else: + parsed = _parse_month_abbr(g) + if parsed: + month = parsed + + # If we found a year and a month, return it + if year is not None and month is not None: + return format_month_year(month, year) + + # Try to find date patterns (YYYY-MM-DD, DD/MM/YYYY, etc.) + date_match = _DATE_PATTERN.search(name) + if date_match: + groups = date_match.groups() + year = None + month = None + day = None + for g in groups: + if g and len(g) == 4: + year = int(g) + elif g and len(g) <= 2: + if day is None: + day = int(g) + else: + month = int(g) + if year and month: + return format_month_year(month, year) + elif year: + return str(year) + + # Try simple year pattern anywhere + year_match = _YEAR_PATTERN.search(name) + if year_match: + return year_match.group(1) + + return None + + +def compute_period_from_date( + emission_date: str | date, + rule: PeriodRule, +) -> str: + """Compute period from emission date and template rule. + + Args: + emission_date: The document emission date (ISO format YYYY-MM-DD or date object). + rule: PeriodRule with granularity and offset. + + Returns: + Human-readable period string. + """ + if isinstance(emission_date, str): + # Parse ISO date + try: + dt = datetime.fromisoformat(emission_date).date() + except ValueError: + # Try other formats + for fmt in ["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y", "%d-%m-%Y"]: + try: + dt = datetime.strptime(emission_date, fmt).date() + break + except ValueError: + continue + else: + raise ValueError(f"Cannot parse emission_date: {emission_date}") + else: + dt = emission_date + + if rule.granularity == PeriodGranularity.DAY: + return f"{dt.day} {_FRENCH_MONTH_NAMES[dt.month]} {dt.year}" + + elif rule.granularity == PeriodGranularity.MONTH: + if rule.offset == PeriodOffset.CURRENT: + return format_month_year(dt.month, dt.year) + else: # NEXT + next_month = dt + relativedelta(months=1) + return format_month_year(next_month.month, next_month.year) + + elif rule.granularity == PeriodGranularity.QUARTER: + quarter = (dt.month - 1) // 3 + 1 + if rule.offset == PeriodOffset.CURRENT: + return format_quarter(quarter, dt.year) + else: # NEXT + next_quarter = quarter + 1 + next_year = dt.year + if next_quarter > 4: + next_quarter = 1 + next_year += 1 + return format_quarter(next_quarter, next_year) + + elif rule.granularity == PeriodGranularity.YEAR: + if rule.offset == PeriodOffset.CURRENT: + return str(dt.year) + else: # NEXT + return str(dt.year + 1) + + else: + raise ValueError(f"Unknown granularity: {rule.granularity}") + + +def format_month_year(month: int, year: int, use_french: bool = True) -> str: + """Format month and year as a human-readable string. + + Uses French month names as per PRD by default. + + Args: + month: Month number (1-12). + year: Year number. + use_french: If True, use French month names; otherwise use English. + + Returns: + Formatted string (e.g., "août 2025" or "August 2025"). + """ + if month < 1 or month > 12: + raise ValueError(f"Invalid month: {month}") + if use_french: + month_name = _FRENCH_MONTH_NAMES[month] + else: + month_name = _MONTH_ABBR[month] + return f"{month_name} {year}" + + +def format_quarter(quarter: int, year: int) -> str: + """Format quarter and year as a human-readable string. + + Args: + quarter: Quarter number (1-4). + year: Year number. + + Returns: + Formatted string (e.g., "Q3 2025"). + """ + if quarter < 1 or quarter > 4: + raise ValueError(f"Invalid quarter: {quarter}") + return f"Q{quarter} {year}" + + +def get_period( + filename: str | None = None, + emission_date: str | date | None = None, + rule: PeriodRule | None = None, +) -> tuple[str, bool]: + """Get the period for a document, trying filename first, then fallback to rule. + + This is the main entry point for period computation. + + Args: + filename: The document filename (optional). + emission_date: The document emission date (optional). + rule: The template period rule (optional, required if emission_date provided). + + Returns: + Tuple of (period_string, is_from_filename). + If period is parsed from filename, is_from_filename is True. + If computed from rule, is_from_filename is False. + + Raises: + ValueError: If no period can be determined and no fallback is possible. + """ + # Try to parse from filename first + if filename: + period = parse_period_from_filename(filename) + if period: + return (period, True) + + # Fallback to rule-based computation + if emission_date and rule: + period = compute_period_from_date(emission_date, rule) + return (period, False) + + # If we have emission_date but no rule, use month-year as fallback + if emission_date: + if isinstance(emission_date, str): + try: + dt = datetime.fromisoformat(emission_date).date() + except ValueError: + for fmt in ["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y", "%d-%m-%Y"]: + try: + dt = datetime.strptime(emission_date, fmt).date() + break + except ValueError: + continue + else: + raise ValueError(f"Cannot parse emission_date: {emission_date}") + else: + dt = emission_date + return (format_month_year(dt.month, dt.year), False) + + raise ValueError("Cannot determine period: no filename, emission_date, or rule provided") + + +def validate_period_consistency( + period: str, + emission_date: str | date, + rule: PeriodRule, +) -> tuple[bool, str | None]: + """Validate that a period is consistent with emission date and template rule. + + Args: + period: The period string to validate. + emission_date: The document emission date. + rule: The template period rule. + + Returns: + Tuple of (is_consistent, error_message). + If consistent, error_message is None. + """ + if isinstance(emission_date, str): + try: + dt = datetime.fromisoformat(emission_date).date() + except ValueError: + for fmt in ["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y", "%d-%m-%Y"]: + try: + dt = datetime.strptime(emission_date, fmt).date() + break + except ValueError: + continue + else: + return (False, f"Cannot parse emission_date: {emission_date}") + else: + dt = emission_date + + # Compute expected period from rule + expected_period = compute_period_from_date(dt, rule) + + # Normalize both periods for comparison + # This is a simple comparison; may need enhancement for different formats + if period.lower() == expected_period.lower(): + return (True, None) + + # Try to parse both and compare + try: + parsed = _parse_period_string(period) + expected_parsed = _parse_period_string(expected_period) + if parsed == expected_parsed: + return (True, None) + except ValueError: + pass + + return ( + False, + f"Period '{period}' is inconsistent with emission_date '{emission_date}' " + f"and rule {rule.granularity.value}/{rule.offset.value}. " + f"Expected: {expected_period}", + ) + + +def _parse_period_string(period: str) -> tuple[int, int, int | None]: + """Parse a period string into (year, month, quarter) tuple. + + Returns: + (year, month, quarter) where month and quarter may be None. + """ + period_lower = period.lower() + + # Try year only + year_match = _YEAR_PATTERN.search(period_lower) + if year_match: + year = int(year_match.group(1)) + return (year, None, None) + + # Try quarter + quarter_match = _QUARTER_PATTERN.search(period_lower) + if quarter_match: + quarter = int(quarter_match.group(1)) + year = int(quarter_match.group(2)) + return (year, None, quarter) + + # Try month-year + for pattern in _MONTH_YEAR_PATTERNS: + match = pattern.search(period_lower) + if match: + groups = match.groups() + year = None + month = None + for g in groups: + if g and g.isdigit() and len(g) == 4: + year = int(g) + elif g and g.isdigit() and len(g) <= 2: + month = int(g) + elif g: + parsed_month = _parse_french_month(g) + if parsed_month: + month = parsed_month + else: + parsed_abbr = _parse_month_abbr(g) + if parsed_abbr: + month = parsed_abbr + + if year and month: + return (year, month, None) + + raise ValueError(f"Cannot parse period string: {period}") diff --git a/automatic-pdf-renamer/app/sidecar.py b/automatic-pdf-renamer/app/sidecar.py new file mode 100644 index 0000000..449ba76 --- /dev/null +++ b/automatic-pdf-renamer/app/sidecar.py @@ -0,0 +1,438 @@ +"""Sidecar module for Automatic PDF Renamer. + +Provides the sidecar data model, schema validation, atomic writes, and I/O helpers. +Sidecars are the source of truth for file metadata (PRD section 7). +""" + +from __future__ import annotations + +import json +from dataclasses import asdict, dataclass, field +from datetime import datetime, timezone +from enum import Enum +from typing import Any + +# Current schema version per PRD section 7 +SCHEMA_VERSION = "1.0" + + +class SidecarStatus(Enum): + """File processing status (PRD section 4).""" + + NOT_PROCESSED = "not processed" + PROCESSED = "processed" + VALIDATED = "validated" + + +class EventType(Enum): + """Event types logged in sidecar history.""" + + CLASSIFIED = "classified" + RENAMED = "renamed" + VALIDATED = "validated" + CORRECTED = "corrected" + IMPORTED = "imported" + REPROCESSED = "reprocessed" + + +@dataclass +class SidecarEvent: + """A single event in the sidecar history.""" + + type: str + payload: dict[str, Any] + timestamp: str + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> SidecarEvent: + return cls( + type=data["type"], + payload=data["payload"], + timestamp=data["timestamp"], + ) + + +@dataclass +class Sidecar: + """Sidecar metadata for a PDF file (PRD section 7).""" + + schema_version: str = SCHEMA_VERSION + sha256: str = "" + original_key: str = "" + current_key: str = "" + status: str = SidecarStatus.NOT_PROCESSED.value + template_name: str = "" + confidence: float = 0.0 + emission_date: str = "" + period: str = "" + emitting_company: str = "" + document_type: str = "" + optional_fields: dict[str, Any] = field(default_factory=dict) + extracted_text: str = "" + created_at: str = "" + updated_at: str = "" + events: list[SidecarEvent] = field(default_factory=list) + + def __post_init__(self): + if not self.created_at: + self.created_at = self._now_iso() + if not self.updated_at: + self.updated_at = self._now_iso() + if not isinstance(self.optional_fields, dict): + self.optional_fields = {} + if not isinstance(self.events, list): + self.events = [] + + @staticmethod + def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + def to_dict(self) -> dict[str, Any]: + data = asdict(self) + data["events"] = [e.to_dict() for e in self.events] + return data + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Sidecar: + events = [SidecarEvent.from_dict(e) for e in data.get("events", [])] + return cls( + schema_version=data.get("schema_version", SCHEMA_VERSION), + sha256=data.get("sha256", ""), + original_key=data.get("original_key", ""), + current_key=data.get("current_key", ""), + status=data.get("status", SidecarStatus.NOT_PROCESSED.value), + template_name=data.get("template_name", ""), + confidence=data.get("confidence", 0.0), + emission_date=data.get("emission_date", ""), + period=data.get("period", ""), + emitting_company=data.get("emitting_company", ""), + document_type=data.get("document_type", ""), + optional_fields=data.get("optional_fields", {}), + extracted_text=data.get("extracted_text", ""), + created_at=data.get("created_at", ""), + updated_at=data.get("updated_at", ""), + events=events, + ) + + def add_event(self, event_type: str, payload: dict[str, Any]) -> None: + """Append a timestamped event to the sidecar history.""" + self.events.append( + SidecarEvent( + type=event_type, + payload=payload, + timestamp=self._now_iso(), + ) + ) + self.updated_at = self._now_iso() + + def set_status(self, status: SidecarStatus) -> None: + """Update the sidecar status.""" + self.status = status.value + self.updated_at = self._now_iso() + + +# Required common fields per PRD section 4.1 +_REQUIRED_COMMON_FIELDS = { + "template_name", + "emission_date", + "emitting_company", + "document_type", + "period", +} + +# All required fields for a valid sidecar +_REQUIRED_SIDECAR_FIELDS = { + "schema_version", + "sha256", + "original_key", + "current_key", + "status", + *"template_name emission_date emitting_company document_type period".split(), + "created_at", + "updated_at", + "events", +} + + +def _check_field_presence(data: dict[str, Any], required: set[str]) -> list[str]: + """Return list of missing required fields.""" + return [f for f in required if f not in data] + + +def _check_types(data: dict[str, Any]) -> list[str]: + """Return list of fields with wrong types.""" + errors = [] + if "schema_version" in data and not isinstance(data["schema_version"], str): + errors.append("schema_version must be str") + if "sha256" in data and not isinstance(data["sha256"], str): + errors.append("sha256 must be str") + if "confidence" in data and not isinstance(data["confidence"], (int, float)): + errors.append("confidence must be numeric") + if "optional_fields" in data and not isinstance(data["optional_fields"], dict): + errors.append("optional_fields must be dict") + if "events" in data and not isinstance(data["events"], list): + errors.append("events must be list") + for i, event in enumerate(data.get("events", [])): + if not isinstance(event, dict): + errors.append(f"event[{i}] must be dict") + continue + if "type" not in event or not isinstance(event["type"], str): + errors.append(f"event[{i}].type must be str") + if "timestamp" not in event or not isinstance(event["timestamp"], str): + errors.append(f"event[{i}].timestamp must be str") + if "payload" not in event or not isinstance(event["payload"], dict): + errors.append(f"event[{i}].payload must be dict") + return errors + + +def validate_sidecar(data: dict[str, Any]) -> tuple[bool, list[str]]: + """Validate a sidecar dict against PRD schema. + + Returns: + tuple of (is_valid, list of error messages) + """ + errors: list[str] = [] + + # Check required fields + missing = _check_field_presence(data, _REQUIRED_SIDECAR_FIELDS) + if missing: + errors.append(f"Missing required fields: {', '.join(missing)}") + + # Check common fields presence (subset of required, but explicitly validated) + missing_common = _check_field_presence(data, _REQUIRED_COMMON_FIELDS) + if missing_common: + errors.append( + f"Missing required common fields: {', '.join(missing_common)}" + ) + + # Check types + type_errors = _check_types(data) + errors.extend(type_errors) + + # Check schema version + if data.get("schema_version") != SCHEMA_VERSION: + errors.append( + f"Unsupported schema_version: {data.get('schema_version')} " + f"(expected {SCHEMA_VERSION})" + ) + + # Check status is valid + valid_statuses = {s.value for s in SidecarStatus} + if data.get("status") not in valid_statuses: + errors.append( + f"Invalid status '{data.get('status')}'. " + f"Must be one of: {', '.join(valid_statuses)}" + ) + + return (len(errors) == 0, errors) + + +def build_sidecar( + sha256: str, + original_key: str, + current_key: str, + status: SidecarStatus, + template_name: str, + emission_date: str, + period: str, + emitting_company: str, + document_type: str, + confidence: float = 0.0, + optional_fields: dict[str, Any] | None = None, + extracted_text: str = "", +) -> Sidecar: + """Construct a new Sidecar with required fields. + + Args: + sha256: SHA256 hash of the PDF binary. + original_key: Original S3 key of the file (before any renames). + current_key: Current S3 key of the file. + status: Current processing status. + template_name: Document family/template name. + emission_date: Date the document was emitted (ISO format YYYY-MM-DD). + period: Human-readable period string. + emitting_company: Company that emitted the document. + document_type: Fine-grained document type within the template. + confidence: Classification confidence score (0.0 to 1.0). + optional_fields: Dict of template-specific optional fields. + extracted_text: First-page extracted text. + + Returns: + A new Sidecar instance. + """ + sidecar = Sidecar( + sha256=sha256, + original_key=original_key, + current_key=current_key, + status=status.value, + template_name=template_name, + confidence=confidence, + emission_date=emission_date, + period=period, + emitting_company=emitting_company, + document_type=document_type, + optional_fields=optional_fields or {}, + extracted_text=extracted_text, + ) + return sidecar + + +def serialize(sidecar: Sidecar, indent: int = 2) -> str: + """Serialize a Sidecar to JSON string. + + Args: + sidecar: The Sidecar to serialize. + indent: JSON indentation level. + + Returns: + JSON string representation. + """ + return json.dumps(sidecar.to_dict(), indent=indent, ensure_ascii=False, sort_keys=True) + + +def deserialize(json_str: str) -> Sidecar: + """Deserialize a JSON string to a Sidecar. + + Args: + json_str: JSON string of a sidecar. + + Returns: + Sidecar instance. + + Raises: + json.JSONDecodeError: If the JSON is invalid. + ValueError: If required fields are missing. + """ + data = json.loads(json_str) + return Sidecar.from_dict(data) + + +def get_sidecar_key(pdf_key: str) -> str: + """Return the sidecar key for a given PDF key. + + Args: + pdf_key: The S3 key of the PDF file. + + Returns: + The corresponding sidecar key (e.g., 'files/doc.pdf' -> 'files/doc.pdf.meta.json'). + """ + return f"{pdf_key}.meta.json" + + +# Default temp prefix for atomic writes +_DEFAULT_TEMP_PREFIX = "tmp/" + + +def write_atomic( + storage_client: Any, + sidecar: Sidecar, + target_key: str, + temp_prefix: str = _DEFAULT_TEMP_PREFIX, +) -> None: + """Write a sidecar atomically using temp-key + copy + delete pattern. + + This ensures crash safety: if the script dies mid-write, the original + sidecar (if any) remains intact, and the temp key can be cleaned up. + + Args: + storage_client: An object with put_object, copy_object, delete_object methods. + Expected interface: + - put_object(key: str, body: str | bytes) + - copy_object(src_key: str, dst_key: str) + - delete_object(key: str) + - object_exists(key: str) -> bool + sidecar: The Sidecar to write. + target_key: The final destination key (e.g., 'files/doc.pdf.meta.json'). + temp_prefix: Prefix for temporary keys (default: 'tmp/'). + + Raises: + RuntimeError: If any step fails. + """ + import uuid + + # Generate a unique temp key + temp_suffix = str(uuid.uuid4()) + temp_key = f"{temp_prefix}{temp_suffix}.meta.json" + + try: + # Step 1: Write to temp key + json_body = serialize(sidecar) + storage_client.put_object(temp_key, json_body) + + # Step 2: Copy temp key to target key (atomic in S3) + # If target exists, this overwrites it atomically + storage_client.copy_object(temp_key, target_key) + + # Step 3: Delete temp key + storage_client.delete_object(temp_key) + + except Exception as e: + # Clean up temp key if it exists + try: + if hasattr(storage_client, "object_exists") and storage_client.object_exists( + temp_key + ): + storage_client.delete_object(temp_key) + except Exception: + pass # Best effort cleanup + raise RuntimeError(f"Atomic sidecar write failed: {e}") from e + + +def read_sidecar(storage_client: Any, pdf_key: str) -> Sidecar | None: + """Read a sidecar from storage. + + Args: + storage_client: An object with get_object method. + Expected interface: + - get_object(key: str) -> str | bytes | None + pdf_key: The S3 key of the PDF file. + + Returns: + Sidecar if found, None otherwise. + """ + sidecar_key = get_sidecar_key(pdf_key) + try: + json_body = storage_client.get_object(sidecar_key) + if json_body is None: + return None + if isinstance(json_body, bytes): + json_body = json_body.decode("utf-8") + return deserialize(json_body) + except (json.JSONDecodeError, ValueError): + return None + + +def sidecar_exists(storage_client: Any, pdf_key: str) -> bool: + """Check if a sidecar exists for the given PDF key. + + Args: + storage_client: An object with object_exists method. + pdf_key: The S3 key of the PDF file. + + Returns: + True if sidecar exists, False otherwise. + """ + sidecar_key = get_sidecar_key(pdf_key) + return storage_client.object_exists(sidecar_key) + + +def matching_sha256(storage_client: Any, pdf_key: str, expected_sha: str) -> bool: + """Check if the sidecar for pdf_key has the expected SHA256. + + Used for resumability: skip documents already imported with matching hash. + + Args: + storage_client: An object with get_object method. + pdf_key: The S3 key of the PDF file. + expected_sha: The expected SHA256 hash. + + Returns: + True if sidecar exists and sha256 matches, False otherwise. + """ + sidecar = read_sidecar(storage_client, pdf_key) + if sidecar is None: + return False + return sidecar.sha256 == expected_sha diff --git a/automatic-pdf-renamer/app/storage/__init__.py b/automatic-pdf-renamer/app/storage/__init__.py new file mode 100644 index 0000000..cc2ef59 --- /dev/null +++ b/automatic-pdf-renamer/app/storage/__init__.py @@ -0,0 +1,119 @@ +"""Storage layer for Automatic PDF Renamer. + +Provides a unified interface for object storage operations. +Implementations: +- S3Storage: For Garage S3 (production) +- FileStorage: For local filesystem (testing) +""" + +from abc import ABC, abstractmethod +from typing import Any + + +class StorageBackend(ABC): + """Abstract base class for storage backends. + + All storage implementations must provide these methods. + """ + + @abstractmethod + def put_object(self, key: str, body: str | bytes) -> None: + """Upload an object to the storage. + + Args: + key: The object key/path. + body: The object content as string or bytes. + """ + pass + + @abstractmethod + def get_object(self, key: str) -> str | bytes | None: + """Retrieve an object from storage. + + Args: + key: The object key/path. + + Returns: + The object content as string or bytes, or None if not found. + """ + pass + + @abstractmethod + def delete_object(self, key: str) -> None: + """Delete an object from storage. + + Args: + key: The object key/path. + """ + pass + + @abstractmethod + def copy_object(self, src_key: str, dst_key: str) -> None: + """Copy an object within the storage. + + Args: + src_key: Source object key/path. + dst_key: Destination object key/path. + """ + pass + + @abstractmethod + def object_exists(self, key: str) -> bool: + """Check if an object exists in storage. + + Args: + key: The object key/path. + + Returns: + True if the object exists, False otherwise. + """ + pass + + @abstractmethod + def list_objects(self, prefix: str = "") -> list[str]: + """List all object keys under a prefix. + + Args: + prefix: The prefix to filter by. + + Returns: + List of object keys (full paths). + """ + pass + + @abstractmethod + def head_object(self, key: str) -> dict[str, Any] | None: + """Get metadata for an object. + + Args: + key: The object key/path. + + Returns: + Dict with metadata (e.g., size, etag, content_type), or None if not found. + """ + pass + + +# Convenience function to get the appropriate storage backend +# This will be configured by the application +def get_storage_backend(backend_type: str, **kwargs) -> StorageBackend: + """Factory function to get a storage backend instance. + + Args: + backend_type: Either 's3' or 'file'. + **kwargs: Backend-specific configuration. + + Returns: + A StorageBackend instance. + + Raises: + ValueError: If backend_type is not recognized. + """ + if backend_type == "s3": + from .s3_storage import S3Storage + return S3Storage(**kwargs) + elif backend_type == "file": + from .file_storage import FileStorage + return FileStorage(**kwargs) + else: + raise ValueError(f"Unknown storage backend type: {backend_type}") diff --git a/automatic-pdf-renamer/app/storage/file_storage.py b/automatic-pdf-renamer/app/storage/file_storage.py new file mode 100644 index 0000000..e910952 --- /dev/null +++ b/automatic-pdf-renamer/app/storage/file_storage.py @@ -0,0 +1,197 @@ +"""File Storage Backend for testing. + +Local filesystem implementation for development and testing purposes. +Mimics S3 behavior using a base directory. +""" + +import logging +import os +from pathlib import Path +from typing import Any + +from . import StorageBackend + +logger = logging.getLogger(__name__) + + +class FileStorage(StorageBackend): + """Filesystem-based storage backend for testing. + + Uses a base directory to store objects as files. + Keys are mapped to filesystem paths relative to the base directory. + """ + + def __init__(self, base_dir: str = "./storage"): + """Initialize file storage backend. + + Args: + base_dir: Base directory for storing objects. + """ + self.base_dir = Path(base_dir).resolve() + self.base_dir.mkdir(parents=True, exist_ok=True) + logger.info(f"FileStorage initialized at {self.base_dir}") + + def _make_path(self, key: str) -> Path: + """Convert a storage key to a filesystem path. + + Args: + key: The object key/path (may contain forward slashes). + + Returns: + Absolute Path for the object file. + """ + # Normalize key: remove leading slash, replace slashes with OS separator + normalized = key.lstrip("/") + if os.sep != "/": + normalized = normalized.replace("/", os.sep) + return self.base_dir / normalized + + def _ensure_parent_dir(self, path: Path) -> None: + """Ensure the parent directory of a path exists.""" + path.parent.mkdir(parents=True, exist_ok=True) + + def put_object(self, key: str, body: str | bytes) -> None: + """Upload an object to the filesystem. + + Args: + key: The object key/path. + body: The object content as string or bytes. + """ + path = self._make_path(key) + self._ensure_parent_dir(path) + + if isinstance(body, str): + body = body.encode("utf-8") + + with open(path, "wb") as f: + f.write(body) + logger.debug(f"Uploaded {key} to {path}") + + def get_object(self, key: str) -> str | bytes | None: + """Retrieve an object from the filesystem. + + Args: + key: The object key/path. + + Returns: + The object content as bytes, or None if not found. + """ + path = self._make_path(key) + + if not path.exists(): + logger.debug(f"Object {key} not found at {path}") + return None + + with open(path, "rb") as f: + return f.read() + + def delete_object(self, key: str) -> None: + """Delete an object from the filesystem. + + Args: + key: The object key/path. + """ + path = self._make_path(key) + + if path.exists(): + path.unlink() + logger.debug(f"Deleted {key} from {path}") + else: + logger.debug(f"Object {key} not found at {path}, nothing to delete") + + def copy_object(self, src_key: str, dst_key: str) -> None: + """Copy an object within the filesystem. + + Args: + src_key: Source object key/path. + dst_key: Destination object key/path. + """ + src_path = self._make_path(src_key) + dst_path = self._make_path(dst_key) + + self._ensure_parent_dir(dst_path) + + if not src_path.exists(): + raise FileNotFoundError(f"Source object {src_key} not found at {src_path}") + + # Use shutil.copy2 to preserve metadata + import shutil + shutil.copy2(str(src_path), str(dst_path)) + logger.debug(f"Copied {src_key} to {dst_key}") + + def object_exists(self, key: str) -> bool: + """Check if an object exists in the filesystem. + + Args: + key: The object key/path. + + Returns: + True if the object exists, False otherwise. + """ + path = self._make_path(key) + return path.exists() + + def list_objects(self, prefix: str = "") -> list[str]: + """List all object keys under a prefix. + + Args: + prefix: The prefix to filter by. + + Returns: + List of object keys (full paths, with forward slashes). + """ + # Convert prefix to path + prefix_path = self._make_path(prefix) + + # Ensure prefix_path is relative to base_dir + if not str(prefix_path).startswith(str(self.base_dir)): + raise ValueError(f"Prefix {prefix} escapes base directory") + + objects: list[str] = [] + + # Walk the directory tree starting from prefix_path + if prefix_path.exists() and prefix_path.is_dir(): + for root, _, files in os.walk(prefix_path): + for file in files: + full_path = Path(root) / file + # Convert back to storage key format (forward slashes, relative to base_dir) + rel_path = full_path.relative_to(self.base_dir) + key = str(rel_path).replace(os.sep, "/") + objects.append(key) + elif prefix_path.exists() and prefix_path.is_file(): + # If prefix is a file, return it + rel_path = prefix_path.relative_to(self.base_dir) + key = str(rel_path).replace(os.sep, "/") + objects.append(key) + + return objects + + def head_object(self, key: str) -> dict[str, Any] | None: + """Get metadata for an object. + + Args: + key: The object key/path. + + Returns: + Dict with metadata (size, etag, content_type, last_modified), or None if not found. + """ + path = self._make_path(key) + + if not path.exists(): + return None + + stat = path.stat() + + # Simple etag: use inode + size (not cryptographic, but sufficient for testing) + etag = f"{stat.st_ino}-{stat.st_size}" + + return { + "size": stat.st_size, + "etag": etag, + "content_type": "application/octet-stream", # Default for files + "last_modified": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(), + } + + +# Import datetime for head_object +from datetime import datetime, timezone diff --git a/automatic-pdf-renamer/app/storage/s3_storage.py b/automatic-pdf-renamer/app/storage/s3_storage.py new file mode 100644 index 0000000..4a0d9cc --- /dev/null +++ b/automatic-pdf-renamer/app/storage/s3_storage.py @@ -0,0 +1,258 @@ +"""S3 Storage Backend for Garage S3. + +Production storage implementation using boto3 for Garage S3 compatibility. +""" + +import logging +from typing import Any + +try: + import boto3 + from botocore.client import BaseClient + from botocore.exceptions import ClientError + HAS_BOTO3 = True +except ImportError: + HAS_BOTO3 = False + BaseClient = None + ClientError = Exception + +from . import StorageBackend + +logger = logging.getLogger(__name__) + + +class S3Storage(StorageBackend): + """S3-compatible storage backend for Garage. + + Uses boto3 to interact with Garage S3 API. + Configuration via environment variables or constructor kwargs. + """ + + def __init__( + self, + endpoint_url: str | None = None, + aws_access_key_id: str | None = None, + aws_secret_access_key: str | None = None, + bucket_name: str | None = None, + region_name: str | None = None, + ): + """Initialize S3 storage backend. + + Args: + endpoint_url: Garage S3 endpoint URL. + aws_access_key_id: S3 access key. + aws_secret_access_key: S3 secret key. + bucket_name: Name of the S3 bucket. + region_name: AWS region (default: us-east-1). + """ + self.endpoint_url = endpoint_url + self.aws_access_key_id = aws_access_key_id + self.aws_secret_access_key = aws_secret_access_key + self.bucket_name = bucket_name + self.region_name = region_name or "us-east-1" + + # Initialize boto3 client + self._client: BaseClient | None = None + self._connect() + + def _connect(self) -> None: + """Initialize the boto3 S3 client.""" + if not HAS_BOTO3: + raise ImportError("boto3 is required for S3Storage. Install with: pip install boto3") + self._client = boto3.client( + "s3", + endpoint_url=self.endpoint_url, + aws_access_key_id=self.aws_access_key_id, + aws_secret_access_key=self.aws_secret_access_key, + region_name=self.region_name, + ) + logger.info( + f"Connected to S3 at {self.endpoint_url} with bucket {self.bucket_name}" + ) + + def _ensure_bucket(self) -> None: + """Ensure the bucket is accessible.""" + if self.bucket_name is None: + raise ValueError("bucket_name is required for S3Storage") + + def _make_key(self, key: str) -> str: + """Normalize a key by removing leading slashes.""" + return key.lstrip("/") + + def put_object(self, key: str, body: str | bytes) -> None: + """Upload an object to S3. + + Args: + key: The object key/path. + body: The object content as string or bytes. + """ + self._ensure_bucket() + normalized_key = self._make_key(key) + + if isinstance(body, str): + body = body.encode("utf-8") + + try: + self._client.put_object( + Bucket=self.bucket_name, + Key=normalized_key, + Body=body, + ) + logger.debug(f"Uploaded {normalized_key} to bucket {self.bucket_name}") + except ClientError as e: + logger.error(f"Failed to upload {normalized_key}: {e}") + raise + + def get_object(self, key: str) -> str | bytes | None: + """Retrieve an object from S3. + + Args: + key: The object key/path. + + Returns: + The object content as bytes, or None if not found. + """ + self._ensure_bucket() + normalized_key = self._make_key(key) + + try: + response = self._client.get_object( + Bucket=self.bucket_name, + Key=normalized_key, + ) + return response["Body"].read() + except ClientError as e: + if e.response["Error"]["Code"] == "NoSuchKey": + logger.debug(f"Object {normalized_key} not found") + return None + logger.error(f"Failed to get {normalized_key}: {e}") + raise + + def delete_object(self, key: str) -> None: + """Delete an object from S3. + + Args: + key: The object key/path. + """ + self._ensure_bucket() + normalized_key = self._make_key(key) + + try: + self._client.delete_object( + Bucket=self.bucket_name, + Key=normalized_key, + ) + logger.debug(f"Deleted {normalized_key} from bucket {self.bucket_name}") + except ClientError as e: + logger.error(f"Failed to delete {normalized_key}: {e}") + raise + + def copy_object(self, src_key: str, dst_key: str) -> None: + """Copy an object within S3. + + Uses S3's copy_object operation which is atomic. + + Args: + src_key: Source object key/path. + dst_key: Destination object key/path. + """ + self._ensure_bucket() + src = self._make_key(src_key) + dst = self._make_key(dst_key) + + # Get the source bucket (for cross-bucket copy support) + # For now, assume same bucket + copy_source = {"Bucket": self.bucket_name, "Key": src} + + try: + self._client.copy_object( + Bucket=self.bucket_name, + Key=dst, + CopySource=copy_source, + ) + logger.debug(f"Copied {src} to {dst} in bucket {self.bucket_name}") + except ClientError as e: + logger.error(f"Failed to copy {src} to {dst}: {e}") + raise + + def object_exists(self, key: str) -> bool: + """Check if an object exists in S3. + + Args: + key: The object key/path. + + Returns: + True if the object exists, False otherwise. + """ + self._ensure_bucket() + normalized_key = self._make_key(key) + + try: + self._client.head_object( + Bucket=self.bucket_name, + Key=normalized_key, + ) + return True + except ClientError as e: + if e.response["Error"]["Code"] == "404": + return False + logger.error(f"Failed to check existence of {normalized_key}: {e}") + raise + + def list_objects(self, prefix: str = "") -> list[str]: + """List all object keys under a prefix. + + Args: + prefix: The prefix to filter by. + + Returns: + List of object keys (full paths, without leading slash). + """ + self._ensure_bucket() + normalized_prefix = self._make_key(prefix) + + objects: list[str] = [] + paginator = self._client.get_paginator("list_objects_v2") + + try: + for page in paginator.paginate( + Bucket=self.bucket_name, + Prefix=normalized_prefix, + ): + if "Contents" in page: + for obj in page["Contents"]: + # Return keys without leading slash + objects.append(obj["Key"].lstrip("/")) + return objects + except ClientError as e: + logger.error(f"Failed to list objects with prefix {normalized_prefix}: {e}") + raise + + def head_object(self, key: str) -> dict[str, Any] | None: + """Get metadata for an object. + + Args: + key: The object key/path. + + Returns: + Dict with metadata (size, etag, content_type, last_modified), or None if not found. + """ + self._ensure_bucket() + normalized_key = self._make_key(key) + + try: + response = self._client.head_object( + Bucket=self.bucket_name, + Key=normalized_key, + ) + return { + "size": response.get("ContentLength"), + "etag": response.get("ETag"), + "content_type": response.get("ContentType"), + "last_modified": response.get("LastModified"), + } + except ClientError as e: + if e.response["Error"]["Code"] == "404": + return None + logger.error(f"Failed to get metadata for {normalized_key}: {e}") + raise diff --git a/automatic-pdf-renamer/scripts/README.md b/automatic-pdf-renamer/scripts/README.md new file mode 100644 index 0000000..af6b52d --- /dev/null +++ b/automatic-pdf-renamer/scripts/README.md @@ -0,0 +1,279 @@ +# Scripts + +Standalone Python scripts for the Automatic PDF Renamer project. + +## Available Scripts + +| Script | Purpose | Dependencies | +|--------|---------|--------------| +| [`import_paperless.py`](import_paperless.py) | Import documents from Paperless-ngx to storage | requests, boto3, pymupdf, python-dateutil | + +--- + +## `import_paperless.py` - Paperless-ngx Initial Import + +**User Story 0**: Standalone script that imports existing paperless-ngx documents into a Garage S3 bucket (or local filesystem) with their metadata as sidecars. + +### Features + +- ✅ Connects to Paperless-ngx REST API +- ✅ Connects to Garage S3 or local filesystem +- ✅ Lists all documents via paginated API +- ✅ Maps Paperless metadata to sidecar fields +- ✅ Downloads PDFs and extracts first-page text +- ✅ Atomic sidecar writes (temp-key + copy pattern) +- ✅ Resumability (skip if already imported with matching SHA256) +- ✅ Dry-run mode for testing +- ✅ Summary reporting +- ✅ Proper exit codes + +### Requirements + +Install dependencies: + +```bash +pip install -r scripts/requirements.txt +``` + +Or individually: + +```bash +pip install requests boto3 pymupdf python-dateutil PyYAML +``` + +### Usage + +#### Environment Variables (Recommended) + +Set these before running: + +```bash +# Paperless-ngx + export PAPERLESS_URL="http://paperless:8000" + export PAPERLESS_TOKEN="your-api-token" + +# S3 Storage (Garage) + export S3_ENDPOINT_URL="http://garage:7777" + export S3_ACCESS_KEY_ID="your-access-key" + export S3_SECRET_ACCESS_KEY="your-secret-key" + export S3_BUCKET_NAME="pdf-renamer" + +# File Storage (for testing) + export FILE_BASE_DIR="./import_storage" +``` + +#### Command Line + +```bash +# S3 storage (default) +python scripts/import_paperless.py + +# File storage for testing +python scripts/import_paperless.py \ + --storage-backend file \ + --file-base-dir ./test_storage + +# Dry run (list what would be imported) +python scripts/import_paperless.py --dry-run + +# Limited import (test with 5 documents) +python scripts/import_paperless.py --limit 5 + +# Verbose mode +python scripts/import_paperless.py --verbose + +# Full options +python scripts/import_paperless.py \ + --paperless-url http://paperless:8000 \ + --paperless-token your-token \ + --storage-backend s3 \ + --s3-endpoint http://garage:7777 \ + --s3-access-key your-key \ + --s3-secret-key your-secret \ + --s3-bucket pdf-renamer \ + --dry-run \ + --limit 10 \ + --verbose +``` + +### Storage Backends + +The import script supports two storage backends: + +| Backend | Description | Use Case | +|---------|-------------|----------| +| `s3` | Garage S3 via boto3 | Production | +| `file` | Local filesystem | Testing, development | + +#### S3 Backend + +Requires: +- `--s3-endpoint` or `S3_ENDPOINT_URL` +- `--s3-access-key` or `S3_ACCESS_KEY_ID` +- `--s3-secret-key` or `S3_SECRET_ACCESS_KEY` +- `--s3-bucket` or `S3_BUCKET_NAME` + +The script will refuse to run if the `files/` prefix is not empty (to prevent accidental overwrites). + +#### File Backend + +Requires: +- `--file-base-dir` or `FILE_BASE_DIR` (default: `./import_storage`) + +Creates a directory structure matching the S3 layout: +``` +file_base_dir/ + files/ + Template1/ + document1.pdf + document1.pdf.meta.json + Template2/ + document2.pdf + document2.pdf.meta.json + tmp/ + (temporary files for atomic writes) +``` + +### Metadata Mapping + +Paperless-ngx fields are mapped to sidecar fields as follows: + +| Sidecar Field | Paperless Source | Notes | +|---------------|------------------|-------| +| `template_name` | `storage_path` (first segment) | Document family | +| `emission_date` | `created` | Document date | +| `emitting_company` | `correspondent.name` | Correspondent | +| `document_type` | `document_type.name` | Document type | +| `period` | Filename (parsed) or `created` (fallback) | Month-year, quarter, year | +| `extracted_text` | PDF first page | Via PyMuPDF | +| `optional_fields` | `labels` | Each label becomes an optional field | +| `sha256` | PDF binary | Computed from downloaded PDF | +| `status` | Hardcoded | `validated` | +| `confidence` | Hardcoded | `1.0` | + +### Output Structure + +Documents are stored at: +``` +files/{template}/{original_filename} +files/{template}/{original_filename}.meta.json +``` + +Example: +``` +files/Banque/Relevé de compte août 2025.pdf +files/Banque/Relevé de compte août 2025.pdf.meta.json +``` + +### Resumability + +The script can be interrupted and rerun safely: +- Documents already imported with matching SHA256 are skipped +- Atomic sidecar writes ensure no partial/corrupted sidecars +- Summary shows imported, skipped, and failed counts + +### Exit Codes + +| Code | Meaning | +|------|---------| +| 0 | Success (all documents imported) | +| 1 | Failure (one or more documents failed) | +| 130 | Interrupted by user (SIGINT) | + +### Example Output + +``` +2026-10-02 12:00:00,000 - INFO - Starting Paperless-ngx import... +2026-10-02 12:00:00,001 - INFO - Using S3 storage backend: http://garage:7777/pdf-renamer +2026-10-02 12:00:00,002 - INFO - Fetching documents from Paperless... +2026-10-02 12:00:05,123 - INFO - Retrieved 150 documents from Paperless +2026-10-02 12:00:05,124 - INFO - Processing document 1: Relevé août 2025.pdf +2026-10-02 12:00:05,456 - INFO - Successfully imported document 1 +2026-10-02 12:00:05,457 - INFO - Processing document 2: Facture Q3 2025.pdf +... + +============================================================ +IMPORT SUMMARY +============================================================ +Storage backend: s3 +Imported: 150 +Skipped: 0 +Failed: 0 +Period fallbacks: 0 + +Per-template counts: + Banque: 75 + Factures: 50 + Contrats: 25 +============================================================ +``` + +### Troubleshooting + +#### "Missing required values" error + +Make sure all required environment variables are set or passed as CLI arguments: + +```bash +# Check what's set +echo $PAPERLESS_URL +echo $S3_ENDPOINT_URL +``` + +#### "Bucket is not empty" error + +The script refuses to run if `files/` prefix already has objects. Options: + +1. **Use a different bucket** +2. **Use file storage for testing**: `--storage-backend file --file-base-dir ./test` +3. **Empty the bucket first** (if you're sure it's safe) + +#### "PyMuPDF not available" warning + +Text extraction will be skipped. Install it for full functionality: + +```bash +pip install pymupdf +``` + +#### Connection errors + +Check that: +- Paperless-ngx is running and accessible +- S3/Garage endpoint is correct +- Credentials are valid +- Network connectivity is available + +### Development / Testing + +For testing without Paperless-ngx or S3: + +```bash +# Use file storage +python scripts/import_paperless.py \ + --storage-backend file \ + --file-base-dir ./test_storage \ + --dry-run + +# Or create a mock Paperless client (see tests/ directory) +``` + +--- + +## Architecture + +The import script uses the same reusable components as the main service: + +- `app/sidecar.py` - Sidecar data model and I/O +- `app/storage/` - Storage backend abstraction +- `app/period.py` - Period parsing and computation +- `app/extractor.py` - PDF text extraction +- `app/config.py` - Configuration utilities + +This ensures consistency between the import script and the main service. + +--- + +## License + +Same as the main project. See [../LICENSE](../LICENSE). diff --git a/automatic-pdf-renamer/scripts/import_paperless.py b/automatic-pdf-renamer/scripts/import_paperless.py new file mode 100755 index 0000000..8be38d2 --- /dev/null +++ b/automatic-pdf-renamer/scripts/import_paperless.py @@ -0,0 +1,919 @@ +#!/usr/bin/env python3 +"""Paperless-ngx Initial Import Script. + +User Story 0: Standalone script to import existing paperless-ngx documents +into a Garage S3 bucket with their metadata as sidecars. + +This script runs once, offline, from the owner's machine or the server. +It is NOT part of the renamer container and has no dashboard surface. + +Storage Backends: +- S3: For production Garage S3 (default) +- File: For local filesystem testing + +Note: Paperless-ngx API returns IDs for correspondent, document_type, and tags. +This script fetches the actual string values from the respective endpoints. +The filename comes from 'archived_file_name', not 'filename'. +""" + +import argparse +import hashlib +import logging +import os +import sys +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +import requests + +# Add parent directory to path for imports +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from app.sidecar import ( + Sidecar, + SidecarStatus, + build_sidecar, + get_sidecar_key, + matching_sha256, + read_sidecar, + sidecar_exists, + write_atomic, +) +from app.storage import get_storage_backend, StorageBackend + +# Configure logging +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(levelname)s - %(message)s", +) +logger = logging.getLogger(__name__) + + +@dataclass +class PaperlessDocument: + """Represents a document from Paperless-ngx API.""" + + id: int + filename: str + storage_path: str + created: str # Document date (emission date) + correspondent_name: str + document_type_name: str + labels: list[str] # List of label names + download_url: str + + +@dataclass +class ImportStats: + """Statistics for the import run.""" + + imported: int = 0 + skipped: int = 0 + failed: int = 0 + period_fallbacks: int = 0 + per_template_counts: dict[str, int] = field(default_factory=dict) + errors: list[str] = field(default_factory=list) + + +class PaperlessClient: + """Client for Paperless-ngx REST API. + + Note: Paperless returns IDs for correspondent, document_type, and tags. + This client fetches the actual string values from the respective endpoints. + """ + + def __init__(self, base_url: str, token: str, timeout: int = 30): + """Initialize the Paperless client. + + Args: + base_url: Paperless-ngx base URL (e.g., http://paperless:8000). + token: API authentication token. + timeout: Request timeout in seconds. + """ + self.base_url = base_url.rstrip("/") + self.token = token + self.timeout = timeout + self.session = requests.Session() + self.session.headers.update({ + "Authorization": f"Token {self.token}", + "Accept": "application/json", + }) + + # Cache for fetched entities to avoid duplicate requests + self._storage_path_cache: dict[int, str] = {} + self._correspondent_cache: dict[int, str] = {} + self._document_type_cache: dict[int, str] = {} + self._tag_cache: dict[int, str] = {} + + def _get(self, endpoint: str, params: dict | None = None) -> dict[str, Any]: + """Make a GET request to the Paperless API.""" + url = f"{self.base_url}{endpoint}" + try: + response = self.session.get(url, params=params, timeout=self.timeout) + response.raise_for_status() + return response.json() + except requests.exceptions.RequestException as e: + logger.error(f"GET {url} failed: {e}") + raise + + def _get_raw(self, endpoint: str) -> bytes: + """Make a GET request and return raw bytes.""" + url = f"{self.base_url}{endpoint}" + try: + response = self.session.get(url, timeout=self.timeout) + response.raise_for_status() + return response.content + except requests.exceptions.RequestException as e: + logger.error(f"GET {url} failed: {e}") + raise + + def list_documents(self, page: int = 1, page_size: int = 100) -> list[dict[str, Any]]: + """List all documents from Paperless (paginated). + + Args: + page: Page number. + page_size: Number of results per page. + + Returns: + List of document dicts. + """ + data = self._get("/api/documents/", params={"page": page, "page_size": page_size}) + return data.get("results", []) + + def get_document(self, doc_id: int) -> dict[str, Any]: + """Get a single document's metadata. + + Args: + doc_id: The document ID. + + Returns: + Document metadata dict. + """ + return self._get(f"/api/documents/{doc_id}/") + + def get_storage_path_name(self, storage_path_id: int) -> str: + """Get storage path name from ID. + + Args: + storage_path_id: The Paperless storage path ID. + + Returns: + Storage path name string. + """ + if not storage_path_id: + return "" + if storage_path_id in self._storage_path_cache: + return self._storage_path_cache[storage_path_id] + try: + storage_path = self._get(f"/api/storage_paths/{storage_path_id}/") + name = storage_path.get("name", f"unknown_storage_path_{storage_path_id}") + self._storage_path_cache[storage_path_id] = name + return name + except Exception as e: + logger.warning(f"Failed to fetch storage path {storage_path_id}: {e}") + return f"unknown_storage_path_{storage_path_id}" + + def get_correspondent_name(self, correspondent_id: int) -> str: + """Get correspondent name from ID. + + Args: + correspondent_id: The Paperless correspondent ID. + + Returns: + Correspondent name string. + """ + if not correspondent_id: + return "" + if correspondent_id in self._correspondent_cache: + return self._correspondent_cache[correspondent_id] + try: + correspondent = self._get(f"/api/correspondents/{correspondent_id}/") + name = correspondent.get("name", f"unknown_correspondent_{correspondent_id}") + self._correspondent_cache[correspondent_id] = name + return name + except Exception as e: + logger.warning(f"Failed to fetch correspondent {correspondent_id}: {e}") + return f"unknown_correspondent_{correspondent_id}" + + def get_document_type_name(self, document_type_id: int) -> str: + """Get document type name from ID. + + Args: + document_type_id: The Paperless document type ID. + + Returns: + Document type name string. + """ + if not document_type_id: + return "" + if document_type_id in self._document_type_cache: + return self._document_type_cache[document_type_id] + try: + doc_type = self._get(f"/api/document_types/{document_type_id}/") + name = doc_type.get("name", f"unknown_type_{document_type_id}") + self._document_type_cache[document_type_id] = name + return name + except Exception as e: + logger.warning(f"Failed to fetch document type {document_type_id}: {e}") + return f"unknown_type_{document_type_id}" + + def get_tag_name(self, tag_id: int) -> str: + """Get tag name from ID. + + Args: + tag_id: The Paperless tag ID. + + Returns: + Tag name string. + """ + if not tag_id: + return "" + if tag_id in self._tag_cache: + return self._tag_cache[tag_id] + try: + tag = self._get(f"/api/tags/{tag_id}/") + name = tag.get("name", f"unknown_tag_{tag_id}") + self._tag_cache[tag_id] = name + return name + except Exception as e: + logger.warning(f"Failed to fetch tag {tag_id}: {e}") + return f"unknown_tag_{tag_id}" + + def get_all_documents(self, limit: int = None) -> list[PaperlessDocument]: + """Get all documents from Paperless (handles pagination). + + Note: Paperless returns IDs for correspondent, document_type, and tags. + This method fetches the actual string values from the respective endpoints. + + Returns: + List of PaperlessDocument objects. + """ + all_docs: list[PaperlessDocument] = [] + page = 1 + page_size = 100 + + while True: + docs = self.list_documents(page=page, page_size=page_size) + if not docs: + break + + for doc in docs: + # Get full document details + full_doc = self.get_document(doc["id"]) + + # Extract fields from Paperless document + # The actual filename is in 'archived_file_name', not 'filename' + archived_file_name = full_doc.get("archived_file_name", "") if full_doc.get("archived_file_name") else full_doc.get("title", "") + ".pdf" + + # Get correspondent name from ID + correspondent_id = full_doc.get("correspondent") + correspondent_name = self.get_correspondent_name(correspondent_id) if correspondent_id else "" + + # Get document type name from ID + document_type_id = full_doc.get("document_type") + document_type_name = self.get_document_type_name(document_type_id) if document_type_id else "" + + # Get storage path + storage_path_id = full_doc.get("storage_path", "") + storage_path = self.get_storage_path_name(storage_path_id) if storage_path_id else "" + + # Get tags/labels - Paperless returns tag IDs, need to fetch names + tag_ids = full_doc.get("tags", []) + labels = [] + for tag_id in tag_ids: + tag_name = self.get_tag_name(tag_id) + if tag_name: + labels.append(tag_name) + + paperless_doc = PaperlessDocument( + id=full_doc["id"], + filename=archived_file_name, + storage_path=storage_path, + created=full_doc.get("created", ""), # This is the document date + correspondent_name=correspondent_name, + document_type_name=document_type_name, + labels=labels, + download_url=f"{self.base_url}/api/documents/{full_doc['id']}/download/", + ) + logger.debug(f"Retrieved document: {paperless_doc}") + all_docs.append(paperless_doc) + + if limit and len(all_docs) >= limit: + logger.info(f"Reached limit of {limit} documents") + return all_docs + + # Check if there are more pages + if len(docs) < page_size: + break + page += 1 + + logger.info(f"Retrieved {len(all_docs)} documents from Paperless") + return all_docs + + def download_document(self, doc_id: int) -> bytes: + """Download a document's PDF. + + Args: + doc_id: The document ID. + + Returns: + PDF content as bytes. + """ + return self._get_raw(f"/api/documents/{doc_id}/download/") + + +class PDFProcessor: + """Processes PDF content for import.""" + + def __init__(self): + """Initialize the PDF processor.""" + # Import extractor (optional dependency) + try: + from app.extractor import extract_first_page_text + + self._extract_text = extract_first_page_text + except ImportError: + logger.warning("PyMuPDF not available, text extraction will be skipped") + self._extract_text = None + + def compute_sha256(self, pdf_bytes: bytes) -> str: + """Compute SHA256 hash of PDF content. + + Args: + pdf_bytes: The PDF content. + + Returns: + Hex-encoded SHA256 hash. + """ + return hashlib.sha256(pdf_bytes).hexdigest() + + def extract_text(self, pdf_bytes: bytes) -> str: + """Extract first page text from PDF. + + Args: + pdf_bytes: The PDF content. + + Returns: + Extracted text, or empty string if extraction fails. + """ + if self._extract_text is None: + logger.warning("Text extraction not available") + return "" + + try: + return self._extract_text(pdf_bytes) + except Exception as e: + logger.warning(f"Text extraction failed: {e}") + return "" + + +class SidecarBuilder: + """Builds sidecar metadata from Paperless documents.""" + + def __init__(self, template_mappings: dict[str, str] | None = None): + """Initialize the sidecar builder. + + Args: + template_mappings: Optional dict mapping Paperless storage_path + prefixes to template names. + """ + self.template_mappings = template_mappings or {} + + def get_template_name(self, storage_path: str) -> str: + """Get template name from storage path. + + Uses the first path segment as the document family (template). + + Args: + storage_path: The Paperless storage path. + + Returns: + Template name (first segment of storage path). + """ + if not storage_path: + return "unknown" + + # Get first path segment + parts = storage_path.strip("/").split("/") + if parts: + return parts[0] + return "unknown" + + def parse_period_from_filename(self, filename: str) -> str | None: + """Parse period from filename. + + Args: + filename: The document filename. + + Returns: + Period string, or None if cannot be parsed. + """ + try: + from app.period import parse_period_from_filename + + return parse_period_from_filename(filename) + except Exception: + return None + + def build_sidecar( + self, + doc: PaperlessDocument, + pdf_bytes: bytes, + sha256: str, + extracted_text: str, + template_name: str | None = None, + ) -> Sidecar: + """Build a sidecar from a Paperless document. + + Args: + doc: The Paperless document. + pdf_bytes: The PDF content. + sha256: The SHA256 hash of the PDF. + extracted_text: The extracted first-page text. + template_name: Optional template name override. + + Returns: + A Sidecar object with all metadata. + """ + # Determine template + if template_name: + final_template = template_name + else: + final_template = self.get_template_name(doc.storage_path) + + # Parse period from filename + period = self.parse_period_from_filename(doc.filename) + if not period: + # Fallback: use emission date's month-year + from datetime import datetime + + try: + dt = datetime.fromisoformat(doc.created).date() + from app.period import format_month_year + + period = format_month_year(dt.month, dt.year) + logger.warning(f"Using fallback period for doc {doc.id}: {period}") + except Exception: + period = doc.created # Last resort + + # Build optional fields from labels + optional_fields = {} + for label in doc.labels: + optional_fields["tag"] = label + + # Construct the original key + if doc.storage_path: + original_key = f"{doc.storage_path}/{doc.filename}" + else: + original_key = doc.filename + + # Current key is the same as original for import (no renames) + current_key = f"files/{final_template}/{doc.filename}" + + # Build the sidecar + sidecar = build_sidecar( + sha256=sha256, + original_key=original_key, + current_key=current_key, + status=SidecarStatus.VALIDATED, + template_name=final_template, + emission_date=doc.created, + period=period, + emitting_company=doc.correspondent_name, + document_type=doc.document_type_name, + confidence=1.0, + optional_fields=optional_fields, + extracted_text=extracted_text, + ) + + # Add imported event + sidecar.add_event( + "imported", + { + "source": "paperless-ngx", + "document_id": doc.id, + "document_url": doc.download_url, + }, + ) + + return sidecar + + +class Importer: + """Main import class that orchestrates the import process.""" + + def __init__( + self, + paperless_url: str, + paperless_token: str, + storage_backend: str = "s3", + s3_endpoint: str | None = None, + s3_access_key: str | None = None, + s3_secret_key: str | None = None, + s3_bucket: str | None = None, + file_base_dir: str | None = None, + dry_run: bool = False, + limit: int | None = None, + template_mappings: dict[str, str] | None = None, + ): + """Initialize the importer. + + Args: + paperless_url: Paperless-ngx API URL. + paperless_token: Paperless-ngx API token. + storage_backend: Storage backend type ('s3' or 'file'). + s3_endpoint: Garage S3 endpoint URL (required for S3). + s3_access_key: S3 access key (required for S3). + s3_secret_key: S3 secret key (required for S3). + s3_bucket: S3 bucket name (required for S3). + file_base_dir: Base directory for file storage (required for file backend). + dry_run: If True, only list what would be imported. + limit: Maximum number of documents to import (for testing). + template_mappings: Optional dict mapping storage_path to template. + """ + self.paperless_url = paperless_url + self.paperless_token = paperless_token + self.storage_backend_type = storage_backend + self.s3_endpoint = s3_endpoint + self.s3_access_key = s3_access_key + self.s3_secret_key = s3_secret_key + self.s3_bucket = s3_bucket + self.file_base_dir = file_base_dir + self.dry_run = dry_run + self.limit = limit + self.template_mappings = template_mappings + + # Initialize clients + self.paperless_client = PaperlessClient(paperless_url, paperless_token) + self.pdf_processor = PDFProcessor() + self.sidecar_builder = SidecarBuilder(template_mappings) + + # Initialize storage - will be set up in _init_storage() + self.storage: StorageBackend | None = None + self._init_storage() + + # Stats + self.stats = ImportStats() + + def _init_storage(self) -> None: + """Initialize the storage backend based on configuration.""" + if self.storage_backend_type == "s3": + if not all([self.s3_endpoint, self.s3_access_key, self.s3_secret_key, self.s3_bucket]): + raise ValueError( + "S3 storage requires: s3_endpoint, s3_access_key, s3_secret_key, s3_bucket" + ) + self.storage = get_storage_backend( + "s3", + endpoint_url=self.s3_endpoint, + aws_access_key_id=self.s3_access_key, + aws_secret_access_key=self.s3_secret_key, + bucket_name=self.s3_bucket, + ) + logger.info(f"Using S3 storage backend: {self.s3_endpoint}/{self.s3_bucket}") + elif self.storage_backend_type == "file": + if not self.file_base_dir: + raise ValueError("File storage requires: file_base_dir") + self.storage = get_storage_backend("file", base_dir=self.file_base_dir) + logger.info(f"Using File storage backend: {self.file_base_dir}") + else: + raise ValueError(f"Unknown storage backend: {self.storage_backend_type}") + + def check_bucket_empty(self, prefix: str = "files/") -> bool: + """Check if the bucket is empty under the given prefix. + + Args: + prefix: The prefix to check. + + Returns: + True if bucket is empty (no objects), False otherwise. + """ + if self.storage is None: + raise RuntimeError("Storage not initialized") + objects = self.storage.list_objects(prefix) + if objects: + logger.error(f"Bucket is not empty under {prefix}: found {len(objects)} objects") + logger.error(f"First few objects: {objects[:5]}") + return False + return True + + def import_document(self, doc: PaperlessDocument) -> bool: + """Import a single document. + + Args: + doc: The Paperless document to import. + + Returns: + True if import succeeded, False otherwise. + """ + if self.storage is None: + raise RuntimeError("Storage not initialized") + + logger.info(f"Processing document {doc.id}: {doc.filename}") + + try: + # Download PDF + pdf_bytes = self.paperless_client.download_document(doc.id) + sha256 = self.pdf_processor.compute_sha256(pdf_bytes) + + # Extract text + extracted_text = self.pdf_processor.extract_text(pdf_bytes) + + # Build sidecar + sidecar = self.sidecar_builder.build_sidecar( + doc, pdf_bytes, sha256, extracted_text + ) + + # Determine keys + pdf_key = sidecar.current_key + sidecar_key = get_sidecar_key(pdf_key) + + # Check for resumability + if sidecar_exists(self.storage, pdf_key) and matching_sha256( + self.storage, pdf_key, sha256 + ): + logger.info(f"Skipping document {doc.id}: already imported with matching SHA256") + self.stats.skipped += 1 + self.stats.per_template_counts[sidecar.template_name] = ( + self.stats.per_template_counts.get(sidecar.template_name, 0) + 1 + ) + return True + + # Dry run: just log what would happen + if self.dry_run: + logger.info(f"[DRY RUN] Would import: {pdf_key}") + logger.info(f"[DRY RUN] SHA256: {sha256}") + logger.info(f"[DRY RUN] Template: {sidecar.template_name}") + logger.info(f"[DRY RUN] Emitting Company: {sidecar.emitting_company}") + logger.info(f"[DRY RUN] Document Type: {sidecar.document_type}") + logger.info(f"[DRY RUN] Emission Date: {sidecar.emission_date}") + logger.info(f"[DRY RUN] Period: {sidecar.period}") + logger.info(f"[DRY RUN] Optional Fields: {sidecar.optional_fields}") + self.stats.imported += 1 + self.stats.per_template_counts[sidecar.template_name] = ( + self.stats.per_template_counts.get(sidecar.template_name, 0) + 1 + ) + return True + + # Upload PDF + self.storage.put_object(pdf_key, pdf_bytes) + logger.debug(f"Uploaded PDF to {pdf_key}") + + # Write sidecar atomically + write_atomic(self.storage, sidecar, sidecar_key) + logger.debug(f"Wrote sidecar to {sidecar_key}") + + # Update stats + self.stats.imported += 1 + self.stats.per_template_counts[sidecar.template_name] = ( + self.stats.per_template_counts.get(sidecar.template_name, 0) + 1 + ) + + logger.info(f"Successfully imported document {doc.id}") + return True + + except Exception as e: + error_msg = f"Failed to import document {doc.id}: {e}" + logger.error(error_msg) + logger.debug("Exception details:", exc_info=True) + self.stats.failed += 1 + self.stats.errors.append(error_msg) + return False + + def run(self) -> bool: + """Run the import process. + + Returns: + True if import succeeded (no failures), False otherwise. + """ + if self.storage is None: + raise RuntimeError("Storage not initialized") + + logger.info("Starting Paperless-ngx import...") + logger.info(f"Storage backend: {self.storage_backend_type}") + + # Check bucket is empty (only for S3, skip for file backend in dry-run) + if not self.dry_run and self.storage_backend_type == "s3": + if not self.check_bucket_empty("files/"): + logger.error("Refusing to run: bucket is not empty under files/") + return False + + # Get all documents from Paperless + logger.info("Fetching documents from Paperless...") + docs = self.paperless_client.get_all_documents(self.limit) + + if not docs: + logger.info("No documents found in Paperless") + return True + + logger.info(f"Found {len(docs)} documents to import") + + # Import each document + for doc in docs: + self.import_document(doc) + + # Print summary + self.print_summary() + + # Return True if no failures + return self.stats.failed == 0 + + def print_summary(self) -> None: + """Print import summary.""" + print("\n" + "=" * 60) + print("IMPORT SUMMARY") + print("=" * 60) + print(f"Storage backend: {self.storage_backend_type}") + print(f"Imported: {self.stats.imported}") + print(f"Skipped: {self.stats.skipped}") + print(f"Failed: {self.stats.failed}") + print(f"Period fallbacks: {self.stats.period_fallbacks}") + print("\nPer-template counts:") + for template, count in sorted(self.stats.per_template_counts.items()): + print(f" {template}: {count}") + + if self.stats.errors: + print("\nErrors:") + for error in self.stats.errors: + print(f" - {error}") + print("=" * 60) + + +def parse_args(): + """Parse command line arguments.""" + parser = argparse.ArgumentParser( + description="Import documents from Paperless-ngx to storage backend", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Environment variables for S3 storage: + S3_ENDPOINT_URL Garage S3 endpoint URL + S3_ACCESS_KEY_ID S3 access key + S3_SECRET_ACCESS_KEY S3 secret key + S3_BUCKET_NAME S3 bucket name + +Environment variables for Paperless: + PAPERLESS_URL Paperless-ngx API URL + PAPERLESS_TOKEN Paperless-ngx API token + +For file storage, use --storage-backend file --file-base-dir /path/to/dir + """, + ) + + # Paperless configuration + parser.add_argument( + "--paperless-url", + help="Paperless-ngx API URL", + default=os.environ.get("PAPERLESS_URL"), + ) + parser.add_argument( + "--paperless-token", + help="Paperless-ngx API token", + default=os.environ.get("PAPERLESS_TOKEN"), + ) + + # Storage backend configuration + parser.add_argument( + "--storage-backend", + choices=["s3", "file"], + default="s3", + help="Storage backend type: 's3' for Garage S3, 'file' for local filesystem (default: s3)", + ) + + # S3 configuration + parser.add_argument( + "--s3-endpoint", + help="Garage S3 endpoint URL", + default=os.environ.get("S3_ENDPOINT_URL"), + ) + parser.add_argument( + "--s3-access-key", + help="S3 access key", + default=os.environ.get("S3_ACCESS_KEY_ID"), + ) + parser.add_argument( + "--s3-secret-key", + help="S3 secret key", + default=os.environ.get("S3_SECRET_ACCESS_KEY"), + ) + parser.add_argument( + "--s3-bucket", + help="S3 bucket name", + default=os.environ.get("S3_BUCKET_NAME"), + ) + + # File storage configuration + parser.add_argument( + "--file-base-dir", + help="Base directory for file storage backend", + default=os.environ.get("FILE_BASE_DIR", "./import_storage"), + ) + + # Import options + parser.add_argument( + "--dry-run", + action="store_true", + help="List what would be imported without actually importing", + ) + parser.add_argument( + "--limit", + type=int, + help="Maximum number of documents to import (for testing)", + ) + parser.add_argument( + "--verbose", + "-v", + action="store_true", + help="Enable verbose logging", + ) + + return parser.parse_args() + + +def validate_env(args) -> None: + """Validate that required environment variables or arguments are set. + + Args: + args: Parsed command line arguments. + + Raises: + ValueError: If required values are missing. + """ + missing = [] + + # Paperless is always required + if not args.paperless_url: + missing.append("Paperless URL (--paperless-url or PAPERLESS_URL)") + if not args.paperless_token: + missing.append("Paperless token (--paperless-token or PAPERLESS_TOKEN)") + + # Storage backend specific requirements + if args.storage_backend == "s3": + if not args.s3_endpoint: + missing.append("S3 endpoint (--s3-endpoint or S3_ENDPOINT_URL)") + if not args.s3_access_key: + missing.append("S3 access key (--s3-access-key or S3_ACCESS_KEY_ID)") + if not args.s3_secret_key: + missing.append("S3 secret key (--s3-secret-key or S3_SECRET_ACCESS_KEY)") + if not args.s3_bucket: + missing.append("S3 bucket (--s3-bucket or S3_BUCKET_NAME)") + elif args.storage_backend == "file": + if not args.file_base_dir: + missing.append("File base dir (--file-base-dir or FILE_BASE_DIR)") + + if missing: + raise ValueError(f"Missing required values:\n " + "\n ".join(missing)) + + +def main(): + """Main entry point.""" + args = parse_args() + + # Configure logging level + if args.verbose: + logging.getLogger().setLevel(logging.DEBUG) + + try: + validate_env(args) + except ValueError as e: + logger.error(str(e)) + sys.exit(1) + + try: + # Create importer + importer = Importer( + paperless_url=args.paperless_url, + paperless_token=args.paperless_token, + storage_backend=args.storage_backend, + s3_endpoint=args.s3_endpoint, + s3_access_key=args.s3_access_key, + s3_secret_key=args.s3_secret_key, + s3_bucket=args.s3_bucket, + file_base_dir=args.file_base_dir, + dry_run=args.dry_run, + limit=args.limit, + ) + + # Run import + success = importer.run() + + # Exit with appropriate code + if success: + logger.info("Import completed successfully") + sys.exit(0) + else: + logger.error("Import completed with failures") + sys.exit(1) + + except KeyboardInterrupt: + logger.info("Import interrupted by user") + sys.exit(130) # SIGINT exit code + except Exception as e: + logger.error(f"Import failed: {e}") + import traceback + + traceback.print_exc() + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/automatic-pdf-renamer/scripts/requirements.txt b/automatic-pdf-renamer/scripts/requirements.txt new file mode 100644 index 0000000..73efc3e --- /dev/null +++ b/automatic-pdf-renamer/scripts/requirements.txt @@ -0,0 +1,17 @@ +# Requirements for Paperless-ngx import script (User Story 0) +# These are the dependencies needed to run scripts/import_paperless.py + +# HTTP client for Paperless-ngx API +requests>=2.28.0 + +# S3 client for Garage +boto3>=1.26.0 + +# PDF text extraction (PyMuPDF) +pymupdf>=1.22.0 + +# Date parsing utilities +python-dateutil>=2.8.0 + +# YAML parsing for config (optional, for template loading) +PyYAML>=6.0