import asyncio import base64 import contextlib import io import json import mimetypes import os import posixpath import queue import sqlite3 import threading import time import traceback import urllib.parse from dataclasses import dataclass, field from datetime import datetime, timezone from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any, Dict, List, Optional import qrcode import qrcode.image.svg from telethon import TelegramClient from telethon.errors import SessionPasswordNeededError BASE_DIR = Path(__file__).resolve().parent DATA_DIR = BASE_DIR / "data" WEBUI_DIR = BASE_DIR / "webui" STATE_FILE = DATA_DIR / "state.json" DEFAULT_HOST = os.environ.get("TELEGRAM_SCRAPER_HOST", "0.0.0.0") DEFAULT_PORT = int(os.environ.get("TELEGRAM_SCRAPER_PORT", "8080")) SESSION_DIR = BASE_DIR / "session" SESSION_DIR.mkdir(exist_ok=True) def utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat() def load_state() -> Dict[str, Any]: if STATE_FILE.exists(): with STATE_FILE.open("r", encoding="utf-8") as handle: return json.load(handle) return { "api_id": None, "api_hash": None, "channels": {}, "channel_names": {}, "scrape_media": True, "forwarding_rules": [], } def save_state(state: Dict[str, Any]) -> None: DATA_DIR.mkdir(parents=True, exist_ok=True) with STATE_FILE.open("w", encoding="utf-8") as handle: json.dump(state, handle, ensure_ascii=False, indent=2) 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]: if not media_path: return None suffix = Path(media_path).suffix.lower() if suffix in {".jpg", ".jpeg", ".png", ".webp", ".gif"}: return "image" if suffix in {".mp4", ".webm", ".mov", ".m4v"}: return "video" if suffix in {".mp3", ".ogg", ".wav", ".m4a"}: return "audio" if media_type == "MessageMediaPhoto": return "image" return "file" def normalize_media_url(media_path: Optional[str]) -> Optional[str]: if not media_path: return None path = Path(media_path) if path.is_absolute(): try: relative = path.resolve().relative_to(DATA_DIR.resolve()) except ValueError: return None else: normalized = Path(media_path) if normalized.parts and normalized.parts[0] == "data": normalized = Path(*normalized.parts[1:]) relative = normalized 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 = { "message_count": 0, "last_date": None, "first_date": None, "media_count": 0, "has_database": db_path.exists(), } if not db_path.exists(): return summary conn = sqlite3.connect(str(db_path)) try: cursor = conn.cursor() cursor.execute( "SELECT COUNT(*), MIN(date), MAX(date), " "SUM(CASE WHEN media_type IS NOT NULL AND media_type != 'MessageMediaWebPage' THEN 1 ELSE 0 END) " "FROM messages" ) row = cursor.fetchone() or (0, None, None, 0) summary.update( { "message_count": row[0] or 0, "first_date": row[1], "last_date": row[2], "media_count": row[3] or 0, } ) return summary finally: conn.close() 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() channels: List[Dict[str, Any]] = [] for channel_id, last_message_id in state.get("channels", {}).items(): summary = database_summary(channel_id) channels.append( { "channel_id": channel_id, "name": channel_display_name(state, channel_id), "last_message_id": last_message_id, **summary, } ) channels.sort( key=lambda item: ( item["last_date"] or "", item["message_count"], ), reverse=True, ) return channels def load_messages( channel_id: str, limit: int = 120, before_message_id: Optional[int] = None ) -> List[Dict[str, Any]]: db_path = channel_db_path(channel_id) if not db_path.exists(): return [] conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row try: query = ( "SELECT message_id, date, sender_id, first_name, last_name, username, message, " "media_type, media_path, reply_to, post_author, views, forwards, reactions " "FROM messages " ) params: List[Any] = [] if before_message_id is not None: query += "WHERE message_id < ? " params.append(before_message_id) query += "ORDER BY message_id DESC LIMIT ?" params.append(limit) rows = conn.execute(query, params).fetchall() finally: conn.close() messages: List[Dict[str, Any]] = [] for row in reversed(rows): sender_name = " ".join( part for part in [row["first_name"], row["last_name"]] if part ).strip() if not sender_name: sender_name = row["username"] or row["post_author"] or channel_id messages.append( { "message_id": row["message_id"], "date": row["date"], "sender_id": row["sender_id"], "sender_name": sender_name, "username": row["username"], "text": row["message"] or "", "media_type": row["media_type"], "media_path": row["media_path"], "media_url": normalize_media_url(row["media_path"]), "media_kind": guess_media_kind(row["media_path"], row["media_type"]), "reply_to": row["reply_to"], "post_author": row["post_author"], "views": row["views"], "forwards": row["forwards"], "reactions": row["reactions"], } ) return messages @dataclass class Job: job_id: str job_type: str title: str payload: Dict[str, Any] status: str = "queued" created_at: str = field(default_factory=utc_now_iso) started_at: Optional[str] = None finished_at: Optional[str] = None logs: str = "" error: Optional[str] = None def to_dict(self) -> Dict[str, Any]: return { "job_id": self.job_id, "job_type": self.job_type, "title": self.title, "payload": self.payload, "status": self.status, "created_at": self.created_at, "started_at": self.started_at, "finished_at": self.finished_at, "logs": self.logs, "error": self.error, } class JobRunner: def __init__(self) -> None: self.jobs: Dict[str, Job] = {} self.job_order: List[str] = [] self.queue: "queue.Queue[Job]" = queue.Queue() self.lock = threading.Lock() self.worker = threading.Thread(target=self._run, daemon=True) self.worker.start() def create_job(self, job_type: str, title: str, payload: Dict[str, Any]) -> Job: job_id = f"job-{int(time.time() * 1000)}" job = Job(job_id=job_id, job_type=job_type, title=title, payload=payload) with self.lock: self.jobs[job_id] = job self.job_order.insert(0, job_id) self.job_order = self.job_order[:20] self.queue.put(job) return job def recent_jobs(self) -> List[Dict[str, Any]]: with self.lock: return [self.jobs[job_id].to_dict() for job_id in self.job_order] def get_job(self, job_id: str) -> Optional[Dict[str, Any]]: with self.lock: job = self.jobs.get(job_id) return job.to_dict() if job else None def _run(self) -> None: while True: job = self.queue.get() self._execute(job) self.queue.task_done() def _execute(self, job: Job) -> None: with self.lock: job.status = "running" job.started_at = utc_now_iso() buffer = io.StringIO() try: with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer): run_job(job.job_type, job.payload) with self.lock: job.status = "done" except Exception as exc: with self.lock: job.status = "failed" job.error = str(exc) traceback.print_exc(file=buffer) finally: with self.lock: job.logs = buffer.getvalue() job.finished_at = utc_now_iso() 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.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(), } def _run_loop(self) -> None: asyncio.set_event_loop(self.loop) self.loop.run_forever() def _run(self, coro): 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 snapshot(self) -> Dict[str, Any]: auth = auth_status() with self.lock: snapshot = dict(self.state) 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() 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 _make_qr_image(self, qr_url: str) -> str: qr = qrcode.QRCode(border=1, box_size=8) qr.add_data(qr_url) qr.make(fit=True) image = qr.make_image(image_factory=qrcode.image.svg.SvgPathImage) buffer = io.BytesIO() image.save(buffer) 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() self._set_state( phase="authorized", status="authorized", details="Telegram session is authorized.", qr_url=None, qr_image=None, phone=None, ) except SessionPasswordNeededError: self._set_state( phase="password_required", status="password_required", details="Two-factor authentication is enabled. Enter your Telegram password.", ) except Exception as exc: self._set_state( phase="error", status="error", details=f"QR login failed: {exc}", qr_url=None, qr_image=None, ) async def _start_qr_login(self) -> 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() 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() sent = await client.send_code_request(phone) self.phone = phone self.phone_code_hash = sent.phone_code_hash self._set_state( phase="code_required", status="code_required", details=f"Code sent to {phone}. Enter it below.", phone=phone, ) return self.snapshot() def request_phone_code(self, phone: str) -> Dict[str, Any]: return self._run(self._request_phone_code(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: raise RuntimeError("Request a phone code first.") try: await client.sign_in( phone=self.phone, code=code, phone_code_hash=self.phone_code_hash ) self._set_state( phase="authorized", status="authorized", details="Telegram session is authorized.", qr_url=None, qr_image=None, ) except SessionPasswordNeededError: self._set_state( phase="password_required", status="password_required", details="Two-factor authentication is enabled. Enter your Telegram password.", ) return self.snapshot() def submit_phone_code(self, code: str) -> Dict[str, Any]: return self._run(self._submit_phone_code(code)) async def _submit_password(self, password: str) -> Dict[str, Any]: client = await self._get_client() await client.sign_in(password=password) self._set_state( phase="authorized", status="authorized", details="Telegram session is authorized.", qr_url=None, qr_image=None, ) return self.snapshot() def submit_password(self, password: str) -> Dict[str, Any]: return self._run(self._submit_password(password)) class ContinuousScrapeManager: def __init__(self) -> None: self.lock = threading.RLock() self.thread: Optional[threading.Thread] = None self.stop_event = threading.Event() self.config: Dict[str, Any] = { "enabled": True, "interval_minutes": 1, "channels": [], "run_all_tracked": True, } self.status: Dict[str, Any] = { "running": False, "last_started_at": None, "last_finished_at": None, "last_iteration_at": None, "last_error": None, "logs": [], "log_entries": [], } 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) log_entries.append(entry) if len(logs) > 300: del logs[:-300] if len(log_entries) > 300: del log_entries[:-300] def snapshot(self) -> Dict[str, Any]: with self.lock: return { "config": dict(self.config), "status": { **self.status, "logs": list(self.status.get("logs", [])), "log_entries": list(self.status.get("log_entries", [])), }, } def update( self, enabled: bool, interval_minutes: int, channels: List[str], run_all_tracked: bool, ) -> Dict[str, Any]: interval_minutes = max(1, int(interval_minutes)) with self.lock: self.config = { "enabled": bool(enabled), "interval_minutes": interval_minutes, "channels": channels, "run_all_tracked": bool(run_all_tracked), } if enabled: self.start() else: self.stop() return self.snapshot() def start(self) -> None: with self.lock: if self.status["running"]: self._log("Continuous scraping is already running.", "warn") return self.stop_event.clear() self.status["running"] = True self.status["last_started_at"] = utc_now_iso() self.status["last_error"] = None self.thread = threading.Thread(target=self._run_loop, daemon=True) self.thread.start() self._log("Continuous scraping started.", "info") def stop(self) -> None: self.stop_event.set() with self.lock: was_running = self.status["running"] self.status["running"] = False if was_running: self._log("Continuous scraping stop requested.", "warn") def _resolve_channels(self) -> List[str]: state = load_state() with self.lock: run_all_tracked = self.config["run_all_tracked"] configured = list(self.config["channels"]) if run_all_tracked: return list(state.get("channels", {}).keys()) tracked = set(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"] if not channels: self._log("No channels configured for continuous scraping.", "warn") else: 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}) output = buffer.getvalue().strip() if output: for line in output.splitlines(): self._log(line) with self.lock: self.status["last_iteration_at"] = utc_now_iso() self.status["last_finished_at"] = utc_now_iso() self.status["last_error"] = None self._log("Iteration finished.", "success") except Exception as exc: output = buffer.getvalue().strip() if output: for line in output.splitlines(): self._log(line) self._log(f"Iteration failed: {exc}", "error") with self.lock: self.status["last_error"] = str(exc) sleep_seconds = max(5, interval_minutes * 60) self._log(f"Sleeping for {interval_minutes} minute(s).", "debug") interrupted = self.stop_event.wait(timeout=sleep_seconds) if interrupted: break with self.lock: self.status["running"] = False self._log("Continuous scraping stopped.", "warn") def import_scraper_class(): from telegram_scraper_with_forwarding import OptimizedTelegramScraper return OptimizedTelegramScraper def run_job(job_type: str, payload: Dict[str, Any]) -> None: if job_type == "set_scrape_media": state = load_state() state["scrape_media"] = bool(payload["value"]) save_state(state) print(f"Media scraping set to {state['scrape_media']}") return async def _async_job() -> None: ScraperClass = import_scraper_class() scraper = ScraperClass() 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." ) try: if job_type == "scrape_channel": channel_id = payload["channel_id"] state = scraper.load_state() offset = int(state.get("channels", {}).get(channel_id, 0) or 0) await scraper.scrape_channel(channel_id, offset) elif job_type == "scrape_all": channels = list(scraper.state.get("channels", {}).keys()) for channel_id in channels: offset = int( scraper.state.get("channels", {}).get(channel_id, 0) or 0 ) await scraper.scrape_channel(channel_id, offset) elif job_type == "scrape_selected": channels = [ str(channel_id) for channel_id in payload.get("channels", []) ] for channel_id in channels: offset = int( scraper.state.get("channels", {}).get(channel_id, 0) or 0 ) await scraper.scrape_channel(channel_id, offset) elif job_type == "export_all": await scraper.export_data() elif job_type == "export_channel": channel_id = payload["channel_id"] scraper.export_to_csv(channel_id) scraper.export_to_json(channel_id) elif job_type == "rescrape_media": await scraper.rescrape_media(payload["channel_id"]) elif job_type == "fix_missing_media": await scraper.fix_missing_media(payload["channel_id"]) elif job_type == "refresh_dialogs": await scraper.list_channels() else: raise RuntimeError(f"Unsupported job type: {job_type}") finally: scraper.close_db_connections() if scraper.client: await scraper.client.disconnect() asyncio.run(_async_job()) def auth_status() -> Dict[str, Any]: state = load_state() status = { "has_api_credentials": bool(state.get("api_id") and state.get("api_hash")), "telethon_available": False, "session_ready": False, "status": "read_only", "details": "", } try: import_scraper_class() status["telethon_available"] = True except Exception as exc: status["details"] = f"Python deps are missing: {exc}" return status session_file = BASE_DIR / "session" / "session.session" status["session_ready"] = 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." ) 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." return status def dashboard_payload(job_runner: JobRunner) -> Dict[str, Any]: state = load_state() channels = list_channels_snapshot() return { "state": { "scrape_media": bool(state.get("scrape_media", True)), "channel_count": len(state.get("channels", {})), "forwarding_rules": state.get("forwarding_rules", []), }, "auth": auth_status(), "channels": channels, "jobs": job_runner.recent_jobs(), } def openapi_payload() -> Dict[str, Any]: json_response = { "200": { "description": "JSON response", "content": {"application/json": {"schema": {"type": "object"}}}, } } accepted_response = { "202": { "description": "Job accepted", "content": {"application/json": {"schema": {"type": "object"}}}, } } error_response = {"400": {"description": "Bad request"}} def json_body(properties: Dict[str, Any], required: Optional[List[str]] = None): return { "required": True, "content": { "application/json": { "schema": { "type": "object", "properties": properties, "required": required or [], "example": { key: value.get("example") for key, value in properties.items() if "example" in value }, } } }, } return { "openapi": "3.0.3", "info": { "title": "Telegram Scraper Web UI API", "version": "0.1.0", "description": "Local endpoints used by the Telegram Scraper web interface.", }, "servers": [{"url": "/"}], "paths": { "/api/dashboard": { "get": { "summary": "Dashboard snapshot", "description": "Returns state summary, channels, auth status, and recent jobs.", "responses": json_response, } }, "/api/auth": { "get": { "summary": "Authentication status", "description": "Returns saved credential flags, login phase, and QR login data.", "responses": json_response, } }, "/api/auth/credentials": { "post": { "summary": "Save Telegram API credentials", "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/auth/qr/start": { "post": { "summary": "Start QR login", "requestBody": json_body({}), "responses": {**json_response, **error_response}, } }, "/api/auth/phone/request": { "post": { "summary": "Request phone login code", "requestBody": json_body( {"phone": {"type": "string", "example": "+1234567890"}}, ["phone"], ), "responses": {**json_response, **error_response}, } }, "/api/auth/phone/submit": { "post": { "summary": "Submit phone login code", "requestBody": json_body( {"code": {"type": "string", "example": "12345"}}, ["code"], ), "responses": {**json_response, **error_response}, } }, "/api/auth/password": { "post": { "summary": "Submit 2FA password", "requestBody": json_body( {"password": {"type": "string", "example": "password"}}, ["password"], ), "responses": {**json_response, **error_response}, } }, "/api/continuous": { "get": { "summary": "Continuous scraping status", "responses": json_response, }, "post": { "summary": "Update continuous scraping", "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}, }, }, "/health/continuous": { "get": { "summary": "Continuous scraping health", "responses": json_response, } }, "/api/channels": { "get": { "summary": "List tracked channels", "responses": json_response, } }, "/api/channels/{channel_id}/messages": { "get": { "summary": "List channel messages", "parameters": [ { "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/channels/add": { "post": { "summary": "Track a channel", "requestBody": json_body( { "channel_id": {"type": "string", "example": "-1001234567890"}, "name": {"type": "string", "example": "Research feed"}, }, ["channel_id"], ), "responses": {**json_response, **error_response}, } }, "/api/channels/remove": { "post": { "summary": "Remove a tracked channel", "requestBody": json_body( {"channel_id": {"type": "string", "example": "-1001234567890"}}, ["channel_id"], ), "responses": {**json_response, **error_response}, } }, "/api/settings/media": { "post": { "summary": "Toggle media scraping", "requestBody": json_body( {"value": {"type": "boolean", "example": True}}, ["value"], ), "responses": {**accepted_response, **error_response}, } }, "/api/jobs": { "get": { "summary": "Recent jobs", "responses": json_response, } }, "/api/jobs/{job_id}": { "get": { "summary": "Job details", "parameters": [ { "name": "job_id", "in": "path", "required": True, "schema": {"type": "string"}, } ], "responses": {**json_response, "404": {"description": "Job not found"}}, } }, "/api/jobs/scrape": { "post": { "summary": "Scrape one channel or all tracked channels", "requestBody": json_body( {"channel_id": {"type": "string", "example": "-1001234567890"}} ), "responses": accepted_response, } }, "/api/jobs/export": { "post": { "summary": "Export all tracked channels", "requestBody": json_body({}), "responses": accepted_response, } }, "/api/jobs/export-channel": { "post": { "summary": "Export one channel", "requestBody": json_body( {"channel_id": {"type": "string", "example": "-1001234567890"}}, ["channel_id"], ), "responses": {**accepted_response, **error_response}, } }, "/api/jobs/rescrape-media": { "post": { "summary": "Rescrape media for a channel", "requestBody": json_body( {"channel_id": {"type": "string", "example": "-1001234567890"}}, ["channel_id"], ), "responses": {**accepted_response, **error_response}, } }, "/api/jobs/fix-media": { "post": { "summary": "Fix missing media for a channel", "requestBody": json_body( {"channel_id": {"type": "string", "example": "-1001234567890"}}, ["channel_id"], ), "responses": {**accepted_response, **error_response}, } }, "/api/jobs/refresh-dialogs": { "post": { "summary": "Refresh Telegram dialogs", "requestBody": json_body({}), "responses": accepted_response, } }, "/openapi.json": { "get": { "summary": "OpenAPI document", "responses": json_response, } }, }, } class TelegramScraperRequestHandler(BaseHTTPRequestHandler): server_version = "TelegramScraperWebUI/0.1" @property def app(self) -> "TelegramScraperWebServer": return self.server # type: ignore[return-value] def do_GET(self) -> None: parsed = urllib.parse.urlparse(self.path) path = parsed.path query = urllib.parse.parse_qs(parsed.query) 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" ) if path in {"/swagger", "/swigger", "/docs"}: 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) if path.startswith("/media/"): relative = path.removeprefix("/media/") return self.serve_media(relative) if path == "/openapi.json": return self.send_json(openapi_payload()) 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 path == "/api/continuous": return self.send_json(self.app.continuous_manager.snapshot()) if path == "/health/continuous": return self.send_json(self.app.continuous_manager.snapshot()) if path == "/api/jobs": return self.send_json(self.app.job_runner.recent_jobs()) if path.startswith("/api/jobs/"): job_id = path.rsplit("/", 1)[-1] job = self.app.job_runner.get_job(job_id) if not job: return self.send_error_json(HTTPStatus.NOT_FOUND, "Job not found") return self.send_json(job) if path == "/api/channels": return self.send_json(list_channels_snapshot()) if path.startswith("/api/channels/") and path.endswith("/messages"): parts = path.split("/") channel_id = parts[3] 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( channel_id, limit=limit, before_message_id=before_message_id ), "channel": next( ( item for item in list_channels_snapshot() if item["channel_id"] == channel_id ), None, ), } return self.send_json(payload) return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found") 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 ) if path == "/viewer": 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, ) if path.startswith("/static/"): relative = path.removeprefix("/static/") return self.serve_static(relative, head_only=True) if path.startswith("/media/"): relative = path.removeprefix("/media/") return self.serve_media(relative, head_only=True) if ( path in { "/api/dashboard", "/api/auth", "/api/continuous", "/api/jobs", "/api/channels", "/health/continuous", "/openapi.json", } or path.startswith("/api/jobs/") or (path.startswith("/api/channels/") and path.endswith("/messages")) ): self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() return return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found") def do_POST(self) -> None: parsed = urllib.parse.urlparse(self.path) path = parsed.path body = self.read_json_body() if body is None: return self.send_error_json(HTTPStatus.BAD_REQUEST, "Expected JSON body") if path == "/api/settings/media": value = bool(body.get("value")) job = self.app.job_runner.create_job( "set_scrape_media", "Update media scraping setting", {"value": value}, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) if path == "/api/continuous": 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_manager.update( 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) if path == "/api/auth/credentials": 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(int(api_id), api_hash) 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) 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) 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) 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) 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" ) 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() 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() state = load_state() existed = channel_id in state.get("channels", {}) state.get("channels", {}).pop(channel_id, None) save_state(state) return self.send_json({"ok": existed, "channel_id": channel_id}) if path == "/api/jobs/scrape": channel_id = body.get("channel_id") if channel_id: job = self.app.job_runner.create_job( "scrape_channel", f"Scrape channel {channel_id}", {"channel_id": str(channel_id)}, ) else: job = self.app.job_runner.create_job( "scrape_all", "Scrape all tracked channels", {}, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) if path == "/api/jobs/export": job = self.app.job_runner.create_job( "export_all", "Export all tracked channels", {}, ) 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" ) job = self.app.job_runner.create_job( "export_channel", f"Export channel {channel_id}", {"channel_id": channel_id}, ) 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" ) job = self.app.job_runner.create_job( "rescrape_media", f"Rescrape media for {channel_id}", {"channel_id": channel_id}, ) 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" ) job = self.app.job_runner.create_job( "fix_missing_media", f"Fix missing media for {channel_id}", {"channel_id": channel_id}, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) if path == "/api/jobs/refresh-dialogs": job = self.app.job_runner.create_job( "refresh_dialogs", "Refresh Telegram dialogs list", {}, ) return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED) return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found") def read_json_body(self) -> Optional[Dict[str, Any]]: length = int(self.headers.get("Content-Length", "0")) if length <= 0: return {} raw = self.rfile.read(length) try: return json.loads(raw.decode("utf-8")) except json.JSONDecodeError: return None def serve_static(self, relative_path: str, head_only: bool = False) -> None: file_path = (WEBUI_DIR / relative_path).resolve() try: file_path.relative_to(WEBUI_DIR.resolve()) except ValueError: 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" ) return self.serve_file(file_path, mime_type, head_only=head_only) def serve_media(self, relative_path: str, head_only: bool = False) -> None: clean = Path(posixpath.normpath(urllib.parse.unquote(relative_path))) file_path = (DATA_DIR / clean).resolve() try: file_path.relative_to(DATA_DIR.resolve()) except ValueError: 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" ) 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: data = file_path.read_bytes() self.send_response(HTTPStatus.OK) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(data))) self.end_headers() if not head_only: self.wfile.write(data) def send_json(self, payload: Any, status: HTTPStatus = HTTPStatus.OK) -> None: raw = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(raw))) self.end_headers() self.wfile.write(raw) def send_error_json(self, status: HTTPStatus, message: str) -> None: self.send_json({"error": message, "status": status}, status=status) def log_message(self, format: str, *args: Any) -> None: return 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() self.continuous_manager.start() def run_server(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT) -> None: server = TelegramScraperWebServer((host, port)) print(f"Telegram Scraper Web UI listening on http://{host}:{port}") print("Press Ctrl+C to stop.") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() if __name__ == "__main__": run_server()