"""Filing-record actions for modelo calculation revisions.
:func:`~aeat.application.modelo.file_modelo_revision` promotes a verified
:class:`CalculationRevision` into a current
:class:`ModeloRecord` after the
:class:`WorkflowEngine` preflight gate passes. Filing
transitions and audit entries are persisted through the
:class:`BucketEventHistoryRepository` path shared by the
modelo revision services.
The action records the operator's local/internal filing state only. It never
submits to AEAT, never marks AEAT acceptance, and never fabricates official
external evidence. A successful transition sets the target revision to
``PRESENTADO``, creates a ``VIGENTE``
:class:`ModeloRecord` with ``aeat_accepted=False``, and
delegates cross-period carry projection to
:func:`~aeat.application.modelo._revision_persistence.persist_filed_revision`,
which stamps locally-filed observations as non-official ``app_filing`` evidence.
See Also:
:func:`~aeat.application.modelo.import_external_filing_evidence`:
Separate AEAT-attested import path that creates
:class:`ExternalEvidence` baselines; this local
filing action deliberately does not.
:func:`~aeat.application.modelo._revision_persistence.persist_filed_revision`:
Persists the filing catalogue, revision state, work-unit pointers,
bucket events, participation index rows, and optional carry observation.
:func:`~aeat.application.modelo._filed_revision_observation.persist_filed_revision_observation`:
Projects filed casillas into non-official cross-period observations.
:func:`~aeat.application.modelo._result_disposition_resolution.resolve_modelo_result_disposition`:
Resolves the shared Modelo 303 refund/carry disposition before the file
transition persists.
:func:`~aeat.application.modelo._verification_actions._require_cross_period_clean_state`:
Rechecks cross-period dependencies before local filing state is written.
"""
from __future__ import annotations
from collections.abc import Iterable
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING
from ...adapters.persistence.profile.buckets import BucketEventHistoryRepository
from ...adapters.persistence.profile.modelos_calculation import CalculationRevisionCatalogueRepository
from ...adapters.persistence.profile.modelos_filing import ModeloRecordCatalogueRepository
from ...adapters.persistence.profile.modelos_verification_reports import VerificationReportCatalogueRepository
from ...adapters.persistence.profile.modelos_work_units import WorkUnitCatalogueRepository
from ...core import RefundElection
from ...core.config import Settings
from ...core.time import now as _utc_now
from ...domain.buckets import BucketEventHistoryRepositoryProtocol
from ...domain.calculations.registry import derive_modelo_202_modality
from ...domain.deadlines import TaxpayerProfile
from ...domain.modelos import (
CalculationRevision,
CalculationRevisionCatalogueRepositoryProtocol,
CalculationRevisionState,
ModeloCode,
ModeloError,
ModeloRecord,
ModeloRecordCatalogue,
ModeloRecordCatalogueRepositoryProtocol,
ModeloRecordStatus,
VerificationReport,
VerificationReportCatalogueRepositoryProtocol,
WorkUnit,
WorkUnitCatalogueRepositoryProtocol,
)
from ..calculations import CalculationObservationRepository, CrossPeriodExpectedMemberSet
from ..workflow import WorkflowEngine, WorkflowRunRepository
from ._action_errors import (
CalculationRevisionNotFoundError,
CalculationRevisionStateError,
ModeloRecordNotFoundError,
VerificationReportNotFoundError,
WorkUnitNotFoundError,
)
from ._iva_wallet_gate import (
require_persisted_iva_compensation_decision_matches_revision as _require_iva_compensation_revision_match,
)
from ._ledger_evidence_gate import raise_if_deductible_vat_evidence_missing
from ._required_binding_gate import (
require_persisted_revision_required_bindings_resolved as _require_persisted_required_bindings_resolved,
)
from ._result_disposition_resolution import revision_is_refund_disposition
from ._revision_persistence import persist_filed_revision
from ._verification_actions import (
_cross_period_expected_member_sets_from_profile,
_require_cross_period_clean_state,
derive_taxpayer_files_economic_activity,
)
from ._workflow_gate import build_revision_workflow_engine as _build_revision_workflow_engine
from ._workflow_gate import run_revision_workflow_gate as _run_revision_workflow_gate
if TYPE_CHECKING:
from ..calculations import IvaWalletDecisionRepository
[docs]
class ModeloFilingEvidenceMissingError(ModeloError):
"""Raised when internal filing would seal deductible IVA without evidence."""
def _existing_vigente_filing_record(
catalogue: ModeloRecordCatalogue,
calculation_revision_id: str,
) -> ModeloRecord | None:
"""Return the current (VIGENTE) filing record for a filed revision, if any.
Used by the idempotent re-file no-op: a revision already in ``PRESENTADO``
state is the current filed answer, so its VIGENTE
:class:`ModeloRecord` is returned unchanged. A
``PRESENTADO`` revision must have exactly one VIGENTE record; ``None``
signals an inconsistent state the caller refuses rather than masking.
"""
for record in catalogue.values():
if record.calculation_revision_id == calculation_revision_id and record.status is ModeloRecordStatus.VIGENTE:
return record
return None
[docs]
def file_modelo_revision(
calculation_revision_id: str,
*,
actor: str,
workflow_profile: TaxpayerProfile,
notes: str | None = None,
refund_election: RefundElection = RefundElection.COMPENSAR,
work_unit_repository: WorkUnitCatalogueRepositoryProtocol | None = None,
calculation_repository: CalculationRevisionCatalogueRepositoryProtocol | None = None,
filing_repository: ModeloRecordCatalogueRepositoryProtocol | None = None,
verification_repository: VerificationReportCatalogueRepositoryProtocol | None = None,
bucket_event_repository: BucketEventHistoryRepositoryProtocol | None = None,
iva_compensation_decision_repository: IvaWalletDecisionRepository | None = None,
calculation_observation_repository: CalculationObservationRepository | None = None,
cross_period_expected_member_sets: Iterable[CrossPeriodExpectedMemberSet] = (),
workflow_engine: WorkflowEngine | None = None,
workflow_runs_dir: Path | None = None,
settings: Settings | None = None,
clock: datetime | None = None,
) -> ModeloRecord:
"""Mark a verified-complete revision as the current internal filed answer.
This is the application service behind ``aeat app modelo work file``. It is
a local state transition, not an AEAT presentation: the resulting
:class:`ModeloRecord` has ``aeat_accepted=False`` and no
external evidence.
Preconditions and state changes:
1. Verify the revision is in ``VERIFICADO_COMPLETO`` state.
2. Recheck profile readiness, persisted Modelo 303 IVA-wallet decision
compatibility, and cross-period clean state.
3. Run the :class:`WorkflowEngine` gate for the
revision's modelo and period.
4. Resolve the Modelo 303 refund/carry disposition from
:class:`RefundElection` and
:class:`TaxpayerProfile`.
5. If a prior current filing exists, mark its
:class:`ModeloRecord` as ``SUPERSEDIDO`` and its
prior :class:`CalculationRevision` as
``PRESENTADO_SUPERSEDIDO``.
6. Create the new filing record with status ``VIGENTE`` and transition the
target calculation revision from ``VERIFICADO_COMPLETO`` to
``PRESENTADO``.
7. Advance the work unit's ``filed_calculation_revision_id`` and
``current_filing_record_id`` pointers, emit ``MODELO_FILED``/
``MODELO_FILED_SUPERSEDED`` bucket events, and persist any local
``app_filing`` carry observation.
Args:
calculation_revision_id: The id of the verified-complete revision
to file.
actor: Operator identifier recorded in the filing record and audit
trail.
workflow_profile: The :class:`TaxpayerProfile`
used to evaluate :class:`WorkflowEngine`
gate conditions and cross-period clean-state applicability.
notes: Optional operator-supplied filing notes.
refund_election: The operator's per-filing Modelo 303 negative-result
disposition election. Defaults to ``COMPENSAR`` (carry the credit
forward). ``DEVOLVER`` requests the credit back as a refund and is
honoured only when the period is a lawful refund period (the year's
last filing period for a non-REDEME taxpayer; every period for REDEME).
An out-of-window ``DEVOLVER`` is refused.
work_unit_repository: Optional work-unit catalogue repository override.
calculation_repository: Optional calculation-revision catalogue
repository override.
filing_repository: Optional filing-record catalogue repository
override.
verification_repository: Optional verification-report catalogue
repository override used by the cross-period clean-state proof.
bucket_event_repository: Optional bucket-event history repository
override.
iva_compensation_decision_repository: Optional IVA wallet decision
repository override used to require that a persisted decision still
matches the target revision.
calculation_observation_repository: Optional calculation-observation
repository override used by the cross-period clean-state proof and
the non-official local carry projection.
cross_period_expected_member_sets: Optional expected grupo member
rosters used by the cross-period clean-state proof.
workflow_engine: Optional workflow engine override for the preflight
gate.
workflow_runs_dir: Optional workflow runs directory override.
settings: Optional settings override.
clock: Optional UTC timestamp override.
Returns:
The newly created local :class:`ModeloRecord` in
``VIGENTE`` status.
Raises:
CalculationRevisionNotFoundError: When the revision id is
absent.
CalculationRevisionStateError: When the revision is not in
``VERIFICADO_COMPLETO`` state.
WorkUnitNotFoundError: When the revision's parent work
unit cannot be loaded.
See Also:
:func:`~aeat.application.modelo._revision_persistence.persist_filed_revision`:
Performs the repository writes once all gates pass.
:func:`~aeat.application.modelo.import_external_filing_evidence`:
Creates official-evidence baselines for imported filings; use that
path when a :class:`ExternalEvidence` reference
must be carried.
:func:`~aeat.application.modelo._filed_revision_observation.persist_filed_revision_observation`:
Saves the non-official ``app_filing`` observation used by later
``previous_filing`` calculations.
:func:`~aeat.application.modelo.export_modelo_revision`:
Sibling local finish line that writes the fichero-BOE artefact
without requiring this internal file marker.
"""
wu_repo = work_unit_repository or WorkUnitCatalogueRepository()
cr_repo = calculation_repository or CalculationRevisionCatalogueRepository()
fr_repo = filing_repository or ModeloRecordCatalogueRepository()
vr_repo = verification_repository or VerificationReportCatalogueRepository()
obs_repo = calculation_observation_repository or CalculationObservationRepository()
bv_repo: BucketEventHistoryRepositoryProtocol = bucket_event_repository or BucketEventHistoryRepository()
_concrete_bv = bv_repo if isinstance(bv_repo, BucketEventHistoryRepository) else BucketEventHistoryRepository()
run_repo = WorkflowRunRepository(objects=_concrete_bv.secure_object_repository)
revisions = cr_repo.load()
target = revisions.get(calculation_revision_id)
if target is None:
raise CalculationRevisionNotFoundError(
translated_message="application.modelo.errors.calculation_revision_not_found",
context={"calculation_revision_id": calculation_revision_id},
)
if target.state is CalculationRevisionState.PRESENTADO:
# Idempotent re-file: this revision is already the current filed answer.
# A retry of a completed single-subject file returns the existing VIGENTE
# filing record as a clean no-op - no new filing record, no duplicate
# MODELO_FILED lifecycle event, and (filing is a local transition, never
# an AEAT submission per aeat-safety-legal-gates) no write/submit path is
# touched. Mirrors the verify-report content-pinned idempotency. The CLI
# surfaces the no-op as an info Notice. A PRESENTADO revision with no
# VIGENTE record is an inconsistent state, so it falls through to the
# hard refusal below rather than fabricating a record.
existing = _existing_vigente_filing_record(fr_repo.load(), calculation_revision_id)
if existing is not None:
return existing
if target.state is not CalculationRevisionState.VERIFICADO_COMPLETO:
raise CalculationRevisionStateError(
f"calculation revision {calculation_revision_id!r} is in state "
f"{target.state.value!r}; only VERIFICADO_COMPLETO revisions can be filed",
)
work_units = wu_repo.load()
work_unit = work_units.get(target.work_unit_id)
if work_unit is None:
raise WorkUnitNotFoundError(
f"calculation revision {calculation_revision_id!r} references missing work_unit_id={target.work_unit_id!r}",
)
_require_filing_preconditions(
work_unit=work_unit,
target=target,
workflow_profile=workflow_profile,
observation_repository=obs_repo,
filing_repository=fr_repo,
calculation_repository=cr_repo,
verification_repository=vr_repo,
iva_compensation_decision_repository=iva_compensation_decision_repository,
cross_period_expected_member_sets=cross_period_expected_member_sets,
)
now = clock or _utc_now()
gate_engine = workflow_engine or _build_revision_workflow_engine(
revision=target,
work_unit=work_unit,
profile=workflow_profile,
actor=actor.strip(),
clock=now,
settings=settings,
)
_run_revision_workflow_gate(
engine=gate_engine,
profile=workflow_profile,
work_unit=work_unit,
today=now.date(),
runs_dir=workflow_runs_dir,
run_repository=run_repo,
)
refunded = _filed_revision_refunded(
work_unit=work_unit,
target=target,
workflow_profile=workflow_profile,
refund_election=refund_election,
)
return persist_filed_revision(
target=target,
work_unit=work_unit,
work_units=work_units,
notes=notes,
actor=actor,
now=now,
calculation_repository=cr_repo,
filing_repository=fr_repo,
work_unit_repository=wu_repo,
bucket_event_repository=bv_repo,
calculation_observation_repository=obs_repo,
refunded=refunded,
taxpayer_nif=workflow_profile.tax_id,
)
def _filed_revision_refunded(
*,
work_unit: WorkUnit,
target: CalculationRevision,
workflow_profile: TaxpayerProfile,
refund_election: RefundElection,
) -> bool:
# Determine the filing disposition ONCE from the same shared resolver the
# export reads. A Modelo 303 devolución period returns the credit, so the
# cross-period carry persisted by filing must generate zero compensación
# rather than double-claim the refunded credit into the next period.
return revision_is_refund_disposition(
work_unit=work_unit,
revision=target,
workflow_profile=workflow_profile,
period=work_unit.period,
refund_election=refund_election,
)
def _require_filing_preconditions(
*,
work_unit: WorkUnit,
target: CalculationRevision,
workflow_profile: TaxpayerProfile,
observation_repository: CalculationObservationRepository,
filing_repository: ModeloRecordCatalogueRepositoryProtocol,
calculation_repository: CalculationRevisionCatalogueRepositoryProtocol,
verification_repository: VerificationReportCatalogueRepositoryProtocol,
iva_compensation_decision_repository: IvaWalletDecisionRepository | None,
cross_period_expected_member_sets: Iterable[CrossPeriodExpectedMemberSet],
) -> None:
from ._profile_readiness_gate import require_profile_ready_for_work_unit
raise_if_deductible_vat_evidence_missing(
target,
error_type=ModeloFilingEvidenceMissingError,
surface="internal filing",
suggestion=(
"aeat app ledger evidence add PATH; "
"aeat app ledger attach TRANSACTION_ID --purchase-invoice-evidence-id EVIDENCE_ID"
),
)
require_profile_ready_for_work_unit(work_unit)
_require_persisted_required_bindings_resolved(
work_unit=work_unit,
revision=target,
action="file",
)
iva_compensation_decision = _require_iva_compensation_revision_match(
work_unit,
target,
repository=iva_compensation_decision_repository,
)
_require_cross_period_clean_state(
work_unit,
observation_repository=observation_repository,
filing_repository=filing_repository,
calculation_repository=calculation_repository,
verification_repository=verification_repository,
iva_compensation_decision=iva_compensation_decision,
expected_member_sets=_cross_period_expected_member_sets_from_profile(
workflow_profile,
cross_period_expected_member_sets,
),
taxpayer_tax_id=workflow_profile.tax_id,
activity_start_date=workflow_profile.activity_start_date,
modelo_202_modality=derive_modelo_202_modality(workflow_profile).modality,
taxpayer_files_economic_activity=derive_taxpayer_files_economic_activity(workflow_profile),
workflow_profile=workflow_profile,
target_revision=target,
)
[docs]
def list_filing_records(
*,
bucket_id: str | None = None,
modelo: str | ModeloCode | None = None,
include_superseded: bool = False,
filing_repository: ModeloRecordCatalogueRepositoryProtocol | None = None,
) -> tuple[ModeloRecord, ...]:
"""List :class:`ModeloRecord` rows, optionally filtered to a bucket and modelo.
Superseded records are excluded unless ``include_superseded``
is true. Results are sorted by ``(bucket_id, filing_year,
modelo, period, filed_at)``.
"""
fr_repo = filing_repository or ModeloRecordCatalogueRepository()
catalogue = fr_repo.load()
modelo_code = ModeloCode(str(modelo)) if modelo is not None else None
records = tuple(
record
for record in catalogue.values()
if (bucket_id is None or record.bucket_id == bucket_id)
and (modelo_code is None or record.modelo == modelo_code)
and (include_superseded or record.status is ModeloRecordStatus.VIGENTE)
)
return tuple(
sorted(
records,
key=lambda r: (r.bucket_id, r.filing_year, str(r.modelo), r.period.registry_token, r.filed_at),
),
)
[docs]
def get_filing_record(
filing_record_id: str,
*,
filing_repository: ModeloRecordCatalogueRepositoryProtocol | None = None,
) -> ModeloRecord:
"""Return the :class:`ModeloRecord` for the given id, or raise."""
fr_repo = filing_repository or ModeloRecordCatalogueRepository()
catalogue = fr_repo.load()
record = catalogue.get(filing_record_id)
if record is None:
raise ModeloRecordNotFoundError(
translated_message="application.modelo.errors.filing_record_not_found",
context={"filing_record_id": filing_record_id},
)
return record
[docs]
def list_verification_reports(
*,
calculation_revision_id: str | None = None,
verification_repository: VerificationReportCatalogueRepositoryProtocol | None = None,
) -> tuple[VerificationReport, ...]:
"""List :class:`VerificationReport` records.
Optionally filtered to one
:class:`CalculationRevision`. The
:class:`VerificationReportCatalogueRepositoryProtocol`
supplies the persisted report catalogue. Results are sorted by
``(calculation_revision_id, run_at)``.
"""
vr_repo = verification_repository or VerificationReportCatalogueRepository()
catalogue = vr_repo.load()
reports = tuple(
r
for r in catalogue.values()
if calculation_revision_id is None or r.calculation_revision_id == calculation_revision_id
)
return tuple(sorted(reports, key=lambda r: (r.calculation_revision_id, r.run_at)))
[docs]
def get_verification_report(
verification_report_id: str,
*,
verification_repository: VerificationReportCatalogueRepositoryProtocol | None = None,
) -> VerificationReport:
"""Return one :class:`VerificationReport` by id, or raise.
The optional
:class:`VerificationReportCatalogueRepositoryProtocol`
supplies the persisted report catalogue for tests or alternate storage
boundaries.
"""
vr_repo = verification_repository or VerificationReportCatalogueRepository()
catalogue = vr_repo.load()
report = catalogue.get(verification_report_id)
if report is None:
raise VerificationReportNotFoundError(
translated_message="application.modelo.errors.verification_report_not_found",
context={"verification_report_id": verification_report_id},
)
return report