"""Regression checks for the review and the 8788 black-box report."""
from __future__ import annotations

import json
import os
import socket
import subprocess
import sys
import tempfile
import threading
import unittest
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from unittest import mock

if "asrservice.config" not in sys.modules:
    _boot = tempfile.mkdtemp(prefix="asr-harden-")
    os.environ["ASR_DATA_DIR"] = str(Path(_boot) / "data")
    os.environ["ASR_STATE_DIR"] = str(Path(_boot) / "state")
    os.environ["ASR_VAULT"] = str(Path(_boot) / "vault")
    os.environ["ASR_NOTIFY"] = "0"
    os.environ["ASR_GIT_SYNC"] = "0"
    os.environ["ASR_AGENT_TOKEN"] = "test-agent-token"

from asrpipe import locking, refine, review, webauth  # noqa: E402
from asrpipe.config import Config  # noqa: E402
from asrpipe import providers  # noqa: E402
from asrservice import config as svc_cfg  # noqa: E402
from asrservice import db as svc_db  # noqa: E402
from asrservice import web as svc_web  # noqa: E402
from asrservice import worker  # noqa: E402


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise urllib.error.HTTPError(req.full_url, code, msg, headers, fp)


def _opener(follow=True):
    if follow:
        return urllib.request.build_opener(urllib.request.HTTPCookieProcessor())
    return urllib.request.build_opener(_NoRedirect)


def _start_server():
    server = ThreadingHTTPServer(("127.0.0.1", 0), svc_web.Handler)
    thread = threading.Thread(target=server.serve_forever, daemon=True)
    thread.start()
    return server, server.server_address[1]


def _raw(port: int, payload: bytes) -> bytes:
    sock = socket.create_connection(("127.0.0.1", port), timeout=5)
    try:
        sock.sendall(payload)
        sock.shutdown(socket.SHUT_WR)
        chunks = []
        while True:
            part = sock.recv(65536)
            if not part:
                break
            chunks.append(part)
        return b"".join(chunks)
    finally:
        sock.close()


