"""Capture filed-declaration evidence after the register reader selects a row.
Core types:
:class:`~cadrumo.domain.calculations.registry.schema.RegistrySnapshot`.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
from .....core.casilla_id import CasillaId
from .....core.config import Settings
from .....core.observed_header_fact import ObservedHeaderFact
from .....core.period import Period
from .....domain.calculations.registry.bindings_previous_filing import previous_filing_observation_requirements
from .....domain.calculations.registry.errors import RegistryValidationError
from .....domain.calculations.registry.relations import relation_source_requirements, source_presence_gaps
from .....domain.calculations.registry.remote_state_guard import RemoteStateGuardPolicy
from .....domain.calculations.registry.schema import RegistrySnapshot
from .._playwright import BrowserContext, Locator, Page, Playwright
from ._declarations_fetch import (
capture_row_pdf_artefact,
capture_submitted_file_artefact,
listing_url_for,
origin_of,
)
from .declarations_observations import (
FiledDeclaracionArtefactSink,
_declaration_pdf_extraction_profile_provisional,
_observed_casillas_from_declaration_pdf,
_read_guard_policy_from_snapshot,
_register_row_artefact,
_registry_snapshot_for_declaration,
_store_artefact,
_submitted_file_coverage_for_casillas,
_submitted_file_layout_refusal,
_with_derived_303_compensation_available_observation,
observed_casillas_from_submitted_file,
observed_header_facts_from_submitted_file,
)
from .declarations_schema import Declaracion
from .errors import JustificanteFetchError, SedeNavigationError, SedeParseError
from .schema import FiledDeclaracionArtefact, FiledDeclaracionObservation, ObservedCasillaValue
if TYPE_CHECKING:
from .....application.auth.session_types import AeatSession
from .....domain.calculations.registry.authority import PinnedAuthorityOperation
from .....domain.calculations.registry.schema import ModeloRevision
def _record_submitted_file_extraction_error(
metadata: dict[str, str],
error: RegistryValidationError | SedeParseError,
) -> None:
"""Persist the adapter's own submitted-file parser refusal verbatim."""
metadata["submitted_file_extraction_error"] = str(error)
async def _capture_declaration_copy_if_available(
*,
context: BrowserContext,
row_locator: Locator,
declaration: Declaracion,
read_policy: RemoteStateGuardPolicy,
observation_key: tuple[str, int, Period, str],
artefact_sink: FiledDeclaracionArtefactSink | None,
) -> tuple[tuple[FiledDeclaracionArtefact, ...], bytes | None]:
"""Capture the optional declaration-copy PDF, retaining artefact order."""
if not declaration.declaration_copy_link_text or declaration.declaration_copy_cell_index is None:
return (), None
declaration_pdf, declaration_pdf_body = await capture_row_pdf_artefact(
context=context,
row_locator=row_locator,
declaration=declaration,
cell_index=declaration.declaration_copy_cell_index,
kind="declaration_pdf",
read_policy=read_policy,
)
stored_declaration_pdf = _store_artefact(
artefact_sink,
observation_key=observation_key,
artefact=declaration_pdf,
body=declaration_pdf_body,
)
return (stored_declaration_pdf,), declaration_pdf_body
async def _capture_submitted_file_if_available(
*,
context: BrowserContext,
page: Page,
row_locator: Locator,
declaration: Declaracion,
read_policy: RemoteStateGuardPolicy,
snapshot: RegistrySnapshot,
operation: PinnedAuthorityOperation,
observation_key: tuple[str, int, Period, str],
artefact_sink: FiledDeclaracionArtefactSink | None,
metadata: dict[str, str],
) -> tuple[
tuple[FiledDeclaracionArtefact, ...],
tuple[ObservedCasillaValue, ...],
tuple[ObservedHeaderFact, ...],
dict[str, float],
]:
"""Capture and interpret the optional submitted file, recording soft refusals."""
artefacts: tuple[FiledDeclaracionArtefact, ...] = ()
casillas: tuple[ObservedCasillaValue, ...] = ()
headers: tuple[ObservedHeaderFact, ...] = ()
extraction_coverage: dict[str, float] = {}
if declaration.archive_link_text and declaration.archive_cell_index is not None:
try:
submitted_artefact, submitted_body = await capture_submitted_file_artefact(
context=context,
page=page,
row_locator=row_locator,
declaration=declaration,
cell_index=declaration.archive_cell_index,
read_policy=read_policy,
)
except (JustificanteFetchError, SedeNavigationError) as exc:
metadata["submitted_file_capture_error"] = str(exc)
else:
submitted_artefact = _store_artefact(
artefact_sink,
observation_key=observation_key,
artefact=submitted_artefact,
body=submitted_body,
)
artefacts = (submitted_artefact,)
try:
casillas = observed_casillas_from_submitted_file(
snapshot=snapshot,
declaration=declaration,
body=submitted_body,
artefact=submitted_artefact,
operation=operation,
)
try:
extraction_coverage["submitted_file"] = _submitted_file_coverage_for_casillas(
snapshot=snapshot,
body=submitted_body,
casillas=casillas,
operation=operation,
)
except RegistryValidationError as exc:
raise _submitted_file_layout_refusal(
snapshot=snapshot,
declaration=declaration,
artefact=submitted_artefact,
reason=str(exc),
) from exc
headers = observed_header_facts_from_submitted_file(
snapshot=snapshot,
body=submitted_body,
operation=operation,
)
except (RegistryValidationError, SedeParseError) as exc:
_record_submitted_file_extraction_error(metadata, exc)
return artefacts, casillas, headers, extraction_coverage
def _resolve_captured_casillas(
*,
casillas: tuple[ObservedCasillaValue, ...],
declaration_pdf_body: bytes | None,
declaration: Declaracion,
snapshot: RegistrySnapshot,
metadata: dict[str, str],
extraction_coverage: dict[str, float],
) -> tuple[tuple[ObservedCasillaValue, ...], dict[str, float]]:
"""Choose submitted-file observations, PDF fallback, or refuse no source."""
if not casillas and declaration_pdf_body is not None:
casillas = _observed_casillas_from_declaration_pdf(
snapshot=snapshot,
declaration=declaration,
body=declaration_pdf_body,
)
extraction_coverage["declaration_pdf"] = 1.0
if _declaration_pdf_extraction_profile_provisional(snapshot):
metadata["declaration_pdf_extraction_profile_provisional"] = "true"
elif not casillas and not declaration.archive_link_text and declaration_pdf_body is None:
raise SedeParseError(
f"AEAT declaration {declaration.expediente_id!r} did not expose submitted-file or declaration-copy data",
)
return casillas, extraction_coverage
[docs]
async def capture_filed_declaration_observation_from_row(
session: AeatSession,
declaration: Declaracion,
*,
row_locator: Locator,
page: Page,
context: BrowserContext,
registry_snapshot: RegistrySnapshot | None,
operation: PinnedAuthorityOperation,
artefact_sink: FiledDeclaracionArtefactSink | None,
) -> FiledDeclaracionObservation:
authenticated_identity = (session.identity_nif or "").strip()
snapshot = registry_snapshot or _registry_snapshot_for_declaration(declaration, operation=operation)
read_policy = _read_guard_policy_from_snapshot(snapshot)
filing_period = declaration.period
observation_key = (declaration.modelo, declaration.ejercicio, filing_period, declaration.expediente_id)
from pydantic import AnyHttpUrl
listing_url = AnyHttpUrl(
listing_url_for(
origin_of(getattr(page, "url", None)),
modelo=declaration.modelo,
ejercicio=declaration.ejercicio,
),
)
register_row, register_row_body = _register_row_artefact(declaration, source_url=listing_url)
artefacts: list[FiledDeclaracionArtefact] = [
_store_artefact(artefact_sink, observation_key=observation_key, artefact=register_row, body=register_row_body),
]
casillas: tuple[ObservedCasillaValue, ...] = ()
headers: tuple[ObservedHeaderFact, ...] = ()
extraction_coverage: dict[str, float] = {}
metadata = {"tipo_solicitud": declaration.tipo_solicitud or "", "observaciones": declaration.observaciones or ""}
justificante, justificante_body = await capture_row_pdf_artefact(
context=context,
row_locator=row_locator,
declaration=declaration,
cell_index=declaration.justificante_cell_index,
kind="justificante_pdf",
read_policy=read_policy,
)
artefacts.append(
_store_artefact(artefact_sink, observation_key=observation_key, artefact=justificante, body=justificante_body),
)
declaration_copy_artefacts, declaration_pdf_body = await _capture_declaration_copy_if_available(
context=context,
row_locator=row_locator,
declaration=declaration,
read_policy=read_policy,
observation_key=observation_key,
artefact_sink=artefact_sink,
)
artefacts.extend(declaration_copy_artefacts)
submitted_artefacts, casillas, headers, extraction_coverage = await _capture_submitted_file_if_available(
context=context,
page=page,
row_locator=row_locator,
declaration=declaration,
read_policy=read_policy,
snapshot=snapshot,
operation=operation,
observation_key=observation_key,
artefact_sink=artefact_sink,
metadata=metadata,
)
artefacts.extend(submitted_artefacts)
casillas, extraction_coverage = _resolve_captured_casillas(
casillas=casillas,
declaration_pdf_body=declaration_pdf_body,
declaration=declaration,
snapshot=snapshot,
metadata=metadata,
extraction_coverage=extraction_coverage,
)
return FiledDeclaracionObservation(
modelo=declaration.modelo,
ejercicio=declaration.ejercicio,
period=filing_period,
expediente_id=declaration.expediente_id,
status=declaration.estado,
presented_at=declaration.presented_at,
authenticated_identity=authenticated_identity,
artefacts=tuple(artefacts),
casillas=casillas,
headers=headers,
metadata=metadata,
extraction_coverage=extraction_coverage,
registry_snapshot_ref=snapshot.snapshot_ref,
)
[docs]
async def capture_previous_filing_observations(
session: AeatSession,
revision: ModeloRevision,
*,
filing_year: int,
period: Period,
operation: PinnedAuthorityOperation,
settings: Settings | None = None,
playwright: Playwright | None = None,
artefact_sink: FiledDeclaracionArtefactSink | None = None,
) -> tuple[FiledDeclaracionObservation, ...]:
"""Capture observations required by registry previous-filing bindings.
The binding requirements choose the source modelo, filing year, and period;
this function selects the authoritative filed row and refuses incomplete
observed-source coverage before returning its evidence.
Core types:
:class:`~cadrumo.domain.calculations.registry.schema.ModeloRevision`.
"""
from .declarations import open_declarations_register
observations: list[FiledDeclaracionObservation] = []
async with open_declarations_register(
session,
operation=operation,
settings=settings,
playwright=playwright,
) as register:
for requirement in previous_filing_observation_requirements(
revision,
filing_year=filing_year,
period=period.registry_token,
):
source_period = requirement.periods[0]
rows = await register.walk(modelo=requirement.source_modelo, ejercicio=requirement.filing_year)
declaration = _select_authoritative_declaration(
tuple(row for row in rows if row.period.registry_token == source_period),
modelo=requirement.source_modelo,
ejercicio=requirement.filing_year,
period_token=source_period,
context="previous-filing requirement",
)
observation = _with_derived_303_compensation_available_observation(
await register.capture_observation(declaration, artefact_sink=artefact_sink),
operation=operation,
)
observed_casillas: set[CasillaId] = {casilla.casilla_id for casilla in observation.casillas}
missing, missing_presence_groups = source_presence_gaps(
required_source_casilla_ids=requirement.enforced_source_casilla_ids,
source_presence_groups=requirement.source_presence_groups,
observed_source_casilla_ids=observed_casillas,
)
if missing:
raise SedeParseError(
f"previous-filing requirement {requirement.source_modelo!r}/{requirement.filing_year}/"
f"{source_period!r} missing observed casillas {missing!r}",
)
if missing_presence_groups:
raise SedeParseError(
f"previous-filing requirement {requirement.source_modelo!r}/{requirement.filing_year}/"
f"{source_period!r} is missing required source-presence groups {list(missing_presence_groups)!r}",
)
observations.append(observation)
return tuple(observations)
[docs]
async def capture_relation_source_observations(
session: AeatSession,
revision: ModeloRevision,
*,
filing_year: int,
period: Period,
operation: PinnedAuthorityOperation,
settings: Settings | None = None,
playwright: Playwright | None = None,
artefact_sink: FiledDeclaracionArtefactSink | None = None,
) -> tuple[FiledDeclaracionObservation, ...]:
"""Capture observations required by registry cross-model relations.
A relation may require several source periods. Every selected filed row is
validated for the casillas the relation declares before it is returned.
Core types:
:class:`~cadrumo.domain.calculations.registry.schema.ModeloRevision`.
"""
from .declarations import open_declarations_register
required_source_casilla_ids: dict[tuple[str, int, str], set[CasillaId]] = {}
for requirement in relation_source_requirements(revision, filing_year=filing_year, period=period.registry_token):
for source_period in requirement.periods:
key = (requirement.source_modelo, requirement.filing_year, source_period)
required_source_casilla_ids.setdefault(key, set()).update(requirement.source_casilla_ids)
observations: list[FiledDeclaracionObservation] = []
async with open_declarations_register(
session,
operation=operation,
settings=settings,
playwright=playwright,
) as register:
for (modelo, source_year, source_period), source_casilla_ids in sorted(required_source_casilla_ids.items()):
rows = await register.walk(modelo=modelo, ejercicio=source_year)
declaration = _select_authoritative_declaration(
tuple(row for row in rows if row.period.registry_token == source_period),
modelo=modelo,
ejercicio=source_year,
period_token=source_period,
context="relation source requirement",
)
observation = _with_derived_303_compensation_available_observation(
await register.capture_observation(declaration, artefact_sink=artefact_sink),
operation=operation,
)
observed_casillas: set[CasillaId] = {casilla.casilla_id for casilla in observation.casillas}
missing = sorted(source_casilla_ids.difference(observed_casillas))
if missing:
raise SedeParseError(
f"relation source requirement {modelo!r}/{source_year}/{source_period!r} "
f"missing observed casillas {missing!r}",
)
observations.append(observation)
return tuple(observations)
def _select_authoritative_declaration(
declarations: tuple[Declaracion, ...],
*,
modelo: str,
ejercicio: int,
period_token: str,
context: str,
) -> Declaracion:
"""Select the latest accepted register row for one filed period."""
if not declarations:
raise SedeParseError(f"{context} {modelo!r}/{ejercicio}/{period_token!r} found no filed declaration")
active = tuple(row for row in declarations if row.estado.upper() == "ALTA")
return max(active or declarations, key=lambda row: (row.presented_at, row.expediente_id))
__all__ = [
"capture_previous_filing_observations",
"capture_relation_source_observations",
]