from __future__ import annotations

import json
import os
import tempfile
import threading
import unittest
import urllib.parse
import urllib.request
from http.server import ThreadingHTTPServer
from pathlib import Path
from unittest import mock

TMP = tempfile.TemporaryDirectory()
os.environ["ASR_DATA_DIR"] = str(Path(TMP.name) / "data")
os.environ["ASR_STATE_DIR"] = str(Path(TMP.name) / "state")
os.environ["ASR_VAULT"] = str(Path(TMP.name) / "vault")
os.environ["ASR_RECORDINGS_ROOT"] = str(Path(TMP.name) / "recordings")
os.environ["ASR_AGENT_TOKEN"] = "test-agent-token"
os.environ["ASR_NOTIFY"] = "0"
os.environ["ASR_GIT_SYNC"] = "0"

from asrpipe import lexicon, providers, review, segment  # noqa: E402
from asrpipe.config import CFG  # noqa: E402
from asrservice import config as svc_cfg, db as svc_db, web as svc_web  # noqa: E402


class ProviderAndServiceTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)
        svc_cfg.ensure_dirs()
        svc_db.connect().close()

    def tearDown(self):
        self.tmp.cleanup()

    def test_sqlite_settings_override_config_dynamically(self):
        svc_db.save_settings({
            "asr_provider": "soniox",
            "litellm_base_url": "https://api.opfo.xyz/v1",
            "litellm_api_key": "sk-sqlite-dynamic-key",
            "clean_model": "aistudio/gemini-flash-latest",
            "draft_model": "openai/gpt-sol-latest",
            "soniox_api_key": "soniox-db-secret",
            "use_hindsight": "0",
        })
        self.assertEqual(CFG.asr_provider, "soniox")
        self.assertEqual(CFG.litellm_base_url, "https://api.opfo.xyz/v1")
        self.assertEqual(CFG.endpoint, "https://api.opfo.xyz/v1/chat/completions")
        self.assertEqual(CFG.transcriptions_endpoint, "https://api.opfo.xyz/v1/audio/transcriptions")
        self.assertEqual(CFG.api_key(), "sk-sqlite-dynamic-key")
        self.assertEqual(CFG.clean_model, "aistudio/gemini-flash-latest")
        self.assertEqual(CFG.draft_model, "openai/gpt-sol-latest")
        self.assertEqual(CFG.soniox_api_key, "soniox-db-secret")
        self.assertFalse(CFG.use_hindsight)
        # Restore default for subsequent tests
        svc_db.save_settings({"asr_provider": "qwen3cam", "use_hindsight": "1"})

    def test_soniox_token_diarization_and_segment_parse(self):
        payload = {
            "tokens": [
                {"text": "大家", "speaker": "1"},
                {"text": "好，", "speaker": "1"},
                {"text": "今天", "speaker": "1"},
                {"text": "讨论", "speaker": "1"},
                {"text": "预算。", "speaker": "1"},
                {"text": "收到，", "speaker": "2"},
                {"text": "我来", "speaker": "2"},
                {"text": "汇报。", "speaker": "2"},
            ]
        }
        text = providers.transcript_from_soniox_response(payload, model="stt-async-v5")
        with_intake = '<!-- intake: {"job_id": "abc", "asr_provider": "soniox", "asr_model": "stt-async-v5"} -->\n' + text
        parsed_turns, meta = segment.parse(with_intake)
        self.assertEqual(meta["asr_provider"], "soniox")
        self.assertEqual(meta["asr_model"], "stt-async-v5")
        self.assertEqual(meta["backend"], "soniox")
        self.assertEqual(len(parsed_turns), 2)
        self.assertEqual(parsed_turns[0].speaker, "speaker_0")
        self.assertEqual(parsed_turns[0].text, "大家好，今天讨论预算。")
        self.assertEqual(parsed_turns[1].speaker, "speaker_1")
        self.assertEqual(parsed_turns[1].text, "收到，我来汇报。")

        with self.assertRaises(providers.EmptyAudioError):
            providers.transcript_from_soniox_response({"tokens": [], "text": ""}, model="stt-async-v5")

    def test_soniox_context_includes_lexicon_and_intake(self):
        lexicon.learn_alias("lite网关", "LiteLLM", note="test")
        ctx = providers.build_soniox_context(
            title="网关评审",
            user_context="讨论 LiteLLM 与 Soniox 集成",
        )
        self.assertIn("terms", ctx)
        self.assertIn("LiteLLM", ctx["terms"])
        self.assertIn("讨论 LiteLLM", ctx.get("text", ""))

    def test_openai_and_grok_diarized_transcriptions(self):
        audio = self.root / "sample.mp3"
        audio.write_bytes(b"ID3-fake-audio")

        def fake_openai_multipart(url, *, fields, file_field, file_path, headers=None, timeout=900):
            return {
                "segments": [
                    {"speaker": "A", "text": "第一句发言。"},
                    {"speaker": "A", "text": "继续补充。"},
                    {"speaker": "B", "text": "第二位发言。"},
                ]
            }

        with mock.patch("asrpipe.providers._http_multipart", side_effect=fake_openai_multipart), \
             mock.patch("asrpipe.providers.probe_duration_seconds", return_value=30), \
             mock.patch.object(providers.CFG, "api_key", return_value="sk-test"):
            res = providers.transcribe_openai(
                audio,
                title="OpenAI 测试",
                model="openai/gpt-4o-transcribe-diarize",
            )
        turns, meta = segment.parse(res.transcript_text)
        self.assertEqual(meta["asr_provider"], "openai")
        self.assertEqual(len(turns), 2)
        self.assertIn("第一句发言。继续补充。", turns[0].text)
        self.assertEqual(turns[1].speaker, "speaker_1")

        def fake_grok_multipart(url, *, fields, file_field, file_path, headers=None, timeout=900):
            return {
                "utterances": [
                    {"speaker": 0, "text": "Grok 说话人零。"},
                    {"speaker": 1, "text": "Grok 说话人一。"},
                ]
            }

        with mock.patch("asrpipe.providers._http_multipart", side_effect=fake_grok_multipart), \
             mock.patch("asrpipe.providers.probe_duration_seconds", return_value=30), \
             mock.patch.object(providers.CFG, "grok_api_key", "xai-test"):
            gres = providers.transcribe_grok(
                audio,
                title="Grok 测试",
            )
        gturns, gmeta = segment.parse(gres.transcript_text)
        self.assertEqual(gmeta["asr_provider"], "grok")
        self.assertEqual(len(gturns), 2)
        self.assertEqual(gturns[0].speaker, "speaker_0")
        self.assertEqual(gturns[1].speaker, "speaker_1")

    def test_online_speaker_confirmation_updates_draft_without_voiceprint_server(self):
        drafts_dir = review.CFG.drafts
        drafts_dir.mkdir(parents=True, exist_ok=True)
        draft = drafts_dir / "online-draft.md"
        draft.write_text(
            "---\n"
            "title: 线上模型会议\n"
            "status: draft\n"
            "asr_provider: soniox\n"
            "voiceprint_session: ''\n"
            "voiceprint_speakers: {}\n"
            "---\n\n"
            "【speaker_1】发言内容\n",
            encoding="utf-8",
        )
        with mock.patch.object(review.CFG, "voiceprint_enroll_url", "http://127.0.0.1:9/unreachable"):
            enrolled = review._confirm_voiceprint(str(draft), "speaker_1", "张三")
        self.assertFalse(enrolled)
        updated = draft.read_text(encoding="utf-8")
        self.assertIn("speaker_1: 张三", updated)

    def test_lexicon_crud_operations(self):
        # Entity CRUD
        lexicon.upsert_entity("测试生物", "公司", company="测试集团", role="子公司", related="李四")
        ents = lexicon.list_entities(q="测试生物")
        self.assertEqual(len(ents), 1)
        self.assertEqual(ents[0]["company"], "测试集团")

        lexicon.update_entity("测试生物", "测试生物科技", "公司", company="新集团", role="总部", related="李四,王五")
        ents2 = lexicon.list_entities(q="测试生物科技")
        self.assertEqual(len(ents2), 1)
        self.assertEqual(ents2[0]["company"], "新集团")
        self.assertTrue(lexicon.delete_entity("测试生物科技"))
        self.assertEqual(len(lexicon.list_entities(q="测试生物科技")), 0)

        # Alias & Correction CRUD
        lexicon.learn_alias("老测试", "测试人", kind="人名", origin="manual", note="初始")
        lexicon.update_alias("老测试", "测试人", "小测试", "测试人", kind="人名", confirmed=1, note="已改")
        aliases = lexicon.list_aliases(q="小测试")
        self.assertEqual(len(aliases), 1)
        self.assertEqual(aliases[0]["note"], "已改")
        self.assertTrue(lexicon.delete_alias("小测试", "测试人"))

        lexicon.learn_correction("误识词", "正词", note="c1")
        lexicon.update_correction("误识词", "正词", "新误识词", "正词", confirmed=1, note="c2")
        cors = lexicon.list_corrections(q="新误识词")
        self.assertEqual(len(cors), 1)
        self.assertEqual(cors[0]["note"], "c2")
        self.assertTrue(lexicon.delete_correction("新误识词", "正词"))

        # Relation CRUD
        lexicon.upsert_relation("张三", "任职于", "洛启生物", confidence=0.8, source="manual")
        lexicon.update_relation("张三", "任职于", "洛启生物", "manual", "张三", "实控人", "洛启生物", confidence=0.95, active=1)
        rels = lexicon.list_relations(q="张三")
        self.assertTrue(any(r["predicate"] == "实控人" for r in rels))
        self.assertTrue(lexicon.delete_relation("张三", "实控人", "洛启生物", "manual"))

    def test_db_web_endpoints_multi_review_and_job_delete(self):
        job1_id = "0123456789abcdef01234567"
        job2_id = "fedcba987654321001234567"
        t1 = svc_cfg.TRANSCRIPTS / f"{job1_id}.txt"
        t2 = svc_cfg.TRANSCRIPTS / f"{job2_id}.txt"
        t1.write_text("【speaker_0】录音一内容\n", encoding="utf-8")
        t2.write_text("【speaker_0】录音二内容\n", encoding="utf-8")

        svc_db.create_job(
            job_id=job1_id,
            original_name="meeting1.m4a",
            stored_name=f"20261102-{job1_id}.m4a",
            title="第一场尽调会",
            context="验证多录音确认",
            size=2048,
            asr_provider="soniox",
            asr_model="stt-async-v5",
        )
        svc_db.update_job(job1_id, "review", "待确认", transcript=str(t1))

        svc_db.create_job(
            job_id=job2_id,
            original_name="meeting2.m4a",
            stored_name=f"20261102-{job2_id}.m4a",
            title="第二场投决会",
            context="验证批量操作",
            size=4096,
            asr_provider="qwen3cam",
        )
        svc_db.update_job(job2_id, "review", "待确认", transcript=str(t2))

        q1 = review.enqueue(str(t1), "第一场问题A？", {"kind": "draft_question", "candidates": ["选项A1", "选项A2"]})
        q2 = review.enqueue(str(t1), "第一场问题B？", {"kind": "draft_question", "candidates": ["选项B1"]})
        q3 = review.enqueue(str(t2), "第二场问题C？", {"kind": "draft_question", "candidates": ["选项C1"]})

        os.environ["ASR_WEB_PASSWORD"] = "test-web-password"
        os.environ.pop("ASR_WEB_ALLOW_ANON", None)
        svc_web.AGENT_TOKEN = "test-agent-token"
        server = ThreadingHTTPServer(("127.0.0.1", 0), svc_web.Handler)
        port = server.server_address[1]
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        try:
            import http.cookiejar
            from asrpipe import webauth
            jar = http.cookiejar.CookieJar()
            opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
            login_body = urllib.parse.urlencode({"password": "test-web-password"}).encode()
            opener.open(urllib.request.Request(
                f"http://127.0.0.1:{port}/login", data=login_body, method="POST",
            ), timeout=5)

            def fetch(path: str, headers: dict[str, str] | None = None) -> tuple[int, str]:
                req = urllib.request.Request(
                    f"http://127.0.0.1:{port}{path}",
                    headers=headers or {},
                )
                with opener.open(req, timeout=5) as resp:
                    return resp.status, resp.read().decode("utf-8")

            def post_form(path: str, data: dict[str, str]) -> tuple[int, str]:
                payload = dict(data)
                payload.setdefault("csrf", webauth.csrf_token())
                encoded = urllib.parse.urlencode(payload).encode("utf-8")
                req = urllib.request.Request(
                    f"http://127.0.0.1:{port}{path}",
                    data=encoded,
                    headers={"Content-Type": "application/x-www-form-urlencoded"},
                    method="POST",
                )
                with opener.open(req, timeout=5) as resp:
                    return resp.status, resp.read().decode("utf-8")

            status, body = fetch("/health")
            self.assertEqual(status, 200)
            self.assertEqual(json.loads(body)["ok"], True)

            status, body = fetch(
                "/api/v1/providers",
                headers={"Authorization": "Bearer test-agent-token"},
            )
            self.assertEqual(status, 200)
            prov_ids = [p["id"] for p in json.loads(body)["providers"]]
            self.assertEqual(prov_ids, ["qwen3cam", "soniox", "gemini", "openai", "grok"])

            # Multi-recording review page shows both recordings grouped
            status, body = fetch("/review/")
            self.assertEqual(status, 200)
            self.assertIn("第一场尽调会", body)
            self.assertIn("第二场投决会", body)
            self.assertIn("第一场问题A？", body)
            self.assertIn("第二场问题C？", body)

            # Batch submit both questions for recording 1
            status, body = post_form(f"/jobs/{job1_id}/review/", {
                "question_ids": f"{q1},{q2}",
                f"preset_{q1}": "选项A1",
                f"custom_{q2}": "自定义回答B",
                "batch_submit": "1",
            })
            self.assertEqual(status, 200)
            self.assertEqual(len(svc_web.pending(svc_db.get_job(job1_id))), 0)
            self.assertEqual(len(svc_web.pending(svc_db.get_job(job2_id))), 1)

            # Edit job metadata
            status, body = post_form(f"/jobs/{job1_id}/meta", {
                "title": "第一场尽调会（已改名）",
                "context": "更新后的背景",
                "provider": "openai",
                "model": "openai/gpt-4o-transcribe-diarize",
                "next_action": "save",
            })
            self.assertEqual(status, 200)
            self.assertIn("第一场尽调会（已改名）", body)

            # Save settings via Web UI POST
            status, body = post_form("/settings", {
                "litellm_base_url": "http://192.168.50.49:4000/v1",
                "clean_model": "aistudio/gemini-flash-lite-latest",
                "keyword_model": "aistudio/gemini-flash-lite-latest",
                "draft_model": "codex/gpt-sol-latest",
                "asr_provider": "soniox",
                "use_hindsight": "1",
            })
            self.assertEqual(status, 200)
            self.assertEqual(CFG.asr_provider, "soniox")

            # Delete job 2 and verify its pending review question is also cleaned up
            status, _ = post_form(f"/jobs/{job2_id}/delete", {"delete_files": "1"})
            self.assertEqual(status, 200)
            self.assertIsNone(svc_db.get_job(job2_id))
            self.assertFalse(t2.exists())
            self.assertNotIn(q3, [r["id"] for r in review.pending(200)])
        finally:
            server.shutdown()
            server.server_close()


if __name__ == "__main__":
    unittest.main()