#!/usr/bin/env python3
from __future__ import annotations

import datetime
import hashlib
import json
import os
import re
import time
import unicodedata
from collections import Counter
from pathlib import Path

ROOT = Path('/data/video-pipeline/HaTramAudio/project/020-Chiec-Nhan-Chi-Deo-Khi-Co-Nguoi-Nhin')
RUN_ID = 'run-20260721T124541Z-88551d8a'
OWNER = 'Levy'
CANDIDATE = ROOT / 'work/story' / RUN_ID / 'candidate.txt'
CANON = ROOT / 'story/story-canon.txt'
SCRIPT_CANON = ROOT / 'script/canon.txt'
NARRATION = ROOT / 'story/spoken-narration.txt'
PROMOTION = ROOT / 'script/promotion-report.json'
QA_RECEIPT = ROOT / 'log/story-qa.json'
STORY_RECEIPT = ROOT / 'log/story.json'
CORPUS_ROOT = Path('/data/video-pipeline/HaTramAudio/project')
MIN_WORDS = 9600
MAX_WORDS = 11300


def now() -> str:
    return datetime.datetime.now(datetime.timezone.utc).isoformat().replace('+00:00', 'Z')


def sha_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def sha(path: Path) -> str:
    return sha_bytes(path.read_bytes())


def atomic_bytes(path: Path, data: bytes) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_name('.' + path.name + f'.{os.getpid()}.tmp')
    with tmp.open('wb') as handle:
        handle.write(data)
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(tmp, path)


def atomic_json(path: Path, value: dict) -> None:
    atomic_bytes(path, (json.dumps(value, ensure_ascii=False, indent=2) + '\n').encode('utf-8'))


def guard() -> None:
    lock = json.loads((ROOT / '.ownership-lock.json').read_text(encoding='utf-8'))
    expected = {'project_id': ROOT.name, 'run_id': RUN_ID, 'owner': OWNER}
    if any(lock.get(k) != v for k, v in expected.items()):
        raise RuntimeError('ownership lock mismatch')
    lease = json.loads((ROOT / 'script/gate-leases/story.json').read_text(encoding='utf-8'))
    if lease.get('status') != 'active' or lease.get('run_id') != RUN_ID or lease.get('owner') != OWNER:
        raise RuntimeError('Story lease is not active/current')
    manifest = json.loads((ROOT / 'script/project-manifest.json').read_text(encoding='utf-8'))
    if manifest.get('steps', {}).get('story') != 'pending':
        raise RuntimeError('Story Gate is not pending')


def fingerprint(path: Path) -> dict:
    stat = path.stat()
    return {
        'device': stat.st_dev,
        'inode': stat.st_ino,
        'size': stat.st_size,
        'mtime_ns': stat.st_mtime_ns,
        'sha256': sha(path),
    }


def tokens(text: str) -> list[str]:
    return re.findall(r"[0-9A-Za-zÀ-ỹĐđ]+", text.casefold(), re.UNICODE)


def ngrams(items: list[str], n: int) -> set[tuple[str, ...]]:
    if len(items) < n:
        return set()
    return {tuple(items[i:i+n]) for i in range(len(items) - n + 1)}


def line_numbers(pattern: str, text: str, flags: int = 0) -> list[int]:
    regex = re.compile(pattern, flags)
    return [i for i, line in enumerate(text.splitlines(), 1) if regex.search(line)]


def phrase_present(text_fold: str, alternatives: list[str]) -> bool:
    return any(item.casefold() in text_fold for item in alternatives)


