From 897c7dad38325d477225641a512238dac32a2138 Mon Sep 17 00:00:00 2001 From: A R R R Associates Date: Mon, 13 Jul 2026 15:02:58 +0530 Subject: [PATCH] Add queued bank statement analysis with global concurrency limits --- .../20260713_bank_statement_analysis_queue.py | 50 ++ app/modules/bank_statement_analyzer/models.py | 38 ++ .../bank_statement_analyzer/service.py | 465 +++++++++++++----- .../bank_statement_analyzer/index.html | 51 +- .../bank_statement_analyzer/jobs.html | 4 + app/modules/bank_statement_analyzer/ui.py | 129 +++-- 6 files changed, 517 insertions(+), 220 deletions(-) create mode 100644 alembic/versions/20260713_bank_statement_analysis_queue.py create mode 100644 app/modules/bank_statement_analyzer/models.py create mode 100644 app/modules/bank_statement_analyzer/templates/bank_statement_analyzer/jobs.html diff --git a/alembic/versions/20260713_bank_statement_analysis_queue.py b/alembic/versions/20260713_bank_statement_analysis_queue.py new file mode 100644 index 0000000..a0e5f94 --- /dev/null +++ b/alembic/versions/20260713_bank_statement_analysis_queue.py @@ -0,0 +1,50 @@ +"""bank statement analysis queue + +Revision ID: 20260713_bank_stmt_queue +Revises: 20260711_reconcile_perm_doc_udin_aqmm +""" +from alembic import op +import sqlalchemy as sa + +revision = "20260713_bank_stmt_queue" +down_revision = "20260711_reconcile_perm_doc_udin_aqmm" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "bank_statement_analysis_jobs", + sa.Column("id", sa.String(length=32), primary_key=True), + sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id", ondelete="CASCADE"), nullable=True), + sa.Column("branch_id", sa.Integer(), sa.ForeignKey("branches.id", ondelete="SET NULL"), nullable=True), + sa.Column("user_id", sa.Integer(), sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False), + sa.Column("role_bucket", sa.String(length=30), nullable=False), + sa.Column("selected_bank", sa.String(length=40), nullable=False, server_default="auto"), + sa.Column("financial_year", sa.String(length=20), nullable=True), + sa.Column("customer_override", sa.String(length=255), nullable=True), + sa.Column("account_override", sa.String(length=100), nullable=True), + sa.Column("classification_enabled", sa.Boolean(), nullable=False, server_default=sa.true()), + sa.Column("status", sa.String(length=20), nullable=False, server_default="queued"), + sa.Column("progress_percent", sa.Integer(), nullable=False, server_default="0"), + sa.Column("file_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("input_files_json", sa.Text(), nullable=False), + sa.Column("job_directory", sa.Text(), nullable=False), + sa.Column("output_file", sa.Text(), nullable=True), + sa.Column("summary_json", sa.Text(), nullable=True), + sa.Column("error_message", sa.Text(), nullable=True), + sa.Column("submitted_at_utc", sa.DateTime(timezone=True), nullable=False), + sa.Column("started_at_utc", sa.DateTime(timezone=True), nullable=True), + sa.Column("completed_at_utc", sa.DateTime(timezone=True), nullable=True), + sa.Column("expires_at_utc", sa.DateTime(timezone=True), nullable=True), + ) + op.create_index("ix_bank_stmt_jobs_tenant_id", "bank_statement_analysis_jobs", ["tenant_id"]) + op.create_index("ix_bank_stmt_jobs_branch_id", "bank_statement_analysis_jobs", ["branch_id"]) + op.create_index("ix_bank_stmt_jobs_user_id", "bank_statement_analysis_jobs", ["user_id"]) + op.create_index("ix_bank_stmt_jobs_status", "bank_statement_analysis_jobs", ["status"]) + op.create_index("ix_bank_stmt_jobs_submitted", "bank_statement_analysis_jobs", ["submitted_at_utc"]) + op.create_index("ix_bank_stmt_jobs_expires", "bank_statement_analysis_jobs", ["expires_at_utc"]) + + +def downgrade(): + op.drop_table("bank_statement_analysis_jobs") diff --git a/app/modules/bank_statement_analyzer/models.py b/app/modules/bank_statement_analyzer/models.py new file mode 100644 index 0000000..93a9917 --- /dev/null +++ b/app/modules/bank_statement_analyzer/models.py @@ -0,0 +1,38 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from sqlalchemy import Boolean, DateTime, ForeignKey, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.db.common import CommonBase + + +class BankStatementAnalysisJob(CommonBase): + __tablename__ = "bank_statement_analysis_jobs" + + id: Mapped[str] = mapped_column(String(32), primary_key=True) + tenant_id: Mapped[int | None] = mapped_column(ForeignKey("tenants.id", ondelete="CASCADE"), nullable=True, index=True) + branch_id: Mapped[int | None] = mapped_column(ForeignKey("branches.id", ondelete="SET NULL"), nullable=True, index=True) + user_id: Mapped[int] = mapped_column(ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True) + role_bucket: Mapped[str] = mapped_column(String(30), nullable=False) + + selected_bank: Mapped[str] = mapped_column(String(40), nullable=False, default="auto") + financial_year: Mapped[str | None] = mapped_column(String(20), nullable=True) + customer_override: Mapped[str | None] = mapped_column(String(255), nullable=True) + account_override: Mapped[str | None] = mapped_column(String(100), nullable=True) + classification_enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + + status: Mapped[str] = mapped_column(String(20), nullable=False, default="queued", index=True) + progress_percent: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + file_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + input_files_json: Mapped[str] = mapped_column(Text, nullable=False) + job_directory: Mapped[str] = mapped_column(Text, nullable=False) + output_file: Mapped[str | None] = mapped_column(Text, nullable=True) + summary_json: Mapped[str | None] = mapped_column(Text, nullable=True) + error_message: Mapped[str | None] = mapped_column(Text, nullable=True) + + submitted_at_utc: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(timezone.utc), index=True) + started_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + completed_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + expires_at_utc: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True, index=True) diff --git a/app/modules/bank_statement_analyzer/service.py b/app/modules/bank_statement_analyzer/service.py index a30a6bd..0b2feef 100644 --- a/app/modules/bank_statement_analyzer/service.py +++ b/app/modules/bank_statement_analyzer/service.py @@ -1,58 +1,63 @@ from __future__ import annotations import json +import math import os import re import shutil +import threading +import time import uuid +from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Iterable from fastapi import UploadFile +from sqlalchemy import func, select, text + +from app.core.db.common import CommonSessionLocal from .analyzer import analyze_files, export_excel +from .models import BankStatementAnalysisJob 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")) +MAX_GLOBAL_PROCESSING = 3 +MAX_PENDING_PER_USER = 3 +COMPLETED_RETENTION_HOURS = 24 +FAILED_RETENTION_HOURS = 24 +POLL_SECONDS = 2 _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() +_QUEUE_LOCK_ID = 820260713 +_executor = ThreadPoolExecutor(max_workers=MAX_GLOBAL_PROCESSING, thread_name_prefix="bank-analyzer") +_worker_lock = threading.Lock() +_worker_started = False def _now() -> datetime: return datetime.now(timezone.utc) +def _resolve_document_storage_root() -> Path: + configured = (os.getenv("DOCUMENT_STORAGE_ROOT") or "").strip() + if configured: + configured_path = Path(configured).expanduser() + return configured_path if configured_path.is_absolute() else (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 _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 + text_value = _SAFE.sub("_", str(value or default).strip()).strip("._-") + return text_value[:80] or default def role_bucket(roles: Iterable[str]) -> str | None: @@ -75,26 +80,11 @@ def _user_root(user, roles: Iterable[str]) -> Path: 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) +def create_job_folder(user, roles: Iterable[str]) -> tuple[str, Path, Path]: job_id = uuid.uuid4().hex - job = _user_root(user, roles) / job_id - input_dir = job / "Input" - output_dir = job / "Output" + job_dir = _user_root(user, roles) / job_id + input_dir = job_dir / "Input" + output_dir = job_dir / "Output" input_dir.mkdir(parents=True, exist_ok=False) output_dir.mkdir(parents=True, exist_ok=False) return job_id, input_dir, output_dir @@ -106,100 +96,321 @@ def _validate_pdf_header(data: bytes) -> None: 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()] + usable = [item for item in files if item and (item.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 + try: + 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.") + target = input_dir / f"{index:02d}_{_segment(Path(name).stem, f'statement_{index}')}.pdf" + 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 + except Exception: + shutil.rmtree(input_dir.parent, ignore_errors=True) + raise -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(), +def pending_count_for_user(user_id: int) -> int: + db = CommonSessionLocal() + try: + return int(db.scalar(select(func.count()).select_from(BankStatementAnalysisJob).where(BankStatementAnalysisJob.user_id == int(user_id), BankStatementAnalysisJob.status.in_(["queued", "processing"]))) or 0) + finally: + db.close() + + +def enqueue_job(*, user, roles: Iterable[str], job_id: str, paths: list[Path], job_dir: Path, bank_selection: str, financial_year: str, customer_override: str, account_override: str, classification_enabled: bool) -> BankStatementAnalysisJob: + if pending_count_for_user(int(user.id)) >= MAX_PENDING_PER_USER: + shutil.rmtree(job_dir, ignore_errors=True) + raise ValueError("You already have three queued or processing analyses. Please wait for one to complete before submitting another.") + bucket = role_bucket(roles) + if not bucket: + shutil.rmtree(job_dir, ignore_errors=True) + raise PermissionError("You do not have access to the Bank Statement Analyzer.") + job = BankStatementAnalysisJob( + id=job_id, + tenant_id=getattr(user, "tenant_id", None), + branch_id=getattr(user, "branch_id", None), + user_id=int(user.id), + role_bucket=bucket, selected_bank=bank_selection, - classification_enabled=classification_enabled, + financial_year=(financial_year or "").strip() or None, + customer_override=(customer_override or "").strip() or None, + account_override=(account_override or "").strip() or None, + classification_enabled=bool(classification_enabled), + status="queued", + progress_percent=0, + file_count=len(paths), + input_files_json=json.dumps([str(path) for path in paths]), + job_directory=str(job_dir), + submitted_at_utc=_now(), ) - 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 + db = CommonSessionLocal() + try: + db.add(job) + db.commit() + db.refresh(job) + except Exception: + db.rollback() + shutil.rmtree(job_dir, ignore_errors=True) + raise + finally: + db.close() + ensure_worker_started() + return job -def resolve_owned_job(user, roles: Iterable[str], job_id: str) -> tuple[Path, dict]: +def _claim_jobs() -> list[str]: + db = CommonSessionLocal() + try: + try: + db.execute(text("SELECT pg_advisory_xact_lock(:lock_id)"), {"lock_id": _QUEUE_LOCK_ID}) + except Exception: + db.rollback() + processing = int(db.scalar(select(func.count()).select_from(BankStatementAnalysisJob).where(BankStatementAnalysisJob.status == "processing")) or 0) + capacity = max(0, MAX_GLOBAL_PROCESSING - processing) + if capacity <= 0: + return [] + stmt = select(BankStatementAnalysisJob).where(BankStatementAnalysisJob.status == "queued").order_by(BankStatementAnalysisJob.submitted_at_utc.asc()).limit(capacity) + try: + stmt = stmt.with_for_update(skip_locked=True) + except Exception: + pass + jobs = list(db.scalars(stmt).all()) + now = _now() + for job in jobs: + job.status = "processing" + job.progress_percent = 5 + job.started_at_utc = now + job.error_message = None + db.commit() + return [job.id for job in jobs] + except Exception: + db.rollback() + return [] + finally: + db.close() + + +def _process_job(job_id: str) -> None: + db = CommonSessionLocal() + try: + job = db.get(BankStatementAnalysisJob, job_id) + if not job or job.status != "processing": + return + paths = [Path(value) for value in json.loads(job.input_files_json)] + output_dir = Path(job.job_directory) / "Output" + job.progress_percent = 15 + db.commit() + metas, all_df, unique_df = analyze_files( + paths, + job.customer_override or "", + job.account_override or "", + bank_hint=job.selected_bank, + classification_enabled=job.classification_enabled, + ) + job.progress_percent = 75 + db.commit() + output = output_dir / "Bank_Statement_Analysis.xlsx" + export_excel( + output, + metas, + all_df, + unique_df, + financial_year=job.financial_year or "", + selected_bank=job.selected_bank, + classification_enabled=job.classification_enabled, + ) + summary = { + "job_id": job.id, + "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 and "review_note" in unique_df.columns else 0, + "categories": int(unique_df.category.nunique()) if not unique_df.empty and "category" in unique_df.columns 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": job.financial_year or "", + "classification_enabled": job.classification_enabled, + } + for path in paths: + try: + path.unlink(missing_ok=True) + except OSError: + pass + input_dir = Path(job.job_directory) / "Input" + try: + input_dir.rmdir() + except OSError: + pass + job.output_file = str(output) + job.summary_json = json.dumps(summary) + job.status = "completed" + job.progress_percent = 100 + job.completed_at_utc = _now() + job.expires_at_utc = _now() + timedelta(hours=COMPLETED_RETENTION_HOURS) + db.commit() + except Exception as exc: + db.rollback() + job = db.get(BankStatementAnalysisJob, job_id) + if job: + job.status = "failed" + job.progress_percent = 100 + job.error_message = str(exc)[:4000] + job.completed_at_utc = _now() + job.expires_at_utc = _now() + timedelta(hours=FAILED_RETENTION_HOURS) + db.commit() + finally: + db.close() + + +def _cleanup_expired() -> None: + db = CommonSessionLocal() + try: + jobs = list(db.scalars(select(BankStatementAnalysisJob).where(BankStatementAnalysisJob.expires_at_utc.is_not(None), BankStatementAnalysisJob.expires_at_utc < _now())).all()) + for job in jobs: + shutil.rmtree(job.job_directory, ignore_errors=True) + db.delete(job) + db.commit() + except Exception: + db.rollback() + finally: + db.close() + + +def _worker_loop() -> None: + while True: + try: + _cleanup_expired() + for job_id in _claim_jobs(): + _executor.submit(_process_job, job_id) + except Exception: + pass + time.sleep(POLL_SECONDS) + + +def ensure_worker_started() -> None: + global _worker_started + with _worker_lock: + if _worker_started: + return + thread = threading.Thread(target=_worker_loop, name="bank-analyzer-queue", daemon=True) + thread.start() + _worker_started = True + + +def _summary(job: BankStatementAnalysisJob) -> dict: + return json.loads(job.summary_json) if job.summary_json else {} + + +def average_duration_minutes() -> float: + db = CommonSessionLocal() + try: + jobs = list(db.scalars(select(BankStatementAnalysisJob).where(BankStatementAnalysisJob.status == "completed", BankStatementAnalysisJob.started_at_utc.is_not(None), BankStatementAnalysisJob.completed_at_utc.is_not(None)).order_by(BankStatementAnalysisJob.completed_at_utc.desc()).limit(20)).all()) + durations = [(job.completed_at_utc - job.started_at_utc).total_seconds() / 60 for job in jobs if job.completed_at_utc and job.started_at_utc] + return max(1.0, sum(durations) / len(durations)) if durations else 4.0 + finally: + db.close() + + +def queue_position(job: BankStatementAnalysisJob) -> int | None: + if job.status != "queued": + return None + db = CommonSessionLocal() + try: + return int(db.scalar(select(func.count()).select_from(BankStatementAnalysisJob).where(BankStatementAnalysisJob.status == "queued", BankStatementAnalysisJob.submitted_at_utc <= job.submitted_at_utc)) or 1) + finally: + db.close() + + +def estimated_wait(job: BankStatementAnalysisJob) -> str: + if job.status == "processing": + return "Processing now" + if job.status != "queued": + return "" + position = queue_position(job) or 1 + waves = max(1, math.ceil(position / MAX_GLOBAL_PROCESSING)) + average = average_duration_minutes() + low = max(1, math.ceil((waves - 1) * average)) + high = max(5, math.ceil(waves * average + 2)) + return f"Approximately {low}-{high} minutes" + + +def get_owned_job(user_id: int, job_id: str) -> BankStatementAnalysisJob | None: 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 + return None + db = CommonSessionLocal() + try: + job = db.scalar(select(BankStatementAnalysisJob).where(BankStatementAnalysisJob.id == job_id, BankStatementAnalysisJob.user_id == int(user_id))) + if job: + db.expunge(job) + return job + finally: + db.close() -def delete_job(job: Path) -> None: - shutil.rmtree(job, ignore_errors=True) +def list_user_jobs(user_id: int, limit: int = 25) -> list[BankStatementAnalysisJob]: + db = CommonSessionLocal() + try: + jobs = list(db.scalars(select(BankStatementAnalysisJob).where(BankStatementAnalysisJob.user_id == int(user_id)).order_by(BankStatementAnalysisJob.submitted_at_utc.desc()).limit(limit)).all()) + for job in jobs: + db.expunge(job) + return jobs + finally: + db.close() + + +def job_view(job: BankStatementAnalysisJob) -> dict: + return { + "id": job.id, + "status": job.status, + "progress_percent": job.progress_percent, + "file_count": job.file_count, + "selected_bank": job.selected_bank, + "financial_year": job.financial_year or "", + "submitted_at": job.submitted_at_utc, + "completed_at": job.completed_at_utc, + "expires_at": job.expires_at_utc, + "error_message": job.error_message or "", + "queue_position": queue_position(job), + "estimated_wait": estimated_wait(job), + "summary": _summary(job), + "download_ready": job.status == "completed" and bool(job.output_file) and Path(job.output_file).is_file(), + } + + +def delete_owned_job(user_id: int, job_id: str) -> bool: + db = CommonSessionLocal() + try: + job = db.scalar(select(BankStatementAnalysisJob).where(BankStatementAnalysisJob.id == job_id, BankStatementAnalysisJob.user_id == int(user_id))) + if not job: + return False + if job.status == "processing": + raise ValueError("A processing analysis cannot be deleted. Please wait for it to finish.") + shutil.rmtree(job.job_directory, ignore_errors=True) + db.delete(job) + db.commit() + return True + finally: + db.close() diff --git a/app/modules/bank_statement_analyzer/templates/bank_statement_analyzer/index.html b/app/modules/bank_statement_analyzer/templates/bank_statement_analyzer/index.html index 44badba..0d4a790 100644 --- a/app/modules/bank_statement_analyzer/templates/bank_statement_analyzer/index.html +++ b/app/modules/bank_statement_analyzer/templates/bank_statement_analyzer/index.html @@ -1,37 +1,42 @@ {% extends "ui/templates/base/layout.html" %} {% block content %} -
+
-

