cadrumo.application.user_profile.censo_sync module

Censo-derived profile signals, and the censal-read reconciliation.

Two things live here. The first is the read-only projection the ledger proportional-deduction path consumes: the home-office afectación ratio derived from the operator-declared vivienda_office m² facts on the encrypted profile.

The second is the censal autofill: AEAT publishes the taxpayer’s own censal state at the Mis Datos Censales consulta, and censal_facts_from_read() projects that read onto declared profile paths so an operator does not retype what the authority already holds. reconcile_censal_read() splits the projection against what the record already carries — adopting only paths the operator left blank and reporting every disagreement instead of overwriting a declared answer — and the canonical user-profile.censo-review operation commits its encrypted reviewed operand through the single cotejo apply authority.

CENSO_SOURCE_TAG marks an AEAT-verified censo fact and the overview calendar reads it to decide whether censo enrolment is verified. It was dormant for as long as nothing stamped it; the censal read is what finally does, so a profile carrying these facts drops the censo.enrolment_unverified advisory for censo-derived obligations. That is correct only because the consulta is an official AEAT read — anything parsed out of an operator-supplied artefact or a Censos WEB editing surface earns the non-official token instead, never this one.

CENSO_SOURCE_TAG: Final

UserProfileFact.source value marking an AEAT-verified censo fact.

CENSAL_ADOPTABLE_PATHS: Final

Declared profile paths the censal consulta can fill, in emit order. Every entry is a key declared by the user-profile schema; the reconciliation refuses to invent one.

identity.tax_id is deliberately ABSENT. Once the ownership refusal is in force the path can never carry information: a read either belongs to this profile, in which case its identity can only agree and adopting it is a no-op, or it does not, in which case the whole read refuses before any path is considered. A field that cannot carry information should not sit in an adoptable set, where its presence implies it might.

exception CensalIdentityMismatchError(message=None, *, context=None, translated_message=None)

Bases: CensoSyncError

The read belongs to a different taxpayer than the profile.

Parameters:

• message (str | None)

• context (Mapping[str, object] | None)

• translated_message (str | None)

Return type:

None

code

class CensalReconciliation(**data)

Bases: BaseModel

The split of a censal read against what the profile already holds.

adopted are facts for paths the record left blank, safe to write. divergences are paths where AEAT and the operator disagree: the operator’s declared answer stands and the authority’s value is recorded as evidence, never silently substituted.

Parameters:

data (Any)

adopted: tuple[UserProfileFact, ...]

divergences: tuple[tuple[str, str], ...]

censal_facts_from_read(result)

Project an AEAT censal consulta read onto declared profile facts.

Every emitted fact carries CENSO_SOURCE_TAG, the declared provenance token for an AEAT-verified censal read, and targets a path the user-profile schema declares.

Deliberately absent: identity.name and identity.surnames. AEAT renders one combined Apellidos y Nombre field, and recovering the split requires assuming the Spanish two-surname convention. That assumption silently mangles any name not following it — a single-surname holder with two given names parses as two surnames and one given name, reversing the operator’s identity. There is no authority in the read for where the boundary falls, so the projection declines to guess rather than write a plausible-looking wrong name.

The fiscal identity is NOT projected. The reconciliation needs it to decide whether the read belongs to this profile at all, but it takes it from the read directly rather than from here: carrying it in this tuple made a collection named for adoption silently load-bearing for an ownership refusal, so removing a path from the tuple would have switched that refusal off with nothing failing.

Parameters:

result (CensalObservation) – The parsed censal consulta read.

Return type:

tuple[UserProfileFact, ...]

Returns:

The projected facts in CENSAL_ADOPTABLE_PATHS order, omitting any path the read left empty.

reconcile_censal_read(record, facts, *, incoming_identity)

Split projected censal facts into safe adoptions and reported disagreements.

Every decision below is taken against the operator’s own UserProfileRecord, read as the ordered fact history rather than a flattened value map: which side wrote a path, and whether it was later cleared, are facts about that history and are invisible in a value-only projection.

A path the record leaves blank is adopted. A path whose recorded value already equals the read is a no-op and is emitted as neither. A path where the two differ is decided by WHO wrote the recorded value:

