import ast
import hashlib
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]


def _string_constants(path: Path) -> dict[str, str]:
    tree = ast.parse(path.read_text(encoding="utf-8"))
    values: dict[str, str] = {}
    for node in tree.body:
        if not isinstance(node, ast.Assign) or len(node.targets) != 1:
            continue
        target = node.targets[0]
        if isinstance(target, ast.Name) and isinstance(node.value, ast.Constant):
            if isinstance(node.value.value, str):
                values[target.id] = node.value.value
    return values


def test_remote_model_and_speaker_encoder_are_revision_pinned():
    app_path = ROOT / "app.py"
    source = app_path.read_text(encoding="utf-8")
    constants = _string_constants(app_path)

    assert constants["MODEL_REVISION"] == "aaf1a0878e37875382bb0e5c8a3a2ba43be67297"
    assert constants["ECAPA_REVISION"] == "0f99f2d0ebe89ac095bcc5903c4dd8f72b367286"
    assert "snapshot_download(REPO_ID, revision=MODEL_REVISION)" in source
    assert "snapshot_download(ECAPA_REPO_ID, revision=ECAPA_REVISION)" in source
    assert "source=ECAPA_DIR" in source
    assert 'overrides={"pretrained_path": ECAPA_DIR}' in source


def test_tts_runtime_is_vendored_from_the_frozen_commit():
    requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8").splitlines()
    provenance = (ROOT / "bluemagpie" / "UPSTREAM_RUNTIME.md").read_text(
        encoding="utf-8"
    )

    assert not any("BlueMagpie-TTS.git" in line for line in requirements)
    assert "ce384c8cc54efea1aaba7b9f1d7ded6c1c99aa9a" in provenance
    assert (ROOT / "bluemagpie" / "LICENSE.upstream").is_file()
    assert (ROOT / "bluemagpie" / "_vendor" / "voxcpm" / "LICENSE").is_file()
    pinned_hashes = {
        "model.py": "91810524212b34f727880154d90653fab4ae1b75eb3471b86cafd92c75514fef",
        "loading.py": "e3407544e9bc888018fe5771edc01d954469e2af548b566873e0ef7f2afe6dca",
        "_vendor/voxcpm/model/utils.py": (
            "cea16e1ab57f15129a7f5dec13c428bd14a771221abcf17dd0a90b3d65e763a2"
        ),
    }
    for relative_path, expected_hash in pinned_hashes.items():
        payload = (ROOT / "bluemagpie" / relative_path).read_bytes()
        assert hashlib.sha256(payload).hexdigest() == expected_hash


def test_barbet_runtime_dependency_is_commit_pinned():
    requirements = (ROOT / "requirements.txt").read_text(encoding="utf-8").splitlines()
    barbet_lines = [line for line in requirements if "OpenFormosa/Barbet.git" in line]

    assert barbet_lines == [
        "git+https://github.com/OpenFormosa/Barbet.git@"
        "6fcd7ce4aa37f2250a3242995bef0fbc3b026ba8"
    ]


def test_quality_asr_is_revision_pinned():
    constants = _string_constants(ROOT / "quality_runtime.py")
    app_source = (ROOT / "app.py").read_text(encoding="utf-8")

    assert constants["WHISPER_MODEL_ID"] == "openai/whisper-large-v3-turbo"
    assert constants["WHISPER_REVISION"] == "41f01f3fe87f28c78e2fbf8b568835947dd65ed9"
    assert "snapshot_download(WHISPER_MODEL_ID, revision=WHISPER_REVISION)" in app_source


def test_quality_runtime_dependencies_are_version_pinned():
    requirements = set(
        (ROOT / "requirements.txt").read_text(encoding="utf-8").splitlines()
    )

    assert "huggingface_hub==0.36.0" in requirements
    assert "opencc-python-reimplemented==0.1.7" in requirements
    assert "transformers==4.57.6" in requirements
    assert "accelerate==1.12.0" in requirements
    assert "einops==0.8.2" in requirements
    assert "pydantic==2.11.10" in requirements
    assert "numpy==2.3.5" in requirements
    assert "scipy==1.17.1" in requirements
    assert "numexpr==2.14.1" in requirements
    assert "bottleneck==1.6.0" in requirements
    assert "tqdm==4.68.2" in requirements
    assert "safetensors==0.8.0" in requirements
    assert "librosa==0.11.0" in requirements
    assert "soundfile==0.14.0" in requirements
    assert "speechbrain==1.0.3" in requirements


def test_readme_describes_conditional_cascade_and_unwired_experimental_scores():
    readme = (ROOT / "README.md").read_text(encoding="utf-8")

    assert "same-seed whole trajectory first" in readme
    assert "1→5→10 sequence fallback" in readme
    assert "floor(20 / chunk count)" in readme
    assert "at most 20 generated TTS chunks" in readme
    assert "不使用 reference-style distribution score" in readme
    assert "尚未啟用 normalized median-F0 transition" in readme
    assert "candidate 0" in readme
    assert "5.2 CJK / 4.6 ASCII" in readme
    assert "candidate 1–9" in readme
    assert "4.6 CJK / 4.0 ASCII" in readme


