import asyncio 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, Callable, Dict, List, Optional 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")) 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() 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 and 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 == "export_all": await scraper.export_data() 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(), } 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.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 == "/api/dashboard": return self.send_json(dashboard_payload(self.app.job_runner)) 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.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/jobs", "/api/channels"} 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/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/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() 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()