Add graceful shutdown, logging, streaming with Range Requests, and state.json cache

- Graceful shutdown: SIGTERM/SIGINT handlers stop continuous scraper,
  disconnect Telegram client, drain running jobs with timeout
- Logging: replace print() with logging module throughout web server,
  basicConfig in main.py with structured log format
- Streaming: chunked file reads (64KB) instead of loading entire files
  into memory; full HTTP Range Request support (206 Partial Content)
  for video seeking in the browser
- Cache: TTL-based in-memory cache (1s) for state.json reads in
  StateStore, thread-safe with existing RLock
- Bugfix: send_error_json now converts HTTPStatus to int for JSON

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()
+                result = self._default_state()
-            return self._merge_defaults(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:

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()
 
 

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))

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:
+        while not self._shutdown_flag:
-            job = self.queue.get()
+            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()
+        stat = file_path.stat()
-        self.send_response(HTTPStatus.OK)
+        file_size = stat.st_size
-        self.send_header("Content-Type", content_type)
+        last_modified = stat.st_mtime
-        self.send_header("Content-Length", str(len(data)))
+
-        self.end_headers()
+        range_header = self.headers.get("Range", "").strip()
-        if not head_only:
+        if range_header.startswith("bytes="):
-            self.wfile.write(data)
+            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}")
+    shutdown_event = threading.Event()
-    print("Press Ctrl+C to stop.")
+
+    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__":