from __future__ import annotations import json import os import re import shutil import uuid from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Iterable from fastapi import UploadFile from app.modules.documents.services import DEFAULT_STORAGE_ROOT from .analyzer import analyze_files, export_excel ALLOWED_ROLES = {"Partner", "Manager", "Branch Manager", "Staff", "Employee", "Consultant"} MAX_FILES = int(os.getenv("BANK_ANALYZER_MAX_FILES", "24")) MAX_FILE_BYTES = int(os.getenv("BANK_ANALYZER_MAX_FILE_MB", "50")) * 1024 * 1024 RETENTION_HOURS = int(os.getenv("BANK_ANALYZER_FAILED_RETENTION_HOURS", "24")) _SAFE = re.compile(r"[^A-Za-z0-9._-]+") def _now() -> datetime: return datetime.now(timezone.utc) def _root() -> Path: """Return the automatic ERP work-storage root. The document module already resolves its storage location for the current deployment. Bank-statement jobs use a sibling ``Work`` folder so no new environment variable or separate path configuration is required. """ return DEFAULT_STORAGE_ROOT.parent / "Work" def _segment(value: object, default: str = "NA") -> str: text = _SAFE.sub("_", str(value or default).strip()).strip("._-") return text[:80] or default def role_bucket(roles: Iterable[str]) -> str | None: role_set = set(roles) for role, bucket in (("Partner", "Partner"), ("Manager", "Manager"), ("Branch Manager", "Manager"), ("Staff", "Staff"), ("Employee", "Staff"), ("Consultant", "Consultant")): if role in role_set: return bucket return None def can_use(roles: Iterable[str]) -> bool: return bool(set(roles) & ALLOWED_ROLES) def _user_root(user, roles: Iterable[str]) -> Path: bucket = role_bucket(roles) if not bucket: raise PermissionError("Bank Statement Analyzer is available only to Partner, Manager, Staff and Consultant roles.") label = _segment(getattr(user, "full_name", None) or getattr(user, "email", None) or f"user_{user.id}") return _root() / bucket / f"{int(user.id)}_{label}" / "Bank_Statement_Analyzer" def cleanup_expired(user, roles: Iterable[str]) -> None: base = _user_root(user, roles) if not base.exists(): return cutoff = _now() - timedelta(hours=RETENTION_HOURS) for child in base.iterdir(): try: modified = datetime.fromtimestamp(child.stat().st_mtime, tz=timezone.utc) if child.is_dir() and modified < cutoff: shutil.rmtree(child, ignore_errors=True) except OSError: continue def create_job(user, roles: Iterable[str]) -> tuple[str, Path, Path]: cleanup_expired(user, roles) job_id = uuid.uuid4().hex job = _user_root(user, roles) / job_id input_dir = job / "Input" output_dir = job / "Output" input_dir.mkdir(parents=True, exist_ok=False) output_dir.mkdir(parents=True, exist_ok=False) return job_id, input_dir, output_dir def _validate_pdf_header(data: bytes) -> None: if not data.startswith(b"%PDF-"): raise ValueError("Only genuine PDF files are allowed.") async def save_uploads(files: list[UploadFile], input_dir: Path) -> list[Path]: usable = [f for f in files if f and (f.filename or "").strip()] if not usable: raise ValueError("Please select at least one PDF bank statement.") if len(usable) > MAX_FILES: raise ValueError(f"A maximum of {MAX_FILES} PDF files can be analyzed in one job.") saved: list[Path] = [] for index, upload in enumerate(usable, start=1): name = Path(upload.filename or f"statement_{index}.pdf").name if Path(name).suffix.lower() != ".pdf": raise ValueError(f"{name}: only PDF files are allowed.") safe_name = f"{index:02d}_{_segment(Path(name).stem, f'statement_{index}')}.pdf" target = input_dir / safe_name size = 0 first = b"" with target.open("wb") as handle: while True: chunk = await upload.read(1024 * 1024) if not chunk: break if not first: first = chunk[:8] size += len(chunk) if size > MAX_FILE_BYTES: raise ValueError(f"{name}: file exceeds the {MAX_FILE_BYTES // (1024 * 1024)} MB limit.") handle.write(chunk) _validate_pdf_header(first) saved.append(target) return saved def analyze_job(*, user, roles: Iterable[str], job_id: str, paths: list[Path], output_dir: Path, customer_override: str = "", account_override: str = "") -> dict: metas, all_df, unique_df = analyze_files(paths, customer_override.strip(), account_override.strip()) output = output_dir / "Bank_Statement_Analysis.xlsx" export_excel(output, metas, all_df, unique_df) summary = { "job_id": job_id, "owner_user_id": int(user.id), "created_at": _now().isoformat(), "statement_count": len(metas), "rows_extracted": int(len(all_df)), "unique_transactions": int(len(unique_df)), "exact_duplicate_rows": int(all_df.exact_duplicate.sum()) if not all_df.empty else 0, "possible_duplicate_rows": int(all_df.possible_duplicate.sum()) if not all_df.empty else 0, "banks": sorted({m.bank_name for m in metas}), "customer_name": next((m.customer_name for m in metas if m.customer_name), ""), "account_number": next((m.account_number for m in metas if m.account_number), ""), "output_file": output.name, } (output_dir.parent / "job.json").write_text(json.dumps(summary, indent=2), encoding="utf-8") return summary def resolve_owned_job(user, roles: Iterable[str], job_id: str) -> tuple[Path, dict]: if not re.fullmatch(r"[a-f0-9]{32}", job_id or ""): raise FileNotFoundError("Analysis job not found.") job = _user_root(user, roles) / job_id meta_path = job / "job.json" if not meta_path.is_file(): raise FileNotFoundError("Analysis job not found.") meta = json.loads(meta_path.read_text(encoding="utf-8")) if int(meta.get("owner_user_id", 0)) != int(user.id): raise PermissionError("You cannot access another user's analysis job.") return job, meta def delete_job(job: Path) -> None: shutil.rmtree(job, ignore_errors=True)