Manage daily reports by date and refresh monthly reports after deletion

arr_ingestion/postgres.py

@@ -8,7 +8,7 @@ import os
 import re
 import time
 from dataclasses import dataclass
-from datetime import date
+from datetime import date, timedelta
 from decimal import Decimal, InvalidOperation
 from typing import Any, Callable, Dict, Mapping, Optional, Sequence, Tuple
 
@@ -215,6 +215,119 @@ class PostgresIngestionRepository(IngestionRepository):
                 "database connection failed",
             ) from None
 
+    def retire_daily_job(
+        self, job_id: str, business_date: date, actor_username: str
+    ) -> Dict[str, Any]:
+        """Retire one exact run, retaining all source, facts and audit lineage."""
+        if not isinstance(job_id, str) or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,127}", job_id) is None:
+            raise IngestionError("DAILY_JOB_NOT_FOUND", "日报记录不存在")
+        if type(business_date) is not date:
+            raise IngestionError("DAILY_DATE_CONFLICT", "报表日期无效，请刷新后重试")
+        actor = self._review_actor(actor_username)
+        return self._run_transaction(
+            lambda cursor: self._retire_daily_job(cursor, job_id, business_date, actor),
+            "daily report could not be retired",
+        )
+
+    @staticmethod
+    def _retirement_receipt(job_id: str, business_date: date, current_removed: bool) -> Dict[str, Any]:
+        return {
+            "job_id": job_id,
+            "business_date": business_date.isoformat(),
+            "is_retired": True,
+            "current_removed": current_removed,
+            "monthly_refresh_required": current_removed,
+        }
+
+    def _retire_daily_job(self, cursor: Any, job_id: str, business_date: date, actor: str) -> Dict[str, Any]:
+        cursor.execute(
+            """SELECT id, business_date, run_status
+               FROM ingestion.processing_runs
+               WHERE run_key = %s AND pipeline_type = 'opera_daily'
+               FOR UPDATE""", (job_id,),
+        )
+        run = cursor.fetchone()
+        if run is None:
+            raise IngestionError("DAILY_JOB_NOT_FOUND", "日报记录不存在")
+        run_id, stored_date, run_status = int(run[0]), run[1], str(run[2])
+        if stored_date != business_date:
+            raise IngestionError("DAILY_DATE_CONFLICT", "报表日期已变化，请刷新后重试")
+        cursor.execute(
+            """SELECT current_removed FROM ingestion.daily_run_retirements
+               WHERE processing_run_id = %s""", (run_id,),
+        )
+        retired = cursor.fetchone()
+        if retired is not None:
+            return self._retirement_receipt(job_id, business_date, bool(retired[0]))
+        if run_status not in {"accepted", "rejected", "failed", "cancelled", "awaiting_review"}:
+            raise IngestionError("DAILY_JOB_BUSY", "任务正在处理，请完成后再删除")
+        cursor.execute(
+            "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
+            (f"finance-daily:{business_date.isoformat()}",),
+        )
+        cursor.execute(
+            """SELECT id, version_status FROM finance.daily_versions
+               WHERE processing_run_id = %s FOR UPDATE""", (run_id,),
+        )
+        versions = cursor.fetchall()
+        if len(versions) > 1:
+            raise IngestionError("DATABASE_STATE_INVALID", "日报版本关联无效")
+        version_id = int(versions[0][0]) if versions else None
+        cursor.execute(
+            """SELECT daily_version_id FROM finance.current_daily_versions
+               WHERE business_date = %s FOR UPDATE""", (business_date,),
+        )
+        current = cursor.fetchone()
+        current_removed = version_id is not None and current is not None and int(current[0]) == version_id
+        if run_status == "awaiting_review":
+            cursor.execute(
+                """SELECT case_key, revision, case_status FROM ingestion.daily_review_cases
+                   WHERE processing_run_id = %s FOR UPDATE""", (run_id,),
+            )
+            case = cursor.fetchone()
+            if case is None or str(case[2]) not in {"open", "generation_failed"}:
+                raise IngestionError("DAILY_JOB_BUSY", "任务正在生成日报，请完成后再删除")
+            self._cancel_price_review(cursor, job_id, str(case[0]), int(case[1]), actor)
+        if current_removed:
+            period_start = business_date.replace(day=1)
+            period_end = (period_start.replace(day=28) + timedelta(days=4)).replace(day=1) - timedelta(days=1)
+            cursor.execute(
+                "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
+                (f"monthly_channel:{period_start.isoformat()}:{period_end.isoformat()}:*",),
+            )
+            cursor.execute(
+                """DELETE FROM finance.current_daily_versions
+                   WHERE business_date = %s AND daily_version_id = %s""", (business_date, version_id),
+            )
+            if cursor.rowcount != 1:
+                raise IngestionError("DAILY_DATE_CONFLICT", "该日有效报表已变化，请刷新后重试")
+            cursor.execute(
+                """UPDATE finance.daily_versions SET version_status = 'superseded', superseded_at = now()
+                   WHERE id = %s AND version_status = 'active'""", (version_id,),
+            )
+            if cursor.rowcount != 1:
+                raise IngestionError("DATABASE_STATE_INVALID", "该日有效报表状态无效")
+            # Withdraw downloadable snapshots immediately, before asynchronous rebuilding.
+            cursor.execute(
+                """UPDATE reporting.monthly_runs
+                   SET report_status = 'withdrawn', withdrawn_at = now(), updated_at = now()
+                   WHERE period_start = %s AND report_status IN ('reserved', 'active', 'superseded', 'failed')""",
+                (period_start,),
+            )
+            self._insert_outbox(
+                cursor, f"daily-version:{version_id}:retired", "daily_version", version_id,
+                "arr.daily_scope_changed",
+                {"daily_version_id": version_id, "business_date": business_date.isoformat(),
+                 "period_start": period_start.isoformat(), "action": "delete"},
+            )
+        cursor.execute(
+            """INSERT INTO ingestion.daily_run_retirements
+               (processing_run_id, business_date, daily_version_id, current_removed, actor_username)
+               VALUES (%s, %s, %s, %s, %s)""",
+            (run_id, business_date, version_id, current_removed, actor),
+        )
+        return self._retirement_receipt(job_id, business_date, current_removed)
+
     def _run_transaction(
         self,
         operation: Callable[[Any], Any],
@@ -401,6 +514,44 @@ class PostgresIngestionRepository(IngestionRepository):
                                      "ARR processing requires migration 021")
         self._run_transaction(check, "ARR PM schema could not be checked")
 