• an operator-declared value is reported as a divergence and left standing — an autofill must not overwrite a declared answer just because the authority disagrees;

• a value a previous censal read adopted is REFRESHED, because there is no operator answer to protect. Both sides are the authority’s, one stale and one current, so reporting it as a disagreement would both strand the profile on the old value forever and assert a conflict that does not exist. A censal address change is the likeliest reason to pull at all, so this is the path that keeps a profile current.

A path the operator explicitly CLEARED is never re-adopted, whatever wrote it originally. A clear is a declaration — I do not want this on my profile — and it is the one form of declaration a value-only projection cannot show, because a cleared path still projects its previous value. The authority’s value is reported as a divergence instead of being written back: the operator’s deletion stands, and they still learn that AEAT holds something there, which matters because they file against AEAT’s record and not against their profile.

Comparison is exact after a whitespace strip. No per-field normalisation is applied: deciding that two differently-written cadastral references or postcodes mean the same thing is a guess about an authority’s format, and a spurious divergence is recoverable by the operator where a silent equation of two genuinely different values is not.

Parameters:

• record (UserProfileRecord | None) – The active profile record, or None for a fresh profile.

• facts (Sequence[UserProfileFact]) – The projected censal facts, which are the adoptable paths and nothing else.

• incoming_identity (str | None) – The fiscal identity the read carries, taken from the read itself. It is a parameter rather than something recovered from facts because the ownership refusal below depends on it: routing it through the projection made a collection named for adoption silently load-bearing for a guard, so removing a path from that collection would have switched the guard off with nothing failing.

Return type:

CensalReconciliation

Returns:

The CensalReconciliation split.

class CensoSyncService(*, bucket_id, profiles)

Bases: object

Read-only censo-derived signals for the active profile bucket.

The service no longer captures, compares, or applies censo snapshots (the live scrape was retired); it exposes the single surviving read: the home-office afectación ratio the ledger ratios and preflight paths consume via bound_raw_afectacion_ratio(), derived from the operator-declared vivienda_office m² facts the operator maintains through config profile edit.

Parameters:

• bucket_id (str)

• profiles (ProfileRecordRepository)

property bucket_id: str

Return the normalized bucket identifier bound to this service.

bound_raw_afectacion_ratio(*, profile_id)

Return office_m2 / total_m2 from the operator-declared profile facts.

Used by the ledger ratios CLI and the manual-transaction classify path to apply the legally-effective cadrumo.application.ledger.ratios.censo_override_warning() and cadrumo.application.ledger.ratios.censo_business_pct_for() helpers without each consumer re-implementing the profile-fact lookup. Reads the encrypted profile record’s canonical path-value projection — the same operator-declared facts config profile edit writes and the deadline engine hydrates — so updating the vivienda_office m² through the CLI drives this ratio directly. Returns None when the profile has no persisted record OR when either vivienda_office.total_m2 / vivienda_office.office_m2 is absent / non-decimal / zero.

Return type:

Decimal | None

Parameters:

profile_id (str)

raw_afectacion_ratio_for_record(record)

Return office_m2 / total_m2 from an already-loaded profile record, if declared.

Core types: UserProfileRecord.

Return type:

Decimal | None

Parameters:

record (UserProfileRecord)

bound_raw_afectacion_ratio(*, bucket_id, profile_id, operation)

Read one encrypted profile’s dwelling ratio through the caller’s pin.

Return type:

Decimal | None

Parameters:

• bucket_id (str)

• profile_id (str)

• operation (PinnedAuthorityOperation)

bound_raw_afectacion_ratio_for_bucket(bucket_id, *, operation)

Return the bucket’s own censo-declared office_m2 / total_m2, if any.

The bucket IS the profile for this question, so the two identifiers the service takes separately collapse to one. Two call sites had each spelled that collapse out, which is one place too many for a rule about which id means what.

The CLI listing deliberately does NOT use this: it resolves a profile_id that can differ from the bucket, and folding it in here would quietly answer a different question than it asked.

Parameters:

• bucket_id (str) – The bucket whose profile facts are read.

• operation (PinnedAuthorityOperation) – The caller-owned authority generation used to decode the encrypted profile record.

Return type:

Decimal | None

Returns:

The raw afectación proportion, or None when the profile is absent or declares no usable m².