import asyncio
import os
import sqlite3
import sys
import tempfile
import unittest
from contextlib import closing
from pathlib import Path
from types import ModuleType, SimpleNamespace


def install_import_stubs() -> None:
    class DummyRouter:
        def message(self, *args, **kwargs):
            def decorator(func):
                return func

            return decorator

    def identity_factory(*args, **kwargs):
        return object()

    modules = {
        "feedparser": ModuleType("feedparser"),
        "httpx": ModuleType("httpx"),
        "yaml": ModuleType("yaml"),
        "uvicorn": ModuleType("uvicorn"),
        "apscheduler": ModuleType("apscheduler"),
        "apscheduler.schedulers": ModuleType("apscheduler.schedulers"),
        "apscheduler.schedulers.asyncio": ModuleType("apscheduler.schedulers.asyncio"),
        "bs4": ModuleType("bs4"),
        "dotenv": ModuleType("dotenv"),
        "aiogram": ModuleType("aiogram"),
        "aiogram.enums": ModuleType("aiogram.enums"),
        "aiogram.exceptions": ModuleType("aiogram.exceptions"),
        "aiogram.filters": ModuleType("aiogram.filters"),
        "aiogram.types": ModuleType("aiogram.types"),
        "aiogram.client": ModuleType("aiogram.client"),
        "aiogram.client.default": ModuleType("aiogram.client.default"),
        "fastapi": ModuleType("fastapi"),
        "fastapi.responses": ModuleType("fastapi.responses"),
    }
    modules["apscheduler.schedulers.asyncio"].AsyncIOScheduler = object
    modules["bs4"].BeautifulSoup = object
    modules["dotenv"].load_dotenv = lambda *args, **kwargs: None
    modules["yaml"].safe_load = lambda stream: {"bot": {"spam_filter": {"enabled": True, "keywords": []}}}
    modules["yaml"].safe_dump = lambda data, **kwargs: str(data)
    modules["aiogram"].Bot = object
    modules["aiogram"].Dispatcher = object
    modules["aiogram"].F = object()
    modules["aiogram"].Router = DummyRouter
    modules["aiogram.enums"].ParseMode = SimpleNamespace(HTML="HTML")
    modules["aiogram.exceptions"].TelegramAPIError = Exception
    modules["aiogram.filters"].Command = identity_factory
    modules["aiogram.filters"].CommandObject = object
    modules["aiogram.types"].Message = object
    modules["aiogram.client.default"].DefaultBotProperties = identity_factory
    modules["fastapi"].Depends = identity_factory
    modules["fastapi"].FastAPI = object
    modules["fastapi"].Form = identity_factory
    modules["fastapi"].HTTPException = Exception
    modules["fastapi"].Request = object
    modules["fastapi"].Response = object
    modules["fastapi"].status = object()
    modules["fastapi.responses"].HTMLResponse = object
    modules["fastapi.responses"].RedirectResponse = object
    modules["fastapi.responses"].PlainTextResponse = object
    modules["uvicorn"].Server = object
    modules["uvicorn"].Config = identity_factory
    sys.modules.update({name: sys.modules.get(name, module) for name, module in modules.items()})


install_import_stubs()
import app


class FakeBot:
    def __init__(self) -> None:
        self.deleted: list[tuple[int, int]] = []
        self.sent_texts: list[str] = []
        self.sent_chat_ids: list[int] = []
        self.fail_chat_ids: set[int] = set()

    async def delete_message(self, chat_id: int, message_id: int) -> None:
        self.deleted.append((chat_id, message_id))

    async def send_message(self, chat_id: int, text: str, disable_web_page_preview: bool = False):
        if chat_id in self.fail_chat_ids:
            raise RuntimeError("send failed")
        self.sent_chat_ids.append(chat_id)
        self.sent_texts.append(text)
        return SimpleNamespace(message_id=3003)


