cadrumo.adapters.outbound.aeat.sede._adapter_utils module

Shared private helpers for AEAT sede browser drivers.

Hosts the small text-normalisation, error-formatting, and selector-probe helpers consumed by every sede driver. Drivers (_groi_check, _nif_iva_check, future siblings) inject their own surface label and share one factual shape-change failure without re-implementing the same logic per driver.

make_locate_helper() and assert_query_browser_action_for() factor out the two private helper shapes that each checker driver used to duplicate.

assert_read_landing() is the package’s landing refusal: the one wall that sees where AEAT actually dispatched a read, rather than where the driver asked to go. Every driver that fills a form and clicks a control needs it, because the click issues a browser form POST that the first-party HTTP guard never sees and that the forbidden-verb source scan deliberately permits.

is_aeat_auth_gate_redirect(current_url)[source]

Return whether current_url is AEAT’s configured auth-gate landing.

The detector accepts the configured AEAT host or a real subdomain, but not a user-info or port-shaped authority that merely ends in that suffix. Callers retain responsibility for translating an affirmative result into their surface-specific navigation error.

Return type:

bool

Parameters:

current_url (str)

assert_query_browser_action_for(policy, action)[source]

Assert that action is permitted under policy.

Raises RegistryValidationError via assert_remote_operation_allowed() if the action pattern is not allowed. Both the GROI and NIF-IVA drivers close over their own RemoteStateGuardPolicy objects; this shared helper removes the duplicate _assert_query_browser_action bodies they used to carry.

Parameters:

  • policy (RemoteStateGuardPolicy) – The guard policy the driver was initialised with.

  • action (str) – Browser action label to validate (e.g. "open-groi-form").

Return type:

None

assert_read_http_for(policy, method, url)[source]

Assert that a wire-crossing read of url is permitted under policy.

The http counterpart of assert_query_browser_action_for(), and the one body behind every sede reader’s fail-closed no-write guard. The censal, notifications and expedientes-walker readers each carried an identical copy differing only in the module-level policy it named; each still declares its OWN policy — the surfaces genuinely differ in which hosts they admit — and passes it here, so the refusal RULE has one home while the per-surface posture stays local.

A guard that refuses writes is the wrong thing to hold three copies of: a correction applied to one leaves the other two admitting what it now refuses, and nothing fails until a write reaches AEAT.

Parameters:

  • policy (RemoteStateGuardPolicy) – The read policy the calling surface declared.

  • method (str) – HTTP method of the navigation about to be made.

  • url (str) – Absolute target URL.

Raises:

RegistryValidationError – The navigation is not a permitted read.

Return type:

None

require_playwright_page(raw_page)[source]

Return raw_page as a Playwright Page or raise a typed adapter error.

Return type:

Page

Parameters:

raw_page (object)

make_locate_helper(surface_label)[source]

Return a _locate coroutine pre-bound to surface_label.

Both the GROI and NIF-IVA drivers wrap first_visible_locator() with the same body, differing only in the surface_label they inject. This factory eliminates that duplicate; each driver calls:

_locate = make_locate_helper("GROI")

and then uses _locate(page, selectors, stage=..., description=..., timeout_ms=...) directly.

Parameters:

surface_label (str) – Sede surface name for log and error messages.

Return type:

Callable[[Page, tuple[str, ...], str, str, int], Coroutine[Any, Any, Locator]]

Returns:

An async callable with the same signature as the internal _locate helpers the drivers previously defined individually.

landed_origin(landed_url)[source]

Return the scheme://host a read actually landed on, or None.

The single extraction every sede reader uses to answer “which host answered this read”. Returns None when the landing carries no usable scheme and host – an empty or absent page.url, about:blank, or a relative path – leaving the REFUSAL to the caller, which is the only layer that knows what its surface is allowed to do with the absence.

Callers must not substitute a constructed origin for a None result. A recorded URL is a claim about where a read happened, and a read whose host cannot be established has no such claim to make.

Return type:

str | None

Parameters:

landed_url (str | None)

assert_read_landing(landing_url, *, surface, policy, allowed_path_prefixes)[source]

Refuse a landing that is not one of a read surface’s declared read pages.

The landing refusal is the only wall that sees where AEAT actually DISPATCHED. The sibling walls each miss a browser-driven navigation by construction: the package’s forbidden-verb source scan deliberately permits click / fill / press, and the HTTP guard is consulted only for a first-party request, never for the form POST a click issues. A driver that fills a form and clicks a control therefore has no mechanism telling it where it ended up unless it calls this.

The rule is an ALLOW-list, not a denylist, and it fails closed at every step: a landing with no readable origin is refused, a landing the surface’s own guard policy does not admit is refused, a surface that declares no read pages refuses everything, and a landing outside the declared pages is refused. An unrecognised landing is a refusal — the guard never has to enumerate the write surfaces it has not seen, which is what a denylist has to get exhaustively right.

