admin_bot: add emergency pause, manual fallback configuration, and admin management UI

2026-08-28 19:34:05 +03:30
parent a7c1c37f01
commit cbd5247e2a
4 changed files with 2435 additions and 383 deletions
@@ -0,0 +1,89 @@
+"""Channel lists render as one button menu, and each button opens that channel's card."""
+import asyncio
+import sys
+
+sys.path.insert(0, "/app")
+
+from db.database import init_db, close_db_pool
+from db.repository import Repository
+from services.admin_bot import AdminBotService
+
+SRC = -1009999000030
+TRG = -1009999000031
+
+
+async def _cleanup(repo):
+    pool = await repo._get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute("DELETE FROM posts WHERE source_channel_id = $1;", SRC)
+        await conn.execute("DELETE FROM sources WHERE channel_id = $1;", SRC)
+        await conn.execute("DELETE FROM targets WHERE channel_id = $1;", TRG)
+
+
+def button_data(rows):
+    return [b.data.decode() for row in rows for b in row]
+
+
+async def run_tests():
+    await init_db()
+    repo = Repository()
+    await _cleanup(repo)
+    try:
+        await _run_assertions(repo)
+    finally:
+        await _cleanup(repo)
+        await close_db_pool()
+    print("All admin menu tests passed successfully!")
+
+
+async def _run_assertions(repo):
+    bot = AdminBotService(repo=repo, review_channel_id=0, admin_user_ids=[1])
+
+    src_id = await repo.add_source(SRC, "Menu Source", "menusrc")
+    target_id = await repo.add_target(TRG, "Menu Target", None)
+    await repo.create_raw_post(SRC, 1, "a post", content_hash="menu_1")
+
+    # Source list: one button per source, pointing at its detail view.
+    text, buttons = await bot._render_source_list()
+    assert "Menu Source" not in text, "the list message itself should stay a header, not a card dump"
+    data = button_data(buttons)
+    assert f"src_view:{src_id}" in data, f"source button missing: {data}"
+
+    # Target list likewise, and it flags auto-routing.
+    text, buttons = await bot._render_target_list()
+    data = button_data(buttons)
+    assert f"trg_view:{target_id}" in data, f"target button missing: {data}"
+    def label_for(rows, want):
+        for row in rows:
+            for b in row:
+                if b.data.decode() == want:
+                    return b.text
+        raise AssertionError(f"button {want} not found")
+
+    assert "🤖" not in label_for(buttons, f"trg_view:{target_id}"), "no auto-route configured yet"
+    await repo.set_target_auto_sources(target_id, [SRC])
+    _, buttons = await bot._render_target_list()
+    assert "🤖" in label_for(buttons, f"trg_view:{target_id}"), "auto-routed target should be flagged"
+
+    # Source detail card carries fetch buttons and a way back.
+    card, buttons = await bot._render_source_config(src_id)
+    assert "Menu Source" in card and str(SRC) in card
+    assert "<b>1</b>" in card, f"collected count missing from card: {card}"
+    assert "Menu Target" in card, "auto-routed target should be listed on the source card"
+    data = button_data(buttons)
+    assert f"hist:{SRC}:20" in data and f"histn:{SRC}" in data, f"fetch buttons missing: {data}"
+    assert "list_src" in data, f"back button missing: {data}"
+
+    # Target detail card can get back to its list too.
+    card, buttons = await bot._render_target_config(target_id)
+    data = button_data(buttons)
+    assert f"auto_src:{target_id}" in data, f"auto-route button missing: {data}"
+    assert "list_trg" in data, f"back button missing: {data}"
+
+    # Missing ids degrade gracefully instead of raising.
+    card, buttons = await bot._render_source_config(999999)
+    assert "یافت نشد" in card and buttons == []
+
+
+if __name__ == "__main__":
+    asyncio.run(run_tests())
@@ -0,0 +1,91 @@
+"""Errors can be grouped, marked fixed, and reported as metrics."""
+import asyncio
+import sys
+
+sys.path.insert(0, "/app")
+
+from db.database import init_db, close_db_pool
+from db.repository import Repository
+from services.admin_bot import AdminBotService, clamp_for_telegram, MAX_MEDIA_CAPTION, MAX_TEXT_MESSAGE
+
+SVC = "test.tracking_service"
+
+
+async def _cleanup(repo):
+    pool = await repo._get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute("DELETE FROM error_logs WHERE service_name LIKE 'test.tracking%';")
+
+
+async def _log(repo, service, etype, msg):
+    pool = await repo._get_pool()
+    async with pool.acquire() as conn:
+        await conn.execute(
+            "INSERT INTO error_logs (service_name, error_type, error_message) VALUES ($1,$2,$3);",
+            service, etype, msg)
+
+
+def test_clamping():
+    # Short text is untouched.
+    assert clamp_for_telegram("hello", True) == "hello"
+
+    long_text = "x" * 5000
+    media = clamp_for_telegram(long_text, has_media=True)
+    assert len(media) <= MAX_MEDIA_CAPTION, f"media caption still {len(media)} chars"
+    assert media.endswith("</i>"), "truncation marker missing"
+
+    plain = clamp_for_telegram(long_text, has_media=False)
+    assert len(plain) <= MAX_TEXT_MESSAGE, f"text message still {len(plain)} chars"
+
+    # A message that exactly fits must not be altered.
+    exact = "y" * MAX_MEDIA_CAPTION
+    assert clamp_for_telegram(exact, True) == exact
+
+
+async def run_tests():
+    test_clamping()
+
+    await init_db()
+    repo = Repository()
+    await _cleanup(repo)
+
+    await _log(repo, SVC, "ValueError", "bad value one")
+    await _log(repo, SVC, "ValueError", "bad value two")
+    await _log(repo, SVC, "KeyError", "missing key")
+    await _log(repo, SVC + "_b", "ValueError", "other service")
+
+    groups = await repo.get_open_error_summary(limit=50)
+    mine = {(g["service_name"], g["error_type"]): g for g in groups if g["service_name"].startswith("test.tracking")}
+    assert len(mine) == 3, f"expected 3 groups, got {list(mine)}"
+    assert mine[(SVC, "ValueError")]["occurrences"] == 2
+    assert mine[(SVC, "ValueError")]["last_message"] == "bad value two", "last_message should be the newest"
+
+    # Resolving one group leaves the others untouched.
+    fixed = await repo.resolve_errors(service_name=SVC, error_type="ValueError", note="fixed in tests")
+    assert fixed == 2, f"expected 2 rows resolved, got {fixed}"
+    groups = await repo.get_open_error_summary(limit=50)
+    mine = {(g["service_name"], g["error_type"]) for g in groups if g["service_name"].startswith("test.tracking")}
+    assert (SVC, "ValueError") not in mine
+    assert (SVC, "KeyError") in mine and (SVC + "_b", "ValueError") in mine
+
+    # Resolving again is a no-op, not a double count.
+    assert await repo.resolve_errors(service_name=SVC, error_type="ValueError") == 0
+
+    # The gauge reflects what is still open.
+    bot = AdminBotService(repo=repo, review_channel_id=0, admin_user_ids=[1])
+    total = await bot.refresh_error_metrics()
+    assert total == await repo.count_open_errors(), "gauge total must match the open count"
+
+    # The report renders and caches groups for the callback buttons.
+    text, buttons = await bot._render_error_report()
+    assert "KeyError" in text
+    assert bot.error_group_cache, "groups must be cached for errfix callbacks"
+    assert any(b.data.decode().startswith("errfix:") for row in buttons for b in row)
+
+    await _cleanup(repo)
+    await close_db_pool()
+    print("All error-tracking tests passed successfully!")
+
+
+if __name__ == "__main__":
+    asyncio.run(run_tests())
@@ -0,0 +1,146 @@
+import asyncio
+from unittest.mock import AsyncMock, patch, MagicMock
+from db.models import AIProviderProfile
+from core.llm import LLMClient
+from services.admin_bot import get_persian_main_menu, AdminBotService
+
+
+def test_main_menu_pause_toggle():
+    menu_running = get_persian_main_menu(is_paused=False)
+    button_texts_running = [getattr(b.button, 'text', '') for row in menu_running for b in row]
+    assert "🛑 توقف اضطراری سیستم" in button_texts_running
+    assert "▶️ راه‌اندازی و ادامه سیستم" not in button_texts_running
+
+    menu_paused = get_persian_main_menu(is_paused=True)
+    button_texts_paused = [getattr(b.button, 'text', '') for row in menu_paused for b in row]
+    assert "▶️ راه‌اندازی و ادامه سیستم" in button_texts_paused
+    assert "🛑 توقف اضطراری سیستم" not in button_texts_paused
+
+
+async def test_llm_fallback_chain_success_on_fallback():
+    # Setup 2 providers: Primary (fails) -> Fallback (succeeds)
+    profile_1 = AIProviderProfile(
+        id=1,
+        name="Primary AI",
+        provider_type="openai",
+        model="gpt-4o",
+        base_url="https://api.openai.com/v1",
+        api_key="key1",
+        is_active=True,
+        fallback_provider_id=2
+    )
+    profile_2 = AIProviderProfile(
+        id=2,
+        name="Backup AI",
+        provider_type="gemini",
+        model="gemini-1.5-flash",
+        base_url="",
+        api_key="key2",
+        is_active=False,
+        fallback_provider_id=None
+    )
+
+    repo_mock = AsyncMock()
+    repo_mock.get_active_provider_profile.return_value = profile_1
+    repo_mock.get_provider_profiles.return_value = [profile_1, profile_2]
+    repo_mock.record_ai_log = AsyncMock()
+
+    fallback_alerts = []
+
+    async def on_fallback(from_p, to_p, err, step):
+        fallback_alerts.append((from_p.name, to_p.name, err, step))
+
+    client = LLMClient(
+        repo=repo_mock,
+        on_fallback_alert=on_fallback,
+    )
+    client.max_retries_per_model = 0  # fail fast for test
+
+    # Mock _call_openai to fail and _call_gemini to succeed
+    with patch.object(client, "_call_openai", side_effect=RuntimeError("Primary Provider Timeout")):
+        with patch.object(client, "_call_gemini", return_value={"decision": "accept", "rewritten_text": "Success from backup"}):
+            res = await client.generate_json("Test prompt", system_prompt="Test sys")
+            assert res["decision"] == "accept"
+            assert res["rewritten_text"] == "Success from backup"
+            assert len(fallback_alerts) == 1
+            assert fallback_alerts[0][0] == "Primary AI"
+            assert fallback_alerts[0][1] == "Backup AI"
+            assert "Primary Provider Timeout" in fallback_alerts[0][2]
+            assert fallback_alerts[0][3] == 1
+
+
+async def test_llm_fallback_chain_all_fail_alert():
+    profile_1 = AIProviderProfile(
+        id=1,
+        name="Primary AI",
+        provider_type="openai",
+        model="gpt-4o",
+        is_active=True,
+        fallback_provider_id=2
+    )
+    profile_2 = AIProviderProfile(
+        id=2,
+        name="Backup AI 1",
+        provider_type="openai",
+        model="gpt-4o-mini",
+        is_active=False,
+        fallback_provider_id=3
+    )
+    profile_3 = AIProviderProfile(
+        id=3,
+        name="Backup AI 2",
+        provider_type="gemini",
+        model="gemini-1.5-flash",
+        is_active=False,
+        fallback_provider_id=None
+    )
+
+    repo_mock = AsyncMock()
+    repo_mock.get_active_provider_profile.return_value = profile_1
+    repo_mock.get_provider_profiles.return_value = [profile_1, profile_2, profile_3]
+    repo_mock.record_ai_log = AsyncMock()
+
+    fallback_alerts = []
+    chain_failure_alerts = []
+
+    async def on_fallback(from_p, to_p, err, step):
+        fallback_alerts.append((from_p.name, to_p.name, err, step))
+
+    async def on_chain_failure(chain, err):
+        chain_failure_alerts.append((len(chain), err))
+
+    client = LLMClient(
+        repo=repo_mock,
+        on_fallback_alert=on_fallback,
+        on_chain_failure_alert=on_chain_failure,
+    )
+    client.max_retries_per_model = 0
+
+    with patch.object(client, "_call_openai", side_effect=RuntimeError("OpenAI Error")):
+        with patch.object(client, "_call_gemini", side_effect=RuntimeError("Gemini Quota Exceeded")):
+            try:
+                await client.generate_json("Test prompt")
+                assert False, "Should have raised exception"
+            except Exception as e:
+                assert "Gemini Quota Exceeded" in str(e) or "OpenAI Error" in str(e)
+
+            # Check fallbacks: 1 -> 2, 2 -> 3
+            assert len(fallback_alerts) == 2
+            assert fallback_alerts[0][0] == "Primary AI"
+            assert fallback_alerts[0][1] == "Backup AI 1"
+            assert fallback_alerts[1][0] == "Backup AI 1"
+            assert fallback_alerts[1][1] == "Backup AI 2"
+
+            # Check chain failure alert triggered
+            assert len(chain_failure_alerts) == 1
+            assert chain_failure_alerts[0][0] == 3
+
+
+async def main():
+    test_main_menu_pause_toggle()
+    await test_llm_fallback_chain_success_on_fallback()
+    await test_llm_fallback_chain_all_fail_alert()
+    print("All system stop and multi-provider AI fallback chain tests passed successfully!")
+
+if __name__ == "__main__":
+    asyncio.run(main())