#!/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/022-Tiem-Giat-Mo-Cua-Luc-Bon-Gio-Muoi-Bay')
RUN_ID = 'run-20260721T172022Z-9462493a'
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()
    last = ' '.join(text.split()[-1200:]).casefold()
    climax_marker = fold.rfind('tĩnh an đặt một chiếc giỏ thép')
    climax = fold[climax_marker:] if climax_marker >= 0 else last
    return {
        'gate0_hook_early': phrase_present(first, ['bốn giờ mười bảy', 'váy cưới', 'ảnh', 'một lần sửa quá khứ']),
        'gate1_narrator_early': phrase_present(first, ['tôi là giang vãn đường', 'giang vãn đường', 'vãn đường']),
        'gate1_place_early': phrase_present(first, ['hạnh phúc lý', 'thành đô', 'tiệm sửa áo cưới']),
        'gate1_goal_early': phrase_present(first, ['hợp đồng', 'giữ tiệm', 'giao xong chiếc váy']),
        'activation_rule': phrase_present(fold, ['bốn giờ mười bảy']) and phrase_present(fold, ['bốn giờ ba mươi']),
        'thirteen_minute_rule': phrase_present(fold, ['mười ba phút']) and phrase_present(fold, ['một hành động hoặc một câu nói']),
        'four_use_limit': phrase_present(fold, ['tối đa bốn lần', 'lần thứ tư']),
        'cost_rule': phrase_present(fold, ['mất một ký ức chung', 'mất vĩnh viễn một ký ức chung']),
        'first_cost_verified': phrase_present(fold, ['khăn tay màu xám', 'khăn xám']) and phrase_present(fold, ['cơn mưa đầu tiên', 'trú mưa']),
        'second_cost_verified': phrase_present(fold, ['tạp dề xám', 'tạp dề màu xanh']) and phrase_present(fold, ['bữa sáng đầu tiên']),
        'third_cost_verified': phrase_present(fold, ['dải ruy băng']) and phrase_present(fold, ['ngày chúng tôi bắt đầu yêu', 'ngày hai người xác nhận yêu']),
        'midpoint_self_reveal': phrase_present(fold, ['chính tôi đã mang chúng đến quầy', 'chính tôi đã dùng tiệm ba lần']),
        'female_lead_stops_machine': phrase_present(climax, ['lấy áo ra', 'nút trắng vẫn chưa sáng']) and phrase_present(climax, ['không giao dịch']),
        'truth_and_accountability': phrase_present(climax, ['tôi làm vì tôi sợ', 'tôi làm vậy vì tôi sợ']) and phrase_present(climax, ['không thể trả ký ức', 'ngừng sửa thêm']),
        'no_instant_forgiveness': phrase_present(climax, ['anh chưa thể tha thứ', 'không tha thứ ngay']),
        'consensual_rebuild': phrase_present(last, ['mỗi tuần một lần', 'tiếp tục gặp anh', 'người đang đi cạnh tôi']),
        'no_memory_reset': phrase_present(last, ['không ký ức nào quay lại', 'những khoảng trống ấy vẫn đau']) and not phrase_present(last, ['mọi ký ức trở lại', 'ký ức đã được khôi phục']),
        'blank_fourth_receipt_coda': phrase_present(last, ['ô phiếu thứ tư vẫn trống', 'ô thứ tư để trống']),
    }


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 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 = [
        'Tống Vãn Hà', 'Lục Thừa An', 'ga Bắc Hà', 'chuyến tàu cuối',
        'nhà ga cũ', 'kho đồ thất lạc', 'chìa khóa không có ổ khóa',
        'Tạ Minh Dao', 'Phó Cảnh Tuyên', 'Vân Đình', 'quyền biểu quyết',
    ]
    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': 'High-concept',
        'story_family_secondary': None,
        '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',
            'high_concept_rule_cost_limit_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': 'High-concept',
        'story_family_secondary': None,
        '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()