Routing the landed URL through policy is deliberate rather than redundant: that evaluation carries the canonical AEAT write-verb token scan (presentar, firmar, pagar, submit and the rest) and the authority checks that refuse a credentialed, off-port or off-host landing. What it CANNOT see is an AEAT write path whose URL contains no write verb at all — BUGC-JDIT/ModifDomiDual and BU36-ASIS/M036/index.zul are both writes and neither carries a token the scan knows. The path allow-list is what closes that half, so the two checks are complementary and both are required.

A prefix grants everything beneath it, and an AEAT application directory is not a read-only unit: an authenticated capture of the live sede found a payment launcher served from the SAME /wlpl/ application directory as a read-only consulta, under sibling paths. Widening a prefix to the application segment therefore admits whatever else AEAT chose to host there, which the write-verb scan will not necessarily catch. Take each prefix at the narrowest level that still covers the intended surface, and prefer the endpoint over the directory unless a surface has per-surface evidence that it genuinely reads across the directory.

The censal reader (censal_datos) keeps its own marker-keyed refusal in addition to this one. Its consulta page and its modification siblings are reached through the same launcher, so it names the write paths outright; that predicate is exported for conformance gates and is not superseded here.

Parameters:

  • landing_url (str | None) – The URL AEAT actually served, after redirects — read off the page, never the URL that was requested.

  • surface (str) – Sede surface label used in the refusal message and context (e.g. "GROI"), so a refusal names which driver refused.

  • policy (RemoteStateGuardPolicy) – The surface’s own remote-state guard policy.

  • allowed_path_prefixes (tuple[str, ...]) – Absolute path prefixes this surface is allowed to land on. Empty refuses every landing.

Raises:

SedeNavigationError – When the landing is unreadable, off-policy, or outside the surface’s declared read pages.

Return type:

None

response_media_type(content_type)[source]

Return the bare media type from a Content-Type header value.

Strips any parameter tail (; charset=binary), surrounding whitespace, and case, so a header may carry parameters without changing what media type it names.

Return type:

str

Parameters:

content_type (str)

assert_pdf_response(*, status, content_type, body, subject)[source]

Validate one AEAT PDF download response, or raise.

The single contract every sede PDF capture path shares: a 2xx status, a non-empty body, and a Content-Type whose media type IS PDF_MIME_TYPE.

The media-type comparison is equality on the parameter-stripped header rather than a substring test. A substring test admits any type that merely CONTAINS the token: application/notpdf and text/pdf satisfy a pdf in ... substring check, and x-application/pdf-trap satisfies even a substring check against PDF_MIME_TYPE itself. None of those is a PDF, and a captured artefact is stored as filing evidence, so admitting one records a non-PDF body under a PDF kind. Equality on the media type still admits the parameterised application/pdf; charset=binary AEAT actually sends.

Parameters:

  • status (int) – HTTP status code of the PDF response.

  • content_type (str) – Raw Content-Type header value, possibly parameterised.

  • body (bytes) – Raw response body bytes.

  • subject (str) – Caller-supplied identification of what was fetched, embedded verbatim in the failure message (e.g. "CSV='ABC123'") so each capture path keeps its own diagnostic handle.

Raises:

JustificanteFetchError – On a non-2xx status, an empty body, or a content type whose media type is not application/pdf.

Return type:

None

normalize_response_text(text)[source]

Casefold + strip diacritics + collapse whitespace for marker matching.

Return type:

str

Parameters:

text (str)

normalize_display_text(text)[source]

Collapse a page value’s whitespace while keeping it readable as printed.

The sibling of normalize_response_text() for the other half of the job: that one casefolds and folds diacritics so a marker can be MATCHED, which makes its output unfit to show or store. This one only tidies the spacing — AEAT pages lay values out with non-breaking spaces and wrapped indentation — so "Cuota Disponible" survives as written rather than arriving as "cuota disponible".

Reach for this whenever the normalised text is the VALUE; reach for normalize_response_text() only when it is a lookup key.

Return type:

str

Parameters:

text (str)

cell_text(cells, index)[source]

Return the cell at index, or None when it is absent or empty.

One concept that was implemented twice, five lines apart in shape: the declaraciones listbox parser’s _cell_text and the notifications table parser’s _safe_cell. They read as divergent because one stripped and one did not, but both callers build their cell lists with get_text(" ", strip=True), so the strip was a no-op and the two were functionally identical on every real input.

The strip is KEPT rather than dropped as dead. “Absent or empty” is this helper’s contract, and an implementation that only satisfies it while every caller happens to pre-strip is coupled to an invariant stated nowhere. It costs nothing and it means the contract holds for a caller that arrives later without that habit.

A missing index and an empty cell collapse to the same None on purpose: AEAT tables omit a column and blank it interchangeably, and no caller distinguishes them.

Return type:

str | None

Parameters:
bounded_text(value, *, max_length=120)[source]

Return value as tidy text, truncated to at most max_length characters.

What a page-shape diagnostic records for an attribute it does not interpret, so a pathological page cannot put an unbounded string into a dump file or an error context.