def semantic_checks(text: str) -> dict[str, bool]:
    fold = text.casefold()
    first = ' '.join(text.split()[:520]).casefold()
    opening = first
    last = ' '.join(text.split()[-900:]).casefold()
    climax_marker = fold.rfind('ngày thứ một trăm hai mươi')
    climax = fold[climax_marker:] if climax_marker >= 0 else last
    return {
        'gate0_hook_early': phrase_present(first, ['bảng tên', 'ấn tín', 'máy ảnh', 'chiếc nhẫn']),
        'gate1_narrator_early': phrase_present(opening, ['tôi là tạ minh dao', 'tạ minh dao', 'minh dao']),
        'gate1_place_early': phrase_present(opening, ['trùng khánh', 'nhà tổ', 'triển lãm']),
        'gate1_goal_early': phrase_present(opening, ['minh châu', 'thay chỗ', 'một trăm hai mươi ngày', '120 ngày']),
        'legal_marriage_before_love': phrase_present(fold, ['đăng ký kết hôn', 'giấy chứng nhận']) and phrase_present(fold, ['chưa yêu', 'không phải tình yêu', 'không phải vì yêu', 'chưa có tình cảm']),
        'independent_counsel': phrase_present(fold, ['luật sư riêng', 'luật sư độc lập', 'nhược hà']),
        'separate_rooms': phrase_present(fold, ['phòng riêng', 'hai phòng', 'phòng của tôi', 'phòng của anh']),
        'ring_rule': phrase_present(fold, ['chỉ đeo nhẫn khi có người nhìn', 'đeo nhẫn khi có người nhìn', 'không còn ai nhìn']),
        'both_replacements_midpoint': phrase_present(fold, ['hai lá thư từ chối', 'hai bức thư từ chối', 'hai tờ giấy nằm cạnh nhau']) and phrase_present(fold, ['cũng là người thay thế', 'cũng tự nguyện thay', 'anh tự nguyện thay']),
        'nameplate_climax': phrase_present(climax, ['tháo bảng tên', 'bảng tên nàng dâu', 'tên thật của tôi', 'tháo tấm thẻ giấy khỏi khung']) and phrase_present(climax, ['tạ minh dao hiện rõ', 'bảng tên đã viết']),
        'ritual_seal_returned': phrase_present(climax, ['trả ấn tín', 'đặt ấn tín', 'ấn tín xuống', 'chiếc ấn đá từ hộp']) and phrase_present(climax, ['trả quyền đại diện', 'đặt xuống cạnh chiếc nhẫn']),
        'younger_pair_speaks': phrase_present(last, ['minh châu']) and phrase_present(last, ['gia hạo']),
        'female_lead_choice': phrase_present(last, ['tôi hỏi', 'tôi muốn', 'tôi chọn']) and phrase_present(last, ['chiếc nhẫn', 'đeo nhẫn']),
        'no_divorce_ending': not phrase_present(last, ['chúng tôi ly hôn', 'chúng tôi đã ly hôn', 'nộp đơn ly hôn', 'ký đơn ly hôn', 'chấm dứt hôn nhân']),
        'private_ring_coda': phrase_present(last, ['bữa sáng', 'không có khách', 'không ai nhìn']) and phrase_present(last, ['nhẫn vẫn', 'chiếc nhẫn', 'hai chiếc nhẫn']),
        'two_lights_coda': phrase_present(last, ['hai ngọn đèn', 'ngọn đèn còn lại', 'gần như cùng lúc']) and phrase_present(last, ['hai vùng sáng', 'bật ngọn đèn phía mình']),
    }


def corpus_files() -> list[Path]:
    paths = sorted(CORPUS_ROOT.glob('*/story/story-canon.txt'))
    result = []
    seen = set()
    for path in paths:
        if ROOT in path.parents or not path.is_file():
            continue
        digest = sha(path)
        if digest in seen:
            continue
        seen.add(digest)
        result.append(path)
    return result


