← Files VeraARCHIVED FILE
modules/journal-bank-reconciliation/scripts/review_session.py
48.8 KB · Oct 2, 2026 · 00:29 UTC
from __future__ import annotations
import json
import re
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Sequence
from excel_sanitization import sanitize_excel_string
__all__ = [
"ReviewSessionResult",
"RunIntakeResult",
"WORKBOOK_REQUIRED_HEADERS",
"WORKBOOK_REQUIRED_SHEETS",
"review_notes_copy",
"refresh_final_artifacts",
"refresh_review_execution_trace",
"write_review_session_artifacts",
"write_run_intake",
]
SCHEMA_VERSION = "1.0"
PLUGIN_NAME = "journal-bank-reconciliation"
WORKFLOW_NAME = "journal-bank-reconciliation"
MAX_MATCH_ITEMS = 200
MAX_UNMATCHED_ITEMS = 500
FINAL_ARTIFACT_EXCLUDED_NAMES = frozenset(
{
"run_intake.json",
"review_payload.json",
"ui_decisions.json",
"final_artifacts.json",
}
)
RECONCILIATION_COMMAND = (
"python",
"plugins/journal-bank-reconciliation/scripts/run_reconciliation.py",
)
def review_notes_copy(language: object) -> dict[str, str]:
"""Return the same fixed language copy for rendering and output checks."""
locale = str(language).lower().replace("_", "-").split("-", 1)[0]
rows = {
"it": (
"Revisione della riconciliazione bancaria",
"Lingua",
"Movimenti bancari",
"Registrazioni contabili",
"Corrispondenze",
"Movimenti bancari non riconciliati",
"Registrazioni non riconciliate",
"Corrispondenze per metodo",
"nessuna",
"Controlli da svolgere",
"Verifica conto, periodo e segno degli importi. Segui una corrispondenza nei due documenti e usa gli elenchi dei movimenti non riconciliati per richiedere le prove mancanti. Il confronto riguarda solo i file forniti; non verifica il trattamento contabile o fiscale dell'operazione.",
),
"en": (
"Journal-Bank Reconciliation Review Notes",
"Language",
"Bank rows",
"Journal rows",
"Matched rows",
"Unmatched bank rows",
"Unmatched journal rows",
"Stage Counts",
"none",
"Review Policy",
"Check the account, period and amount signs. Trace a match to both documents and use the unmatched lists to request missing evidence. The comparison covers only the supplied files; it does not verify the accounting or tax treatment of the transaction.",
),
"fr": (
"Revue du rapprochement bancaire",
"Langue",
"Mouvements bancaires",
"Écritures comptables",
"Correspondances",
"Mouvements bancaires non rapprochés",
"Écritures non rapprochées",
"Correspondances par méthode",
"aucune",
"Contrôles à effectuer",
"Vérifiez le compte, la période et le signe des montants. Retrouvez une correspondance dans les deux documents et utilisez les listes des mouvements non rapprochés pour demander les justificatifs manquants. La comparaison porte uniquement sur les fichiers fournis ; elle ne vérifie pas le traitement comptable ou fiscal de l'opération.",
),
"de": (
"Prüfung der Bankabstimmung",
"Sprache",
"Bankbewegungen",
"Buchungen",
"Zuordnungen",
"Nicht zugeordnete Bankbewegungen",
"Nicht zugeordnete Buchungen",
"Zuordnungen nach Methode",
"keine",
"Weitere Prüfung",
"Prüfen Sie Konto, Zeitraum und Vorzeichen der Beträge. Verfolgen Sie eine Zuordnung in beiden Dokumenten und fordern Sie anhand der nicht zugeordneten Bewegungen fehlende Belege an. Der Vergleich umfasst nur die bereitgestellten Dateien; er prüft nicht die buchhalterische oder steuerliche Behandlung des Vorgangs.",
),
"es": (
"Notas de revisión de la conciliación entre diario y banco",
"Idioma",
"Movimientos bancarios",
"Asientos del diario",
"Filas conciliadas",
"Movimientos bancarios sin conciliar",
"Asientos del diario sin conciliar",
"Recuento por etapa",
"ninguno",
"Política de revisión",
"Compruebe la cuenta, el período y el signo de los importes. Siga una coincidencia en ambos documentos y utilice las listas de movimientos sin conciliar para solicitar los justificantes que faltan. La comparación abarca solo los archivos facilitados; no verifica el tratamiento contable o fiscal de la operación.",
),
}
keys = (
"title",
"language",
"bank_row_count",
"journal_row_count",
"matched_count",
"unmatched_bank_count",
"unmatched_journal_count",
"stages",
"none",
"review",
"guidance",
)
return dict(zip(keys, rows.get(locale, rows["en"]), strict=True))
@dataclass(frozen=True)
class RunIntakeResult:
"""Run intake artifact written before journal-bank reconciliation."""
run_id: str
path: Path
@dataclass(frozen=True)
class ReviewSessionResult:
"""Review-session artifacts for one journal-bank reconciliation run."""
run_id: str
run_intake_path: Path
review_payload_path: Path
ui_decisions_path: Path
final_artifacts_path: Path
review_item_count: int
def _utc_now() -> str:
return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def _safe_slug(value: str) -> str:
slug = re.sub(r"[^a-zA-Z0-9_.-]+", "-", value).strip("-._").lower()
return slug or "run"
def _run_id(bank_path: Path, journal_path: Path) -> str:
timestamp = re.sub(r"[^0-9]", "", _utc_now())
return (
f"{PLUGIN_NAME}-{_safe_slug(bank_path.stem)}-"
f"{_safe_slug(journal_path.stem)}-{timestamp}"
)
def _write_json(path: Path, payload: dict[str, Any]) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(payload, ensure_ascii=False, indent=2, default=str) + "\n",
encoding="utf-8",
)
return path
def _write_review_handoff_card(
output_dir: Path,
*,
run_id: str,
title: str,
validate_tool: str,
render_tool: str,
save_tool: str,
apply_tool: str,
language: str,
) -> Path:
path = output_dir / "review_handoff.md"
if _is_spanish(language):
lines = [
f"# Entrega para revisión: {title}",
"",
f"- ID de ejecución: `{run_id}`",
"- Datos de revisión: `review_payload.json`",
"- Datos de entrada de la ejecución: `run_intake.json`",
"- Decisiones pendientes: `ui_decisions.json`",
"- Decisiones aplicadas: `applied_decisions.json`",
"- Artefactos finales: `final_artifacts.json`",
"",
"## Revisión en Codex",
f"1. Valide los datos con `{validate_tool}`.",
f"2. Muestre el espacio de revisión con `{render_tool}`.",
f"3. Guarde las acciones de revisión con `{save_tool}`.",
f"4. Aplique las acciones de revisión con `{apply_tool}`.",
"",
"El guardado y la aplicación persistentes requieren la interfaz de revisión MCP o del servidor local. "
"La alternativa HTML estática solo permite copiar o descargar el JSON de decisiones.",
"",
"<!-- Review Handoff -->",
]
else:
lines = [
f"# {title} Review Handoff",
"",
f"- Run ID: `{run_id}`",
"- Review payload: `review_payload.json`",
"- Run intake: `run_intake.json`",
"- Pending decisions: `ui_decisions.json`",
"- Applied decisions: `applied_decisions.json`",
"- Final artifacts: `final_artifacts.json`",
"",
"## Review In Codex",
f"1. Validate the payload with `{validate_tool}`.",
f"2. Render the review workbench with `{render_tool}`.",
f"3. Save reviewer actions with `{save_tool}`.",
f"4. Apply reviewer actions with `{apply_tool}`.",
"",
"Persistent save/apply requires the MCP or local-server review surface. "
"Static HTML fallback can copy or download decision JSON only.",
]
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
return path
def _review_handoff_output_record(path: Path, language: str) -> dict[str, Any]:
return {
"path": path.name,
"size_bytes": path.stat().st_size,
"kind": "md",
"status": "written",
"required_text": [
"Entrega para revisión" if _is_spanish(language) else "Review Handoff",
*(["Review Handoff"] if _is_spanish(language) else []),
"review_payload.json",
"ui_decisions.json",
"applied_decisions.json",
"final_artifacts.json",
],
"qa_checks": ["nonempty_text", "required_text"],
}
def _local_output_refs(final_artifacts_path: Path) -> list[str]:
refs = [
"run_intake.json",
"review_payload.json",
"ui_decisions.json",
"final_artifacts.json",
]
payload = json.loads(final_artifacts_path.read_text(encoding="utf-8"))
outputs = payload.get("outputs")
if isinstance(outputs, list):
for output in outputs:
if not isinstance(output, dict):
continue
path_value = output.get("path")
if (
isinstance(path_value, str)
and path_value.strip()
and "://" not in path_value
):
refs.append(path_value.strip())
return list(dict.fromkeys(refs))
def _append_execution_trace(
run_intake_path: Path,
final_artifacts_path: Path,
*,
command: Sequence[str],
) -> None:
from vera_assurance.serialization import build_review_execution_step
payload = json.loads(run_intake_path.read_text(encoding="utf-8"))
step = build_review_execution_step(payload, WORKFLOW_NAME, command)
step["outputs"] = _local_output_refs(final_artifacts_path)
payload["execution_trace"] = [step]
_write_json(run_intake_path, payload)
def refresh_review_execution_trace(
run_intake_path: Path,
final_artifacts_path: Path,
) -> None:
"""Refresh the review-session trace against the current final manifest."""
_append_execution_trace(
run_intake_path,
final_artifacts_path,
command=RECONCILIATION_COMMAND,
)
def _as_output_ref(path: str | Path | None, output_dir: Path) -> str | None:
if path is None:
return None
candidate = Path(path)
try:
return candidate.relative_to(output_dir).as_posix()
except ValueError:
return candidate.as_posix()
def _json_object_if_present(path: Path) -> dict[str, Any]:
if not path.is_file():
return {}
payload = json.loads(path.read_text(encoding="utf-8"))
return payload if isinstance(payload, dict) else {}
def _clean_text(value: Any) -> str:
return " ".join(str(value or "").strip().split())
def _is_spanish(language: object) -> bool:
return (
str(language or "").strip().lower().replace("_", "-").split("-", 1)[0] == "es"
)
def _rows(frame: Any) -> list[dict[str, Any]]:
if frame is None:
return []
to_dicts = getattr(frame, "to_dicts", None)
if callable(to_dicts):
return [row for row in to_dicts() if isinstance(row, dict)]
if isinstance(frame, list):
return [row for row in frame if isinstance(row, dict)]
return []
def _base_item(
item_id: str,
item_type: str,
title: str,
*,
allowed_actions: Sequence[str],
recommended_action: str,
source_path: str | None = None,
output_path: str | None = None,
evidence: Sequence[dict[str, Any]] = (),
data: dict[str, Any] | None = None,
) -> dict[str, Any]:
return {
"id": item_id,
"item_type": item_type,
"title": title,
"source_path": source_path,
"output_path": output_path,
"allowed_actions": list(allowed_actions),
"recommended_action": recommended_action,
"evidence": list(evidence),
"data": data or {},
"status": "needs_review",
}
def _review_columns(language: str) -> list[dict[str, str]]:
if _is_spanish(language):
return [
{"field": "item_type", "label": "Tipo"},
{"field": "title", "label": "Movimiento"},
{"field": "recommended_action", "label": "Acción sugerida"},
{"field": "source_path", "label": "Fuente"},
{"field": "output_path", "label": "Salida"},
{"field": "status", "label": "Estado"},
]
return [
{"field": "item_type", "label": "Type"},
{"field": "title", "label": "Movement"},
{"field": "recommended_action", "label": "Suggested action"},
{"field": "source_path", "label": "Source"},
{"field": "output_path", "label": "Output"},
{"field": "status", "label": "Status"},
]
def _transaction_title(row: dict[str, Any], fallback: str) -> str:
parts = [
_clean_text(row.get("transaction_date")),
_clean_text(row.get("amount_signed") or row.get("amount_abs")),
_clean_text(row.get("reference") or row.get("movement_number")),
_clean_text(row.get("beneficiary") or row.get("description")),
]
return " | ".join(part for part in parts if part) or fallback
def _match_title(row: dict[str, Any], index: int, language: str) -> str:
parts = [
_clean_text(row.get("bank_amount")),
_clean_text(row.get("shared_references")),
_clean_text(row.get("stage")),
]
fallback = (
f"Par conciliado {index}" if _is_spanish(language) else f"Matched pair {index}"
)
return " | ".join(part for part in parts if part) or fallback
def _requested_reconciliation_evidence(
row: dict[str, Any], side: str, language: str
) -> tuple[str, str]:
localized = {
"it": (
"movimento non riconciliato",
"Scrittura del giornale o del mastro a supporto del movimento bancario {ref}",
"Il movimento bancario non ha una corrispondenza nel giornale secondo i controlli eseguiti.",
"Estratto conto o prova del pagamento per il movimento del giornale {ref}",
"Il movimento del giornale non ha una corrispondenza bancaria secondo i controlli eseguiti.",
),
"fr": (
"mouvement non rapproché",
"Écriture du journal ou du grand livre justifiant le mouvement bancaire {ref}",
"Le mouvement bancaire n’a pas de correspondance dans le journal selon les contrôles effectués.",
"Relevé bancaire ou preuve de paiement pour le mouvement du journal {ref}",
"Le mouvement du journal n’a pas de correspondance bancaire selon les contrôles effectués.",
),
"de": (
"nicht abgestimmte Buchung",
"Journal- oder Hauptbuchbeleg für die Bankbuchung {ref}",
"Die Bankbuchung hat nach den durchgeführten Prüfungen keine Zuordnung im Journal.",
"Kontoauszug oder Zahlungsnachweis für die Journalbuchung {ref}",
"Die Journalbuchung hat nach den durchgeführten Prüfungen keine Bankzuordnung.",
),
}
if language in localized:
fallback, bank_request, bank_reason, journal_request, journal_reason = (
localized[language]
)
descriptor = (
_clean_text(row.get("reference") or row.get("movement_number"))
or _clean_text(row.get("amount_signed") or row.get("amount_abs"))
or fallback
)
return (
(bank_request.format(ref=descriptor), bank_reason)
if side == "bank"
else (journal_request.format(ref=descriptor), journal_reason)
)
reference = _clean_text(row.get("reference") or row.get("movement_number"))
amount = _clean_text(row.get("amount_signed") or row.get("amount_abs"))
descriptor = (
reference
or amount
or (
"transacción sin conciliar"
if _is_spanish(language)
else "unmatched transaction"
)
)
if side == "bank":
return (
(
f"Justificante del diario o del mayor para la transacción bancaria {descriptor}"
if _is_spanish(language)
else f"Journal or ledger support for bank transaction {descriptor}"
),
(
"La transacción bancaria no tiene una correspondencia determinista en el diario."
if _is_spanish(language)
else "Bank transaction has no deterministic journal match."
),
)
return (
(
f"Extracto bancario o justificante de pago para la transacción del diario {descriptor}"
if _is_spanish(language)
else f"Bank statement or payment evidence for journal transaction {descriptor}"
),
(
"La transacción del diario no tiene una correspondencia bancaria determinista."
if _is_spanish(language)
else "Journal transaction has no deterministic bank match."
),
)
def _unmatched_items(
rows: Sequence[dict[str, Any]],
*,
side: str,
output_path: str,
language: str,
) -> list[dict[str, Any]]:
item_type = "unmatched_bank" if side == "bank" else "unmatched_journal"
if _is_spanish(language):
source_label = "Banco" if side == "bank" else "Diario"
row_label = "fila"
else:
source_label = "Bank" if side == "bank" else "Journal"
row_label = "row"
items: list[dict[str, Any]] = []
for index, row in enumerate(rows[:MAX_UNMATCHED_ITEMS], start=1):
requested_document, reason = _requested_reconciliation_evidence(
row, side, language
)
data = dict(row)
data["requested_document"] = requested_document
data["reason"] = reason
items.append(
_base_item(
f"{item_type}-{index}",
item_type,
_transaction_title(row, f"{source_label} row {index}"),
source_path="; ".join(
part
for part in (
_clean_text(row.get("source_file")),
(
f"{row_label} {_clean_text(row.get('source_row'))}"
if _clean_text(row.get("source_row"))
else ""
),
)
if part
)
or None,
output_path=output_path,
allowed_actions=(
"accept",
"edit",
"mark_unclear",
"request_more_documents",
"skip",
),
recommended_action="request_more_documents",
evidence=[
{
"kind": "unmatched_transaction",
"side": side,
"transaction_id": row.get("transaction_id"),
"amount_abs": row.get("amount_abs"),
"reference": row.get("reference"),
"movement_number": row.get("movement_number"),
},
{
"kind": "missing_reconciliation_evidence",
"side": side,
"requested_document": requested_document,
"reason": reason,
"status": "needs_evidence",
},
],
data=data,
)
)
if len(rows) > MAX_UNMATCHED_ITEMS:
items.append(
_base_item(
f"{item_type}-truncated",
"review_artifact",
(
f"Filas sin conciliar del {source_label.lower()} truncadas en el widget"
if _is_spanish(language)
else f"{source_label} unmatched rows truncated in widget"
),
output_path=output_path,
allowed_actions=("accept", "mark_unclear", "skip"),
recommended_action="mark_unclear",
data={
"shown_count": MAX_UNMATCHED_ITEMS,
"total_count": len(rows),
"full_results": output_path,
},
)
)
return items
def _match_items(rows: Sequence[dict[str, Any]], language: str) -> list[dict[str, Any]]:
items: list[dict[str, Any]] = []
for index, row in enumerate(rows[:MAX_MATCH_ITEMS], start=1):
data = dict(row)
data["target_artifact"] = "reconciliation_matches.csv"
data["target_id_field"] = "bank_transaction_id"
data["target_record_id"] = str(row.get("bank_transaction_id") or "")
data["target_field"] = "review_note"
data["edit_hint"] = (
"Editar este par conciliado actualiza review_note en reconciliation_matches.csv para el bank_transaction_id correspondiente."
if _is_spanish(language)
else "Editing this matched pair updates review_note in reconciliation_matches.csv for the matching bank_transaction_id."
)
items.append(
_base_item(
f"matched-pair-{index}",
"matched_pair",
_match_title(row, index, language),
output_path="reconciliation_matches.csv",
allowed_actions=("accept", "edit", "mark_unclear", "skip"),
recommended_action="accept",
evidence=[
{
"kind": "deterministic_match",
"stage": row.get("stage"),
"amount_delta": row.get("amount_delta"),
"date_diff_days": row.get("date_diff_days"),
"shared_references": row.get("shared_references"),
}
],
data=data,
)
)
return items
def _artifact_items(
audit: dict[str, Any], output_dir: Path, language: str
) -> list[dict[str, Any]]:
outputs = audit.get("outputs") if isinstance(audit.get("outputs"), dict) else {}
spanish = _is_spanish(language)
labels = {
"normalized_bank_csv": (
"review_artifact",
"CSV bancario normalizado" if spanish else "Normalized bank CSV",
),
"normalized_journal_csv": (
"review_artifact",
"CSV del diario normalizado" if spanish else "Normalized journal CSV",
),
"reconciliation_matches_csv": (
"review_artifact",
(
"CSV de coincidencias de conciliación"
if spanish
else "Reconciliation matches CSV"
),
),
"relationship_residuals_csv": (
"review_artifact",
(
"CSV de residuales de la relación"
if spanish
else "Relationship residuals CSV"
),
),
"unmatched_bank_csv": (
"review_artifact",
"CSV bancario sin conciliar" if spanish else "Unmatched bank CSV",
),
"unmatched_journal_csv": (
"review_artifact",
"CSV del diario sin conciliar" if spanish else "Unmatched journal CSV",
),
"workbook_xlsx": (
"workpaper_artifact",
(
"Libro de conciliación entre diario y banco"
if spanish
else "Journal-bank reconciliation workbook"
),
),
"audit_json": (
"review_artifact",
(
"JSON de auditoría de la conciliación"
if spanish
else "Reconciliation audit JSON"
),
),
"review_notes_md": (
"review_artifact",
"Notas de revisión" if spanish else "Review notes",
),
"material_value_ledger_json": (
"review_artifact",
(
"Libro de direcciones de valores materiales"
if spanish
else "Material-value address ledger"
),
),
}
items: list[dict[str, Any]] = []
for index, (field, (item_type, title)) in enumerate(labels.items(), start=1):
path_value = outputs.get(field)
if not path_value:
continue
path_ref = _as_output_ref(path_value, output_dir)
exists = Path(path_value).exists()
items.append(
_base_item(
f"artifact-{index}",
item_type,
title,
output_path=path_ref,
allowed_actions=("accept", "edit", "mark_unclear", "skip"),
recommended_action="accept" if exists else "mark_unclear",
evidence=[
{
"kind": "artifact_status",
"field": field,
"path": path_ref,
"exists": exists,
}
],
data={"field": field, "path": path_ref, "exists": exists},
)
)
return items
MATCH_WORKBOOK_COLUMNS = [
"status",
"stage",
"bank_transaction_id",
"journal_transaction_id",
"bank_date",
"journal_date",
"date_diff_days",
"bank_amount",
"journal_amount",
"amount_delta",
"bank_description",
"journal_description",
"shared_references",
"review_note",
]
RESIDUAL_WORKBOOK_COLUMNS = [
"side",
"record_ref",
"transaction_id",
"record_amount",
"allocated_amount",
"residual",
"currency",
"unit",
"entity_ref",
"party_ref",
]
TRANSACTION_WORKBOOK_COLUMNS = [
"side",
"transaction_id",
"transaction_date",
"amount_signed",
"amount_abs",
"description",
"beneficiary",
"reference",
"movement_number",
"account",
"currency",
"unit",
"entity_ref",
"party_ref",
"direction",
"source_file",
"source_sheet",
"source_row",
]
NON_MOVEMENT_WORKBOOK_COLUMNS = [
"side",
"source_file",
"source_sheet",
"source_row",
"classification",
"reason",
"transaction_date",
"amount_signed",
"amount_abs",
"description",
]
WORKBOOK_REQUIRED_HEADERS = {
"matches": MATCH_WORKBOOK_COLUMNS,
"relationship_residuals": RESIDUAL_WORKBOOK_COLUMNS,
"unmatched_bank": TRANSACTION_WORKBOOK_COLUMNS,
"unmatched_journal": TRANSACTION_WORKBOOK_COLUMNS,
"bank_pdf_non_movements": NON_MOVEMENT_WORKBOOK_COLUMNS,
"normalized_bank": TRANSACTION_WORKBOOK_COLUMNS,
"normalized_journal": TRANSACTION_WORKBOOK_COLUMNS,
}
WORKBOOK_REQUIRED_SHEETS = [
"matches",
"relationship_residuals",
"unmatched_bank",
"unmatched_journal",
"bank_pdf_non_movements",
"normalized_bank",
"normalized_journal",
]
def _column_letters(index: int) -> str:
letters = ""
while index > 0:
index, remainder = divmod(index - 1, 26)
letters = chr(65 + remainder) + letters
return letters
def _cell_reference(columns: Sequence[str], field: str, row: int) -> str:
return f"{_column_letters(list(columns).index(field) + 1)}{row}"
def _add_cell_check(cells: dict[str, str], reference: str, value: object) -> None:
text = sanitize_excel_string(_clean_text(value))
if text:
cells[reference] = text
def _required_sheet_cells(
*,
columns: Sequence[str],
fields: Sequence[str],
first_row: dict[str, Any] | None,
) -> dict[str, str]:
cells: dict[str, str] = {}
for field in fields:
if field not in columns:
continue
cells[_cell_reference(columns, field, 1)] = field
if first_row:
_add_cell_check(
cells, _cell_reference(columns, field, 2), first_row.get(field)
)
return cells
def _workbook_required_cells(
match_rows: Sequence[dict[str, Any]],
relationship_residual_rows: Sequence[dict[str, Any]],
unmatched_bank_rows: Sequence[dict[str, Any]],
unmatched_journal_rows: Sequence[dict[str, Any]],
bank_pdf_non_movement_rows: Sequence[dict[str, Any]],
) -> dict[str, dict[str, str]]:
return {
"matches": _required_sheet_cells(
columns=MATCH_WORKBOOK_COLUMNS,
fields=[
"status",
"stage",
"bank_transaction_id",
"journal_transaction_id",
"shared_references",
],
first_row=match_rows[0] if match_rows else None,
),
"relationship_residuals": _required_sheet_cells(
columns=RESIDUAL_WORKBOOK_COLUMNS,
fields=[
"side",
"record_ref",
"transaction_id",
"record_amount",
"allocated_amount",
"residual",
],
first_row=(
relationship_residual_rows[0] if relationship_residual_rows else None
),
),
"unmatched_bank": _required_sheet_cells(
columns=TRANSACTION_WORKBOOK_COLUMNS,
fields=["side", "transaction_id", "transaction_date", "reference"],
first_row=unmatched_bank_rows[0] if unmatched_bank_rows else None,
),
"unmatched_journal": _required_sheet_cells(
columns=TRANSACTION_WORKBOOK_COLUMNS,
fields=["side", "transaction_id", "transaction_date", "reference"],
first_row=unmatched_journal_rows[0] if unmatched_journal_rows else None,
),
"bank_pdf_non_movements": _required_sheet_cells(
columns=NON_MOVEMENT_WORKBOOK_COLUMNS,
fields=[
"source_file",
"source_sheet",
"source_row",
"classification",
"description",
"amount_abs",
],
first_row=(
bank_pdf_non_movement_rows[0] if bank_pdf_non_movement_rows else None
),
),
}
def _output_records(
output_dir: Path,
audit: dict[str, Any],
*,
match_rows: Sequence[dict[str, Any]],
relationship_residual_rows: Sequence[dict[str, Any]],
unmatched_bank_rows: Sequence[dict[str, Any]],
unmatched_journal_rows: Sequence[dict[str, Any]],
bank_pdf_non_movement_rows: Sequence[dict[str, Any]],
) -> list[dict[str, Any]]:
outputs: list[dict[str, Any]] = []
for path in sorted(output_dir.rglob("*")):
if not path.is_file() or path.name in FINAL_ARTIFACT_EXCLUDED_NAMES:
continue
relative = path.relative_to(output_dir).as_posix()
output = {
"path": relative,
"size_bytes": path.stat().st_size,
"kind": path.suffix.lower().lstrip(".") or "file",
"status": "written",
}
if relative == "journal_bank_reconciliation.xlsx":
output["required_sheets"] = WORKBOOK_REQUIRED_SHEETS
output["required_sheet_headers"] = WORKBOOK_REQUIRED_HEADERS
output["required_cells"] = _workbook_required_cells(
match_rows,
relationship_residual_rows,
unmatched_bank_rows,
unmatched_journal_rows,
bank_pdf_non_movement_rows,
)
output["source_row_counts"] = {
"matches": int(audit.get("matched_count", 0)),
"relationship_residuals": int(
audit.get("relationship_residual_row_count", 0)
),
"unmatched_bank": int(audit.get("unmatched_bank_count", 0)),
"unmatched_journal": int(audit.get("unmatched_journal_count", 0)),
"bank_pdf_non_movements": int(
audit.get("bank_pdf_non_movement_row_count", 0)
),
"normalized_bank": int(audit.get("bank_row_count", 0)),
"normalized_journal": int(audit.get("journal_row_count", 0)),
}
output["qa_checks"] = [
"office_zip",
"workbook_xml",
"required_sheets",
"required_sheet_headers",
"required_cells",
]
elif relative == "reconciliation_matches.csv":
output["row_count"] = int(audit.get("matched_count", 0))
output["required_columns"] = [
"status",
"bank_transaction_id",
"journal_transaction_id",
"amount_delta",
]
elif relative == "relationship_residuals.csv":
output["row_count"] = int(audit.get("relationship_residual_row_count", 0))
output["required_columns"] = RESIDUAL_WORKBOOK_COLUMNS
elif relative == "unmatched_bank.csv":
output["row_count"] = int(audit.get("unmatched_bank_count", 0))
output["required_columns"] = [
"transaction_id",
"transaction_date",
"amount_abs",
]
elif relative == "unmatched_journal.csv":
output["row_count"] = int(audit.get("unmatched_journal_count", 0))
output["required_columns"] = [
"transaction_id",
"transaction_date",
"amount_abs",
]
elif relative == "bank_pdf_non_movement_rows.csv":
output["row_count"] = int(audit.get("bank_pdf_non_movement_row_count", 0))
output["required_columns"] = [
"source_file",
"source_row",
"classification",
"description",
"amount_abs",
]
elif relative == "review_notes.md":
words = review_notes_copy(audit.get("language"))
output["required_text"] = [
f"# {words['title']}",
f"## {words['stages']}",
f"## {words['review']}",
]
output["qa_checks"] = ["nonempty_text", "required_text"]
outputs.append(output)
return outputs
def refresh_final_artifacts(final_artifacts_path: Path) -> dict[str, Any]:
"""Refresh the manifest against the current output files without losing QA."""
payload = json.loads(final_artifacts_path.read_text(encoding="utf-8"))
if not isinstance(payload, dict):
raise ValueError("final_artifacts.json must contain an object")
outputs = payload.get("outputs")
existing_by_path = (
{
str(output["path"]): dict(output)
for output in outputs
if isinstance(output, dict) and isinstance(output.get("path"), str)
}
if isinstance(outputs, list)
else {}
)
refreshed: list[dict[str, Any]] = []
output_dir = final_artifacts_path.parent
for path in sorted(output_dir.rglob("*")):
if not path.is_file() or path.name in FINAL_ARTIFACT_EXCLUDED_NAMES:
continue
relative = path.relative_to(output_dir).as_posix()
output = existing_by_path.get(relative, {})
output.update(
{
"path": relative,
"size_bytes": path.stat().st_size,
"kind": path.suffix.lower().lstrip(".") or "file",
}
)
output.setdefault("status", "written")
refreshed.append(output)
payload["outputs"] = refreshed
_write_json(final_artifacts_path, payload)
return payload
def write_run_intake(
output_dir: Path,
*,
bank_path: Path,
journal_path: Path,
recipe_path: Path | None,
sample_path: Path | None,
language: str,
document_language: str,
tolerance: str,
date_window_days: int,
client_run_id: str | None = None,
client_run_root: Path | None = None,
) -> RunIntakeResult:
"""Write run intake before deterministic matching."""
run_id = client_run_id or _run_id(bank_path, journal_path)
spanish = _is_spanish(language)
def run_reference(path_value: Path) -> str:
if client_run_root is None:
return path_value.as_posix()
run_root = client_run_root.expanduser().resolve()
try:
relative = path_value.expanduser().resolve().relative_to(run_root)
except ValueError as exc:
raise ValueError(
"Journal-Bank Reconciliation path is outside the run root."
) from exc
if not relative.parts:
raise ValueError(
"Journal-Bank Reconciliation path must identify a run artifact."
)
return relative.as_posix()
bank_ref = run_reference(bank_path)
journal_ref = run_reference(journal_path)
sample_ref = run_reference(sample_path) if sample_path is not None else None
recipe_ref = run_reference(recipe_path) if recipe_path is not None else None
output_ref = run_reference(output_dir)
local_files_read = [bank_ref, journal_ref]
if recipe_path is not None:
local_files_read.append(recipe_ref)
if sample_path is not None:
local_files_read.append(sample_ref)
payload = {
"schema_version": SCHEMA_VERSION,
"plugin": PLUGIN_NAME,
"workflow": WORKFLOW_NAME,
"run_id": run_id,
**(
{"path_reference": "run_root_relative"}
if client_run_root is not None
else {}
),
"created_at": _utc_now(),
"language": language,
"input_paths": [
bank_ref,
journal_ref,
*([sample_ref] if sample_ref else []),
],
"output_dir": output_ref,
"inferred_task": "journal_bank_reconciliation_review_payload",
"assumptions": {
"bank_path": bank_ref,
"journal_path": journal_ref,
"sample_path": sample_ref,
"recipe_path": recipe_ref,
"language": language,
"document_language": document_language,
"currency": "EUR",
"tolerance": tolerance,
"date_window_days": date_window_days,
},
"unresolved_questions": [],
"dependency_check": {
"status": "not_run_by_script",
"note": (
"Codex debe ejecutar scripts/check_dependencies.py antes de los scripts auxiliares."
if spanish
else "Codex should run scripts/check_dependencies.py before helper scripts."
),
},
"data_posture": {
"local_files_read": local_files_read,
"external_connectors_used": [],
"upload_paths_used": [],
"remote_sql_execution_used": False,
"hosted_notebook_execution_used": False,
"notes": [
(
"Los scripts de conciliación leen localmente los archivos bancarios, del diario, de la receta opcional y de la muestra opcional."
if spanish
else "Matching scripts read bank, journal, optional recipe, and optional sample files locally."
),
(
"De forma predeterminada no se utiliza ningún conector externo, ruta de carga, SQL remoto ni cuaderno alojado."
if spanish
else "No external connector, upload path, remote SQL, or hosted notebook execution is used by default."
),
],
},
"status": "ready_for_reconciliation_run",
}
return RunIntakeResult(
run_id=run_id,
path=_write_json(output_dir / "run_intake.json", payload),
)
def write_review_session_artifacts(
output_dir: Path,
*,
run_id: str,
run_intake_path: Path,
matches: Any,
unmatched_bank: Any,
unmatched_journal: Any,
audit: dict[str, Any],
bank_pdf_non_movements: Any = None,
relationship_residuals: Any = None,
) -> ReviewSessionResult:
"""Write review payload, pending decisions, and final artifacts."""
match_rows = _rows(matches)
unmatched_bank_rows = _rows(unmatched_bank)
unmatched_journal_rows = _rows(unmatched_journal)
bank_pdf_non_movement_rows = _rows(bank_pdf_non_movements)
relationship_residual_rows = _rows(relationship_residuals)
language = str(audit.get("language") or "en")
items: list[dict[str, Any]] = []
items.extend(
_unmatched_items(
unmatched_bank_rows,
side="bank",
output_path="unmatched_bank.csv",
language=language,
)
)
items.extend(
_unmatched_items(
unmatched_journal_rows,
side="journal",
output_path="unmatched_journal.csv",
language=language,
)
)
items.extend(_match_items(match_rows, language))
items.extend(_artifact_items(audit, output_dir, language))
gate_register = _json_object_if_present(output_dir / "assurance_gates.json")
gate_entries = gate_register.get("gates")
gate_statuses = (
{
str(name): str(gate.get("status") or "not_assessed")
for name, gate in gate_entries.items()
if isinstance(gate, dict)
}
if isinstance(gate_entries, dict)
else {}
)
review_payload = {
"schema_version": SCHEMA_VERSION,
"plugin": PLUGIN_NAME,
"workflow": WORKFLOW_NAME,
"run_id": run_id,
"created_at": _utc_now(),
"language": language,
"source_paths": [
audit.get("bank_path"),
audit.get("journal_path"),
audit.get("sample_path"),
],
"review_type": "journal_bank_reconciliation_review",
"items": items,
"item_count": len(items),
"columns": _review_columns(language),
"source_artifacts": {
"run_intake": _as_output_ref(run_intake_path, output_dir),
"audit": "reconciliation_audit.json",
"input_receipts": "input_receipts.json",
"source_qualifications": "source_qualifications.json",
"reviewed_decisions": "reviewed_decisions.json",
"lineage": "lineage.json",
"relationship_ledger": "relationship_ledger.json",
"relationship_residuals": "relationship_residuals.csv",
"material_value_ledger": "material_value_ledger.json",
"assurance_gates": "assurance_gates.json",
"artifact_receipts": "artifact_receipts.json",
"review_notes": "review_notes.md",
"matches": "reconciliation_matches.csv",
"unmatched_bank": "unmatched_bank.csv",
"unmatched_journal": "unmatched_journal.csv",
"bank_pdf_non_movements": "bank_pdf_non_movement_rows.csv",
"workbook": "journal_bank_reconciliation.xlsx",
},
"allowed_actions": [
"accept",
"reject",
"edit",
"mark_unclear",
"request_more_documents",
"skip",
],
"status": "ready_for_review",
"assurance": {
"gate_statuses": gate_statuses,
"report_ready": bool(gate_register.get("report_ready", False)),
"relationship_balanced": bool(audit.get("relationship_balanced", False)),
"unresolved_relationship_rows": int(audit.get("unmatched_bank_count") or 0)
+ int(audit.get("unmatched_journal_count") or 0),
},
"summary": {
"bank_row_count": audit.get("bank_row_count", len(match_rows)),
"journal_row_count": audit.get("journal_row_count", 0),
"matched_count": audit.get("matched_count", len(match_rows)),
"relationship_residual_row_count": audit.get(
"relationship_residual_row_count",
len(relationship_residual_rows),
),
"unmatched_bank_count": audit.get(
"unmatched_bank_count", len(unmatched_bank_rows)
),
"unmatched_journal_count": audit.get(
"unmatched_journal_count", len(unmatched_journal_rows)
),
"bank_pdf_non_movement_row_count": audit.get(
"bank_pdf_non_movement_row_count", len(bank_pdf_non_movement_rows)
),
"bank_pdf_non_movement_classifications": audit.get(
"bank_pdf_non_movement_classifications", {}
),
"stage_counts": audit.get("stage_counts", {}),
"sample_movement_count": audit.get("sample_movement_count", 0),
"tolerance": audit.get("tolerance"),
"date_window_days": audit.get("date_window_days"),
},
}
review_payload_path = _write_json(
output_dir / "review_payload.json",
review_payload,
)
ui_decisions_path = _write_json(
output_dir / "ui_decisions.json",
{
"schema_version": SCHEMA_VERSION,
"plugin": PLUGIN_NAME,
"workflow": WORKFLOW_NAME,
"run_id": run_id,
"decided_at": None,
"decision_source": "not_collected",
"review_payload_path": review_payload_path.name,
"decisions": [],
"decision_count": 0,
"status": "pending_review",
},
)
review_handoff_path = _write_review_handoff_card(
output_dir,
run_id=run_id,
title=(
"Conciliación entre diario y banco"
if _is_spanish(language)
else "Journal-Bank Reconciliation"
),
validate_tool="validate_journal_bank_review",
render_tool="render_journal_bank_review",
save_tool="save_journal_bank_decisions",
apply_tool="apply_journal_bank_decisions",
language=language,
)
outputs = _output_records(
output_dir,
audit,
match_rows=match_rows,
relationship_residual_rows=relationship_residual_rows,
unmatched_bank_rows=unmatched_bank_rows,
unmatched_journal_rows=unmatched_journal_rows,
bank_pdf_non_movement_rows=bank_pdf_non_movement_rows,
)
outputs = [
output
for output in outputs
if not (
isinstance(output, dict) and output.get("path") == review_handoff_path.name
)
]
outputs.append(_review_handoff_output_record(review_handoff_path, language))
spanish = _is_spanish(language)
final_artifacts_path = _write_json(
output_dir / "final_artifacts.json",
{
"schema_version": SCHEMA_VERSION,
"plugin": PLUGIN_NAME,
"workflow": WORKFLOW_NAME,
"run_id": run_id,
"completed_at": _utc_now(),
"outputs": outputs,
"caveats": [
(
"Las coincidencias deterministas solo se aceptan conforme a las reglas del script; las filas sin conciliar requieren la interpretación explícita de Codex o de la persona revisora."
if spanish
else "Deterministic matches are accepted only by the script rules; unmatched rows require explicit Codex or reviewer interpretation."
),
(
"Los datos de revisión MCP tienen un límite; use las salidas CSV, XLSX y JSON como conjunto completo de evidencias."
if spanish
else "The MCP review payload is bounded; use CSV/XLSX/JSON outputs as the complete evidence set."
),
(
"ui_decisions.json queda pendiente hasta que Codex, el widget MCP o la revisión alternativa registren las decisiones."
if spanish
else "ui_decisions.json is pending until Codex, the MCP widget, or fallback review records decisions."
),
],
"next_actions": [
(
"Ejecute validate_journal_bank_review y, cuando MCP esté disponible, render_journal_bank_review."
if spanish
else "Call validate_journal_bank_review, then render_journal_bank_review when MCP is available."
),
(
"Revise las filas bancarias y del diario sin conciliar antes de considerar completo el paquete."
if spanish
else "Review unmatched bank and journal rows before treating the package as complete."
),
(
"No convierta filas ambiguas en coincidencias sin modificar las reglas deterministas y volver a ejecutar el proceso."
if spanish
else "Do not promote ambiguous rows to matched without changing deterministic rules and rerunning."
),
],
"status": "written_pending_review",
},
)
refresh_review_execution_trace(run_intake_path, final_artifacts_path)
return ReviewSessionResult(
run_id=run_id,
run_intake_path=run_intake_path,
review_payload_path=review_payload_path,
ui_decisions_path=ui_decisions_path,
final_artifacts_path=final_artifacts_path,
review_item_count=len(items),
)
SHA-256: f890ad8853d182e5355c157db7dbf3d134481a04e2d4a3a1b0784792890b0e1d