"""발행 검수 게이트 — 사이트가 나가기 전에 반드시 통과해야 하는 검사.""" from dataclasses import dataclass, field from common.category_schema import get_schema from common.enums import PUBLISHABLE_FACT_STATUSES, FactStatus, PlaceCategory, PublishRejectReason @dataclass class GateResult: """검수 결과.""" passed: bool reason: PublishRejectReason | None = None detail: dict = field(default_factory=dict) def as_log(self) -> dict: return {"reason": self.reason.name if self.reason else None, **self.detail} def check_facts_verified(facts: list) -> GateResult: """규칙 1 — 노출 대상 fact 가 전부 검증됐는가.""" bad = [ {"key": f["key"] if isinstance(f, dict) else f.key, "status": FactStatus(f["status"] if isinstance(f, dict) else f.status).name} for f in facts if FactStatus(f["status"] if isinstance(f, dict) else f.status) not in PUBLISHABLE_FACT_STATUSES ] if bad: return GateResult(False, PublishRejectReason.UNVERIFIED_FACT, {"unverified": bad[:20], "count": len(bad)}) return GateResult(True) def check_required_fields(category: PlaceCategory, facts: list) -> GateResult: """업종 스키마의 required 필드가 다 있는가.""" schema = get_schema(category) required = set(schema.required_keys("place")) have = { (f["key"] if isinstance(f, dict) else f.key) for f in facts if str(f["value"] if isinstance(f, dict) else f.value or "").strip() } missing = sorted(required - have) if missing: return GateResult( False, PublishRejectReason.REQUIRED_FACT_MISSING, {"missing": missing, "labels": [schema.get(k).label for k in missing if schema.get(k)]}, ) return GateResult(True) def check_unique_content(unique_content_count: int | None) -> GateResult: """렌더러가 '고유 콘텐츠 0건' 으로 거부했을 때 사유 코드를 붙이는 자리.""" if unique_content_count is None: return GateResult(True) if unique_content_count <= 0: return GateResult(False, PublishRejectReason.NO_UNIQUE_CONTENT, {"unique_content_count": unique_content_count}) return GateResult(True) def check_jsonld_matches(mismatches: list) -> GateResult: """규칙 3 — 구조화 데이터 값이 화면 값과 같은가.""" if mismatches: return GateResult(False, PublishRejectReason.JSONLD_MISMATCH, {"mismatches": list(mismatches)[:20]}) return GateResult(True) def evaluate( category: PlaceCategory, facts: list, unique_content_count: int | None, mismatches: list ) -> GateResult: """게이트 전체.""" for result in ( check_facts_verified(facts), check_jsonld_matches(mismatches), ): if not result.passed: return result # 통과했지만 비어 있는 필수 항목은 경고로 실어 보낸다(발행은 진행된다). required = check_required_fields(category, facts) if not required.passed: return GateResult(True, None, {"warning": PublishRejectReason.REQUIRED_FACT_MISSING.name, **(required.detail or {})}) return GateResult(True)