"""Read-only walker over AEAT's *Consultar declaraciones presentadas* surface.
The expediente-tree walker (`_walker.walk_expedientes_tree`) walks
``Mis Expedientes`` (``/wlpl/TEWV-CORE/ResumenVlt``) which is a
*procedures* surface — sanciones, recursos, gestión recaudación —
not the canonical filings register. To read the operator's actual
filing history across supported periodic, annual, retention, and
informative declarations, we drive the
``Consultar declaraciones presentadas`` form at
``/wlpl/SCEJ-MANT/CONSUL/index.zul``.
The form is built on the ZK framework. Input ids are generated by the
server and are not stable selectors; automation binds on label text
and visible row structure. The submit action is a real ``Buscar``
button issuing a ZK AJAX RPC; URL parameters alone do not drive
results.
Parsed filings are grounded against the registry: each declaration is
resolved to its :class:`ModeloRevision` and a :class:`RegistrySnapshot`
(loaded through :class:`ValidatedRegistryAuthority`), and reported casilla
values are surfaced as :class:`CasillaObservation` rows carrying their
legal and source references.
"""
from __future__ import annotations
import re
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from typing import TYPE_CHECKING, Final
from .....core.config import Settings
from .....core.filing_year import FILING_YEAR_MAX, FILING_YEAR_MIN
from .....core.i18n.render import tr
from .....core.logging import get_logger
from .....core.time.clock import now
from .....domain.calculations.registry.remote_state_guard import RemoteStateGuardPolicy
# Importing the renta package registers the first-slice routing
# cross-domain snapshot check with the registry validator. build_snapshot
# of a Modelo 100 revision fails loudly if that check is unregistered, so
# the M100 routing referential-integrity gate runs on this declarations path.
from .....domain.calculations.registry.schema import RegistrySnapshot
from .._html import parse_html
from .._playwright import BrowserContext, Locator, Page, Playwright, PlaywrightError
from ..browser.factory import opened_browser_page, shared_playwright_runtime
from ..browser.profile import Profile
from ._adapter_utils import assert_read_landing
from ._auth_state import storage_state_for_session
from ._browser_constants import (
PLAYWRIGHT_WAIT_NETWORKIDLE as _WAIT_NETWORKIDLE,
)
from ._browser_constants import (
navigation_timeout_ms,
)
from ._declarations_diagnostics import (
declarations_page_shape_context as _declarations_page_shape_context,
)
from ._declarations_diagnostics import (
declarations_page_shape_context_from_page as _declarations_page_shape_context_from_page,
)
from ._declarations_fetch import (
READ_GUARD_POLICY,
SEDE_BASE,
assert_declarations_read_browser_action,
assert_declarations_read_http,
get_buscar_settle_ms,
get_form_interaction_timeout_ms,
)
from ._declarations_listbox import _has_class, _parse_listbox, _parse_presented_at
from .declarations_observations import (
FiledDeclaracionArtefactSink,
_read_guard_policy_from_snapshot,
_registry_snapshot_for_declaration,
)
from .declarations_schema import Declaracion
from .errors import SedeFailureMode, SedeNavigationError, SedeParseError
from .schema import (
FiledDeclaracionObservation,
FiledDeclarationAvailability,
FiledDeclarationAvailabilityReport,
)
if TYPE_CHECKING:
from .....application.auth.session_types import AeatSession
from .....domain.calculations.registry.authority import PinnedAuthorityOperation
log = get_logger(__name__)
EXTERNAL = Settings.external_constants()
_LISTING_URL = f"{SEDE_BASE}{EXTERNAL.aeat.sede_paths.declarations_listing}"
_DECLARATIONS_LISTING_PATH_PREFIX = EXTERNAL.aeat.sede_paths.declarations_listing.removesuffix("/index.zul")
# The register application directory. Buscar is a ZK AJAX RPC, so the page is
# expected not to move across the search -- the same prefix the pre-search
# navigation check already requires is therefore the post-search read surface
# too, rather than a new claim about where AEAT may serve results.
_DECLARATIONS_READ_PATH_PREFIXES: tuple[str, ...] = (_DECLARATIONS_LISTING_PATH_PREFIX,)
# The two combobox option shapes, taken from the option_match values the shipped
# selection path already uses against the live form: a modelo option carries
# "<code> -" followed by a description, and an ejercicio option is a bare year.
# Requiring the dash on one and anchoring the other end-to-end keeps them
# disjoint, which matters because AEAT renders BOTH popups into one DOM snapshot
# -- the bundled real capture carries the modelo list and the ejercicio list
# together, along with a "-- Seleccione --" placeholder in each.
#
# The description is deliberately NOT required to be non-empty. The same real
# capture offers "174 -" with a blank description, and a pattern demanding a
# following character drops that modelo out of the offered set entirely -- a
# silent narrowing of the very signal this read exists to widen.
#
# Codes are not digits-only: the capture offers ATF, BSS, DT2, FCV, RIB, VEG and
# 14A alongside the numeric ones.
_MODELO_OPTION_RE: Final[re.Pattern[str]] = re.compile(r"^(?P<modelo>[0-9A-Z]{2,8})\s*-")
_EJERCICIO_OPTION_RE: Final[re.Pattern[str]] = re.compile(r"^(?P<ejercicio>[0-9]{4})$")
[docs]
def assert_declarations_read_landing(
landing_url: str,
*,
policy: RemoteStateGuardPolicy = READ_GUARD_POLICY,
) -> None:
"""Refuse a landing that is not the declaraciones register.
The navigation INTO the register already refused a landing outside
this prefix, which is why the rule is that same prefix rather than a
wider one. What was missing is the other side of the search: Buscar
is a real submit control, and its landing was LOGGED and never
checked, so a search that left the register was recorded and then
scraped. That failure is silent -- the row parser finds no rows, and
an empty listing is indistinguishable from a taxpayer who filed
nothing that ejercicio.
Args:
landing_url: The URL AEAT actually served, read off the page.
policy: The read guard policy; defaults to the module's own.
Raises:
SedeNavigationError: When the landing is not the register.
"""
assert_read_landing(
landing_url,
surface="declaraciones register",
policy=policy,
allowed_path_prefixes=_DECLARATIONS_READ_PATH_PREFIXES,
)
[docs]
@asynccontextmanager
async def shared_playwright(
session: AeatSession,
) -> AsyncGenerator[Playwright]:
"""Yield a long-lived Playwright instance for bulk register sweeps.
Standalone register reads spin up their own Playwright + BrowserSession.
Bulk callers
(e.g. ``aeat sede capture-corpus``) pay ~1s per iteration on
Playwright startup; wrapping repeated reads in this helper amortises
that cost across the entire run.
Usage:
.. code-block:: python
async with shared_playwright(session) as pw:
for modelo in modelos:
rows = await walk_declarations_register(
session, modelo=modelo, ejercicio=ejercicio,
playwright=pw,
)
async with open_declarations_register(
session, operation=operation, playwright=pw,
) as register:
for declaration in rows:
observation = await register.capture_observation(declaration)
Args:
session: Authenticated AEAT session. Validated upfront so
bulk reads fail fast rather than per-iteration.
Yields:
A live :class:`Playwright` instance. Cleaned up automatically
on exit.
"""
storage_state_for_session(session)
async with shared_playwright_runtime() as pw:
yield pw
def _register_rows_from_snapshot(
html: str,
*,
modelo: str,
ejercicio: int,
) -> tuple[Declaracion, ...]:
"""Parse one register snapshot into rows, refusing a page the grid says is partial.
The register is read from a single ``Buscar`` result snapshot, and the walker
does not click through pages. When the grid's own pager label declares more
records than the snapshot rendered, the rows in hand are an undercount of the
operator's filing history — and every downstream ``row_count`` is a plain
``len()`` over them, so returning the short tuple would report a partial
capture with exactly the confidence of a complete one. That is refused
loudly instead: a truncated register read is an error, never a quiet
shortfall.
Raises:
SedeParseError: When the rendered row count falls short of the record
total the grid's pager label declares.
"""
page = _parse_listbox(html, modelo=modelo, ejercicio=ejercicio)
if page.truncated:
raise SedeParseError(
# Kept short and fact-first: the bulk sweep bounds a failure row's
# message, so a long preamble would truncate away the counts.
f"declaraciones register modelo {modelo} ejercicio {ejercicio} rendered "
f"{len(page.rows)} row(s) but its pager declares {page.declared_total} in total; "
"refusing an under-reported filing history",
context={
"modelo": modelo,
"ejercicio": ejercicio,
"rendered_count": len(page.rows),
"declared_total": page.declared_total,
},
)
return page.rows
def _combobox_option_texts(html: str) -> tuple[str, ...]:
"""Return every rendered combobox option text in one register-page snapshot.
ZK keeps combobox popups in the DOM whether or not they are open, so a
snapshot can carry the modelo popup's options and the ejercicio popup's
options at once. The two callers below therefore do NOT rely on "only the
open popup is present"; they discriminate by option SHAPE instead, and the
two shapes are disjoint (see :func:`filed_register_modelo_options` and
:func:`filed_register_ejercicio_options`).
Raises:
SedeParseError: When the snapshot cannot be parsed at all.
"""
try:
soup = parse_html(html)
except Exception as exc:
raise SedeParseError(
f"failed to parse declaraciones combobox HTML: {exc}",
translated_message=tr("adapters.sede.errors.parse_failed"),
) from exc
return tuple(
text
for option in soup.find_all(class_=_has_class("z-comboitem-text"))
for text in (option.get_text(" ", strip=True),)
if text
)
[docs]
def filed_register_modelo_options(html: str) -> tuple[str, ...]:
"""Return the modelo codes the register's modelo combobox offers.
The shape is the one the shipped selection path already matches against the
live form: a modelo option reads ``"<code> - <description>"``, which is why
:func:`_drive_search` selects on ``f"{modelo} -"``. Requiring the dash is
what keeps this disjoint from an ejercicio option, which is a bare year.
Duplicates are collapsed in first-seen order, so the result is what AEAT
offered rather than how many times it rendered it.
Raises:
SedeParseError: When the snapshot carries no modelo-shaped option. An
empty modelo list is not a taxpayer with no history — the register
always offers modelos — so it means the form did not render or AEAT
changed the option shape, and reporting it as "nothing offered"
would turn a shape change into a silently empty history.
"""
codes_list: list[str] = []
for text in _combobox_option_texts(html):
match = _MODELO_OPTION_RE.match(text)
if match is None:
continue
modelo = match.group("modelo")
if isinstance(modelo, str):
codes_list.append(modelo)
codes = tuple(dict.fromkeys(codes_list))
if not codes:
raise SedeParseError(
"declaraciones register rendered no modelo-shaped combobox option; "
"refusing to report an empty offered-modelo set",
translated_message=tr("adapters.sede.errors.parse_failed"),
failure_mode=SedeFailureMode.EXTERNAL_SHAPE_CHANGED,
)
return codes
[docs]
def filed_register_ejercicio_options(html: str) -> tuple[int, ...]:
"""Return the ejercicios the register's ejercicio combobox offers, newest first.
An ejercicio option is a bare four-digit year, which is what
:func:`_drive_search` selects on. Unlike the modelo list, an EMPTY result is
a legitimate answer: the register genuinely offers no ejercicio for a modelo
the authenticated taxpayer has nothing under, and that is one of the two
readings this signal cannot distinguish between. So this returns ``()``
rather than refusing, and the caller must not read the emptiness as proof
that nothing was filed.
"""
years = {
int(match.group("ejercicio"))
for text in _combobox_option_texts(html)
for match in (_EJERCICIO_OPTION_RE.match(text),)
if match is not None
}
return tuple(sorted((year for year in years if FILING_YEAR_MIN <= year <= FILING_YEAR_MAX), reverse=True))
[docs]
async def discover_filed_declaration_availability(
session: AeatSession,
*,
settings: Settings | None = None,
playwright: Playwright | None = None,
) -> FiledDeclarationAvailabilityReport:
"""Read what the declaraciones register OFFERS, without querying any pair.
Opens the register form once, enumerates the modelo combobox's full option
set, then selects each modelo in turn and enumerates the ejercicio
combobox's option set for it. Nothing is searched, downloaded or persisted:
this is a read of the form's own controls.
The result is tagged
:attr:`~core.filed_history_discovery_signal.FiledHistoryDiscoverySignal.AEAT_REGISTER_OPTIONS` and MUST be
consumed as AEAT's *offered* option set. Whether those option lists are
scoped to the authenticated NIF or are a static universal catalogue is
unconfirmed, and settling it needs a live authenticated probe against an
account with real filing history under explicit operator authorisation. So
a caller may UNION this into a walk grid — widening coverage costs nothing —
but may never measure completeness against it alone, and may never report a
pair absent from it as a filing the taxpayer does not have.
A modelo whose ejercicio enumeration fails is skipped with a logged
diagnostic rather than failing the whole read, because this signal is
additive by construction: losing one modelo's options narrows a bonus, while
refusing the read would deny the caller the widening entirely.
Args:
session: Authenticated AEAT session whose ``storage_state_path`` carries
valid cookies.
settings: Optional :class:`Settings` override.
playwright: Optional pre-started Playwright instance, typically from
:func:`shared_playwright`. Caller owns its lifetime.
Returns:
A :class:`FiledDeclarationAvailabilityReport` of the offered pairs.
Raises:
SedeNavigationError: When the register form does not load.
SedeParseError: When the modelo combobox renders no modelo-shaped option.
"""
items: list[FiledDeclarationAvailability] = []
async with _open_register_page(session, settings=settings, playwright=playwright) as (page, _context):
await _open_register_form(page)
await _open_combobox(page, label_text="Modelo (*)")
modelos = filed_register_modelo_options(await page.content())
log.info("discover_filed_declaration_availability: register offers %d modelo option(s)", len(modelos))
for modelo in modelos:
try:
if not await _select_combobox_value(page, label_text="Modelo (*)", option_match=f"{modelo} -"):
log.info(
"discover_filed_declaration_availability: modelo option vanished between reads modelo=%s",
modelo,
)
continue
await _open_combobox(page, label_text="Ejercicio (*)")
ejercicios = filed_register_ejercicio_options(await page.content())
except (SedeNavigationError, SedeParseError) as exc:
log.info(
"discover_filed_declaration_availability: ejercicio enumeration failed modelo=%s error=%s",
modelo,
exc.__class__.__name__,
)
continue
items.append(FiledDeclarationAvailability(modelo=modelo, ejercicios=ejercicios))
return FiledDeclarationAvailabilityReport(items=tuple(items), discovered_at=now())
[docs]
class DeclaracionesRegisterSession:
"""Reusable read-only session for AEAT's filed-declarations register."""
def __init__(
self,
session: AeatSession,
page: Page,
context: BrowserContext,
*,
operation: PinnedAuthorityOperation,
) -> None:
"""Bind the live session, page and browser context this reader drives."""
self.session = session
self._page = page
self._context = context
self._operation = operation
[docs]
async def walk(self, *, modelo: str, ejercicio: int) -> tuple[Declaracion, ...]:
"""Return :class:`Declaracion` rows for one ``(modelo, ejercicio)`` query.
Raises:
SedeParseError: When the grid declares more records than it rendered,
per :func:`_register_rows_from_snapshot`.
"""
if not await _drive_search(self._page, modelo=modelo, ejercicio=ejercicio):
log.info(
"DeclaracionesRegisterSession.walk: ejercicio unavailable modelo=%s ejercicio=%d",
modelo,
ejercicio,
)
return ()
results = _register_rows_from_snapshot(await self._page.content(), modelo=modelo, ejercicio=ejercicio)
log.info(
"DeclaracionesRegisterSession.walk: found %d declaration(s) modelo=%s ejercicio=%d",
len(results),
modelo,
ejercicio,
)
return results
[docs]
async def capture_observation(
self,
declaration: Declaracion,
*,
registry_snapshot: RegistrySnapshot | None = None,
artefact_sink: FiledDeclaracionArtefactSink | None = None,
) -> FiledDeclaracionObservation:
"""Capture a normalized :class:`FiledDeclaracionObservation` using the active page.
Args:
declaration: The :class:`Declaracion` row to observe.
registry_snapshot: Optional pre-built :class:`RegistrySnapshot`. When
omitted, the snapshot is resolved from the declaration coordinates.
artefact_sink: Optional callable storing each captured artefact.
"""
if not (self.session.identity_nif or "").strip():
raise SedeNavigationError(
"AeatSession.identity_nif is empty; cannot bind live filing observation",
translated_message=tr("adapters.sede.errors.empty_identity_nif"),
)
snapshot = registry_snapshot or _registry_snapshot_for_declaration(
declaration,
operation=self._operation,
)
read_policy = _read_guard_policy_from_snapshot(snapshot)
if not await _drive_search(
self._page,
modelo=declaration.modelo,
ejercicio=declaration.ejercicio,
read_policy=read_policy,
):
raise SedeNavigationError(
f"AEAT declarations register does not offer ejercicio {declaration.ejercicio} "
f"for modelo {declaration.modelo}",
translated_message=tr("adapters.sede.errors.ejercicio_unavailable"),
)
row_locator = _row_locator_for_expediente(
self._page,
expediente_id=declaration.expediente_id,
)
from .declarations_capture import capture_filed_declaration_observation_from_row
return await capture_filed_declaration_observation_from_row(
self.session,
declaration,
row_locator=row_locator,
page=self._page,
context=self._context,
registry_snapshot=snapshot,
operation=self._operation,
artefact_sink=artefact_sink,
)
[docs]
@asynccontextmanager
async def open_declarations_register(
session: AeatSession,
*,
operation: PinnedAuthorityOperation,
settings: Settings | None = None,
playwright: Playwright | None = None,
) -> AsyncGenerator[DeclaracionesRegisterSession]:
"""Open a :class:`DeclaracionesRegisterSession` for repeated filed-declaration register reads."""
async with _open_register_page(session, settings=settings, playwright=playwright) as (
page,
context,
):
yield DeclaracionesRegisterSession(session, page, context, operation=operation)
@asynccontextmanager
async def _open_register_page(
session: AeatSession,
*,
settings: Settings | None = None,
playwright: Playwright | None = None,
) -> AsyncGenerator[tuple[Page, BrowserContext]]:
"""Yield a Playwright ``(page, context)`` bound to the AEAT session.
Standalone register walks and reusable register sessions need the same
bringup: validate persisted session state,
build a :class:`Profile`, spin up Playwright, and create a
context preloaded with the session cookies. Centralising the
helper keeps the two callers structurally identical and
fixes a latent leak where the walker did not close the
underlying :class:`BrowserSession` after use.
Args:
session: Authenticated AEAT session.
settings: Optional :class:`Settings` override.
playwright: Optional pre-started :class:`Playwright` instance
(typically produced by :func:`shared_playwright`). When
provided, the helper reuses it instead of starting its
own — saves ~1s per iteration in bulk sweeps. Caller owns
the lifetime; this helper does not stop a passed-in
instance.
Yields:
A ``(page, context)`` pair. The page is a fresh
:class:`Page` inside the storage-state-loaded context.
Raises:
SedeNavigationError: When the session has no encrypted browser state.
"""
settings = settings or Settings()
storage_state = storage_state_for_session(session)
storage_state_path = session.storage_state_path
if storage_state_path is None:
raise SedeNavigationError(
"AeatSession has no persisted auth session; run `aeat config auth status` first",
translated_message=tr("adapters.sede.errors.no_auth_session"),
)
from .....core.bucket_pointer import require_active_bucket_id
profile = Profile(name=require_active_bucket_id())
if playwright is not None:
async with _opened_browser_session(playwright, settings, profile, storage_state) as (page, context):
yield page, context
return
async with (
shared_playwright_runtime() as pw,
_opened_browser_session(pw, settings, profile, storage_state) as (page, context),
):
yield page, context
@asynccontextmanager
async def _opened_browser_session(
pw: Playwright,
settings: Settings,
profile: Profile,
storage_state: dict[str, object],
) -> AsyncGenerator[tuple[Page, BrowserContext]]:
"""Inner helper: create + tear down a BrowserSession + context."""
async with opened_browser_page(pw, settings, profile, storage_state=storage_state) as (page, context):
yield page, context
[docs]
async def walk_declarations_register(
session: AeatSession,
*,
modelo: str,
ejercicio: int,
settings: Settings | None = None,
playwright: Playwright | None = None,
) -> tuple[Declaracion, ...]:
"""Drive the *Consultar declaraciones presentadas* form for one query.
Args:
session: Authenticated AEAT session whose
``storage_state_path`` carries valid cookies.
modelo: Modelo code to query. The form's modelo combobox is matched on the leading
``"<modelo> -"`` text.
ejercicio: Tax year to query (``2024``).
settings: Optional :class:`Settings` override.
playwright: Optional pre-started Playwright instance
(typically from :func:`shared_playwright`). Reused
across the call to amortise the ~1s startup cost in
bulk sweeps. When ``None``, a fresh instance is started
and torn down per call.
Returns:
Tuple of :class:`Declaracion` records, one per filing row.
Empty when AEAT returns "No se han encontrado resultados".
Raises:
SedeParseError: When the grid declares more records than it rendered,
per :func:`_register_rows_from_snapshot`.
"""
async with _open_register_page(session, settings=settings, playwright=playwright) as (
page,
_context,
):
if not await _drive_search(page, modelo=modelo, ejercicio=ejercicio):
log.info(
"walk_declarations_register: ejercicio unavailable modelo=%s ejercicio=%d",
modelo,
ejercicio,
)
return ()
results = _register_rows_from_snapshot(await page.content(), modelo=modelo, ejercicio=ejercicio)
log.info(
"walk_declarations_register: found %d declaration(s) modelo=%s ejercicio=%d",
len(results),
modelo,
ejercicio,
)
return results
async def _drive_search(
page: Page,
*,
modelo: str,
ejercicio: int,
read_policy: RemoteStateGuardPolicy = READ_GUARD_POLICY,
) -> bool:
"""Fill the form and click Buscar.
Raises :class:`SedeNavigationError` early when AEAT redirects to
the login wall instead of the form (session expired) so callers
get an actionable error rather than an opaque combobox click
timeout.
"""
await _open_register_form(page, modelo=modelo, ejercicio=ejercicio, read_policy=read_policy)
if not await _select_combobox_value(
page,
label_text="Modelo (*)",
option_match=f"{modelo} -",
read_policy=read_policy,
):
raise SedeNavigationError(
f"AEAT declarations register does not offer modelo {modelo!r}",
translated_message=tr("adapters.sede.errors.modelo_unavailable"),
failure_mode=SedeFailureMode.EXTERNAL_SHAPE_CHANGED,
context=await _declarations_page_shape_context_from_page(
page,
stage="modelo_option_missing",
modelo=modelo,
ejercicio=ejercicio,
),
)
if not await _select_combobox_value(
page,
label_text="Ejercicio (*)",
option_match=str(ejercicio),
read_policy=read_policy,
):
return False
try:
assert_declarations_read_browser_action("buscar-declaraciones-presentadas", policy=read_policy)
await (
page.locator("button.z-button")
.filter(has_text="Buscar")
.click(
timeout=get_form_interaction_timeout_ms(),
)
)
except PlaywrightError as exc:
raise SedeNavigationError(
f"clicking Buscar failed: {exc}",
translated_message="adapters.sede.errors.playwright_buscar_click_failed",
failure_mode=SedeFailureMode.LIVE_NAVIGATION_FAILED,
context=await _declarations_page_shape_context_from_page(
page,
stage="buscar_click",
modelo=modelo,
ejercicio=ejercicio,
),
) from exc
await page.wait_for_timeout(get_buscar_settle_ms())
# Buscar is a real submit control. Its landing was previously LOGGED and
# never checked, so a search that left the register was recorded and then
# scraped: the row parser would find no rows, and an empty listing is
# indistinguishable from a taxpayer who filed nothing that ejercicio.
assert_declarations_read_landing(getattr(page, "url", "") or "", policy=read_policy)
log.info(
"declarations register search completed modelo=%s ejercicio=%d shape=%s",
modelo,
ejercicio,
_declarations_page_shape_context(
await page.content(),
landing_url=getattr(page, "url", "") or "",
stage="post_buscar",
modelo=modelo,
ejercicio=ejercicio,
),
)
return True
async def _open_register_form(
page: Page,
*,
modelo: str | None = None,
ejercicio: int | None = None,
read_policy: RemoteStateGuardPolicy = READ_GUARD_POLICY,
) -> None:
"""Navigate to the register form and refuse anything that is not it.
Extracted so the search driver and the availability reader share ONE way in.
A second navigation path would carry its own copy of the login-wall check,
the alert-modal dismissal and the landing prefix rule, and the landing-refusal
enrollment gate only proves that a rule is CALLED — not that a copy of it
stayed in step with this one.
``modelo`` and ``ejercicio`` are diagnostic context only; the availability
reader has no pair yet and passes neither.
Raises:
SedeNavigationError: When navigation fails, AEAT serves something other
than the register, or the form does not render its Modelo label.
"""
try:
assert_declarations_read_http("GET", _LISTING_URL, policy=read_policy)
await page.goto(
_LISTING_URL,
wait_until=_WAIT_NETWORKIDLE,
timeout=navigation_timeout_ms(),
)
except PlaywrightError as exc:
raise SedeNavigationError(
f"goto {_LISTING_URL!r} failed: {exc}",
translated_message=tr("adapters.sede.errors.listing_nav_failed"),
failure_mode=SedeFailureMode.LIVE_NAVIGATION_FAILED,
context=await _declarations_page_shape_context_from_page(
page,
stage="listing_goto",
modelo=modelo,
ejercicio=ejercicio,
),
) from exc
await page.wait_for_timeout(1500)
await _continue_alert_modal(page, read_policy=read_policy)
final_url = page.url
if _DECLARATIONS_LISTING_PATH_PREFIX not in final_url:
log.warning(
"declaraciones register did not load; session may have expired (final_url=%r)",
final_url,
)
raise SedeNavigationError(
f"declaraciones register did not load (final URL: {final_url!r}); "
"session likely expired — run `aeat config auth test` and retry",
translated_message=tr("adapters.sede.errors.session_expired_nav_failed"),
failure_mode=SedeFailureMode.LIVE_NAVIGATION_FAILED,
context=await _declarations_page_shape_context_from_page(
page,
stage="listing_final_url",
modelo=modelo,
ejercicio=ejercicio,
),
)
# Defensive: even when the URL matches, AEAT sometimes serves a
# blank shell with no Modelo label until the JS finishes booting.
try:
await page.get_by_text("Modelo (*)", exact=True).first.wait_for(
state="visible",
timeout=get_form_interaction_timeout_ms(),
)
except PlaywrightError as exc:
raise SedeNavigationError(
"declaraciones register form did not render the 'Modelo (*)' "
f"label within {get_form_interaction_timeout_ms()}ms; "
"session likely expired or AEAT served a maintenance page",
translated_message=tr("adapters.sede.errors.form_render_timeout"),
failure_mode=SedeFailureMode.LIVE_NAVIGATION_FAILED,
context=await _declarations_page_shape_context_from_page(
page,
stage="form_render",
modelo=modelo,
ejercicio=ejercicio,
),
) from exc
async def _open_combobox(
page: Page,
*,
label_text: str,
read_policy: RemoteStateGuardPolicy = READ_GUARD_POLICY,
) -> None:
"""Click open the combobox rendered after ``label_text``.
Extracted from :func:`_select_combobox_value` so the availability reader
opens a combobox through the same locator and the same read-action assertion
that every selection already uses, rather than growing a parallel one.
Raises:
SedeNavigationError: When the combobox button cannot be clicked.
"""
label = page.get_by_text(label_text, exact=True).first
button = label.locator('xpath=following::a[contains(@class,"z-combobox-button")][1]')
try:
assert_declarations_read_browser_action(f"select-{label_text.split()[0].lower()}", policy=read_policy)
await button.click(timeout=get_form_interaction_timeout_ms())
except PlaywrightError as exc:
raise SedeNavigationError(
f"opening combobox after label {label_text!r} failed: {exc}",
translated_message="adapters.sede.errors.playwright_combobox_open_failed",
) from exc
await page.wait_for_timeout(400)
async def _select_combobox_value(
page: Page,
*,
label_text: str,
option_match: str,
read_policy: RemoteStateGuardPolicy = READ_GUARD_POLICY,
) -> bool:
"""Open the combobox after ``label_text`` and pick an option matching ``option_match``."""
await _open_combobox(page, label_text=label_text, read_policy=read_policy)
options = page.locator(".z-comboitem-text")
matching_options = options.filter(has_text=option_match)
if await matching_options.count() == 0:
log.info(
"AEAT combobox option unavailable label=%s option=%s",
label_text,
option_match,
)
return False
target = matching_options.first
try:
assert_declarations_read_browser_action(f"select-option-{option_match}", policy=read_policy)
await target.click(timeout=get_form_interaction_timeout_ms())
except PlaywrightError as exc:
raise SedeNavigationError(
f"selecting option {option_match!r} for {label_text!r} failed: {exc}",
translated_message="adapters.sede.errors.playwright_combobox_select_failed",
) from exc
await page.wait_for_timeout(300)
return True
async def _continue_alert_modal(
page: Page,
*,
read_policy: RemoteStateGuardPolicy = READ_GUARD_POLICY,
) -> None:
"""Dismiss AEAT's informational alert modal without opening alert actions."""
try:
assert_declarations_read_browser_action("alert-modal-continuar", policy=read_policy)
result = await page.evaluate(
"""
() => {
const modal = document.querySelector("#alertsModal");
if (!modal) return "absent";
const style = window.getComputedStyle(modal);
if (style.display === "none" || style.visibility === "hidden") return "hidden";
const buttons = [...modal.querySelectorAll("button")];
const continueButton = buttons.find((button) =>
(button.textContent || "").trim().toLowerCase() === "continuar"
);
if (!continueButton) {
throw new Error("AEAT alert modal has no Continuar button");
}
continueButton.click();
return "continued";
}
""",
)
except PlaywrightError as exc:
raise SedeNavigationError(
f"continuing AEAT alert modal failed: {exc}",
translated_message="adapters.sede.errors.playwright_alert_modal_failed",
) from exc
if result == "continued":
await page.wait_for_timeout(500)
def _row_locator_for_expediente(page: Page, *, expediente_id: str) -> Locator:
"""Return a Playwright locator pointing at the listitem whose ``Expediente`` cell text equals ``expediente_id``.
The match is anchored rather than a substring. This locator scopes every
per-row artefact fetch to one register row, so it is the mechanism that binds
a fetched justificante to the declaration it is recorded against; a
substring filter would let a row whose expediente id merely contains the
target satisfy it. AEAT expediente ids are long tracking numbers with no
known reachable collision, so this closes a latent hazard rather than an
observed one, which is the right time to close it for a sole mechanism.
Surrounding whitespace is tolerated because the cell text comes from
rendered markup, but nothing else is.
"""
exact_cell_text = re.compile(rf"^\s*{re.escape(expediente_id)}\s*$")
return page.locator(".z-listitem").filter(
has=page.locator(".z-listcell").filter(has_text=exact_cell_text),
)
__all__ = [
"Declaracion",
"DeclaracionesRegisterSession",
"_parse_presented_at",
"open_declarations_register",
"shared_playwright",
"walk_declarations_register",
]