+    def assert_daily_lifecycle_schema(self) -> None:
+        """Do not expose date history/deletion with an incomplete 022 upgrade."""
+        def check(cursor):
+            cursor.execute("""SELECT
+                to_regclass('ingestion.daily_run_retirements') IS NOT NULL,
+                (SELECT count(*) = 7 FROM information_schema.columns
+                 WHERE table_schema = 'ingestion' AND table_name = 'daily_run_retirements'
+                   AND column_name IN ('id','processing_run_id','business_date','daily_version_id',
+                                       'current_removed','actor_username','retired_at')),
+                EXISTS (SELECT 1 FROM information_schema.columns
+                    WHERE table_schema = 'reporting' AND table_name = 'monthly_runs'
+                      AND column_name = 'withdrawn_at' AND data_type = 'timestamp with time zone'),
+                EXISTS (SELECT 1 FROM pg_constraint
+                    WHERE conrelid = 'reporting.monthly_runs'::regclass
+                      AND conname = 'monthly_runs_status_shape'
+                      AND pg_get_constraintdef(oid) LIKE '%withdrawn_at%'),
+                EXISTS (SELECT 1 FROM pg_index AS index_row
+                    JOIN pg_class AS index_name ON index_name.oid = index_row.indexrelid
+                    WHERE index_row.indrelid = 'reporting.monthly_runs'::regclass
+                      AND index_name.relname = 'monthly_runs_live_snapshot_unique'
+                      AND index_row.indisunique AND index_row.indisvalid
+                      AND pg_get_indexdef(index_row.indexrelid) LIKE '%period_start, source_snapshot_sha256%'
+                      AND pg_get_expr(index_row.indpred,index_row.indrelid) LIKE '%report_status <>%withdrawn%'),
+                EXISTS (SELECT 1 FROM pg_trigger
+                    WHERE tgrelid = to_regclass('ingestion.daily_run_retirements')
+                      AND tgname = 'daily_run_retirements_immutable' AND tgenabled = 'O'),
+                EXISTS (SELECT 1 FROM pg_proc
+                    WHERE oid = to_regprocedure('finance.validate_current_daily_version()')
+                      AND prosrc LIKE '%daily_run_retirements%'),
+                EXISTS (SELECT 1 FROM pg_proc
+                    WHERE oid = to_regprocedure('reporting.protect_published_monthly_child()')
+                      AND prosrc LIKE '%withdrawn%')""")
+            row = cursor.fetchone()
+            if not row or len(row) != 8 or any(value is not True for value in row):
+                raise IngestionError("DATABASE_MIGRATION_MISSING",
+                                     "ARR daily report lifecycle requires migration 022")
+        self._run_transaction(check, "ARR daily lifecycle schema could not be checked")
+
     def register_job(self, registration: JobRegistration) -> None:
         if (
             registration.source.role not in {"source_xml", "source_data"}
@@ -1280,6 +1431,17 @@ class PostgresIngestionRepository(IngestionRepository):
                     "processing job identity conflicts",
                 )
             processing_run_id = int(existing[0])
+            # Pre-022 ingestion remains compatible; once retirement is enabled,
+            # an old run identity can never be reused to undo a user's deletion.
+            cursor.execute("SELECT to_regclass('ingestion.daily_run_retirements')")
+            retirement_table = cursor.fetchone()
+            if retirement_table is not None and retirement_table[0] is not None:
+                cursor.execute(
+                    "SELECT 1 FROM ingestion.daily_run_retirements WHERE processing_run_id = %s",
+                    (processing_run_id,),
+                )
+                if cursor.fetchone() is not None:
+                    raise IngestionError("JOB_RETIRED", "该报表记录已删除，请重新提交新任务")
             cursor.execute(
                 """
                 SELECT idempotency_key
@@ -2275,6 +2437,14 @@ class PostgresIngestionRepository(IngestionRepository):
         business_date: date,
         daily_version_id: int,
     ) -> None:
+        # Publication and retirement share this month lock. A successful daily
+        # replacement cannot slip between a monthly source check and activation.
+        period_start = business_date.replace(day=1)
+        period_end = (period_start.replace(day=28) + timedelta(days=4)).replace(day=1) - timedelta(days=1)
+        cursor.execute(
+            "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
+            (f"monthly_channel:{period_start.isoformat()}:{period_end.isoformat()}:*",),
+        )
         cursor.execute(
             """
             SELECT daily_version_id