diff --git a/app_state.py b/app_state.py index e89fc10..8f02720 100644 --- a/app_state.py +++ b/app_state.py @@ -1,7 +1,8 @@ import json import threading +import time from pathlib import Path -from typing import Any, Callable, Dict +from typing import Any, Callable, Dict, Optional DEFAULT_STATE: Dict[str, Any] = { @@ -24,17 +25,32 @@ class StateStore: def __init__(self, path: Path): self.path = path self.lock = threading.RLock() + self._cache: Optional[Dict[str, Any]] = None + self._cache_time: float = 0 + self._cache_ttl: float = 1.0 def load(self) -> Dict[str, Any]: with self.lock: + now = time.time() + if self._cache is not None and (now - self._cache_time) < self._cache_ttl: + return dict(self._cache) if not self.path.exists(): - return self._default_state() + result = self._default_state() + self._cache = result + self._cache_time = now + return result try: with self.path.open("r", encoding="utf-8") as handle: state = json.load(handle) except (json.JSONDecodeError, OSError): - return self._default_state() - return self._merge_defaults(state) + result = self._default_state() + self._cache = result + self._cache_time = now + return result + result = self._merge_defaults(state) + self._cache = result + self._cache_time = now + return result def save(self, state: Dict[str, Any]) -> None: with self.lock: @@ -44,6 +60,7 @@ class StateStore: json.dump(self._merge_defaults(state), handle, ensure_ascii=False, indent=2) handle.write("\n") tmp_path.replace(self.path) + self._cache = None def update(self, mutator: Callable[[Dict[str, Any]], None]) -> Dict[str, Any]: with self.lock: diff --git a/main.py b/main.py index ab1159e..e622697 100644 --- a/main.py +++ b/main.py @@ -1,7 +1,16 @@ +import logging +import sys + from webui_server import run_server def main(): + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + stream=sys.stdout, + ) run_server() diff --git a/scraper_jobs.py b/scraper_jobs.py index 1ebd546..44ac6b2 100644 --- a/scraper_jobs.py +++ b/scraper_jobs.py @@ -1,8 +1,11 @@ import asyncio +import logging from typing import Any, Dict, List from app_state import StateStore +logger = logging.getLogger(__name__) + class ScraperJobService: def __init__(self, state_store: StateStore): @@ -16,7 +19,7 @@ class ScraperJobService: state["scrape_media"] = value self.state_store.update(mutate) - print(f"Media scraping set to {value}") + logger.info("Media scraping set to %s", value) return asyncio.run(self._run_async(job_type, payload)) diff --git a/webui_server.py b/webui_server.py index fae401b..bd1af75 100644 --- a/webui_server.py +++ b/webui_server.py @@ -3,10 +3,12 @@ import base64 import contextlib import io import json +import logging import mimetypes import os import posixpath import queue +import signal import sqlite3 import threading import time @@ -40,6 +42,8 @@ SESSION_DIR.mkdir(exist_ok=True) STATE_STORE = StateStore(STATE_FILE) SCRAPER_JOBS = ScraperJobService(STATE_STORE) +logger = logging.getLogger(__name__) + def utc_now_iso() -> str: return datetime.now(timezone.utc).isoformat() @@ -238,10 +242,13 @@ class JobRunner: self.job_order: List[str] = [] self.queue: "queue.Queue[Job]" = queue.Queue() self.lock = threading.Lock() + self._shutdown_flag = False 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: + 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) with self.lock: @@ -260,9 +267,36 @@ class JobRunner: job = self.jobs.get(job_id) return job.to_dict() if job else None + def shutdown(self, timeout: float = 10.0) -> None: + self._shutdown_flag = True + running_jobs = [] + with self.lock: + for job_id, job in list(self.jobs.items()): + if job.status == "running": + running_jobs.append(job) + if 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: + still_running = [j for j in running_jobs if j.status == "running"] + if not still_running: + break + time.sleep(0.2) + still_running = [j for j in running_jobs if j.status == "running"] + if still_running: + logger.warning( + "%d job(s) did not finish within %ss timeout", + len(still_running), + timeout, + ) + def _run(self) -> None: - while True: - job = self.queue.get() + while not self._shutdown_flag: + try: + job = self.queue.get(timeout=1) + except queue.Empty: + continue self._execute(job) self.queue.task_done() @@ -484,6 +518,19 @@ class TelegramAuthManager: def submit_password(self, password: str) -> Dict[str, Any]: return self._run(self._submit_password(password)) + def shutdown(self, timeout: float = 5.0) -> None: + async def _disconnect(): + if self.client: + await self.client.disconnect() + if self.client: + try: + future = asyncio.run_coroutine_threadsafe(_disconnect(), self.loop) + future.result(timeout=timeout) + except Exception: + pass + self.loop.call_soon_threadsafe(self.loop.stop) + self.thread.join(timeout=timeout) + class ContinuousScrapeManager: def __init__(self) -> None: @@ -1321,13 +1368,67 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): 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) + stat = file_path.stat() + file_size = stat.st_size + last_modified = stat.st_mtime + + range_header = self.headers.get("Range", "").strip() + if range_header.startswith("bytes="): + try: + start, end = self._parse_range(range_header, file_size) + except ValueError: + self.send_error_json(HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE, "Invalid range") + return + content_length = end - start + 1 + self.send_response(HTTPStatus.PARTIAL_CONTENT) + self.send_header("Content-Type", content_type) + self.send_header("Accept-Ranges", "bytes") + self.send_header("Last-Modified", time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", time.gmtime(last_modified) + )) + self.send_header("Content-Range", f"bytes {start}-{end}/{file_size}") + self.send_header("Content-Length", str(content_length)) + self.end_headers() + if not head_only: + self._write_file_range(file_path, start, content_length) + else: + self.send_response(HTTPStatus.OK) + self.send_header("Content-Type", content_type) + self.send_header("Accept-Ranges", "bytes") + self.send_header("Last-Modified", time.strftime( + "%a, %d %b %Y %H:%M:%S GMT", time.gmtime(last_modified) + )) + self.send_header("Content-Length", str(file_size)) + self.end_headers() + if not head_only: + self._write_file_range(file_path, 0, file_size) + + def _parse_range(self, range_header: str, file_size: int) -> tuple[int, int]: + range_val = range_header.removeprefix("bytes=").strip() + if "-" not in range_val: + raise ValueError("missing dash") + parts = range_val.split("-", 1) + if parts[0] == "": + end = int(parts[1]) + start = max(0, file_size - end) + else: + start = int(parts[0]) + end = int(parts[1]) if parts[1] else file_size - 1 + if start < 0 or start >= file_size or end >= file_size or start > end: + raise ValueError("out of bounds") + return start, end + + def _write_file_range(self, file_path: Path, offset: int, length: int) -> None: + chunk_size = 65536 + with open(file_path, "rb") as f: + f.seek(offset) + remaining = length + while remaining > 0: + chunk = f.read(min(chunk_size, remaining)) + if not chunk: + break + self.wfile.write(chunk) + remaining -= len(chunk) def send_json(self, payload: Any, status: HTTPStatus = HTTPStatus.OK) -> None: raw = json.dumps(payload, ensure_ascii=False).encode("utf-8") @@ -1338,10 +1439,15 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler): self.wfile.write(raw) def send_error_json(self, status: HTTPStatus, message: str) -> None: - self.send_json({"error": message, "status": status}, status=status) + self.send_json({"error": message, "status": int(status)}, status=status) def log_message(self, format: str, *args: Any) -> None: - return + logger.info( + "%s %s — %s", + self.command, + self.path, + args[1] if len(args) > 1 else "-", + ) class TelegramScraperWebServer(ThreadingHTTPServer): @@ -1356,14 +1462,30 @@ class TelegramScraperWebServer(ThreadingHTTPServer): 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.") + shutdown_event = threading.Event() + + def _shutdown(signum=None, frame=None): + if shutdown_event.is_set(): + return + logger.warning("Shutdown signal received, shutting down gracefully...") + shutdown_event.set() + server.continuous_manager.stop() + server.auth_manager.shutdown() + server.job_runner.shutdown() + + signal.signal(signal.SIGTERM, _shutdown) + signal.signal(signal.SIGINT, _shutdown) + + logger.info("Telegram Scraper Web UI listening on http://%s:%s", host, port) + logger.info("Press Ctrl+C to stop.") try: - server.serve_forever() + while not shutdown_event.is_set(): + server.handle_request() except KeyboardInterrupt: - pass + _shutdown() finally: server.server_close() + logger.info("Server stopped.") if __name__ == "__main__":