arrr-erp/app/modules/bank_statement_analyzer/service.py

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 .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 _resolve_document_storage_root() -> Path:
    """Resolve the same automatic document root without importing documents.services.

    Importing app.modules.documents.services here creates a circular import through
    clients.__init__ -> clients.ui -> documents.services. This local resolver
    intentionally mirrors the existing document module's storage-root rules while
    keeping the bank analyzer independent of document models and UI modules.
    """
    configured = (os.getenv("DOCUMENT_STORAGE_ROOT") or "").strip()
    if configured:
        configured_path = Path(configured).expanduser()
        if configured_path.is_absolute():
            return configured_path
        return (Path.cwd() / configured_path).resolve()

    if os.name == "nt":
        anchor = Path.cwd().anchor or (os.environ.get("SystemDrive", "C:") + "\\")
        return Path(anchor) / "AuditFirmERPDocuments" / "engagement_documents"

    return (Path.cwd() / "documents" / "engagement_documents").resolve()


def _now() -> datetime:
    return datetime.now(timezone.utc)


def _root() -> Path:
    return _resolve_document_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 = [file for file in files if file and (file.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 = "",
    bank_selection: str = "auto",
    financial_year: str = "",
    classification_enabled: bool = True,
) -> dict:
    metas, all_df, unique_df = analyze_files(
        paths,
        customer_override.strip(),
        account_override.strip(),
        bank_hint=bank_selection,
        classification_enabled=classification_enabled,
    )
    output = output_dir / "Bank_Statement_Analysis.xlsx"
    export_excel(
        output,
        metas,
        all_df,
        unique_df,
        financial_year=financial_year.strip(),
        selected_bank=bank_selection,
        classification_enabled=classification_enabled,
    )
    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,
        "review_items": int(unique_df.review_note.fillna("").ne("").sum()) if not unique_df.empty else 0,
        "categories": int(unique_df.category.nunique()) if not unique_df.empty else 0,
        "banks": sorted({meta.bank_name for meta in metas}),
        "customer_name": next((meta.customer_name for meta in metas if meta.customer_name), ""),
        "account_number": next((meta.account_number for meta in metas if meta.account_number), ""),
        "financial_year": financial_year.strip(),
        "classification_enabled": classification_enabled,
        "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)