#!/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/025-Nguoi-Giu-Cho-Cuoi-Cung')
RUN_ID = 'run-20260721T232244Z-a0837ee4'
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('sáng hôm ấy, mười tám người dùng mẫu vẫn xuất hiện')
    climax = fold[climax_marker:] if climax_marker >= 0 else last
    return {
        'gate0_hook_fake_queue': phrase_present(first, ['xé danh sách mười tám người dùng mẫu', 'hàng chờ đẹp']),
        'gate1_narrator_early': phrase_present(first, ['lâm vãn ý', 'vãn ý']),
        'gate1_profession_early': phrase_present(first, ['giữ mười hai chỗ', 'nghề của tôi', 'dịch vụ thay thời gian']),
        'gate1_place_early': phrase_present(first, ['trung tâm hành chính minh châu']),
        'digital_queue_conflict': phrase_present(fold, ['mã hẹn', 'luồng thủ công']) and phrase_present(fold, ['lịch số']),
        'mrs_khuong_agency': phrase_present(fold, ['đừng nói qua cô', 'tôi muốn chính cậu ấy nói']) and phrase_present(fold, ['tôi không cần cô giữ chỗ lần này']),
        'research_consent_contract': phrase_present(fold, ['quyền dừng', 'không ghi âm']) and phrase_present(fold, ['dữ liệu ẩn danh', 'rút lời']),
        'role_swap_learning': phrase_present(fold, ['hôm nay đổi vai']) and phrase_present(fold, ['tiết lộ thông tin khách']),
        'accessibility_user_agency': phrase_present(fold, ['trình đọc màn hình']) and phrase_present(fold, ['hỏi tôi cần hoàn thành việc gì']),
        'midpoint_structural_reveal': phrase_present(fold, ['sáu mươi tám phần trăm']) and phrase_present(fold, ['khoảng cách số']) and phrase_present(fold, ['ba mươi bảy giao dịch']),
        'female_accountability': phrase_present(fold, ['hoàn tiền', 'danh sách người đang hoạt động']) and phrase_present(fold, ['tất cả khoản chênh']),
        'rupture_has_boundary': phrase_present(fold, ['tôi không muốn làm việc với anh hôm nay']) and phrase_present(fold, ['bao lâu?']),
        'community_redesign': phrase_present(fold, ['quầy hỗ trợ cộng đồng']) and phrase_present(fold, ['hội đồng người dùng']),
        'female_refuses_staged_queue': phrase_present(climax, ['không có hàng mẫu từ chúng tôi', 'ngừng thuê người đóng vai']),
        'male_shuts_dashboard': phrase_present(climax, ['anh tắt dashboard', 'dữ liệu đang được phân loại lại theo nguồn']),
        'public_truth_and_cost': phrase_present(climax, ['tôi đang thừa nhận phần trách nhiệm của tôi']) and phrase_present(climax, ['hợp đồng của cảnh thâm bị tạm dừng']),
        'no_instant_romance': phrase_present(last, ['sau khi nộp báo cáo', 'dự án đã chuyển giao']) and phrase_present(last, ['thứ bảy']),
        'self_queue_coda': phrase_present(last, ['tôi tự bước tới', 'tự lấy số', 'phiếu số b một trăm mười bảy']) and phrase_present(last, ['đợi ở đây']),
    }


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 = [
        'phục dựng mùi hương', 'lọ nước hoa cuối cùng', 'di chúc khứu giác',
        'Tạ Minh Dao', 'Phó Cảnh Tuyên', 'Vân Đình', 'quyền biểu quyết',
        'ga Bắc Hà', 'chìa khóa không có ổ khóa', 'tiệm giặt mười ba phú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': 'Oan gia nghề nghiệp',
        'story_family_secondary': 'Chữa lành đời thường',
        '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': 'Oan gia nghề nghiệp',
        'story_family_secondary': 'Chữa lành đời thường',
        '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()