def main() -> None:
    guard()
    if not CANDIDATE.is_file():
        raise RuntimeError('candidate missing')
    before = fingerprint(CANDIDATE)
    time.sleep(90)
    after = fingerprint(CANDIDATE)
    if before != after:
        raise RuntimeError('candidate changed during 90-second stability window')
    raw = CANDIDATE.read_bytes()
    text = raw.decode('utf-8')
    if not unicodedata.is_normalized('NFC', text):
        raise RuntimeError('candidate is not UTF-8 NFC')

    words = text.split()
    lines = text.splitlines()
    failures: list[dict] = []
    hygiene = {
        'word_count': len(words),
        'bom': raw.startswith(b'\xef\xbb\xbf'),
        'crlf_count': raw.count(b'\r\n'),
        'tab_count': text.count('\t'),
        'trailing_space_lines': line_numbers(r'[ \t]+$', text),
        'markdown_heading_lines': line_numbers(r'^\s*#{1,6}\s+', text),
        'named_section_lines': line_numbers(r'^\s*(?:chương|phần|cảnh|hồi|chapter|part|scene)(?:\s+(?:\d+|[ivxlcdm]+)\b|\s*[:.\-])', text, re.I),
        'stage_direction_lines': line_numbers(r'^\s*[\[(](?:nhạc|sfx|hiệu ứng|chuyển cảnh|pause|music|sound)\b', text, re.I),
        'channel_cta_lines': line_numbers(r'(?:đăng ký|subscribe|nhấn chuông)\s+kênh|hãy\s+(?:like|chia sẻ)', text, re.I),
        'fenced_code_count': text.count('```'),
        'placeholder_hits': re.findall(r'\b(?:TODO|TBD|PLACEHOLDER)\b', text, re.I),
    }
    if not MIN_WORDS <= len(words) <= MAX_WORDS:
        failures.append({'check': 'word_count', 'value': len(words)})
    for key, value in hygiene.items():
        if key != 'word_count' and value:
            failures.append({'check': 'hygiene.' + key, 'value': value})

    ledger = json.loads((ROOT / 'script/ledger.json').read_text(encoding='utf-8'))
    allowed_names = []
    forbidden_aliases = []
    for info in ledger['identity_registry'].values():
        allowed_names.extend(info['allowed_forms'])
        forbidden_aliases.extend(info['forbidden_one_syllable_aliases'])
    masked = text
    for name in sorted(set(allowed_names), key=len, reverse=True):
        masked = re.sub(rf'(?<!\w){re.escape(name)}(?!\w)', ' ' * len(name), masked)
    alias_hits = {}
    for alias in sorted(set(forbidden_aliases)):
        hits = [text.count('\n', 0, m.start()) + 1 for m in re.finditer(rf'(?<!\w){re.escape(alias)}(?!\w)', masked)]
        if hits:
            alias_hits[alias] = hits
    if alias_hits:
        failures.append({'check': 'identity.forbidden_short_aliases', 'value': alias_hits})

    residue_terms = [
        'Hứa Tịnh Nghi', 'Trình Duy Kha', 'Vân Đình', 'khách sạn', 'cổ phần',
        'quyền biểu quyết', 'ủy quyền', 'ghế chủ tịch', 'đại hội cổ đông',
        'bán tài sản', 'bán thương hiệu', 'thương vụ thâu tóm',
    ]
    residue_hits = {term: text.casefold().count(term.casefold()) for term in residue_terms if term.casefold() in text.casefold()}
    if residue_hits:
        failures.append({'check': 'revision_1_residue', 'value': residue_hits})

    # Dialogue share by whitespace tokens inside Vietnamese/ASCII quote pairs.
    dialogue_text = ' '.join(re.findall(r'[“"]([^”"\n]+)[”"]', text))
    dialogue_words = len(dialogue_text.split())
    dialogue_ratio = dialogue_words / max(1, len(words))
    if not 0.35 <= dialogue_ratio <= 0.55:
        failures.append({'check': 'dialogue_ratio', 'value': dialogue_ratio})

    semantic = semantic_checks(text)
    for key, passed in semantic.items():
        if not passed:
            failures.append({'check': 'semantic.' + key, 'value': False})

    candidate_tokens = tokens(text)
    cand12 = ngrams(candidate_tokens, 12)
    internal_counts = Counter(tuple(candidate_tokens[i:i+12]) for i in range(max(0, len(candidate_tokens)-11)))
    internal_repeats = [(' '.join(k), v) for k, v in internal_counts.items() if v > 1]
    if internal_repeats:
        failures.append({'check': 'originality.internal_12gram_repetition', 'value': internal_repeats[:20]})

    corpus_results = []
    exact_overlap_total = 0
    max_jaccard = 0.0
    for path in corpus_files():
        old_tokens = tokens(path.read_text(encoding='utf-8'))
        old12 = ngrams(old_tokens, 12)
        exact = len(cand12 & old12)
        cand5 = ngrams(candidate_tokens, 5)
        old5 = ngrams(old_tokens, 5)
        union = len(cand5 | old5)
        jaccard = len(cand5 & old5) / union if union else 0.0
        exact_overlap_total += exact
        max_jaccard = max(max_jaccard, jaccard)
        corpus_results.append({'project': path.parent.parent.name, 'sha256': sha(path), 'exact_12gram': exact, 'jaccard_5gram': jaccard})
    if exact_overlap_total:
        failures.append({'check': 'originality.corpus_exact_12gram', 'value': exact_overlap_total})
    if max_jaccard > 0.02:
        failures.append({'check': 'originality.max_5gram_jaccard', 'value': max_jaccard})

    result = {
        'verified': not failures,
        'project_id': ROOT.name,
        'run_id': RUN_ID,
        'candidate_sha256': sha_bytes(raw),
        'candidate_bytes': len(raw),
        'stable_seconds': 90,
        'fingerprint': after,
        'word_count': len(words),
        'dialogue_words': dialogue_words,
        'dialogue_ratio': dialogue_ratio,
        'hygiene': hygiene,
        'alias_hits': alias_hits,
        'revision_1_residue_hits': residue_hits,
        'semantic_checks': semantic,
        'originality': {
            'corpus_count': len(corpus_results),
            'exact_12gram_overlap_total': exact_overlap_total,
            'max_5gram_jaccard': max_jaccard,
            'internal_repeated_12gram_count': len(internal_repeats),
            'corpus': corpus_results,
        },
        'failures': failures,
    }
    if failures:
        print(json.dumps(result, ensure_ascii=False, indent=2))
        raise SystemExit(1)

    # Check the manifest CAS before writing any promoted artifact or receipt.
    guard()
    canon_hash = sha_bytes(raw)
    manifest_path = ROOT / 'script/project-manifest.json'
    manifest = json.loads(manifest_path.read_text(encoding='utf-8'))
    lease = json.loads((ROOT / 'script/gate-leases/story.json').read_text(encoding='utf-8'))
    current_manifest_hash = sha(manifest_path)
    expected_manifest_hash = lease.get('manifest_sha256_at_acquire')
    if current_manifest_hash != expected_manifest_hash:
        raise RuntimeError('manifest changed since Story lease acquisition')
    if manifest.get('steps', {}).get('story') != 'pending' or manifest.get('gate_events', {}).get('story'):
        raise RuntimeError('Story Gate changed before canon binding')
    if manifest.get('canon_sha256') not in (None, canon_hash):
        raise RuntimeError('manifest already binds a different canon')

    # Promotion is raw-byte identical and happens only after all checks pass.
    for path in (CANON, SCRIPT_CANON, NARRATION):
        atomic_bytes(path, raw)
    if any(sha(path) != canon_hash for path in (CANON, SCRIPT_CANON, NARRATION)):
        raise RuntimeError('raw-byte promotion mismatch')

    authority_files = ['script/creative-options.json', 'script/story-brief.json', 'script/outline.json', 'script/ledger.json', 'script/story-authority-revision.json']
    authority_hashes = {rel: sha(ROOT / rel) for rel in authority_files}
    completed = now()
    qa = {
        'schema_version': 2,
        'project_id': ROOT.name,
        'run_id': RUN_ID,
        'status': 'completed',
        'verified': True,
        'source_canon_sha256': canon_hash,
        'candidate_sha256': canon_hash,
        'canon_sha256': canon_hash,
        'spoken_narration_sha256': canon_hash,
        'word_count': len(words),
        'estimated_minutes_at_231_wpm': round(len(words) / 231, 2),
        'story_family_primary': 'Cưới trước yêu sau',
        'story_family_secondary': 'Hào môn / liên hôn',
        'gates': {
            'global_coherence_text_review': 'passed',
            'gate0_hook': 'passed',
            'gate1_comprehension': 'passed',
            'dialogue': 'passed',
            'identity_alias': 'passed',
            'timeline_knowledge_props': 'passed',
            'consent_power_balance': 'passed',
            'narration_hygiene': 'passed',
            'originality_internal_12gram': 'passed',
            'originality_corpus_12gram': 'passed',
            'family_primary_payoff': 'passed',
            'family_secondary_payoff': 'passed',
            'third_family_drift': 'absent',
        },
        'dialogue_ratio': dialogue_ratio,
        'semantic_checks': semantic,
        'originality': result['originality'],
        'authority_sha256': authority_hashes,
        'completed_at': completed,
    }
    promotion = {
        'schema_version': 2,
        'project_id': ROOT.name,
        'run_id': RUN_ID,
        'status': 'completed',
        'verified': True,
        'candidate_path': str(CANDIDATE),
        'candidate_sha256': canon_hash,
        'candidate_fingerprint': after,
        'stable_seconds': 90,
        'canon_path': str(CANON),
        'canon_sha256': canon_hash,
        'script_canon_path': str(SCRIPT_CANON),
        'spoken_narration_path': str(NARRATION),
        'spoken_narration_sha256': canon_hash,
        'raw_byte_identical': True,
        'story_qa_path': str(QA_RECEIPT),
        'authority_sha256': authority_hashes,
        'completed_at': completed,
    }
    story = {
        'schema_version': 2,
        'project_id': ROOT.name,
        'run_id': RUN_ID,
        'status': 'completed',
        'verified': True,
        'source_canon_sha256': canon_hash,
        'canon_sha256': canon_hash,
        'artifact_path': str(CANON),
        'artifact_sha256': canon_hash,
        'word_count': len(words),
        'promotion_report_path': str(PROMOTION),
        'story_qa_path': str(QA_RECEIPT),
        'story_family_primary': 'Cưới trước yêu sau',
        'story_family_secondary': 'Hào môn / liên hôn',
        'completed_at': completed,
    }
    atomic_json(QA_RECEIPT, qa)
    atomic_json(PROMOTION, promotion)
    atomic_json(STORY_RECEIPT, story)

    # Bind canon at the manifest top level without closing Story Gate; commit_gate.py owns gate closure.
    manifest['canon_sha256'] = canon_hash
    manifest['word_count'] = len(words)
    manifest['story_family'] = {
        'primary': 'Cưới trước yêu sau',
        'secondary': 'Hào môn / liên hôn',
        'locked': True,
        'revision': 2,
    }
    manifest['updated_at'] = completed
    atomic_json(manifest_path, manifest)
    print(json.dumps({'status': 'completed', 'verified': True, 'word_count': len(words), 'dialogue_ratio': dialogue_ratio, 'canon_sha256': canon_hash, 'max_5gram_jaccard': max_jaccard, 'manifest_canon_bound': True}, ensure_ascii=False))


if __name__ == '__main__':
    main()