diff --git a/app/modules/bank_statement_analyzer/service.py b/app/modules/bank_statement_analyzer/service.py index ac90d2b..a30a6bd 100644 --- a/app/modules/bank_statement_analyzer/service.py +++ b/app/modules/bank_statement_analyzer/service.py @@ -11,8 +11,6 @@ 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"} @@ -22,18 +20,34 @@ 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 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" + return _resolve_document_storage_root().parent / "Work" def _segment(value: object, default: str = "NA") -> str: @@ -92,7 +106,7 @@ def _validate_pdf_header(data: bytes) -> None: 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()] + 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: @@ -122,10 +136,36 @@ async def save_uploads(files: list[UploadFile], input_dir: Path) -> list[Path]: 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()) +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) + 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), @@ -135,9 +175,13 @@ def analyze_job(*, user, roles: Iterable[str], job_id: str, paths: list[Path], o "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), ""), + "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")