Delete all files

Dockerfile

@@ -1,33 +0,0 @@
-FROM python:3.11-slim
-
-# Set working directory
-WORKDIR /app
-
-# Install system dependencies
-RUN apt-get update && apt-get install -y \
-    tesseract-ocr \
-    tesseract-ocr-eng \
-    poppler-utils \
-    && rm -rf /var/lib/apt/lists/*
-
-# Copy requirements first for better caching
-COPY requirements.txt .
-
-# Install Python dependencies
-RUN pip install --no-cache-dir -r requirements.txt
-
-# Copy application code
-COPY . .
-
-# Create necessary directories
-RUN mkdir -p uploads outputs static config
-
-# Expose port (Hugging Face Spaces uses port 7860)
-EXPOSE 7860
-
-# Set environment variables for Gradio
-ENV GRADIO_SERVER_NAME="0.0.0.0" \
-    GRADIO_SERVER_PORT=7860
-
-# Run the application
-CMD ["python", "app_gradio.py"]

README.md

@@ -1,39 +0,0 @@
----
-title: T1 Tax PDF Processor
-emoji: 📄
-colorFrom: purple
-colorTo: blue
-sdk: gradio
-sdk_version: 4.19.2
-app_file: app_gradio.py
-pinned: false
-license: mit
----
-
-# T1 Tax PDF Processor
-
-Automatically extract data from Canadian T1 tax return PDFs and fill PDF forms.
-
-## How to Use
-
-1. **Upload Source PDF** - Your T1 tax return (with line numbers and data)
-2. **Upload Template PDF** - Empty fillable form you want to fill
-3. **Click "Process PDFs"** - Automatic extraction and filling
-4. **Download Result** - Get your filled PDF
-
-## Features
-
-- 📄 Extract text from T1 PDFs
-- 🔍 Smart T1 line detection (15000, 23600, 26000, etc.)
-- 🤖 OCR support for scanned PDFs
-- 📝 Automatic form filling
-- 🗺️ Configurable field mappings
-
-## Tech Stack
-
-Python · FastAPI · Gradio · pdfplumber · Tesseract OCR
-
-## ⚠️ Privacy Notice
-
-**Demo Only** - This is a demonstration application.
-Do not upload real tax documents with sensitive personal information to public demos.

app/__init__.py

@@ -1,7 +0,0 @@
-"""
-T1 Tax PDF Processor Application.
-
-A complete automation system for extracting data from Canadian T1 tax return PDFs
-and filling other PDF forms automatically.
-"""
-__version__ = "1.0.0"

app/api/__init__.py

@@ -1,4 +0,0 @@
-"""API module with FastAPI routes."""
-from app.api.routes import router
-
-__all__ = ["router"]

app/api/models.py

@@ -1,120 +0,0 @@
-"""
-Pydantic models for API request/response validation.
-"""
-from datetime import datetime
-from typing import Optional
-from pydantic import BaseModel, Field
-
-
-class UploadResponse(BaseModel):
-    """Response model for file upload endpoint."""
-    success: bool
-    session_id: str = Field(..., description="Unique session identifier for tracking files")
-    source_filename: str
-    target_filename: Optional[str] = None
-    message: str
-
-
-class ExtractedData(BaseModel):
-    """Model for extracted T1 tax data."""
-    line_values: dict[str, Optional[str]] = Field(
-        default_factory=dict,
-        description="Extracted line numbers and their values"
-    )
-    extraction_method: str = Field(
-        default="text",
-        description="Method used: 'text' or 'ocr'"
-    )
-    has_text: bool = Field(
-        default=True,
-        description="Whether PDF had extractable text"
-    )
-    raw_text: Optional[str] = Field(
-        default=None,
-        description="Raw extracted text (optional)"
-    )
-
-
-class ProcessingRequest(BaseModel):
-    """Request model for processing endpoint."""
-    session_id: str = Field(..., description="Session ID from upload")
-    line_numbers: Optional[list[str]] = Field(
-        default=None,
-        description="Specific line numbers to extract. If None, extracts all."
-    )
-    use_ocr: bool = Field(
-        default=False,
-        description="Force OCR processing"
-    )
-    include_raw_text: bool = Field(
-        default=False,
-        description="Include raw extracted text in response"
-    )
-
-
-class ProcessingResponse(BaseModel):
-    """Response model for processing endpoint."""
-    success: bool
-    session_id: str
-    extracted_data: Optional[ExtractedData] = None
-    mapped_fields: Optional[dict[str, str]] = None
-    output_filename: Optional[str] = None
-    errors: list[str] = Field(default_factory=list)
-    warnings: list[str] = Field(default_factory=list)
-
-
-class BatchUploadItem(BaseModel):
-    """Model for a single item in batch upload."""
-    source_filename: str
-    target_filename: Optional[str] = None
-
-
-class BatchProcessingRequest(BaseModel):
-    """Request model for batch processing."""
-    session_ids: list[str] = Field(..., description="List of session IDs to process")
-    line_numbers: Optional[list[str]] = None
-    use_ocr: bool = False
-
-
-class BatchProcessingResponse(BaseModel):
-    """Response model for batch processing."""
-    success: bool
-    total: int
-    processed: int
-    failed: int
-    results: list[ProcessingResponse]
-
-
-class FieldMapping(BaseModel):
-    """Model for a field mapping entry."""
-    line_number: str
-    field_name: str
-    description: Optional[str] = None
-
-
-class MappingConfigRequest(BaseModel):
-    """Request model for updating mapping configuration."""
-    mappings: list[FieldMapping]
-
-
-class MappingConfigResponse(BaseModel):
-    """Response model for mapping configuration."""
-    success: bool
-    mappings: dict[str, dict]
-    message: str
-
-
-class HealthResponse(BaseModel):
-    """Response model for health check endpoint."""
-    status: str
-    version: str
-    timestamp: datetime
-    tesseract_available: bool
-    openai_configured: bool
-
-
-class ErrorResponse(BaseModel):
-    """Standard error response model."""
-    success: bool = False
-    error: str
-    details: Optional[dict] = None

app/api/processor_service.py

@@ -1,274 +0,0 @@
-"""
-PDF processing service that orchestrates extraction, mapping, and filling.
-"""
-import uuid
-import shutil
-from pathlib import Path
-from typing import Optional
-from datetime import datetime
-
-from app.config import settings
-from app.extractor import PDFExtractor
-from app.ocr import OCRProcessor
-from app.mapper import FieldMapper
-from app.filler import PDFFiller
-from app.utils.logging import get_logger
-from app.utils.exceptions import (
-    PDFProcessorError,
-    PDFExtractionError,
-    OCRError,
-    MappingError,
-    PDFFillingError,
-    FileValidationError
-)
-
-logger = get_logger(__name__)
-
-
-class ProcessingSession:
-    """Represents a single PDF processing session."""
-
-    def __init__(self, session_id: str):
-        self.session_id = session_id
-        self.created_at = datetime.utcnow()
-        self.source_pdf: Optional[Path] = None
-        self.template_pdf: Optional[Path] = None
-        self.output_pdf: Optional[Path] = None
-        self.extracted_data: Optional[dict] = None
-        self.mapped_fields: Optional[dict] = None
-        self.status: str = "initialized"
-        self.errors: list[str] = []
-        self.warnings: list[str] = []
-
-    @property
-    def session_dir(self) -> Path:
-        return settings.upload_dir / self.session_id
-
-
-class PDFProcessorService:
-    """
-    Main service class that coordinates PDF processing.
-    """
-
-    def __init__(self, mapping_config_path: Optional[Path] = None):
-        self.extractor = PDFExtractor()
-        self.filler = PDFFiller()
-        self.mapper = FieldMapper(config_path=mapping_config_path)
-
-        # Initialize OCR (may not be available)
-        try:
-            self.ocr = OCRProcessor()
-            self.ocr_available = self.ocr.is_available()
-        except Exception as e:
-            logger.warning(f"OCR not available: {e}")
-            self.ocr = None
-            self.ocr_available = False
-
-        # Track active sessions
-        self.sessions: dict[str, ProcessingSession] = {}
-
-    def create_session(self) -> ProcessingSession:
-        """Create a new processing session."""
-        session_id = str(uuid.uuid4())
-        session = ProcessingSession(session_id)
-        session.session_dir.mkdir(parents=True, exist_ok=True)
-        self.sessions[session_id] = session
-        logger.info(f"Created new session: {session_id}")
-        return session
-
-    def get_session(self, session_id: str) -> ProcessingSession:
-        """Get an existing session."""
-        if session_id not in self.sessions:
-            # Try to restore from disk
-            session_dir = settings.upload_dir / session_id
-            if session_dir.exists():
-                session = ProcessingSession(session_id)
-                self.sessions[session_id] = session
-                return session
-            raise FileValidationError(f"Session not found: {session_id}")
-        return self.sessions[session_id]
-
-    def save_uploaded_file(
-        self,
-        session: ProcessingSession,
-        file_content: bytes,
-        filename: str,
-        file_type: str
-    ) -> Path:
-        """
-        Save an uploaded file to the session directory.
-
-        Args:
-            session: Processing session.
-            file_content: File content as bytes.
-            filename: Original filename.
-            file_type: 'source' or 'template'.
-
-        Returns:
-            Path to saved file.
-        """
-        # Validate extension
-        ext = Path(filename).suffix.lower()
-        if ext not in settings.allowed_extensions:
-            raise FileValidationError(
-                f"Invalid file extension: {ext}. Allowed: {settings.allowed_extensions}"
-            )
-
-        # Validate file size
-        size_mb = len(file_content) / (1024 * 1024)
-        if size_mb > settings.max_file_size_mb:
-            raise FileValidationError(
-                f"File too large: {size_mb:.2f}MB. Max: {settings.max_file_size_mb}MB"
-            )
-
-        # Save file
-        safe_filename = f"{file_type}_{filename}"
-        file_path = session.session_dir / safe_filename
-        file_path.write_bytes(file_content)
-
-        if file_type == "source":
-            session.source_pdf = file_path
-        elif file_type == "template":
-            session.template_pdf = file_path
-
-        logger.info(f"Saved {file_type} file: {file_path}")
-        return file_path
-
-    def process(
-        self,
-        session: ProcessingSession,
-        line_numbers: Optional[list[str]] = None,
-        use_ocr: Optional[bool] = None,
-        flatten: bool = False
-    ) -> dict:
-        """
-        Process the uploaded PDFs.
-
-        Args:
-            session: Processing session.
-            line_numbers: Specific line numbers to extract.
-            use_ocr: Force OCR processing.
-            flatten: Flatten output PDF form.
-
-        Returns:
-            Processing result dictionary.
-        """
-        if not session.source_pdf:
-            raise FileValidationError("No source PDF uploaded")
-        if not session.template_pdf:
-            raise FileValidationError("No template PDF uploaded")
-
-        session.status = "processing"
-
-        try:
-            # Step 1: Extract data from source PDF
-            extracted = self._extract_data(session, line_numbers, use_ocr)
-            session.extracted_data = extracted
-
-            # Step 2: Map extracted values to form fields
-            mapped = self.mapper.map_values(extracted["line_values"])
-            session.mapped_fields = mapped
-
-            # Step 3: Fill the template PDF
-            output_filename = f"filled_{session.session_id}.pdf"
-            output_path = settings.output_dir / output_filename
-
-            self.filler.fill_form(
-                template_path=session.template_pdf,
-                output_path=output_path,
-                field_values=mapped,
-                flatten=flatten
-            )
-            session.output_pdf = output_path
-
-            session.status = "completed"
-            logger.info(f"Processing completed for session {session.session_id}")
-
-            return {
-                "status": "success",
-                "extracted_data": extracted,
-                "mapped_fields": mapped,
-                "output_filename": output_filename
-            }
-
-        except Exception as e:
-            session.status = "failed"
-            session.errors.append(str(e))
-            logger.error(f"Processing failed for session {session.session_id}: {e}")
-            raise
-
-    def _extract_data(
-        self,
-        session: ProcessingSession,
-        line_numbers: Optional[list[str]] = None,
-        use_ocr: Optional[bool] = None
-    ) -> dict:
-        """Extract data from source PDF."""
-        # First try text extraction
-        has_text = self.extractor.has_text_content(session.source_pdf)
-
-        if has_text and not use_ocr:
-            # Use text extraction
-            result = self.extractor.extract_all_data(session.source_pdf, line_numbers)
-            result["extraction_method"] = "text"
-        elif self.ocr_available:
-            # Use OCR
-            logger.info("Using OCR for extraction")
-            text = self.ocr.process_pdf(session.source_pdf)
-            line_values = self.extractor.extract_line_values(text, line_numbers)
-            result = {
-                "text": text,
-                "line_values": line_values,
-                "has_text": False,
-                "extraction_method": "ocr"
-            }
-        else:
-            # No text and no OCR available
-            session.warnings.append("PDF appears to be scanned but OCR is not available")
-            result = self.extractor.extract_all_data(session.source_pdf, line_numbers)
-            result["extraction_method"] = "text_fallback"
-
-        # Add text preview (first 500 chars)
-        if result.get("text"):
-            result["text_preview"] = result["text"][:500]
-
-        return result
-
-    def get_output_path(self, session: ProcessingSession) -> Optional[Path]:
-        """Get the output PDF path for a session."""
-        if session.output_pdf and session.output_pdf.exists():
-            return session.output_pdf
-        return None
-
-    def get_template_fields(self, session: ProcessingSession) -> dict:
-        """Get form fields from the template PDF."""
-        if not session.template_pdf:
-            raise FileValidationError("No template PDF uploaded")
-        return self.filler.get_form_fields(session.template_pdf)
-
-    def update_mapping(self, line_number: str, field_name: str, description: str = "") -> None:
-        """Update field mapping."""
-        self.mapper.add_mapping(line_number, field_name, description)
-
-    def get_all_mappings(self) -> dict:
-        """Get all current mappings."""
-        return self.mapper.get_all_mappings()
-
-    def save_mapping_config(self, path: Optional[Path] = None) -> None:
-        """Save current mappings to config file."""
-        if path is None:
-            path = settings.config_dir / settings.default_mapping_file
-        self.mapper.save_config(path)
-
-    def cleanup_session(self, session_id: str) -> None:
-        """Clean up session files."""
-        if session_id in self.sessions:
-            session = self.sessions[session_id]
-            if session.session_dir.exists():
-                shutil.rmtree(session.session_dir)
-            del self.sessions[session_id]
-            logger.info(f"Cleaned up session: {session_id}")
-
-
-# Global service instance
-processor_service = PDFProcessorService()

app/api/routes.py

@@ -1,286 +0,0 @@
-"""
-FastAPI route handlers for PDF processing API.
-"""
-from pathlib import Path
-from typing import Optional
-
-from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks
-from fastapi.responses import FileResponse
-
-from app.api.schemas import (
-    ProcessingRequest,
-    ProcessingResult,
-    UploadResponse,
-    BatchProcessingRequest,
-    BatchProcessingResult,
-    MappingConfig,
-    MappingEntry,
-    TemplateValidation,
-    FormFieldInfo,
-    ExtractedData,
-    HealthResponse
-)
-from app.api.processor_service import processor_service
-from app.config import settings
-from app.utils.logging import get_logger
-from app.utils.exceptions import PDFProcessorError, FileValidationError
-
-logger = get_logger(__name__)
-
-router = APIRouter(prefix="/api/v1", tags=["pdf"])
-
-
-@router.get("/health", response_model=HealthResponse)
-async def health_check():
-    """Check API health and service availability."""
-    return HealthResponse(
-        status="healthy",
-        version=settings.app_version,
-        ocr_available=processor_service.ocr_available,
-        gemini_available=settings.google_api_key is not None
-    )
-
-
-@router.post("/upload", response_model=UploadResponse)
-async def upload_files(
-    source_pdf: UploadFile = File(..., description="T1 tax return PDF to extract data from"),
-    template_pdf: UploadFile = File(..., description="Target PDF form to fill")
-):
-    """
-    Upload source T1 PDF and target template PDF.
-
-    Returns a session ID for subsequent processing.
-    """
-    try:
-        # Create new session
-        session = processor_service.create_session()
-
-        # Save source PDF
-        source_content = await source_pdf.read()
-        processor_service.save_uploaded_file(
-            session, source_content, source_pdf.filename, "source"
-        )
-
-        # Save template PDF
-        template_content = await template_pdf.read()
-        processor_service.save_uploaded_file(
-            session, template_content, template_pdf.filename, "template"
-        )
-
-        return UploadResponse(
-            session_id=session.session_id,
-            source_filename=source_pdf.filename,
-            template_filename=template_pdf.filename,
-            message="Files uploaded successfully. Use /process to extract and fill data."
-        )
-
-    except FileValidationError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-    except Exception as e:
-        logger.error(f"Upload failed: {e}")
-        raise HTTPException(status_code=500, detail=f"Upload failed: {str(e)}")
-
-
-@router.post("/process", response_model=ProcessingResult)
-async def process_pdfs(request: ProcessingRequest):
-    """
-    Process uploaded PDFs: extract data and fill template.
-
-    Extracts T1 tax line values from the source PDF and fills the template PDF form.
-    """
-    try:
-        session = processor_service.get_session(request.session_id)
-
-        result = processor_service.process(
-            session=session,
-            line_numbers=request.line_numbers,
-            use_ocr=request.use_ocr,
-            flatten=request.flatten_output
-        )
-
-        extracted = result["extracted_data"]
-        return ProcessingResult(
-            session_id=session.session_id,
-            status="success",
-            extracted_data=ExtractedData(
-                line_values=extracted["line_values"],
-                extraction_method=extracted["extraction_method"],
-                has_text_content=extracted["has_text"],
-                raw_text_preview=extracted.get("text_preview")
-            ),
-            mapped_fields=result["mapped_fields"],
-            output_filename=result["output_filename"],
-            errors=session.errors,
-            warnings=session.warnings
-        )
-
-    except FileValidationError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-    except PDFProcessorError as e:
-        logger.error(f"Processing error: {e}")
-        raise HTTPException(status_code=422, detail=str(e))
-    except Exception as e:
-        logger.error(f"Unexpected error: {e}")
-        raise HTTPException(status_code=500, detail=f"Processing failed: {str(e)}")
-
-
-@router.get("/download/{session_id}")
-async def download_result(session_id: str):
-    """
-    Download the filled PDF for a completed session.
-    """
-    try:
-        session = processor_service.get_session(session_id)
-        output_path = processor_service.get_output_path(session)
-
-        if not output_path:
-            raise HTTPException(
-                status_code=404,
-                detail="Output file not found. Ensure processing is complete."
-            )
-
-        return FileResponse(
-            path=output_path,
-            filename=f"filled_t1_{session_id[:8]}.pdf",
-            media_type="application/pdf"
-        )
-
-    except FileValidationError as e:
-        raise HTTPException(status_code=404, detail=str(e))
-    except Exception as e:
-        logger.error(f"Download error: {e}")
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-@router.post("/batch/process", response_model=BatchProcessingResult)
-async def batch_process(request: BatchProcessingRequest):
-    """
-    Process multiple uploaded PDF sessions in batch.
-    """
-    results = []
-    successful = 0
-    failed = 0
-
-    for session_id in request.session_ids:
-        try:
-            session = processor_service.get_session(session_id)
-            result = processor_service.process(
-                session=session,
-                line_numbers=request.line_numbers,
-                use_ocr=request.use_ocr,
-                flatten=request.flatten_output
-            )
-
-            extracted = result["extracted_data"]
-            results.append(ProcessingResult(
-                session_id=session_id,
-                status="success",
-                extracted_data=ExtractedData(
-                    line_values=extracted["line_values"],
-                    extraction_method=extracted["extraction_method"],
-                    has_text_content=extracted["has_text"],
-                    raw_text_preview=extracted.get("text_preview")
-                ),
-                mapped_fields=result["mapped_fields"],
-                output_filename=result["output_filename"],
-                errors=session.errors,
-                warnings=session.warnings
-            ))
-            successful += 1
-
-        except Exception as e:
-            logger.error(f"Batch processing failed for {session_id}: {e}")
-            results.append(ProcessingResult(
-                session_id=session_id,
-                status="failed",
-                errors=[str(e)]
-            ))
-            failed += 1
-
-    return BatchProcessingResult(
-        total=len(request.session_ids),
-        successful=successful,
-        failed=failed,
-        results=results
-    )
-
-
-@router.get("/template/fields/{session_id}", response_model=TemplateValidation)
-async def get_template_fields(session_id: str):
-    """
-    Get form fields from the uploaded template PDF.
-    """
-    try:
-        session = processor_service.get_session(session_id)
-        fields = processor_service.get_template_fields(session)
-
-        field_list = [
-            FormFieldInfo(
-                name=name,
-                field_type=str(info.get("type", "Unknown")),
-                current_value=str(info.get("value", "")) if info.get("value") else None
-            )
-            for name, info in fields.items()
-        ]
-
-        # Check which mapped fields are missing
-        all_mappings = processor_service.get_all_mappings()
-        mapped_field_names = {m["field"] for m in all_mappings.values()}
-        existing_field_names = set(fields.keys())
-        missing = list(mapped_field_names - existing_field_names)
-
-        return TemplateValidation(
-            valid=len(missing) == 0,
-            total_fields=len(fields),
-            fields=field_list,
-            missing_fields=missing
-        )
-
-    except FileValidationError as e:
-        raise HTTPException(status_code=400, detail=str(e))
-    except Exception as e:
-        logger.error(f"Template fields error: {e}")
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-@router.get("/mappings")
-async def get_mappings():
-    """
-    Get all current T1 line to field mappings.
-    """
-    return processor_service.get_all_mappings()
-
-
-@router.post("/mappings")
-async def update_mappings(config: MappingConfig):
-    """
-    Update T1 line to field mappings.
-    """
-    try:
-        for entry in config.mappings:
-            processor_service.update_mapping(
-                entry.line_number,
-                entry.field_name,
-                entry.description or ""
-            )
-
-        # Save to config file
-        processor_service.save_mapping_config()
-
-        return {"message": f"Updated {len(config.mappings)} mappings"}
-
-    except Exception as e:
-        logger.error(f"Mapping update error: {e}")
-        raise HTTPException(status_code=500, detail=str(e))
-
-
-@router.delete("/session/{session_id}")
-async def cleanup_session(session_id: str, background_tasks: BackgroundTasks):
-    """
-    Clean up session files and data.
-    """
-    try:
-        background_tasks.add_task(processor_service.cleanup_session, session_id)
-        return {"message": f"Session {session_id} cleanup scheduled"}
-    except Exception as e:
-        raise HTTPException(status_code=500, detail=str(e))

app/api/schemas.py

@@ -1,108 +0,0 @@
-"""
-Pydantic models for API request/response validation.
-"""
-from pathlib import Path
-from typing import Optional
-from pydantic import BaseModel, Field
-
-
-class ProcessingRequest(BaseModel):
-    """Request model for PDF processing."""
-    session_id: str = Field(..., description="Session ID from upload")
-    line_numbers: Optional[list[str]] = Field(
-        None,
-        description="Specific T1 line numbers to extract. If None, extracts all."
-    )
-    use_ocr: Optional[bool] = Field(
-        None,
-        description="Force OCR processing. Auto-detected if None."
-    )
-    flatten_output: bool = Field(
-        False,
-        description="Flatten the output PDF form fields."
-    )
-
-
-class ExtractedData(BaseModel):
-    """Model for extracted T1 tax data."""
-    line_values: dict[str, Optional[str]]
-    extraction_method: str
-    has_text_content: bool
-    raw_text_preview: Optional[str] = None
-
-
-class ProcessingResult(BaseModel):
-    """Response model for processing results."""
-    session_id: str
-    status: str
-    extracted_data: Optional[ExtractedData] = None
-    mapped_fields: Optional[dict[str, str]] = None
-    output_filename: Optional[str] = None
-    errors: list[str] = []
-    warnings: list[str] = []
-
-
-class UploadResponse(BaseModel):
-    """Response model for file upload."""
-    session_id: str
-    source_filename: str
-    template_filename: str
-    message: str
-
-
-class DownloadInfo(BaseModel):
-    """Response model for download information."""
-    session_id: str
-    filename: str
-    download_url: str
-
-
-class BatchProcessingRequest(BaseModel):
-    """Request model for batch processing."""
-    session_ids: list[str]
-    line_numbers: Optional[list[str]] = None
-    use_ocr: Optional[bool] = None
-    flatten_output: bool = False
-
-
-class BatchProcessingResult(BaseModel):
-    """Response model for batch processing."""
-    total: int
-    successful: int
-    failed: int
-    results: list[ProcessingResult]
-
-
-class MappingEntry(BaseModel):
-    """Model for a single mapping entry."""
-    line_number: str
-    field_name: str
-    description: Optional[str] = ""
-
-
-class MappingConfig(BaseModel):
-    """Model for mapping configuration."""
-    mappings: list[MappingEntry]
-
-
-class FormFieldInfo(BaseModel):
-    """Model for PDF form field information."""
-    name: str
-    field_type: str
-    current_value: Optional[str] = None
-
-
-class TemplateValidation(BaseModel):
-    """Response model for template validation."""
-    valid: bool
-    total_fields: int
-    fields: list[FormFieldInfo]
-    missing_fields: list[str] = []
-
-
-class HealthResponse(BaseModel):
-    """Health check response."""
-    status: str
-    version: str
-    ocr_available: bool
-    gemini_available: bool

app/config.py

@@ -1,53 +0,0 @@
-"""
-Application configuration settings.
-"""
-import os
-from pathlib import Path
-from typing import Optional
-
-from pydantic_settings import BaseSettings
-
-
-class Settings(BaseSettings):
-    """Application settings loaded from environment variables."""
-
-    # Application
-    app_name: str = "T1 Tax PDF Processor"
-    app_version: str = "1.0.0"
-    debug: bool = False
-
-    # Paths
-    base_dir: Path = Path(__file__).parent.parent
-    upload_dir: Path = base_dir / "uploads"
-    output_dir: Path = base_dir / "outputs"
-    config_dir: Path = base_dir / "config"
-    static_dir: Path = base_dir / "static"
-    templates_dir: Path = base_dir / "templates"
-
-    # OCR Settings
-    tesseract_cmd: Optional[str] = None  # Path to tesseract executable
-    ocr_language: str = "eng"
-    ocr_dpi: int = 300
-
-    # PDF Processing
-    max_file_size_mb: int = 50
-    allowed_extensions: list[str] = [".pdf"]
-
-    # Google Generative AI (optional)
-    google_api_key: Optional[str] = None
-    gemini_model: str = "gemini-2.5-flash"
-
-    # Mapping
-    default_mapping_file: str = "t1_mapping.json"
-
-    class Config:
-        env_file = ".env"
-        env_file_encoding = "utf-8"
-
-
-settings = Settings()
-
-# Create directories if they don't exist
-settings.upload_dir.mkdir(parents=True, exist_ok=True)
-settings.output_dir.mkdir(parents=True, exist_ok=True)
-settings.config_dir.mkdir(parents=True, exist_ok=True)

app/extractor/__init__.py

@@ -1,6 +0,0 @@
-"""
-PDF Extractor module for extracting text and data from Canadian T1 tax PDFs.
-"""
-from .pdf_extractor import PDFExtractor
-
-__all__ = ["PDFExtractor"]

app/extractor/pdf_extractor.py

@@ -1,186 +0,0 @@
-"""
-PDF text extraction module.
-Extracts text from PDFs using pdfplumber with PyMuPDF as fallback.
-"""
-import re
-from pathlib import Path
-from typing import Optional
-
-import pdfplumber
-import fitz  # PyMuPDF
-
-from app.utils.logging import get_logger
-from app.utils.exceptions import PDFExtractionError
-
-logger = get_logger(__name__)
-
-
-class PDFExtractor:
-    """
-    Extracts text content and specific T1 tax line values from PDFs.
-    Uses pdfplumber as primary extractor with PyMuPDF fallback.
-    """
-
-    # Common T1 tax form line patterns
-    # Lines can appear in various formats:
-    # "15000 Total income: 50000.00"
-    # "Line 15000: $50,000.00"
-    # "15000  50000"
-    # "Total income (line 15000) 50000.00"
-    LINE_PATTERNS = [
-        # Pattern: "15000 Description: Value" or "15000 Value"
-        r"(?:line\s*)?(\d{5})\s*[:\-]?\s*(?:[A-Za-z\s,]+[:\-])?\s*\$?\s*([\d,]+\.?\d*)",
-        # Pattern: "Description (line 15000): Value"
-        r"[A-Za-z\s]+\(line\s*(\d{5})\)[:\s]*\$?\s*([\d,]+\.?\d*)",
-        # Pattern: "Line 15000 Description Value"
-        r"(?:Line|line)\s+(\d{5})\s+[A-Za-z\s]*?\s+\$?\s*([\d,]+\.?\d*)",
-        # Pattern: Simple "15000 50000.00"
-        r"^(\d{5})\s+\$?\s*([\d,]+\.?\d*)$",
-        # Pattern: With dollar sign "15000 $50,000.00"
-        r"(\d{5})\s+\$\s*([\d,]+\.?\d*)",
-    ]
-
-    def __init__(self):
-        self.compiled_patterns = [re.compile(p, re.MULTILINE | re.IGNORECASE) for p in self.LINE_PATTERNS]
-
-    def extract_text(self, pdf_path: Path) -> str:
-        """
-        Extract all text from a PDF file.
-
-        Args:
-            pdf_path: Path to the PDF file.
-
-        Returns:
-            Extracted text content.
-
-        Raises:
-            PDFExtractionError: If extraction fails.
-        """
-        pdf_path = Path(pdf_path)
-        if not pdf_path.exists():
-            raise PDFExtractionError(f"PDF file not found: {pdf_path}")
-
-        text = self._extract_with_pdfplumber(pdf_path)
-
-        # If pdfplumber returns empty or minimal text, try PyMuPDF
-        if not text or len(text.strip()) < 50:
-            logger.info("pdfplumber returned minimal text, trying PyMuPDF")
-            text_pymupdf = self._extract_with_pymupdf(pdf_path)
-            if len(text_pymupdf) > len(text):
-                text = text_pymupdf
-
-        return text
-
-    def _extract_with_pdfplumber(self, pdf_path: Path) -> str:
-        """Extract text using pdfplumber."""
-        try:
-            text_parts = []
-            with pdfplumber.open(pdf_path) as pdf:
-                for page_num, page in enumerate(pdf.pages, 1):
-                    page_text = page.extract_text() or ""
-                    text_parts.append(page_text)
-                    logger.debug(f"Extracted {len(page_text)} chars from page {page_num}")
-            return "\n".join(text_parts)
-        except Exception as e:
-            logger.warning(f"pdfplumber extraction failed: {e}")
-            return ""
-
-    def _extract_with_pymupdf(self, pdf_path: Path) -> str:
-        """Extract text using PyMuPDF (fitz)."""
-        try:
-            text_parts = []
-            with fitz.open(pdf_path) as doc:
-                for page_num, page in enumerate(doc, 1):
-                    page_text = page.get_text()
-                    text_parts.append(page_text)
-                    logger.debug(f"PyMuPDF extracted {len(page_text)} chars from page {page_num}")
-            return "\n".join(text_parts)
-        except Exception as e:
-            logger.warning(f"PyMuPDF extraction failed: {e}")
-            return ""
-
-    def has_text_content(self, pdf_path: Path) -> bool:
-        """
-        Check if PDF has extractable text content.
-        Used to determine if OCR is needed.
-
-        Args:
-            pdf_path: Path to the PDF file.
-
-        Returns:
-            True if PDF has text content, False if likely scanned.
-        """
-        text = self.extract_text(pdf_path)
-        # If we got less than 100 characters, probably needs OCR
-        return len(text.strip()) > 100
-
-    def extract_line_values(self, text: str, line_numbers: Optional[list[str]] = None) -> dict[str, Optional[str]]:
-        """
-        Extract specific T1 tax line values from text.
-
-        Args:
-            text: Text content from PDF.
-            line_numbers: List of line numbers to extract (e.g., ["15000", "23600"]).
-                         If None, extracts all found line values.
-
-        Returns:
-            Dictionary mapping line numbers to their values.
-        """
-        extracted = {}
-
-        # If specific lines requested, initialize them
-        if line_numbers:
-            for line in line_numbers:
-                extracted[line] = None
-
-        # Apply all patterns
-        for pattern in self.compiled_patterns:
-            matches = pattern.findall(text)
-            for match in matches:
-                if len(match) >= 2:
-                    line_num = match[0]
-                    value = self._clean_value(match[1])
-
-                    # Only include if no filter or line is in filter
-                    if line_numbers is None or line_num in line_numbers:
-                        # Keep the value if we found one (don't overwrite with None)
-                        if value and (line_num not in extracted or extracted[line_num] is None):
-                            extracted[line_num] = value
-                            logger.debug(f"Extracted line {line_num}: {value}")
-
-        return extracted
-
-    def _clean_value(self, value: str) -> Optional[str]:
-        """Clean extracted value by removing formatting."""
-        if not value:
-            return None
-        # Remove commas and extra spaces
-        cleaned = value.replace(",", "").strip()
-        # Verify it's a valid number
-        try:
-            float(cleaned)
-            return cleaned
-        except ValueError:
-            return None
-
-    def extract_all_data(self, pdf_path: Path, line_numbers: Optional[list[str]] = None) -> dict:
-        """
-        Extract all data from a PDF including text and line values.
-
-        Args:
-            pdf_path: Path to the PDF file.
-            line_numbers: Optional list of specific line numbers to extract.
-
-        Returns:
-            Dictionary with 'text', 'line_values', and 'has_text' keys.
-        """
-        text = self.extract_text(pdf_path)
-        has_text = len(text.strip()) > 100
-        line_values = self.extract_line_values(text, line_numbers)
-
-        return {
-            "text": text,
-            "line_values": line_values,
-            "has_text": has_text,
-            "extraction_method": "text"
-        }

app/filler/__init__.py

@@ -1,4 +0,0 @@
-"""PDF Filler module for populating PDF forms."""
-from app.filler.pdf_filler import PDFFiller
-
-__all__ = ["PDFFiller"]

app/filler/pdf_filler.py

@@ -1,252 +0,0 @@
-"""
-PDF form filler module using pypdf and pdfrw.
-"""
-from pathlib import Path
-from typing import Optional
-import io
-
-from pypdf import PdfReader, PdfWriter
-from pypdf.generic import NameObject, BooleanObject
-
-from app.utils.logging import get_logger
-from app.utils.exceptions import PDFFillingError
-
-logger = get_logger(__name__)
-
-
-class PDFFiller:
-    """
-    Fills PDF form fields with extracted T1 tax data.
-    Supports AcroForm fillable PDFs.
-    """
-
-    def __init__(self):
-        pass
-
-    def get_form_fields(self, pdf_path: Path) -> dict[str, dict]:
-        """
-        Get all form fields from a PDF.
-
-        Args:
-            pdf_path: Path to the PDF file.
-
-        Returns:
-            Dictionary of field names to field info.
-        """
-        pdf_path = Path(pdf_path)
-        if not pdf_path.exists():
-            raise PDFFillingError(f"PDF file not found: {pdf_path}")
-
-        try:
-            reader = PdfReader(pdf_path)
-            fields = reader.get_fields()
-
-            if not fields:
-                logger.warning(f"No form fields found in {pdf_path}")
-                return {}
-
-            field_info = {}
-            for name, field in fields.items():
-                field_info[name] = {
-                    "type": field.get("/FT", "Unknown"),
-                    "value": field.get("/V", ""),
-                    "default": field.get("/DV", ""),
-                }
-            logger.info(f"Found {len(field_info)} form fields in {pdf_path}")
-            return field_info
-
-        except Exception as e:
-            raise PDFFillingError(f"Failed to read form fields: {e}")
-
-    def fill_form(
-        self,
-        template_path: Path,
-        output_path: Path,
-        field_values: dict[str, str],
-        flatten: bool = False
-    ) -> Path:
-        """
-        Fill PDF form fields with values.
-
-        Args:
-            template_path: Path to the template PDF with form fields.
-            output_path: Path for the output filled PDF.
-            field_values: Dictionary mapping field names to values.
-            flatten: If True, flatten the form (make fields non-editable).
-
-        Returns:
-            Path to the filled PDF file.
-        """
-        template_path = Path(template_path)
-        output_path = Path(output_path)
-
-        if not template_path.exists():
-            raise PDFFillingError(f"Template PDF not found: {template_path}")
-
-        output_path.parent.mkdir(parents=True, exist_ok=True)
-
-        try:
-            reader = PdfReader(template_path)
-            writer = PdfWriter()
-
-            # Copy all pages
-            for page in reader.pages:
-                writer.add_page(page)
-
-            # Get existing form fields
-            existing_fields = reader.get_fields() or {}
-
-            # Update fields with values
-            fields_filled = 0
-            fields_missing = []
-
-            for field_name, value in field_values.items():
-                if field_name in existing_fields:
-                    writer.update_page_form_field_values(
-                        writer.pages[0],
-                        {field_name: str(value)},
-                        auto_regenerate=True
-                    )
-                    fields_filled += 1
-                    logger.debug(f"Filled field '{field_name}' with '{value}'")
-                else:
-                    fields_missing.append(field_name)
-                    logger.warning(f"Field '{field_name}' not found in PDF")
-
-            if fields_missing:
-                logger.info(f"Missing fields: {fields_missing}")
-
-            # Flatten if requested
-            if flatten:
-                for page in writer.pages:
-                    if "/Annots" in page:
-                        for annotation in page["/Annots"]:
-                            annotation_obj = annotation.get_object()
-                            if annotation_obj.get("/Subtype") == "/Widget":
-                                annotation_obj[NameObject("/Ff")] = BooleanObject(True)
-
-            # Write output
-            with open(output_path, "wb") as f:
-                writer.write(f)
-
-            logger.info(f"Filled {fields_filled} fields, saved to {output_path}")
-            return output_path
-
-        except Exception as e:
-            raise PDFFillingError(f"Failed to fill PDF form: {e}")
-
-    def fill_form_bytes(
-        self,
-        template_bytes: bytes,
-        field_values: dict[str, str],
-        flatten: bool = False
-    ) -> bytes:
-        """
-        Fill PDF form from bytes and return filled PDF as bytes.
-
-        Args:
-            template_bytes: Template PDF as bytes.
-            field_values: Dictionary mapping field names to values.
-            flatten: If True, flatten the form.
-
-        Returns:
-            Filled PDF as bytes.
-        """
-        try:
-            reader = PdfReader(io.BytesIO(template_bytes))
-            writer = PdfWriter()
-
-            for page in reader.pages:
-                writer.add_page(page)
-
-            existing_fields = reader.get_fields() or {}
-
-            for field_name, value in field_values.items():
-                if field_name in existing_fields:
-                    writer.update_page_form_field_values(
-                        writer.pages[0],
-                        {field_name: str(value)},
-                        auto_regenerate=True
-                    )
-
-            output = io.BytesIO()
-            writer.write(output)
-            return output.getvalue()
-
-        except Exception as e:
-            raise PDFFillingError(f"Failed to fill PDF from bytes: {e}")
-
-    def validate_template(self, pdf_path: Path, required_fields: list[str]) -> dict:
-        """
-        Validate that a PDF template has the required form fields.
-
-        Args:
-            pdf_path: Path to the template PDF.
-            required_fields: List of required field names.
-
-        Returns:
-            Dictionary with validation results.
-        """
-        existing_fields = self.get_form_fields(pdf_path)
-        existing_names = set(existing_fields.keys())
-        required_names = set(required_fields)
-
-        present = required_names & existing_names
-        missing = required_names - existing_names
-        extra = existing_names - required_names
-
-        return {
-            "valid": len(missing) == 0,
-            "present_fields": list(present),
-            "missing_fields": list(missing),
-            "extra_fields": list(extra),
-            "total_fields": len(existing_fields)
-        }
-
-    def create_field_mapping_report(
-        self,
-        pdf_path: Path,
-        line_values: dict[str, str],
-        field_mapping: dict[str, str]
-    ) -> dict:
-        """
-        Create a report showing how T1 lines map to PDF fields.
-
-        Args:
-            pdf_path: Path to the template PDF.
-            line_values: Extracted T1 line values.
-            field_mapping: Mapping from line numbers to field names.
-
-        Returns:
-            Report dictionary.
-        """
-        existing_fields = self.get_form_fields(pdf_path)
-
-        report = {
-            "mappable_lines": [],
-            "unmappable_lines": [],
-            "unfilled_fields": []
-        }
-
-        mapped_fields = set()
-        for line_num, value in line_values.items():
-            field_name = field_mapping.get(line_num)
-            if field_name and field_name in existing_fields:
-                report["mappable_lines"].append({
-                    "line": line_num,
-                    "field": field_name,
-                    "value": value
-                })
-                mapped_fields.add(field_name)
-            else:
-                report["unmappable_lines"].append({
-                    "line": line_num,
-                    "value": value,
-                    "reason": "No matching field" if not field_name else "Field not in PDF"
-                })
-
-        for field_name in existing_fields:
-            if field_name not in mapped_fields:
-                report["unfilled_fields"].append(field_name)
-
-        return report

app/main.py

@@ -1,63 +0,0 @@
-"""
-Main FastAPI application entry point.
-"""
-from fastapi import FastAPI
-from fastapi.middleware.cors import CORSMiddleware
-from fastapi.staticfiles import StaticFiles
-from fastapi.responses import FileResponse
-
-from app.api import router
-from app.config import settings
-from app.utils.logging import setup_logging
-
-# Setup logging
-setup_logging()
-
-# Create FastAPI app
-app = FastAPI(
-    title=settings.app_name,
-    version=settings.app_version,
-    description="API for extracting data from Canadian T1 tax return PDFs and filling target PDF forms.",
-    docs_url="/docs",
-    redoc_url="/redoc"
-)
-
-# CORS middleware for frontend
-app.add_middleware(
-    CORSMiddleware,
-    allow_origins=["*"],
-    allow_credentials=True,
-    allow_methods=["*"],
-    allow_headers=["*"],
-)
-
-# Include API routes
-app.include_router(router)
-
-# Mount static files
-app.mount("/static", StaticFiles(directory=str(settings.static_dir)), name="static")
-
-
-@app.get("/")
-async def root():
-    """Serve the main frontend page."""
-    return FileResponse(settings.templates_dir / "index.html")
-
-
-@app.get("/favicon.ico")
-async def favicon():
-    """Serve favicon."""
-    favicon_path = settings.static_dir / "favicon.ico"
-    if favicon_path.exists():
-        return FileResponse(favicon_path)
-    return {"detail": "Not found"}
-
-
-if __name__ == "__main__":
-    import uvicorn
-    uvicorn.run(
-        "app.main:app",
-        host="0.0.0.0",
-        port=8000,
-        reload=settings.debug
-    )

app/mapper/__init__.py

@@ -1,4 +0,0 @@
-"""Mapper module for T1 line to PDF field mapping."""
-from app.mapper.field_mapper import FieldMapper
-
-__all__ = ["FieldMapper"]

app/mapper/field_mapper.py

@@ -1,135 +0,0 @@
-"""
-Field mapping module for T1 tax line to PDF form field mapping.
-"""
-import json
-from pathlib import Path
-from typing import Optional
-
-from app.config import settings
-from app.utils.logging import get_logger
-from app.utils.exceptions import MappingError
-
-logger = get_logger(__name__)
-
-
-class FieldMapper:
-    """
-    Maps T1 tax line numbers to PDF form field names.
-    Supports loading mappings from JSON configuration files.
-    """
-
-    DEFAULT_MAPPINGS = {
-        "10100": {"field": "employment_income", "description": "Employment income"},
-        "10400": {"field": "other_employment_income", "description": "Other employment income"},
-        "11500": {"field": "other_pensions", "description": "Other pensions and superannuation"},
-        "12100": {"field": "interest_income", "description": "Interest and other investment income"},
-        "12700": {"field": "taxable_capital_gains", "description": "Taxable capital gains"},
-        "13000": {"field": "other_income", "description": "Other income"},
-        "13500": {"field": "self_employment_income", "description": "Self-employment income"},
-        "14400": {"field": "workers_compensation", "description": "Workers compensation benefits"},
-        "14500": {"field": "social_assistance", "description": "Social assistance payments"},
-        "15000": {"field": "total_income", "description": "Total income"},
-        "20700": {"field": "registered_pension_plan", "description": "Registered pension plan deduction"},
-        "20800": {"field": "rrsp_deduction", "description": "RRSP/PRPP deduction"},
-        "21200": {"field": "union_dues", "description": "Annual union, professional, or like dues"},
-        "21900": {"field": "moving_expenses", "description": "Moving expenses"},
-        "22000": {"field": "support_payments", "description": "Support payments made"},
-        "22100": {"field": "carrying_charges", "description": "Carrying charges and interest expenses"},
-        "22900": {"field": "other_employment_expenses", "description": "Other employment expenses"},
-        "23200": {"field": "other_deductions", "description": "Other deductions"},
-        "23300": {"field": "cpp_self_employed", "description": "CPP/QPP contributions on self-employment"},
-        "23600": {"field": "net_income", "description": "Net income"},
-        "26000": {"field": "taxable_income", "description": "Taxable income"},
-        "30000": {"field": "basic_personal_amount", "description": "Basic personal amount"},
-        "31200": {"field": "ei_premiums", "description": "Employment insurance premiums"},
-        "31800": {"field": "disability_amount_self", "description": "Disability amount (for self)"},
-        "33500": {"field": "medical_expenses", "description": "Medical expenses"},
-        "34900": {"field": "donations_gifts", "description": "Donations and gifts"},
-        "35000": {"field": "total_non_refundable_credits", "description": "Total federal non-refundable tax credits"},
-        "42000": {"field": "net_federal_tax", "description": "Net federal tax"},
-        "43500": {"field": "total_payable", "description": "Total payable"},
-        "43700": {"field": "total_income_tax_deducted", "description": "Total income tax deducted"},
-        "48400": {"field": "refund", "description": "Refund"},
-        "48500": {"field": "balance_owing", "description": "Balance owing"}
-    }
-
-    def __init__(self, config_path: Optional[Path] = None):
-        self.mappings: dict[str, dict[str, str]] = {}
-        self._load_default_mappings()
-        if config_path:
-            self.load_config(config_path)
-
-    def _load_default_mappings(self) -> None:
-        self.mappings = self.DEFAULT_MAPPINGS.copy()
-        logger.debug(f"Loaded {len(self.mappings)} default mappings")
-
-    def load_config(self, config_path: Path) -> None:
-        config_path = Path(config_path)
-        if not config_path.exists():
-            raise MappingError(f"Mapping config file not found: {config_path}")
-
-        try:
-            with open(config_path, "r", encoding="utf-8") as f:
-                config = json.load(f)
-
-            for line_num, value in config.items():
-                if isinstance(value, str):
-                    self.mappings[line_num] = {"field": value, "description": ""}
-                elif isinstance(value, dict):
-                    self.mappings[line_num] = value
-                else:
-                    logger.warning(f"Invalid mapping format for line {line_num}")
-
-            logger.info(f"Loaded mapping config from {config_path}: {len(config)} mappings")
-        except json.JSONDecodeError as e:
-            raise MappingError(f"Invalid JSON in config file: {e}")
-        except Exception as e:
-            raise MappingError(f"Failed to load config: {e}")
-
-    def save_config(self, config_path: Path) -> None:
-        config_path = Path(config_path)
-        config_path.parent.mkdir(parents=True, exist_ok=True)
-        with open(config_path, "w", encoding="utf-8") as f:
-            json.dump(self.mappings, f, indent=2)
-        logger.info(f"Saved mapping config to {config_path}")
-
-    def get_field_name(self, line_number: str) -> Optional[str]:
-        mapping = self.mappings.get(line_number)
-        return mapping["field"] if mapping else None
-
-    def get_mapping(self, line_number: str) -> Optional[dict[str, str]]:
-        return self.mappings.get(line_number)
-
-    def map_values(self, line_values: dict[str, str]) -> dict[str, str]:
-        mapped = {}
-        unmapped_lines = []
-        for line_num, value in line_values.items():
-            if value is None:
-                continue
-            field_name = self.get_field_name(line_num)
-            if field_name:
-                mapped[field_name] = value
-                logger.debug(f"Mapped line {line_num} -> {field_name}: {value}")
-            else:
-                unmapped_lines.append(line_num)
-                logger.warning(f"No mapping found for line {line_num}")
-        if unmapped_lines:
-            logger.info(f"Unmapped lines: {unmapped_lines}")
-        return mapped
-
-    def add_mapping(self, line_number: str, field_name: str, description: str = "") -> None:
-        self.mappings[line_number] = {"field": field_name, "description": description}
-        logger.debug(f"Added mapping: {line_number} -> {field_name}")
-
-    def remove_mapping(self, line_number: str) -> bool:
-        if line_number in self.mappings:
-            del self.mappings[line_number]
-            logger.debug(f"Removed mapping for line {line_number}")
-            return True
-        return False
-
-    def get_all_mappings(self) -> dict[str, dict[str, str]]:
-        return self.mappings.copy()
-
-    def get_required_lines(self) -> list[str]:
-        return list(self.mappings.keys())

app/ocr/__init__.py

@@ -1,4 +0,0 @@
-"""OCR module for processing scanned PDFs."""
-from app.ocr.ocr_processor import OCRProcessor
-
-__all__ = ["OCRProcessor"]

app/ocr/ocr_processor.py

@@ -1,188 +0,0 @@
-"""
-OCR module for processing scanned PDF documents using Tesseract.
-"""
-import tempfile
-from pathlib import Path
-from typing import Optional
-
-import pytesseract
-from pdf2image import convert_from_path
-from PIL import Image
-
-from app.config import settings
-from app.utils.logging import get_logger
-from app.utils.exceptions import OCRError
-
-logger = get_logger(__name__)
-
-
-class OCRProcessor:
-    """
-    Processes scanned PDFs using Tesseract OCR.
-    Converts PDF pages to images and extracts text via OCR.
-    """
-
-    def __init__(self, tesseract_cmd: Optional[str] = None, language: str = "eng"):
-        """
-        Initialize the OCR processor.
-
-        Args:
-            tesseract_cmd: Path to tesseract executable. Auto-detected if None.
-            language: OCR language code (default: "eng" for English).
-        """
-        self.language = language
-        self.dpi = settings.ocr_dpi
-
-        # Configure tesseract path if provided
-        if tesseract_cmd or settings.tesseract_cmd:
-            pytesseract.pytesseract.tesseract_cmd = tesseract_cmd or settings.tesseract_cmd
-
-        # Verify tesseract is available
-        self._verify_tesseract()
-
-    def _verify_tesseract(self) -> None:
-        """Verify that Tesseract is installed and accessible."""
-        try:
-            version = pytesseract.get_tesseract_version()
-            logger.info(f"Tesseract version: {version}")
-        except Exception as e:
-            logger.warning(
-                f"Tesseract not found or not accessible: {e}. "
-                "OCR will fail if invoked. Install Tesseract to enable OCR support."
-            )
-
-    def process_pdf(self, pdf_path: Path) -> str:
-        """
-        Extract text from a scanned PDF using OCR.
-
-        Args:
-            pdf_path: Path to the PDF file.
-
-        Returns:
-            Extracted text from all pages.
-
-        Raises:
-            OCRError: If OCR processing fails.
-        """
-        pdf_path = Path(pdf_path)
-        if not pdf_path.exists():
-            raise OCRError(f"PDF file not found: {pdf_path}")
-
-        logger.info(f"Starting OCR processing for: {pdf_path}")
-
-        try:
-            # Convert PDF pages to images
-            images = self._pdf_to_images(pdf_path)
-            logger.info(f"Converted PDF to {len(images)} images")
-
-            # Process each image with OCR
-            text_parts = []
-            for i, image in enumerate(images, 1):
-                page_text = self._ocr_image(image)
-                text_parts.append(page_text)
-                logger.debug(f"OCR extracted {len(page_text)} chars from page {i}")
-
-            full_text = "\n\n".join(text_parts)
-            logger.info(f"OCR completed. Total characters extracted: {len(full_text)}")
-            return full_text
-
-        except Exception as e:
-            raise OCRError(f"OCR processing failed: {str(e)}", {"pdf_path": str(pdf_path)})
-
-    def _pdf_to_images(self, pdf_path: Path) -> list[Image.Image]:
-        """
-        Convert PDF pages to PIL images.
-
-        Args:
-            pdf_path: Path to the PDF file.
-
-        Returns:
-            List of PIL Image objects, one per page.
-        """
-        try:
-            with tempfile.TemporaryDirectory() as temp_dir:
-                images = convert_from_path(
-                    pdf_path,
-                    dpi=self.dpi,
-                    output_folder=temp_dir,
-                    fmt="png",
-                    thread_count=2,
-                )
-                # Convert to list to ensure images are loaded before temp dir cleanup
-                return list(images)
-        except Exception as e:
-            raise OCRError(f"Failed to convert PDF to images: {str(e)}")
-
-    def _ocr_image(self, image: Image.Image) -> str:
-        """
-        Perform OCR on a single image.
-
-        Args:
-            image: PIL Image object.
-
-        Returns:
-            Extracted text from the image.
-        """
-        # Preprocess image for better OCR results
-        processed = self._preprocess_image(image)
-
-        # Perform OCR
-        custom_config = r"--oem 3 --psm 6"
-        text = pytesseract.image_to_string(
-            processed,
-            lang=self.language,
-            config=custom_config
-        )
-        return text
-
-    def _preprocess_image(self, image: Image.Image) -> Image.Image:
-        """
-        Preprocess image to improve OCR accuracy.
-
-        Args:
-            image: Original PIL Image.
-
-        Returns:
-            Preprocessed PIL Image.
-        """
-        # Convert to grayscale
-        if image.mode != "L":
-            image = image.convert("L")
-
-        # Optional: Apply additional preprocessing
-        # - Contrast enhancement
-        # - Noise reduction
-        # - Deskewing
-
-        return image
-
-    def process_image(self, image_path: Path) -> str:
-        """
-        Extract text from a single image file.
-
-        Args:
-            image_path: Path to the image file.
-
-        Returns:
-            Extracted text.
-
-        Raises:
-            OCRError: If OCR processing fails.
-        """
-        image_path = Path(image_path)
-        if not image_path.exists():
-            raise OCRError(f"Image file not found: {image_path}")
-
-        try:
-            with Image.open(image_path) as image:
-                return self._ocr_image(image)
-        except Exception as e:
-            raise OCRError(f"Failed to process image: {str(e)}")
-
-    def is_available(self) -> bool:
-        """Check if Tesseract OCR is available."""
-        try:
-            pytesseract.get_tesseract_version()
-            return True
-        except Exception:
-            return False

app/utils/__init__.py

@@ -1,25 +0,0 @@
-"""
-Utility modules for the T1 PDF Processor.
-"""
-from .exceptions import (
-    PDFProcessorError,
-    PDFExtractionError,
-    OCRError,
-    MappingError,
-    PDFFillingError,
-    FileValidationError,
-    ConfigurationError,
-)
-from .logging import setup_logging, get_logger
-
-__all__ = [
-    "PDFProcessorError",
-    "PDFExtractionError",
-    "OCRError",
-    "MappingError",
-    "PDFFillingError",
-    "FileValidationError",
-    "ConfigurationError",
-    "setup_logging",
-    "get_logger",
-]

app/utils/exceptions.py

@@ -1,43 +0,0 @@
-"""
-Custom exception classes for the application.
-"""
-from typing import Optional, Any
-
-
-class PDFProcessorError(Exception):
-    """Base exception for PDF processor errors."""
-
-    def __init__(self, message: str, details: Optional[dict[str, Any]] = None):
-        super().__init__(message)
-        self.message = message
-        self.details = details or {}
-
-
-class PDFExtractionError(PDFProcessorError):
-    """Raised when PDF text extraction fails."""
-    pass
-
-
-class OCRError(PDFProcessorError):
-    """Raised when OCR processing fails."""
-    pass
-
-
-class MappingError(PDFProcessorError):
-    """Raised when field mapping fails."""
-    pass
-
-
-class PDFFillingError(PDFProcessorError):
-    """Raised when PDF form filling fails."""
-    pass
-
-
-class FileValidationError(PDFProcessorError):
-    """Raised when file validation fails."""
-    pass
-
-
-class ConfigurationError(PDFProcessorError):
-    """Raised when configuration is invalid."""
-    pass

app/utils/gemini_parser.py

@@ -1,129 +0,0 @@
-"""
-Google Gemini-based fallback parser for when regex extraction fails.
-"""
-import json
-from typing import Optional
-
-from app.config import settings
-from app.utils.logging import get_logger
-from app.utils.exceptions import PDFExtractionError
-
-logger = get_logger(__name__)
-
-
-class GeminiParser:
-    """
-    Uses Google Gemini models to extract T1 tax line values when regex fails.
-    """
-
-    def __init__(self, api_key: Optional[str] = None):
-        self.api_key = api_key or settings.google_api_key
-        self.model_name = settings.gemini_model
-        self._model = None
-
-        if self.api_key:
-            try:
-                import google.generativeai as genai
-                genai.configure(api_key=self.api_key)
-                self._model = genai.GenerativeModel(self.model_name)
-                logger.info(f"Gemini client initialized with model: {self.model_name}")
-            except ImportError:
-                logger.warning("google-generativeai package not installed")
-            except Exception as e:
-                logger.warning(f"Failed to initialize Gemini client: {e}")
-
-    def is_available(self) -> bool:
-        """Check if Gemini parsing is available."""
-        return self._model is not None
-
-    def extract_line_values(
-        self,
-        text: str,
-        line_numbers: Optional[list[str]] = None
-    ) -> dict[str, Optional[str]]:
-        """
-        Use Gemini to extract T1 tax line values from text.
-
-        Args:
-            text: Text content from PDF.
-            line_numbers: Specific line numbers to extract.
-
-        Returns:
-            Dictionary mapping line numbers to values.
-        """
-        if not self._model:
-            raise PDFExtractionError("Gemini client not available")
-
-        if line_numbers:
-            lines_str = ", ".join(line_numbers)
-            target_lines = f"Extract values for these specific lines: {lines_str}"
-        else:
-            target_lines = "Extract all T1 tax line values you can find"
-
-        prompt = f"""You are a Canadian T1 tax form data extractor.
-
-Given the following text extracted from a T1 tax return PDF, extract the line values.
-
-{target_lines}
-
-T1 tax lines are 5-digit numbers (like 15000, 23600, 26000) followed by dollar amounts.
-The format may vary:
-- "15000 Total Income: $50,000.00"
-- "Line 15000  50000"
-- "Total income (line 15000) 50,000"
-
-Return ONLY a valid JSON object mapping line numbers to their numeric values (without $ or commas).
-Example: {{"15000": "50000.00", "23600": "45000.00"}}
-
-If a line is not found, omit it from the response.
-
-TEXT:
-{text[:8000]}
-"""
-
-        try:
-            response = self._model.generate_content(
-                prompt,
-                generation_config={
-                    "temperature": 0.1,
-                    "max_output_tokens": 1000,
-                }
-            )
-
-            result_text = response.text.strip()
-
-            # Clean up markdown code blocks if present
-            if result_text.startswith("```"):
-                result_text = result_text.split("```")[1]
-                if result_text.startswith("json"):
-                    result_text = result_text[4:]
-                result_text = result_text.strip()
-
-            extracted = json.loads(result_text)
-            logger.info(f"Gemini extracted {len(extracted)} line values")
-
-            # Clean and validate values
-            cleaned = {}
-            for line_num, value in extracted.items():
-                if isinstance(value, (int, float)):
-                    cleaned[line_num] = str(value)
-                elif isinstance(value, str):
-                    clean_value = value.replace(",", "").replace("$", "").strip()
-                    try:
-                        float(clean_value)
-                        cleaned[line_num] = clean_value
-                    except ValueError:
-                        logger.warning(f"Invalid value for line {line_num}: {value}")
-
-            return cleaned
-
-        except json.JSONDecodeError as e:
-            logger.error(f"Failed to parse Gemini response as JSON: {e}")
-            raise PDFExtractionError(f"Gemini returned invalid JSON: {e}")
-        except Exception as e:
-            logger.error(f"Gemini extraction failed: {e}")
-            raise PDFExtractionError(f"Gemini extraction failed: {e}")
-
-
-# Global parser instance
-gemini_parser = GeminiParser()

app/utils/logging.py

@@ -1,41 +0,0 @@
-"""
-Logging utility module for consistent logging across the application.
-"""
-import sys
-from pathlib import Path
-
-from loguru import logger
-
-from app.config import settings
-
-
-def setup_logging() -> None:
-    """Configure application logging with loguru."""
-    # Remove default handler
-    logger.remove()
-
-    # Console handler
-    logger.add(
-        sys.stderr,
-        format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
-        level="DEBUG" if settings.debug else "INFO",
-        colorize=True,
-    )
-
-    # File handler
-    log_file = settings.base_dir / "logs" / "app.log"
-    log_file.parent.mkdir(parents=True, exist_ok=True)
-
-    logger.add(
-        log_file,
-        format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
-        level="DEBUG",
-        rotation="10 MB",
-        retention="7 days",
-        compression="zip",
-    )
-
-
-def get_logger(name: str):
-    """Get a logger instance with the given name."""
-    return logger.bind(name=name)

app_gradio.py

@@ -1,231 +0,0 @@
-"""
-Gradio UI for T1 Tax PDF Processor - Hugging Face Spaces Version
-"""
-import gradio as gr
-import tempfile
-import json
-from pathlib import Path
-
-from app.extractor import PDFExtractor
-from app.ocr import OCRProcessor
-from app.mapper import FieldMapper
-from app.filler import PDFFiller
-from app.utils.logging import get_logger
-
-logger = get_logger(__name__)
-
-# Initialize components
-extractor = PDFExtractor()
-mapper = FieldMapper()
-filler = PDFFiller()
-
-# Try to initialize OCR
-try:
-    ocr = OCRProcessor()
-    ocr_available = ocr.is_available()
-except:
-    ocr = None
-    ocr_available = False
-
-
-def process_pdfs(source_pdf, template_pdf, use_ocr=False, flatten_output=False):
-    """
-    Process PDFs: extract from source and fill template.
-
-    Args:
-        source_pdf: Path to source T1 PDF
-        template_pdf: Path to template PDF form
-        use_ocr: Force OCR processing
-        flatten_output: Flatten output PDF
-
-    Returns:
-        tuple: (output_pdf_path, extraction_table_data, status_message)
-    """
-    try:
-        if not source_pdf or not template_pdf:
-            return None, [], "❌ Please upload both PDFs"
-
-        source_path = Path(source_pdf.name)
-        template_path = Path(template_pdf.name)
-
-        # Step 1: Extract data
-        has_text = extractor.has_text_content(source_path)
-
-        if has_text and not use_ocr:
-            result = extractor.extract_all_data(source_path)
-            method = "text"
-        elif ocr_available:
-            text = ocr.process_pdf(source_path)
-            line_values = extractor.extract_line_values(text)
-            result = {
-                "text": text,
-                "line_values": line_values,
-                "has_text": False,
-                "extraction_method": "ocr"
-            }
-            method = "ocr"
-        else:
-            result = extractor.extract_all_data(source_path)
-            method = "text_fallback"
-
-        line_values = result["line_values"]
-
-        if not line_values:
-            return None, [], "⚠️ No T1 line values found in source PDF"
-
-        # Step 2: Map values
-        mapped = mapper.map_values(line_values)
-
-        # Step 3: Fill template
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as output_file:
-            output_path = Path(output_file.name)
-
-        filler.fill_form(
-            template_path=template_path,
-            output_path=output_path,
-            field_values=mapped,
-            flatten=flatten_output
-        )
-
-        # Create table data
-        table_data = []
-        for line_num, value in line_values.items():
-            if value:
-                field_name = mapper.get_field_name(line_num) or "unmapped"
-                table_data.append([line_num, field_name, value])
-
-        status = f"✅ Success! Extracted {len(line_values)} lines using {method} method"
-
-        return str(output_path), table_data, status
-
-    except Exception as e:
-        logger.error(f"Processing error: {e}")
-        return None, [], f"❌ Error: {str(e)}"
-
-
-def get_current_mappings():
-    """Get current field mappings as formatted text."""
-    mappings = mapper.get_all_mappings()
-    output = "## Current Field Mappings\n\n"
-    output += "| Line # | Field Name | Description |\n"
-    output += "|--------|------------|-------------|\n"
-
-    for line_num, info in sorted(mappings.items())[:20]:  # Show first 20
-        field = info.get("field", "")
-        desc = info.get("description", "")
-        output += f"| {line_num} | {field} | {desc} |\n"
-
-    if len(mappings) > 20:
-        output += f"\n... and {len(mappings) - 20} more mappings"
-
-    return output
-
-
-# Create Gradio Interface
-with gr.Blocks(theme=gr.themes.Soft(), title="T1 Tax PDF Processor") as demo:
-
-    gr.Markdown("""
-    # 📄 T1 Tax PDF Processor
-
-    Extract data from Canadian T1 tax return PDFs and automatically fill PDF forms.
-
-    **How it works:**
-    1. Upload your T1 tax return (source) and target form template (template)
-    2. Click "Process PDFs" to extract and fill
-    3. Download the filled PDF
-
-    ⚠️ **Demo Only** - Do not upload real sensitive tax documents to public demos!
-    """)
-
-    with gr.Row():
-        with gr.Column():
-            source_pdf = gr.File(
-                label="📄 Source T1 PDF (with data)",
-                file_types=[".pdf"],
-                type="filepath"
-            )
-
-            template_pdf = gr.File(
-                label="📝 Target Template PDF (to fill)",
-                file_types=[".pdf"],
-                type="filepath"
-            )
-
-    with gr.Row():
-        use_ocr = gr.Checkbox(
-            label="Force OCR (for scanned PDFs)",
-            value=False,
-            info="Enable if source PDF is scanned/image-based"
-        )
-
-        flatten_output = gr.Checkbox(
-            label="Flatten output PDF",
-            value=False,
-            info="Make filled fields non-editable"
-        )
-
-    process_btn = gr.Button("🚀 Process PDFs", variant="primary", size="lg")
-
-    status_text = gr.Markdown("Ready to process...")
-
-    with gr.Row():
-        output_pdf = gr.File(
-            label="📥 Download Filled PDF",
-            type="filepath"
-        )
-
-    with gr.Row():
-        extraction_table = gr.Dataframe(
-            headers=["Line Number", "Field Name", "Value"],
-            label="📊 Extracted T1 Data",
-            wrap=True
-        )
-
-    # Process button action
-    process_btn.click(
-        fn=process_pdfs,
-        inputs=[source_pdf, template_pdf, use_ocr, flatten_output],
-        outputs=[output_pdf, extraction_table, status_text]
-    )
-
-    # Mappings section
-    with gr.Accordion("🗺️ Field Mappings Configuration", open=False):
-        gr.Markdown("""
-        View current T1 line number to PDF field mappings.
-        To modify mappings, update `config/t1_mapping.json` in the repository.
-        """)
-
-        mappings_display = gr.Markdown(get_current_mappings())
-
-    gr.Markdown("""
-    ---
-
-    ### 📚 Common T1 Line Numbers
-
-    | Line | Description |
-    |------|-------------|
-    | 10100 | Employment income |
-    | 15000 | Total income |
-    | 23600 | Net income |
-    | 26000 | Taxable income |
-    | 42000 | Net federal tax |
-    | 48400 | Refund |
-    | 48500 | Balance owing |
-
-    ### 🔧 Tech Stack
-
-    Built with: FastAPI, pdfplumber, PyMuPDF, Tesseract OCR, Gradio
-
-    ### ⚠️ Privacy Notice
-
-    This is a demonstration application. Files are processed temporarily and not stored permanently.
-    **Do not upload real tax documents containing sensitive personal information.**
-    """)
-
-# Launch
-if __name__ == "__main__":
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        share=False
-    )

config/t1_mapping.json

@@ -1,50 +0,0 @@
-{
-  "15000": {
-    "field": "total_income",
-    "description": "Total income before deductions"
-  },
-  "23600": {
-    "field": "net_income",
-    "description": "Net income after deductions"
-  },
-  "26000": {
-    "field": "taxable_income",
-    "description": "Taxable income"
-  },
-  "10100": {
-    "field": "employment_income",
-    "description": "Employment income (line 10100)"
-  },
-  "12100": {
-    "field": "interest_income",
-    "description": "Interest and other investment income"
-  },
-  "20800": {
-    "field": "rrsp_deduction",
-    "description": "RRSP/PRPP deduction"
-  },
-  "30000": {
-    "field": "basic_personal_amount",
-    "description": "Basic personal amount"
-  },
-  "42000": {
-    "field": "net_federal_tax",
-    "description": "Net federal tax"
-  },
-  "43500": {
-    "field": "total_payable",
-    "description": "Total tax payable"
-  },
-  "43700": {
-    "field": "total_tax_deducted",
-    "description": "Total income tax deducted"
-  },
-  "48400": {
-    "field": "refund",
-    "description": "Refund amount"
-  },
-  "48500": {
-    "field": "balance_owing",
-    "description": "Balance owing"
-  }
-}

requirements.txt

@@ -1,38 +0,0 @@
-# Web Framework
-fastapi==0.109.2
-uvicorn[standard]==0.27.1
-python-multipart==0.0.9
-gradio==5.0.2
-
-# PDF Processing
-pdfplumber==0.10.4
-PyMuPDF==1.24.14
-pdfrw==0.4
-pypdf==4.0.2
-reportlab==4.1.0
-
-# OCR
-pytesseract==0.3.10
-pdf2image==1.17.0
-Pillow==10.4.0
-
-
-# Data Processing
-pydantic==2.8.2
-pydantic-settings==2.4.0
-
-# Google Generative AI (optional fallback)
-google-generativeai==0.4.1
-
-# Utilities
-python-dotenv==1.0.1
-aiofiles==23.2.1
-jinja2==3.1.3
-
-# Logging
-loguru==0.7.2
-
-# Testing
-pytest==8.0.1
-pytest-asyncio==0.23.5
-httpx==0.26.0