diff --git a/webui_server.py b/webui_server.py index bac2aaf..5f9ce9f 100644 --- a/webui_server.py +++ b/webui_server.py @@ -23,7 +23,16 @@ 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 @@ -57,13 +66,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 +98,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 +134,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 +158,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 +175,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 +275,8 @@ def load_messages( return messages +# ── Job ────────────────────────────────────────────────────────────────── + @dataclass class Job: job_id: str @@ -251,6 +289,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 +303,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 +323,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 +338,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 +358,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 +404,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 +423,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 +455,101 @@ 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: + 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 +559,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 +574,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 +624,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 +686,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 +705,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 +762,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 +798,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 +819,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 +853,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 +937,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 +958,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 +1131,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 +1286,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 +1583,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 +1593,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 +1607,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 +1649,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 +1661,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 +1796,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 +1811,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 +1820,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 +1842,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 +1860,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 +1909,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 +2326,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 +2338,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 +2351,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 +2426,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 +2461,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()