"""auto_approve.py — 자동 승인 5조건 검증 콘텐츠가 자동 승인 가능한지 5조건을 검증한다. 1. 사전 승인 템플릿 20종 매칭 여부 2. 저작권 안전 목록 체크 3. thread-hook-formula 체크리스트 통과 (5감정 중 2개+) 4. 블랙리스트 미저촉 5. 72시간 중복 아님 """ import json import hashlib import time from dataclasses import dataclass, field from pathlib import Path from typing import Optional from .hook_scorer import score as hook_score, check_blacklist_keywords PIPELINE_DIR = Path(__file__).parent TEMPLATES_DIR = PIPELINE_DIR / "templates" APPROVED_DIR = PIPELINE_DIR / "approved" PUBLISHED_DIR = PIPELINE_DIR / "published" @dataclass class ApprovalResult: """승인 결과""" condition_1_template_match: bool = False matched_template: Optional[str] = None condition_2_copyright_safe: bool = False condition_3_hook_pass: bool = False hook_emotion_count: int = 0 condition_4_blacklist_clear: bool = False blacklist_violations: list[str] = field(default_factory=list) condition_5_no_duplicate: bool = False duplicate_hash: Optional[str] = None all_passed: bool = False fail_reasons: list[str] = field(default_factory=list) # 저작권 안전 소스 목록 COPYRIGHT_SAFE_SOURCES = [ "자체 작성", "금융감독원 공시", "보험개발원 통계", "국민건강보험공단", "생명보험협회", "손해보험협회", "금융위원회 보도자료", "한국은행 통계", "통계청 자료", "공공데이터포털", ] def _load_templates() -> list[dict]: """사전 승인 템플릿 목록 로드""" templates = [] if not TEMPLATES_DIR.exists(): return templates for f in sorted(TEMPLATES_DIR.glob("*.json")): with open(f, "r", encoding="utf-8") as fp: templates.append(json.load(fp)) return templates def _content_hash(text: str) -> str: """콘텐츠 해시 생성 (중복 체크용)""" normalized = text.strip().lower() return hashlib.sha256(normalized.encode("utf-8")).hexdigest()[:16] def _load_recent_hashes(hours: int = 72) -> set[str]: """최근 N시간 내 게시된 콘텐츠 해시 로드""" hashes: set[str] = set() cutoff = time.time() - (hours * 3600) for directory in [APPROVED_DIR, PUBLISHED_DIR]: if not directory.exists(): continue for f in directory.glob("*.json"): try: with open(f, "r", encoding="utf-8") as fp: data = json.load(fp) created_at = data.get("created_at", 0) if created_at > cutoff: content_hash = data.get("content_hash", "") if content_hash: hashes.add(content_hash) except (json.JSONDecodeError, KeyError): continue return hashes def check_template_match(text: str, templates: Optional[list[dict]] = None) -> tuple[bool, Optional[str]]: """조건 1: 사전 승인 템플릿 매칭 여부 텍스트가 사전 승인된 20종 템플릿 중 하나와 매칭되는지 확인. 매칭 기준: 텍스트에 템플릿의 required_keywords가 모두 포함 """ if templates is None: templates = _load_templates() for tpl in templates: tpl_id = tpl.get("id", "unknown") required_keywords = tpl.get("required_keywords", []) if not required_keywords: continue if all(kw in text for kw in required_keywords): return True, tpl_id return False, None def check_copyright_safe(source: str) -> bool: """조건 2: 저작권 안전 목록 체크 콘텐츠 소스가 저작권 안전 목록에 포함되는지 확인. """ source_lower = source.strip().lower() if source else "" for safe in COPYRIGHT_SAFE_SOURCES: if safe.lower() in source_lower or source_lower in safe.lower(): return True # "자체 작성"은 항상 안전 if "자체" in source_lower or "원본" in source_lower or "직접" in source_lower: return True return False def check_hook_formula(text: str) -> tuple[bool, int]: """조건 3: thread-hook-formula 체크리스트 통과 5감정 중 최소 2개 이상 포함 여부 판정. """ result = hook_score(text) return result.has_min_emotions, result.emotion_count def check_blacklist(text: str) -> tuple[bool, list[str]]: """조건 4: 블랙리스트 미저촉 블랙리스트 키워드 및 회사명 저촉 여부 확인. Returns: (통과 여부, 위반 키워드 목록) - True = 미저촉(통과) """ has_violation, violations = check_blacklist_keywords(text) return not has_violation, violations def check_no_duplicate(text: str, hours: int = 72) -> tuple[bool, Optional[str]]: """조건 5: 72시간 중복 아님 최근 72시간 내 동일/유사 콘텐츠가 게시되지 않았는지 확인. """ content_hash = _content_hash(text) recent_hashes = _load_recent_hashes(hours) is_duplicate = content_hash in recent_hashes return not is_duplicate, content_hash if is_duplicate else None def approve(text: str, source: str = "자체 작성") -> ApprovalResult: """5조건 종합 검증 Args: text: 콘텐츠 텍스트 source: 콘텐츠 소스 (저작권 확인용) Returns: ApprovalResult 검증 결과 """ result = ApprovalResult() # 조건 1: 템플릿 매칭 result.condition_1_template_match, result.matched_template = check_template_match(text) # 조건 2: 저작권 안전 result.condition_2_copyright_safe = check_copyright_safe(source) # 조건 3: 훅 공식 통과 result.condition_3_hook_pass, result.hook_emotion_count = check_hook_formula(text) # 조건 4: 블랙리스트 미저촉 result.condition_4_blacklist_clear, result.blacklist_violations = check_blacklist(text) # 조건 5: 72시간 중복 아님 result.condition_5_no_duplicate, result.duplicate_hash = check_no_duplicate(text) # 종합 판정 result.fail_reasons = [] if not result.condition_1_template_match: result.fail_reasons.append("사전 승인 템플릿과 매칭되지 않음") if not result.condition_2_copyright_safe: result.fail_reasons.append(f"저작권 안전 목록 미포함: {source}") if not result.condition_3_hook_pass: result.fail_reasons.append( f"훅 공식 미통과: 감정 {result.hook_emotion_count}개 (최소 2개 필요)" ) if not result.condition_4_blacklist_clear: result.fail_reasons.append( f"블랙리스트 저촉: {', '.join(result.blacklist_violations)}" ) if not result.condition_5_no_duplicate: result.fail_reasons.append("72시간 내 중복 콘텐츠 존재") result.all_passed = len(result.fail_reasons) == 0 return result if __name__ == "__main__": sample_text = ( "7월부터 수수료 구조가 완전히 바뀌는데, 아직도 모르는 설계사가 47,382명이나 된다니. " "이거 겪어본 사람? 안 하면 진짜 큰일남." ) result = approve(sample_text, source="자체 작성") print(f"자동 승인: {result.all_passed}") for i, reason in enumerate(result.fail_reasons, 1): print(f" 실패 {i}: {reason}")