"""Repository contracts and an offline reference implementation for ARR ingestion.""" from __future__ import annotations import copy import hashlib import json import re import threading from dataclasses import dataclass, replace from datetime import date from decimal import Decimal, InvalidOperation from typing import Any, Dict, Mapping, Optional, Protocol, Tuple from arr_ingestion.contracts import ArtifactRef, IngestionError from arr_ingestion.validation import VerifiedDelivery MANUAL_PRICE_RE = re.compile(r"^(?:0|[1-9][0-9]{0,15})$") @dataclass(frozen=True) class JobRegistration: job_id: str source: ArtifactRef processor_version: str rule_set_sha256: str attempt_no: int idempotency_key: str uploaded_filename: Optional[str] = None @dataclass(frozen=True) class IngestionOutcome: status: str job_id: str business_date: Optional[date] daily_version_id: Optional[int] version_no: Optional[int] review_case_id: Optional[str] = None review_revision: Optional[int] = None review_completed_items: Optional[int] = None review_total_items: Optional[int] = None @dataclass(frozen=True) class ReviewGeneration: status: str job_id: str review_case_id: str attempt_no: Optional[int] business_date: date manual_override_bytes: Optional[bytes] manual_override_sha256: Optional[str] daily_version_id: Optional[int] = None version_no: Optional[int] = None source: Optional[ArtifactRef] = None class IngestionRepository(Protocol): def register_job(self, registration: JobRegistration) -> None: ... def mark_running(self, job_id: str, attempt_no: int) -> None: ... def record_failure(self, job_id: str, attempt_no: int, failure_code: str) -> None: ... def commit_delivery(self, delivery: VerifiedDelivery) -> IngestionOutcome: ... def get_price_review(self, job_id: str, limit: int, offset: int) -> Dict[str, Any]: ... 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]: ... 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: ... def record_price_review_generation_failure( self, job_id: str, case_id: str, attempt_no: int, failure_code: str, ) -> None: ... def cancel_price_review( self, job_id: str, case_id: str, revision: int, actor_username: str, ) -> Dict[str, Any]: ... @dataclass class _MemoryJob: registration: JobRegistration status: str = "queued" failure_code: Optional[str] = None business_date: Optional[date] = None @dataclass class _MemoryReviewCase: case_id: str business_date: date processor_version: str rule_set_sha256: str source_sha256: str status: str revision: int items: list[Dict[str, Any]] manifest_bytes: Optional[bytes] = None manifest_sha256: Optional[str] = None @dataclass class _MemoryDailyVersion: id: int business_date: date version_no: int identity: Tuple[str, date, str, str, Optional[str]] status: str records: Tuple[Mapping[str, Any], ...] channels: Tuple[Mapping[str, Any], ...] class InMemoryIngestionRepository: """Thread-safe vertical-slice repository; never used as the production fact store.""" def __init__(self) -> None: self._lock = threading.RLock() self._jobs: Dict[str, _MemoryJob] = {} self._callbacks: Dict[str, Tuple[str, IngestionOutcome]] = {} self._versions: Dict[int, _MemoryDailyVersion] = {} self._identity_versions: Dict[Tuple[str, date, str, str, Optional[str]], int] = {} self._current: Dict[date, int] = {} self._reviews: Dict[str, _MemoryReviewCase] = {} self._next_review_item_id = 1 self._next_version_id = 1 def register_job(self, registration: JobRegistration) -> None: if registration.source.role != "source_xml" or registration.attempt_no < 1: raise IngestionError("JOB_INVALID", "processing job registration is invalid") with self._lock: existing = self._jobs.get(registration.job_id) if existing is not None: if ( existing.registration.source != registration.source or existing.registration.processor_version != registration.processor_version or existing.registration.rule_set_sha256 != registration.rule_set_sha256 or existing.registration.uploaded_filename != registration.uploaded_filename ): raise IngestionError("JOB_CONFLICT", "processing job identity conflicts") if registration.attempt_no == existing.registration.attempt_no: if existing.registration != registration: raise IngestionError("JOB_CONFLICT", "processing attempt identity conflicts") return if ( existing.status != "awaiting_review" or registration.attempt_no != existing.registration.attempt_no + 1 ): raise IngestionError("JOB_CONFLICT", "processing attempt sequence conflicts") existing.registration = registration existing.status = "queued" return self._jobs[registration.job_id] = _MemoryJob(registration=registration) def mark_running(self, job_id: str, attempt_no: int) -> None: with self._lock: job = self._jobs.get(job_id) if job is None: raise IngestionError("JOB_NOT_FOUND", "processing job was not registered") if attempt_no != job.registration.attempt_no: raise IngestionError("JOB_NOT_FOUND", "processing attempt was not registered") if job.status in {"succeeded", "failed", "cancelled"}: raise IngestionError("JOB_TERMINAL", "processing job is already terminal") job.status = "running" def record_failure(self, job_id: str, attempt_no: int, failure_code: str) -> None: with self._lock: job = self._jobs.get(job_id) if job is None: raise IngestionError("JOB_NOT_FOUND", "processing job was not registered") if attempt_no != job.registration.attempt_no: raise IngestionError("JOB_NOT_FOUND", "processing attempt was not registered") if job.status == "succeeded": raise IngestionError("JOB_TERMINAL", "processing job is already terminal") job.status = "failed" job.failure_code = failure_code review_case = self._reviews.get(job_id) if review_case is not None and review_case.status == "processing": review_case.status = "failed" def commit_delivery(self, delivery: VerifiedDelivery) -> IngestionOutcome: envelope = delivery.envelope with self._lock: existing_callback = self._callbacks.get(envelope.delivery_id) if existing_callback is not None: existing_hash, outcome = existing_callback if existing_hash != delivery.envelope_sha256: raise IngestionError("CALLBACK_CONFLICT", "delivery identifier conflicts") return outcome job = self._jobs.get(envelope.job_id) if job is None: raise IngestionError("JOB_NOT_FOUND", "processing job was not registered") registration = job.registration source = envelope.artifacts["source_xml"] if ( source is None or source.sha256 != registration.source.sha256 or source.byte_size != registration.source.byte_size or source.object_key != registration.source.object_key or envelope.attempt_no != registration.attempt_no or envelope.processor_version != registration.processor_version or envelope.rule_set_sha256 != registration.rule_set_sha256 ): raise IngestionError("JOB_DELIVERY_MISMATCH", "delivery does not match its job") if envelope.status == "failed": job.status = "failed" errors = delivery.structured_payload.get("errors") first_error = errors[0] if isinstance(errors, list) and errors else {} code = first_error.get("code") if isinstance(first_error, Mapping) else None job.failure_code = str(code or "PROCESSING_FAILED") review_case = self._reviews.get(envelope.job_id) if review_case is not None and review_case.status == "processing": review_case.status = "failed" outcome = IngestionOutcome( status="recorded_failure", job_id=envelope.job_id, business_date=envelope.business_date, daily_version_id=None, version_no=None, ) self._callbacks[envelope.delivery_id] = ( delivery.envelope_sha256, outcome, ) return outcome if envelope.status == "review_required": if envelope.business_date is None: raise IngestionError("RESULT_CONTRACT_INVALID", "review delivery has no date") issues = delivery.structured_payload.get("review_issues") if not isinstance(issues, list) or not issues: raise IngestionError("RESULT_CONTRACT_INVALID", "review issues are invalid") case_id = "dailyreview-" + hashlib.sha256( f"{envelope.job_id}\x00{envelope.delivery_id}".encode("utf-8") ).hexdigest()[:32] if envelope.job_id in self._reviews: case = self._reviews[envelope.job_id] else: items: list[Dict[str, Any]] = [] for issue in issues: if not isinstance(issue, Mapping): raise IngestionError("RESULT_CONTRACT_INVALID", "review issue is invalid") item = copy.deepcopy(dict(issue)) item["item_id"] = self._next_review_item_id self._next_review_item_id += 1 item["real_price"] = None item["revision"] = 0 items.append(item) case = _MemoryReviewCase( case_id=case_id, business_date=envelope.business_date, processor_version=envelope.processor_version, rule_set_sha256=envelope.rule_set_sha256, source_sha256=registration.source.sha256, status="open", revision=0, items=items, ) self._reviews[envelope.job_id] = case job.status = "awaiting_review" job.business_date = envelope.business_date completed = sum(1 for item in case.items if item["real_price"] is not None) outcome = IngestionOutcome( status="recorded_review", job_id=envelope.job_id, business_date=envelope.business_date, daily_version_id=None, version_no=None, review_case_id=case.case_id, review_revision=case.revision, review_completed_items=completed, review_total_items=len(case.items), ) self._callbacks[envelope.delivery_id] = (delivery.envelope_sha256, outcome) return outcome review_case_id = delivery.structured_payload.get("review_case_id") review_case: Optional[_MemoryReviewCase] = None if review_case_id is not None: review_case = self._reviews.get(envelope.job_id) manual_ref = envelope.artifacts.get("manual_override_json") if ( review_case is None or review_case.case_id != review_case_id or review_case.status != "processing" or manual_ref is None or review_case.manifest_sha256 != manual_ref.sha256 ): raise IngestionError("REVIEW_STATE_INVALID", "人工复核确认状态无效") if envelope.business_date is None: raise IngestionError("RESULT_CONTRACT_INVALID", "successful delivery has no date") manual_override_sha256 = delivery.structured_payload.get("manual_override_sha256") if manual_override_sha256 is not None and not isinstance(manual_override_sha256, str): raise IngestionError("RESULT_CONTRACT_INVALID", "人工复核清单哈希无效") identity = ( registration.source.sha256, envelope.business_date, envelope.processor_version, envelope.rule_set_sha256, manual_override_sha256, ) existing_version_id = self._identity_versions.get(identity) if existing_version_id is not None: version = self._versions[existing_version_id] job.status = "succeeded" job.business_date = envelope.business_date if review_case is not None: review_case.status = "completed" outcome = IngestionOutcome( status="already_committed", job_id=envelope.job_id, business_date=envelope.business_date, daily_version_id=version.id, version_no=version.version_no, ) self._callbacks[envelope.delivery_id] = ( delivery.envelope_sha256, outcome, ) return outcome version_no = 1 + max( ( version.version_no for version in self._versions.values() if version.business_date == envelope.business_date ), default=0, ) version_id = self._next_version_id self._next_version_id += 1 previous_id = self._current.get(envelope.business_date) if previous_id is not None: self._versions[previous_id].status = "superseded" records = delivery.structured_payload.get("records") channels = delivery.structured_payload.get("channels") version = _MemoryDailyVersion( id=version_id, business_date=envelope.business_date, version_no=version_no, identity=identity, status="active", records=tuple(copy.deepcopy(records if isinstance(records, list) else [])), channels=tuple(copy.deepcopy(channels if isinstance(channels, list) else [])), ) self._versions[version_id] = version self._identity_versions[identity] = version_id self._current[envelope.business_date] = version_id job.status = "succeeded" job.business_date = envelope.business_date if review_case is not None: review_case.status = "completed" outcome = IngestionOutcome( status="committed", job_id=envelope.job_id, business_date=envelope.business_date, daily_version_id=version_id, version_no=version_no, ) self._callbacks[envelope.delivery_id] = ( delivery.envelope_sha256, outcome, ) return outcome def current_version_id(self, business_date: date) -> Optional[int]: with self._lock: return self._current.get(business_date) def version_status(self, version_id: int) -> Optional[str]: with self._lock: version = self._versions.get(version_id) return version.status if version else None def version_records(self, version_id: int) -> Tuple[Mapping[str, Any], ...]: with self._lock: version = self._versions.get(version_id) return tuple(copy.deepcopy(version.records)) if version else tuple() @staticmethod def _money(value: object, *, two_places: bool = False) -> Decimal: try: amount = Decimal(str(value)) except (InvalidOperation, ValueError): raise IngestionError("REVIEW_PRICE_INVALID", "人工价格格式无效") from None if not amount.is_finite() or amount < 0: raise IngestionError("REVIEW_PRICE_INVALID", "人工价格必须为非负金额") if two_places: if not re.fullmatch(r"(?:0|[1-9][0-9]{0,15})\.[0-9]{2}", str(value)): raise IngestionError("REVIEW_PRICE_INVALID", "人工价格必须使用两位小数字符串") return amount.quantize(Decimal("0.01")) return amount.normalize() @staticmethod def _decimal_text(value: Decimal) -> str: return format(value.normalize(), "f") @classmethod def _manual_price(cls, value: object) -> Decimal: if not isinstance(value, str) or MANUAL_PRICE_RE.fullmatch(value) is None: raise IngestionError("REVIEW_PRICE_INVALID", "人工价格必须是非负整数字符串") return cls._money(value).quantize(Decimal("0.01")) @classmethod def _review_price_text(cls, value: object) -> str: amount = cls._money(value) if amount != amount.to_integral_value(): raise IngestionError("DATABASE_STATE_INVALID", "人工价格不是整数") return format(amount.quantize(Decimal("1")), "f") @classmethod def _manifest_for(cls, job_id: str, case: _MemoryReviewCase) -> bytes: keys = sorted( ( str(item["company_key"]), str(item["rate_code"]), cls._money(item["effective_rate_amount"]), ) for item in case.items ) issues = [ { "company_key": company_key, "rate_code": rate_code, "effective_rate_amount": cls._decimal_text(amount), } for company_key, rate_code, amount in keys ] by_key = { ( str(item["company_key"]), str(item["rate_code"]), cls._money(item["effective_rate_amount"]), ): item for item in case.items } overrides = [] for company_key, rate_code, amount in keys: price = cls._money(by_key[(company_key, rate_code, amount)]["real_price"], two_places=True) overrides.append( { "company_key": company_key, "rate_code": rate_code, "effective_rate_amount": cls._decimal_text(amount), "real_price": format(price, ".2f"), } ) payload = { "review_version": "1.0", "review_case_id": case.case_id, "job_id": job_id, "source_sha256": case.source_sha256, "business_date": case.business_date.isoformat(), "processor_version": case.processor_version, "rule_set_sha256": case.rule_set_sha256, "issues": issues, "overrides": overrides, } return json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False, ).encode("utf-8") @classmethod def _review_response(cls, case: _MemoryReviewCase, limit: int, offset: int) -> Dict[str, Any]: if limit < 1 or offset < 0: raise IngestionError("REVIEW_PAGE_INVALID", "复核分页参数无效") total = len(case.items) sliced = case.items[offset : offset + limit] completed = sum(1 for item in case.items if item["real_price"] is not None) safe_items = [] for item in sliced: safe_items.append( { "item_id": item["item_id"], "company_key": item["company_key"], "rate_code": item["rate_code"], "effective_rate_amount": item["effective_rate_amount"], "candidate_prices": copy.deepcopy(item["candidate_prices"]), "affected_records": item["affected_records"], "affected_rooms": item["affected_rooms"], "affected_room_nights": item["affected_room_nights"], "real_price": ( cls._review_price_text(item["real_price"]) if item["real_price"] is not None else None ), "revision": item["revision"], } ) return { "case_id": case.case_id, "case_status": case.status, "revision": case.revision, "business_date": case.business_date.isoformat(), "completed_items": completed, "total_items": total, "items": safe_items, "pagination": { "limit": limit, "offset": offset, "total": total, "has_next": offset + len(safe_items) < total, }, } def get_price_review(self, job_id: str, limit: int, offset: int) -> Dict[str, Any]: with self._lock: case = self._reviews.get(job_id) if case is None: raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在") return self._review_response(case, limit, offset) 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]: del actor_username with self._lock: case = self._reviews.get(job_id) if case is None or case.case_id != case_id: raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在") if case.status != "open": raise IngestionError("REVIEW_IMMUTABLE", "复核清单已冻结，不能修改") if revision != case.revision: raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新") item = next((value for value in case.items if value["item_id"] == item_id), None) if item is None: raise IngestionError("REVIEW_ITEM_NOT_FOUND", "待人工处理项不存在") price = self._manual_price(real_price) item["real_price"] = format(price, ".2f") case.revision += 1 item["revision"] = case.revision return self._review_response(case, len(case.items), 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: del actor_username with self._lock: job = self._jobs.get(job_id) case = self._reviews.get(job_id) if job is None or case is None or case.case_id != case_id: raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在") if case.processor_version != processor_version or case.rule_set_sha256 != rule_set_sha256: raise IngestionError("REVIEW_RULESET_CHANGED", "处理器或规则已变更，请取消后重新上传") if case.status == "completed": version_id = self._current.get(case.business_date) version = self._versions.get(version_id) if version_id is not None else None return ReviewGeneration( "completed", job_id, case.case_id, None, case.business_date, None, case.manifest_sha256, version.id if version else None, version.version_no if version else None ) if case.status == "processing": return ReviewGeneration( "processing", job_id, case.case_id, job.registration.attempt_no, case.business_date, None, case.manifest_sha256 ) if case.status not in {"open", "generation_failed"}: raise IngestionError("REVIEW_NOT_OPEN", "复核任务不能确认生成") if revision != case.revision: raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新") if any(item["real_price"] is None for item in case.items): raise IngestionError("REVIEW_INCOMPLETE", "请先填写全部缺价项") if case.manifest_bytes is None: case.manifest_bytes = self._manifest_for(job_id, case) case.manifest_sha256 = hashlib.sha256(case.manifest_bytes).hexdigest() next_attempt = job.registration.attempt_no + 1 job.registration = replace( job.registration, attempt_no=next_attempt, idempotency_key=idempotency_key, ) job.status = "queued" case.status = "processing" return ReviewGeneration( "ready", job_id, case.case_id, next_attempt, case.business_date, case.manifest_bytes, case.manifest_sha256, source=job.registration.source, ) def record_price_review_generation_failure( self, job_id: str, case_id: str, attempt_no: int, failure_code: str, ) -> None: del attempt_no, failure_code with self._lock: job = self._jobs.get(job_id) case = self._reviews.get(job_id) if job is None or case is None or case.case_id != case_id: raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在") if case.status == "completed": return case.status = "generation_failed" job.status = "awaiting_review" def cancel_price_review( self, job_id: str, case_id: str, revision: int, actor_username: str, ) -> Dict[str, Any]: del actor_username with self._lock: job = self._jobs.get(job_id) case = self._reviews.get(job_id) if job is None or case is None or case.case_id != case_id: raise IngestionError("REVIEW_NOT_FOUND", "待人工处理任务不存在") if case.status not in {"open", "generation_failed"}: raise IngestionError("REVIEW_NOT_CANCELLABLE", "复核任务当前不能取消") if revision != case.revision: raise IngestionError("REVIEW_REVISION_CONFLICT", "复核清单已被其他会话更新") case.status = "cancelled" case.revision += 1 job.status = "cancelled" return self._review_response(case, len(case.items), 0)