aeat.application.user_profile._censo_sync module

Operator-facing Modelo 036 censo-sync service.

CensoSyncService exposes the four-verb surface the CLI mounts under aeat config profile censo {refresh, show, compare, apply}. Persisted profile facts are stamped onto a UserProfileRecord on apply. AEAT is the binding legal source of truth for censo data; this service is the only application path that captures censo facts into the secure store and stamps them onto the operator’s profile.

The service composes:

The sede G313 live fetch is injected as a fact_source callable so the same service body backs both the production sede adapter and real-behavior tests without conditional code paths. This service owns profile-fact writes and CENSO_REFRESHED / CENSO_APPLIED event emission. It does not mutate work units, calculation revisions, drafts, or filing records directly; consumers that need downstream stale-state handling react to the bucket events outside this module.

CENSO_SOURCE_TAG: Final

UserProfileFact.source value stamped on every censo-derived fact.

CENSO_DERIVED_SOURCE_TAG: Final

UserProfileFact.source for facts derived from censo plus profile evidence.

class CensoComparisonStatus(*values)[source]

Bases: StrEnum

Per-field comparison outcome between snapshot and profile.

Variables:

  • MATCHES – Censo and profile values are identical.

  • DIVERGES – Both present, values differ.

  • PROFILE_ONLY – Operator has a value the censo does not publish.

  • CENSO_ONLY – AEAT publishes a value the operator’s profile does not yet record.

MATCHES
DIVERGES
PROFILE_ONLY
CENSO_ONLY
class CensoFieldComparison(**data)[source]

Bases: BaseModel

One field-by-field row of a CensoProfileComparison.

Parameters:
path: str
censo_value: str | None
profile_value: str | None
status: CensoComparisonStatus
class CensoProfileComparison(**data)[source]

Bases: BaseModel

Result of censo compare: full field-by-field diff payload.

Parameters:

  • snapshot_id (str)

  • profile_id (Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=36, pattern=^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$, ascii_only=None)])

  • captured_at (datetime)

  • rows (tuple[CensoFieldComparison, ...])

snapshot_id: str
profile_id: ProfileId
captured_at: datetime
rows: tuple[CensoFieldComparison, ...]
property diverging: tuple[CensoFieldComparison, ...]
property censo_only: tuple[CensoFieldComparison, ...]
property profile_only: tuple[CensoFieldComparison, ...]
class CensoApplyResult(**data)[source]

Bases: BaseModel

Result of censo apply: which facts landed on the profile.

Parameters:

  • snapshot_id (str)

  • profile_id (Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=36, pattern=^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$, ascii_only=None)])

  • written_paths (tuple[str, ...])

  • unchanged_paths (tuple[str, ...])

  • derived_paths (tuple[str, ...])

  • seeded_home_office_categories (tuple[str, ...])

snapshot_id: str
profile_id: ProfileId
written_paths: tuple[str, ...]
unchanged_paths: tuple[str, ...]
derived_paths: tuple[str, ...]
seeded_home_office_categories: tuple[str, ...]
CensoFactSource

Callable returning the AEAT-side censo facts for one refresh.

In production this is wired to the sede G313 adapter; in tests it is a constant callable returning a fixture dictionary. The service stays sede-agnostic so the same body covers both call paths.

alias of Callable[[], Mapping[str, str]]

class CensoSyncService(*, bucket_id, snapshots=None, profiles=None, events=None)[source]

Bases: object

Four-verb operator-facing service over censo snapshots.

refresh captures a new snapshot from AEAT, show returns the active or named snapshot, compare diffs the snapshot against the operator’s current profile, and apply writes the snapshot facts onto the profile under the aeat_censo_read provenance tag.

Parameters:
property bucket_id: str
refresh_censo(*, profile_id, source_url, fact_source)[source]

Fetch fresh censo facts, capture them as the new ACTIVE snapshot, and return the CensoSnapshot.

Raises CensoNotAvailableError when fact_source returns an empty mapping — AEAT publishes no censo for the operator’s NIF, so the caller should re-run after enrolment (or confirm the certificate is registered against the NIF).

Return type:

CensoSnapshot

Parameters:
async refresh_censo_from_sede(*, profile_id)[source]

Drive the live G313 Playwright fetch and persist the snapshot.

Acquires (or refreshes) an authenticated AeatSession, navigates to the documented G313 launcher, parses the response into a CensoFactSet, projects it into the dotted snapshot mapping, and captures via refresh_censo().

Parameters:

profile_id (str) – The profile id the captured censo snapshot is scoped to.

Return type:

CensoSnapshot

Returns:

The persisted CensoSnapshot for the profile.

Raises:

CensoNotAvailableError – when AEAT publishes no censo for the operator’s NIF (empty CensoFactSet).

show_censo(*, profile_id, snapshot_id=None)[source]

Return one CensoSnapshot — the latest ACTIVE by default.

Return type:

CensoSnapshot

Parameters:

  • profile_id (str)

  • snapshot_id (str | None)

compare_censo_with_profile(*, profile_id, snapshot_id=None)[source]

Compare the active (or named) snapshot to the current profile.

Returns a CensoProfileComparison with one row per censo-tracked path, classified into matches / diverges / profile_only / censo_only.

Return type:

CensoProfileComparison

Parameters:

  • profile_id (str)

  • snapshot_id (str | None)

apply_censo_to_profile(*, profile_id, snapshot_id=None)[source]

Stamp the snapshot facts onto the profile, replacing prior aeat_censo_read facts.

Every raw censo fact lands as a UserProfileFact with source = "aeat_censo_read". Facts defensibly derived from the censo/profile pair land with source = "aeat_censo_derived". Pre-existing facts from either censo source are replaced; facts from other sources (manual_cli, wizard) are preserved untouched so operator-entered values stay addressable for the compare verb.

Emits CENSO_APPLIED on the bucket-event-history catalogue itself so downstream maintenance can react without the CLI owning event-enrolment policy.

Raises CensoApplyConflictError when the profile is absent — there is nothing to stamp facts onto.

Returns:

The apply result details.

Return type:

CensoApplyResult

Parameters:

  • profile_id (str)

  • snapshot_id (str | None)

bound_raw_afectacion_ratio(*, profile_id)[source]

Return office_m2 / total_m2 from the active censo snapshot.

Used by the ledger ratios CLI and the manual-transaction classify path to apply the legally-effective aeat.application.ledger._ratios.censo_override_warning() and aeat.application.ledger._ratios.censo_business_pct_for() helpers without each consumer re-implementing the snapshot lookup. Returns None when no ACTIVE snapshot exists OR when either vivienda_office.total_m2 / vivienda_office.office_m2 is absent / non-decimal / zero.

Return type:

Decimal | None

Parameters:

profile_id (str)