class MonitorMessageCleanupTest(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.old_db_path = app.DB_PATH
        app.DB_PATH = Path(self.temp_dir.name) / "test.sqlite3"
        app.init_db()

    def tearDown(self) -> None:
        app.DB_PATH = self.old_db_path
        self.temp_dir.cleanup()

    def test_monitor_notification_send_is_recorded_for_later_deletion(self) -> None:
        old_bot = app.bot
        old_admin_chat_id = app.admin_chat_id
        old_admin_chat_ids = app.admin_chat_ids
        old_config = app.config
        fake_bot = FakeBot()
        app.bot = fake_bot
        app.admin_chat_id = 1001
        app.admin_chat_ids = []
        app.config = {"cleanup": {"monitor_message_delete_after_minutes": 1}}
        try:
            sent = asyncio.run(app.admin_send_monitor("monitor hit", "NodeSeek 新帖"))
            self.assertTrue(sent)
            self.assertEqual(["monitor hit"], fake_bot.sent_texts)
            with closing(sqlite3.connect(app.DB_PATH)) as conn:
                row = conn.execute(
                    "SELECT chat_id, message_id, monitor_name, delete_after_seconds FROM monitor_messages"
                ).fetchone()
            self.assertEqual((1001, 3003, "NodeSeek 新帖", 60), row)
        finally:
            app.bot = old_bot
            app.admin_chat_id = old_admin_chat_id
            app.admin_chat_ids = old_admin_chat_ids
            app.config = old_config

    def test_monitor_event_history_is_recorded(self) -> None:
        app.record_monitor_event("NodeSeek 新帖", "title", "https://example.com", ["关键词"], False)
        with closing(sqlite3.connect(app.DB_PATH)) as conn:
            row = conn.execute("SELECT monitor_name, title, pushed FROM monitor_events").fetchone()
        self.assertEqual(("NodeSeek 新帖", "title", 0), row)

    def test_monitor_notification_is_sent_to_all_admins(self) -> None:
        old_bot = app.bot
        old_admin_chat_ids = app.admin_chat_ids
        old_config = app.config
        fake_bot = FakeBot()
        app.bot = fake_bot
        app.admin_chat_ids = [1001, 1002, 1003]
        app.config = {"cleanup": {"monitor_message_delete_after_minutes": 1}}
        try:
            self.assertTrue(asyncio.run(app.admin_send_monitor("monitor hit", "NodeSeek 新帖")))
            self.assertEqual([1001, 1002, 1003], fake_bot.sent_chat_ids)
        finally:
            app.bot = old_bot
            app.admin_chat_ids = old_admin_chat_ids
            app.config = old_config

    def test_monitor_notification_continues_when_one_admin_fails(self) -> None:
        old_bot = app.bot
        old_admin_chat_ids = app.admin_chat_ids
        old_config = app.config
        fake_bot = FakeBot()
        fake_bot.fail_chat_ids.add(1002)
        app.bot = fake_bot
        app.admin_chat_ids = [1001, 1002, 1003]
        app.config = {"cleanup": {"monitor_message_delete_after_minutes": 1}}
        try:
            with self.assertLogs("tg-watchbot", level="ERROR"):
                self.assertTrue(asyncio.run(app.admin_send_monitor("monitor hit", "NodeSeek 新帖")))
            self.assertEqual([1001, 1003], fake_bot.sent_chat_ids)
        finally:
            app.bot = old_bot
            app.admin_chat_ids = old_admin_chat_ids
            app.config = old_config

    def test_outbound_message_is_recorded_in_conversation_log(self) -> None:
        app.upsert_user(2001, "User", "user")
        outbox_id = app.create_outbox_message(2001, "reply text", "web:inbox", 4004)
        with closing(sqlite3.connect(app.DB_PATH)) as conn:
            conn.row_factory = sqlite3.Row
            row = conn.execute("SELECT direction, source, text, forwarded FROM inbox_messages WHERE id=?", (outbox_id,)).fetchone()
        self.assertEqual(("out", "web:inbox", "reply text", 1), (row["direction"], row["source"], row["text"], row["forwarded"]))

    def test_save_message_map_supports_message_id_only_payload(self) -> None:
        app.save_message_map(1001, 3003, 2001, 4004)
        with closing(sqlite3.connect(app.DB_PATH)) as conn:
            row = conn.execute(
                "SELECT admin_chat_id, admin_message_id, user_id, user_message_id FROM message_map"
            ).fetchone()
        self.assertEqual((1001, 3003, 2001, 4004), row)

    def test_expired_monitor_message_is_deleted_and_removed_from_queue(self) -> None:
        app.record_monitor_message(1001, 2002, "NodeSeek 新帖", delete_after_seconds=60, sent_at_ts=1000)

        fake_bot = FakeBot()
        deleted_count = asyncio.run(app.delete_expired_monitor_messages(fake_bot, now_ts=1061))

        self.assertEqual(1, deleted_count)
        self.assertEqual([(1001, 2002)], fake_bot.deleted)
        with closing(sqlite3.connect(app.DB_PATH)) as conn:
            remaining = conn.execute("SELECT COUNT(*) FROM monitor_messages").fetchone()[0]
        self.assertEqual(0, remaining)

    def test_unexpired_monitor_message_is_kept(self) -> None:
        app.record_monitor_message(1001, 2002, "NodeSeek 新帖", delete_after_seconds=60, sent_at_ts=1000)

        fake_bot = FakeBot()
        deleted_count = asyncio.run(app.delete_expired_monitor_messages(fake_bot, now_ts=1059))

        self.assertEqual(0, deleted_count)
        self.assertEqual([], fake_bot.deleted)
        with closing(sqlite3.connect(app.DB_PATH)) as conn:
            remaining = conn.execute("SELECT COUNT(*) FROM monitor_messages").fetchone()[0]
        self.assertEqual(1, remaining)


class BotConfigurationTest(unittest.TestCase):
    def test_parse_admin_chat_ids_keeps_unique_first_three(self) -> None:
        self.assertEqual([1, 2, 3], app.parse_admin_chat_ids("1,2 2;3,4"))

    def test_bot_is_not_configured_without_token_or_admin_chat_id(self) -> None:
        old_token = os.environ.pop("TELEGRAM_BOT_TOKEN", None)
        old_admin = os.environ.pop("ADMIN_CHAT_ID", None)
        try:
            self.assertFalse(app.bot_env_configured())
        finally:
            if old_token is not None:
                os.environ["TELEGRAM_BOT_TOKEN"] = old_token
            if old_admin is not None:
                os.environ["ADMIN_CHAT_ID"] = old_admin

    def test_bot_is_configured_with_token_and_admin_chat_id(self) -> None:
        old_token = os.environ.get("TELEGRAM_BOT_TOKEN")
        old_admin = os.environ.get("ADMIN_CHAT_ID")
        os.environ["TELEGRAM_BOT_TOKEN"] = "123456:test-token"
        os.environ["ADMIN_CHAT_ID"] = "1001"
        try:
            self.assertTrue(app.bot_env_configured())
        finally:
            if old_token is None:
                os.environ.pop("TELEGRAM_BOT_TOKEN", None)
            else:
                os.environ["TELEGRAM_BOT_TOKEN"] = old_token
            if old_admin is None:
                os.environ.pop("ADMIN_CHAT_ID", None)
            else:
                os.environ["ADMIN_CHAT_ID"] = old_admin

    def test_write_env_values_preserves_existing_session_secret(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            old_env_path = app.ENV_PATH
            app.ENV_PATH = Path(temp_dir) / ".env"
            app.ENV_PATH.write_text("WEB_PANEL_SESSION_SECRET=keep-me\n", encoding="utf-8")
            try:
                app.write_env_values({
                    "TELEGRAM_BOT_TOKEN": "123456:test-token",
                    "ADMIN_CHAT_ID": "1001",
                    "WEB_PANEL_USER": "admin",
                    "WEB_PANEL_PASSWORD": "change-me",
                })
                self.assertIn(
                    "WEB_PANEL_SESSION_SECRET=keep-me",
                    app.ENV_PATH.read_text(encoding="utf-8"),
                )
            finally:
                app.ENV_PATH = old_env_path


class PanelHtmlContractTest(unittest.TestCase):
    def test_login_form_keeps_expected_fields(self) -> None:
        html = app.login_page()
        self.assertIn("action=/login", html)
        self.assertIn("name=username", html)
        self.assertIn("name=password", html)

    def test_monitor_form_keeps_backend_field_names(self) -> None:
        html = app.monitor_form_html()
        for expected in [
            "action='/monitor/create'",
            "name=name",
            "name=mtype",
            "name=url",
            "name=interval_seconds",
            "name=keywords",
            "name=item_selector",
            "name=title_selector",
            "name=link_selector",
            "name=keyword_match",
            "name=new_item",
            "name=price_change",
            "name=stock_change",
            "name=notify_telegram",
        ]:
            self.assertIn(expected, html)

    def test_monitor_form_can_disable_telegram_notification(self) -> None:
        monitor = {
            "type": "rss",
            "interval_seconds": 60,
            "notify_telegram": False,
            "notify_on": {"keyword_match": True},
        }
        html = app.monitor_form_html(monitor)
        self.assertIn("name=notify_telegram", html)
        self.assertNotIn("name=notify_telegram checked", html)

    def test_layout_groups_navigation_by_domain(self) -> None:
        html = app.layout("测试", "<p>ok</p>")
        for expected in ["<b>消息</b>", "<b>监控</b>", "<b>配置</b>", "<b>系统</b>", "私聊广告拦截", "TG 群监听"]:
            self.assertIn(expected, html)

    def test_inbox_copy_describes_two_way_conversation(self) -> None:
        source = Path("app.py").read_text(encoding="utf-8")
        self.assertIn("这里显示双向机器人对话记录", source)
        self.assertIn("管理员 -> 用户", source)

    def test_users_page_keeps_shared_settings_form(self) -> None:
        source = Path("app.py").read_text(encoding="utf-8")
        self.assertIn("action='/users/settings'", source)
        self.assertIn("这里和“Bot / 面板设置”共用同一份 .env", source)

    def test_group_monitor_form_keeps_backend_field_names(self) -> None:
        html = app.group_monitor_form_html()
        for expected in [
            "action='/group-monitors/create'",
            "name=name",
            "name=chat_id",
            "name=keywords",
            "name=exclude_keywords",
            "name=enabled",
            "name=notify_telegram",
            "name=listen_source",
            "name=summary_mode",
            "name=ai_interface",
            "name=ai_base_url",
            "name=ai_api_key",
            "name=ai_model",
            "name=ai_temperature",
            "name=ai_timeout_seconds",
            "name=ai_prompt",
            "name=ai_min_interval_seconds",
            "name=ai_dedupe_window_seconds",
        ]:
            self.assertIn(expected, html)


class SpamAndTemplateConfigTest(unittest.TestCase):
    def test_spam_keyword_hits_follow_config(self) -> None:
        old_config = app.config
        app.config = {"bot": {"spam_filter": {"enabled": True, "keywords": ["博彩", "投资"]}}}
        try:
            self.assertEqual(["博彩"], app.spam_keyword_hits("这里有博彩广告"))
        finally:
            app.config = old_config

    def test_quick_replies_are_loaded_from_config(self) -> None:
        old_config = app.config
        app.config = {"bot": {"quick_replies": [{"title": "收到", "text": "稍后处理"}]}}
        try:
            self.assertEqual("收到", app.list_quick_replies()[0]["title"])
        finally:
            app.config = old_config

    def test_update_spam_keywords_writes_config(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            old_config_path = app.CONFIG_PATH
            old_config = app.config
            app.CONFIG_PATH = Path(temp_dir) / "config.yaml"
            app.CONFIG_PATH.write_text("bot:\n  spam_filter:\n    enabled: true\n    keywords: []\n", encoding="utf-8")
            app.config = {"bot": {"spam_filter": {"enabled": True, "keywords": []}}}
            try:
                self.assertEqual(["广告"], app.update_spam_keywords("add", "广告"))
                self.assertEqual([], app.update_spam_keywords("delete", "广告"))
            finally:
                app.CONFIG_PATH = old_config_path
                app.config = old_config


class GroupMonitorTest(unittest.TestCase):
    def test_ai_api_url_supports_v1_and_plain_base(self) -> None:
        self.assertEqual("https://api.example.com/v1/responses", app.ai_api_url("https://api.example.com", "/responses"))
        self.assertEqual("https://api.example.com/v1/chat/completions", app.ai_api_url("https://api.example.com/v1", "/chat/completions"))

    def test_extract_responses_text_and_chat_text(self) -> None:
        self.assertEqual(
            "hello",
            app.extract_responses_text({"output_text": "hello"}),
        )
        self.assertEqual(
            "a\nb",
            app.extract_responses_text(
                {"output": [{"content": [{"text": "a"}, {"content": "b"}]}]}
            ),
        )
        self.assertEqual(
            "ok",
            app.extract_chat_text({"choices": [{"message": {"content": "ok"}}]}),
        )

    def test_cfg_save_normalizes_group_monitor_ai_fields(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            old_config_path = app.CONFIG_PATH
            old_config = app.config
            old_reload = app.reload_scheduler_jobs
            app.CONFIG_PATH = Path(temp_dir) / "config.yaml"
            app.reload_scheduler_jobs = lambda: None
            try:
                cfg = {
                    "monitors": [],
                    "group_monitors": [
                        {
                            "enabled": True,
                            "chat_id": "-10099",
                            "keywords": ["vps"],
                            "exclude_keywords": [],
                            "summary_mode": "bad-mode",
                            "ai_interface": "bad-iface",
                            "ai_temperature": "x",
                            "ai_timeout_seconds": "0",
                        }
                    ],
                }
                app.cfg_save(cfg)
                saved = app.config["group_monitors"][0]
                self.assertEqual(-10099, saved["chat_id"])
                self.assertEqual("template", saved["summary_mode"])
                self.assertEqual("responses", saved["ai_interface"])
                self.assertEqual("bot", saved["listen_source"])
                self.assertEqual(0.2, saved["ai_temperature"])
                self.assertEqual(1, saved["ai_timeout_seconds"])
                self.assertEqual(app.DEFAULT_GROUP_AI_MIN_INTERVAL_SECONDS, saved["ai_min_interval_seconds"])
                self.assertEqual(app.DEFAULT_GROUP_AI_DEDUPE_WINDOW_SECONDS, saved["ai_dedupe_window_seconds"])
            finally:
                app.CONFIG_PATH = old_config_path
                app.config = old_config
                app.reload_scheduler_jobs = old_reload

    def test_build_group_ai_system_prompt_allows_custom_prompt(self) -> None:
        text = app.build_group_ai_system_prompt("请按项目符号输出")
        self.assertIn("Telegram 群消息摘要助手", text)
        self.assertIn("请按项目符号输出", text)

    def test_group_monitor_allow_send_applies_interval_and_dedupe(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            old_db_path = app.DB_PATH
            app.DB_PATH = Path(temp_dir) / "test.sqlite3"
            app.init_db()
            monitor = {
                "name": "测试群",
                "ai_min_interval_seconds": 30,
                "ai_dedupe_window_seconds": 120,
            }
            try:
                ok1, reason1 = app.group_monitor_allow_send(monitor, "fp1", now_ts=1000)
                ok2, reason2 = app.group_monitor_allow_send(monitor, "fp2", now_ts=1010)
                ok3, reason3 = app.group_monitor_allow_send(monitor, "fp1", now_ts=1040)
                self.assertTrue(ok1)
                self.assertEqual("", reason1)
                self.assertFalse(ok2)
                self.assertIn("min-interval", reason2)
                self.assertFalse(ok3)
                self.assertIn("dedupe", reason3)
            finally:
                app.DB_PATH = old_db_path


class MonitorRuntimeAndUpdateTest(unittest.TestCase):
    def setUp(self) -> None:
        self.temp_dir = tempfile.TemporaryDirectory()
        self.old_db_path = app.DB_PATH
        app.DB_PATH = Path(self.temp_dir.name) / "test.sqlite3"
        app.init_db()

    def tearDown(self) -> None:
        app.DB_PATH = self.old_db_path
        self.temp_dir.cleanup()

    def test_record_monitor_runtime_tracks_failures(self) -> None:
        app.record_monitor_runtime("m1", ok=False, duration_ms=120, sent_count=0, error="oops")
        app.record_monitor_runtime("m1", ok=False, duration_ms=90, sent_count=0, error="oops2")
        app.record_monitor_runtime("m1", ok=True, duration_ms=70, sent_count=2)
        data = app.list_monitor_runtime_status()["m1"]
        self.assertEqual(0, data["consecutive_failures"])
        self.assertEqual(2, data["last_sent_count"])
        self.assertEqual(70, data["last_duration_ms"])

    def test_git_update_status_parses_ahead_behind_and_dirty(self) -> None:
        old_git_run = app.git_run

        class FakeResult:
            def __init__(self, out: str):
                self.stdout = out

        def fake_git_run(repo_dir, args, check=True):
            cmd = " ".join(args)
            if cmd.startswith("fetch "):
                return FakeResult("")
            if cmd == "rev-parse HEAD":
                return FakeResult("abc123\n")
            if cmd == "rev-parse origin/main":
                return FakeResult("def456\n")
            if cmd.startswith("rev-list --left-right --count"):
                return FakeResult("2 5\n")
            if cmd == "status --porcelain":
                return FakeResult(" M app.py\n")
            raise AssertionError(f"unexpected git command: {cmd}")

        app.git_run = fake_git_run
        try:
            st = app.git_update_status(Path("."), "main", fetch_remote=True)
            self.assertEqual("abc123", st["head"])
            self.assertEqual("def456", st["remote_head"])
            self.assertEqual(2, st["ahead"])
            self.assertEqual(5, st["behind"])
            self.assertTrue(st["dirty"])
        finally:
            app.git_run = old_git_run

    def test_group_monitor_for_chat_returns_enabled_target(self) -> None:
        old_config = app.config
        app.config = {
            "group_monitors": [
                {"enabled": True, "chat_id": -10001, "keywords": ["vps"], "exclude_keywords": []},
                {"enabled": False, "chat_id": -10002, "keywords": ["api"]},
            ]
        }
        try:
            monitor = app.group_monitor_for_chat(-10001)
            self.assertIsNotNone(monitor)
            self.assertEqual(-10001, monitor["chat_id"])
            self.assertIsNone(app.group_monitor_for_chat(-10002))
        finally:
            app.config = old_config

    def test_group_monitor_for_chat_and_source_returns_matched_monitor(self) -> None:
        old_config = app.config
        app.config = {
            "group_monitors": [
                {"enabled": True, "chat_id": -10001, "listen_source": "bot", "keywords": ["vps"]},
                {"enabled": True, "chat_id": -10001, "listen_source": "user_session", "keywords": ["api"]},
            ]
        }
        try:
            monitor_bot = app.group_monitor_for_chat_and_source(-10001, "bot")
            monitor_session = app.group_monitor_for_chat_and_source(-10001, "user_session")
            self.assertIsNotNone(monitor_bot)
            self.assertIsNotNone(monitor_session)
            self.assertEqual("bot", monitor_bot["listen_source"])
            self.assertEqual("user_session", monitor_session["listen_source"])
        finally:
            app.config = old_config

    def test_handle_group_keyword_message_sends_summary_to_admin(self) -> None:
        old_config = app.config
        old_bot = app.bot
        old_admin_chat_ids = app.admin_chat_ids
        fake_bot = FakeBot()
        app.bot = fake_bot
        app.admin_chat_ids = [9001]
        app.config = {
            "group_monitors": [
                {
                    "enabled": True,
                    "name": "测试群",
                    "chat_id": -100100100,
                    "keywords": ["vps", "优惠"],
                    "exclude_keywords": ["求带"],
                    "notify_telegram": True,
                }
            ]
        }
        msg = SimpleNamespace(
            chat=SimpleNamespace(id=-100100100, username="groupdemo", title="测试群"),
            from_user=SimpleNamespace(id=123, first_name="Alice", last_name="", username="alice"),
            text="今晚 vps 有优惠",
            caption=None,
            reply_to_message=None,
            message_id=777,
            content_type="text",
        )
        try:
            ok = asyncio.run(app.handle_group_keyword_message(msg))
            self.assertTrue(ok)
            self.assertEqual([9001], fake_bot.sent_chat_ids)
            self.assertIn("[群关键词命中]", fake_bot.sent_texts[0])
            self.assertIn("命中：vps, 优惠", fake_bot.sent_texts[0])
        finally:
            app.config = old_config
            app.bot = old_bot
            app.admin_chat_ids = old_admin_chat_ids

    def test_handle_group_keyword_message_respects_exclude_keywords(self) -> None:
        old_config = app.config
        old_bot = app.bot
        old_admin_chat_ids = app.admin_chat_ids
        fake_bot = FakeBot()
        app.bot = fake_bot
        app.admin_chat_ids = [9001]
        app.config = {
            "group_monitors": [
                {
                    "enabled": True,
                    "chat_id": -100100100,
                    "keywords": ["vps"],
                    "exclude_keywords": ["求带"],
                    "notify_telegram": True,
                }
            ]
        }
        msg = SimpleNamespace(
            chat=SimpleNamespace(id=-100100100, username="groupdemo", title="测试群"),
            from_user=SimpleNamespace(id=123, first_name="Alice", last_name="", username="alice"),
            text="vps 求带",
            caption=None,
            reply_to_message=None,
            message_id=777,
            content_type="text",
        )
        try:
            ok = asyncio.run(app.handle_group_keyword_message(msg))
            self.assertFalse(ok)
            self.assertEqual([], fake_bot.sent_chat_ids)
        finally:
            app.config = old_config
            app.bot = old_bot
            app.admin_chat_ids = old_admin_chat_ids

    def test_group_ai_summary_fallback_to_template_when_ai_fails(self) -> None:
        old_config = app.config
        old_bot = app.bot
        old_admin_chat_ids = app.admin_chat_ids
        old_ai = app.summarize_group_message_ai
        fake_bot = FakeBot()
        app.bot = fake_bot
        app.admin_chat_ids = [9001]
        app.config = {
            "group_monitors": [
                {
                    "enabled": True,
                    "name": "测试群",
                    "chat_id": -100100100,
                    "keywords": ["vps"],
                    "exclude_keywords": [],
                    "notify_telegram": True,
                    "summary_mode": "ai",
                    "ai_base_url": "https://api.example.com/v1",
                    "ai_api_key": "sk-test",
                    "ai_model": "gpt-4o-mini",
                    "ai_interface": "responses",
                }
            ]
        }

        async def fail_ai(message, monitor, hits):
            raise RuntimeError("ai failed")

        app.summarize_group_message_ai = fail_ai
        msg = SimpleNamespace(
            chat=SimpleNamespace(id=-100100100, username="groupdemo", title="测试群"),
            from_user=SimpleNamespace(id=123, first_name="Alice", last_name="", username="alice"),
            text="今晚 vps 有货",
            caption=None,
            reply_to_message=None,
            message_id=888,
            content_type="text",
        )
        try:
            ok = asyncio.run(app.handle_group_keyword_message(msg))
            self.assertTrue(ok)
            self.assertEqual([9001], fake_bot.sent_chat_ids)
            self.assertIn("[群AI总结失败，已使用模板]", fake_bot.sent_texts[0])
            self.assertIn("[群关键词命中]", fake_bot.sent_texts[0])
        finally:
            app.config = old_config
            app.bot = old_bot
            app.admin_chat_ids = old_admin_chat_ids
            app.summarize_group_message_ai = old_ai

    def test_record_and_list_discovered_group_chats(self) -> None:
        msg = SimpleNamespace(
            chat=SimpleNamespace(id=-100123, type="supergroup", title="测试群A", username="group_a"),
            text="hello",
            caption=None,
            reply_to_message=None,
            message_id=1,
            from_user=SimpleNamespace(id=11, first_name="u", last_name="", username="u1"),
            content_type="text",
        )
        app.record_discovered_group_chat(msg)
        rows = app.list_discovered_group_chats()
        self.assertTrue(rows)
        self.assertEqual(-100123, rows[0]["chat_id"])
        self.assertEqual("测试群A", rows[0]["title"])

    def test_group_monitors_page_keeps_discovered_chat_actions_markup(self) -> None:
        source = Path("app.py").read_text(encoding="utf-8")
        self.assertIn("已发现群聊", source)
        self.assertIn("用此群创建监听", source)
        self.assertIn("/group-monitors/new?chat_id=", source)


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