diff --git a/README.md b/README.md index e594767..0ce0cef 100644 --- a/README.md +++ b/README.md @@ -3,9 +3,9 @@ Telegram scraper on top of Telethon with: - CLI workflow for scraping, exports, media recovery, and forwarding -- lightweight Web UI served by Python +- lightweight multi-account Web UI served by Python - SQLite storage per tracked chat/channel -- Docker/Compose setup for "run it and leave it on the server" +- Docker/Compose / Kubernetes setup for "run it and leave it on the server" ## What It Does @@ -15,20 +15,31 @@ Telegram scraper on top of Telethon with: - Exports to CSV and JSON - Supports continuous scraping - Supports forwarding rules -- Lets you browse tracked channels and local exports in a web panel +- Multi-account support — multiple Telegram sessions side by side +- Lets you browse tracked channels and local exports in a web panel with account tabs ## Project Layout ```text . ├── main.py # starts the web server -├── webui_server.py # HTTP server + API +├── webui_server.py # HTTP server + REST API ├── webui/ # plain HTML/CSS/JS frontend +│ ├── index.html / app.js # main dashboard +│ ├── viewer.html / viewer.js # message viewer with account selector +│ └── swagger.html / swagger.js # API docs ├── telegram_scraper_with_forwarding.py +├── app_state.py # multi-account state management +├── scraper_jobs.py # async job runner +├── health.py # health check endpoints ├── data/ -│ ├── state.json # shared settings and credentials -│ └── /.db # SQLite databases -└── session/ # Telethon session files +│ ├── state.json # global state (accounts list) +│ └── accounts// # per-account data +│ ├── state.json # per-account settings & credentials +│ └── / # SQLite DB + media +└── session/ + ├── .session # per-account Telethon session + └── ... # (WAL journal mode for concurrency) ``` ## Requirements @@ -97,11 +108,12 @@ http://:8080 The current web panel includes: -- tracked channels overview +- multi-account support with account tabs +- tracked channels overview per account - background job queue -- scrape/export/media actions +- scrape/export/media actions per account - shared `scrape_media` toggle -- local message viewer powered by SQLite + media files +- local message viewer with account selector - API docs at `/swagger` and `/openapi.json` - health checks at `/health` and `/health/continuous` @@ -113,48 +125,67 @@ If you still want the old interactive mode: python telegram_scraper_with_forwarding.py ``` -## Docker +## Docker / Kubernetes -Build and run: +### Docker Compose ```bash docker compose up -d --build ``` +### Kubernetes + +```bash +kubectl apply -f k8s/telegram-scraper.yaml +``` + Then open: ```text http://:8080 ``` -### Volumes +### Volumes (Docker) - `./data:/app/data` for databases, exports, and `state.json` - `session:/app/session` for Telethon sessions If you already logged in before with the same compose volume and did not remove it, the session should be reused. +## Multi-Account + +The app supports multiple Telegram accounts side by side. Each account has: + +- its own `state.json` under `data/accounts//` +- its own Telethon session file `session/.session` +- its own channel databases and media + +Use the Web UI account tabs to switch between accounts, or create new ones in Settings. + +Legacy single-account data is automatically migrated to the `default` account on first startup. + ## Shared State The Web UI and CLI share the same files: -- `data/state.json` -- `session/session.session` +- `data/state.json` (global — accounts list) +- `data/accounts//state.json` (per-account credentials and settings) +- `session/.session` (per-account Telethon session, WAL journal mode) That means: -- credentials can stay in `state.json` -- both entry points use the same tracked channels -- both entry points can reuse the same Telegram login session +- credentials persist in state files +- both entry points use the same tracked channels per account +- both entry points can reuse the same Telegram login sessions ## Data Storage ### SQLite -Each tracked channel gets its own database: +Each tracked channel per account gets its own database: ```text -data//.db +data/accounts///.db ``` Main fields include: @@ -179,20 +210,22 @@ Main fields include: Media files are stored in: ```text -data//media/ +data/accounts///media/ ``` ### Exports Generated into the same channel folder: -- `data//_.csv` -- `data//_.json` +- `data/accounts///_.csv` +- `data/accounts///_.json` ## Notes - The web server is intentionally simple: plain HTML/CSS/JS, no frontend framework -- The viewer currently reads local SQLite data, not Telegram export HTML directly +- The viewer reads local SQLite data and supports account switching via a dropdown +- Session SQLite files use WAL journal mode to prevent "database is locked" errors when concurrent scraping and jobs run +- The active account tab is persisted in `localStorage` across page reloads - If dependencies are installed but the Telegram session is missing, the Web UI falls back to read-only status until login is available ## Disclaimer diff --git a/app_state.py b/app_state.py index 6d27411..3d6a910 100644 --- a/app_state.py +++ b/app_state.py @@ -1,11 +1,23 @@ import json +import logging +import shutil import threading import time +from copy import deepcopy from pathlib import Path -from typing import Any, Callable, Dict, Optional +from typing import Any, Callable, Dict, List, Optional +logger = logging.getLogger(__name__) -DEFAULT_STATE: Dict[str, Any] = { +# ── defaults ────────────────────────────────────────────────────────── + +GLOBAL_DEFAULTS: Dict[str, Any] = { + "accounts": [], + "version": 2, +} + +ACCOUNT_DEFAULTS: Dict[str, Any] = { + "label": "", "api_id": None, "api_hash": None, "channels": {}, @@ -20,10 +32,15 @@ DEFAULT_STATE: Dict[str, Any] = { }, } +# ── StateStore (thread-safe JSON) ───────────────────────────────────── + class StateStore: - def __init__(self, path: Path): + """Thread-safe JSON state store with TTL cache.""" + + def __init__(self, path: Path, defaults: Optional[Dict[str, Any]] = None): self.path = path + self.defaults = defaults or {} self.lock = threading.RLock() self._cache: Optional[Dict[str, Any]] = None self._cache_time: float = 0 @@ -35,15 +52,15 @@ class StateStore: if self._cache is not None and (now - self._cache_time) < self._cache_ttl: return dict(self._cache) if not self.path.exists(): - result = self._default_state() + result = deepcopy(self.defaults) self._cache = result self._cache_time = now return result try: with self.path.open("r", encoding="utf-8") as handle: - state = json.load(handle) + state: Dict[str, Any] = json.load(handle) except (json.JSONDecodeError, OSError): - result = self._default_state() + result = deepcopy(self.defaults) self._cache = result self._cache_time = now return result @@ -56,10 +73,9 @@ class StateStore: with self.lock: self.path.parent.mkdir(parents=True, exist_ok=True) tmp_path = self.path.with_suffix(self.path.suffix + ".tmp") + merged = self._merge_defaults(state) with tmp_path.open("w", encoding="utf-8") as handle: - json.dump( - self._merge_defaults(state), handle, ensure_ascii=False, indent=2 - ) + json.dump(merged, handle, ensure_ascii=False, indent=2) handle.write("\n") tmp_path.replace(self.path) self._cache = None @@ -73,9 +89,7 @@ class StateStore: def continuous_config(self) -> Dict[str, Any]: state = self.load() - return dict( - state.get("continuous_scraping") or DEFAULT_STATE["continuous_scraping"] - ) + return deepcopy(state.get("continuous_scraping") or ACCOUNT_DEFAULTS["continuous_scraping"]) def save_continuous_config(self, config: Dict[str, Any]) -> Dict[str, Any]: def mutate(state: Dict[str, Any]) -> None: @@ -92,14 +106,208 @@ class StateStore: return self.update(mutate)["continuous_scraping"] - def _default_state(self) -> Dict[str, Any]: - return json.loads(json.dumps(DEFAULT_STATE)) - def _merge_defaults(self, state: Dict[str, Any]) -> Dict[str, Any]: - merged = self._default_state() + merged = deepcopy(self.defaults) for key, value in state.items(): - if key == "continuous_scraping" and isinstance(value, dict): - merged[key].update(value) + if isinstance(value, dict) and isinstance(merged.get(key), dict): + nested = deepcopy(merged[key]) + nested.update(value) + merged[key] = nested else: merged[key] = value return merged + + +# ── global state helpers ────────────────────────────────────────────── + +_GLOBAL_STORE: Optional[StateStore] = None + + +def get_global_store(data_dir: Path) -> StateStore: + global _GLOBAL_STORE + if _GLOBAL_STORE is None: + _GLOBAL_STORE = StateStore(data_dir / "state.json", defaults=GLOBAL_DEFAULTS) + return _GLOBAL_STORE + + +def load_global(data_dir: Path) -> Dict[str, Any]: + return get_global_store(data_dir).load() + + +def save_global(data_dir: Path, state: Dict[str, Any]) -> None: + get_global_store(data_dir).save(state) + + +def list_accounts(data_dir: Path) -> List[str]: + return list(load_global(data_dir).get("accounts", [])) + + +def account_exists(data_dir: Path, account_id: str) -> bool: + return account_id in list_accounts(data_dir) + + +# ── per-account state helpers ───────────────────────────────────────── + +_ACCOUNT_STORES: Dict[str, StateStore] = {} +_account_stores_lock = threading.Lock() + + +def get_account_store(data_dir: Path, account_id: str) -> StateStore: + global _ACCOUNT_STORES + with _account_stores_lock: + key = f"{data_dir}:{account_id}" + if key not in _ACCOUNT_STORES: + store = StateStore( + data_dir / "accounts" / account_id / "state.json", + defaults=ACCOUNT_DEFAULTS, + ) + _ACCOUNT_STORES[key] = store + return _ACCOUNT_STORES[key] + + +def load_account(data_dir: Path, account_id: str) -> Dict[str, Any]: + return get_account_store(data_dir, account_id).load() + + +def save_account(data_dir: Path, account_id: str, state: Dict[str, Any]) -> None: + get_account_store(data_dir, account_id).save(state) + + +def account_data_dir(data_dir: Path, account_id: str) -> Path: + """Per-account directory for channel DBs, media, exports.""" + return data_dir / "accounts" / account_id + + +def account_session_path(session_dir: Path, account_id: str) -> str: + """Per-account Telethon session file. Always under session/.session.""" + return str(session_dir / f"{account_id}.session") + + +# ── MIGRATION (with data copy) ───────────────────────────────────────── + + +def _copy_channel_data(src_root: Path, dst_root: Path, channel_id: str) -> None: + """Copy a single channel's DB + media from src_root to dst_root.""" + src_ch = src_root / channel_id + dst_ch = dst_root / channel_id + if not src_ch.exists(): + return + dst_ch.mkdir(parents=True, exist_ok=True) + + # SQLite DB + db_name = f"{channel_id}.db" + src_db = src_ch / db_name + dst_db = dst_ch / db_name + if src_db.exists() and not dst_db.exists(): + shutil.copy2(src_db, dst_db) + logger.info(" copied %s", db_name) + + # Media directory + src_media = src_ch / "media" + dst_media = dst_ch / "media" + if src_media.exists() and src_media.is_dir(): + dst_media.mkdir(parents=True, exist_ok=True) + for item in src_media.iterdir(): + if item.is_file(): + dst_file = dst_media / item.name + if not dst_file.exists(): + shutil.copy2(item, dst_file) + + # Exports (csv, json) + for ext in (".csv", ".json"): + for f in src_ch.glob(f"*{ext}"): + dst_f = dst_ch / f.name + if not dst_f.exists(): + shutil.copy2(f, dst_f) + + +def migrate_legacy_state(data_dir: Path, session_dir: Path) -> bool: + """ + Run once at startup. Detects legacy data/state.json that has api_id + and migrates to multi-account format, COPYING channel DBs/media from + data// → data/accounts/default// . + + Returns True if migration was performed. + """ + state_path = data_dir / "state.json" + if not state_path.exists(): + return False + + try: + with state_path.open("r", encoding="utf-8") as f: + raw: Dict[str, Any] = json.load(f) + except (json.JSONDecodeError, OSError): + return False + + # Already migrated or fresh install with no legacy api_id + if raw.get("version") == 2 or not raw.get("api_id"): + return False + + logger.info( + "=== Legacy state detected — migrating to multi-account format ===" + ) + logger.info("This may take a while if you have many channels with media.") + + # ── 1. Create per-account state for "default" ────────────────────── + acc_dir = data_dir / "accounts" / "default" + acc_dir.mkdir(parents=True, exist_ok=True) + + acc_state = { + "label": "Default", + "api_id": raw.get("api_id"), + "api_hash": raw.get("api_hash"), + "channels": raw.get("channels", {}), + "channel_names": raw.get("channel_names", {}), + "scrape_media": raw.get("scrape_media", True), + "forwarding_rules": raw.get("forwarding_rules", []), + "continuous_scraping": raw.get( + "continuous_scraping", + { + "enabled": True, + "interval_minutes": 1, + "channels": [], + "run_all_tracked": True, + }, + ), + } + + acc_state_path = acc_dir / "state.json" + with open(acc_state_path, "w", encoding="utf-8") as f: + json.dump(acc_state, f, ensure_ascii=False, indent=2) + f.write("\n") + logger.info(" ✓ data/accounts/default/state.json created") + + # ── 2. Copy channel data ─────────────────────────────────────────── + legacy_channels = list(raw.get("channels", {}).keys()) + if legacy_channels: + logger.info( + " Copying %d channel(s) from data// → data/accounts/default// ...", + len(legacy_channels), + ) + for i, channel_id in enumerate(legacy_channels, 1): + logger.info(" [%d/%d] channel %s", i, len(legacy_channels), channel_id) + _copy_channel_data(data_dir, acc_dir, channel_id) + logger.info(" ✓ Channel data copied") + else: + logger.info(" (no tracked channels to copy)") + + # ── 3. Rewrite global state.json ─────────────────────────────────── + global_state = { + "accounts": ["default"], + "version": 2, + } + with open(state_path, "w", encoding="utf-8") as f: + json.dump(global_state, f, ensure_ascii=False, indent=2) + f.write("\n") + logger.info(" ✓ data/state.json rewritten (accounts list only)") + + # ── 4. Rename legacy session file ───────────────────────────────── + legacy_session = session_dir / "session.session" + if legacy_session.exists(): + new_session = session_dir / "default.session" + if not new_session.exists(): + legacy_session.rename(new_session) + logger.info(" ✓ session/session.session → session/default.session") + + logger.info("=== Migration complete ===") + return True diff --git a/health.py b/health.py index f7a1ba8..2bdaed8 100644 --- a/health.py +++ b/health.py @@ -1,6 +1,6 @@ import sqlite3 from pathlib import Path -from typing import Any, Dict +from typing import Any, Dict, List, Optional from app_state import StateStore @@ -11,6 +11,7 @@ def health_payload( state_store: StateStore, continuous_snapshot: Dict[str, Any], job_queue_size: int, + account_ids: Optional[List[str]] = None, ) -> Dict[str, Any]: checks = { "data_dir": _dir_check(data_dir, writable=True), @@ -20,6 +21,22 @@ def health_payload( "continuous": _continuous_check(continuous_snapshot), "job_queue": {"ok": True, "size": job_queue_size}, } + + # Per-account health + if account_ids: + account_checks: Dict[str, Any] = {} + for acc_id in account_ids: + acc_dir = data_dir / "accounts" / acc_id + session_file = session_dir / f"{acc_id}.session" + account_checks[acc_id] = { + "data_dir": _dir_check(acc_dir), + "session_file": { + "ok": session_file.exists(), + "path": str(session_file), + }, + } + checks["accounts"] = account_checks + ok = all(item.get("ok", False) for item in checks.values()) return {"ok": ok, "status": "ok" if ok else "degraded", "checks": checks} @@ -63,11 +80,19 @@ def _sqlite_check() -> Dict[str, Any]: def _continuous_check(snapshot: Dict[str, Any]) -> Dict[str, Any]: - status = snapshot.get("status", {}) - config = snapshot.get("config", {}) + running_accounts = snapshot.get("running_accounts") + if running_accounts is None: + running_accounts = snapshot return { "ok": True, - "enabled": bool(config.get("enabled")), - "running": bool(status.get("running")), - "last_error": status.get("last_error"), + "account_count": len(running_accounts), + "accounts": { + aid: { + "running": bool((info.get("status") or info).get("running")), + "enabled": bool((info.get("config") or {}).get("enabled", False)), + "last_error": (info.get("status") or info).get("last_error"), + } + for aid, info in running_accounts.items() + if isinstance(info, dict) + }, } diff --git a/main.py b/main.py index e622697..7bc8875 100644 --- a/main.py +++ b/main.py @@ -1,5 +1,6 @@ import logging import sys +from pathlib import Path from webui_server import run_server @@ -11,6 +12,20 @@ def main(): datefmt="%Y-%m-%d %H:%M:%S", stream=sys.stdout, ) + + logger = logging.getLogger(__name__) + + # Run legacy migration before starting the server + from app_state import migrate_legacy_state + + data_dir = Path("data") + session_dir = Path("session") + try: + if migrate_legacy_state(data_dir, session_dir): + logger.info("Legacy migration completed successfully.") + except Exception as exc: + logger.error("Migration failed: %s", exc) + run_server() diff --git a/scraper_jobs.py b/scraper_jobs.py index 2b50bde..216d981 100644 --- a/scraper_jobs.py +++ b/scraper_jobs.py @@ -1,6 +1,6 @@ import asyncio import logging -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional from app_state import StateStore @@ -25,12 +25,21 @@ class ScraperJobService: asyncio.run(self._run_async(job_type, payload)) async def _run_async(self, job_type: str, payload: Dict[str, Any]) -> None: + # Extract account_id from payload, default to None (legacy) + account_id: Optional[str] = payload.get("account_id") ScraperClass = self._import_scraper_class() - scraper = ScraperClass() + scraper = ScraperClass(account_id=account_id) + + if account_id: + from app_state import load_account + acc_state = load_account(self.state_store.path.parent, account_id) + # Load per-account state into scraper state + scraper.state = acc_state + initialized = await scraper.initialize_client(interactive=False) if not initialized: raise RuntimeError( - "Telegram client is not ready. Check credentials, session, and write access to /app/session." + "Telegram client is not ready. Check credentials, session, and write access." ) try: if job_type == "scrape_channel": diff --git a/telegram_scraper_with_forwarding.py b/telegram_scraper_with_forwarding.py index b6bfe51..fe4cc1b 100644 --- a/telegram_scraper_with_forwarding.py +++ b/telegram_scraper_with_forwarding.py @@ -21,7 +21,14 @@ from telethon.tl.types import ( ) from telethon.errors import FloodWaitError, SessionPasswordNeededError import qrcode -from app_state import StateStore +from app_state import ( + StateStore, + account_data_dir, + account_session_path, + get_account_store, + load_account, + save_account, +) warnings.filterwarnings( "ignore", message="Using async sessions support is an experimental feature" @@ -72,14 +79,35 @@ class ForwardingRule: enabled: bool = True +def _ensure_session_wal(session_path: str) -> None: + db_path = session_path if session_path.endswith(".session") else f"{session_path}.session" + if not Path(db_path).exists(): + return + try: + conn = sqlite3.connect(db_path, timeout=1) + conn.execute("PRAGMA journal_mode=WAL;") + conn.execute("PRAGMA synchronous=NORMAL;") + conn.close() + except Exception: + pass + + class OptimizedTelegramScraper: - def __init__(self): - self.DATA_DIR = Path("data") + def __init__(self, account_id: Optional[str] = None): + self.account_id = account_id self.SESSION_DIR = Path("session") - self.DATA_DIR.mkdir(exist_ok=True) self.SESSION_DIR.mkdir(exist_ok=True) + + if account_id: + self.DATA_DIR = Path("data") / "accounts" / account_id + self.state_store = get_account_store(Path("data"), account_id) + else: + self.DATA_DIR = Path("data") + self.state_store = StateStore(self.DATA_DIR / "state.json") + + self.DATA_DIR.mkdir(parents=True, exist_ok=True) self.STATE_FILE = str(self.DATA_DIR / "state.json") - self.state_store = StateStore(self.DATA_DIR / "state.json") + self.state = self.load_state() self.client = None self.continuous_scraping_active = False @@ -1480,8 +1508,10 @@ class OptimizedTelegramScraper: print("Invalid API ID. Must be a number.") return False + session = account_session_path(self.SESSION_DIR, self.account_id or "session") + _ensure_session_wal(session) self.client = TelegramClient( - str(self.SESSION_DIR / "session"), + session, self.state["api_id"], self.state["api_hash"], ) diff --git a/webui/app.js b/webui/app.js index 8af9b4d..1fabdfb 100644 --- a/webui/app.js +++ b/webui/app.js @@ -1,9 +1,17 @@ +/* ── State ───────────────────────────────────────────── */ + const state = { - dashboard: null, - auth: null, - continuous: null, + accounts: [], + activeAccount: null, + dashboards: {}, + continuous: {}, + channels: {}, + jobs: {}, + authStates: {}, }; +/* ── API helper ─────────────────────────────────────── */ + async function api(path, options = {}) { const response = await fetch(path, { headers: { "Content-Type": "application/json" }, @@ -16,6 +24,8 @@ async function api(path, options = {}) { return data; } +/* ── Format helpers ─────────────────────────────────── */ + function formatDate(value) { if (!value) return "-"; return value.replace("T", " ").replace("+00:00", " UTC"); @@ -58,47 +68,166 @@ function confirmAction(message) { return window.confirm(message); } -function renderAuth(auth) { - document.getElementById("auth-status").textContent = auth.status; - document.getElementById("auth-details").textContent = auth.details || ""; +function isAccountAuthorized(auth) { + const status = auth?.auth_status || auth || {}; + return ( + auth?.phase === "authorized" || + auth?.status === "authorized" || + status.status === "ready" || + status.status === "authorized" + ); } -function toggleHidden(id, hidden) { - document.getElementById(id).classList.toggle("hidden", hidden); +/* ── Account tab management ─────────────────────────── */ + +function renderAccountTabs(accounts) { + const container = document.getElementById("account-tabs"); + const template = document.getElementById("account-tab-template"); + container.innerHTML = ""; + + accounts.forEach((acc) => { + const node = template.content.firstElementChild.cloneNode(true); + node.dataset.accountId = acc.id; + node.querySelector(".account-tab-name").textContent = acc.label || acc.id; + const statusEl = node.querySelector(".account-tab-status"); + if (acc.auth) { + const authOk = isAccountAuthorized(acc.auth); + statusEl.textContent = authOk ? "●" : "○"; + statusEl.className = "account-tab-status " + (authOk ? "ok" : ""); + } + if (acc.id === state.activeAccount) { + node.classList.add("active"); + } + node.addEventListener("click", () => switchAccount(acc.id)); + container.appendChild(node); + }); } -function renderAuthPanel(authData) { - state.auth = authData; - renderAuth(authData.auth_status); +function renderAccountPanel(accountId) { + const container = document.getElementById("account-panels"); + const template = document.getElementById("account-panel-template"); - const apiId = authData.saved_credentials?.api_id ?? ""; - document.getElementById("api-id-input").value = apiId || ""; - document.getElementById("api-hash-input").placeholder = authData.saved_credentials?.api_hash_present - ? "API Hash saved" - : "API Hash"; + // Don't duplicate + if (document.getElementById(`panel-${accountId}`)) return; - const showQr = Boolean(authData.qr_image); - toggleHidden("qr-wrap", !showQr); - if (showQr) { - document.getElementById("qr-image").src = authData.qr_image; + const node = template.content.firstElementChild.cloneNode(true); + node.id = `panel-${accountId}`; + node.dataset.accountId = accountId; + + // Add/remove channel form + const addForm = node.querySelector(".add-channel-form"); + addForm.addEventListener("submit", async (event) => { + event.preventDefault(); + const channelId = addForm.querySelector(".add-channel-id").value.trim(); + const name = addForm.querySelector(".add-channel-name").value.trim(); + if (!channelId) return; + await api(`/api/accounts/${accountId}/channels/add`, { + method: "POST", + body: JSON.stringify({ channel_id: channelId, name }), + }); + addForm.reset(); + await refreshAccount(accountId); + }); + + // Continuous form + const contForm = node.querySelector(".continuous-form"); + contForm.addEventListener("submit", async (event) => { + event.preventDefault(); + const enabled = contForm.querySelector(".continuous-enabled").checked; + const intervalMinutes = contForm.querySelector(".continuous-interval").value; + const runAllTracked = contForm.querySelector(".continuous-all").checked; + const channels = Array.from(contForm.querySelectorAll(".continuous-channel-checkbox:checked")).map( + (item) => item.value + ); + if (!runAllTracked && channels.length === 0 && enabled) { + if (!confirmAction("Continuous scraping enabled with no selected channels. Save anyway?")) return; + } + await api(`/api/accounts/${accountId}/continuous`, { + method: "POST", + body: JSON.stringify({ enabled, interval_minutes: intervalMinutes, run_all_tracked: runAllTracked, channels }), + }); + await refreshAccount(accountId); + }); + + // "All tracked" checkbox toggles individual checkboxes + const contAll = contForm.querySelector(".continuous-all"); + contAll.addEventListener("change", () => { + const checkboxes = contForm.querySelectorAll(".continuous-channel-checkbox"); + checkboxes.forEach((cb) => { cb.disabled = contAll.checked; }); + }); + + // Content tabs within account panel + const contentTabs = node.querySelectorAll(".content-tab"); + contentTabs.forEach((tab) => { + tab.addEventListener("click", () => { + const targetId = tab.dataset.tabTarget; + contentTabs.forEach((t) => t.classList.toggle("active", t === tab)); + node.querySelectorAll(".tab-panel").forEach((p) => { + p.classList.toggle("active", p.dataset.panel === targetId); + }); + }); + }); + + // Refresh jobs + node.querySelector(".refresh-jobs-btn").addEventListener("click", () => refreshAccount(accountId)); + + container.appendChild(node); +} + +function switchAccount(accountId) { + if (state.activeAccount === accountId) return; + state.activeAccount = accountId; + localStorage.setItem("activeAccount", accountId); + + // Update tabs + document.querySelectorAll(".account-tab").forEach((tab) => { + tab.classList.toggle("active", tab.dataset.accountId === accountId); + }); + + // Show/hide panels + document.querySelectorAll(".account-panel").forEach((panel) => { + panel.classList.toggle("hidden", panel.dataset.accountId !== accountId); + }); + + // Ensure panel exists + if (!document.getElementById(`panel-${accountId}`)) { + renderAccountPanel(accountId); } - toggleHidden("code-form", authData.phase !== "code_required"); - toggleHidden("password-form", authData.phase !== "password_required"); + // Update sidebar + updateSidebarAccount(); + + // Refresh data + refreshAccount(accountId); + + // Update settings auth section + updateAuthSection(accountId); } -function renderSummary(data) { - document.getElementById("channel-count").textContent = String(data.state.channel_count); - document.getElementById("forwarding-count").textContent = String(data.state.forwarding_rules.length); - const toggle = document.getElementById("scrape-media-toggle"); - toggle.checked = Boolean(data.state.scrape_media); - document.getElementById("scrape-media-label").textContent = toggle.checked ? "ON" : "OFF"; +function updateSidebarAccount() { + const acc = state.accounts.find((a) => a.id === state.activeAccount); + const nameEl = document.getElementById("active-account-name"); + const statusEl = document.getElementById("active-account-status"); + if (acc) { + nameEl.textContent = acc.label || acc.id; + const authOk = isAccountAuthorized(acc.auth); + statusEl.textContent = authOk ? "Authorized session" : "Needs login"; + statusEl.style.color = authOk ? "var(--ok)" : "var(--danger)"; + } else { + nameEl.textContent = "None"; + statusEl.textContent = "Add an account in Settings"; + statusEl.style.color = "var(--dim)"; + } } -function renderChannels(channels) { - const tbody = document.getElementById("channels-table"); - tbody.innerHTML = ""; +/* ── Dashboard rendering ────────────────────────────── */ + +function renderChannels(accountId, channels) { + const panel = document.getElementById(`panel-${accountId}`); + if (!panel) return; + const tbody = panel.querySelector(".channels-table"); const template = document.getElementById("channel-row-template"); + tbody.innerHTML = ""; channels.forEach((channel) => { const node = template.content.firstElementChild.cloneNode(true); @@ -109,50 +238,55 @@ function renderChannels(channels) { node.querySelector(".last-date").textContent = channel.last_date || "-"; node.querySelector(".scrape-btn").addEventListener("click", async () => { - await api("/api/jobs/scrape", { + await api(`/api/accounts/${accountId}/jobs/scrape`, { method: "POST", body: JSON.stringify({ channel_id: channel.channel_id }), }); - await refreshDashboard(); + await refreshAccount(accountId); }); node.querySelector(".export-btn").addEventListener("click", async () => { - await api("/api/jobs/export-channel", { + await api(`/api/accounts/${accountId}/jobs/export-channel`, { method: "POST", body: JSON.stringify({ channel_id: channel.channel_id }), }); - await refreshDashboard(); + await refreshAccount(accountId); }); node.querySelector(".export-view-btn").addEventListener("click", () => { - window.location.href = `/viewer?channel=${encodeURIComponent(channel.channel_id)}`; + window.location.href = `/viewer?channel=${encodeURIComponent(channel.channel_id)}&account=${encodeURIComponent(accountId)}`; }); node.querySelector(".media-btn").addEventListener("click", async () => { - await api("/api/jobs/rescrape-media", { + await api(`/api/accounts/${accountId}/jobs/rescrape-media`, { method: "POST", body: JSON.stringify({ channel_id: channel.channel_id }), }); - await refreshDashboard(); + await refreshAccount(accountId); }); node.querySelector(".remove-btn").addEventListener("click", async () => { if (!confirmAction(`Remove ${channel.name} from tracked channels?`)) return; - await api("/api/channels/remove", { + await api(`/api/accounts/${accountId}/channels/remove`, { method: "POST", body: JSON.stringify({ channel_id: channel.channel_id }), }); - await refreshDashboard(); + await refreshAccount(accountId); }); tbody.appendChild(node); }); + + // Update channel count stat + panel.querySelector(".channel-count").textContent = String(channels.length); } -function renderJobs(jobs) { - const root = document.getElementById("jobs-list"); - root.innerHTML = ""; +function renderJobs(accountId, jobs) { + const panel = document.getElementById(`panel-${accountId}`); + if (!panel) return; + const root = panel.querySelector(".jobs-list"); const template = document.getElementById("job-template"); + root.innerHTML = ""; if (!jobs.length) { root.innerHTML = '
No jobs yet.
'; @@ -171,18 +305,25 @@ function renderJobs(jobs) { }); } -function renderContinuousChannelPicker(channels, config) { - const root = document.getElementById("continuous-channel-list"); +function renderContinuous(accountId, data) { + const panel = document.getElementById(`panel-${accountId}`); + if (!panel) return; + const config = data.config || {}; + const status = data.status || {}; + + const contForm = panel.querySelector(".continuous-form"); + contForm.querySelector(".continuous-enabled").checked = Boolean(config.enabled); + contForm.querySelector(".continuous-interval").value = config.interval_minutes || 1; + contForm.querySelector(".continuous-all").checked = Boolean(config.run_all_tracked); + + // Channel picker + const chList = panel.querySelector(".continuous-channel-list"); + const channels = state.channels[accountId] || []; const runAll = Boolean(config.run_all_tracked); const selected = new Set(config.channels || []); - root.innerHTML = ""; + chList.innerHTML = ""; - if (!channels.length) { - root.innerHTML = '
No tracked channels.
'; - return; - } - - channels.forEach((channel) => { + channels.forEach((ch) => { const label = document.createElement("label"); label.className = "checkbox-row"; label.classList.toggle("disabled", runAll); @@ -190,15 +331,59 @@ function renderContinuousChannelPicker(channels, config) { const checkbox = document.createElement("input"); checkbox.type = "checkbox"; checkbox.className = "continuous-channel-checkbox"; - checkbox.value = channel.channel_id; - checkbox.checked = runAll || selected.has(channel.channel_id); + checkbox.value = ch.channel_id; + checkbox.checked = runAll || selected.has(ch.channel_id); checkbox.disabled = runAll; const text = document.createElement("span"); - text.textContent = `${channel.name} (${channel.channel_id})`; + text.textContent = `${ch.name} (${ch.channel_id})`; label.append(checkbox, text); - root.appendChild(label); + chList.appendChild(label); + }); + + // Meta + panel.querySelector(".continuous-status").textContent = status.running ? "running" : "stopped"; + panel.querySelector(".continuous-last-iteration").textContent = status.last_iteration_at + ? `${relativeTime(status.last_iteration_at)} (${displayTime(status.last_iteration_at)})` + : "-"; + panel.querySelector(".continuous-last-error").textContent = status.last_error || "-"; + + // Logs + const logsRoot = panel.querySelector(".continuous-logs"); + const entries = (status.log_entries || status.logs || []).map(normalizeLogEntry); + logsRoot.innerHTML = ""; + + if (!entries.length) { + logsRoot.innerHTML = '
-INFONo logs yet.
'; + return; + } + + entries.slice().reverse().forEach((entry) => { + const row = document.createElement("div"); + row.className = `log-line ${entry.level}`; + const time = entry.timestamp ? displayTime(entry.timestamp).split(", ").pop() : entry.legacyTime || "-"; + const absolute = entry.timestamp ? displayTime(entry.timestamp) : entry.legacyTime || "-"; + row.title = absolute; + + const timeNode = document.createElement("span"); + timeNode.className = "log-time"; + timeNode.textContent = time; + + const ageNode = document.createElement("span"); + ageNode.className = "log-age"; + ageNode.textContent = entry.timestamp ? relativeTime(entry.timestamp) : "-"; + + const levelNode = document.createElement("span"); + levelNode.className = "log-level"; + levelNode.textContent = entry.level.toUpperCase(); + + const messageNode = document.createElement("span"); + messageNode.className = "log-message"; + messageNode.textContent = entry.message; + + row.append(timeNode, ageNode, levelNode, messageNode); + logsRoot.appendChild(row); }); } @@ -230,246 +415,361 @@ function normalizeLogEntry(entry) { }; } -function renderContinuousLogs(status) { - const root = document.getElementById("continuous-logs"); - const entries = (status.log_entries || status.logs || []).map(normalizeLogEntry); - root.innerHTML = ""; +function renderSummary(accountId, data) { + const panel = document.getElementById(`panel-${accountId}`); + if (!panel) return; + const d = data.dashboard || data.state || {}; + panel.querySelector(".forwarding-count").textContent = String((d.forwarding_rules || []).length); + const toggle = panel.querySelector(".scrape-media-label"); + toggle.textContent = d.scrape_media ? "ON" : "OFF"; +} - if (!entries.length) { - const empty = document.createElement("div"); - empty.className = "log-line"; - empty.innerHTML = '--INFONo logs yet.'; - root.appendChild(empty); +/* ── Account data loading ───────────────────────────── */ + +async function refreshAccount(accountId) { + try { + const [dashboard, authData, continuousData] = await Promise.all([ + api(`/api/accounts/${accountId}`), + api(`/api/accounts/${accountId}/auth`).catch(() => ({})), + api(`/api/accounts/${accountId}/continuous`).catch(() => ({})), + ]); + + state.dashboards[accountId] = dashboard; + state.continuous[accountId] = continuousData; + state.authStates[accountId] = authData; + state.channels[accountId] = dashboard.channels || []; + + renderSummary(accountId, dashboard); + renderChannels(accountId, dashboard.channels || []); + renderJobs(accountId, dashboard.jobs || []); + renderContinuous(accountId, continuousData); + + // Update tab status indicator + const tab = document.querySelector(`.account-tab[data-account-id="${accountId}"]`); + if (tab) { + const statusEl = tab.querySelector(".account-tab-status"); + const authOk = isAccountAuthorized(authData); + statusEl.textContent = authOk ? "●" : "○"; + statusEl.className = "account-tab-status " + (authOk ? "ok" : ""); + } + } catch (err) { + console.error(`Failed to refresh account ${accountId}:`, err); + } +} + +async function loadAccounts() { + let accounts = []; + try { + const resp = await api("/api/accounts"); + accounts = resp.accounts || []; + } catch { + // Legacy fallback — check /api/auth + try { + const legacy = await api("/api/auth"); + if (legacy.saved_credentials?.api_id) { + accounts = [{ id: "default", label: "Default", auth: legacy.auth_status || legacy }]; + } + } catch { + // No accounts at all + } + } + + state.accounts = accounts; + + if (accounts.length === 0) { + document.getElementById("no-accounts-msg").classList.remove("hidden"); return; } - entries.slice().reverse().forEach((entry) => { - const row = document.createElement("div"); - row.className = `log-line ${entry.level}`; - const time = entry.timestamp ? displayTime(entry.timestamp).split(", ").pop() : entry.legacyTime || "-"; - const absolute = entry.timestamp ? displayTime(entry.timestamp) : entry.legacyTime || "-"; - row.title = absolute; + document.getElementById("no-accounts-msg").classList.add("hidden"); - const timeNode = document.createElement("span"); - timeNode.className = "log-time"; - timeNode.textContent = time; + // Render tabs + renderAccountTabs(accounts); - const ageNode = document.createElement("span"); - ageNode.className = "log-age"; - ageNode.textContent = entry.timestamp ? relativeTime(entry.timestamp) : "-"; + // Render panels for each account + accounts.forEach((acc) => renderAccountPanel(acc.id)); - const levelNode = document.createElement("span"); - levelNode.className = "log-level"; - levelNode.textContent = entry.level.toUpperCase(); + // Determine active account (persisted across reloads) + const savedId = localStorage.getItem("activeAccount"); + const activeId = (savedId && accounts.some(a => a.id === savedId)) ? savedId : (state.activeAccount || accounts[0].id); + switchAccount(activeId); - const messageNode = document.createElement("span"); - messageNode.className = "log-message"; - messageNode.textContent = entry.message; - - row.append(timeNode, ageNode, levelNode, messageNode); - root.appendChild(row); - }); + // Update settings + renderSettingsAccounts(); } -function renderContinuous(data) { - state.continuous = data; - const config = data.config || {}; - const status = data.status || {}; +/* ── Settings: Accounts list ────────────────────────── */ - document.getElementById("continuous-enabled").checked = Boolean(config.enabled); - document.getElementById("continuous-interval").value = config.interval_minutes || 1; - document.getElementById("continuous-all").checked = Boolean(config.run_all_tracked); - renderContinuousChannelPicker(state.dashboard?.channels || [], config); +function renderSettingsAccounts() { + const container = document.getElementById("accounts-list"); + const template = document.getElementById("account-list-item-template"); + container.innerHTML = ""; - document.getElementById("continuous-status").textContent = status.running ? "running" : "stopped"; - document.getElementById("continuous-last-iteration").textContent = status.last_iteration_at - ? `${relativeTime(status.last_iteration_at)} (${displayTime(status.last_iteration_at)})` - : "-"; - document.getElementById("continuous-last-error").textContent = status.last_error || "-"; - renderContinuousLogs(status); -} + state.accounts.forEach((acc) => { + const node = template.content.firstElementChild.cloneNode(true); + node.querySelector(".account-list-label").textContent = acc.label || acc.id; + node.querySelector(".account-list-id").textContent = acc.id; -function setupTabs() { - const tabs = document.querySelectorAll(".content-tab"); - const panels = document.querySelectorAll(".tab-panel"); + const authStatus = node.querySelector(".account-list-auth-status"); + const authOk = isAccountAuthorized(acc.auth); + authStatus.textContent = authOk ? "Authorized" : "Needs auth"; + authStatus.style.color = authOk ? "var(--ok)" : "var(--danger)"; + authStatus.style.fontSize = "0.78rem"; - tabs.forEach((tab) => { - tab.addEventListener("click", () => { - const targetId = tab.dataset.tabTarget; - tabs.forEach((item) => item.classList.toggle("active", item === tab)); - panels.forEach((panel) => panel.classList.toggle("active", panel.id === targetId)); + node.querySelector(".account-select-btn").addEventListener("click", () => { + switchAccount(acc.id); + document.getElementById("settings-dialog").close(); }); + + node.querySelector(".account-remove-btn").addEventListener("click", async () => { + if (!confirmAction(`Remove account "${acc.label || acc.id}"? All its data will be deleted.`)) return; + try { + await api(`/api/accounts/${acc.id}`, { method: "DELETE" }); + await loadAccounts(); + if (state.activeAccount === acc.id) { + state.activeAccount = null; + } + } catch (err) { + alert(`Failed to remove account: ${err.message}`); + } + }); + + container.appendChild(node); }); } -async function refreshDashboard() { - const data = await api("/api/dashboard"); - const authData = await api("/api/auth"); - const continuousData = await api("/api/continuous"); - state.dashboard = data; - renderAuthPanel(authData); - renderSummary(data); - renderChannels(data.channels); - renderJobs(data.jobs); - renderContinuous(continuousData); +/* ── Auth (operates on active account) ───────────────── */ + +function updateAuthSection(accountId) { + const label = document.getElementById("auth-account-label"); + const acc = state.accounts.find((a) => a.id === accountId); + label.textContent = acc ? (acc.label || acc.id) : "-"; } +async function saveCredentials(accountId) { + const apiId = document.getElementById("api-id-input").value.trim(); + const apiHash = document.getElementById("api-hash-input").value.trim(); + if (!apiId || !apiHash) return; + await api(`/api/accounts/${accountId}/auth/credentials`, { + method: "POST", + body: JSON.stringify({ api_id: apiId, api_hash: apiHash }), + }); + await loadAccounts(); +} + +async function startQrLogin(accountId) { + const data = await api(`/api/accounts/${accountId}/auth/qr/start`, { + method: "POST", + body: JSON.stringify({}), + }); + if (data.qr_image) { + document.getElementById("qr-image").src = data.qr_image; + document.getElementById("qr-wrap").classList.remove("hidden"); + } + await loadAccounts(); +} + +async function requestPhoneCode(accountId) { + const phone = document.getElementById("phone-input").value.trim(); + if (!phone) return; + await api(`/api/accounts/${accountId}/auth/phone/request`, { + method: "POST", + body: JSON.stringify({ phone }), + }); + await loadAccounts(); +} + +async function submitPhoneCode(accountId) { + const code = document.getElementById("code-input").value.trim(); + if (!code) return; + await api(`/api/accounts/${accountId}/auth/phone/submit`, { + method: "POST", + body: JSON.stringify({ code }), + }); + document.getElementById("code-input").value = ""; + await loadAccounts(); +} + +async function submitPassword(accountId) { + const password = document.getElementById("password-input").value.trim(); + if (!password) return; + await api(`/api/accounts/${accountId}/auth/password`, { + method: "POST", + body: JSON.stringify({ password }), + }); + document.getElementById("password-input").value = ""; + await loadAccounts(); +} + +/* ── Scrape / Export all (for active account) ───────── */ + +async function scrapeAll() { + if (!state.activeAccount) return; + if (!confirmAction("Queue scraping for all tracked channels?")) return; + await api(`/api/accounts/${state.activeAccount}/jobs/scrape`, { + method: "POST", + body: JSON.stringify({}), + }); + await refreshAccount(state.activeAccount); +} + +async function exportAll() { + if (!state.activeAccount) return; + if (!confirmAction("Queue export for all tracked channels?")) return; + await api(`/api/accounts/${state.activeAccount}/jobs/export`, { + method: "POST", + body: JSON.stringify({}), + }); + await refreshAccount(state.activeAccount); +} + +async function refreshDialogs() { + if (!state.activeAccount) return; + await api(`/api/accounts/${state.activeAccount}/jobs/refresh-dialogs`, { + method: "POST", + body: JSON.stringify({}), + }); + await refreshAccount(state.activeAccount); +} + +async function toggleMedia(value) { + if (!state.activeAccount) return; + await api(`/api/accounts/${state.activeAccount}/settings/media`, { + method: "POST", + body: JSON.stringify({ value }), + }); + await refreshAccount(state.activeAccount); +} + +/* ── Main ───────────────────────────────────────────── */ + async function main() { - setupTabs(); + // ── Sidebar action buttons ── + document.getElementById("scrape-all-btn").addEventListener("click", async () => { + try { await scrapeAll(); } catch (err) { alert("Scrape error: " + err.message); } + }); + document.getElementById("export-all-btn").addEventListener("click", async () => { + try { await exportAll(); } catch (err) { alert("Export error: " + err.message); } + }); + document.getElementById("refresh-dialogs-btn").addEventListener("click", async () => { + try { await refreshDialogs(); } catch (err) { alert("Dialog refresh error: " + err.message); } + }); + // ── Settings dialog ── const settingsDialog = document.getElementById("settings-dialog"); document.getElementById("open-settings-btn").addEventListener("click", () => { + // Update auth section for active account + if (state.activeAccount) { + updateAuthSection(state.activeAccount); + } + renderSettingsAccounts(); settingsDialog.showModal(); }); document.getElementById("close-settings-btn").addEventListener("click", () => { settingsDialog.close(); }); - document.getElementById("scrape-all-btn").addEventListener("click", async (event) => { - if (!confirmAction("Queue scraping for all tracked channels?")) return; - setBusy(event.currentTarget, true); - try { - await api("/api/jobs/scrape", { method: "POST", body: JSON.stringify({}) }); - await refreshDashboard(); - } finally { - setBusy(event.currentTarget, false); - } - }); - - document.getElementById("export-all-btn").addEventListener("click", async (event) => { - if (!confirmAction("Queue export for all tracked channels?")) return; - setBusy(event.currentTarget, true); - try { - await api("/api/jobs/export", { method: "POST", body: JSON.stringify({}) }); - await refreshDashboard(); - } finally { - setBusy(event.currentTarget, false); - } - }); - - document.getElementById("refresh-dialogs-btn").addEventListener("click", async (event) => { - setBusy(event.currentTarget, true); - try { - await api("/api/jobs/refresh-dialogs", { method: "POST", body: JSON.stringify({}) }); - await refreshDashboard(); - } finally { - setBusy(event.currentTarget, false); - } - }); - - document.getElementById("refresh-jobs-btn").addEventListener("click", refreshDashboard); - - document.getElementById("scrape-media-toggle").addEventListener("change", async (event) => { - const checked = event.currentTarget.checked; - document.getElementById("scrape-media-label").textContent = checked ? "ON" : "OFF"; - await api("/api/settings/media", { - method: "POST", - body: JSON.stringify({ value: checked }), - }); - await refreshDashboard(); - }); - - document.getElementById("add-channel-form").addEventListener("submit", async (event) => { + // ── Add account form ── + document.getElementById("add-account-form").addEventListener("submit", async (event) => { event.preventDefault(); - const channelId = document.getElementById("add-channel-id").value.trim(); - const name = document.getElementById("add-channel-name").value.trim(); - if (!channelId) return; - await api("/api/channels/add", { - method: "POST", - body: JSON.stringify({ channel_id: channelId, name }), - }); - event.currentTarget.reset(); - await refreshDashboard(); + const accountId = document.getElementById("add-account-id").value.trim(); + const label = document.getElementById("add-account-label").value.trim(); + const apiId = document.getElementById("add-account-api-id").value.trim(); + const apiHash = document.getElementById("add-account-api-hash").value.trim(); + if (!accountId) return; + + try { + await api("/api/accounts", { + method: "POST", + body: JSON.stringify({ + account_id: accountId, + label: label || accountId, + api_id: apiId ? parseInt(apiId) : undefined, + api_hash: apiHash || undefined, + }), + }); + event.target.reset(); + await loadAccounts(); + if (!state.activeAccount) { + switchAccount(accountId); + } + } catch (err) { + alert("Failed to add account: " + err.message); + } }); + // ── Credentials form ── document.getElementById("credentials-form").addEventListener("submit", async (event) => { event.preventDefault(); - const apiId = document.getElementById("api-id-input").value.trim(); - const apiHash = document.getElementById("api-hash-input").value.trim(); - await api("/api/auth/credentials", { - method: "POST", - body: JSON.stringify({ api_id: apiId, api_hash: apiHash }), - }); - await refreshDashboard(); + if (!state.activeAccount) return; + try { + await saveCredentials(state.activeAccount); + } catch (err) { + alert("Failed to save credentials: " + err.message); + } }); + // ── QR login ── document.getElementById("start-qr-btn").addEventListener("click", async (event) => { setBusy(event.currentTarget, true); try { - await api("/api/auth/qr/start", { - method: "POST", - body: JSON.stringify({}), - }); - await refreshDashboard(); + if (!state.activeAccount) return; + await startQrLogin(state.activeAccount); + } catch (err) { + alert("QR login error: " + err.message); } finally { setBusy(event.currentTarget, false); } }); + // ── Phone ── document.getElementById("phone-form").addEventListener("submit", async (event) => { event.preventDefault(); - const phone = document.getElementById("phone-input").value.trim(); - await api("/api/auth/phone/request", { - method: "POST", - body: JSON.stringify({ phone }), - }); - await refreshDashboard(); + if (!state.activeAccount) return; + try { + await requestPhoneCode(state.activeAccount); + } catch (err) { + alert("Phone code request error: " + err.message); + } }); + // ── Code ── document.getElementById("code-form").addEventListener("submit", async (event) => { event.preventDefault(); - const code = document.getElementById("code-input").value.trim(); - await api("/api/auth/phone/submit", { - method: "POST", - body: JSON.stringify({ code }), - }); - event.currentTarget.reset(); - await refreshDashboard(); + if (!state.activeAccount) return; + try { + await submitPhoneCode(state.activeAccount); + } catch (err) { + alert("Code submit error: " + err.message); + } }); + // ── Password ── document.getElementById("password-form").addEventListener("submit", async (event) => { event.preventDefault(); - const password = document.getElementById("password-input").value.trim(); - await api("/api/auth/password", { - method: "POST", - body: JSON.stringify({ password }), - }); - event.currentTarget.reset(); - await refreshDashboard(); - }); - - document.getElementById("continuous-form").addEventListener("submit", async (event) => { - event.preventDefault(); - const enabled = document.getElementById("continuous-enabled").checked; - const intervalMinutes = document.getElementById("continuous-interval").value; - const runAllTracked = document.getElementById("continuous-all").checked; - const channels = Array.from(document.querySelectorAll(".continuous-channel-checkbox:checked")).map( - (item) => item.value, - ); - if (!runAllTracked && channels.length === 0 && enabled) { - if (!confirmAction("Continuous scraping is enabled with no selected channels. Save anyway?")) return; + if (!state.activeAccount) return; + try { + await submitPassword(state.activeAccount); + } catch (err) { + alert("Password error: " + err.message); } - await api("/api/continuous", { - method: "POST", - body: JSON.stringify({ - enabled, - interval_minutes: intervalMinutes, - run_all_tracked: runAllTracked, - channels, - }), - }); - await refreshDashboard(); }); - document.getElementById("continuous-all").addEventListener("change", () => { - renderContinuousChannelPicker(state.dashboard?.channels || [], { - ...(state.continuous?.config || {}), - run_all_tracked: document.getElementById("continuous-all").checked, - }); + // ── Media toggle ── + document.getElementById("scrape-media-toggle").addEventListener("change", async (event) => { + const checked = event.currentTarget.checked; + try { await toggleMedia(checked); } catch (err) { alert("Media toggle error: " + err.message); } }); - await refreshDashboard(); - window.setInterval(refreshDashboard, 5000); + // ── Load accounts ── + await loadAccounts(); + + // ── Periodic refresh ── + window.setInterval(() => { + if (state.activeAccount) { + refreshAccount(state.activeAccount); + } + }, 5000); } main().catch((error) => { diff --git a/webui/index.html b/webui/index.html index faea273..b979a7c 100644 --- a/webui/index.html +++ b/webui/index.html @@ -4,7 +4,7 @@ Telegram Scraper Control Panel - +
@@ -22,10 +22,10 @@ Health -
-
Telegram Status
-
Checking...
-

