#!/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/028-Nguoi-Duoc-Goi-Ten-Cuoi-Cung')
RUN_ID = 'run-20260722T063856Z-87753fd5'
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()[:650]).casefold()
    last = ' '.join(text.split()[-1500:]).casefold()
    climax_marker = fold.rfind('đồng hồ trên cánh gà bắt đầu đếm ngược')
    climax = fold[climax_marker:] if climax_marker >= 0 else last
    return {
        'gate0_hook_refusal': phrase_present(first, ['tôi sẽ không đọc bản này']) and phrase_present(first, ['để họ tự nói tên mình']),
        'gate1_narrator_early': phrase_present(first, ['tống vãn nghi', 'vãn nghi']),
        'gate1_place_and_stakes': phrase_present(first, ['lễ tốt nghiệp', 'hội trường']) and phrase_present(first, ['danh sách']),
        'accessibility_agency': phrase_present(fold, ['để tôi tự nói']) and phrase_present(fold, ['hà dịch thanh']),
        'consent_contract': phrase_present(fold, ['bạn có quyền không kể lý do']) and phrase_present(fold, ['consent']),
        'multiple_valid_paths': phrase_present(fold, ['một hệ thống có nhiều đường hợp lệ']),
        'midpoint_empty_seat': phrase_present(fold, ['ghế số hai mươi bảy']) and phrase_present(fold, ['tống vãn nghi']),
        'mutual_vulnerability': phrase_present(fold, ['tôi ghi nhầm giờ']) and phrase_present(fold, ['mẹ tôi mổ cột sống']),
        'female_accountability': phrase_present(fold, ['tôi đã sai']) and phrase_present(fold, ['không trao cho tôi quyền đại diện']),
        'consent_withdrawal_respected': phrase_present(fold, ['mười hai người rút consent']) and phrase_present(fold, ['không bị đưa trở lại']),
        'mature_redesign': phrase_present(fold, ['từng người tự nói tên']) and phrase_present(fold, ['không kể lý lịch']),
        'female_climax_choice': phrase_present(climax, ['tôi sẽ không đọc bản này']) and phrase_present(climax, ['tôi từ chối dùng giọng mình']),
        'male_climax_cost': phrase_present(climax, ['anh tắt đồng hồ']) and phrase_present(climax, ['chịu toàn bộ trách nhiệm']),
        'self_naming_payoff': phrase_present(climax, ['hà dịch thanh']) and phrase_present(climax, ['tống vãn nghi']),
        'procedural_consequence': phrase_present(last, ['nhà trường lập biểu mẫu mới']) and phrase_present(last, ['không ai phải nộp lý do cá nhân']),
        'no_instant_romance': phrase_present(last, ['một tuần sau']) and phrase_present(last, ['người tôi muốn tìm hiểu']),
        'self_naming_coda': phrase_present(last, ['tự cầm micrô']) and phrase_present(last, ['tôi ở đây']),
    }


def corpus_files() -> list[Path]:
    paths = sorted(set(CORPUS_ROOT.glob('*/story/story-canon.txt')) | set(CORPUS_ROOT.glob('*/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 full_name, info in ledger['identity_registry'].items():
        allowed_names.append(full_name)
        allowed_names.extend(info['aliases_allowed'])
        forbidden_aliases.extend(info['forbidden'])
    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 = ledger['forbidden_residue']
    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': 'Ngôn tình học đường / thanh xuân',
        'story_family_secondary': 'Chữa lành',
        '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',
            'secondary_healing_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': 'Ngôn tình học đường / thanh xuân',
        'story_family_secondary': 'Chữa lành',
        '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': 'Ngôn tình học đường / thanh xuân',
        'secondary': 'Chữa lành',
        'locked': True,
        'revision': 3,
    }
    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()