Bank Statement Analyzer

-

Upload one or more PDF bank statements. Select a bank for direct parser validation or keep Auto Detect. The workbook includes reconciliation, duplicate checks, transaction classifications, party summaries, review items and draft bank-basis financial helpers.

+
+

Bank Statement Analyzer

Upload supported PDF statements. Up to three analyses run globally at one time; additional jobs are queued safely.

+ My Analysis Jobs +
- {% if error %}
Analysis could not be completed.
{{ error }}
{% endif %} + + {% if error %}
Analysis could not be submitted.
{{ error }}
{% endif %} +
-
- - -

Manual selection validates the PDF against that bank. Auto Detect preserves the existing behavior.

-
-
- - -
+

Keep Auto Detect or select a bank for direct parser validation.

+
-
- -
+
-
Available banks
Axis Bank, HDFC Bank, IDFC FIRST Bank, Indian Bank, IndusInd Bank, Kotak Mahindra Bank and State Bank of India.
Successful jobs are deleted automatically after the Excel response is sent. Failed or abandoned jobs are cleaned after the configured retention period.
-
+
Queue limits
Maximum three processing jobs across all users. Each user may have up to three queued or processing jobs. Completed workbooks remain available for 24 hours.
+
+ + {% if active_job %} +
+

Current Analysis

