Accounting · Bank Reconciliation
- Compare the normalized transactions from a completed client-bound Bank Analyzer job with one selected Tally Bank ledger. This screen is read-only against Tally; unmatched bank rows can later be handled through the existing Accounting Bank Queue. + Compare normalized Bank Analyzer transactions with Tally Bank ledgers. Phase 20 remembers each client bank-account ↔ Tally-ledger mapping, can reconcile every mapped account in one action, supports wider cheque/clearing timing windows, and keeps reviewer resolutions. Tally access remains read-only.
The Local Agent only reads vouchers from the selected Tally company/date range and filters them to the selected Bank ledger. No accounting entry is posted by this action.
+Choose the completed client-bound analysis whose imported bank transactions you want to reconcile. The job can contain any number of bank accounts.
+Mappings are remembered client-wise and company-wise. Once all accounts are mapped, future reconciliation runs do not require selecting the bank ledger again.
Use a wider timing tolerance for cheque issue/presentation or deposit/clearing differences. Amount and bank direction must still match exactly before the engine considers a Tally voucher as a candidate.
+| Run | Bank Ledger | Period | Status | Result | ||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| Run | Bank / Account | Period | Status | Result | ||||||||||
| #{{ run.id }} | -{{ run.bank_ledger_name }} | +{{ run.bank_ledger_name }} {{ run.account_number or "All imported accounts" }} |
{{ run.date_from }} → {{ run.date_to }} | {{ run.status|replace('_',' ')|title }}{% if run.last_error %} {{ run.last_error }} {% endif %} |
+ {% set s = run_summaries.get(run.id, {}) %}
{% if run.status=='completed' %}
- {% set s = run.summary_json|from_json if false else none %}
- Open run to view matched / probable / bank-only / books-only detail.
+ Matched {{ s.get('matched',0) }} · Probable {{ s.get('probable_match',0) }} · Timing {{ s.get('timing_difference',0) }}
+ Bank only {{ s.get('bank_only',0) }} · Books only {{ s.get('books_only',0) }} · Duplicate {{ s.get('duplicate_candidate',0) }}
+ Unreconciled net difference ₹{{ '%.2f'|format(s.get('unreconciled_net_difference',0)) }}
{% else %}—{% endif %}
|
Open | @@ -144,7 +232,7 @@ @@ -158,6 +246,15 @@ {% elif selected_run.status=='failed' %}
| Result | Bank Statement | Tally | Why | Action | {{ item.match_reason }} | -+ |
+ {% if item.resolution_status and item.resolution_status != 'unresolved' %}
+ Reviewed: {{ item.resolution_status|replace('_',' ')|title }}{% if item.resolution_note %}
+ {% endif %}
+
{% if item.match_status=='bank_only' and item.bank_transaction_id %}
- Review in Accounting Queue
- {% elif item.match_status=='books_only' %}
- Investigate timing / statement coverage.
- {% else %}
- Review if needed.
+ Open Accounting Queue
{% endif %}
{{ item.resolution_note }} {% endif %} |
diff --git a/app/modules/bank_statement_analyzer/client_context.py b/app/modules/bank_statement_analyzer/client_context.py
index 32cc872..f21a50c 100644
--- a/app/modules/bank_statement_analyzer/client_context.py
+++ b/app/modules/bank_statement_analyzer/client_context.py
@@ -365,12 +365,37 @@ def archive_analysis_to_engagement(
return {"status": "not_requested", "documents": []}
archived = []
+
+ reused = []
+ try:
+ reused = json.loads(getattr(job, "stored_source_versions_json", None) or "[]")
+ except Exception:
+ reused = []
+ reused_by_copied_name = {
+ _text(row.get("copied_filename")): row
+ for row in reused
+ if isinstance(row, dict) and _text(row.get("copied_filename"))
+ }
+
meta_by_source = {
Path(_text(getattr(meta, "source_file", ""))).name: meta
for meta in metas
}
for source in source_paths:
+ reused_row = reused_by_copied_name.get(source.name)
+ if reused_row:
+ archived.append(
+ {
+ "document_id": int(reused_row.get("document_id") or 0),
+ "version_id": int(reused_row.get("version_id") or 0),
+ "filename": reused_row.get("original_filename") or source.name,
+ "type": "BANK_STATEMENT",
+ "reused_existing_document": True,
+ }
+ )
+ continue
+
upload = _upload_proxy(source, "application/pdf")
try:
meta = meta_by_source.get(source.name)
diff --git a/app/modules/bank_statement_analyzer/models.py b/app/modules/bank_statement_analyzer/models.py
index d1be60f..ef99058 100644
--- a/app/modules/bank_statement_analyzer/models.py
+++ b/app/modules/bank_statement_analyzer/models.py
@@ -39,6 +39,11 @@ class BankStatementAnalysisJob(CommonBase):
accounting_import_status: Mapped[str] = mapped_column(String(30), nullable=False, default="not_requested", index=True)
accounting_import_json: Mapped[str | None] = mapped_column(Text, nullable=True)
+ # Phase 20: stored-statement reuse. These fields only record provenance;
+ # the physical files remain in the existing Engagement Documents storage.
+ stored_source_versions_json: Mapped[str | None] = mapped_column(Text, nullable=True)
+ source_hashes_json: Mapped[str | None] = mapped_column(Text, nullable=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)
diff --git a/app/modules/bank_statement_analyzer/service.py b/app/modules/bank_statement_analyzer/service.py
index 23cc46b..4929078 100644
--- a/app/modules/bank_statement_analyzer/service.py
+++ b/app/modules/bank_statement_analyzer/service.py
@@ -68,6 +68,80 @@ def _segment(value: object, default: str = "NA") -> str:
+def _statement_period(value):
+ text_value = str(value or "").strip()
+ if not text_value:
+ return None
+ for fmt in (
+ "%Y-%m-%d",
+ "%d/%m/%Y",
+ "%d-%m-%Y",
+ "%d.%m.%Y",
+ "%d-%b-%Y",
+ "%d %b %Y",
+ "%d %B %Y",
+ "%Y/%m/%d",
+ ):
+ try:
+ return datetime.strptime(text_value[:10], fmt).date()
+ except Exception:
+ pass
+ try:
+ return datetime.fromisoformat(text_value.replace("Z", "+00:00")).date()
+ except Exception:
+ return None
+
+
+def _statement_overlap_warnings(metas: list) -> list[dict]:
+ warnings = []
+ groups = {}
+ for meta in metas:
+ account = str(getattr(meta, "account_number", "") or "").strip()
+ bank = str(getattr(meta, "bank_name", "") or "").strip()
+ start = _statement_period(getattr(meta, "period_from", ""))
+ end = _statement_period(getattr(meta, "period_to", ""))
+ if not account or not start or not end:
+ continue
+ if end < start:
+ start, end = end, start
+ groups.setdefault((bank.casefold(), account), []).append(
+ {
+ "bank_name": bank,
+ "account_number": account,
+ "period_from": start,
+ "period_to": end,
+ "source_file": Path(str(getattr(meta, "source_file", "") or "")).name,
+ }
+ )
+
+ for (_bank_key, _account), rows in groups.items():
+ rows.sort(key=lambda row: (row["period_from"], row["period_to"]))
+ for idx, left in enumerate(rows):
+ for right in rows[idx + 1:]:
+ if right["period_from"] > left["period_to"]:
+ break
+ overlap_from = max(left["period_from"], right["period_from"])
+ overlap_to = min(left["period_to"], right["period_to"])
+ if overlap_from <= overlap_to:
+ warnings.append(
+ {
+ "bank_name": left["bank_name"],
+ "account_number": left["account_number"],
+ "first_file": left["source_file"],
+ "second_file": right["source_file"],
+ "overlap_from": overlap_from.isoformat(),
+ "overlap_to": overlap_to.isoformat(),
+ "message": (
+ f"Overlapping statement periods for account "
+ f"{left['account_number']}: {left['source_file']} and "
+ f"{right['source_file']} overlap from "
+ f"{overlap_from.isoformat()} to {overlap_to.isoformat()}."
+ ),
+ }
+ )
+ return warnings
+
+
def _workbook_filename(metas: list) -> str:
"""Build a safe BankName_ClientName.xlsx filename for every bank."""
bank_names = [str(getattr(meta, "bank_name", "") or "").strip() for meta in metas]
@@ -162,7 +236,7 @@ def pending_count_for_user(user_id: int) -> int:
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, client_id: int | None = None, engagement_id: int | None = None, ownership_confirmation: bool = False, purpose: str = "analyze_only") -> BankStatementAnalysisJob:
+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, client_id: int | None = None, engagement_id: int | None = None, ownership_confirmation: bool = False, purpose: str = "analyze_only", stored_source_versions: list[dict] | None = None, source_hashes: list[dict] | None = None) -> 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.")
@@ -192,6 +266,16 @@ def enqueue_job(*, user, roles: Iterable[str], job_id: str, paths: list[Path], j
if purpose in {"accounting_entries", "bank_reconciliation"}
else "not_requested"
),
+ stored_source_versions_json=(
+ json.dumps(stored_source_versions, ensure_ascii=False)
+ if stored_source_versions
+ else None
+ ),
+ source_hashes_json=(
+ json.dumps(source_hashes, ensure_ascii=False)
+ if source_hashes
+ else None
+ ),
status="queued",
progress_percent=0,
file_count=len(paths),
@@ -328,6 +412,13 @@ def _process_job(job_id: str) -> None:
"engagement_id": job.engagement_id,
"ownership_status": job.ownership_status,
"engagement_archive_status": job.engagement_archive_status,
+ "stored_statement_reuse_count": len(
+ json.loads(job.stored_source_versions_json or "[]")
+ ) if getattr(job, "stored_source_versions_json", None) else 0,
+ "source_hash_count": len(
+ json.loads(job.source_hashes_json or "[]")
+ ) if getattr(job, "source_hashes_json", None) else 0,
+ "statement_overlap_warnings": _statement_overlap_warnings(metas),
}
# Keep the original uploaded statements until the job expiry time.
# This applies equally to completed and failed jobs and allows the
@@ -530,6 +621,8 @@ def job_view(job: BankStatementAnalysisJob) -> dict:
"purpose": getattr(job, "purpose", "analyze_only"),
"accounting_import_status": getattr(job, "accounting_import_status", "not_requested"),
"accounting_import": json.loads(job.accounting_import_json) if getattr(job, "accounting_import_json", None) else {},
+ "stored_source_versions": json.loads(job.stored_source_versions_json) if getattr(job, "stored_source_versions_json", None) else [],
+ "source_hashes": json.loads(job.source_hashes_json) if getattr(job, "source_hashes_json", None) else [],
"submitted_at": job.submitted_at_utc,
"completed_at": job.completed_at_utc,
"expires_at": job.expires_at_utc,
diff --git a/app/modules/bank_statement_analyzer/stored_statement_service.py b/app/modules/bank_statement_analyzer/stored_statement_service.py
new file mode 100644
index 0000000..2a81e21
--- /dev/null
+++ b/app/modules/bank_statement_analyzer/stored_statement_service.py
@@ -0,0 +1,254 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import shutil
+from pathlib import Path
+
+from sqlalchemy import select
+from sqlalchemy.orm import selectinload
+
+from app.modules.documents.models import (
+ DocumentDownloadRequest,
+ EngagementDocument,
+ EngagementDocumentVersion,
+)
+from app.modules.documents.services import (
+ create_download_request_for_version,
+ download_request_cache_path,
+ version_absolute_path,
+)
+
+from .client_context import analyzer_client_context
+
+
+def _sha256(path: Path) -> str:
+ hasher = hashlib.sha256()
+ with path.open("rb") as handle:
+ while True:
+ chunk = handle.read(1024 * 1024)
+ if not chunk:
+ break
+ hasher.update(chunk)
+ return hasher.hexdigest()
+
+
+def _allowed_engagement_ids(db, *, request, user, roles, client_id: int) -> set[int]:
+ context = analyzer_client_context(db, request=request, user=user, roles=roles)
+ visible_clients = {int(row.id) for row in context["clients"]}
+ if int(client_id) not in visible_clients:
+ raise PermissionError("Selected client is not visible for Bank Statement Analyzer.")
+ return {
+ int(row.id)
+ for row in context["engagements"]
+ if int(row.client_id) == int(client_id)
+ }
+
+
+def list_stored_bank_statements(
+ db,
+ *,
+ request,
+ user,
+ roles,
+ client_id: int,
+ limit: int = 200,
+):
+ allowed_engagements = _allowed_engagement_ids(
+ db,
+ request=request,
+ user=user,
+ roles=roles,
+ client_id=client_id,
+ )
+ if not allowed_engagements:
+ return []
+
+ docs = list(
+ db.execute(
+ select(EngagementDocument)
+ .options(selectinload(EngagementDocument.versions))
+ .where(
+ EngagementDocument.client_id == int(client_id),
+ EngagementDocument.engagement_id.in_(allowed_engagements),
+ EngagementDocument.document_type == "BANK_STATEMENT",
+ EngagementDocument.is_deleted.is_(False),
+ )
+ .order_by(EngagementDocument.updated_at_utc.desc(), EngagementDocument.id.desc())
+ .limit(max(1, min(500, int(limit))))
+ ).scalars().all()
+ )
+
+ result = []
+ for doc in docs:
+ version = doc.versions[0] if doc.versions else None
+ if not version:
+ continue
+ result.append(
+ {
+ "document_id": int(doc.id),
+ "version_id": int(version.id),
+ "engagement_id": int(doc.engagement_id),
+ "title": doc.title,
+ "filename": version.original_filename,
+ "size_bytes": int(version.file_size_bytes or 0),
+ "sha256": version.file_hash_sha256,
+ "uploaded_at": version.uploaded_at_utc.isoformat() if version.uploaded_at_utc else "",
+ "storage_status": version.storage_status,
+ }
+ )
+ return result
+
+
+def _ready_cached_path(db, *, version_id: int, user_id: int):
+ req = db.execute(
+ select(DocumentDownloadRequest)
+ .where(
+ DocumentDownloadRequest.version_id == int(version_id),
+ DocumentDownloadRequest.requested_by_user_id == int(user_id),
+ DocumentDownloadRequest.request_status == "ready",
+ )
+ .order_by(DocumentDownloadRequest.fulfilled_at_utc.desc(), DocumentDownloadRequest.id.desc())
+ .limit(1)
+ ).scalar_one_or_none()
+ if not req:
+ return None
+ path = download_request_cache_path(req)
+ return path if path and path.exists() else None
+
+
+def prepare_stored_bank_statements(
+ db,
+ *,
+ request,
+ user,
+ roles,
+ client_id: int,
+ version_ids: list[int],
+ input_dir: Path,
+):
+ selected_ids = list(dict.fromkeys(int(value) for value in version_ids if int(value) > 0))
+ if not selected_ids:
+ return [], [], []
+
+ allowed_engagements = _allowed_engagement_ids(
+ db,
+ request=request,
+ user=user,
+ roles=roles,
+ client_id=client_id,
+ )
+
+ versions = list(
+ db.execute(
+ select(EngagementDocumentVersion)
+ .join(
+ EngagementDocument,
+ EngagementDocument.id == EngagementDocumentVersion.document_id,
+ )
+ .where(
+ EngagementDocumentVersion.id.in_(selected_ids),
+ EngagementDocumentVersion.client_id == int(client_id),
+ EngagementDocumentVersion.engagement_id.in_(allowed_engagements),
+ EngagementDocument.document_type == "BANK_STATEMENT",
+ EngagementDocument.is_deleted.is_(False),
+ )
+ ).scalars().all()
+ )
+ by_id = {int(row.id): row for row in versions}
+
+ if set(selected_ids) != set(by_id):
+ raise PermissionError("One or more stored bank statements are not available to this user/client.")
+
+ prepared = []
+ provenance = []
+ pending = []
+
+ for order, version_id in enumerate(selected_ids, start=1):
+ version = by_id[version_id]
+ source = version_absolute_path(version)
+ if not source.exists():
+ source = _ready_cached_path(
+ db,
+ version_id=version.id,
+ user_id=user.id,
+ )
+
+ if source is None or not source.exists():
+ req = create_download_request_for_version(
+ db,
+ version=version,
+ user=user,
+ request=request,
+ )
+ if req is not None and req.id is None:
+ db.flush()
+ if req is not None and req.request_status == "ready":
+ cached = download_request_cache_path(req)
+ if cached is None or not cached.exists():
+ # A stale ready request must be made retrievable again.
+ req.request_status = "retry"
+ req.cached_relative_path = None
+ req.cached_hash_sha256 = None
+ req.last_error = "Cached retrieval copy was missing; queued again for branch Local Agent."
+ db.add(req)
+ db.flush()
+ if not req:
+ raise FileNotFoundError(
+ f"{version.original_filename}: the stored file is not currently available "
+ "on ERP or a registered branch storage node."
+ )
+ pending.append(
+ {
+ "version_id": int(version.id),
+ "filename": version.original_filename,
+ "request_id": int(req.id) if req.id else None,
+ "status": req.request_status,
+ }
+ )
+ continue
+
+ actual_hash = _sha256(source)
+ expected = (version.file_hash_sha256 or "").strip().lower()
+ if expected and actual_hash.lower() != expected:
+ raise ValueError(
+ f"{version.original_filename}: stored file hash verification failed."
+ )
+
+ safe_name = Path(version.original_filename or f"statement_{version.id}.pdf").name
+ if Path(safe_name).suffix.lower() != ".pdf":
+ raise ValueError(f"{safe_name}: only stored PDF bank statements can be reused.")
+
+ target = input_dir / f"stored_{order:03d}_v{version.id}_{safe_name}"
+ shutil.copy2(source, target)
+ prepared.append(target)
+ provenance.append(
+ {
+ "version_id": int(version.id),
+ "document_id": int(version.document_id),
+ "engagement_id": int(version.engagement_id),
+ "original_filename": version.original_filename,
+ "copied_filename": target.name,
+ "sha256": actual_hash,
+ }
+ )
+
+ db.flush()
+ return prepared, provenance, pending
+
+
+def deduplicate_source_paths(paths: list[Path]):
+ unique = []
+ hashes = []
+ seen = set()
+
+ for path in paths:
+ digest = _sha256(path)
+ if digest in seen:
+ path.unlink(missing_ok=True)
+ continue
+ seen.add(digest)
+ unique.append(path)
+ hashes.append({"filename": path.name, "sha256": digest})
+
+ return unique, hashes
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 a40c421..99d9097 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
@@ -65,7 +65,7 @@
I confirm that all statements uploaded in this analysis belong to the selected client or one of its ERP-recorded legal/trade/business names.
-
|---|