"""PostgreSQL implementation of validate-before-write ARR daily ingestion.""" from __future__ import annotations import hashlib import json import os import re import time from dataclasses import dataclass from datetime import date from decimal import Decimal, InvalidOperation from typing import Any, Callable, Dict, Mapping, Optional, Sequence, Tuple from arr_ingestion.contracts import ArtifactRef, IngestionError, SHA256_RE from arr_ingestion.repository import ( IngestionOutcome, IngestionRepository, JobRegistration, ReviewGeneration, ) from arr_ingestion.validation import VerifiedDelivery TARGET_DATABASE = "booking_test" DATABASE_ENV = "ARR_DATABASE_URL" ARTIFACT_STORAGE_PROVIDER = "oss" ARTIFACT_BUCKET_ALIAS = "arr-private" TRANSIENT_SQLSTATES = frozenset({"23505", "40001", "40P01", "55P03"}) MAX_TRANSACTION_ATTEMPTS = 4 @dataclass(frozen=True) class DatabaseConfig: dsn: str @classmethod def from_environment(cls) -> "DatabaseConfig": dsn = os.environ.get(DATABASE_ENV, "").strip() if not dsn: raise IngestionError( "DATABASE_CONFIG_MISSING", f"{DATABASE_ENV} is required", ) return cls(dsn=dsn) def _default_connect(dsn: str) -> Any: try: import psycopg # type: ignore[import-not-found] except ImportError: raise IngestionError( "DATABASE_DRIVER_UNAVAILABLE", "PostgreSQL driver is unavailable; install requirements-arr-ingestion.txt", ) from None try: return psycopg.connect(dsn, autocommit=False) except Exception: raise IngestionError( "DATABASE_UNAVAILABLE", "database connection failed", ) from None def _decimal_or_none(value: Any) -> Optional[Decimal]: if value is None: return None try: number = Decimal(str(value)) except (InvalidOperation, ValueError): raise IngestionError( "RESULT_CONTRACT_INVALID", "numeric result value is invalid", ) from None if not number.is_finite(): raise IngestionError( "RESULT_CONTRACT_INVALID", "numeric result value is invalid", ) return number def _date_or_none(value: Any) -> Optional[date]: if value is None: return None if isinstance(value, date): return value if isinstance(value, str): try: return date.fromisoformat(value) except ValueError: pass raise IngestionError( "RESULT_CONTRACT_INVALID", "date result value is invalid", ) MANUAL_PRICE_RE = re.compile(r"^(?:0|[1-9][0-9]{0,15})$") def _manual_price(value: Any) -> Decimal: if not isinstance(value, str) or MANUAL_PRICE_RE.fullmatch(value) is None: raise IngestionError( "REVIEW_PRICE_INVALID", "人工价格必须是非负整数字符串", ) try: amount = Decimal(value) except InvalidOperation: raise IngestionError("REVIEW_PRICE_INVALID", "人工价格格式无效") from None if amount < 0: raise IngestionError("REVIEW_PRICE_INVALID", "人工价格不得为负数") return amount.quantize(Decimal("0.01")) def _review_price_text(value: Any) -> str: amount = _decimal_or_none(value) if amount is None or amount < 0 or amount != amount.to_integral_value(): raise IngestionError("DATABASE_STATE_INVALID", "人工价格不是整数") return format(amount.quantize(Decimal("1")), "f") def _canonical_amount(value: Any) -> str: amount = _decimal_or_none(value) if amount is None or amount < 0: raise IngestionError("RESULT_CONTRACT_INVALID", "复核Opera价格无效") return format(amount.normalize(), "f") def _canonical_json_bytes(value: Mapping[str, Any]) -> bytes: return json.dumps( value, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") def _safe_failure_code(payload: Mapping[str, Any]) -> str: errors = payload.get("errors") if isinstance(errors, list) and errors and isinstance(errors[0], Mapping): value = errors[0].get("code") if isinstance(value, str) and re.fullmatch( r"[A-Z][A-Z0-9_]{0,63}", value, ): return value return "PROCESSING_FAILED" def _is_transient_database_error(error: Exception) -> bool: sqlstate = getattr(error, "sqlstate", None) if not isinstance(sqlstate, str): diagnostic = getattr(error, "diag", None) sqlstate = getattr(diagnostic, "sqlstate", None) return isinstance(sqlstate, str) and sqlstate in TRANSIENT_SQLSTATES def _outcome_counts(payload: Mapping[str, Any]) -> Tuple[int, int, int, int, int]: values = payload.get("outcome_counts") if not isinstance(values, Mapping): raise IngestionError( "RESULT_CONTRACT_INVALID", "structured outcome counts are invalid", ) try: counts = tuple( int(values[name]) for name in ( "retained", "excluded_rate_code", "duplicate", "validation_failed", "price_unmatched", ) ) except (KeyError, TypeError, ValueError): raise IngestionError( "RESULT_CONTRACT_INVALID", "structured outcome counts are invalid", ) from None if any(value < 0 for value in counts): raise IngestionError( "RESULT_CONTRACT_INVALID", "structured outcome counts are invalid", ) return counts # type: ignore[return-value] class PostgresIngestionRepository(IngestionRepository): def __init__( self, config: DatabaseConfig, connect: Optional[Callable[[str], Any]] = None, ) -> None: self._config = config self._connect = connect or _default_connect def _open(self) -> Any: try: return self._connect(self._config.dsn) except IngestionError: raise except Exception: raise IngestionError( "DATABASE_UNAVAILABLE", "database connection failed", ) from None def _run_transaction( self, operation: Callable[[Any], Any], failure_message: str, ) -> Any: for attempt_index in range(MAX_TRANSACTION_ATTEMPTS): connection = self._open() retry = False try: with connection.transaction(): with connection.cursor() as cursor: self._begin(cursor) return operation(cursor) except IngestionError: raise except Exception as error: retry = ( attempt_index + 1 < MAX_TRANSACTION_ATTEMPTS and _is_transient_database_error(error) ) if not retry: raise IngestionError( "DATABASE_WRITE_FAILED", failure_message, ) from None finally: connection.close() if retry: time.sleep(0.02 * (2**attempt_index)) raise IngestionError( "DATABASE_WRITE_FAILED", failure_message, ) @staticmethod def _begin(cursor: Any) -> None: cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE") cursor.execute("SET LOCAL lock_timeout = '10s'") cursor.execute("SET LOCAL statement_timeout = '30s'") cursor.execute("SELECT current_database()") row = cursor.fetchone() if not row or row[0] != TARGET_DATABASE: raise IngestionError( "DATABASE_TARGET_INVALID", "database target is not booking_test", ) @staticmethod def _ensure_artifact(cursor: Any, reference: ArtifactRef) -> int: cursor.execute( """ SELECT id, artifact_kind, original_filename, sha256, byte_size, mime_type FROM ingestion.artifacts WHERE storage_provider = %s AND bucket_alias = %s AND object_key = %s AND object_version_id IS NULL FOR SHARE """, ( ARTIFACT_STORAGE_PROVIDER, ARTIFACT_BUCKET_ALIAS, reference.object_key, ), ) row = cursor.fetchone() if row: if ( str(row[1]) != reference.file_kind or str(row[2]) != reference.original_filename or str(row[3]) != reference.sha256 or int(row[4]) != reference.byte_size or (row[5] or "") != reference.mime_type ): raise IngestionError( "ARTIFACT_CONFLICT", "stored artifact identity conflicts", ) return int(row[0]) cursor.execute( """ INSERT INTO ingestion.artifacts ( artifact_kind, storage_provider, bucket_alias, object_key, original_filename, sha256, byte_size, mime_type ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s) RETURNING id """, ( reference.file_kind, ARTIFACT_STORAGE_PROVIDER, ARTIFACT_BUCKET_ALIAS, reference.object_key, reference.original_filename, reference.sha256, reference.byte_size, reference.mime_type, ), ) return int(cursor.fetchone()[0]) def register_job(self, registration: JobRegistration) -> None: if ( registration.source.role != "source_xml" or registration.attempt_no < 1 or not SHA256_RE.fullmatch(registration.idempotency_key) ): raise IngestionError( "JOB_INVALID", "processing job registration is invalid", ) self._run_transaction( lambda cursor: self._register_job(cursor, registration), "processing job could not be stored", ) def mark_running(self, job_id: str, attempt_no: int) -> None: self._run_transaction( lambda cursor: self._mark_running(cursor, job_id, attempt_no), "processing job could not be started", ) def _mark_running(self, cursor: Any, job_id: str, attempt_no: int) -> None: run_id, run_status, attempt_id, attempt_status = self._lock_attempt( cursor, job_id, attempt_no ) if run_status in {"accepted", "rejected", "failed", "cancelled"}: raise IngestionError("JOB_TERMINAL", "processing job is already terminal") if attempt_status in {"succeeded", "failed", "cancelled"}: raise IngestionError("JOB_TERMINAL", "processing attempt is already terminal") cursor.execute( """ UPDATE ingestion.processing_attempts SET attempt_status = 'running', started_at = COALESCE(started_at, now()) WHERE id = %s """, (attempt_id,), ) cursor.execute( """ UPDATE ingestion.processing_runs SET run_status = 'running', failure_code = NULL, failure_message = NULL, updated_at = now() WHERE id = %s """, (run_id,), ) def record_failure(self, job_id: str, attempt_no: int, failure_code: str) -> None: if not isinstance(failure_code, str) or re.fullmatch( r"[A-Z][A-Z0-9_]{0,63}", failure_code ) is None: failure_code = "PROCESSING_FAILED" self._run_transaction( lambda cursor: self._record_runtime_failure( cursor, job_id, attempt_no, failure_code ), "processing failure could not be stored", ) def _record_runtime_failure( self, cursor: Any, job_id: str, attempt_no: int, failure_code: str, ) -> None: run_id, run_status, attempt_id, _attempt_status = self._lock_attempt( cursor, job_id, attempt_no ) if run_status == "failed": return if run_status in {"accepted", "rejected", "cancelled"}: raise IngestionError("JOB_TERMINAL", "processing job is already terminal") cursor.execute( """ UPDATE ingestion.processing_attempts SET attempt_status = 'failed', failure_code = %s, failure_message = 'programmatic processing failed', finished_at = now() WHERE id = %s """, (failure_code, attempt_id), ) cursor.execute( """ SELECT id, revision FROM ingestion.daily_review_cases WHERE processing_run_id = %s AND case_status = 'processing' FOR UPDATE """, (run_id,), ) active_review = cursor.fetchone() if active_review: review_case_id = int(active_review[0]) cursor.execute( """ UPDATE ingestion.daily_review_cases SET case_status = 'failed', failure_code = %s, failure_message = 'deterministic final generation failed', updated_at = now() WHERE id = %s """, (failure_code, review_case_id), ) self._review_event( cursor, review_case_id, "PRICE_REVIEW_FAILED", "system:runtime", int(active_review[1]), ) cursor.execute( """ UPDATE ingestion.processing_runs SET run_status = 'failed', failure_code = %s, failure_message = 'programmatic processing failed', updated_at = now(), finished_at = now() WHERE id = %s """, (failure_code, run_id), ) self._insert_outbox( cursor, f"processing-run:{run_id}:failed", "processing_run", run_id, "arr.processing_failed", { "job_id": job_id, "business_date": None, "daily_version_id": None, "failure_code": failure_code, }, ) @staticmethod def _review_actor(value: str) -> str: if not isinstance(value, str) or not value.strip() or len(value) > 255: raise IngestionError("REVIEW_ACTOR_INVALID", "复核会话用户无效") return value.strip() @staticmethod def _review_number(value: Any) -> int | float: amount = _decimal_or_none(value) if amount is None: raise IngestionError("DATABASE_STATE_INVALID", "复核金额为空") return int(amount) if amount == amount.to_integral_value() else float(amount) @staticmethod def _review_event( cursor: Any, review_case_id: int, event_type: str, actor_username: str, revision: int, *, review_item_id: Optional[int] = None, previous_real_price: Optional[Decimal] = None, new_real_price: Optional[Decimal] = None, ) -> None: cursor.execute( """ INSERT INTO ingestion.daily_review_events ( review_case_id, review_item_id, event_type, actor_username, previous_real_price, new_real_price, revision ) VALUES (%s, %s, %s, %s, %s, %s, %s) """, ( review_case_id, review_item_id, event_type, actor_username, previous_real_price, new_real_price, revision, ), ) @staticmethod def _lock_review_case(cursor: Any, job_id: str, case_key: str) -> Tuple[Any, ...]: cursor.execute( """ SELECT review.id, review.case_key, review.case_status, review.revision, review.business_date, review.processor_version, review.rule_set_sha256, review.source_sha256, review.manual_override_json, review.manual_override_sha256, review.manual_override_artifact_id, run.id, run.run_status, run.requested_processor_version, run.requested_rule_set_sha256 FROM ingestion.daily_review_cases AS review JOIN ingestion.processing_runs AS run ON run.id = review.processing_run_id WHERE run.run_key = %s AND review.case_key = %s FOR UPDATE OF review, run """, (job_id, case_key), ) row = cursor.fetchone() if not row: raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在") return tuple(row) def get_price_review(self, job_id: str, limit: int, offset: int) -> Dict[str, Any]: if ( not isinstance(limit, int) or isinstance(limit, bool) or not 1 <= limit <= 200 or not isinstance(offset, int) or isinstance(offset, bool) or offset < 0 ): raise IngestionError("REVIEW_PAGE_INVALID", "复核分页参数无效") return self._run_transaction( lambda cursor: self._get_price_review(cursor, job_id, limit, offset), "review case could not be read", ) def _get_price_review( self, cursor: Any, job_id: str, limit: int, offset: int ) -> Dict[str, Any]: cursor.execute( """ SELECT review.id, review.case_key, review.case_status, review.revision, review.business_date, count(item.id) AS total_items, count(item.id) FILTER (WHERE item.real_price IS NOT NULL) AS completed_items FROM ingestion.daily_review_cases AS review JOIN ingestion.processing_runs AS run ON run.id = review.processing_run_id LEFT JOIN ingestion.daily_review_items AS item ON item.review_case_id = review.id WHERE run.run_key = %s GROUP BY review.id, review.case_key, review.case_status, review.revision, review.business_date """, (job_id,), ) case = cursor.fetchone() if not case: raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在") review_case_id = int(case[0]) total = int(case[5]) completed = int(case[6]) cursor.execute( """ SELECT id, company_key, rate_code, effective_rate_amount, candidate_prices, affected_records, affected_rooms, affected_room_nights, real_price, revision FROM ingestion.daily_review_items WHERE review_case_id = %s ORDER BY company_key, rate_code, effective_rate_amount, id LIMIT %s OFFSET %s """, (review_case_id, limit, offset), ) items = [] for row in cursor.fetchall(): real_price = row[8] items.append( { "item_id": int(row[0]), "company_key": str(row[1]), "rate_code": str(row[2]), "effective_rate_amount": self._review_number(row[3]), "candidate_prices": row[4] if isinstance(row[4], list) else [], "affected_records": int(row[5]), "affected_rooms": int(row[6]), "affected_room_nights": int(row[7]), "real_price": (_review_price_text(real_price) if real_price is not None else None), "revision": int(row[9]), } ) return { "case_id": str(case[1]), "case_status": str(case[2]), "revision": int(case[3]), "business_date": _date_or_none(case[4]).isoformat() if case[4] is not None else None, "completed_items": completed, "total_items": total, "items": items, "pagination": { "limit": limit, "offset": offset, "total": total, "has_next": offset + len(items) < total, }, } def update_price_review_item( self, job_id: str, item_id: int, case_id: str, revision: int, real_price: str, actor_username: str, ) -> Dict[str, Any]: if not isinstance(item_id, int) or isinstance(item_id, bool) or item_id < 1: raise IngestionError("REVIEW_ITEM_NOT_FOUND", "待人工处理项不存在") if not isinstance(revision, int) or isinstance(revision, bool) or revision < 0: raise IngestionError("REVIEW_REVISION_CONFLICT", "复核版本无效") price = _manual_price(real_price) actor = self._review_actor(actor_username) return self._run_transaction( lambda cursor: self._update_price_review_item( cursor, job_id, item_id, case_id, revision, price, actor ), "review item could not be updated", ) def _update_price_review_item( self, cursor: Any, job_id: str, item_id: int, case_id: str, revision: int, price: Decimal, actor_username: str, ) -> Dict[str, Any]: case = self._lock_review_case(cursor, job_id, case_id) review_case_id, _key, status, current_revision = int(case[0]), str(case[1]), str(case[2]), int(case[3]) if status != "open": raise IngestionError("REVIEW_IMMUTABLE", "复核清单已冻结，不能修改") if revision != current_revision: raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新") cursor.execute( """ SELECT real_price FROM ingestion.daily_review_items WHERE id = %s AND review_case_id = %s FOR UPDATE """, (item_id, review_case_id), ) item = cursor.fetchone() if not item: raise IngestionError("REVIEW_ITEM_NOT_FOUND", "待人工处理项不存在") previous = _decimal_or_none(item[0]) next_revision = current_revision + 1 cursor.execute( """ UPDATE ingestion.daily_review_items SET real_price = %s, revision = %s, updated_at = now() WHERE id = %s """, (price, next_revision, item_id), ) cursor.execute( """ UPDATE ingestion.daily_review_cases SET revision = %s, updated_at = now() WHERE id = %s """, (next_revision, review_case_id), ) self._review_event( cursor, review_case_id, "PRICE_REVIEW_UPDATED", actor_username, next_revision, review_item_id=item_id, previous_real_price=previous, new_real_price=price, ) return self._get_price_review(cursor, job_id, 200, 0) def begin_price_review_generation( self, job_id: str, case_id: str, revision: int, actor_username: str, processor_version: str, rule_set_sha256: str, idempotency_key: str, ) -> ReviewGeneration: if not isinstance(revision, int) or isinstance(revision, bool) or revision < 0: raise IngestionError("REVIEW_REVISION_CONFLICT", "复核版本无效") if not SHA256_RE.fullmatch(idempotency_key): raise IngestionError("JOB_INVALID", "复核生成幂等键无效") actor = self._review_actor(actor_username) return self._run_transaction( lambda cursor: self._begin_price_review_generation( cursor, job_id, case_id, revision, actor, processor_version, rule_set_sha256, idempotency_key, ), "review final generation could not be started", ) def _begin_price_review_generation( self, cursor: Any, job_id: str, case_id: str, revision: int, actor_username: str, processor_version: str, rule_set_sha256: str, idempotency_key: str, ) -> ReviewGeneration: case = self._lock_review_case(cursor, job_id, case_id) ( review_case_id, case_key, status, current_revision, business_date, case_processor_version, case_rule_set_sha256, source_sha256, stored_manifest, stored_manifest_sha256, _stored_manifest_artifact_id, processing_run_id, _run_status, requested_processor_version, requested_rule_set_sha256, ) = case review_case_id = int(review_case_id) business_date = _date_or_none(business_date) assert business_date is not None if ( str(case_processor_version) != processor_version or str(case_rule_set_sha256) != rule_set_sha256 or requested_processor_version != processor_version or str(requested_rule_set_sha256) != rule_set_sha256 ): raise IngestionError( "REVIEW_RULESET_CHANGED", "处理器或规则已变更，请取消后重新上传", ) if status == "completed": cursor.execute( "SELECT id, version_no FROM finance.daily_versions WHERE review_case_id = %s", (review_case_id,), ) version = cursor.fetchone() return ReviewGeneration( "completed", job_id, str(case_key), None, business_date, None, str(stored_manifest_sha256) if stored_manifest_sha256 else None, int(version[0]) if version else None, int(version[1]) if version else None, ) if status == "processing": cursor.execute( """ SELECT attempt_no FROM ingestion.processing_attempts WHERE processing_run_id = %s ORDER BY attempt_no DESC LIMIT 1 """, (int(processing_run_id),), ) attempt = cursor.fetchone() return ReviewGeneration( "processing", job_id, str(case_key), int(attempt[0]) if attempt else None, business_date, None, str(stored_manifest_sha256) if stored_manifest_sha256 else None, ) if status not in {"open", "generation_failed"}: raise IngestionError("REVIEW_NOT_OPEN", "复核任务当前不能确认生成") if int(current_revision) != revision: raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新") cursor.execute( """ SELECT company_key, rate_code, effective_rate_amount, real_price FROM ingestion.daily_review_items WHERE review_case_id = %s ORDER BY company_key, rate_code, effective_rate_amount, id FOR UPDATE """, (review_case_id,), ) items = cursor.fetchall() if not items or any(row[3] is None for row in items): raise IngestionError("REVIEW_INCOMPLETE", "请先填写全部缺价项") if stored_manifest is None: issue_rows = [ { "company_key": str(row[0]), "rate_code": str(row[1]), "effective_rate_amount": _canonical_amount(row[2]), } for row in items ] overrides = [ { **issue, "real_price": format(Decimal(str(row[3])), ".2f"), } for issue, row in zip(issue_rows, items) ] manifest_payload: Dict[str, Any] = { "review_version": "1.0", "review_case_id": str(case_key), "job_id": job_id, "source_sha256": str(source_sha256), "business_date": business_date.isoformat(), "processor_version": processor_version, "rule_set_sha256": rule_set_sha256, "issues": issue_rows, "overrides": overrides, } manifest_bytes = _canonical_json_bytes(manifest_payload) manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest() cursor.execute( """ UPDATE ingestion.daily_review_cases SET case_status = 'processing', manual_override_json = %s::jsonb, manual_override_sha256 = %s, frozen_at = now(), failure_code = NULL, failure_message = NULL, updated_at = now() WHERE id = %s """, (manifest_bytes.decode("utf-8"), manifest_sha256, review_case_id), ) else: manifest_payload = stored_manifest if isinstance(stored_manifest, Mapping) else None if manifest_payload is None: raise IngestionError("DATABASE_STATE_INVALID", "冻结人工价格清单无效") manifest_bytes = _canonical_json_bytes(dict(manifest_payload)) manifest_sha256 = hashlib.sha256(manifest_bytes).hexdigest() if manifest_sha256 != str(stored_manifest_sha256): raise IngestionError("DATABASE_STATE_INVALID", "冻结人工价格清单哈希不一致") cursor.execute( """ UPDATE ingestion.daily_review_cases SET case_status = 'processing', failure_code = NULL, failure_message = NULL, updated_at = now() WHERE id = %s """, (review_case_id,), ) cursor.execute( """ SELECT COALESCE(max(attempt_no), 0) + 1 FROM ingestion.processing_attempts WHERE processing_run_id = %s """, (int(processing_run_id),), ) attempt_no = int(cursor.fetchone()[0]) cursor.execute( """ INSERT INTO ingestion.processing_attempts ( processing_run_id, attempt_no, attempt_status, idempotency_key ) VALUES (%s, %s, 'queued', %s) """, (int(processing_run_id), attempt_no, idempotency_key), ) self._review_event( cursor, review_case_id, "PRICE_REVIEW_FINALIZED", actor_username, int(current_revision), ) cursor.execute( """ SELECT artifact.artifact_kind, artifact.object_key, artifact.original_filename, artifact.sha256, artifact.byte_size, artifact.mime_type FROM ingestion.processing_runs AS run JOIN ingestion.artifacts AS artifact ON artifact.id = run.source_artifact_id WHERE run.id = %s FOR SHARE OF artifact """, (int(processing_run_id),), ) source_row = cursor.fetchone() if ( source_row is None or str(source_row[0]) != "opera_xml" or not isinstance(source_row[1], str) or not isinstance(source_row[2], str) or not isinstance(source_row[3], str) or not isinstance(source_row[4], int) or not isinstance(source_row[5], str) ): raise IngestionError("DATABASE_STATE_INVALID", "复核源XML制品无效") source_ref = ArtifactRef( role="source_xml", file_kind="opera_xml", object_key=source_row[1], original_filename=source_row[2], sha256=source_row[3], byte_size=source_row[4], mime_type=source_row[5], ) if source_ref.sha256 != str(source_sha256): raise IngestionError("DATABASE_STATE_INVALID", "复核源XML身份不一致") return ReviewGeneration( "ready", job_id, str(case_key), attempt_no, business_date, manifest_bytes, manifest_sha256, source=source_ref, ) def record_price_review_generation_failure( self, job_id: str, case_id: str, attempt_no: int, failure_code: str, ) -> None: if not isinstance(failure_code, str) or re.fullmatch(r"[A-Z][A-Z0-9_]{0,63}", failure_code) is None: failure_code = "REVIEW_GENERATION_FAILED" self._run_transaction( lambda cursor: self._record_price_review_generation_failure( cursor, job_id, case_id, attempt_no, failure_code ), "review generation failure could not be recorded", ) def _record_price_review_generation_failure( self, cursor: Any, job_id: str, case_id: str, attempt_no: int, failure_code: str ) -> None: case = self._lock_review_case(cursor, job_id, case_id) review_case_id, _case_key, status, revision, _date_value, _processor, _rules, _source, _json, _sha, _artifact, run_id, _run_status, _requested_processor, _requested_rules = case if str(status) == "completed": return cursor.execute( """ SELECT id FROM ingestion.processing_attempts WHERE processing_run_id = %s AND attempt_no = %s FOR UPDATE """, (int(run_id), attempt_no), ) attempt = cursor.fetchone() if not attempt: raise IngestionError("JOB_NOT_FOUND", "复核生成尝试不存在") cursor.execute( """ UPDATE ingestion.processing_attempts SET attempt_status = 'failed', failure_code = %s, failure_message = 'manual review final generation failed', finished_at = now() WHERE id = %s """, (failure_code, int(attempt[0])), ) cursor.execute( """ UPDATE ingestion.daily_review_cases SET case_status = 'generation_failed', failure_code = %s, failure_message = 'manual review final generation failed', updated_at = now() WHERE id = %s """, (failure_code, int(review_case_id)), ) cursor.execute( """ UPDATE ingestion.processing_runs SET run_status = 'awaiting_review', failure_code = NULL, failure_message = NULL, finished_at = NULL, updated_at = now() WHERE id = %s """, (int(run_id),), ) self._review_event( cursor, int(review_case_id), "PRICE_REVIEW_GENERATION_FAILED", "system:processor", int(revision), ) def cancel_price_review( self, job_id: str, case_id: str, revision: int, actor_username: str, ) -> Dict[str, Any]: actor = self._review_actor(actor_username) return self._run_transaction( lambda cursor: self._cancel_price_review(cursor, job_id, case_id, revision, actor), "review case could not be cancelled", ) def _cancel_price_review( self, cursor: Any, job_id: str, case_id: str, revision: int, actor_username: str ) -> Dict[str, Any]: case = self._lock_review_case(cursor, job_id, case_id) review_case_id, _case_key, status, current_revision, _date_value, _processor, _rules, _source, _json, _sha, _artifact, run_id, _run_status, _requested_processor, _requested_rules = case if str(status) not in {"open", "generation_failed"}: raise IngestionError("REVIEW_NOT_CANCELLABLE", "复核任务当前不能取消") if not isinstance(revision, int) or revision != int(current_revision): raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新") next_revision = int(current_revision) + 1 cursor.execute( """ UPDATE ingestion.daily_review_cases SET case_status = 'cancelled', revision = %s, cancelled_at = now(), updated_at = now() WHERE id = %s """, (next_revision, int(review_case_id)), ) cursor.execute( """ UPDATE ingestion.processing_runs SET run_status = 'cancelled', failure_code = NULL, failure_message = NULL, finished_at = now(), updated_at = now() WHERE id = %s """, (int(run_id),), ) self._review_event( cursor, int(review_case_id), "PRICE_REVIEW_CANCELLED", actor_username, next_revision, ) return self._get_price_review(cursor, job_id, 200, 0) @staticmethod def _lock_attempt( cursor: Any, job_id: str, attempt_no: int, ) -> Tuple[int, str, int, str]: cursor.execute( """ SELECT run.id, run.run_status, attempt.id, attempt.attempt_status FROM ingestion.processing_runs AS run JOIN ingestion.processing_attempts AS attempt ON attempt.processing_run_id = run.id AND attempt.attempt_no = %s WHERE run.run_key = %s FOR UPDATE OF run, attempt """, (attempt_no, job_id), ) row = cursor.fetchone() if not row: raise IngestionError("JOB_NOT_FOUND", "processing job was not registered") return int(row[0]), str(row[1]), int(row[2]), str(row[3]) def _register_job(self, cursor: Any, registration: JobRegistration) -> None: source_artifact_id = self._ensure_artifact(cursor, registration.source) cursor.execute( """ SELECT id, source_artifact_id, requested_processor_version, requested_rule_set_sha256, uploaded_filename, run_status FROM ingestion.processing_runs WHERE run_key = %s FOR UPDATE """, (registration.job_id,), ) existing = cursor.fetchone() if existing is None: if registration.attempt_no != 1: raise IngestionError( "JOB_CONFLICT", "first processing attempt must be attempt 1", ) cursor.execute( """ INSERT INTO ingestion.processing_runs ( run_key, pipeline_type, source_artifact_id, run_status, requested_processor_version, requested_rule_set_sha256, uploaded_filename ) VALUES (%s, 'opera_daily', %s, 'queued', %s, %s, %s) RETURNING id """, ( registration.job_id, source_artifact_id, registration.processor_version, registration.rule_set_sha256, registration.uploaded_filename, ), ) processing_run_id = int(cursor.fetchone()[0]) else: if ( int(existing[1]) != source_artifact_id or existing[2] != registration.processor_version or str(existing[3]) != registration.rule_set_sha256 or existing[4] != registration.uploaded_filename ): raise IngestionError( "JOB_CONFLICT", "processing job identity conflicts", ) processing_run_id = int(existing[0]) cursor.execute( """ SELECT idempotency_key FROM ingestion.processing_attempts WHERE processing_run_id = %s AND attempt_no = %s """, (processing_run_id, registration.attempt_no), ) attempt = cursor.fetchone() if attempt is not None: if str(attempt[0]) != registration.idempotency_key: raise IngestionError( "JOB_CONFLICT", "processing attempt identity conflicts", ) return if str(existing[5]) in { "accepted", "rejected", "failed", "cancelled", }: raise IngestionError( "JOB_TERMINAL", "processing job is already terminal", ) cursor.execute( """ SELECT COALESCE(max(attempt_no), 0) + 1 FROM ingestion.processing_attempts WHERE processing_run_id = %s """, (processing_run_id,), ) if int(cursor.fetchone()[0]) != registration.attempt_no: raise IngestionError( "JOB_CONFLICT", "processing attempt sequence conflicts", ) cursor.execute( """ INSERT INTO ingestion.processing_attempts ( processing_run_id, attempt_no, attempt_status, idempotency_key ) VALUES (%s, %s, 'queued', %s) """, ( processing_run_id, registration.attempt_no, registration.idempotency_key, ), ) def commit_delivery(self, delivery: VerifiedDelivery) -> IngestionOutcome: return self._run_transaction( lambda cursor: self._commit(cursor, delivery), "validated delivery was not committed", ) def _commit( self, cursor: Any, delivery: VerifiedDelivery, ) -> IngestionOutcome: envelope = delivery.envelope cursor.execute( """ SELECT run.id, run.source_artifact_id, run.requested_processor_version, run.requested_rule_set_sha256, run.run_status, attempt.id, attempt.attempt_status FROM ingestion.processing_runs AS run JOIN ingestion.processing_attempts AS attempt ON attempt.processing_run_id = run.id AND attempt.attempt_no = %s WHERE run.run_key = %s FOR UPDATE OF run, attempt """, (envelope.attempt_no, envelope.job_id), ) run_row = cursor.fetchone() if not run_row: raise IngestionError( "JOB_NOT_FOUND", "processing job was not registered", ) processing_run_id = int(run_row[0]) attempt_id = int(run_row[5]) cursor.execute( """ SELECT delivery.id, delivery.envelope_sha256, delivery.delivery_status, delivery.daily_version_id, delivery.review_case_id, delivery.business_date, version.version_no, version.processing_run_id FROM ingestion.processing_deliveries AS delivery LEFT JOIN finance.daily_versions AS version ON version.id = delivery.daily_version_id WHERE delivery.delivery_key = %s FOR UPDATE OF delivery """, (envelope.delivery_id,), ) delivery_row = cursor.fetchone() if delivery_row: if str(delivery_row[1]) != delivery.envelope_sha256: raise IngestionError( "CALLBACK_CONFLICT", "delivery identifier conflicts", ) delivery_status = str(delivery_row[2]) if delivery_status == "recorded_failure": return IngestionOutcome( "recorded_failure", envelope.job_id, delivery_row[5], ( int(delivery_row[3]) if delivery_row[3] is not None else None ), None, ) if delivery_status == "committed": disposition = ( "committed" if ( delivery_row[7] is not None and int(delivery_row[7]) == processing_run_id ) else "already_committed" ) return IngestionOutcome( disposition, envelope.job_id, delivery_row[5], int(delivery_row[3]), int(delivery_row[6]), ) if delivery_status == "recorded_review": return IngestionOutcome( "recorded_review", envelope.job_id, delivery_row[5], None, None, ) if delivery_status == "rejected": raise IngestionError( "CALLBACK_REJECTED", "delivery was previously rejected", ) delivery_id = int(delivery_row[0]) else: delivery_id = None if ( delivery_id is None and str(run_row[4]) in {"accepted", "rejected", "failed", "cancelled"} ): raise IngestionError( "JOB_TERMINAL", "processing job is already terminal", ) if ( run_row[2] != envelope.processor_version or str(run_row[3]) != envelope.rule_set_sha256 ): raise IngestionError( "JOB_DELIVERY_MISMATCH", "delivery does not match its job", ) source_ref = envelope.artifacts["source_xml"] result_ref = envelope.artifacts["result_json"] structured_ref = envelope.artifacts["structured_result_json"] if source_ref is None or result_ref is None or structured_ref is None: raise IngestionError( "RESULT_CONTRACT_INVALID", "required artifact is missing", ) source_artifact_id = self._ensure_artifact(cursor, source_ref) if source_artifact_id != int(run_row[1]): raise IngestionError( "JOB_DELIVERY_MISMATCH", "delivery does not match its job", ) daily_ref = envelope.artifacts["daily_report"] exception_ref = envelope.artifacts["exception_report"] manual_override_ref = envelope.artifacts.get("manual_override_json") daily_artifact_id = ( self._ensure_artifact(cursor, daily_ref) if daily_ref is not None else None ) result_artifact_id = self._ensure_artifact(cursor, result_ref) structured_artifact_id = self._ensure_artifact( cursor, structured_ref, ) exception_artifact_id = ( self._ensure_artifact(cursor, exception_ref) if exception_ref is not None else None ) manual_override_artifact_id = ( self._ensure_artifact(cursor, manual_override_ref) if manual_override_ref is not None else None ) envelope_json = json.dumps( envelope.to_dict(), ensure_ascii=False, separators=(",", ":"), sort_keys=True, ) if delivery_id is None: cursor.execute( """ INSERT INTO ingestion.processing_deliveries ( delivery_key, processing_run_id, attempt_id, envelope_sha256, envelope_json, delivery_status, result_status, processor_version, rule_set_sha256, result_schema_version, business_date, source_artifact_id, daily_report_artifact_id, result_json_artifact_id, structured_result_artifact_id, exception_report_artifact_id, manual_override_artifact_id, validated_at ) VALUES ( %s, %s, %s, %s, %s::jsonb, 'validating', %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now() ) RETURNING id """, ( envelope.delivery_id, processing_run_id, attempt_id, delivery.envelope_sha256, envelope_json, envelope.status, envelope.processor_version, envelope.rule_set_sha256, envelope.result_schema_version, envelope.business_date, source_artifact_id, daily_artifact_id, result_artifact_id, structured_artifact_id, exception_artifact_id, manual_override_artifact_id, ), ) delivery_id = int(cursor.fetchone()[0]) else: cursor.execute( """ UPDATE ingestion.processing_deliveries SET delivery_status = 'validating', result_status = %s, processor_version = %s, rule_set_sha256 = %s, result_schema_version = %s, business_date = %s, source_artifact_id = %s, daily_report_artifact_id = %s, result_json_artifact_id = %s, structured_result_artifact_id = %s, exception_report_artifact_id = %s, manual_override_artifact_id = %s, validated_at = now() WHERE id = %s """, ( envelope.status, envelope.processor_version, envelope.rule_set_sha256, envelope.result_schema_version, envelope.business_date, source_artifact_id, daily_artifact_id, result_artifact_id, structured_artifact_id, exception_artifact_id, manual_override_artifact_id, delivery_id, ), ) cursor.execute( """ UPDATE ingestion.processing_runs SET run_status = 'validating', result_artifact_id = %s, delivered_processor_version = %s, delivered_rule_set_sha256 = %s, result_schema_version = %s, delivery_sha256 = %s, delivery_json = %s::jsonb, business_date = %s, failure_code = NULL, failure_message = NULL, updated_at = now(), validated_at = now() WHERE id = %s """, ( structured_artifact_id, envelope.processor_version, envelope.rule_set_sha256, envelope.result_schema_version, delivery.envelope_sha256, envelope_json, envelope.business_date, processing_run_id, ), ) cursor.execute( """ UPDATE ingestion.processing_attempts SET attempt_status = 'delivered' WHERE id = %s """, (attempt_id,), ) if envelope.status == "failed": return self._record_failure( cursor, delivery, processing_run_id, attempt_id, delivery_id, source_artifact_id, result_artifact_id, structured_artifact_id, exception_artifact_id, ) if envelope.status == "review_required": return self._record_review( cursor, delivery, processing_run_id, attempt_id, delivery_id, source_artifact_id, result_artifact_id, structured_artifact_id, ) if envelope.business_date is None or daily_artifact_id is None: raise IngestionError( "RESULT_CONTRACT_INVALID", "successful delivery is incomplete", ) return self._commit_success( cursor, delivery, processing_run_id, attempt_id, delivery_id, source_artifact_id, daily_artifact_id, result_artifact_id, structured_artifact_id, manual_override_artifact_id, ) def _record_review( self, cursor: Any, delivery: VerifiedDelivery, processing_run_id: int, attempt_id: int, delivery_id: int, source_artifact_id: int, result_artifact_id: int, structured_artifact_id: int, ) -> IngestionOutcome: envelope = delivery.envelope payload = delivery.structured_payload business_date = envelope.business_date if business_date is None: raise IngestionError("RESULT_CONTRACT_INVALID", "review delivery has no business date") source_ref = envelope.artifacts["source_xml"] issues = payload.get("review_issues") if source_ref is None or not isinstance(issues, list) or not issues: raise IngestionError("RESULT_CONTRACT_INVALID", "review delivery is incomplete") case_key = "dailyreview-" + hashlib.sha256( f"{envelope.job_id}\x00{envelope.delivery_id}".encode("utf-8") ).hexdigest()[:32] cursor.execute( """ INSERT INTO ingestion.daily_review_cases ( case_key, processing_run_id, initial_delivery_id, business_date, source_sha256, processor_version, rule_set_sha256, review_version, case_status ) VALUES (%s, %s, %s, %s, %s, %s, %s, '1.0', 'open') RETURNING id """, ( case_key, processing_run_id, delivery_id, business_date, source_ref.sha256, envelope.processor_version, envelope.rule_set_sha256, ), ) review_case_id = int(cursor.fetchone()[0]) for issue in issues: if not isinstance(issue, Mapping): raise IngestionError("RESULT_CONTRACT_INVALID", "review issue is invalid") try: company_key = issue["company_key"] rate_code = issue["rate_code"] opera_amount = _decimal_or_none(issue["effective_rate_amount"]) candidate_prices = issue["candidate_prices"] affected_records = int(issue["affected_records"]) affected_rooms = int(issue["affected_rooms"]) affected_room_nights = int(issue["affected_room_nights"]) except (KeyError, TypeError, ValueError): raise IngestionError("RESULT_CONTRACT_INVALID", "review issue is invalid") from None if ( not isinstance(company_key, str) or not company_key or not isinstance(rate_code, str) or not rate_code or opera_amount is None or opera_amount < 0 or not isinstance(candidate_prices, list) or affected_records < 1 or affected_rooms < 1 or affected_room_nights < 0 ): raise IngestionError("RESULT_CONTRACT_INVALID", "review issue is invalid") cursor.execute( """ INSERT INTO ingestion.daily_review_items ( review_case_id, company_key, rate_code, effective_rate_amount, candidate_prices, affected_records, affected_rooms, affected_room_nights ) VALUES (%s, %s, %s, %s, %s::jsonb, %s, %s, %s) """, ( review_case_id, company_key, rate_code, opera_amount, json.dumps(candidate_prices, ensure_ascii=False, separators=(",", ":")), affected_records, affected_rooms, affected_room_nights, ), ) self._review_event( cursor, review_case_id, "PRICE_REVIEW_REQUIRED", "system:processor", 0, ) cursor.execute( """ UPDATE ingestion.processing_deliveries SET delivery_status = 'recorded_review', review_case_id = %s, committed_at = now() WHERE id = %s """, (review_case_id, delivery_id), ) cursor.execute( """ UPDATE ingestion.processing_attempts SET attempt_status = 'review_required', failure_code = NULL, failure_message = NULL, finished_at = now() WHERE id = %s """, (attempt_id,), ) cursor.execute( """ UPDATE ingestion.processing_runs SET run_status = 'awaiting_review', result_artifact_id = %s, business_date = %s, failure_code = NULL, failure_message = NULL, finished_at = NULL, updated_at = now() WHERE id = %s """, (structured_artifact_id, business_date, processing_run_id), ) return IngestionOutcome( "recorded_review", envelope.job_id, business_date, None, None, review_case_id=case_key, review_revision=0, review_completed_items=0, review_total_items=len(issues), ) def _record_failure( self, cursor: Any, delivery: VerifiedDelivery, processing_run_id: int, attempt_id: int, delivery_id: int, source_artifact_id: int, result_artifact_id: int, structured_artifact_id: int, exception_artifact_id: Optional[int], ) -> IngestionOutcome: envelope = delivery.envelope payload = delivery.structured_payload code = _safe_failure_code(payload) counts = _outcome_counts(payload) records = payload.get("records") if not isinstance(records, list): raise IngestionError( "RESULT_CONTRACT_INVALID", "structured records are invalid", ) cursor.execute( """ INSERT INTO finance.daily_versions ( business_date, version_no, processing_run_id, source_artifact_id, result_json_artifact_id, structured_result_artifact_id, exception_report_artifact_id, version_status, processor_version, rule_set_sha256, result_schema_version, result_sha256, source_rows, retained_rows, excluded_rate_code_rows, duplicate_rows, validation_failed_rows, price_unmatched_rows, failure_code, failure_message, validated_at ) VALUES ( NULL, NULL, %s, %s, %s, %s, %s, 'rejected', %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'deterministic processing failed', now() ) RETURNING id """, ( processing_run_id, source_artifact_id, result_artifact_id, structured_artifact_id, exception_artifact_id, envelope.processor_version, envelope.rule_set_sha256, envelope.result_schema_version, envelope.artifacts["structured_result_json"].sha256, int(payload["source_rows"]), *counts, code, ), ) daily_version_id = int(cursor.fetchone()[0]) booking_links = self._load_booking_links(cursor, records) self._insert_records( cursor, daily_version_id, records, booking_links, ) cursor.execute( """ UPDATE ingestion.processing_deliveries SET delivery_status = 'recorded_failure', daily_version_id = %s, failure_code = %s, failure_message = 'deterministic processing failed', committed_at = now() WHERE id = %s """, (daily_version_id, code, delivery_id), ) cursor.execute( """ UPDATE ingestion.processing_attempts SET attempt_status = 'failed', failure_code = %s, failure_message = 'deterministic processing failed', finished_at = now() WHERE id = %s """, (code, attempt_id), ) cursor.execute( """ SELECT id, revision FROM ingestion.daily_review_cases WHERE processing_run_id = %s AND case_status = 'processing' FOR UPDATE """, (processing_run_id,), ) active_review = cursor.fetchone() if active_review: review_case_id = int(active_review[0]) cursor.execute( """ UPDATE ingestion.daily_review_cases SET case_status = 'failed', failure_code = %s, failure_message = 'deterministic final generation failed', updated_at = now() WHERE id = %s """, (code, review_case_id), ) self._review_event( cursor, review_case_id, "PRICE_REVIEW_FAILED", "system:processor", int(active_review[1]), ) cursor.execute( """ UPDATE ingestion.processing_runs SET run_status = 'failed', failure_code = %s, failure_message = 'deterministic processing failed', updated_at = now(), finished_at = now() WHERE id = %s """, (code, processing_run_id), ) self._insert_outbox( cursor, f"processing-run:{processing_run_id}:failed", "processing_run", processing_run_id, "arr.processing_failed", { "job_id": envelope.job_id, "business_date": ( envelope.business_date.isoformat() if envelope.business_date else None ), "daily_version_id": daily_version_id, "failure_code": code, }, ) return IngestionOutcome( "recorded_failure", envelope.job_id, envelope.business_date, daily_version_id, None, ) def _commit_success( self, cursor: Any, delivery: VerifiedDelivery, processing_run_id: int, attempt_id: int, delivery_id: int, source_artifact_id: int, daily_artifact_id: int, result_artifact_id: int, structured_artifact_id: int, manual_override_artifact_id: Optional[int], ) -> IngestionOutcome: envelope = delivery.envelope business_date = envelope.business_date assert business_date is not None payload = delivery.structured_payload review_case_db_id: Optional[int] = None manual_override_sha256: Optional[str] = None manually_priced_rows = 0 if manual_override_artifact_id is not None: review_case_key = payload.get("review_case_id") manual_override_sha256 = payload.get("manual_override_sha256") manually_priced_rows = int(payload.get("manually_priced_rows", 0)) if ( not isinstance(review_case_key, str) or not isinstance(manual_override_sha256, str) or not SHA256_RE.fullmatch(manual_override_sha256) or manually_priced_rows < 1 ): raise IngestionError("RESULT_CONTRACT_INVALID", "人工复核结果lineage无效") case = self._lock_review_case(cursor, envelope.job_id, review_case_key) if ( str(case[2]) != "processing" or int(case[11]) != processing_run_id or str(case[7]) != envelope.artifacts["source_xml"].sha256 or str(case[5]) != envelope.processor_version or str(case[6]) != envelope.rule_set_sha256 or str(case[9]) != manual_override_sha256 ): raise IngestionError("REVIEW_STATE_INVALID", "人工复核确认状态无效") review_case_db_id = int(case[0]) elif payload.get("review_case_id") is not None: raise IngestionError("RESULT_CONTRACT_INVALID", "人工复核结果缺少冻结清单产物") cursor.execute( "SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))", (f"finance-daily:{business_date.isoformat()}",), ) cursor.execute( """ SELECT id, version_no, version_status FROM finance.daily_versions WHERE source_artifact_id = %s AND business_date = %s AND processor_version = %s AND rule_set_sha256 = %s AND manual_override_sha256 IS NOT DISTINCT FROM %s FOR UPDATE """, ( source_artifact_id, business_date, envelope.processor_version, envelope.rule_set_sha256, manual_override_sha256, ), ) existing = cursor.fetchone() if existing: if str(existing[2]) not in {"active", "superseded"}: raise IngestionError( "DATABASE_STATE_INVALID", "existing daily version is invalid", ) daily_version_id = int(existing[0]) version_no = int(existing[1]) disposition = "already_committed" else: payload = delivery.structured_payload counts = _outcome_counts(payload) cursor.execute( """ SELECT COALESCE(MAX(version_no), 0) + 1 FROM finance.daily_versions WHERE business_date = %s """, (business_date,), ) version_no = int(cursor.fetchone()[0]) cursor.execute( """ INSERT INTO finance.daily_versions ( business_date, version_no, processing_run_id, review_case_id, manual_override_sha256, manually_priced_rows, source_artifact_id, daily_report_artifact_id, result_json_artifact_id, structured_result_artifact_id, version_status, processor_version, rule_set_sha256, result_schema_version, result_sha256, source_rows, retained_rows, excluded_rate_code_rows, duplicate_rows, validation_failed_rows, price_unmatched_rows, validated_at ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'validated', %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now() ) RETURNING id """, ( business_date, version_no, processing_run_id, review_case_db_id, manual_override_sha256, manually_priced_rows, source_artifact_id, daily_artifact_id, result_artifact_id, structured_artifact_id, envelope.processor_version, envelope.rule_set_sha256, envelope.result_schema_version, envelope.artifacts["structured_result_json"].sha256, int(payload["source_rows"]), *counts, ), ) daily_version_id = int(cursor.fetchone()[0]) records = payload.get("records") if not isinstance(records, list): raise IngestionError( "RESULT_CONTRACT_INVALID", "structured records are invalid", ) booking_links = self._load_booking_links(cursor, records) self._insert_records( cursor, daily_version_id, records, booking_links, ) channels = payload.get("channels") if not isinstance(channels, list): raise IngestionError( "RESULT_CONTRACT_INVALID", "structured channels are invalid", ) self._insert_channel_metrics( cursor, daily_version_id, channels, ) self._activate_daily_version( cursor, business_date, daily_version_id, ) disposition = "committed" cursor.execute( """ UPDATE ingestion.processing_deliveries SET delivery_status = 'committed', daily_version_id = %s, review_case_id = %s, committed_at = now() WHERE id = %s """, (daily_version_id, review_case_db_id, delivery_id), ) if review_case_db_id is not None: cursor.execute( """ UPDATE ingestion.daily_review_cases SET case_status = 'completed', manual_override_artifact_id = %s, failure_code = NULL, failure_message = NULL, completed_at = now(), updated_at = now() WHERE id = %s """, (manual_override_artifact_id, review_case_db_id), ) cursor.execute( """ UPDATE ingestion.processing_attempts SET attempt_status = 'succeeded', failure_code = NULL, failure_message = NULL, finished_at = now() WHERE id = %s """, (attempt_id,), ) cursor.execute( """ UPDATE ingestion.processing_runs SET run_status = 'accepted', business_date = %s, failure_code = NULL, failure_message = NULL, updated_at = now(), finished_at = now() WHERE id = %s """, (business_date, processing_run_id), ) self._insert_outbox( cursor, f"processing-run:{processing_run_id}:accepted", "processing_run", processing_run_id, "arr.daily_version_committed", { "job_id": envelope.job_id, "business_date": business_date.isoformat(), "daily_version_id": daily_version_id, "version_no": version_no, "disposition": disposition, }, ) return IngestionOutcome( disposition, envelope.job_id, business_date, daily_version_id, version_no, ) @staticmethod def _insert_channel_metrics( cursor: Any, daily_version_id: int, channels: Sequence[Mapping[str, Any]], ) -> None: for channel_order, channel in enumerate(channels, 1): cursor.execute( """ INSERT INTO finance.daily_channel_metrics ( daily_version_id, channel_key, channel_order, row_count ) VALUES (%s, %s, %s, %s) """, ( daily_version_id, str(channel["worksheet"]), channel_order, int(channel["rows"]), ), ) @staticmethod def _activate_daily_version( cursor: Any, business_date: date, daily_version_id: int, ) -> None: cursor.execute( """ SELECT daily_version_id FROM finance.current_daily_versions WHERE business_date = %s FOR UPDATE """, (business_date,), ) current = cursor.fetchone() if current and int(current[0]) != daily_version_id: cursor.execute( """ UPDATE finance.daily_versions SET version_status = 'superseded', superseded_at = now() WHERE id = %s AND version_status = 'active' """, (int(current[0]),), ) cursor.execute( """ UPDATE finance.daily_versions SET version_status = 'active', activated_at = now() WHERE id = %s AND version_status = 'validated' """, (daily_version_id,), ) cursor.execute( """ INSERT INTO finance.current_daily_versions ( business_date, daily_version_id, activated_at ) VALUES (%s, %s, now()) ON CONFLICT (business_date) DO UPDATE SET daily_version_id = EXCLUDED.daily_version_id, activated_at = EXCLUDED.activated_at """, (business_date, daily_version_id), ) @staticmethod def _load_booking_links( cursor: Any, records: Sequence[Mapping[str, Any]], ) -> Mapping[str, Tuple[str, int]]: group_codes = sorted( { str(record["group_code_key"]) for record in records if record.get("group_code_key") } ) if not group_codes: return {} cursor.execute( """ SELECT source.group_code_key, count(DISTINCT source.id) FROM booking.source_rows AS source JOIN booking.current_source_batch AS active_source ON active_source.source_batch_id = source.source_batch_id AND active_source.singleton JOIN booking.source_batches AS batch ON batch.id = source.source_batch_id AND batch.batch_status = 'accepted' JOIN booking.current_row_parses AS current_parse ON current_parse.source_row_id = source.id JOIN booking.parse_versions AS parsed ON parsed.id = current_parse.parse_version_id AND parsed.parse_status = 'accepted' WHERE source.group_code_key = ANY(%s) GROUP BY source.group_code_key """, (group_codes,), ) matches: Dict[str, Tuple[str, int]] = {} for group_code, count in cursor.fetchall(): match_count = int(count) matches[str(group_code)] = ("matched", match_count) for group_code in group_codes: matches.setdefault(group_code, ("unmatched", 0)) return matches @staticmethod def _insert_records( cursor: Any, daily_version_id: int, records: Sequence[Mapping[str, Any]], booking_links: Mapping[str, Tuple[str, int]], ) -> None: inserted_ids: Dict[int, int] = {} for record in records: source_sequence = int(record["source_sequence"]) duplicate_sequence = record.get( "duplicate_of_source_sequence", ) duplicate_id = ( inserted_ids.get(int(duplicate_sequence)) if duplicate_sequence is not None else None ) if duplicate_sequence is not None and duplicate_id is None: raise IngestionError( "RESULT_CONTRACT_INVALID", "duplicate lineage is invalid", ) group_code = record.get("group_code_key") if group_code is None: source_status, source_match_count = ( "missing_group_code", 0, ) else: source_status, source_match_count = booking_links.get( str(group_code), ("unmatched", 0), ) cursor.execute( """ INSERT INTO finance.daily_records ( daily_version_id, source_sequence, source_location, source_worksheet, source_row_no, outcome, decision_codes, duplicate_of_record_id, block_code, adults, children, company_name, company_key, confirmation_no, disp_room_no, effective_rate_amount, full_name, res_comment, trace_text, no_of_rooms, products, rate_code, room_category_label, arrival, departure, nights, real_price, total_price, kb_amount, channel_key, pricing_method, booking_source_match_status, booking_source_match_count ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s ) RETURNING id """, ( daily_version_id, source_sequence, record.get("source_location"), record.get("source_worksheet"), record.get("source_row_no"), record.get("outcome"), list(record.get("decision_codes") or []), duplicate_id, record.get("block_code"), record.get("adults"), record.get("children"), record.get("company_name"), record.get("company_key"), record.get("confirmation_no"), record.get("disp_room_no"), _decimal_or_none(record.get("effective_rate_amount")), record.get("full_name"), record.get("res_comment"), record.get("trace_text"), record.get("no_of_rooms"), record.get("products"), record.get("rate_code"), record.get("room_category_label"), _date_or_none(record.get("arrival")), _date_or_none(record.get("departure")), record.get("nights"), _decimal_or_none(record.get("real_price")), _decimal_or_none(record.get("total_price")), _decimal_or_none(record.get("kb_amount")), record.get("channel_key"), record.get("pricing_method"), source_status, source_match_count, ), ) inserted_ids[source_sequence] = int(cursor.fetchone()[0]) @staticmethod def _insert_outbox( cursor: Any, event_key: str, aggregate_type: str, aggregate_id: int, event_type: str, payload: Mapping[str, Any], ) -> None: cursor.execute( """ INSERT INTO ingestion.outbox_events ( event_key, aggregate_type, aggregate_id, event_type, payload ) VALUES (%s, %s, %s, %s, %s::jsonb) ON CONFLICT (event_key) DO NOTHING """, ( event_key, aggregate_type, aggregate_id, event_type, json.dumps( payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True, ), ), )