You may leave this page. The job continues in the background.

{{ active_job.status|title }}
+
Files
{{ active_job.file_count }}
Queue position
{{ active_job.queue_position or '—' }}
Estimated wait
{{ active_job.estimated_wait or '—' }}
Progress
{{ active_job.progress_percent }}%
+
+ {% if active_job.status == 'completed' %} +

Analysis completed

Statements
{{ active_job.summary.statement_count or 0 }}
Transactions extracted
{{ active_job.summary.rows_extracted or 0 }}
Exact duplicates
{{ active_job.summary.exact_duplicate_rows or 0 }}
Review items
{{ active_job.summary.review_items or 0 }}
Download ExcelAnalyze Another Bank

The workbook remains available until {{ active_job.expires_at or '24 hours after completion' }}.

+ {% elif active_job.status == 'failed' %}
Analysis failed.
{{ active_job.error_message }}
Analyze another statement
+ {% else %}
{% if active_job.status == 'queued' %}Your job is queued. You may safely leave this page and return through My Analysis Jobs.{% else %}Your statements are being processed.{% endif %}
{% endif %} +
+ {% endif %} + + {% if recent_jobs %}

Recent Analyses

View all
{% for item in recent_jobs %}{% endfor %}
SubmittedBankFilesStatusAction
{{ item.submitted_at }}{{ item.selected_bank|replace('_',' ')|title }}{{ item.file_count }}{{ item.status|title }}{% if item.queue_position %} · Position {{ item.queue_position }}{% endif %}View
{% endif %}
+{% if active_job and active_job.status in ['queued','processing'] %}{% endif %} {% endblock %} diff --git a/app/modules/bank_statement_analyzer/templates/bank_statement_analyzer/jobs.html b/app/modules/bank_statement_analyzer/templates/bank_statement_analyzer/jobs.html new file mode 100644 index 0000000..c2650ce --- /dev/null +++ b/app/modules/bank_statement_analyzer/templates/bank_statement_analyzer/jobs.html @@ -0,0 +1,4 @@ +{% extends "ui/templates/base/layout.html" %} +{% block content %} +