class HardeningTests(unittest.TestCase):
    def setUp(self):
        os.environ.pop("ASR_WEB_ALLOW_ANON", None)
        os.environ["ASR_WEB_PASSWORD"] = "test-web-password"
        os.environ["ASR_NOTIFY"] = "0"
        os.environ["ASR_GIT_SYNC"] = "0"
        svc_cfg.ensure_dirs()
        svc_db.connect().close()
        self._hindsight = Config.use_hindsight.__get__(svc_web.CFG, type(svc_web.CFG))
        svc_web.CFG.use_hindsight = False

    def tearDown(self):
        svc_web.CFG.use_hindsight = self._hindsight

    def test_closed_without_password_and_csrf_and_bad_length(self):
        os.environ["ASR_WEB_PASSWORD"] = ""
        server, port = _start_server()
        try:
            opener = _opener(False)
            with self.assertRaises(urllib.error.HTTPError) as caught:
                opener.open(f"http://127.0.0.1:{port}/review/", timeout=5)
            self.assertEqual(caught.exception.code, 303)
            self.assertIn("/login", caught.exception.headers.get("Location", ""))
            page = urllib.request.urlopen(f"http://127.0.0.1:{port}/login", timeout=5).read().decode()
            self.assertIn("ASR_WEB_PASSWORD", page)
            raw = _raw(port, b"POST /login HTTP/1.0\r\nContent-Length: abc\r\n\r\n")
            self.assertIn(b" 400 ", raw)
            self.assertNotIn(b"Empty", raw)
        finally:
            server.shutdown()
            server.server_close()
            os.environ["ASR_WEB_PASSWORD"] = "test-web-password"

    def test_pages_hide_secrets_escape_titles_and_reject_bad_edits(self):
        svc_db.save_settings({
            "litellm_api_key": "sk-SECRET-abcd-wxyz",
            "diarize_url": "http://192.168.50.100:10099/asr",
        })
        jid = "abc123abc123abc123abc123"
        title = "X');window.__xss=1;//"
        svc_db.create_job(jid, "录音.wav", "sample.wav", title=title, size=100, source="web")
        audio = svc_cfg.PROCESSED / "sample.wav"
        audio.write_bytes(b"0123456789" * 10)
        draft = svc_cfg.DRAFTS / f"{jid}.md"
        draft.write_text("原文一行\n", encoding="utf-8")
        result = svc_cfg.RESULTS / "2026-纪要.md"
        result.write_text("# 纪要\n", encoding="utf-8")
        svc_db.update_job(jid, "complete", "完成", draft=str(draft), result=str(result))
        q1 = review.enqueue("rec-a", "第一题？", {"kind": "draft_question", "candidates": ["甲"]})
        q2 = review.enqueue("rec-b", "第二题？", {"kind": "draft_question", "candidates": ["乙"]})
        server, port = _start_server()
        try:
            opener = _opener(True)
            opener.open(urllib.request.Request(
                f"http://127.0.0.1:{port}/login",
                data=urllib.parse.urlencode({"password": "test-web-password"}).encode(),
                method="POST",
            ), timeout=5)
            home = opener.open(f"http://127.0.0.1:{port}/", timeout=5).read().decode()
            self.assertNotIn('onclick="delOne(', home)
            self.assertIn("data-del", home)
            settings = opener.open(f"http://127.0.0.1:{port}/settings", timeout=5).read().decode()
            self.assertNotIn("192.168.50.100", settings)
            self.assertNotIn("sk-SECRET", settings)
            self.assertIn("已配置", settings)
            review_html = opener.open(f"http://127.0.0.1:{port}/review/", timeout=5).read().decode()
            forms = review_html.split('<form class="q-item"')
            self.assertGreaterEqual(len(forms), 3)
            for form in forms[1:]:
                head = form.split("</form>", 1)[0]
                self.assertLess(head.find('name="single_custom"'), head.find("custom_"))
            missing = _raw(
                port,
                (
                    "POST /jobs/%s/meta HTTP/1.0\r\nHost: 127.0.0.1\r\n"
                    "Content-Type: application/x-www-form-urlencoded\r\n"
                    "Cookie: asr_session=%s\r\nContent-Length: 10\r\n\r\ntitle=nope"
                    % (jid, webauth.session_token())
                ).encode(),
            )
            self.assertIn(b" 403 ", missing)
            edit = urllib.parse.urlencode({
                "csrf": webauth.csrf_token(),
                "target": "../../x",
                "content": "x",
            }).encode()
            req = urllib.request.Request(
                f"http://127.0.0.1:{port}/jobs/{jid}/edit", data=edit, method="POST",
            )
            body = opener.open(req, timeout=5).read().decode()
            self.assertIn("未保存", body)
            self.assertEqual(draft.read_text(encoding="utf-8"), "原文一行\n")
            empty = urllib.parse.urlencode({
                "csrf": webauth.csrf_token(), "target": "draft", "content": "  ",
            }).encode()
            body = opener.open(urllib.request.Request(
                f"http://127.0.0.1:{port}/jobs/{jid}/edit", data=empty, method="POST",
            ), timeout=5).read().decode()
            self.assertIn("内容为空", body)
            header = (
                f"POST /jobs/{jid}/edit HTTP/1.0\r\nCookie: asr_session={webauth.session_token()}\r\n"
                "Content-Type: application/x-www-form-urlencoded\r\nContent-Length: 3000000\r\n\r\n"
            ).encode()
            huge = _raw(port, header)
            self.assertIn(b" 413 ", huge)
            art = urllib.request.Request(f"http://127.0.0.1:{port}/jobs/{jid}/artifact/result")
            with opener.open(art, timeout=5) as resp:
                disp = resp.headers.get("Content-Disposition", "")
                text = resp.read().decode()
            disp.encode("latin-1")
            self.assertIn("filename*=UTF-8''", disp)
            self.assertIn("filename=\"download.md\"", disp)
            self.assertIn("纪要", text)
            ranged = urllib.request.Request(
                f"http://127.0.0.1:{port}/jobs/{jid}/audio",
                headers={"Range": "bytes=80-10"},
            )
            try:
                opener.open(ranged, timeout=5)
                self.fail("reverse range should be rejected")
            except urllib.error.HTTPError as exc:
                self.assertEqual(exc.code, 416)
                self.assertIn("bytes */100", exc.headers.get("Content-Range", ""))
            suffix = urllib.request.Request(
                f"http://127.0.0.1:{port}/jobs/{jid}/audio",
                headers={"Range": "bytes=-10"},
            )
            with opener.open(suffix, timeout=5) as resp:
                self.assertEqual(resp.status, 206)
                self.assertEqual(resp.headers.get("Content-Range"), "bytes 90-99/100")
                self.assertEqual(len(resp.read()), 10)
            try:
                opener.open(f"http://127.0.0.1:{port}/no-such-page", timeout=5)
                self.fail("missing page should be 404")
            except urllib.error.HTTPError as exc:
                self.assertEqual(exc.code, 404)
                self.assertIn("页面不存在", exc.read().decode())
        finally:
            server.shutdown()
            server.server_close()
            review.delete_for_job(transcript_path="rec-a")
            review.delete_for_job(transcript_path="rec-b")
            self.assertIsNotNone(q1)
            self.assertIsNotNone(q2)

    def test_busy_retry_does_not_move_processing_audio(self):
        jid = "fedcba9876543210fedcba98"
        name = "busy.wav"
        svc_db.create_job(jid, name, name, title="进行中", size=4, source="web")
        audio = svc_cfg.PROCESSING / name
        audio.write_bytes(b"wave")
        svc_db.update_job(jid, "transcribing", "正在转写")
        server, port = _start_server()
        try:
            opener = _opener(True)
            opener.open(urllib.request.Request(
                f"http://127.0.0.1:{port}/login",
                data=urllib.parse.urlencode({"password": "test-web-password"}).encode(),
                method="POST",
            ), timeout=5)
            body = urllib.parse.urlencode({
                "csrf": webauth.csrf_token(),
                "title": "进行中",
                "provider": "not-a-real-engine",
                "next_action": "retry",
            }).encode()
            page = opener.open(urllib.request.Request(
                f"http://127.0.0.1:{port}/jobs/{jid}/meta", data=body, method="POST",
            ), timeout=5).read().decode()
            self.assertIn("未重新排队", page)
            self.assertIn("未知引擎", page)
            self.assertTrue(audio.is_file())
            self.assertFalse((svc_cfg.INBOX / name).exists())
            self.assertEqual(svc_db.get_job(jid)["status"], "transcribing")
        finally:
            server.shutdown()
            server.server_close()

    def test_revision_failure_keeps_marker_and_answers(self):
        jid = "a1b2c3d4e5f60718293a4b5c"
        transcript = svc_cfg.TRANSCRIPTS / f"{jid}.txt"
        transcript.write_text("【speaker_0】原文\n", encoding="utf-8")
        draft = svc_cfg.DRAFTS / f"{jid}.md"
        draft.write_text("---\ntype: minutes\n---\n# 稿\n", encoding="utf-8")
        svc_db.create_job(jid, "a.wav", "a.wav", title="重精修", size=1, source="web")
        svc_db.update_job(jid, "review", "待确认", transcript=str(transcript), draft=str(draft))
        qid = review.enqueue(str(transcript), "已确认的问题？", {
            "kind": "draft_question", "draft_path": str(draft), "candidates": ["保留"],
        })
        review.answer(qid, "保留")
        pending_id = review.enqueue(str(transcript), "仍待确认？", {
            "kind": "draft_question", "draft_path": str(draft), "candidates": ["甲"],
        })
        with mock.patch("asrservice.worker.refine.refine", side_effect=RuntimeError("llm down")):
            with self.assertRaises(RuntimeError):
                worker.rerefine_job(svc_db.get_job(jid))
        answered = review.recent_answered(20)
        self.assertIn(qid, [row["id"] for row in answered])
        self.assertIn(pending_id, [row["id"] for row in review.pending(50)])
        marker = svc_cfg.REVISION_QUEUE / f"revise-{jid}.json"
        marker.write_text(json.dumps({
            "action": "revise", "job_id": jid, "draft": "/missing/draft.md",
        }), encoding="utf-8")
        worker.process_revisions()
        self.assertTrue(marker.is_file())
        payload = json.loads(marker.read_text(encoding="utf-8"))
        self.assertEqual(payload["attempts"], 1)
        self.assertEqual(svc_db.get_job(jid)["status"], "review")

    def test_recover_processing_and_shared_lock_and_stream(self):
        jid = "00112233445566778899aabb"
        name = f"stuck-{jid[:8]}.wav"
        audio = svc_cfg.PROCESSING / name
        audio.write_bytes(b"stuck")
        svc_db.create_job(jid, name, name, title="卡住", size=5, source="web")
        svc_db.update_job(jid, "transcribing", "中断")
        worker.recover()
        self.assertFalse(audio.exists())
        parked = svc_cfg.INBOX / name
        self.assertTrue(parked.is_file())
        self.assertEqual(svc_db.get_job(jid)["status"], "queued")
        worker.beat()
        self.assertTrue((svc_cfg.STATE / "worker.heartbeat").is_file())

        lock_path = Path(tempfile.mkdtemp()) / "consumer.lock"
        handle = locking.acquire_recording_lock(lock_path)
        try:
            with self.assertRaises(SystemExit):
                locking.acquire_recording_lock(lock_path)
        finally:
            handle.close()

        blob = b"STREAMTEST" + b"q" * (256 * 1024)
        audio_path = Path(tempfile.mkdtemp()) / "stream.wav"
        audio_path.write_bytes(blob)
        seen = {}

        class Capture(BaseHTTPRequestHandler):
            def do_POST(self):
                n = int(self.headers["Content-Length"])
                body = self.rfile.read(n)
                seen["n"] = len(body)
                seen["head"] = b"STREAMTEST" in body
                seen["tail"] = body.endswith(b"q" + b"\r\n--" + body.split(b"----ASRBoundary", 1)[0][:0] or b"q")
                self.send_response(200)
                self.send_header("Content-Type", "application/json")
                self.end_headers()
                self.wfile.write(b'{"ok": true}')

            def log_message(self, *_):
                pass

        cap = ThreadingHTTPServer(("127.0.0.1", 0), Capture)
        threading.Thread(target=cap.serve_forever, daemon=True).start()
        try:
            out = providers._http_multipart(
                f"http://127.0.0.1:{cap.server_address[1]}/up",
                fields={"model": "unit"},
                file_field="file",
                file_path=audio_path,
                timeout=10,
                retries=1,
            )
        finally:
            cap.shutdown()
            cap.server_close()
        self.assertTrue(out["ok"])
        self.assertGreater(seen["n"], len(blob))
        self.assertTrue(seen["head"])

    def test_wording_timezone_git_default_and_lexicon_page(self):
        self.assertEqual(
            refine.scrub_internal_wording("卡片未提供职务，卡片提及张三"),
            "记录未写明职务，录音中提到张三",
        )
        stamp = datetime(2026, 11, 2, 13, 8, 26, tzinfo=timezone.utc).timestamp()
        self.assertEqual(svc_web.fmt_ts(stamp), "2026-11-02 21:08:26")
        header = worker.intake_header({"context": "a-->b } c"})
        self.assertEqual(header.count("-->"), 1)
        self.assertTrue(header.endswith("-->"))
        self.assertNotIn("a-->", header)
        self.assertTrue(header.startswith("<!-- intake:"))
        for i in range(3):
            from asrpipe import lexicon
            lexicon.learn_alias(f"别名{i}", f"规范{i}", origin="manual")
        from asrpipe import lexicon
        self.assertGreaterEqual(lexicon.count_aliases(q="别名"), 3)
        page = lexicon.list_aliases(q="别名", limit=2, offset=0)
        self.assertEqual(len(page), 2)
        env = os.environ.copy()
        env.pop("ASR_GIT_SYNC", None)
        env.pop("ASR_NOTIFY", None)
        env["ASR_STATE_DIR"] = tempfile.mkdtemp()
        env["PYTHONPATH"] = str(Path(__file__).resolve().parents[1])
        out = subprocess.check_output(
            [sys.executable, "-c",
             "from asrpipe.config import Config; c=Config(); print(int(c.git_sync), int(c.notify))"],
            env=env, text=True,
        ).strip()
        self.assertEqual(out, "0 0")


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