from __future__ import annotations import re import shutil from dataclasses import dataclass from pathlib import Path from sqlalchemy import select from core.db import SessionLocal from core.models import Candidate from services.ingestion_service import IngestionService DEFAULT_REFERENCE_RESUMES_DIR = ( Path(__file__).resolve().parents[1] / "data" / "reference_resumes" ) FALLBACK_REFERENCE_RESUME_DIRS = [ Path("E:/SaturnIQ/engazeiq-adk") / "multiagent_platform_backend" / "services" / "agent_services" / "mcp_tools" / "organized_docs" / "resumes", Path("E:/SaturnIQ/engazeiq-adk") / "multiagent_platform_backend" / "services" / "agent_services" / "mcp_tools" / "organized_docs" / "resumes_", ] @dataclass class ReferenceIngestionSummary: source_dir: str requested_limit: int files_seen: int ingested: int skipped: int failed: int failures: list[dict[str, str]] def _resolve_source_dir(source_dir: str | None) -> Path: if source_dir: explicit = Path(source_dir) if explicit.exists() and explicit.is_dir(): return explicit raise FileNotFoundError(f"Reference resume directory not found: {explicit}") candidate_dirs = [DEFAULT_REFERENCE_RESUMES_DIR, *FALLBACK_REFERENCE_RESUME_DIRS] for directory in candidate_dirs: if directory.exists() and directory.is_dir() and any(directory.rglob("*.pdf")): return directory checked = "\n".join(str(path) for path in candidate_dirs) raise FileNotFoundError(f"No usable reference resume directory found. Checked:\n{checked}") def _normalize_external_id(file_path: Path) -> str: normalized = re.sub(r"[^a-zA-Z0-9]+", "_", file_path.stem).strip("_") if not normalized: normalized = "resume" return f"ref_{normalized.lower()}" def sync_reference_resumes_to_local( *, max_files: int = 120, source_dir: str | None = None, ) -> dict[str, object]: resolved_source = _resolve_source_dir(source_dir) DEFAULT_REFERENCE_RESUMES_DIR.mkdir(parents=True, exist_ok=True) copied = 0 skipped = 0 seen_names: set[str] = set() pdf_files = sorted(resolved_source.rglob("*.pdf"))[: max(1, max_files)] for idx, file_path in enumerate(pdf_files, start=1): base_name = file_path.name target_name = base_name same_name_target = DEFAULT_REFERENCE_RESUMES_DIR / base_name if same_name_target.exists() and same_name_target.stat().st_size == file_path.stat().st_size: skipped += 1 seen_names.add(base_name.lower()) continue if target_name.lower() in seen_names or (DEFAULT_REFERENCE_RESUMES_DIR / target_name).exists(): target_name = f"{idx:04d}_{base_name}" seen_names.add(target_name.lower()) target_path = DEFAULT_REFERENCE_RESUMES_DIR / target_name if target_path.exists(): skipped += 1 continue shutil.copy2(file_path, target_path) copied += 1 return { "source_dir": str(resolved_source), "target_dir": str(DEFAULT_REFERENCE_RESUMES_DIR), "files_seen": len(pdf_files), "copied": copied, "skipped": skipped, } def ingest_reference_resumes( *, limit: int = 25, source_dir: str | None = None, force_reingest: bool = False, ) -> ReferenceIngestionSummary: base_dir = _resolve_source_dir(source_dir) service = IngestionService() pdf_files = sorted(base_dir.rglob("*.pdf"))[: max(1, limit)] failures: list[dict[str, str]] = [] ingested = 0 skipped = 0 with SessionLocal() as db: for pdf in pdf_files: external_id = _normalize_external_id(pdf) if not force_reingest: exists = db.execute( select(Candidate.id).where(Candidate.external_id == external_id) ).scalar_one_or_none() if exists is not None: skipped += 1 continue try: file_bytes = pdf.read_bytes() service.ingest_pdf( db, file_name=pdf.name, file_bytes=file_bytes, external_id=external_id, ) ingested += 1 except Exception as exc: failures.append({"file": str(pdf), "error": str(exc)}) return ReferenceIngestionSummary( source_dir=str(base_dir), requested_limit=max(1, limit), files_seen=len(pdf_files), ingested=ingested, skipped=skipped, failed=len(failures), failures=failures, )