My Analysis Jobs

Queued, processing, completed and failed bank-statement analyses.

New Analysis
{% for item in jobs %}{% else %}{% endfor %}
SubmittedBankFilesStatusEstimate / ExpiryAction
{{ item.submitted_at }}{{ item.selected_bank|replace('_',' ')|title }}{{ item.file_count }}{{ item.status|title }}{% if item.queue_position %}
Queue position {{ item.queue_position }}
{% endif %}		{% if item.status in ['queued','processing'] %}{{ item.estimated_wait }}{% elif item.status == 'completed' %}Available for 24 hours{% else %}{{ item.error_message[:80] }}{% endif %}
View{% if item.download_ready %}Download{% endif %}{% if item.status != 'processing' %}
{% endif %}
No analysis jobs yet.
+{% endblock %} diff --git a/app/modules/bank_statement_analyzer/ui.py b/app/modules/bank_statement_analyzer/ui.py index 25a669e..bbff089 100644 --- a/app/modules/bank_statement_analyzer/ui.py +++ b/app/modules/bank_statement_analyzer/ui.py @@ -3,8 +3,7 @@ from __future__ import annotations from pathlib import Path from fastapi import APIRouter, File, Form, Request, UploadFile -from fastapi.responses import FileResponse, RedirectResponse -from starlette.background import BackgroundTask +from fastapi.responses import FileResponse, JSONResponse, RedirectResponse from app.core.db.common import CommonSessionLocal from app.core.http_responses import ui_access_denied, not_found_response @@ -14,7 +13,7 @@ from app.core.templating import templates from app.modules.core.rbac.deps import get_user_permissions, get_user_roles from .parsers.registry import BANK_OPTIONS -from .service import can_use, create_job, save_uploads, analyze_job, resolve_owned_job, delete_job +from .service import can_use, create_job_folder, delete_owned_job, enqueue_job, ensure_worker_started, get_owned_job, job_view, list_user_jobs, save_uploads router = APIRouter(prefix="/tools/bank-statement-analyzer", tags=["bank-statement-analyzer-ui"]) @@ -47,105 +46,99 @@ def _auth(request, db): @router.get("") -def index(request: Request): +def index(request: Request, job: str | None = None): + ensure_worker_started() db = CommonSessionLocal() try: user, roles, denied = _auth(request, db) if denied: return denied - return templates.TemplateResponse( - "modules/bank_statement_analyzer/templates/bank_statement_analyzer/index.html", - _ctx(request, db, user, error=""), - ) + selected_job = get_owned_job(user.id, job) if job else None + recent = [job_view(item) for item in list_user_jobs(user.id, limit=8)] + return templates.TemplateResponse("modules/bank_statement_analyzer/templates/bank_statement_analyzer/index.html", _ctx(request, db, user, error="", active_job=job_view(selected_job) if selected_job else None, recent_jobs=recent)) finally: db.close() @router.post("/analyze") -async def analyze( - request: Request, - csrf_token: str = Form(...), - bank_selection: str = Form("auto"), - financial_year: str = Form(""), - customer_name: str = Form(""), - account_number: str = Form(""), - enable_classification: str | None = Form(None), - statements: list[UploadFile] = File(...), -): +async def analyze(request: Request, csrf_token: str = Form(...), bank_selection: str = Form("auto"), financial_year: str = Form(""), customer_name: str = Form(""), account_number: str = Form(""), enable_classification: str | None = Form(None), statements: list[UploadFile] = File(...)): db = CommonSessionLocal() - job_dir: Path | None = None selected_bank = bank_selection if bank_selection in dict(BANK_OPTIONS) else "auto" classification_enabled = enable_classification == "1" + job_dir: Path | None = None try: user, roles, denied = _auth(request, db) if denied: return denied validate_csrf(request, csrf_token) - job_id, input_dir, output_dir = create_job(user, roles) + job_id, input_dir, _output_dir = create_job_folder(user, roles) job_dir = input_dir.parent paths = await save_uploads(statements, input_dir) - summary = analyze_job( - user=user, - roles=roles, - job_id=job_id, - paths=paths, - output_dir=output_dir, - customer_override=customer_name, - account_override=account_number, - bank_selection=selected_bank, - financial_year=financial_year, - classification_enabled=classification_enabled, - ) - return templates.TemplateResponse( - "modules/bank_statement_analyzer/templates/bank_statement_analyzer/result.html", - _ctx(request, db, user, summary=summary), - ) + enqueue_job(user=user, roles=roles, job_id=job_id, paths=paths, job_dir=job_dir, bank_selection=selected_bank, financial_year=financial_year, customer_override=customer_name, account_override=account_number, classification_enabled=classification_enabled) + return RedirectResponse(f"/tools/bank-statement-analyzer?job={job_id}#analysis-status", status_code=303) except Exception as exc: + if job_dir: + import shutil + shutil.rmtree(job_dir, ignore_errors=True) user = get_current_user(request, db=db) if not user: return RedirectResponse("/login", status_code=303) - return templates.TemplateResponse( - "modules/bank_statement_analyzer/templates/bank_statement_analyzer/index.html", - _ctx( - request, - db, - user, - error=str(exc), - selected_bank=selected_bank, - financial_year=financial_year, - classification_enabled=classification_enabled, - ), - status_code=400, - ) + recent = [job_view(item) for item in list_user_jobs(user.id, limit=8)] + return templates.TemplateResponse("modules/bank_statement_analyzer/templates/bank_statement_analyzer/index.html", _ctx(request, db, user, error=str(exc), active_job=None, recent_jobs=recent, selected_bank=selected_bank, financial_year=financial_year, classification_enabled=classification_enabled), status_code=400) finally: db.close() -@router.get("/{job_id}/download") +@router.get("/jobs") +def jobs(request: Request): + ensure_worker_started() + db = CommonSessionLocal() + try: + user, roles, denied = _auth(request, db) + if denied: + return denied + items = [job_view(item) for item in list_user_jobs(user.id, limit=100)] + return templates.TemplateResponse("modules/bank_statement_analyzer/templates/bank_statement_analyzer/jobs.html", _ctx(request, db, user, jobs=items)) + finally: + db.close() + + +@router.get("/jobs/{job_id}/status") +def status(job_id: str, request: Request): + ensure_worker_started() + db = CommonSessionLocal() + try: + user, roles, denied = _auth(request, db) + if denied: + return JSONResponse({"detail": "Access denied"}, status_code=403) + job = get_owned_job(user.id, job_id) + if not job: + return JSONResponse({"detail": "Analysis job not found"}, status_code=404) + view = job_view(job) + return JSONResponse({"id": view["id"], "status": view["status"], "progress_percent": view["progress_percent"], "queue_position": view["queue_position"], "estimated_wait": view["estimated_wait"], "download_ready": view["download_ready"], "error_message": view["error_message"]}) + finally: + db.close() + + +@router.get("/jobs/{job_id}/download") def download(job_id: str, request: Request): db = CommonSessionLocal() try: user, roles, denied = _auth(request, db) if denied: return denied - try: - job, meta = resolve_owned_job(user, roles, job_id) - except FileNotFoundError: - return not_found_response(request, "Analysis job not found or already cleaned up.") - output = job / "Output" / meta["output_file"] - if not output.is_file(): - return not_found_response(request, "Analysis workbook not found.") - return FileResponse( - path=output, - filename="Bank_Statement_Analysis.xlsx", - media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", - background=BackgroundTask(delete_job, job), - ) + job = get_owned_job(user.id, job_id) + if not job: + return not_found_response(request, "Analysis job not found or expired.") + view = job_view(job) + if not view["download_ready"]: + return not_found_response(request, "Analysis workbook is not ready or has expired.") + return FileResponse(path=job.output_file, filename="Bank_Statement_Analysis.xlsx", media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet") finally: db.close() -@router.post("/{job_id}/delete") +@router.post("/jobs/{job_id}/delete") def delete(job_id: str, request: Request, csrf_token: str = Form(...)): db = CommonSessionLocal() try: @@ -153,11 +146,7 @@ def delete(job_id: str, request: Request, csrf_token: str = Form(...)): if denied: return denied validate_csrf(request, csrf_token) - try: - job, _ = resolve_owned_job(user, roles, job_id) - delete_job(job) - except FileNotFoundError: - pass - return RedirectResponse("/tools/bank-statement-analyzer", status_code=303) + delete_owned_job(user.id, job_id) + return RedirectResponse("/tools/bank-statement-analyzer/jobs", status_code=303) finally: db.close()