624 lines
23 KiB
Python
624 lines
23 KiB
Python
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()
|