cadrumo.adapters.outbound.aeat.sede.notifications module

Read-only notifications/messages reader for the authenticated AEAT sede.

Captured against a production account; the sede exposes two notification surfaces:

• Summary (/wlpl/GNNO-JDIT/ResumenInteresados) — unread notifications + unread communications, two tables keyed by número de certificado.

• Query (/wlpl/GNNO-JDIT/SvInteresadosQuery?VEZ=BUSCAR1) — a full-text search form plus a results table with one row per (notification, communication, or pending) item.

Both surfaces are reachable on the www6 Cl@ve-dispatched subdomain; an earlier assumption that notifications lived only on www1 (cert-only) was wrong.

The reader is structurally read-only: no form submission, no state-changing URL. Acknowledgement (acuse) is a strictly local concern (the local inbox tracks read/unread); the reader never tells AEAT “this was read”.

Public surface: RemoteNotification, NotificationsSnapshot, parse_notifications_query(), fetch_notifications_query(), notifications_query_url().

BrowserSessionFactory

Async factory returning an object exposing create_context(storage_state=...) + close().

Production passes default_browser_session_factory(). The callable seam keeps browser-session construction owned by the outbound adapter composition boundary.

alias of Callable[[Settings], Awaitable[Any]]

class SedeNotificationTipo(*values)

Bases: StrEnum

What kind of item a sede notifications row represents.

NOTIFICACION

A notificacion, which starts a legal response window.

COMUNICACION

A comunicacion, which does not.

PENDIENTE

Listed as pending, with its kind not yet resolved.

UNKNOWN

The row did not say, and nothing may be inferred from that.

SedeNotificationTipoValue

Every kind, for the strict observation field that records what a row was.

The model validates strictly and is built from scraped text, so it takes the literal over the members rather than the bare enum, which would refuse the plain token.

alias of Literal[NOTIFICACION, COMUNICACION, PENDIENTE, UNKNOWN]

class RemoteNotification(**data)

Bases: BaseModel

One row of AEAT’s notifications/communications surface.

tipo distinguishes a formal Notificación (legally binding, triggers the ten-day acuse window) from a lighter-weight Comunicación. The pendiente value marks the placeholder state AEAT shows for items that have been issued but not yet delivered.

Variables:

• certificado_id – Nº de certificado — AEAT’s per-notification identifier, typed AeatCertificadoId.

• tipo – Row class — "notificacion" | "comunicacion" | "pendiente" | "unknown".

• concepto – Free-text concepto / subject line (may be empty for pending rows).

• titular_nif – NIF / NIE of the titular (verbatim from AEAT).

• titular_nombre – Free-text nombre / razón social of the titular.

• destinatario_nif – NIF of the destinatario (may equal titular).

• destinatario_nombre – Free-text nombre / razón social of the destinatario.

• fecha_emision – Fecha de emisión as a local date.

• fecha_notificacion – Fecha de notificación when the row has been delivered, else None.

• modo_notificacion – Text AEAT prints for the delivery channel, or None when not yet delivered.

• leida – True if AEAT marks the row as “Leída”, False otherwise, None for pending rows with no column value.

• source_url – URL the row was scraped from.

• mode – Structural read-only marker.

Parameters:

data (Any)

certificado_id: AeatCertificadoId

tipo: SedeNotificationTipoValue

concepto: str

titular_nif: str

titular_nombre: str

destinatario_nif: str

destinatario_nombre: str

fecha_emision: date

fecha_notificacion: date | None

modo_notificacion: str | None

leida: bool | None

source_url: AnyHttpUrl

mode: Literal['read']

class NotificationsSnapshot(**data)

Bases: BaseModel

One sede-side capture of the entire notifications surface.

Variables:

• rows – Every notification / communication the sede returned.

• captured_at – UTC timestamp at snapshot completion.

• source_url – The sede URL the rows were scraped from (summary or query, depending on caller choice).

• mode – Structural read-only marker.

Parameters:

data (Any)

rows: tuple[RemoteNotification, ...]

captured_at: datetime

source_url: AnyHttpUrl

mode: Literal['read']

parse_notifications_query(html, *, source_url)

Parse the full SvInteresadosQuery results table.

