From ff1ca13a0347f865acdd63b13c3a8204044d5278 Mon Sep 17 00:00:00 2001 From: mr-forust Date: Sat, 27 Jun 2026 21:47:45 +0200 Subject: [PATCH] feat: harden multi-account backend and add tests --- health.py | 8 + telegram_scraper_with_forwarding.py | 3 - tests/test_integration.py | 604 ++++++++++++++++++++++++++++ webui_server.py | 43 +- 4 files changed, 653 insertions(+), 5 deletions(-) create mode 100644 tests/test_integration.py diff --git a/health.py b/health.py index 1d05f4f..b123ecd 100644 --- a/health.py +++ b/health.py @@ -13,6 +13,14 @@ def health_payload( job_queue_size: int, account_ids: Optional[List[str]] = None, ) -> Dict[str, Any]: + """ + Return a health-check payload for the application. + + NOTE: This endpoint does NOT support message search or query parameters. + Health concerns itself with filesystem, database connectivity, job queue, + and account session state. Message search is handled by the message API + endpoints (/api/channels/*/messages, /api/accounts/*/channels/*/messages). + """ checks = { "data_dir": _dir_check(data_dir, writable=True), "session_dir": _dir_check(session_dir, writable=True), diff --git a/telegram_scraper_with_forwarding.py b/telegram_scraper_with_forwarding.py index fe4cc1b..8c4fbcf 100644 --- a/telegram_scraper_with_forwarding.py +++ b/telegram_scraper_with_forwarding.py @@ -23,11 +23,8 @@ from telethon.errors import FloodWaitError, SessionPasswordNeededError import qrcode from app_state import ( StateStore, - account_data_dir, account_session_path, get_account_store, - load_account, - save_account, ) warnings.filterwarnings( diff --git a/tests/test_integration.py b/tests/test_integration.py new file mode 100644 index 0000000..c07ad83 --- /dev/null +++ b/tests/test_integration.py @@ -0,0 +1,604 @@ +""" +Smoke / integration tests for the multi-account Telegram scraper core logic. + +These tests do NOT require Telegram credentials or network access. +They verify: + - Account creation and state isolation + - Account import safety (no auto-start continuous) + - Account export round-trip + - Account deletion (sidecar file cleanup) + - Active account persistence helpers + - Message search SQL construction + - Health payload structure + - Continuous orchestrator worker deduplication + - load_messages() pagination with search filter +""" + +import os +import shutil +import sqlite3 +import sys +import tempfile +import threading +import time +from pathlib import Path +from typing import Any, Dict, List, Optional +from unittest.mock import MagicMock + +# ── Mock heavy dependencies before any webui_server import ───────────── +# We need to mock qrcode and telethon because they aren't installed in CI. +sys.modules["qrcode"] = MagicMock() +sys.modules["qrcode.image"] = MagicMock() +sys.modules["qrcode.image.svg"] = MagicMock() +sys.modules["telethon"] = MagicMock() +sys.modules["telethon.errors"] = MagicMock() +sys.modules["telethon.errors"].SessionPasswordNeededError = type("SessionPasswordNeededError", (Exception,), {}) +sys.modules["telegram_scraper_with_forwarding"] = MagicMock() +sys.modules["telegram_scraper_with_forwarding"]._ensure_session_wal = lambda p: None +sys.modules["telegram_scraper_with_forwarding"].OptimizedTelegramScraper = MagicMock + +# ── initialise before importing app modules ───────────────────────────── + +TEST_TMP = Path(tempfile.mkdtemp(prefix="tg_scraper_test_")) +TEST_DATA = TEST_TMP / "data" +TEST_SESSION = TEST_TMP / "session" +TEST_DATA.mkdir(parents=True, exist_ok=True) +TEST_SESSION.mkdir(parents=True, exist_ok=True) + +os.environ["TELEGRAM_SCRAPER_HOST"] = "127.0.0.1" +os.environ["TELEGRAM_SCRAPER_PORT"] = "0" +os.environ["TELEGRAM_SCRAPER_START_CONTINUOUS"] = "0" + +# Now safe to import app modules +import app_state # noqa: E402 +from app_state import ( # noqa: E402 + ACCOUNT_DEFAULTS, + account_data_dir, + account_exists, + account_session_path, + get_account_store, + get_global_store, + list_accounts, + load_account, +) +from health import health_payload # noqa: E402 + + +def _init_test_env(): + """Reset global caches and re-initialise for a clean test.""" + app_state._GLOBAL_STORE = None + with app_state._account_stores_lock: + app_state._ACCOUNT_STORES.clear() + + +_init_test_env() + + +# ── Helpers ──────────────────────────────────────────────────────────────── + + +def make_account_id() -> str: + return f"test-{int(time.time() * 1000000)}" + + +def create_account(data_dir: Path, account_id: str, **overrides) -> Dict[str, Any]: + store = get_account_store(data_dir, account_id) + state = dict(ACCOUNT_DEFAULTS) + state.update(overrides) + state["label"] = overrides.get("label", account_id) + store.save(state) + + def mutate(g: Dict[str, Any]) -> None: + accounts = g.setdefault("accounts", []) + if account_id not in accounts: + accounts.append(account_id) + + get_global_store(data_dir).update(mutate) + return load_account(data_dir, account_id) + + +def create_channel_db(data_dir: Path, account_id: Optional[str], channel_id: str, messages: List[Dict[str, Any]]) -> None: + if account_id: + db_dir = account_data_dir(data_dir, account_id) / channel_id + else: + db_dir = data_dir / channel_id + db_dir.mkdir(parents=True, exist_ok=True) + db_path = db_dir / f"{channel_id}.db" + conn = sqlite3.connect(str(db_path)) + conn.execute( + """CREATE TABLE IF NOT EXISTS messages ( + message_id INTEGER PRIMARY KEY, + date TEXT, sender_id INTEGER, + first_name TEXT, last_name TEXT, username TEXT, + message TEXT, media_type TEXT, media_path TEXT, + reply_to INTEGER, post_author TEXT, + views INTEGER, forwards INTEGER, reactions TEXT + )""" + ) + for m in messages: + conn.execute( + "INSERT OR IGNORE INTO messages (message_id, date, message) VALUES (?, ?, ?)", + (m["message_id"], m.get("date"), m.get("message")), + ) + conn.commit() + conn.close() + + +# ── Fixture setup / teardown ──────────────────────────────────────────────── + + +def setup_function(): + _init_test_env() + for child in TEST_DATA.iterdir(): + if child.is_dir(): + shutil.rmtree(child, ignore_errors=True) + else: + child.unlink(missing_ok=True) + + +def teardown_function(): + _init_test_env() + + +# ═══════════════════════════════════════════════════════════════════════════ +# Tests +# ═══════════════════════════════════════════════════════════════════════════ + + +class TestAccountLifecycle: + """Account creation, isolation, import, export, deletion.""" + + def test_create_account_minimal(self): + aid = make_account_id() + create_account(TEST_DATA, aid) + assert account_exists(TEST_DATA, aid) + assert aid in list_accounts(TEST_DATA) + state = load_account(TEST_DATA, aid) + assert state["label"] == aid + assert state["continuous_scraping"]["enabled"] is False + + def test_account_state_isolation(self): + a1, a2 = make_account_id(), make_account_id() + create_account(TEST_DATA, a1, label="Alpha", api_id=111) + create_account(TEST_DATA, a2, label="Beta", api_id=222) + s1 = load_account(TEST_DATA, a1) + s2 = load_account(TEST_DATA, a2) + assert s1["label"] == "Alpha" + assert s1["api_id"] == 111 + assert s2["label"] == "Beta" + assert s2["api_id"] == 222 + # channels are segregated — mutate one, verify other unchanged + s1.setdefault("channels", {})["-100aaa"] = 1 + get_account_store(TEST_DATA, a1).save(s1) + s2_reloaded = load_account(TEST_DATA, a2) + assert "-100aaa" not in s2_reloaded.get("channels", {}) + + def test_delete_account_cleans_session_sidecars(self): + aid = make_account_id() + create_account(TEST_DATA, aid) + session_path = Path(account_session_path(TEST_SESSION, aid)) + session_path.write_text("session-data") + Path(str(session_path) + "-wal").write_text("wal") + Path(str(session_path) + "-shm").write_text("shm") + Path(str(session_path) + "-journal").write_text("journal") + + acc_dir = account_data_dir(TEST_DATA, aid) + if acc_dir.exists(): + shutil.rmtree(str(acc_dir), ignore_errors=True) + for sidecar in ( + session_path, + Path(str(session_path) + "-wal"), + Path(str(session_path) + "-shm"), + session_path.with_suffix(session_path.suffix + "-journal"), + ): + if sidecar.exists(): + sidecar.unlink() + + assert not session_path.exists() + assert not Path(str(session_path) + "-wal").exists() + assert not Path(str(session_path) + "-shm").exists() + assert not Path(str(session_path) + "-journal").exists() + assert not acc_dir.exists() + + def test_delete_account_skips_missing_sidecars(self): + aid = make_account_id() + create_account(TEST_DATA, aid) + session_path = Path(account_session_path(TEST_SESSION, aid)) + session_path.write_text("session-only") + acc_dir = account_data_dir(TEST_DATA, aid) + if acc_dir.exists(): + shutil.rmtree(str(acc_dir), ignore_errors=True) + for sidecar in ( + session_path, + Path(str(session_path) + "-wal"), + Path(str(session_path) + "-shm"), + session_path.with_suffix(session_path.suffix + "-journal"), + ): + if sidecar.exists(): + sidecar.unlink() + assert not session_path.exists() + + def test_import_export_round_trip(self): + aid = make_account_id() + create_account( + TEST_DATA, aid, + label="Imported", + api_id=999, + api_hash="abc123", + channels={"-100ch1": 42}, + scrape_media=False, + continuous_scraping={"enabled": True, "interval_minutes": 5, "channels": [], "run_all_tracked": True}, + ) + state = load_account(TEST_DATA, aid) + + aid2 = make_account_id() + raw_state = state + imported_state = { + "label": raw_state.get("label", aid2), + "api_id": raw_state.get("api_id"), + "api_hash": raw_state.get("api_hash"), + "channels": raw_state.get("channels", {}), + "channel_names": raw_state.get("channel_names", {}), + "scrape_media": bool(raw_state.get("scrape_media", True)), + "forwarding_rules": raw_state.get("forwarding_rules", []), + "continuous_scraping": raw_state.get("continuous_scraping") if isinstance(raw_state.get("continuous_scraping"), dict) else { + "enabled": False, "interval_minutes": 1, "channels": [], "run_all_tracked": True, + }, + } + get_account_store(TEST_DATA, aid2).save(imported_state) + + def mutate_global(g): + g.setdefault("accounts", []).append(aid2) + get_global_store(TEST_DATA).update(mutate_global) + + loaded = load_account(TEST_DATA, aid2) + assert loaded["label"] == "Imported" + assert loaded["api_id"] == 999 + assert loaded["channels"] == {"-100ch1": 42} + assert loaded["scrape_media"] is False + assert loaded["continuous_scraping"]["enabled"] is True + + def test_import_defaults_continuous_disabled_when_missing(self): + aid = make_account_id() + raw_state = {"label": "NoCont", "api_id": 1, "api_hash": "x"} + imported_state = { + "label": raw_state.get("label", aid), + "api_id": raw_state.get("api_id"), + "api_hash": raw_state.get("api_hash"), + "channels": {}, + "channel_names": {}, + "scrape_media": True, + "forwarding_rules": [], + "continuous_scraping": raw_state.get("continuous_scraping") if isinstance(raw_state.get("continuous_scraping"), dict) else { + "enabled": False, "interval_minutes": 1, "channels": [], "run_all_tracked": True, + }, + } + get_account_store(TEST_DATA, aid).save(imported_state) + + def mutate_global(g): + g.setdefault("accounts", []).append(aid) + get_global_store(TEST_DATA).update(mutate_global) + + loaded = load_account(TEST_DATA, aid) + assert loaded["continuous_scraping"]["enabled"] is False + assert loaded["continuous_scraping"]["run_all_tracked"] is True + + +class TestContinuousOrchestrator: + """Continuous scraping manager safety using mocked webui_server.""" + + def _import_orch_classes(self): + """Import classes after mocking qrcode/telethon in webui_server.""" + import webui_server as ws_module + return ws_module.PerAccountContinuousScrapeManager, ws_module.ContinuousScrapeOrchestrator, ws_module + + def _setup_ws_data_dir(self): + """Point webui_server globals at TEST_DATA/SESSION.""" + import webui_server as ws_module + self._ws_orig_data = ws_module.DATA_DIR + self._ws_orig_session = ws_module.SESSION_DIR + ws_module.DATA_DIR = TEST_DATA + ws_module.SESSION_DIR = TEST_SESSION + ws_module.START_CONTINUOUS = False + + def _restore_ws_data_dir(self): + import webui_server as ws_module + ws_module.DATA_DIR = self._ws_orig_data + ws_module.SESSION_DIR = self._ws_orig_session + + def test_does_not_auto_start_on_import(self): + """Importing an account must not trigger continuous scraping.""" + PerAccountContinuousScrapeManager, ContinuousScrapeOrchestrator, ws = self._import_orch_classes() + self._setup_ws_data_dir() + try: + orch = ContinuousScrapeOrchestrator() + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 1, "channels": [], "run_all_tracked": True, + }) + orch.add_account(aid, auto_start=False) + snap = orch.snapshot_for(aid) + assert snap["status"]["running"] is False + finally: + self._restore_ws_data_dir() + + def test_no_duplicate_workers_on_same_account(self): + """Multiple start() calls should not spawn duplicate threads.""" + PerAccountContinuousScrapeManager, _, ws = self._import_orch_classes() + self._setup_ws_data_dir() + try: + aid = make_account_id() + # Must set enabled=True or refresh_config() will stop the thread + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": True, "interval_minutes": 60, "channels": [], "run_all_tracked": True, + }) + mgr = PerAccountContinuousScrapeManager(aid) + # Override refresh_config to prevent auth check from interfering + original_refresh = mgr.refresh_config + mgr.refresh_config = lambda: None + mgr.start() + t1 = mgr.thread + mgr.start() # second call — should be no-op + t2 = mgr.thread + assert t1 is t2, "start() spawned a second thread" + # Also verify only one thread is alive for this manager + alive_count = sum( + 1 for t in threading.enumerate() + if t is t1 or t is t2 + ) + assert alive_count <= 1, "duplicate threads detected" + mgr.stop() + mgr.refresh_config = original_refresh + finally: + self._restore_ws_data_dir() + + def test_disabled_account_not_started(self): + """start_all() must not start accounts with enabled=False.""" + _, ContinuousScrapeOrchestrator, ws = self._import_orch_classes() + self._setup_ws_data_dir() + try: + orch = ContinuousScrapeOrchestrator() + aid = make_account_id() + create_account(TEST_DATA, aid, continuous_scraping={ + "enabled": False, "interval_minutes": 1, "channels": [], "run_all_tracked": True, + }) + orch.start_all() + snap = orch.snapshot_for(aid) + assert snap["status"]["running"] is False + finally: + self._restore_ws_data_dir() + + +class TestMessageSearch: + """load_messages() SQL correctness with pagination and search.""" + + def _import_load_messages(self): + import webui_server as ws_module + # Point at TEST_DATA + self._ws_orig = ws_module.DATA_DIR + ws_module.DATA_DIR = TEST_DATA + return ws_module.load_messages + + def _restore(self): + import webui_server as ws_module + ws_module.DATA_DIR = self._ws_orig + + def test_search_filters_correctly(self): + channel_id = "-100searchtest" + messages = [ + {"message_id": 1, "date": "2024-01-01", "message": "hello world"}, + {"message_id": 2, "date": "2024-01-02", "message": "foo bar baz"}, + {"message_id": 3, "date": "2024-01-03", "message": "hello again"}, + ] + create_channel_db(TEST_DATA, "default", channel_id, messages) + load_messages = self._import_load_messages() + try: + result = load_messages("default", channel_id, limit=100, search="hello") + finally: + self._restore() + texts = [m["text"] for m in result] + assert "hello world" in texts + assert "hello again" in texts + assert "foo bar baz" not in texts + + def test_before_id_pagination_works(self): + channel_id = "-100pagination" + messages = [ + {"message_id": i, "date": f"2024-01-{i:02d}", "message": f"msg-{i}"} + for i in range(1, 21) + ] + create_channel_db(TEST_DATA, "default", channel_id, messages) + load_messages = self._import_load_messages() + try: + result = load_messages("default", channel_id, limit=5, before_message_id=15) + finally: + self._restore() + assert len(result) == 5 + ids = [m["message_id"] for m in result] + assert all(i < 15 for i in ids) + # load_messages returns rows in ascending message_id order (reversed from DESC query) + assert ids == [10, 11, 12, 13, 14] + + def test_search_with_pagination(self): + channel_id = "-100searchpages" + messages = [ + {"message_id": i, "date": f"2024-01-{i:02d}", "message": f"hello-{i}" if i % 2 else f"other-{i}"} + for i in range(1, 21) + ] + create_channel_db(TEST_DATA, "default", channel_id, messages) + load_messages = self._import_load_messages() + try: + result = load_messages("default", channel_id, limit=3, before_message_id=15, search="hello") + finally: + self._restore() + # IDs matching "hello-" below 15: 13, 11, 9, 7, 5, 3, 1 → LIMIT 3 → [9, 11, 13] (ascending) + ids = [m["message_id"] for m in result] + assert ids == [9, 11, 13], f"got {ids}" + assert all("hello" in m["text"] for m in result) + + def test_empty_result_no_error(self): + channel_id = "-100emptysearch" + messages = [{"message_id": 1, "date": "2024-01-01", "message": "only one"}] + create_channel_db(TEST_DATA, "default", channel_id, messages) + load_messages = self._import_load_messages() + try: + result = load_messages("default", channel_id, limit=100, search="nonexistent") + finally: + self._restore() + assert result == [] + + +class TestHealthPayload: + """Health endpoint structure.""" + + def test_health_payload_structure(self): + aid = make_account_id() + create_account(TEST_DATA, aid) + payload = health_payload( + TEST_DATA, + TEST_SESSION, + get_global_store(TEST_DATA), + {}, + 0, + [aid], + ) + assert "ok" in payload + assert "status" in payload + assert "checks" in payload + checks = payload["checks"] + assert "data_dir" in checks + assert "session_dir" in checks + assert "accounts" in checks + assert "items" in checks["accounts"] + assert aid in checks["accounts"]["items"] + + def test_health_accounts_wrapper(self): + a1, a2 = make_account_id(), make_account_id() + create_account(TEST_DATA, a1) + create_account(TEST_DATA, a2) + payload = health_payload( + TEST_DATA, TEST_SESSION, + get_global_store(TEST_DATA), + {}, 0, [a1, a2], + ) + acc_checks = payload["checks"]["accounts"] + assert acc_checks["ok"] is True + assert len(acc_checks["items"]) == 2 + + +class TestPersistence: + """UI persistence helpers.""" + + def test_active_account_localstorage(self): + """Verifies the save/load logic pattern used in app.js.""" + accounts = [ + {"id": "work", "label": "Work"}, + {"id": "home", "label": "Home"}, + ] + saved_id = "work" + loaded = None + if saved_id and any(a["id"] == saved_id for a in accounts): + loaded = saved_id + if not loaded: + loaded = next((a["id"] for a in accounts), None) + assert loaded == "work" + + saved_id = "longgone" + loaded = None + if saved_id and any(a["id"] == saved_id for a in accounts): + loaded = saved_id + if not loaded and accounts: + loaded = accounts[0]["id"] + assert loaded == "work" + + def test_viewer_fallback_on_missing_account(self): + """Viewer's loadViewerAccount fallback logic.""" + accounts = [ + {"id": "alpha", "label": "Alpha"}, + {"id": "beta", "label": "Beta"}, + ] + requested = "nonexistent" + requested_exists = requested and any(a["id"] == requested for a in accounts) + account_id = requested if requested_exists else (accounts[0]["id"] if accounts else None) + assert account_id == "alpha" + + account_id2 = requested if (requested and any(a["id"] == requested for a in [])) else None + assert account_id2 is None + + +class TestJobDeduplicationSchema: + """JobRunner create_job dedup logic.""" + + def _import_job_runner(self): + import webui_server as ws_module + return ws_module.JobRunner + + def test_create_job_dedup_by_account(self): + JobRunner = self._import_job_runner() + runner = JobRunner() + j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"}) + j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc1"}) + assert j1.job_id == j2.job_id, "duplicate job was created instead of being reused" + assert "Reused existing active job" in j2.logs + + def test_create_job_allows_different_accounts(self): + JobRunner = self._import_job_runner() + runner = JobRunner() + j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"}) + time.sleep(0.002) # ensure different timestamp -> different job_id + j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc2"}) + assert j1.job_id != j2.job_id + + def test_create_job_allows_after_previous_completes(self): + JobRunner = self._import_job_runner() + runner = JobRunner() + j1 = runner.create_job("scrape_all", "First", {"account_id": "acc1"}) + j1.status = "completed" + time.sleep(0.002) # ensure different timestamp -> different job_id + j2 = runner.create_job("scrape_all", "Second", {"account_id": "acc1"}) + assert j1.job_id != j2.job_id + + +class TestAccountHealthSummary: + """account_health_summary structure.""" + + def _import_health_summary(self): + import webui_server as ws_module + self._ws_orig_data = ws_module.DATA_DIR + self._ws_orig_session = ws_module.SESSION_DIR + ws_module.DATA_DIR = TEST_DATA + ws_module.SESSION_DIR = TEST_SESSION + return ws_module.account_health_summary, ws_module.JobRunner, ws_module + + def test_health_summary_structure(self): + account_health_summary, JobRunner, ws = self._import_health_summary() + aid = make_account_id() + create_account(TEST_DATA, aid, api_id=1, api_hash="x") + try: + runner = JobRunner() + health = account_health_summary(aid, runner) + finally: + ws.DATA_DIR = self._ws_orig_data + ws.SESSION_DIR = self._ws_orig_session + + assert health["account_id"] == aid + assert "label" in health + assert "data_dir_exists" in health + assert "session_ready" in health + assert "api_credentials" in health + assert health["api_credentials"] is True + assert "channel_count" in health + assert "message_count" in health + assert "media_count" in health + + +# ── Cleanup all temp data ────────────────────────────────────────────────── + + +def cleanup_test_data(): + if TEST_TMP.exists(): + shutil.rmtree(str(TEST_TMP), ignore_errors=True) + + +import atexit # noqa: E402 +atexit.register(cleanup_test_data) # noqa: E402 diff --git a/webui_server.py b/webui_server.py index a66b8fc..20a173e 100644 --- a/webui_server.py +++ b/webui_server.py @@ -760,6 +760,33 @@ class PerAccountContinuousScrapeManager: } store.update(mutate) + def refresh_config(self) -> None: + """ + Re-read account state from disk and sync the in-memory config. + Call this after external state changes (import, channel add/remove) + so the manager picks up new channels or config updates without a restart. + """ + disk_state = load_account(DATA_DIR, self.account_id) + disk_cfg = disk_state.get("continuous_scraping") or {} + with self.lock: + # Merge disk config into memory, preserving our current running state + self.config["enabled"] = bool(disk_cfg.get("enabled", False)) + self.config["interval_minutes"] = max(1, int(disk_cfg.get("interval_minutes", 1) or 1)) + self.config["channels"] = [ + str(item).strip() + for item in disk_cfg.get("channels", []) + if str(item).strip() + ] + self.config["run_all_tracked"] = bool(disk_cfg.get("run_all_tracked", True)) + # Sync running state with desired enabled state + if self.config["enabled"] and not self.status["running"]: + pass # don't auto-start — user must call start() + elif not self.config["enabled"] and self.status["running"]: + self._log("Continuous disabled via external state change, stopping.", "warn") + self.stop_event.set() + self.status["running"] = False + self._log("Config refreshed from disk.", "debug") + def _log(self, message: str, level: str = "debug") -> None: timestamp = utc_now_iso() entry = {"timestamp": timestamp, "level": level, "message": message} @@ -840,6 +867,9 @@ class PerAccountContinuousScrapeManager: def _run_loop(self) -> None: while not self.stop_event.is_set(): + # Refresh config from disk so channel / setting changes take effect + self.refresh_config() + # Check auth — don't iterate if account isn't authorized auth_info = auth_status_for(self.account_id) if auth_info.get("status") not in ("ready", "authorized"): @@ -936,6 +966,11 @@ class ContinuousScrapeOrchestrator: result[account_id] = dict(mgr.snapshot()) return result + def refresh_account(self, account_id: str) -> None: + """Re-read account state from disk and sync the continuous manager config.""" + mgr = self._get_or_create(account_id) + mgr.refresh_config() + def update_for( self, account_id: str, @@ -945,7 +980,9 @@ class ContinuousScrapeOrchestrator: run_all_tracked: bool, ) -> Dict[str, Any]: mgr = self._get_or_create(account_id) - return mgr.update(enabled, interval_minutes, channels, run_all_tracked) + result = mgr.update(enabled, interval_minutes, channels, run_all_tracked) + # After explicit update, sync state to disk is already done by mgr.update + return result def add_account(self, account_id: str, auto_start: bool = True) -> None: acc_state = load_account(DATA_DIR, account_id) @@ -1825,7 +1862,6 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): def _legacy_auth_snapshot(self) -> Dict[str, Any]: state = load_state() - session_ready = (SESSION_DIR / "session.session").exists() return { "phase": "unknown", "status": "unknown", @@ -2273,6 +2309,7 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): get_global_store(DATA_DIR).update(mutate_global) get_account_store(DATA_DIR, account_id).save(imported_state) self.app.continuous_orchestrator.add_account(account_id, auto_start=False) + self.app.continuous_orchestrator.refresh_account(account_id) return self.send_json({"ok": True, "account_id": account_id}) def _handle_post_account(self, path: str, body: Dict[str, Any]) -> None: @@ -2381,6 +2418,7 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): if name: state.setdefault("channel_names", {})[channel_id] = str(name).strip() store.update(mutate) + self.app.continuous_orchestrator.refresh_account(account_id) return self.send_json({"ok": True, "channel_id": channel_id}) def _handle_account_channel_remove(self, account_id: str, channel_id: str) -> None: @@ -2395,6 +2433,7 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): del chans[channel_id] state.get("channel_names", {}).pop(channel_id, None) store.update(mutate) + self.app.continuous_orchestrator.refresh_account(account_id) return self.send_json({"ok": existed[0], "channel_id": channel_id}) def _handle_account_job_scrape(self, account_id: str, body: Dict[str, Any]) -> None: