Fix concrete pipeline bugs and add fast unit-test CI

.github/workflows/ci.yml

@@ -0,0 +1,31 @@
+name: CI
+
+# Fast, dependency-light checks that run on every PR and push. This is the
+# required gate. The full pipeline smoke test lives in smoke-test.yml and is
+# run manually because it needs the heavy extraction stack and network access.
+on:
+  pull_request:
+    branches:
+      - main
+      - develop
+  push:
+    branches:
+      - main
+
+jobs:
+  unit-tests:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.11'
+
+      - name: Install dev dependencies
+        run: pip install -r requirements-dev.txt
+
+      - name: Run unit tests
+        run: python -m pytest

.github/workflows/smoke-test.yml

@@ -1,28 +1,29 @@
-name: Smoke Test   # Name of the workflow, shown in GitHub Actions tab
+name: Smoke Test (full pipeline)
 
+# The full DVC pipeline smoke test. It downloads filings and runs the heavy
+# extraction stack (torch, docling, layout models), so it is NOT part of the
+# required per-PR gate — it is slow and depends on external downloads. Trigger
+# it manually from the Actions tab when you want an end-to-end run.
 on:
-  pull_request:    # Trigger workflow when a pull request is created/updated
-    branches:
-      - develop    # Run workflow only if PR targets the 'develop' branch
-      - main       # Or if PR targets the 'main' branch
+  workflow_dispatch:
 
 jobs:
-  smoke-test:      
-    runs-on: ubuntu-latest   # Run the job on the latest Ubuntu runner provided by GitHub
+  smoke-test:
+    runs-on: ubuntu-latest
 
-    steps:        
-      - name: Checkout repository   # Fetch the repository code
-        uses: actions/checkout@v4   # Use the official GitHub Action to check out code
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
 
-      - name: Set up Python        
+      - name: Set up Python
         uses: actions/setup-python@v5
         with:
-          python-version: '3.10'    
+          python-version: '3.11'
 
-      - name: Install dependencies  # Install project dependencies
+      - name: Install dependencies
         run: |
-          pip install -r requirements.txt   # Install packages listed in requirements.txt
-          pip install dvc[s3]              # Install DVC with S3 support (change to dvc[gdrive] or dvc[azure] if needed)
+          pip install -r requirements.txt
+          pip install dvc                 # No remote is configured; plain DVC is enough for `dvc repro`.
 
-      - name: Run smoke test        # Execute smoke test script
-        run: bash tests/smoke.sh    # Run the smoke.sh 
+      - name: Run smoke test
+        run: bash tests/smoke.sh

dvc.yaml

@@ -8,7 +8,8 @@ stages:
       - download.filing_types
       - download.output_dir
     outs:
-      - data/raw/META/
+      - data/raw/10-K/
+      - data/raw/10-Q/
 
   extract_text:
     cmd: python src/extractors/text_extractor.py

pytest.ini

@@ -0,0 +1,9 @@
+[pytest]
+# The fast, dependency-light suite. The integration tests under
+# tests/integration and evaluation/ require the heavy extraction stack and are
+# not collected here.
+testpaths = tests/unit
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+addopts = -q

requirements-dev.txt

@@ -0,0 +1,6 @@
+# Lightweight dependencies for the fast unit-test suite (tests/unit).
+# These deliberately exclude the heavy extraction stack (torch, docling,
+# detectron2, pdfplumber) so the suite runs quickly and deterministically in CI.
+pytest>=8.0.0
+requests>=2.31.0
+PyYAML>=6.0

src/downloaders/sec_downloader.py