The ellipsis is one character rather than three dots for an arithmetic reason: the truncation keeps max_length - 1 characters, so a one-character marker lands exactly on the bound while "..." overshoots it by two. The wallet and declarations shape readers each carried a copy of this and had already drifted onto different markers, which is how the overshooting one survived.

Return type:

str

Parameters:
redacted_url(value)[source]

Return value with its query string and fragment removed.

Everything identifying rides in the query on a sede URL — the NIF, the expediente reference, the session token — so a URL bound for a log line, an error context or a diagnostic dump goes through here first. Scheme, host and path survive, which is what makes the record diagnosable.

Distinct from application.auth _redacted_url_summary, which keeps the query KEY names and drops the scheme; that one answers “which parameters did this request carry”, not “where did it go”.

Return type:

str | None

Parameters:

value (object)

SPANISH_NEGATIVE_VERDICT_MARKERS: tuple[str, ...]

Normalised AEAT phrases that reject an identity, shared by every checker.

Every sede identity checker reads the same Spanish rejection vocabulary off the same AEAT template family, so the negative table is one contract rather than a per-driver table. It is deliberately the union of the phrases observed across the checkers: a driver that omits one classifies an explicit rejection as unknown at best, and – when a generic valido substring survives in its positive table – as valid, turning a refusal into a false pass.

Markers are matched against normalize_response_text() output, so they are casefolded, unaccented, and whitespace-collapsed.

extract_marker_verdict(body_text, *, positive_markers, negative_markers=('no consta', 'no valido', 'no es valido', 'no es un nif valido', 'el campo nif no es un nif valido', 'no identificado', 'no esta identificado', 'no se encuentra identificado', 'operador no identificado', 'invalid'))[source]

Classify an AEAT response body as valid, invalid, or unknown.

Negative markers are tested first and win outright. AEAT phrases a rejection by negating the same word it uses to affirm (no es un NIF válido), so a positive-first or negative-incomplete parser reads an explicit refusal as a pass. Precedence, not marker richness, is what makes the classification safe.

Parameters:

  • body_text (str) – Raw response body text scraped from the AEAT page.

  • positive_markers (tuple[str, ...]) – Driver-specific phrases that affirm the identity. Positive vocabulary is surface-specific (a GROI registration phrase does not affirm a VIES NIF-IVA), so it stays per driver.

  • negative_markers (tuple[str, ...]) – Rejection phrases; defaults to the shared SPANISH_NEGATIVE_VERDICT_MARKERS contract.

Return type:

Literal[<IdentityCheckVerdict.VALID: 'valid'>, <IdentityCheckVerdict.INVALID: 'invalid'>, <IdentityCheckVerdict.UNKNOWN: 'unknown'>]

Returns:

"invalid" on any negative marker, "valid" on any positive marker, "unknown" for empty or structurally unanswerable text.

registry_failure_message(exc)[source]

Build a registry-facing error string enriched with the failure_mode context field.

Sede driver exceptions carry a context mapping; this helper extracts failure_mode (falling back to a site_health:<state> label when only a state key is present) and appends it to the base str(exc) so callers wrapping the exception into a RegistryValidationError preserve the diagnostic context. Returns str(exc) unchanged when no failure_mode is derivable.

Return type:

str

Parameters:

exc (BaseException)

async first_visible_locator(page, selectors, *, stage, description, timeout_ms, probe_timeout_ms, surface_label)[source]

Return the first selector in selectors that resolves to a visible element.

Probes each selector in order using a short probe_timeout_ms deadline. If a selector is not visible within that window, PlaywrightError / PlaywrightTimeoutError is caught and the next selector is tried. When no selector resolves, SedeParseError is raised with SedeFailureMode.EXTERNAL_SHAPE_CHANGED so callers can distinguish “AEAT changed the page layout” from transient network timeouts.

Parameters:

  • page (Page) – Playwright Page on which to probe the selectors.

  • selectors (tuple[str, ...]) – CSS selector strings tried in declaration order.

  • stage (str) – Opaque label for the current driver stage, included in the error context.

  • description (str) – Human-readable description of the expected element, used in the error message.

  • timeout_ms (int) – Overall operation timeout (ms); probe_timeout_ms is capped to this value.

  • probe_timeout_ms (int) – Per-selector visibility probe budget (ms).

  • surface_label (str) – Sede surface name included in log and error messages (e.g. "GROI").

Return type:

Locator

Returns:

The first Locator from selectors whose element was visible within probe_timeout_ms.

Raises:

SedeParseError – When every selector probe timed out or failed.

nif_check_operation_tail(expected)[source]

Build the shared per-NIF check + discard-session operation tail.

Both the GROI and NIF/IVA sede drivers close their planned-operation sequence identically: one check-nif-<NIF> browser action per declared NIF (normalised to upper-case and sorted so the operation labels the remote-state guard pre-flight sees on the driverless oracle path match what the live driver emits), followed by one discard-session action. Each driver prepends its own URL/form prologue and appends this tail.

Return type:

tuple[RemoteOperation, ...]

Parameters:

expected (Mapping[str, object])