+
+
Active Account
+
Loading...
+

@@ -38,127 +38,54 @@
-
- - + + + + +
+
-
-
-
-
Channels
-
0
+ + - -
-
-
-

Tracked Channels

-

Stored in data/state.json. Message stats come from SQLite.

-
-
- - - -
-
- -
- - - - - - - - - - - -
ChannelMessagesMediaLast messageActions
-
-
- -
-
-
-

Jobs

-

Queued work runs one job at a time to keep the Telegram session stable.

-
- -
-
-
-
- -
-
-
-
-

Continuous Scraping

-

Runs on a timer and keeps a compact, highlighted run log.

-
-
- -
- - - -
-
Continuous channel set
-
-
- -
- -
-
Status: -
-
Last run: -
-
Last error: -
-
- -
-
+
+
Settings
-

Telegram & Scraping

+

Global & Accounts

+
-
Telegram Login
+
Accounts
+
+ + +
+ + +
+
Account Auth: -
@@ -190,6 +117,7 @@
+
Scraping
 + + + + + + + + + + - + + + + diff --git a/webui/style.css b/webui/style.css index 3898a00..2e40e14 100644 --- a/webui/style.css +++ b/webui/style.css @@ -1,17 +1,24 @@ :root { color-scheme: dark; - --bg-color: #050505; - --panel: #0c0c0c; - --panel-alt: #080808; - --text-color: #e0e0e0; - --accent: #ffffff; - --dim: #6f6f6f; - --line: #2a2a2a; - --danger: #ff6b6b; - --ok: #7ee787; - --warn: #f0c674; - --info: #8ab4f8; - --font-mono: "Courier New", Courier, monospace; + --bg-color: #000000; + --panel: #0b0b0b; + --panel-solid: #101010; + --panel-alt: #050505; + --text-color: #f2f3f5; + --accent: #1fb9aa; + --accent-2: #58a6ff; + --bubble-own: #12342f; + --bubble-other: #171717; + --dim: #9aa0a6; + --line: #242424; + --danger: #ff7373; + --ok: #65e6a4; + --warn: #f6c969; + --info: #8ec7ff; + --radius: 18px; + --shadow: none; + --font-ui: "Segoe UI", "Helvetica Neue", Arial, sans-serif; + --font-mono: "IBM Plex Mono", "SFMono-Regular", Consolas, monospace; } * { @@ -20,11 +27,11 @@ body { margin: 0; - background: var(--bg-color); + background: #000; color: var(--text-color); - font-family: var(--font-mono); - font-size: 16px; - line-height: 1.55; + font-family: var(--font-ui); + font-size: 15px; + line-height: 1.5; } a { @@ -35,8 +42,9 @@ a { a:hover, button:hover { - background: var(--text-color); - color: var(--bg-color); + border-color: #333; + background: #151515; + color: var(--text-color); } code, @@ -50,19 +58,19 @@ button { height: 100vh; overflow: hidden; display: grid; - grid-template-columns: 300px minmax(0, 1fr); + grid-template-columns: 308px minmax(0, 1fr); } .sidebar { - background: var(--panel-alt); - border-right: 1px dashed var(--line); - padding: 28px 24px; + background: #050505; + border-right: 1px solid var(--line); + padding: 24px 20px; } .content { overflow-y: auto; - padding: 28px; - max-width: 1320px; + padding: 24px; + max-width: 1440px; width: 100%; } @@ -70,21 +78,16 @@ button { .dialog-header h2 { margin: 0; color: var(--accent); - font-size: 1.45rem; - text-transform: uppercase; - letter-spacing: 0; + font-size: 1.5rem; + letter-spacing: -0.04em; } .eyebrow, .section-title { color: var(--dim); - font-size: 0.78rem; + font-size: 0.75rem; text-transform: uppercase; -} - -.eyebrow::before, -.section-title::before { - content: "./"; + letter-spacing: 0.08em; } .muted { @@ -112,8 +115,9 @@ input, .content-tab, .viewer-channel-item { border: 1px solid var(--line); - background: var(--panel); + background: #101010; color: var(--text-color); + border-radius: 999px; } .nav-link, @@ -125,21 +129,23 @@ input, min-height: 42px; padding: 9px 13px; cursor: pointer; + transition: background-color 0.18s ease, border-color 0.18s ease, transform 0.18s ease; } .nav-link.active, .button.primary, .content-tab.active { - border-color: var(--accent); - color: var(--accent); - box-shadow: inset 3px 0 0 var(--accent); + border-color: #0f6f67; + color: #ffffff; + background: #0f6f67; + box-shadow: none; } .nav-link.active:hover, .button.primary:hover, .content-tab.active:hover { - background: var(--panel); - color: var(--accent); + background: #128277; + color: #ffffff; } .button.danger { @@ -164,7 +170,9 @@ input, .message-card, .settings-dialog { background: var(--panel); - border: 1px dashed var(--line); + border: 1px solid var(--line); + border-radius: var(--radius); + box-shadow: var(--shadow); } .status-card, @@ -185,10 +193,10 @@ input, .status-badge { display: inline-flex; - padding: 4px 8px; + padding: 6px 10px; border: 1px solid var(--line); + border-radius: 999px; color: var(--ok); - text-transform: uppercase; } .content-tabs { @@ -228,8 +236,9 @@ input, .stat-value { margin-top: 8px; - font-size: 2rem; + font-size: 2.25rem; font-weight: 700; + letter-spacing: -0.05em; } .stat-compact { @@ -250,8 +259,9 @@ input, input { min-width: 160px; - padding: 9px 11px; + padding: 10px 13px; outline: 0; + border-radius: 14px; } input:focus { @@ -286,7 +296,8 @@ input:focus { position: absolute; inset: 0; border: 1px solid var(--line); - background: #101010; + background: rgba(255, 255, 255, 0.06); + border-radius: 999px; } .switch-slider::before { @@ -297,6 +308,7 @@ input:focus { width: 16px; height: 16px; background: var(--dim); + border-radius: 50%; transition: transform 0.16s ease, background-color 0.16s ease; } @@ -323,11 +335,11 @@ td { padding: 12px 10px; text-align: left; vertical-align: top; - border-bottom: 1px dashed var(--line); + border-bottom: 1px solid var(--line); } tbody tr:hover { - background: #101010; + background: #111; } .action-row { @@ -375,7 +387,9 @@ tbody tr:hover { gap: 10px; min-height: 38px; padding: 8px; - border: 1px dashed var(--line); + border: 1px solid var(--line); + border-radius: 14px; + background: #0f0f0f; } .checkbox-row input { @@ -391,7 +405,8 @@ tbody tr:hover { max-height: 420px; overflow: auto; border: 1px solid var(--line); - background: #030303; + background: rgba(0, 0, 0, 0.2); + border-radius: 16px; } .log-line { @@ -404,7 +419,7 @@ tbody tr:hover { } .log-line:hover { - background: #101010; + background: #111; } .log-time, @@ -455,13 +470,14 @@ tbody tr:hover { .settings-section { margin-top: 18px; padding-top: 16px; - border-top: 1px dashed var(--line); + border-top: 1px solid var(--line); } .qr-wrap { margin-top: 14px; padding: 12px; - border: 1px dashed var(--line); + border: 1px solid var(--line); + border-radius: 16px; } .qr-wrap img { @@ -481,18 +497,18 @@ tbody tr:hover { .viewer-shell { display: grid; - grid-template-columns: 300px minmax(0, 1fr); + grid-template-columns: 360px minmax(0, 1fr); height: 100vh; overflow: hidden; } .viewer-sidebar { - background: var(--panel-alt); - border-right: 1px dashed var(--line); - padding: 28px 24px; + background: #050505; + border-right: 1px solid var(--line); + padding: 14px 10px; display: flex; flex-direction: column; - gap: 20px; + gap: 12px; overflow-y: auto; } @@ -505,48 +521,104 @@ tbody tr:hover { align-items: flex-start; justify-content: space-between; gap: 16px; + padding: 10px 6px 8px; } .viewer-sidebar-head h1 { margin: 0; color: var(--accent); font-size: 1.45rem; - text-transform: uppercase; - letter-spacing: 0; + letter-spacing: -0.04em; } .viewer-channel-list { display: grid; - gap: 10px; + gap: 2px; } .viewer-channel-item { display: grid; - gap: 4px; + grid-template-columns: 52px minmax(0, 1fr); + align-items: center; + gap: 10px; width: 100%; - padding: 12px; + min-height: 68px; + padding: 8px 10px; text-align: left; cursor: pointer; - border: 1px solid var(--line); - background: var(--panel); + border: 1px solid transparent; + background: transparent; color: var(--text-color); + border-radius: 12px; } .viewer-channel-item.active { - border-color: var(--accent); - box-shadow: inset 3px 0 0 var(--accent); + border-color: transparent; + background: #12342f; + box-shadow: none; +} + +.viewer-channel-item:hover { + background: #111; +} + +.viewer-channel-avatar { + display: inline-flex; + align-items: center; + justify-content: center; + width: 48px; + height: 48px; + border-radius: 50%; + background: #5f6b70; + color: white; + font-weight: 700; + letter-spacing: -0.03em; +} + +.viewer-channel-copy { + display: grid; + gap: 3px; + min-width: 0; +} + +.viewer-channel-row { + display: grid; + grid-template-columns: minmax(0, 1fr) auto; + align-items: center; + gap: 8px; + min-width: 0; } .viewer-channel-name { - color: var(--accent); + color: var(--text-color); font-weight: 700; + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; } -.viewer-channel-meta { +.viewer-channel-time, +.viewer-channel-preview { color: var(--dim); font-size: 0.82rem; } +.viewer-channel-preview { + overflow: hidden; + text-overflow: ellipsis; + white-space: nowrap; +} + +.viewer-channel-count { + min-width: 22px; + padding: 2px 7px; + border-radius: 999px; + background: #12342f; + color: var(--accent); + font-size: 0.76rem; + text-align: center; +} + /* ---------- Main panel ---------- */ .viewer-main { @@ -554,27 +626,31 @@ tbody tr:hover { flex-direction: column; height: 100vh; overflow: hidden; + background: + radial-gradient(circle at 20px 20px, rgba(255, 255, 255, 0.025) 1px, transparent 1.5px), + radial-gradient(circle at 58px 54px, rgba(255, 255, 255, 0.018) 1px, transparent 1.5px), + #000; + background-size: auto, 84px 84px, 84px 84px, auto; } .viewer-header { flex-shrink: 0; - padding: 20px 28px; - border-bottom: 1px dashed var(--line); - background: var(--panel-alt); + padding: 16px 24px; + border-bottom: 1px solid var(--line); + background: #050505; } .viewer-header h2 { margin: 0; color: var(--accent); font-size: 1.45rem; - text-transform: uppercase; - letter-spacing: 0; + letter-spacing: -0.04em; } .messages-list { flex: 1; overflow-y: auto; - padding: 12px 28px 20px; + padding: 18px 34px 22px; display: flex; flex-direction: column; } @@ -594,9 +670,7 @@ tbody tr:hover { gap: 12px; margin: 12px 0; font-size: 0.78rem; - color: var(--dim); - text-transform: uppercase; - letter-spacing: 0.03em; + color: #d6e6e8; flex-shrink: 0; } @@ -605,7 +679,14 @@ tbody tr:hover { content: ""; flex: 1; height: 1px; - background: var(--line); + background: #242424; +} + +.chat-date-sep { + align-self: center; + padding: 5px 12px; + border-radius: 999px; + background: #111; } .chat-date-sep:empty { @@ -627,15 +708,16 @@ tbody tr:hover { /* ---------- Chat bubble ---------- */ .chat-bubble { - max-width: 75%; - padding: 7px 11px; - background: #1c1c1c; - border: 1px solid var(--line); - border-radius: 1px; - margin-bottom: 4px; + max-width: min(680px, 74%); + padding: 8px 12px 7px; + background: var(--bubble-other); + border: 1px solid #242424; + border-radius: 18px 18px 18px 6px; + margin-bottom: 6px; overflow-wrap: anywhere; word-break: break-word; position: relative; + box-shadow: none; } .chat-bubble::before { @@ -647,19 +729,20 @@ tbody tr:hover { .chat-bubble-wrap:not(.chat-bubble-wrap--own) .chat-bubble::before { left: -12px; - border-right-color: #1c1c1c; + border-right-color: var(--bubble-other); border-left: 0; } .chat-bubble-wrap--own .chat-bubble::before { right: -12px; - border-left-color: #1e3a5f; + border-left-color: var(--bubble-own); border-right: 0; } .chat-bubble-wrap--own .chat-bubble { - background: #1e3a5f; - border-color: #2b5278; + background: var(--bubble-own); + border-color: #1f5b52; + border-radius: 18px 18px 6px 18px; } .chat-bubble--highlight { @@ -675,7 +758,7 @@ tbody tr:hover { .chat-sender { font-size: 0.82rem; - color: var(--info); + color: var(--accent); font-weight: 700; margin-bottom: 2px; } @@ -690,8 +773,10 @@ tbody tr:hover { display: flex; gap: 6px; margin-bottom: 4px; - padding: 4px 0; + padding: 6px 8px; cursor: pointer; + border-radius: 10px; + background: rgba(0, 0, 0, 0.18); } .chat-reply.hidden { @@ -702,7 +787,7 @@ tbody tr:hover { flex-shrink: 0; width: 2px; background: var(--accent); - border-radius: 1px; + border-radius: 999px; opacity: 0.5; } @@ -729,6 +814,7 @@ tbody tr:hover { .chat-text { white-space: pre-wrap; line-height: 1.45; + font-size: 0.96rem; } .chat-text:empty { @@ -749,7 +835,8 @@ tbody tr:hover { .chat-media video { display: block; max-width: 100%; - border: 1px solid var(--line); + border: 1px solid rgba(255, 255, 255, 0.08); + border-radius: 14px; } /* ---------- Footer ---------- */ @@ -783,7 +870,7 @@ tbody tr:hover { } .api-method { - border-top: 1px dashed var(--line); + border-top: 1px solid var(--line); padding-top: 12px; } @@ -827,7 +914,7 @@ tbody tr:hover { .sidebar, .viewer-sidebar { border-right: 0; - border-bottom: 1px dashed var(--line); + border-bottom: 1px solid var(--line); } .panel-grid { @@ -835,6 +922,124 @@ tbody tr:hover { } } +/* ── Account Tabs ─────────────────────────────────────── */ + +.account-tabs { + display: flex; + flex-wrap: wrap; + gap: 8px; + margin-bottom: 16px; + padding-bottom: 10px; + border-bottom: 1px solid var(--line); +} + +.account-tab { + display: inline-flex; + align-items: center; + gap: 8px; + min-height: 38px; + padding: 7px 14px; + border: 1px solid var(--line); + border-radius: 999px; + background: #101010; + color: var(--text-color); + cursor: pointer; + font: inherit; + font-size: 0.88rem; +} + +.account-tab.active { + border-color: #0f6f67; + color: #ffffff; + background: #0f6f67; +} + +.account-tab .account-tab-status { + font-size: 0.72rem; + opacity: 0.7; +} + +.account-tab .account-tab-status.ok { + color: var(--ok); +} + +.account-tab .account-tab-status.errored { + color: var(--danger); +} + +/* ── Account Panels ──────────────────────────────────── */ + +.account-panel { + display: block; +} + +.account-panel.hidden { + display: none; +} + +/* ── Settings: Account List ──────────────────────────── */ + +.accounts-list { + display: grid; + gap: 8px; + margin-bottom: 12px; +} + +.account-list-item { + display: flex; + align-items: center; + justify-content: space-between; + gap: 12px; + padding: 10px 12px; + border: 1px solid var(--line); + border-radius: 16px; + background: #0f0f0f; +} + +.account-list-info { + display: grid; + gap: 2px; + min-width: 0; +} + +.account-list-label { + font-weight: 700; +} + +.account-list-id { + font-size: 0.82rem; +} + +.account-list-auth-status { + font-size: 0.78rem; +} + +.account-list-actions { + display: flex; + gap: 6px; + flex-shrink: 0; +} + +.add-account-form { + display: grid; + gap: 8px; + margin-top: 8px; + padding: 12px; + border: 1px solid var(--line); + border-radius: 16px; + background: #0f0f0f; +} + +/* ── Active Account Card ─────────────────────────────── */ + +#active-account-name { + word-break: break-all; +} + +#active-account-status { + margin-top: 6px; +} + /* ---------- Mobile: tablet & phone ---------- */ @media (max-width: 768px) { @@ -844,13 +1049,13 @@ tbody tr:hover { .viewer-sidebar { position: fixed; - left: -300px; + left: -360px; top: 0; bottom: 0; - width: 280px; + width: 340px; z-index: 100; transition: left 0.2s ease; - border-right: 1px dashed var(--line); + border-right: 1px solid var(--line); border-bottom: 0; } @@ -939,7 +1144,7 @@ tbody tr:hover { } .viewer-sidebar { - width: 260px; + width: 300px; padding: 16px; } diff --git a/webui/viewer.html b/webui/viewer.html index 1dfe75e..96eae71 100644 --- a/webui/viewer.html +++ b/webui/viewer.html @@ -4,7 +4,7 @@ Telegram Scraper Viewer - +
@@ -13,6 +13,7 @@
Export Viewer

Messages

+

Account:

Dashboard @@ -36,8 +37,17 @@ diff --git a/webui/viewer.js b/webui/viewer.js index 4e211b0..ae20025 100644 --- a/webui/viewer.js +++ b/webui/viewer.js @@ -1,6 +1,8 @@ if ("scrollRestoration" in history) history.scrollRestoration = "manual"; const viewerState = { + accountId: null, + accounts: [], channels: [], channelId: null, oldestMessageId: null, @@ -42,6 +44,24 @@ function formatTime(dateStr) { return d.toLocaleTimeString("en-US", { hour: "2-digit", minute: "2-digit" }); } +function chatListTime(dateStr) { + if (!dateStr) return ""; + const d = new Date(dateStr.replace(" ", "T")); + if (isNaN(d.getTime())) return ""; + const now = new Date(); + if (d.toDateString() === now.toDateString()) { + return d.toLocaleTimeString("en-US", { hour: "2-digit", minute: "2-digit" }); + } + return d.toLocaleDateString("en-US", { month: "short", day: "numeric" }); +} + +function initials(value) { + const source = String(value || "?").replace(/^@/, "").trim(); + const words = source.split(/\s+/).filter(Boolean); + if (words.length > 1) return (words[0][0] + words[1][0]).toUpperCase(); + return source.slice(0, 2).toUpperCase(); +} + function dateKey(dateStr) { if (!dateStr) return ""; return dateStr.slice(0, 10); @@ -75,8 +95,14 @@ function renderChannelList() { if (!template) return; viewerState.channels.forEach((channel) => { const node = template.content.firstElementChild.cloneNode(true); + const preview = + channel.last_message_preview || + (channel.has_database ? "No text in the last saved message" : "No local database yet"); + node.querySelector(".viewer-channel-avatar").textContent = initials(channel.name || channel.channel_id); node.querySelector(".viewer-channel-name").textContent = channel.name; - node.querySelector(".viewer-channel-meta").textContent = `${channel.message_count} messages`; + node.querySelector(".viewer-channel-time").textContent = chatListTime(channel.last_date); + node.querySelector(".viewer-channel-preview").textContent = preview; + node.querySelector(".viewer-channel-count").textContent = String(channel.message_count || 0); if (channel.channel_id === viewerState.channelId) { node.classList.add("active"); } @@ -90,6 +116,102 @@ function renderChannelList() { }); } +function messageEndpoint(channelId, before = "") { + if (viewerState.accountId) { + return `/api/accounts/${encodeURIComponent(viewerState.accountId)}/channels/${encodeURIComponent(channelId)}/messages?limit=80${before}`; + } + return `/api/channels/${encodeURIComponent(channelId)}/messages?limit=80${before}`; +} + +function channelsEndpoint() { + if (viewerState.accountId) { + return `/api/accounts/${encodeURIComponent(viewerState.accountId)}/channels`; + } + return "/api/channels"; +} + +function renderAccountSelector() { + const sel = document.getElementById("viewer-account-select"); + if (!sel) return; + sel.innerHTML = ""; + const legacy = viewerState.accounts.length === 0; + if (legacy) { + const opt = document.createElement("option"); + opt.value = ""; + opt.textContent = "Legacy"; + sel.appendChild(opt); + } else { + viewerState.accounts.forEach((acc) => { + const opt = document.createElement("option"); + opt.value = acc.id; + opt.textContent = acc.label || acc.id; + if (acc.id === viewerState.accountId) opt.selected = true; + sel.appendChild(opt); + }); + } +} + +async function switchViewerAccount(accountId) { + viewerState.accountId = accountId || null; + viewerState.accounts.forEach((acc) => { + if (acc.id === accountId) { + const opts = document.getElementById("viewer-account-select")?.options; + if (opts) { + for (let i = 0; i < opts.length; i++) { + opts[i].selected = opts[i].value === accountId; + } + } + } + }); + try { + const authData = viewerState.accountId + ? await api(`/api/accounts/${encodeURIComponent(viewerState.accountId)}/auth`) + : await api("/api/auth"); + viewerState.userId = authData.user_id || null; + } catch (e) { + console.warn("Could not fetch user ID:", e); + } + const channels = await api(channelsEndpoint()); + viewerState.channels = channels; + viewerState.channelId = channels[0]?.channel_id || null; + viewerState.oldestMessageId = null; + viewerState.newestMessageId = null; + renderChannelList(); + if (viewerState.channelId) { + await loadChannel(viewerState.channelId); + } +} + +async function loadViewerAccount(params) { + const requested = params.get("account"); + try { + const payload = await api("/api/accounts"); + viewerState.accounts = payload.accounts || []; + } catch { + viewerState.accounts = []; + } + viewerState.accountId = + requested || + viewerState.accounts[0]?.id || + null; + + renderAccountSelector(); + + const sel = document.getElementById("viewer-account-select"); + if (sel) { + sel.addEventListener("change", () => switchViewerAccount(sel.value)); + } + + try { + const authData = viewerState.accountId + ? await api(`/api/accounts/${encodeURIComponent(viewerState.accountId)}/auth`) + : await api("/api/auth"); + viewerState.userId = authData.user_id || null; + } catch (e) { + console.warn("Could not fetch user ID:", e); + } +} + function buildMessageNode(message, root) { const template = document.getElementById("message-template"); const wrap = template.content.firstElementChild.cloneNode(true); @@ -240,9 +362,7 @@ async function loadChannel(channelId, append = false) { ? `&before=${viewerState.oldestMessageId}` : ""; try { - const payload = await api( - `/api/channels/${encodeURIComponent(channelId)}/messages?limit=80${before}` - ); + const payload = await api(messageEndpoint(channelId, before)); if (append) { const list = document.getElementById("messages-list"); const prevScrollHeight = list.scrollHeight; @@ -285,15 +405,9 @@ function setupInfiniteScroll() { } async function main() { - try { - const authData = await api("/api/auth"); - viewerState.userId = authData.user_id || null; - } catch (e) { - console.warn("Could not fetch user ID:", e); - } - - viewerState.channels = await api("/api/channels"); const params = new URLSearchParams(window.location.search); + await loadViewerAccount(params); + viewerState.channels = await api(channelsEndpoint()); viewerState.channelId = params.get("channel") || viewerState.channels[0]?.channel_id || diff --git a/webui_server.py b/webui_server.py index bac2aaf..031e78d 100644 --- a/webui_server.py +++ b/webui_server.py @@ -23,11 +23,21 @@ from typing import Any, Dict, List, Optional import qrcode import qrcode.image.svg -from app_state import StateStore +from app_state import ( + StateStore, + account_exists, + account_data_dir, + account_session_path, + get_account_store, + get_global_store, + load_account, + list_accounts, +) from health import health_payload from scraper_jobs import ScraperJobService from telethon import TelegramClient from telethon.errors import SessionPasswordNeededError +from telegram_scraper_with_forwarding import _ensure_session_wal BASE_DIR = Path(__file__).resolve().parent @@ -57,13 +67,7 @@ def save_state(state: Dict[str, Any]) -> None: STATE_STORE.save(state) -def channel_db_path(channel_id: str) -> Path: - return DATA_DIR / channel_id / f"{channel_id}.db" - - -def guess_media_kind( - media_path: Optional[str], media_type: Optional[str] -) -> Optional[str]: +def guess_media_kind(media_path: Optional[str], media_type: Optional[str]) -> Optional[str]: if not media_path: return None suffix = Path(media_path).suffix.lower() @@ -95,14 +99,22 @@ def normalize_media_url(media_path: Optional[str]) -> Optional[str]: return "/media/" + urllib.parse.quote(str(relative).replace("\\", "/")) -def database_summary(channel_id: str) -> Dict[str, Any]: - db_path = channel_db_path(channel_id) - summary = { +def channel_db_path(account_id: Optional[str], channel_id: str) -> Path: + if account_id: + return account_data_dir(DATA_DIR, account_id) / channel_id / f"{channel_id}.db" + return DATA_DIR / channel_id / f"{channel_id}.db" + + +def database_summary(account_id: Optional[str], channel_id: str) -> Dict[str, Any]: + db_path = channel_db_path(account_id, channel_id) + summary: Dict[str, Any] = { "message_count": 0, "last_date": None, "first_date": None, "media_count": 0, "has_database": db_path.exists(), + "last_message_preview": "", + "last_sender": "", } if not db_path.exists(): return summary @@ -123,6 +135,21 @@ def database_summary(channel_id: str) -> Dict[str, Any]: "media_count": row[3] or 0, } ) + try: + cursor.execute( + "SELECT message, media_type, first_name, last_name, username, post_author " + "FROM messages ORDER BY message_id DESC LIMIT 1" + ) + latest = cursor.fetchone() + except sqlite3.OperationalError: + latest = None + if latest: + text = (latest[0] or "").strip() + if not text and latest[1]: + text = f"[{latest[1]}]" + sender = " ".join(part for part in [latest[2], latest[3]] if part).strip() + summary["last_message_preview"] = text[:180] + summary["last_sender"] = sender or latest[4] or latest[5] or "" return summary finally: conn.close() @@ -132,11 +159,14 @@ def channel_display_name(state: Dict[str, Any], channel_id: str) -> str: return state.get("channel_names", {}).get(channel_id) or channel_id -def list_channels_snapshot() -> List[Dict[str, Any]]: - state = load_state() +def list_channels_snapshot(account_id: Optional[str] = None) -> List[Dict[str, Any]]: + if account_id: + state = load_account(DATA_DIR, account_id) + else: + state = load_state() channels: List[Dict[str, Any]] = [] for channel_id, last_message_id in state.get("channels", {}).items(): - summary = database_summary(channel_id) + summary = database_summary(account_id, channel_id) channels.append( { "channel_id": channel_id, @@ -146,19 +176,26 @@ def list_channels_snapshot() -> List[Dict[str, Any]]: } ) channels.sort( - key=lambda item: ( - item["last_date"] or "", - item["message_count"], - ), + key=lambda item: (item["last_date"] or "", item["message_count"]), reverse=True, ) return channels +def account_state_for_legacy() -> Optional[Dict[str, Any]]: + legacy = load_state() + if legacy.get("api_id") and legacy.get("api_hash"): + return legacy + return None + + def load_messages( - channel_id: str, limit: int = 120, before_message_id: Optional[int] = None + account_id: Optional[str], + channel_id: str, + limit: int = 120, + before_message_id: Optional[int] = None, ) -> List[Dict[str, Any]]: - db_path = channel_db_path(channel_id) + db_path = channel_db_path(account_id, channel_id) if not db_path.exists(): return [] conn = sqlite3.connect(str(db_path)) @@ -239,6 +276,8 @@ def load_messages( return messages +# ── Job ────────────────────────────────────────────────────────────────── + @dataclass class Job: job_id: str @@ -251,6 +290,7 @@ class Job: finished_at: Optional[str] = None logs: str = "" error: Optional[str] = None + account_id: Optional[str] = None def to_dict(self) -> Dict[str, Any]: return { @@ -264,9 +304,12 @@ class Job: "finished_at": self.finished_at, "logs": self.logs, "error": self.error, + "account_id": self.account_id, } +# ── JobRunner ──────────────────────────────────────────────────────────── + class JobRunner: def __init__(self) -> None: self.jobs: Dict[str, Job] = {} @@ -281,7 +324,14 @@ class JobRunner: if self._shutdown_flag: raise RuntimeError("Server is shutting down, cannot create new jobs") job_id = f"job-{int(time.time() * 1000)}" - job = Job(job_id=job_id, job_type=job_type, title=title, payload=payload) + account_id = payload.get("account_id") + job = Job( + job_id=job_id, + job_type=job_type, + title=title, + payload=payload, + account_id=account_id, + ) with self.lock: self.jobs[job_id] = job self.job_order.insert(0, job_id) @@ -289,9 +339,12 @@ class JobRunner: self.queue.put(job) return job - def recent_jobs(self) -> List[Dict[str, Any]]: + def recent_jobs(self, account_id: Optional[str] = None) -> List[Dict[str, Any]]: with self.lock: - return [self.jobs[job_id].to_dict() for job_id in self.job_order] + result = [self.jobs[job_id].to_dict() for job_id in self.job_order] + if account_id is not None: + result = [j for j in result if j.get("account_id") == account_id] + return result def get_job(self, job_id: str) -> Optional[Dict[str, Any]]: with self.lock: @@ -306,9 +359,7 @@ class JobRunner: if job.status == "running": running_jobs.append(job) if running_jobs: - logger.warning( - "Waiting for %d running job(s) to finish...", len(running_jobs) - ) + logger.warning("Waiting for %d running job(s) to finish...", len(running_jobs)) deadline = time.time() + timeout while time.time() < deadline: with self.lock: @@ -354,27 +405,16 @@ class JobRunner: job.finished_at = utc_now_iso() +# ── TelegramAuthManager (per-account) ──────────────────────────────────── + class TelegramAuthManager: def __init__(self) -> None: self.lock = threading.Lock() self.loop = asyncio.new_event_loop() self.thread = threading.Thread(target=self._run_loop, daemon=True) self.thread.start() - self.client: Optional[TelegramClient] = None - self.user_id: Optional[int] = None - self.qr_login = None - self.qr_wait_task = None - self.phone = None - self.phone_code_hash = None - self.state: Dict[str, Any] = { - "phase": "idle", - "status": "unknown", - "details": "", - "qr_url": None, - "qr_image": None, - "phone": None, - "updated_at": utc_now_iso(), - } + self.clients: Dict[str, TelegramClient] = {} + self.auth_data: Dict[str, Dict[str, Any]] = {} def _run_loop(self) -> None: asyncio.set_event_loop(self.loop) @@ -384,56 +424,27 @@ class TelegramAuthManager: future = asyncio.run_coroutine_threadsafe(coro, self.loop) return future.result() - def _set_state(self, **updates: Any) -> None: - with self.lock: - self.state.update(updates) - self.state["updated_at"] = utc_now_iso() + def _get_auth_data(self, account_id: str) -> Dict[str, Any]: + if account_id not in self.auth_data: + self.auth_data[account_id] = { + "phase": "idle", + "status": "unknown", + "details": "", + "qr_url": None, + "qr_image": None, + "phone": None, + "phone_code_hash": None, + "qr_login": None, + "qr_wait_task": None, + "user_id": None, + "updated_at": utc_now_iso(), + } + return self.auth_data[account_id] - def snapshot(self) -> Dict[str, Any]: - auth = auth_status() - with self.lock: - snapshot = dict(self.state) - snapshot["user_id"] = self.user_id - snapshot["saved_credentials"] = { - "api_id": load_state().get("api_id"), - "api_hash_present": bool(load_state().get("api_hash")), - } - snapshot["auth_status"] = auth - return snapshot - - async def _get_client(self) -> TelegramClient: - state = load_state() - api_id = state.get("api_id") - api_hash = state.get("api_hash") - if not api_id or not api_hash: - raise RuntimeError("Save api_id and api_hash first.") - if self.client is None: - self.client = TelegramClient(str(SESSION_DIR / "session"), api_id, api_hash) - if not self.client.is_connected(): - await self.client.connect() - if self.user_id is None and await self.client.is_user_authorized(): - try: - me = await self.client.get_me() - self.user_id = me.id - except Exception: - pass - return self.client - - async def _is_authorized(self) -> bool: - client = await self._get_client() - return await client.is_user_authorized() - - def save_credentials(self, api_id: int, api_hash: str) -> Dict[str, Any]: - state = load_state() - state["api_id"] = int(api_id) - state["api_hash"] = api_hash.strip() - save_state(state) - self._set_state( - phase="credentials_saved", - status="ready_for_auth", - details="Credentials saved. You can login with QR or phone code.", - ) - return self.snapshot() + def _set_state(self, account_id: str, **updates: Any) -> None: + data = self._get_auth_data(account_id) + data.update(updates) + data["updated_at"] = utc_now_iso() def _make_qr_image(self, qr_url: str) -> str: qr = qrcode.QRCode(border=1, box_size=8) @@ -445,10 +456,102 @@ class TelegramAuthManager: encoded = base64.b64encode(buffer.getvalue()).decode("ascii") return f"data:image/svg+xml;base64,{encoded}" - async def _wait_for_qr_login(self) -> None: - try: - await self.qr_login.wait() + async def _get_client(self, account_id: str) -> TelegramClient: + acc_state = load_account(DATA_DIR, account_id) + api_id = acc_state.get("api_id") + api_hash = acc_state.get("api_hash") + if not api_id or not api_hash: + raise RuntimeError("Save api_id and api_hash first for this account.") + if account_id not in self.clients or self.clients[account_id] is None: + _ensure_session_wal(account_session_path(SESSION_DIR, account_id)) + self.clients[account_id] = TelegramClient( + account_session_path(SESSION_DIR, account_id), + api_id, + api_hash, + ) + client = self.clients[account_id] + if not client.is_connected(): + await client.connect() + data = self._get_auth_data(account_id) + if data.get("user_id") is None and await client.is_user_authorized(): + try: + me = await client.get_me() + data["user_id"] = me.id + except Exception: + pass + return client + + def auth_state(self, account_id: str) -> Dict[str, Any]: + data = self._get_auth_data(account_id) + acc_state = load_account(DATA_DIR, account_id) + snapshot = dict(data) + snapshot.pop("qr_login", None) + snapshot.pop("qr_wait_task", None) + snapshot.pop("phone_code_hash", None) + snapshot["saved_credentials"] = { + "api_id": acc_state.get("api_id"), + "api_hash_present": bool(acc_state.get("api_hash")), + } + snapshot["auth_status"] = auth_status_for(account_id) + return snapshot + + def all_auth_states(self) -> Dict[str, Dict[str, Any]]: + result: Dict[str, Dict[str, Any]] = {} + for account_id in list_accounts(DATA_DIR): + result[account_id] = self.auth_state(account_id) + return result + + def save_credentials(self, account_id: str, api_id: int, api_hash: str) -> Dict[str, Any]: + def mutate(state: Dict[str, Any]) -> None: + state["api_id"] = int(api_id) + state["api_hash"] = api_hash.strip() + + store = get_account_store(DATA_DIR, account_id) + store.update(mutate) + self._set_state( + account_id, + phase="credentials_saved", + status="ready_for_auth", + details="Credentials saved. You can login with QR or phone code.", + ) + return self.auth_state(account_id) + + async def _start_qr_login(self, account_id: str) -> Dict[str, Any]: + client = await self._get_client(account_id) + data = self._get_auth_data(account_id) + if await client.is_user_authorized(): self._set_state( + account_id, + phase="authorized", + status="authorized", + details="Telegram session is already authorized.", + ) + return self.auth_state(account_id) + qr_login = await client.qr_login() + qr_url = qr_login.url + data["qr_login"] = qr_login + self._set_state( + account_id, + phase="qr_waiting", + status="qr_waiting", + details="Scan the QR code in Telegram: Settings -> Devices -> Scan QR.", + qr_url=qr_url, + qr_image=self._make_qr_image(qr_url), + ) + data["qr_wait_task"] = self.loop.create_task( + self._wait_for_qr_login(account_id) + ) + return self.auth_state(account_id) + + async def _wait_for_qr_login(self, account_id: str) -> None: + data = self._get_auth_data(account_id) + qr_login = data.get("qr_login") + if not qr_login: + return + try: + await qr_login.wait() + self._set_state( + account_id, phase="authorized", status="authorized", details="Telegram session is authorized.", @@ -458,12 +561,14 @@ class TelegramAuthManager: ) except SessionPasswordNeededError: self._set_state( + account_id, phase="password_required", status="password_required", details="Two-factor authentication is enabled. Enter your Telegram password.", ) except Exception as exc: self._set_state( + account_id, phase="error", status="error", details=f"QR login failed: {exc}", @@ -471,62 +576,48 @@ class TelegramAuthManager: qr_image=None, ) - async def _start_qr_login(self) -> Dict[str, Any]: - client = await self._get_client() + def start_qr_login(self, account_id: str) -> Dict[str, Any]: + return self._run(self._start_qr_login(account_id)) + + async def _request_phone_code(self, account_id: str, phone: str) -> Dict[str, Any]: + client = await self._get_client(account_id) + data = self._get_auth_data(account_id) if await client.is_user_authorized(): self._set_state( + account_id, phase="authorized", status="authorized", details="Telegram session is already authorized.", ) - return self.snapshot() - self.qr_login = await client.qr_login() - qr_url = self.qr_login.url - self._set_state( - phase="qr_waiting", - status="qr_waiting", - details="Scan the QR code in Telegram: Settings -> Devices -> Scan QR.", - qr_url=qr_url, - qr_image=self._make_qr_image(qr_url), - ) - self.qr_wait_task = self.loop.create_task(self._wait_for_qr_login()) - return self.snapshot() - - def start_qr_login(self) -> Dict[str, Any]: - return self._run(self._start_qr_login()) - - async def _request_phone_code(self, phone: str) -> Dict[str, Any]: - client = await self._get_client() - if await client.is_user_authorized(): - self._set_state( - phase="authorized", - status="authorized", - details="Telegram session is already authorized.", - ) - return self.snapshot() + return self.auth_state(account_id) sent = await client.send_code_request(phone) - self.phone = phone - self.phone_code_hash = sent.phone_code_hash + data["phone"] = phone + data["phone_code_hash"] = sent.phone_code_hash self._set_state( + account_id, phase="code_required", status="code_required", details=f"Code sent to {phone}. Enter it below.", phone=phone, ) - return self.snapshot() + return self.auth_state(account_id) - def request_phone_code(self, phone: str) -> Dict[str, Any]: - return self._run(self._request_phone_code(phone)) + def request_phone_code(self, account_id: str, phone: str) -> Dict[str, Any]: + return self._run(self._request_phone_code(account_id, phone)) - async def _submit_phone_code(self, code: str) -> Dict[str, Any]: - client = await self._get_client() - if not self.phone or not self.phone_code_hash: + async def _submit_phone_code(self, account_id: str, code: str) -> Dict[str, Any]: + client = await self._get_client(account_id) + data = self._get_auth_data(account_id) + if not data.get("phone") or not data.get("phone_code_hash"): raise RuntimeError("Request a phone code first.") try: await client.sign_in( - phone=self.phone, code=code, phone_code_hash=self.phone_code_hash + phone=data["phone"], + code=code, + phone_code_hash=data["phone_code_hash"], ) self._set_state( + account_id, phase="authorized", status="authorized", details="Telegram session is authorized.", @@ -535,38 +626,61 @@ class TelegramAuthManager: ) except SessionPasswordNeededError: self._set_state( + account_id, phase="password_required", status="password_required", details="Two-factor authentication is enabled. Enter your Telegram password.", ) - return self.snapshot() + return self.auth_state(account_id) - def submit_phone_code(self, code: str) -> Dict[str, Any]: - return self._run(self._submit_phone_code(code)) + def submit_phone_code(self, account_id: str, code: str) -> Dict[str, Any]: + return self._run(self._submit_phone_code(account_id, code)) - async def _submit_password(self, password: str) -> Dict[str, Any]: - client = await self._get_client() + async def _submit_password(self, account_id: str, password: str) -> Dict[str, Any]: + client = await self._get_client(account_id) await client.sign_in(password=password) self._set_state( + account_id, phase="authorized", status="authorized", details="Telegram session is authorized.", qr_url=None, qr_image=None, ) - return self.snapshot() + return self.auth_state(account_id) - def submit_password(self, password: str) -> Dict[str, Any]: - return self._run(self._submit_password(password)) + def submit_password(self, account_id: str, password: str) -> Dict[str, Any]: + return self._run(self._submit_password(account_id, password)) + + def delete_account(self, account_id: str) -> None: + if account_id in self.clients: + client = self.clients[account_id] + try: + future = asyncio.run_coroutine_threadsafe( + self._disconnect_client(client), self.loop + ) + future.result(timeout=5) + except Exception: + pass + del self.clients[account_id] + self.auth_data.pop(account_id, None) + + async def _disconnect_client(self, client: TelegramClient) -> None: + if client and client.is_connected(): + await client.disconnect() def shutdown(self, timeout: float = 5.0) -> None: - async def _disconnect(): - if self.client: - await self.client.disconnect() + async def _disconnect_all(): + for client in self.clients.values(): + try: + if client and client.is_connected(): + await client.disconnect() + except Exception: + pass - if self.client: + if self.clients: try: - future = asyncio.run_coroutine_threadsafe(_disconnect(), self.loop) + future = asyncio.run_coroutine_threadsafe(_disconnect_all(), self.loop) future.result(timeout=timeout) except Exception: pass @@ -574,12 +688,15 @@ class TelegramAuthManager: self.thread.join(timeout=timeout) -class ContinuousScrapeManager: - def __init__(self) -> None: +# ── PerAccountContinuousScrapeManager ──────────────────────────────────── + +class PerAccountContinuousScrapeManager: + def __init__(self, account_id: str) -> None: + self.account_id = account_id self.lock = threading.RLock() self.thread: Optional[threading.Thread] = None self.stop_event = threading.Event() - self.config: Dict[str, Any] = STATE_STORE.continuous_config() + self.config: Dict[str, Any] = self._load_config() self.status: Dict[str, Any] = { "running": False, "last_started_at": None, @@ -590,14 +707,37 @@ class ContinuousScrapeManager: "log_entries": [], } + def _load_config(self) -> Dict[str, Any]: + acc_state = load_account(DATA_DIR, self.account_id) + return dict(acc_state.get("continuous_scraping", { + "enabled": True, + "interval_minutes": 1, + "channels": [], + "run_all_tracked": True, + })) + + def _save_config(self) -> None: + store = get_account_store(DATA_DIR, self.account_id) + def mutate(state: Dict[str, Any]) -> None: + state["continuous_scraping"] = { + "enabled": bool(self.config.get("enabled", True)), + "interval_minutes": max(1, int(self.config.get("interval_minutes", 1) or 1)), + "channels": [ + str(item).strip() + for item in self.config.get("channels", []) + if str(item).strip() + ], + "run_all_tracked": bool(self.config.get("run_all_tracked", True)), + } + store.update(mutate) + def _log(self, message: str, level: str = "debug") -> None: timestamp = utc_now_iso() - line = f"[{datetime.now().strftime('%H:%M:%S')}] {message}" entry = {"timestamp": timestamp, "level": level, "message": message} with self.lock: logs = self.status.setdefault("logs", []) log_entries = self.status.setdefault("log_entries", []) - logs.append(line) + logs.append(f"[{datetime.now().strftime('%H:%M:%S')}] {message}") log_entries.append(entry) if len(logs) > 300: del logs[:-300] @@ -624,14 +764,14 @@ class ContinuousScrapeManager: ) -> Dict[str, Any]: interval_minutes = max(1, int(interval_minutes)) with self.lock: - self.config = STATE_STORE.save_continuous_config( - { - "enabled": bool(enabled), - "interval_minutes": interval_minutes, - "channels": channels, - "run_all_tracked": bool(run_all_tracked), - } - ) + self.config = { + "enabled": bool(enabled), + "interval_minutes": interval_minutes, + "channels": channels, + "run_all_tracked": bool(run_all_tracked), + } + self._save_config() + self.config = self._load_config() if enabled: self.start() else: @@ -660,19 +800,20 @@ class ContinuousScrapeManager: self._log("Continuous scraping stop requested.", "warn") def _resolve_channels(self) -> List[str]: - state = load_state() + acc_state = load_account(DATA_DIR, self.account_id) with self.lock: - run_all_tracked = self.config["run_all_tracked"] - configured = list(self.config["channels"]) + run_all_tracked = self.config.get("run_all_tracked", True) + configured = list(self.config.get("channels", [])) if run_all_tracked: - return list(state.get("channels", {}).keys()) - tracked = set(state.get("channels", {}).keys()) + return list(acc_state.get("channels", {}).keys()) + tracked = set(acc_state.get("channels", {}).keys()) return [channel for channel in configured if channel in tracked] def _run_loop(self) -> None: while not self.stop_event.is_set(): channels = self._resolve_channels() - interval_minutes = self.snapshot()["config"]["interval_minutes"] + cfg = self.snapshot()["config"] + interval_minutes = cfg.get("interval_minutes", 1) if not channels: self._log("No channels configured for continuous scraping.", "warn") @@ -680,11 +821,11 @@ class ContinuousScrapeManager: self._log(f"Starting iteration for {len(channels)} channel(s).", "info") buffer = io.StringIO() try: - with ( - contextlib.redirect_stdout(buffer), - contextlib.redirect_stderr(buffer), - ): - run_job("scrape_selected", {"channels": channels}) + with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer): + run_job("scrape_selected", { + "channels": channels, + "account_id": self.account_id, + }) output = buffer.getvalue().strip() if output: for line in output.splitlines(): @@ -714,9 +855,83 @@ class ContinuousScrapeManager: self._log("Continuous scraping stopped.", "warn") +# ── ContinuousScrapeOrchestrator ───────────────────────────────────────── + +class ContinuousScrapeOrchestrator: + def __init__(self) -> None: + self.lock = threading.Lock() + self.managers: Dict[str, PerAccountContinuousScrapeManager] = {} + + def _get_or_create(self, account_id: str) -> PerAccountContinuousScrapeManager: + with self.lock: + if account_id not in self.managers: + self.managers[account_id] = PerAccountContinuousScrapeManager(account_id) + return self.managers[account_id] + + def start_account(self, account_id: str) -> None: + mgr = self._get_or_create(account_id) + mgr.start() + + def stop_account(self, account_id: str) -> None: + with self.lock: + mgr = self.managers.get(account_id) + if mgr: + mgr.stop() + + def snapshot_for(self, account_id: str) -> Dict[str, Any]: + mgr = self._get_or_create(account_id) + return mgr.snapshot() + + def snapshot(self) -> Dict[str, Dict[str, Any]]: + result: Dict[str, Dict[str, Any]] = {} + with self.lock: + for account_id, mgr in self.managers.items(): + result[account_id] = dict(mgr.snapshot()) + missing = set(list_accounts(DATA_DIR)) - set(result.keys()) + for account_id in missing: + mgr = self._get_or_create(account_id) + result[account_id] = dict(mgr.snapshot()) + return result + + def update_for( + self, + account_id: str, + enabled: bool, + interval_minutes: int, + channels: List[str], + run_all_tracked: bool, + ) -> Dict[str, Any]: + mgr = self._get_or_create(account_id) + return mgr.update(enabled, interval_minutes, channels, run_all_tracked) + + def add_account(self, account_id: str) -> None: + acc_state = load_account(DATA_DIR, account_id) + cfg = acc_state.get("continuous_scraping", {}) + if cfg.get("enabled", True) and START_CONTINUOUS: + self.start_account(account_id) + + def remove_account(self, account_id: str) -> None: + self.stop_account(account_id) + with self.lock: + self.managers.pop(account_id, None) + + def stop_all(self) -> None: + with self.lock: + for account_id in list(self.managers.keys()): + self.managers[account_id].stop() + + def start_all(self) -> None: + for account_id in list_accounts(DATA_DIR): + acc_state = load_account(DATA_DIR, account_id) + cfg = acc_state.get("continuous_scraping", {}) + if cfg.get("enabled", True): + self.start_account(account_id) + + +# ── Legacy helpers ─────────────────────────────────────────────────────── + def import_scraper_class(): from telegram_scraper_with_forwarding import OptimizedTelegramScraper - return OptimizedTelegramScraper @@ -724,10 +939,15 @@ def run_job(job_type: str, payload: Dict[str, Any]) -> None: SCRAPER_JOBS.run(job_type, payload) -def auth_status() -> Dict[str, Any]: - state = load_state() - status = { - "has_api_credentials": bool(state.get("api_id") and state.get("api_hash")), +def auth_status_for(account_id: Optional[str] = None) -> Dict[str, Any]: + if account_id: + acc_state = load_account(DATA_DIR, account_id) + session_file = account_session_path(SESSION_DIR, account_id) + else: + acc_state = load_state() + session_file = str(SESSION_DIR / "session.session") + status: Dict[str, Any] = { + "has_api_credentials": bool(acc_state.get("api_id") and acc_state.get("api_hash")), "telethon_available": False, "session_ready": False, "status": "read_only", @@ -740,22 +960,23 @@ def auth_status() -> Dict[str, Any]: status["details"] = f"Python deps are missing: {exc}" return status - session_file = BASE_DIR / "session" / "session.session" - status["session_ready"] = session_file.exists() + status["session_ready"] = Path(session_file).exists() if status["has_api_credentials"] and status["session_ready"]: status["status"] = "ready" - status["details"] = ( - "Scraping actions should be available after dependencies are installed." - ) + status["details"] = "Scraping actions should be available after dependencies are installed." elif status["has_api_credentials"]: status["status"] = "needs_auth" status["details"] = "API keys found, but Telegram session file is missing." else: status["status"] = "needs_credentials" - status["details"] = "api_id/api_hash are missing in data/state.json." + status["details"] = "api_id/api_hash are missing." return status +def auth_status() -> Dict[str, Any]: + return auth_status_for(account_id=None) + + def dashboard_payload(job_runner: JobRunner) -> Dict[str, Any]: state = load_state() channels = list_channels_snapshot() @@ -912,7 +1133,6 @@ def openapi_payload() -> Dict[str, Any]: "/health": { "get": { "summary": "Application health", - "description": "Checks data/session write access, state loading, SQLite, continuous status, and job queue size.", "responses": json_response, } }, @@ -1068,10 +1288,296 @@ def openapi_payload() -> Dict[str, Any]: "responses": json_response, } }, + "/api/accounts": { + "get": { + "summary": "List all accounts", + "responses": json_response, + }, + "post": { + "summary": "Create a new account", + "requestBody": json_body( + { + "account_id": { + "type": "string", + "example": "work", + }, + "label": { + "type": "string", + "example": "Work Account", + }, + "api_id": { + "type": "integer", + "example": 123456, + }, + "api_hash": { + "type": "string", + "example": "0123456789abcdef", + }, + } + ), + "responses": {**json_response, **error_response}, + }, + }, + "/api/accounts/{id}": { + "get": { + "summary": "Account dashboard", + "parameters": [ + { + "name": "id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ], + "responses": json_response, + }, + "delete": { + "summary": "Delete account", + "parameters": [ + { + "name": "id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ], + "responses": json_response, + }, + }, + "/api/accounts/{id}/auth": { + "get": { + "summary": "Account auth snapshot", + "parameters": [ + { + "name": "id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ], + "responses": json_response, + }, + }, + "/api/accounts/{id}/auth/credentials": { + "post": { + "summary": "Save credentials for account", + "requestBody": json_body( + { + "api_id": {"type": "integer", "example": 123456}, + "api_hash": {"type": "string", "example": "0123456789abcdef"}, + }, + ["api_id", "api_hash"], + ), + "responses": {**json_response, **error_response}, + } + }, + "/api/accounts/{id}/auth/qr/start": { + "post": { + "summary": "Start QR login for account", + "requestBody": json_body({}), + "responses": {**json_response, **error_response}, + } + }, + "/api/accounts/{id}/auth/phone/request": { + "post": { + "summary": "Request phone code for account", + "requestBody": json_body( + {"phone": {"type": "string", "example": "+1234567890"}}, + ["phone"], + ), + "responses": {**json_response, **error_response}, + } + }, + "/api/accounts/{id}/auth/phone/submit": { + "post": { + "summary": "Submit phone code for account", + "requestBody": json_body( + {"code": {"type": "string", "example": "12345"}}, + ["code"], + ), + "responses": {**json_response, **error_response}, + } + }, + "/api/accounts/{id}/auth/password": { + "post": { + "summary": "Submit 2FA password for account", + "requestBody": json_body( + {"password": {"type": "string", "example": "password"}}, + ["password"], + ), + "responses": {**json_response, **error_response}, + } + }, + "/api/accounts/{id}/channels": { + "get": { + "summary": "List channels for account", + "parameters": [ + { + "name": "id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ], + "responses": json_response, + } + }, + "/api/accounts/{id}/channels/{channel_id}/messages": { + "get": { + "summary": "List messages for account channel", + "parameters": [ + { + "name": "id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + }, + { + "name": "channel_id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + }, + { + "name": "limit", + "in": "query", + "schema": { + "type": "integer", + "default": 120, + "maximum": 300, + }, + }, + { + "name": "before", + "in": "query", + "schema": {"type": "integer"}, + }, + ], + "responses": json_response, + } + }, + "/api/accounts/{id}/channels/add": { + "post": { + "summary": "Add channel to account", + "requestBody": json_body( + { + "channel_id": {"type": "string", "example": "-1001234567890"}, + "name": {"type": "string", "example": "Research feed"}, + }, + ["channel_id"], + ), + "responses": {**json_response, **error_response}, + } + }, + "/api/accounts/{id}/channels/remove": { + "post": { + "summary": "Remove channel from account", + "requestBody": json_body( + {"channel_id": {"type": "string", "example": "-1001234567890"}}, + ["channel_id"], + ), + "responses": {**json_response, **error_response}, + } + }, + "/api/accounts/{id}/jobs/scrape": { + "post": { + "summary": "Scrape channels for account", + "requestBody": json_body( + {"channel_id": {"type": "string", "example": "-1001234567890"}} + ), + "responses": accepted_response, + } + }, + "/api/accounts/{id}/jobs/export": { + "post": { + "summary": "Export all channels for account", + "requestBody": json_body({}), + "responses": accepted_response, + } + }, + "/api/accounts/{id}/jobs/export-channel": { + "post": { + "summary": "Export one channel for account", + "requestBody": json_body( + {"channel_id": {"type": "string", "example": "-1001234567890"}}, + ["channel_id"], + ), + "responses": {**accepted_response, **error_response}, + } + }, + "/api/accounts/{id}/jobs/rescrape-media": { + "post": { + "summary": "Rescrape media for account channel", + "requestBody": json_body( + {"channel_id": {"type": "string", "example": "-1001234567890"}}, + ["channel_id"], + ), + "responses": {**accepted_response, **error_response}, + } + }, + "/api/accounts/{id}/jobs/fix-media": { + "post": { + "summary": "Fix missing media for account channel", + "requestBody": json_body( + {"channel_id": {"type": "string", "example": "-1001234567890"}}, + ["channel_id"], + ), + "responses": {**accepted_response, **error_response}, + } + }, + "/api/accounts/{id}/jobs/refresh-dialogs": { + "post": { + "summary": "Refresh dialogs for account", + "requestBody": json_body({}), + "responses": accepted_response, + } + }, + "/api/accounts/{id}/settings/media": { + "post": { + "summary": "Toggle media scraping for account", + "requestBody": json_body( + {"value": {"type": "boolean", "example": True}}, + ["value"], + ), + "responses": {**accepted_response, **error_response}, + } + }, + "/api/accounts/{id}/continuous": { + "get": { + "summary": "Continuous scraping status for account", + "parameters": [ + { + "name": "id", + "in": "path", + "required": True, + "schema": {"type": "string"}, + } + ], + "responses": json_response, + }, + "post": { + "summary": "Update continuous scraping for account", + "requestBody": json_body( + { + "enabled": {"type": "boolean", "example": True}, + "interval_minutes": {"type": "integer", "example": 5}, + "run_all_tracked": {"type": "boolean", "example": False}, + "channels": { + "type": "array", + "items": {"type": "string"}, + "example": ["-1001234567890"], + }, + } + ), + "responses": {**json_response, **error_response}, + }, + }, }, } +# ── Request Handler ────────────────────────────────────────────────────── + class TelegramScraperRequestHandler(BaseHTTPRequestHandler): server_version = "TelegramScraperWebUI/0.1" @@ -1079,6 +1585,8 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): def app(self) -> "TelegramScraperWebServer": return self.server # type: ignore[return-value] + # ── GET ────────────────────────────────────────────────────────────── + def do_GET(self) -> None: parsed = urllib.parse.urlparse(self.path) path = parsed.path @@ -1087,13 +1595,9 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): if path == "/": return self.serve_file(WEBUI_DIR / "index.html", "text/html; charset=utf-8") if path == "/viewer": - return self.serve_file( - WEBUI_DIR / "viewer.html", "text/html; charset=utf-8" - ) + return self.serve_file(WEBUI_DIR / "viewer.html", "text/html; charset=utf-8") if path in {"/swagger", "/swigger", "/docs"}: - return self.serve_file( - WEBUI_DIR / "swagger.html", "text/html; charset=utf-8" - ) + return self.serve_file(WEBUI_DIR / "swagger.html", "text/html; charset=utf-8") if path.startswith("/static/"): relative = path.removeprefix("/static/") return self.serve_static(relative) @@ -1105,19 +1609,27 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): if path == "/api/dashboard": return self.send_json(dashboard_payload(self.app.job_runner)) if path == "/api/auth": - return self.send_json(self.app.auth_manager.snapshot()) + if self.app.legacy_account_id: + return self.send_json(self.app.auth_manager.auth_state(self.app.legacy_account_id)) + return self.send_json(self._legacy_auth_snapshot()) if path == "/api/continuous": - return self.send_json(self.app.continuous_manager.snapshot()) + if self.app.legacy_account_id: + return self.send_json(self.app.continuous_orchestrator.snapshot_for(self.app.legacy_account_id)) + return self.send_json({"config": {}, "status": {"running": False, "logs": [], "log_entries": []}}) if path == "/health/continuous": - return self.send_json(self.app.continuous_manager.snapshot()) + if self.app.legacy_account_id: + return self.send_json(self.app.continuous_orchestrator.snapshot_for(self.app.legacy_account_id)) + return self.send_json({"config": {}, "status": {"running": False, "logs": [], "log_entries": []}}) if path == "/health": + cs_snapshot = self.app.continuous_orchestrator.snapshot() return self.send_json( health_payload( DATA_DIR, SESSION_DIR, STATE_STORE, - self.app.continuous_manager.snapshot(), + cs_snapshot, self.app.job_runner.queue.qsize(), + list_accounts(DATA_DIR), ) ) if path == "/api/jobs": @@ -1139,7 +1651,7 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): payload = { "channel_id": channel_id, "messages": load_messages( - channel_id, limit=limit, before_message_id=before_message_id + None, channel_id, limit=limit, before_message_id=before_message_id ), "channel": next( ( @@ -1151,26 +1663,127 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): ), } return self.send_json(payload) + + # ── /api/accounts/* routes ────────────────────────────────────── + if path == "/api/accounts": + return self._handle_get_accounts_list() + if path.startswith("/api/accounts/"): + return self._handle_get_account(path, query) + return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found") + def _legacy_auth_snapshot(self) -> Dict[str, Any]: + state = load_state() + session_ready = (SESSION_DIR / "session.session").exists() + return { + "phase": "unknown", + "status": "unknown", + "details": "", + "qr_url": None, + "qr_image": None, + "phone": None, + "user_id": None, + "saved_credentials": { + "api_id": state.get("api_id"), + "api_hash_present": bool(state.get("api_hash")), + }, + "auth_status": auth_status(), + } + + def _handle_get_accounts_list(self) -> None: + ids = list_accounts(DATA_DIR) + accounts = [] + for account_id in ids: + acc_state = load_account(DATA_DIR, account_id) + auth_info = auth_status_for(account_id) + cs_info = self.app.continuous_orchestrator.snapshot_for(account_id) + accounts.append({ + "id": account_id, + "label": acc_state.get("label", ""), + "auth": auth_info, + "status": auth_info.get("status", "unknown"), + "continuous_running": cs_info.get("status", {}).get("running", False), + }) + return self.send_json({"accounts": accounts}) + + def _handle_get_account(self, path: str, query: Dict[str, List[str]]) -> None: + rest = path[len("/api/accounts/"):] + parts = rest.split("/") + account_id = urllib.parse.unquote(parts[0]) + sub = parts[1:] + + if not self._account_exists_or_404(account_id): + return + + if not sub: + return self._handle_get_account_dashboard(account_id) + if sub == ["auth"]: + return self.send_json(self.app.auth_manager.auth_state(account_id)) + if sub == ["channels"]: + return self._handle_get_account_channels(account_id) + if len(sub) >= 3 and sub[0] == "channels" and sub[-1] == "messages": + channel_id = sub[1] + return self._handle_get_account_channel_messages(account_id, channel_id, query) + if sub == ["continuous"]: + return self.send_json( + self.app.continuous_orchestrator.snapshot_for(account_id) + ) + return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found") + + def _handle_get_account_dashboard(self, account_id: str) -> None: + acc_state = load_account(DATA_DIR, account_id) + channels = list_channels_snapshot(account_id) + auth_info = auth_status_for(account_id) + cs_info = self.app.continuous_orchestrator.snapshot_for(account_id) + payload = { + "account_id": account_id, + "label": acc_state.get("label", ""), + "auth": auth_info, + "auth_detail": self.app.auth_manager.auth_state(account_id), + "channels": channels, + "jobs": self.app.job_runner.recent_jobs(account_id=account_id), + "continuous": cs_info, + "scrape_media": bool(acc_state.get("scrape_media", True)), + } + return self.send_json(payload) + + def _handle_get_account_channels(self, account_id: str) -> None: + return self.send_json(list_channels_snapshot(account_id)) + + def _handle_get_account_channel_messages( + self, account_id: str, channel_id: str, query: Dict[str, List[str]] + ) -> None: + limit = max(1, min(int(query.get("limit", ["120"])[0]), 300)) + before = query.get("before") + before_message_id = int(before[0]) if before else None + payload = { + "channel_id": channel_id, + "messages": load_messages( + account_id, channel_id, limit=limit, before_message_id=before_message_id + ), + "channel": next( + ( + item + for item in list_channels_snapshot(account_id) + if item["channel_id"] == channel_id + ), + None, + ), + } + return self.send_json(payload) + + # ── HEAD ───────────────────────────────────────────────────────────── + def do_HEAD(self) -> None: parsed = urllib.parse.urlparse(self.path) path = parsed.path if path == "/": - return self.serve_file( - WEBUI_DIR / "index.html", "text/html; charset=utf-8", head_only=True - ) + return self.serve_file(WEBUI_DIR / "index.html", "text/html; charset=utf-8", head_only=True) if path == "/viewer": - return self.serve_file( - WEBUI_DIR / "viewer.html", "text/html; charset=utf-8", head_only=True - ) + return self.serve_file(WEBUI_DIR / "viewer.html", "text/html; charset=utf-8", head_only=True) if path in {"/swagger", "/swigger", "/docs"}: - return self.serve_file( - WEBUI_DIR / "swagger.html", - "text/html; charset=utf-8", - head_only=True, - ) + return self.serve_file(WEBUI_DIR / "swagger.html", "text/html; charset=utf-8", head_only=True) if path.startswith("/static/"): relative = path.removeprefix("/static/") return self.serve_static(relative, head_only=True) @@ -1185,12 +1798,14 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): "/api/continuous", "/api/jobs", "/api/channels", + "/api/accounts", "/health", "/health/continuous", "/openapi.json", } or path.startswith("/api/jobs/") or (path.startswith("/api/channels/") and path.endswith("/messages")) + or path.startswith("/api/accounts/") ): self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "application/json; charset=utf-8") @@ -1198,6 +1813,8 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): return return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found") + # ── POST ───────────────────────────────────────────────────────────── + def do_POST(self) -> None: parsed = urllib.parse.urlparse(self.path) path = parsed.path @@ -1205,8 +1822,11 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): if body is None: return self.send_error_json(HTTPStatus.BAD_REQUEST, "Expected JSON body") + # ── Legacy endpoints ──────────────────────────────────────────── if path == "/api/settings/media": value = bool(body.get("value")) + if self.app.legacy_account_id: + return self._handle_account_settings_media(self.app.legacy_account_id, value) job = self.app.job_runner.create_job( "set_scrape_media", "Update media scraping setting", @@ -1224,12 +1844,16 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): if str(item).strip() ] run_all_tracked = bool(body.get("run_all_tracked", True)) - payload = self.app.continuous_manager.update( - enabled=enabled, - interval_minutes=interval_minutes, - channels=channels, - run_all_tracked=run_all_tracked, - ) + if self.app.legacy_account_id: + payload = self.app.continuous_orchestrator.update_for( + account_id=self.app.legacy_account_id, + enabled=enabled, + interval_minutes=interval_minutes, + channels=channels, + run_all_tracked=run_all_tracked, + ) + else: + payload = {"config": {}, "status": {"running": False, "logs": [], "log_entries": []}} except Exception as exc: return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) return self.send_json(payload) @@ -1238,72 +1862,47 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): api_id = body.get("api_id") api_hash = str(body.get("api_hash", "")).strip() if not api_id or not api_hash: - return self.send_error_json( - HTTPStatus.BAD_REQUEST, "api_id and api_hash are required" - ) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "api_id and api_hash are required") try: - payload = self.app.auth_manager.save_credentials(int(api_id), api_hash) + state = load_state() + state["api_id"] = int(api_id) + state["api_hash"] = api_hash + save_state(state) + payload = self._legacy_auth_snapshot() except Exception as exc: return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) return self.send_json(payload) if path == "/api/auth/qr/start": - try: - payload = self.app.auth_manager.start_qr_login() - except Exception as exc: - return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) - return self.send_json(payload) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "Legacy auth is not available in multi-account mode. Use /api/accounts/{id}/auth/qr/start instead.") if path == "/api/auth/phone/request": - phone = str(body.get("phone", "")).strip() - if not phone: - return self.send_error_json(HTTPStatus.BAD_REQUEST, "phone is required") - try: - payload = self.app.auth_manager.request_phone_code(phone) - except Exception as exc: - return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) - return self.send_json(payload) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "Legacy auth is not available in multi-account mode. Use /api/accounts/{id}/auth/phone/request instead.") if path == "/api/auth/phone/submit": - code = str(body.get("code", "")).strip() - if not code: - return self.send_error_json(HTTPStatus.BAD_REQUEST, "code is required") - try: - payload = self.app.auth_manager.submit_phone_code(code) - except Exception as exc: - return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) - return self.send_json(payload) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "Legacy auth is not available in multi-account mode. Use /api/accounts/{id}/auth/phone/submit instead.") if path == "/api/auth/password": - password = str(body.get("password", "")).strip() - if not password: - return self.send_error_json( - HTTPStatus.BAD_REQUEST, "password is required" - ) - try: - payload = self.app.auth_manager.submit_password(password) - except Exception as exc: - return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) - return self.send_json(payload) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "Legacy auth is not available in multi-account mode. Use /api/accounts/{id}/auth/password instead.") if path == "/api/channels/add": channel_id = str(body.get("channel_id", "")).strip() if not channel_id: - return self.send_error_json( - HTTPStatus.BAD_REQUEST, "channel_id is required" - ) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required") + if self.app.legacy_account_id: + return self._handle_account_channel_add(self.app.legacy_account_id, channel_id, body.get("name")) state = load_state() if channel_id not in state["channels"]: state["channels"][channel_id] = 0 if body.get("name"): - state.setdefault("channel_names", {})[channel_id] = str( - body["name"] - ).strip() + state.setdefault("channel_names", {})[channel_id] = str(body["name"]).strip() save_state(state) return self.send_json({"ok": True, "channel_id": channel_id}) if path == "/api/channels/remove": channel_id = str(body.get("channel_id", "")).strip() + if self.app.legacy_account_id: + return self._handle_account_channel_remove(self.app.legacy_account_id, channel_id) state = load_state() existed = channel_id in state.get("channels", {}) state.get("channels", {}).pop(channel_id, None) @@ -1312,77 +1911,405 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): if path == "/api/jobs/scrape": channel_id = body.get("channel_id") + payload = {} + account_id = self.app.legacy_account_id + if account_id: + payload["account_id"] = account_id if channel_id: + payload["channel_id"] = str(channel_id) job = self.app.job_runner.create_job( "scrape_channel", - f"Scrape channel {channel_id}", - {"channel_id": str(channel_id)}, + f"Scrape channel {channel_id}" + (f" [{account_id}]" if account_id else ""), + payload, ) else: job = self.app.job_runner.create_job( "scrape_all", - "Scrape all tracked channels", - {}, + "Scrape all tracked channels" + (f" [{account_id}]" if account_id else ""), + payload, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) if path == "/api/jobs/export": + payload = {} + if self.app.legacy_account_id: + payload["account_id"] = self.app.legacy_account_id job = self.app.job_runner.create_job( "export_all", "Export all tracked channels", - {}, + payload, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) if path == "/api/jobs/export-channel": channel_id = str(body.get("channel_id", "")).strip() if not channel_id: - return self.send_error_json( - HTTPStatus.BAD_REQUEST, "channel_id is required" - ) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required") + payload = {"channel_id": channel_id} + if self.app.legacy_account_id: + payload["account_id"] = self.app.legacy_account_id job = self.app.job_runner.create_job( "export_channel", f"Export channel {channel_id}", - {"channel_id": channel_id}, + payload, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) if path == "/api/jobs/rescrape-media": channel_id = str(body.get("channel_id", "")).strip() if not channel_id: - return self.send_error_json( - HTTPStatus.BAD_REQUEST, "channel_id is required" - ) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required") + payload = {"channel_id": channel_id} + if self.app.legacy_account_id: + payload["account_id"] = self.app.legacy_account_id job = self.app.job_runner.create_job( "rescrape_media", f"Rescrape media for {channel_id}", - {"channel_id": channel_id}, + payload, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) if path == "/api/jobs/fix-media": channel_id = str(body.get("channel_id", "")).strip() if not channel_id: - return self.send_error_json( - HTTPStatus.BAD_REQUEST, "channel_id is required" - ) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required") + payload = {"channel_id": channel_id} + if self.app.legacy_account_id: + payload["account_id"] = self.app.legacy_account_id job = self.app.job_runner.create_job( "fix_missing_media", f"Fix missing media for {channel_id}", - {"channel_id": channel_id}, + payload, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) if path == "/api/jobs/refresh-dialogs": + payload = {} + if self.app.legacy_account_id: + payload["account_id"] = self.app.legacy_account_id job = self.app.job_runner.create_job( "refresh_dialogs", "Refresh Telegram dialogs list", - {}, + payload, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) + # ── /api/accounts/* POST routes ───────────────────────────────── + if path == "/api/accounts": + return self._handle_post_accounts_create(body) + if path.startswith("/api/accounts/"): + return self._handle_post_account(path, body) + return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found") + def _handle_post_accounts_create(self, body: Dict[str, Any]) -> None: + account_id = str(body.get("account_id", "")).strip() + if not account_id: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "account_id is required") + if not account_id.replace("-", "").replace("_", "").isalnum(): + return self.send_error_json(HTTPStatus.BAD_REQUEST, "account_id must be alphanumeric (dashes and underscores allowed)") + + existing = list_accounts(DATA_DIR) + if account_id in existing: + return self.send_error_json(HTTPStatus.BAD_REQUEST, f"Account '{account_id}' already exists") + + label = str(body.get("label", account_id)).strip() + api_id = body.get("api_id") + api_hash = str(body.get("api_hash", "")).strip() + + def mutate_global(state: Dict[str, Any]) -> None: + accounts = state.setdefault("accounts", []) + if account_id not in accounts: + accounts.append(account_id) + + get_global_store(DATA_DIR).update(mutate_global) + + store = get_account_store(DATA_DIR, account_id) + init_state = { + "label": label, + "api_id": int(api_id) if api_id else None, + "api_hash": api_hash if api_id else None, + "channels": {}, + "channel_names": {}, + "scrape_media": True, + "forwarding_rules": [], + "continuous_scraping": { + "enabled": True, + "interval_minutes": 1, + "channels": [], + "run_all_tracked": True, + }, + } + store.save(init_state) + self.app.continuous_orchestrator.add_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: + rest = path[len("/api/accounts/"):] + parts = rest.split("/") + account_id = urllib.parse.unquote(parts[0]) + sub = parts[1:] + + if not sub: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "Missing endpoint") + if not self._account_exists_or_404(account_id): + return + if sub == ["auth", "credentials"]: + return self._handle_post_account_auth_credentials(account_id, body) + if sub == ["auth", "qr", "start"]: + return self._handle_post_account_auth_qr_start(account_id) + if sub == ["auth", "phone", "request"]: + return self._handle_post_account_auth_phone_request(account_id, body) + if sub == ["auth", "phone", "submit"]: + return self._handle_post_account_auth_phone_submit(account_id, body) + if sub == ["auth", "password"]: + return self._handle_post_account_auth_password(account_id, body) + if sub == ["channels", "add"]: + return self._handle_account_channel_add(account_id, str(body.get("channel_id", "")).strip(), body.get("name")) + if sub == ["channels", "remove"]: + return self._handle_account_channel_remove(account_id, str(body.get("channel_id", "")).strip()) + if sub == ["jobs", "scrape"]: + return self._handle_account_job_scrape(account_id, body) + if sub == ["jobs", "export"]: + return self._handle_account_job_export(account_id) + if sub == ["jobs", "export-channel"]: + return self._handle_account_job_export_channel(account_id, body) + if sub == ["jobs", "rescrape-media"]: + return self._handle_account_job_rescrape_media(account_id, body) + if sub == ["jobs", "fix-media"]: + return self._handle_account_job_fix_media(account_id, body) + if sub == ["jobs", "refresh-dialogs"]: + return self._handle_account_job_refresh_dialogs(account_id) + if sub == ["settings", "media"]: + return self._handle_account_settings_media(account_id, bool(body.get("value"))) + if sub == ["continuous"]: + return self._handle_account_continuous(account_id, body) + return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found") + + def _account_exists_or_404(self, account_id: str) -> bool: + if not account_id or not account_exists(DATA_DIR, account_id): + self.send_error_json(HTTPStatus.NOT_FOUND, f"Account '{account_id}' not found") + return False + return True + + def _handle_post_account_auth_credentials(self, account_id: str, body: Dict[str, Any]) -> None: + api_id = body.get("api_id") + api_hash = str(body.get("api_hash", "")).strip() + if not api_id or not api_hash: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "api_id and api_hash are required") + try: + payload = self.app.auth_manager.save_credentials(account_id, int(api_id), api_hash) + except Exception as exc: + return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + return self.send_json(payload) + + def _handle_post_account_auth_qr_start(self, account_id: str) -> None: + try: + payload = self.app.auth_manager.start_qr_login(account_id) + except Exception as exc: + return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + return self.send_json(payload) + + def _handle_post_account_auth_phone_request(self, account_id: str, body: Dict[str, Any]) -> None: + phone = str(body.get("phone", "")).strip() + if not phone: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "phone is required") + try: + payload = self.app.auth_manager.request_phone_code(account_id, phone) + except Exception as exc: + return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + return self.send_json(payload) + + def _handle_post_account_auth_phone_submit(self, account_id: str, body: Dict[str, Any]) -> None: + code = str(body.get("code", "")).strip() + if not code: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "code is required") + try: + payload = self.app.auth_manager.submit_phone_code(account_id, code) + except Exception as exc: + return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + return self.send_json(payload) + + def _handle_post_account_auth_password(self, account_id: str, body: Dict[str, Any]) -> None: + password = str(body.get("password", "")).strip() + if not password: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "password is required") + try: + payload = self.app.auth_manager.submit_password(account_id, password) + except Exception as exc: + return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + return self.send_json(payload) + + def _handle_account_channel_add(self, account_id: str, channel_id: str, name: Optional[str]) -> None: + if not channel_id: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required") + store = get_account_store(DATA_DIR, account_id) + def mutate(state: Dict[str, Any]) -> None: + if channel_id not in state.setdefault("channels", {}): + state["channels"][channel_id] = 0 + if name: + state.setdefault("channel_names", {})[channel_id] = str(name).strip() + store.update(mutate) + return self.send_json({"ok": True, "channel_id": channel_id}) + + def _handle_account_channel_remove(self, account_id: str, channel_id: str) -> None: + if not channel_id: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required") + store = get_account_store(DATA_DIR, account_id) + existed = [False] + def mutate(state: Dict[str, Any]) -> None: + chans = state.get("channels", {}) + if channel_id in chans: + existed[0] = True + del chans[channel_id] + state.get("channel_names", {}).pop(channel_id, None) + store.update(mutate) + 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: + channel_id = body.get("channel_id") + if channel_id: + job = self.app.job_runner.create_job( + "scrape_channel", + f"Scrape channel {channel_id} [{account_id}]", + {"channel_id": str(channel_id), "account_id": account_id}, + ) + else: + job = self.app.job_runner.create_job( + "scrape_all", + f"Scrape all tracked channels [{account_id}]", + {"account_id": account_id}, + ) + return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) + + def _handle_account_job_export(self, account_id: str) -> None: + job = self.app.job_runner.create_job( + "export_all", + f"Export all tracked channels [{account_id}]", + {"account_id": account_id}, + ) + return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) + + def _handle_account_job_export_channel(self, account_id: str, body: Dict[str, Any]) -> None: + channel_id = str(body.get("channel_id", "")).strip() + if not channel_id: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required") + job = self.app.job_runner.create_job( + "export_channel", + f"Export channel {channel_id} [{account_id}]", + {"channel_id": channel_id, "account_id": account_id}, + ) + return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) + + def _handle_account_job_rescrape_media(self, account_id: str, body: Dict[str, Any]) -> None: + channel_id = str(body.get("channel_id", "")).strip() + if not channel_id: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required") + job = self.app.job_runner.create_job( + "rescrape_media", + f"Rescrape media for {channel_id} [{account_id}]", + {"channel_id": channel_id, "account_id": account_id}, + ) + return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) + + def _handle_account_job_fix_media(self, account_id: str, body: Dict[str, Any]) -> None: + channel_id = str(body.get("channel_id", "")).strip() + if not channel_id: + return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required") + job = self.app.job_runner.create_job( + "fix_missing_media", + f"Fix missing media for {channel_id} [{account_id}]", + {"channel_id": channel_id, "account_id": account_id}, + ) + return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) + + def _handle_account_job_refresh_dialogs(self, account_id: str) -> None: + job = self.app.job_runner.create_job( + "refresh_dialogs", + f"Refresh Telegram dialogs [{account_id}]", + {"account_id": account_id}, + ) + return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) + + def _handle_account_settings_media(self, account_id: str, value: bool) -> None: + store = get_account_store(DATA_DIR, account_id) + def mutate(state: Dict[str, Any]) -> None: + state["scrape_media"] = value + store.update(mutate) + job = self.app.job_runner.create_job( + "set_scrape_media", + f"Update media scraping setting [{account_id}]", + {"value": value, "account_id": account_id}, + ) + return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) + + def _handle_account_continuous(self, account_id: str, body: Dict[str, Any]) -> None: + try: + enabled = bool(body.get("enabled")) + interval_minutes = int(body.get("interval_minutes", 1)) + channels = [ + str(item).strip() + for item in body.get("channels", []) + if str(item).strip() + ] + run_all_tracked = bool(body.get("run_all_tracked", True)) + payload = self.app.continuous_orchestrator.update_for( + account_id=account_id, + enabled=enabled, + interval_minutes=interval_minutes, + channels=channels, + run_all_tracked=run_all_tracked, + ) + except Exception as exc: + return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc)) + return self.send_json(payload) + + # ── DELETE ─────────────────────────────────────────────────────────── + + def do_DELETE(self) -> None: + parsed = urllib.parse.urlparse(self.path) + path = parsed.path + + if path.startswith("/api/accounts/"): + rest = path[len("/api/accounts/"):] + parts = rest.split("/") + account_id = urllib.parse.unquote(parts[0]) + if len(parts) == 1: + return self._handle_delete_account(account_id) + return self.send_error_json(HTTPStatus.BAD_REQUEST, "Invalid account path") + + return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found") + + def _handle_delete_account(self, account_id: str) -> None: + ids = list_accounts(DATA_DIR) + if account_id not in ids: + return self.send_error_json(HTTPStatus.NOT_FOUND, f"Account '{account_id}' not found") + + self.app.continuous_orchestrator.remove_account(account_id) + self.app.auth_manager.delete_account(account_id) + + def mutate_global(state: Dict[str, Any]) -> None: + accounts = state.setdefault("accounts", []) + if account_id in accounts: + accounts.remove(account_id) + + get_global_store(DATA_DIR).update(mutate_global) + + acc_dir = account_data_dir(DATA_DIR, account_id) + session_file = Path(account_session_path(SESSION_DIR, account_id)) + if acc_dir.exists(): + import shutil + shutil.rmtree(str(acc_dir), ignore_errors=True) + if session_file.exists(): + try: + session_file.unlink() + except OSError: + pass + + return self.send_json({"ok": True, "account_id": account_id}) + + # ── Helpers ───────────────────────────────────────────────────────── + def read_json_body(self) -> Optional[Dict[str, Any]]: length = int(self.headers.get("Content-Length", "0")) if length <= 0: @@ -1401,9 +2328,7 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): return self.send_error_json(HTTPStatus.FORBIDDEN, "Forbidden") if not file_path.exists() or not file_path.is_file(): return self.send_error_json(HTTPStatus.NOT_FOUND, "File not found") - mime_type = ( - mimetypes.guess_type(str(file_path))[0] or "application/octet-stream" - ) + mime_type = mimetypes.guess_type(str(file_path))[0] or "application/octet-stream" return self.serve_file(file_path, mime_type, head_only=head_only) def serve_media(self, relative_path: str, head_only: bool = False) -> None: @@ -1415,14 +2340,10 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): return self.send_error_json(HTTPStatus.FORBIDDEN, "Forbidden") if not file_path.exists() or not file_path.is_file(): return self.send_error_json(HTTPStatus.NOT_FOUND, "Media file not found") - mime_type = ( - mimetypes.guess_type(str(file_path))[0] or "application/octet-stream" - ) + mime_type = mimetypes.guess_type(str(file_path))[0] or "application/octet-stream" return self.serve_file(file_path, mime_type, head_only=head_only) - def serve_file( - self, file_path: Path, content_type: str, head_only: bool = False - ) -> None: + def serve_file(self, file_path: Path, content_type: str, head_only: bool = False) -> None: stat = file_path.stat() file_size = stat.st_size last_modified = stat.st_mtime @@ -1432,9 +2353,7 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): try: start, end = self._parse_range(range_header, file_size) except ValueError: - self.send_error_json( - HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE, "Invalid range" - ) + self.send_error_json(HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE, "Invalid range") return content_length = end - start + 1 self.send_response(HTTPStatus.PARTIAL_CONTENT) @@ -1509,16 +2428,30 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): ) +# ── Server ─────────────────────────────────────────────────────────────── + class TelegramScraperWebServer(ThreadingHTTPServer): def __init__(self, server_address: tuple[str, int]): super().__init__(server_address, TelegramScraperRequestHandler) self.job_runner = JobRunner() self.auth_manager = TelegramAuthManager() - self.continuous_manager = ContinuousScrapeManager() - if START_CONTINUOUS and self.continuous_manager.snapshot()["config"].get( - "enabled", True - ): - self.continuous_manager.start() + self.continuous_orchestrator = ContinuousScrapeOrchestrator() + self.legacy_account_id: Optional[str] = None + self._detect_legacy_account() + + if START_CONTINUOUS: + self.continuous_orchestrator.start_all() + + def _detect_legacy_account(self) -> None: + ids = list_accounts(DATA_DIR) + if not ids: + legacy = load_state() + if legacy.get("api_id") and legacy.get("api_hash"): + import app_state as as_mod + as_mod.migrate_legacy_state(DATA_DIR, SESSION_DIR) + ids = list_accounts(DATA_DIR) + if ids and len(ids) == 1: + self.legacy_account_id = ids[0] def run_server(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT) -> None: @@ -1530,7 +2463,7 @@ def run_server(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT) -> None: return logger.warning("Shutdown signal received, shutting down gracefully...") shutdown_event.set() - server.continuous_manager.stop() + server.continuous_orchestrator.stop_all() server.auth_manager.shutdown() server.job_runner.shutdown()