@@ -5,6 +5,7 @@
 """
 
 import os
+import re
 import sys
 import yaml
 import requests
@@ -59,33 +60,89 @@ def load_params(self, params_file: str = "params.yaml") -> Dict[str, Any]:
             logger.error(f"Error parsing {params_file}: {e}")
             return {}
 
+    @staticmethod
+    def _extract_gdrive_confirm_token(response: "requests.Response") -> Optional[str]:
+        """Return Google Drive's download-confirmation token, if the response is an interstitial page.
+
+        Large Google Drive files can't be virus-scanned, so the first request returns an
+        HTML page instead of the file. We must resend the request with the confirm token to
+        get the real bytes. Without this the downloader silently saves the HTML page as a PDF.
+        """
+        # Older style: token is set as a cookie named ``download_warning*``.
+        for key, value in response.cookies.items():
+            if key.startswith("download_warning"):
+                return value
+
+        # Newer style: token is embedded in the HTML form of the interstitial page.
+        content_type = response.headers.get("Content-Type", "")
+        if "text/html" in content_type:
+            match = re.search(r'name="confirm"\s+value="([^"]+)"', response.text)
+            if match:
+                return match.group(1)
+            match = re.search(r'confirm=([0-9A-Za-z_\-]+)', response.text)
+            if match:
+                return match.group(1)
+        return None
+
     def download_file(self, url: str, output_path: Path, filename: str) -> bool:
-        """Download a file from URL to the specified path."""
+        """Download a file from URL and verify it is a real PDF before keeping it."""
+        file_path = output_path / filename
         try:
             logger.info(f"Downloading {filename} from {url}")
-            
+
             headers = {
                 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
             }
-
-            response = requests.get(url, headers=headers, stream=True, timeout=30)
+
+            session = requests.Session()
+            response = session.get(url, headers=headers, stream=True, timeout=30)
             response.raise_for_status()
-
-            file_path = output_path / filename
+
+            # Handle Google Drive's large-file confirmation interstitial.
+            if "drive.google.com" in url or "drive.usercontent.google.com" in url:
+                token = self._extract_gdrive_confirm_token(response)
+                if token:
+                    logger.info("Google Drive confirmation required, retrying with token")
+                    response = session.get(
+                        url, headers=headers, params={'confirm': token},
+                        stream=True, timeout=30
+                    )
+                    response.raise_for_status()
+
             with open(file_path, 'wb') as f:
                 for chunk in response.iter_content(chunk_size=8192):
                     if chunk:
                         f.write(chunk)
-
+
+            # Guard against silently saving an HTML error/interstitial page as a ".pdf".
+            if not self._is_pdf(file_path):
+                logger.error(
+                    f"Downloaded {filename} is not a valid PDF (likely an HTML error page). "
+                    "Discarding the file."
+                )
+                file_path.unlink(missing_ok=True)
+                return False
+
             file_size = file_path.stat().st_size
             logger.info(f"Successfully downloaded {filename} ({file_size:,} bytes)")
             return True
-            
+
         except requests.RequestException as e:
             logger.error(f"Failed to download {filename}: {e}")
+            file_path.unlink(missing_ok=True)
             return False
         except Exception as e:
             logger.error(f"Unexpected error downloading {filename}: {e}")
+            file_path.unlink(missing_ok=True)
+            return False
+
+    @staticmethod
+    def _is_pdf(file_path: Path) -> bool:
+        """Return True if the file starts with the PDF magic bytes (``%PDF``)."""
+        try:
+            with open(file_path, 'rb') as f:
+                return f.read(5).startswith(b'%PDF')
+        except OSError:
             return False
 
     def get_filing_url(self, company: str, fiscal_year: int, filing_type: str) -> Optional[str]:
@@ -104,32 +161,36 @@ def get_filing_url(self, company: str, fiscal_year: int, filing_type: str) -> Op
     def download_filings(self, companies: list, fiscal_years: list, filing_types: list) -> Dict[str, bool]:
         """Download specified filings for given companies and years."""
         results = {}
-        
+
         for company in companies:
-            company_dir = self.output_dir / company.upper()
-            company_dir.mkdir(exist_ok=True)
-
             for year in fiscal_years:
                 for filing_type in filing_types:
+                    # Organize downloads by filing type (e.g. data/raw/10-K/) so the
+                    # downstream extractors, which read from data/raw/<FILING_TYPE>/,
+                    # find their inputs. This also keeps dvc.yaml's `outs` consistent
+                    # with what is actually produced.
+                    filing_dir = self.output_dir / filing_type.upper()
+                    filing_dir.mkdir(parents=True, exist_ok=True)
+
                     # Generate filename
                     filename = f"{year}_{company.lower()}_{filing_type.lower()}.pdf"
-                    
+
                     # Check if file already exists
-                    file_path = company_dir / filename
+                    file_path = filing_dir / filename
                     if file_path.exists():
                         logger.info(f"File already exists: {file_path}")
                         results[f"{company}_{year}_{filing_type}"] = True
                         continue
-                    
+
                     # Get download URL
                     url = self.get_filing_url(company, year, filing_type)
                     if not url:
                         logger.error(f"No URL found for {company} {year} {filing_type}")
                         results[f"{company}_{year}_{filing_type}"] = False
                         continue
-                    
+
                     # Download the file
-                    success = self.download_file(url, company_dir, filename)
+                    success = self.download_file(url, filing_dir, filename)
                     results[f"{company}_{year}_{filing_type}"] = success
 
                     # Small delay to be respectful to servers

src/extractors/layout_detector.py

@@ -638,8 +638,8 @@ def _create_comparison_metrics(self, d2_perf: Dict, lmv3_perf: Dict) -> Dict:
     def process_single_10k_pdf(self, max_pages: int = None) -> Dict:
         """Process single 10-K PDF for focused analysis"""
 
-        # Target specific 10-K file
-        raw_dir = Path("/Users/HemanthRayudu/Profession/Assignments/DAMG/Docuparse/data/raw")
+        # Target specific 10-K file (relative to the repo root)
+        raw_dir = Path("data/raw")
         target_pdf = raw_dir / "10-K" / "2024_meta_10-k.pdf"
 
         if not target_pdf.exists():
@@ -794,7 +794,7 @@ def _print_10k_results(self, summary: Dict):
 
 def estimate_processing_time():
     """Estimate processing time by testing one PDF"""
-    raw_dir = Path("/Users/HemanthRayudu/Profession/Assignments/DAMG/Docuparse/data/raw")
+    raw_dir = Path("data/raw")
     pdf_files = list(raw_dir.rglob("*.pdf"))
 
     if not pdf_files:

src/extractors/table_extractor.py

@@ -8,7 +8,7 @@
 from typing import List, Dict, Optional, Tuple, Any
 import logging
 import json
-from datetime import datetime
+from datetime import datetime, timezone
 from dataclasses import dataclass, asdict
 import time
 
@@ -759,7 +759,7 @@ def _save_page_tables(self, doc_id: str, page_num: int, page_result: PageTableEx
                 'rows': table.rows,
                 'cols': table.cols,
                 'extraction_time': table.extraction_time,
-                'timestamp': datetime.utcnow().isoformat(),
+                'timestamp': datetime.now(timezone.utc).isoformat(),
                 'error': table.error,
                 'file_path': str(table_path)
             }
@@ -814,7 +814,7 @@ def _save_extraction_log(self, doc_id: str, results: List[PageTableExtraction]):
 
         log_data = {
             'doc_id': doc_id,
-            'timestamp': datetime.utcnow().isoformat(),
+            'timestamp': datetime.now(timezone.utc).isoformat(),
             'total_pages': len(results),
             'total_tables': sum(r.total_tables for r in results),
             'statistics': {
@@ -1006,7 +1006,7 @@ def process_single_pdf(pdf_file: Path) -> Dict:
                     'quarter': quarter,
                     'company': company,
                     'source_file': pdf_file.name,
-                    'processed_at': datetime.utcnow().isoformat()
+                    'processed_at': datetime.now(timezone.utc).isoformat()
                 },
                 'extraction_stats': results['statistics'],
                 'file_paths': {
@@ -1108,7 +1108,7 @@ def process_single_pdf(pdf_file: Path) -> Dict:
     if not target_pdf.exists():
         logger.warning(f"Target PDF not found: {target_pdf}")
         return {
-            'timestamp': datetime.utcnow().isoformat(),
+            'timestamp': datetime.now(timezone.utc).isoformat(),
             'total_statistics': total_stats,
             'file_results': all_results,
             'output_directory': str(output_dir)
@@ -1143,7 +1143,7 @@ def process_single_pdf(pdf_file: Path) -> Dict:
 
     # Save comprehensive processing log
     processing_log = {
-        'timestamp': datetime.utcnow().isoformat(),
+        'timestamp': datetime.now(timezone.utc).isoformat(),
         'total_statistics': total_stats,
         'file_results': all_results,
         'output_directory': str(output_dir)

src/extractors/text_extractor.py

@@ -7,7 +7,7 @@
 from typing import Dict, List, Optional, Tuple
 import logging
 import json
-from datetime import datetime
+from datetime import datetime, timezone
 import numpy as np
 from dataclasses import dataclass, asdict
 
@@ -238,7 +238,7 @@ def _save_page_text(self, doc_id: str, page_num: int, page_result: PageExtractio
             'word_count': page_result.word_count,
             'char_count': page_result.char_count,
             'extraction_time': page_result.extraction_time,
-            'timestamp': datetime.utcnow().isoformat(),
+            'timestamp': datetime.now(timezone.utc).isoformat(),
             'error': page_result.error
         }
 
@@ -260,7 +260,7 @@ def _save_extraction_log(self, doc_id: str, results: List[PageExtraction]):
 
         log_data = {
             'doc_id': doc_id,
-            'timestamp': datetime.utcnow().isoformat(),
+            'timestamp': datetime.now(timezone.utc).isoformat(),
             'total_pages': len(results),
             'ocr_pages': [r.page_num for r in results if r.ocr_used],
             'failed_pages': [r.page_num for r in results if r.error],
@@ -437,7 +437,7 @@ def process_single_pdf(pdf_file: Path, doc_type: str, year: str, quarter: str =
                     'quarter': quarter,
                     'company': 'META',
                     'source_file': pdf_file.name,
-                    'processed_at': datetime.utcnow().isoformat()
+                    'processed_at': datetime.now(timezone.utc).isoformat()
                 },
                 'extraction_stats': results['statistics'],
                 'file_paths': {
@@ -547,7 +547,7 @@ def process_single_pdf(pdf_file: Path, doc_type: str, year: str, quarter: str =
             serializable_results[key] = result
 
     processing_log = {
-        'timestamp': datetime.utcnow().isoformat(),
+        'timestamp': datetime.now(timezone.utc).isoformat(),
         'total_statistics': total_stats,
         'file_results': serializable_results,
         'output_directory': str(output_dir)

src/representations/metadata_storage_formats.py

@@ -40,8 +40,9 @@ def convert_metadata(input_file: Path, output_dir: Path):
 
 
 if __name__ == "__main__":
-    input_dir = Path(r"C:\Users\Pauline\Desktop\PEI\NEU_assignments\DAMG 7245_(2025 Fall)\data\parsed\metadata")
-    output_dir = Path(r"C:\Users\Pauline\Desktop\PEI\NEU_assignments\DAMG 7245_(2025 Fall)\data\parsed\metadata_representations")
+    # Paths are relative to the repo root so this runs on any machine.
+    input_dir = Path("data/parsed/metadata")
+    output_dir = Path("data/parsed/metadata_representations")
     output_dir.mkdir(parents=True, exist_ok=True)
 
     # find all JSONL files