cadrumo.application.overview.calendar_evidence module

Calendar event dedup and filing-evidence reconciliation for the overview read model.

Extracted from calendar (size-budget split) to keep the calendar module and its build_overview_calendar entry point under their line-count overrides. This module owns two cohesive concerns:

  • Deterministic dedup of observed OverviewCalendarEvent rows across multiple local snapshots (_calendar_event_sort_key, _dedupe_calendar_events), shared by every calendar-events builder.

  • Reconciling already-loaded local and AEAT-side signals – filed Modelo records, observed register events, filed-declaration observations, persisted calculation observations, live justificante captures, and loaded justificante metadata – into typed OverviewCalendarFilingEvidence rows (calendar_filing_evidence_from_sources()), and merging that evidence back onto calendar entries and events.

This module is local-only and pure with respect to I/O, exactly like its parent: it never starts a live AEAT read and never verifies a justificante by fetching external state; it only reconciles evidence the caller has already loaded.

See also

cadrumo.application.overview.calendar

Composes calendar_filing_evidence_from_sources() and the other builders here into build_overview_calendar().

cadrumo.application.overview.calendar

Composes this evidence projection into the overview calendar.

NO_AEAT_HISTORY_NOTICE_CODE

Notice code for a workable profile carrying no official AEAT observation.

no_aeat_history_notice(observation_source_kinds, *, tax_route=None)[source]

Point a workable profile with no AEAT-sourced history at the history pull.

Fires only when NOT ONE persisted calculation observation carries an official AEAT source kind. A profile with even one is not a fresh profile any more, so the nudge would be noise; a profile with none has nothing AEAT-sourced to reconcile against, and until this shipped nothing told the operator that a single verb would fetch it.

Keyed on the official-source predicate rather than on an empty observation list, deliberately. A profile whose only observations are locally filed or operator-entered has the same gap as one with no observations at all – it holds nothing AEAT ever confirmed – and testing for emptiness would leave exactly that taxpayer unprompted.

Returns None when any official observation exists, so an onboarded profile stays quiet.

Takes the source kinds, not the observations, and the narrowing is the point. Reading source_kind off an untyped payload classified EVIDENCE AUTHORITY by attribute NAME, and that name is not unique to this axis: the aggregation observation envelopes carry a field spelled identically that means CAPTURE PROVENANCE – which ingestion path wrote a row – and answers to a different closed set. Nothing in a duck-typed read distinguishes them, so widening this function’s input was a one-line change away from grading capture provenance as AEAT confirmation. Asking for the enum makes the two axes non-interchangeable at the call site, where the caller holds the payload and knows which one it has.

Parameters:

  • observation_source_kinds (tuple[ObservationSourceKind, ...]) – Every persisted calculation observation’s source kind, across every modelo.

  • tax_route (TaxRoute | None) – The active profile’s derive_tax_route() branch, when known. A Sociedades filer (IMPUESTO_SOCIEDADES) has no whole-history sweep to recommend: the bulk filed-data capture planner structurally diverts Modelo 200 and 202 – the Sociedades filer’s own direct-tax obligations – into typed unsupported rows, because this deployment’s registry declares no authenticated filed-declarations read surface for them. Pointing such a taxpayer at the sweep verb recommends a verb that can never fetch what the notice is about, so this route carries no action and a route-specific message instead. None (the default) keeps the whole-history-sweep suggestion for every other route.

Return type:

Notice | None

calendar_filing_evidence_from_sources(*, filing_records=(), observed_events=(), filed_declaration_observations=(), verified_filed_declaration_artefact_refs=(), verified_filed_declaration_artefact_csvs=None, calculation_observations=(), justificante_capture_snapshots=(), justificantes=(), expected_tax_id=None)[source]

Build OverviewCalendarFilingEvidence from local records and observed AEAT signals.

The function is pure and intentionally accepts already-loaded records. CLI/storage code owns I/O; this projection only reconciles the existing local ModeloRecord catalogue, calendar-visible AEAT register events, FiledDeclaracionObservation rows, persisted calculation observations from justificante capture, JustificanteCaptureSnapshot rows, and already-loaded justificante metadata. The filing_records are filed ModeloRecord rows whose external justificante references are only promoted to justificante_verified when matching persisted metadata exists for the same CSV/model/year/period/taxpayer. Filed-declaration observations are only promoted to justificante_verified when their justificante artefact storage reference is listed in verified_filed_declaration_artefact_refs by the storage layer that loaded and hashed the encrypted artefact body.

The result keeps OverviewLocalFilingState and OverviewAeatSubmissionState independent so imported AEAT baselines, local filings, observed submissions, and verified justificantes do not overwrite each other’s meaning.

Return type:

tuple[OverviewCalendarFilingEvidence, ...]

Parameters:
merge_calendar_filing_evidence(*groups)[source]

Join reconciled evidence groups by natural address deterministically.

Return type:

tuple[OverviewCalendarFilingEvidence, ...]

Parameters:

groups (tuple[OverviewCalendarFilingEvidence, ...])

calendar_entry_filing_evidence(*, modelo, filing_year, period, evidence)
Return type:

OverviewCalendarFilingEvidence

Parameters:
calendar_events_with_filing_evidence(events, evidence)
Return type:

tuple[OverviewCalendarEvent, ...]

Parameters:
dedupe_calendar_events(events)

Deduplicate repeated observations across multiple local snapshots.

Return type:

tuple[OverviewCalendarEvent, ...]

Parameters:

events (list[OverviewCalendarEvent])