"""Read-only source projections for the unified review queue.
Each projection reads one application-owned source capability and emits a tuple
of typed :class:`ReviewItem` records. Projections are pure and stateless; they
tolerate missing source records by returning an empty tuple. Severity is
derived per source via a first-match-wins predicate table.
Persistence is supplied by the required :class:`DraftReviewPorts` bundle. This
module owns only source classification and typed review-item projection; the
encrypted repositories remain outer composition concerns.
"""
from __future__ import annotations
from datetime import UTC, datetime, time
from decimal import Decimal
from pathlib import Path
from pydantic import ValidationError
from ...core.config import Settings
from ...core.errors.hierarchy import CadrumoError
from ...core.errors.severity import BaseSeverity
from ...core.i18n.translatable import Translatable as tr
from ...core.logging import get_logger
from ...domain.calculations.registry.authority import PinnedAuthorityOperation
from ...domain.filing.schema import ModeloDraft, ModeloValidationFinding
from ...domain.invoices.enums import PaymentStatus
from ...domain.invoices.models import Invoice, InvoiceCatalogue
from ...domain.submission.models import ModeloDraftStatus
from ...domain.transactions.enums import BusinessClassification, is_classified
from ...domain.transactions.models import Transaction, TransactionCatalogue
from ..filing.draft_review import ModeloApprovalStaleReason
from ..filing.draft_review_ports import DraftReviewPorts, DraftReviewProfileRepositoryProtocol
from ..filing.draft_revision_gate import require_modelo_draft_coordinates_current
from .enums import ReviewSeverity
from .errors import ReviewSourceLoadError
from .models import (
FindingReviewItem,
InvoiceReviewItem,
TransactionReviewItem,
)
_LOGGER = get_logger(__name__)
_SUMMARY_MAX = 80
# ── transactions ──────────────────────────────────────────────────
[docs]
def transactions_pending(
*,
ports: DraftReviewPorts,
catalogue: TransactionCatalogue | None = None,
) -> tuple[TransactionReviewItem, ...]:
"""Return one :class:`TransactionReviewItem` per pending-review transaction.
``catalogue`` is an optional :class:`TransactionCatalogue` override; the repository is
loaded when ``None``.
Skips fully-classified rows (BUSINESS / PERSONAL / MIXED), rows
explicitly skipped by rule (``SKIPPED_BY_RULE``), and rows the
operator reviewed and deliberately excluded (``REVIEWED_EXCLUDED``)
— those have a final disposition and do not want the operator's
attention.
"""
if catalogue is None:
catalogue = load_transactions(ports=ports)
if catalogue is None:
return ()
items: list[TransactionReviewItem] = []
for transaction in catalogue.values():
severity = _classify_transaction(transaction.business_classification)
if severity is None:
continue
items.append(_to_transaction_item(transaction, severity=severity))
return tuple(items)
[docs]
def transactions_low_confidence(
*,
ports: DraftReviewPorts,
threshold: Decimal,
catalogue: TransactionCatalogue | None = None,
) -> tuple[TransactionReviewItem, ...]:
"""Return transactions whose decision confidence sits below a threshold.
Args:
ports: Composed application capabilities for the profile to inspect.
threshold: Minimum acceptable confidence; transactions strictly below
this value are included.
catalogue: Optional :class:`TransactionCatalogue` override; when ``None``
the catalogue is loaded from the composed repository.
Surfaces every transaction whose ``classification_confidence`` is
non-None and strictly less than the threshold, regardless of
classification state. Transactions with ``None`` confidence are excluded
because they have no claim to filter against.
Each element in the returned tuple is a :class:`TransactionReviewItem`.
"""
if catalogue is None:
catalogue = load_transactions(ports=ports)
if catalogue is None:
return ()
items: list[TransactionReviewItem] = []
for transaction in catalogue.values():
confidence = transaction.classification_confidence
if confidence is None or confidence >= threshold:
continue
items.append(_to_transaction_item(transaction, severity=ReviewSeverity.NORMAL))
return tuple(items)
def _classify_transaction(state: BusinessClassification) -> ReviewSeverity | None:
"""First-match-wins severity per the BusinessClassification states.
Returns ``None`` when the state has a final disposition that does
not warrant the operator's attention (classified, rule-excluded, or
reviewed-and-excluded).
"""
if is_classified(state):
return None
if state is BusinessClassification.SKIPPED_BY_RULE:
return None
if state is BusinessClassification.REVIEWED_EXCLUDED:
return None
if state is BusinessClassification.FAILED_VALIDATION:
return ReviewSeverity.CRITICAL
if state is BusinessClassification.PROCESSED_UNCLASSIFIED:
return ReviewSeverity.HIGH
if state is BusinessClassification.NOT_YET_PROCESSED:
return ReviewSeverity.NORMAL
return ReviewSeverity.NORMAL
[docs]
def load_transactions(*, ports: DraftReviewPorts) -> TransactionCatalogue | None:
"""Load the transaction catalogue through the composed application port."""
repository = ports.transaction_repository
try:
if not repository.exists():
_LOGGER.debug("transactions catalogue secure object absent")
return None
return repository.load()
except (CadrumoError, ValidationError, OSError, ValueError) as exc:
raise ReviewSourceLoadError(
message="failed to load transactions catalogue from secure backend",
translated_message="review.adapters.errors.transactions_load_failed",
context=_load_failure_context(exc),
) from exc
def _to_transaction_item(
transaction: Transaction,
*,
severity: ReviewSeverity,
) -> TransactionReviewItem:
raw = transaction.raw
effective_date = raw.value_date or raw.booked_date
since = transaction.classified_at or datetime.combine(effective_date, time.min, tzinfo=UTC)
description = raw.description.strip()
if len(description) > _SUMMARY_MAX:
description = description[: _SUMMARY_MAX - 1] + "…"
del raw # description + amount captured above; nothing else needed
summary = tr("review.transaction.summary")
return TransactionReviewItem(
item_id=transaction.transaction_id,
modelo=None,
severity=severity,
summary=summary,
drill_command=f"aeat app ledger review {transaction.transaction_id}",
since=since,
source=transaction,
)
# ── invoices ──────────────────────────────────────────────────────
[docs]
def invoices_pending(
*,
ports: DraftReviewPorts,
catalogue: InvoiceCatalogue | None = None,
) -> tuple[InvoiceReviewItem, ...]:
"""Return :class:`InvoiceReviewItem` records for unmatched / disputed / pending invoices.
Args:
ports: Composed application capabilities for the profile to inspect.
catalogue: Optional :class:`InvoiceCatalogue` override; the repository
is loaded when ``None``.
"""
if catalogue is None:
catalogue = load_invoices(ports=ports)
if catalogue is None:
return ()
items: list[InvoiceReviewItem] = []
for invoice in catalogue.values():
result = _classify_invoice(invoice)
if result is None:
continue
severity, reason = result
items.append(_to_invoice_item(invoice, severity=severity, reason=reason))
return tuple(items)
[docs]
def load_invoices(*, ports: DraftReviewPorts) -> InvoiceCatalogue | None:
"""Load the invoice catalogue through the composed application port."""
repository = ports.invoice_repository
try:
if not repository.exists():
_LOGGER.debug("invoices catalogue secure object absent")
return None
return repository.load()
except (CadrumoError, ValidationError, OSError, ValueError) as exc:
raise ReviewSourceLoadError(
message="failed to load invoices catalogue from secure backend",
translated_message="review.adapters.errors.invoices_load_failed",
context=_load_failure_context(exc),
) from exc
def _classify_invoice(invoice: Invoice) -> tuple[ReviewSeverity, str] | None:
"""First-match-wins severity + reason for invoices."""
if invoice.linked_transaction_ids == ():
return ReviewSeverity.HIGH, "unmatched"
if invoice.payment_status is PaymentStatus.OVERDUE:
return ReviewSeverity.HIGH, "overdue"
if invoice.payment_status is PaymentStatus.PENDING:
return ReviewSeverity.NORMAL, "payment-pending"
if invoice.payment_status is PaymentStatus.PARTIALLY_PAID:
return ReviewSeverity.NORMAL, "partially-paid"
return None
def _to_invoice_item(invoice: Invoice, *, severity: ReviewSeverity, reason: str) -> InvoiceReviewItem:
del reason # severity already encodes the disposition for the queue line
summary = tr("review.invoice.summary")
since = datetime.combine(invoice.issued_at, time.min, tzinfo=UTC)
return InvoiceReviewItem(
item_id=invoice.invoice_id,
modelo=None,
severity=severity,
summary=summary,
drill_command=f"aeat app review view {invoice.invoice_id}",
since=since,
source=invoice,
)
# ── filing drafts ─────────────────────────────────────────────────
[docs]
def drafts_pending(
settings: Settings,
*,
bucket_id: str,
ports: DraftReviewPorts,
operation: PinnedAuthorityOperation,
drafts: tuple[tuple[Path, ModeloDraft], ...] | None = None,
) -> tuple[FindingReviewItem, ...]:
"""Return :class:`FindingReviewItem` records for findings + unready drafts.
Args:
settings: Active application settings.
bucket_id: Stable bucket identifier for the draft repository to inspect.
ports: Required encrypted persistence capabilities.
operation: Caller-owned generation-pinned indexed authority operation.
drafts: Optional pre-loaded sequence of ``(path, draft)`` pairs where
each draft is a :class:`ModeloDraft`; when ``None`` drafts are
loaded from that bucket's secure storage.
A draft whose ``profile_tax_id`` does not match the active
profile's tax id is not the active profile's data and is skipped.
Callers see only drafts owned by the active profile.
"""
if drafts is None:
drafts = load_drafts(ports=ports, operation=operation)
active_tax_id = _resolve_review_active_tax_id(ports.profile_repository, bucket_id=bucket_id)
if active_tax_id is None:
return ()
items: list[FindingReviewItem] = []
seen: set[tuple[str, str, str]] = set()
for path, stored in drafts:
stored = require_modelo_draft_coordinates_current(stored, operation=operation)
if (stored.profile_tax_id or "") != active_tax_id:
continue
draft, stale_reasons = reviewed_against_current_state(
stored,
bucket_id=bucket_id,
ports=ports,
operation=operation,
)
path_str = str(path)
if draft.findings:
items.extend(_draft_finding_review_items(draft, path_str=path_str, seen=seen))
else:
_append_unready_draft_review_item(
draft,
path_str=path_str,
items=items,
stale_reasons=stale_reasons,
)
return tuple(items)
[docs]
def reviewed_against_current_state(
draft: ModeloDraft,
*,
bucket_id: str,
ports: DraftReviewPorts,
operation: PinnedAuthorityOperation,
) -> tuple[ModeloDraft, tuple[ModeloApprovalStaleReason, ...]]:
"""Return ``draft`` with an aged-out approval reported as aged out.
An approval is a claim about the inputs it was granted over, and the review
queue is the surface whose job is to notice when that claim has stopped
holding. Reading only the stored status cannot notice: an approved draft
stays approved on disk however far the ledger, invoices, taxpayer profile
or registry schema have moved underneath it, so the
``APROBACION_CADUCADA`` row this adapter already knows how to emit was
unreachable and the queue reported nothing to do.
Only ``APROBADO`` drafts are recomputed. Every other status either has no
approval basis to compare or is already downstream of approval, and the
recomputation loads catalogues and a registry snapshot, which is not work
to do for a draft that cannot be stale.
The refreshed draft is NOT written back. This module's adapters are pure
readers, and the verdict is derived state: freezing it into the store on a
read would make the stored status depend on when someone happened to open
the review queue.
Core types:
:class:`~cadrumo.domain.filing.schema.ModeloDraft`.
"""
if draft.status is not ModeloDraftStatus.APROBADO:
return (draft, ())
from ..filing.draft_review import approval_stale_reasons, refresh_review_status
from ..filing.runtime import build_runtime_schema_provider
schema_provider = build_runtime_schema_provider(
filing_year=draft.period.filing_year,
period=draft.period,
modelos=(draft.modelo,),
operation=operation,
)
refreshed = refresh_review_status(
draft,
bucket_id=bucket_id,
schema_provider=schema_provider,
ports=ports,
operation=operation,
)
if refreshed.status is not ModeloDraftStatus.APROBACION_CADUCADA:
return (refreshed, ())
# Recomputed rather than returned by the refresh, which reports the
# transition and logs the reasons without carrying them. The second pass is
# paid only by a draft that HAS aged out, which is the rare case, and the
# alternative is a row that tells an operator something is wrong without
# telling them what moved.
return (
refreshed,
approval_stale_reasons(
draft,
bucket_id=bucket_id,
schema_provider=schema_provider,
ports=ports,
operation=operation,
),
)
def _draft_finding_review_items(
draft: ModeloDraft,
*,
path_str: str,
seen: set[tuple[str, str, str]],
) -> tuple[FindingReviewItem, ...]:
"""Yield one ``FindingReviewItem`` per non-duplicate finding on ``draft``.
Dedup is keyed on ``(draft_id, finding.code, finding.casilla_id)``
so two findings against the same casilla under the same code
surface as a single review row. The ``seen`` set is mutated in
place so dedup spans every draft in the same ``drafts_pending``
pass, not just one draft.
"""
out: list[FindingReviewItem] = []
for finding in draft.findings:
dedup_key = (draft.draft_id, finding.code, finding.casilla_id or "-")
if dedup_key in seen:
continue
seen.add(dedup_key)
out.append(_to_finding_item(draft=draft, path_str=path_str, finding=finding))
return tuple(out)
def _append_unready_draft_review_item(
draft: ModeloDraft,
*,
path_str: str,
items: list[FindingReviewItem],
stale_reasons: tuple[ModeloApprovalStaleReason, ...] = (),
) -> None:
"""Append one review item for a finding-free draft that is not yet ready to file.
DRAFT / VALIDATED drafts get a placeholder review row prompting
the operator to complete the draft. APPROVAL_STALE drafts get a
distinct review row prompting re-approval. Any other status is
a no-op — those drafts are not in the review queue's purview.
"""
if draft.status in {ModeloDraftStatus.BORRADOR, ModeloDraftStatus.VALIDADO}:
items.append(_to_placeholder_item(draft=draft, path_str=path_str))
elif draft.status is ModeloDraftStatus.APROBACION_CADUCADA:
items.append(_to_stale_approval_item(draft=draft, path_str=path_str, stale_reasons=stale_reasons))
def _resolve_review_active_tax_id(
repository: DraftReviewProfileRepositoryProtocol,
*,
bucket_id: str,
) -> str | None:
"""Return the reviewed bucket's tax id through the composed profile port."""
try:
values = repository.load_path_values(bucket_id=bucket_id)
except (CadrumoError, OSError, ValueError):
_LOGGER.debug("review adapters could not resolve profile tax id", exc_info=True)
return None
tax_id = (values or {}).get("identity.tax_id")
return tax_id if isinstance(tax_id, str) and tax_id else None
[docs]
def load_drafts(
*,
ports: DraftReviewPorts,
operation: PinnedAuthorityOperation,
) -> tuple[tuple[Path, ModeloDraft], ...]:
"""Iterate every persisted draft via the composed application capability.
Drafts are ciphertext-at-rest only. The helper returns the secure
backend's logical path marker alongside the typed payload so callers
can identify the draft without consulting a plaintext draft
directory.
"""
repository = ports.draft_repository
out: list[tuple[Path, ModeloDraft]] = []
try:
for draft in repository.iter_drafts():
draft = require_modelo_draft_coordinates_current(draft, operation=operation)
out.append((repository.envelope_path_for(draft.draft_id), draft))
except (CadrumoError, ValidationError, OSError, ValueError) as exc:
raise ReviewSourceLoadError(
message="failed to load filing drafts from secure backend",
translated_message="review.adapters.errors.drafts_load_failed",
context=_load_failure_context(exc),
) from exc
return tuple(out)
def _load_failure_context(exc: BaseException) -> dict[str, str]:
"""Return non-sensitive load-failure context for operator error envelopes."""
return {"error_type": type(exc).__name__}
def _classify_finding(severity: BaseSeverity) -> ReviewSeverity:
"""Map BaseSeverity to ReviewSeverity for findings."""
if severity is BaseSeverity.ERROR:
return ReviewSeverity.CRITICAL
if severity is BaseSeverity.WARNING:
return ReviewSeverity.HIGH
return ReviewSeverity.INFO
def _to_finding_item(
*,
draft: ModeloDraft,
path_str: str,
finding: ModeloValidationFinding,
) -> FindingReviewItem:
casilla = finding.casilla_id or "-"
summary = tr("review.filing.finding_summary")
severity = _classify_finding(finding.severity)
return FindingReviewItem(
item_id=f"{draft.draft_id}:{finding.code}:{casilla}",
modelo=draft.modelo,
severity=severity,
summary=summary,
drill_command=f"aeat app review view {draft.draft_id}:{finding.code}:{casilla}",
since=draft.updated_at,
source=finding,
draft_id=draft.draft_id,
draft_path=path_str,
)
def _to_placeholder_item(*, draft: ModeloDraft, path_str: str) -> FindingReviewItem:
summary = tr("review.filing.draft_placeholder_summary")
return FindingReviewItem(
item_id=f"{draft.draft_id}:_status:{draft.status.value}",
modelo=draft.modelo,
severity=ReviewSeverity.NORMAL,
summary=summary,
drill_command=f"aeat app review view {draft.draft_id}:_status:{draft.status.value}",
since=draft.updated_at,
source=None,
draft_id=draft.draft_id,
draft_path=path_str,
)
def _to_stale_approval_item(
*,
draft: ModeloDraft,
path_str: str,
stale_reasons: tuple[ModeloApprovalStaleReason, ...] = (),
) -> FindingReviewItem:
"""Emit a high-severity item for drafts whose stored approval is stale.
The reasons ride as stable tokens. "This approval is stale" and "this
approval is stale because the transaction catalogue changed" ask the
operator for different next actions, and only the projection that renders
a row may turn the second into words.
"""
summary = tr("review.filing.stale_approval_summary")
return FindingReviewItem(
item_id=f"{draft.draft_id}:_status:APPROVAL_STALE",
modelo=draft.modelo,
severity=ReviewSeverity.HIGH,
summary=summary,
drill_command=f"aeat app review view {draft.draft_id}:_status:APPROVAL_STALE",
since=draft.updated_at,
source=None,
draft_id=draft.draft_id,
draft_path=path_str,
stale_reasons=stale_reasons,
)