This is the canonical list view; every row carries the full column set (tipo, leída, modo de notificación, etc.).

Parameters:

• html (str) – Raw HTML body of a SvInteresadosQuery results page.

• source_url (str) – URL the HTML was scraped from (recorded on the returned snapshot).

Return type:

NotificationsSnapshot

Returns:

A NotificationsSnapshot with one row per item.

notifications_query_url(*, today, lookback_years=None)

Return the query URL carrying an explicit, widened date window.

AEAT’s search defaults fecha desde to one month before today, so a bare request answers “what arrived this month” rather than “what is on the register”. Reading it that way reported a near-empty inbox for an account holding older notifications – the surface was answering the question it was asked. The window is therefore stated rather than inherited.

The filters are sent as GET parameters. The form declares method="post", but the servlet honours the same fields on a query string, so the reader stays a pure navigation and needs no read-POST allowance on its guard policy.

Parameters:

• today (date) – The upper bound of the window, injected rather than read from the clock so the URL is a pure function of its inputs.

• lookback_years (int | None) – Override for the configured lookback.

Return type:

str

Returns:

The absolute query URL with its date window and both filters widened.

class NotificationDocument(**data)

Bases: BaseModel

The PDF AEAT serves for one already-read notification.

Variables:

• certificado_id – The notification the document belongs to.

• pdf_bytes – The document as served. Held in memory by the caller and written only to encrypted storage; never to disk.

• pdf_sha256 – Digest of pdf_bytes, for evidence integrity.

• source_url – The detail endpoint the document was served from.

• mode – Structural read-only marker.

Parameters:

data (Any)

certificado_id: AeatCertificadoId

pdf_bytes: bytes

pdf_sha256: ContentDigest

source_url: AnyHttpUrl

mode: Literal['read']

notification_detail_url(certificado_id)

Return the detail-page URL for one notification.

Return type:

str

Parameters:

certificado_id (str)

assert_notification_content_readable(row)

Refuse to fetch the content of any notification AEAT has not recorded as read.

This guard exists for a legal reason, not a technical one. AEAT serves a notification’s content and performs its comparecencia through the same control. For a notification already read, driving it redisplays a document and changes nothing. For an unread one, that same control is the act by which the notification becomes legally served – it starts the appeal and payment periods, and AEAT requires the taxpayer’s own signature to complete it.

That signature is the taxpayer’s to give. An agent must never stand in for it, must never provoke the flow that asks for it, and must never cause a notification to be served as a side effect of reading a mailbox. So the predicate is not “is this fetchable” but “has AEAT already recorded that the taxpayer read this” – and anything else is refused, including a notification that is already SERVED but still unread. Service and reading are different events, and only the second one licenses this fetch.

Fail-closed by construction: leida is None for pending rows and for rows whose column AEAT did not render, and both refuse here.

Parameters:

row (RemoteNotification) – The notification whose content a caller wants.

Raises:

SedeNavigationError – When the row is anything other than already read.

Return type:

None

async fetch_notification_document(session, row, *, settings=None)

Fetch the PDF of one ALREADY-READ notification.

Guarded by assert_notification_content_readable(), which refuses every row AEAT does not already report as read. The refusal is checked before any request crosses the wire, so a refused row produces no AEAT contact at all.

Parameters:

• session (AeatSession) – An authenticated AeatSession.

• row (RemoteNotification) – The notification to fetch. Must be already read.

• settings (Settings | None) – Optional core.config.Settings override.

Return type:

NotificationDocument

Returns:

The NotificationDocument as served.

Raises:

SedeNavigationError – When the row is not already read, or the landing is off-policy.

async fetch_notifications_query(session, *, settings=None, today=None)

Live-fetch SvInteresadosQuery over a widened date window using Cl@ve.

Parameters:

• session (AeatSession) – An authenticated AeatSession.

• settings (Settings | None) – Optional core.config.Settings override.

• today (date | None) – Upper bound of the search window; defaults to the current date.

Return type:

NotificationsSnapshot

Returns:

A NotificationsSnapshot parsed from the live HTML.

assert_notifications_read_landing(landing_url)

Refuse a DEHú landing outside the exact notification read routes.

Return type:

None

Parameters:

landing_url (str | None)