def test_app_wires_candidate_offset_to_explicit_generation_policy_and_logs_it():
    source = (ROOT / "app.py").read_text(encoding="utf-8")

    assert source.count("policy: GenerationPolicy") == 2
    assert "policy=generation_policy_for_candidate_offset(seed - request_seed)" in source
    assert 'f"name={policy.name}' in source
    assert "chunk_policies={selected_policies}" in source
    assert '"min_len": min_len' in source


def test_internal_synthesize_accepts_only_a_keyword_seed_while_ui_stays_unchanged():
    source = (ROOT / "app.py").read_text(encoding="utf-8")
    tree = ast.parse(source)
    functions = {
        node.name: node
        for node in tree.body
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
    }

    synthesize = functions["_synthesize"]
    assert "request_seed" not in [argument.arg for argument in synthesize.args.args]
    assert [argument.arg for argument in synthesize.args.kwonlyargs][-1] == "request_seed"
    assert isinstance(synthesize.args.kw_defaults[-1], ast.Constant)
    assert synthesize.args.kw_defaults[-1].value is None
    assert "request_seed = resolve_request_seed(request_seed, secrets.randbelow)" in source
    assert "run_adaptive_cascade(\n                chunks,\n                request_seed," in source
    for wrapper_name in ("tts_speaker", "tts_reference", "tts_longform"):
        wrapper = functions[wrapper_name]
        arguments = wrapper.args.args + wrapper.args.kwonlyargs
        assert "request_seed" not in [argument.arg for argument in arguments]


def test_app_reverifies_the_post_join_speed_adjusted_whole_waveform():
    source = (ROOT / "app.py").read_text(encoding="utf-8")
    tree = ast.parse(source)
    functions = {
        node.name: node
        for node in tree.body
        if isinstance(node, ast.FunctionDef)
    }
    assemble_source = ast.get_source_segment(
        source,
        functions["_assemble_trajectory_audio"],
    )
    synthesize_source = ast.get_source_segment(source, functions["_synthesize"])

    assert assemble_source is not None
    assert synthesize_source is not None
    speed_index = assemble_source.index(
        "waveform = _apply_speed(waveform, playback_speed)"
    )
    finish_index = assemble_source.index("return finish_audio(waveform, SR)")
    assert speed_index < finish_index
    assemble_index = synthesize_source.index(
        "waveform = _assemble_trajectory_audio(cascade.trajectory, chunks, speed)"
    )
    verify_index = synthesize_source.index("final_verification = _verify_trajectory_audio(")
    require_index = synthesize_source.index(
        "require_verified_final_output(final_verification)"
    )
    return_index = synthesize_source.index("return SR, waveform")
    assert assemble_index < verify_index < require_index < return_index
    assert "(text,)" in synthesize_source[verify_index:require_index]
    assert "        1.0," in synthesize_source[verify_index:require_index]
    assert "QUALITY_FINAL_ASR_MAX_NEW_TOKENS" in synthesize_source[
        verify_index:require_index
    ]


def test_whole_candidate_qualification_uses_the_exact_return_assembler_after_local_pass():
    source = (ROOT / "app.py").read_text(encoding="utf-8")
    tree = ast.parse(source)
    functions = {
        node.name: node
        for node in tree.body
        if isinstance(node, ast.FunctionDef)
    }
    qualify_source = ast.get_source_segment(
        source,
        functions["_qualify_candidate_trajectory_audio"],
    )
    synthesize_source = ast.get_source_segment(source, functions["_synthesize"])

    assert qualify_source is not None
    assert synthesize_source is not None
    local_index = qualify_source.index("local_verification = _verify_trajectory_audio(")
    local_fail_index = qualify_source.index("if not local_verification.passed:")
    assemble_index = qualify_source.index("waveform = _assemble_trajectory_audio(")
    joined_index = qualify_source.index("joined_verification = _verify_trajectory_audio(")
    assert local_index < local_fail_index < assemble_index < joined_index
    assert "QUALITY_FINAL_ASR_MAX_NEW_TOKENS" in qualify_source[joined_index:]
    assert "qualify_trajectory_with_joined_output(" in qualify_source[joined_index:]
    assert (
        "waveform = _assemble_trajectory_audio(cascade.trajectory, chunks, speed)"
        in synthesize_source
    )
    assert "require_verified_final_output(final_verification)" in synthesize_source


def test_space_wires_bounded_k_best_paths_to_exact_assembled_whole_gate():
    source = (ROOT / "app.py").read_text(encoding="utf-8")
    readme = (ROOT / "README.md").read_text(encoding="utf-8")

    assert "sequence_final_verifier=lambda sequence_result, candidate_chunks:" in source
    assert "_verify_sequence_trajectory_audio(" in source
    assert "max_sequence_paths=3" in source
    assert "waveform = _assemble_trajectory_audio(" in source
    assert "QUALITY_FINAL_ASR_MAX_NEW_TOKENS" in source
    assert "sequence_rank={cascade.sequence_path_rank}" in source
    assert "sequence_paths_checked={cascade.sequence_paths_checked}" in source
    assert "cer={comparison.cer:.6f}" in source
    assert "prefix_cer={comparison.prefix_cer:.6f}" in source
    assert "suffix_cer={comparison.suffix_cer:.6f}" in source
    assert "tail_units={comparison.extra_tail_units}" in source
    assert "最多 3 條不同完整路徑" in readme
    assert "不增加 TTS generation 數量" in readme
    assert "不套用於單 chunk request" in readme