forust
1dfe41950e
- 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
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
import json
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Dict, Optional
|
|
|
|
|
|
DEFAULT_STATE: Dict[str, Any] = {
|
|
"api_id": None,
|
|
"api_hash": None,
|
|
"channels": {},
|
|
"channel_names": {},
|
|
"scrape_media": True,
|
|
"forwarding_rules": [],
|
|
"continuous_scraping": {
|
|
"enabled": True,
|
|
"interval_minutes": 1,
|
|
"channels": [],
|
|
"run_all_tracked": True,
|
|
},
|
|
}
|
|
|
|
|
|
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():
|
|
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):
|
|
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:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp_path = self.path.with_suffix(self.path.suffix + ".tmp")
|
|
with tmp_path.open("w", encoding="utf-8") as handle:
|
|
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:
|
|
state = self.load()
|
|
mutator(state)
|
|
self.save(state)
|
|
return state
|
|
|
|
def continuous_config(self) -> Dict[str, Any]:
|
|
state = self.load()
|
|
return dict(state.get("continuous_scraping") or DEFAULT_STATE["continuous_scraping"])
|
|
|
|
def save_continuous_config(self, config: Dict[str, Any]) -> Dict[str, Any]:
|
|
def mutate(state: Dict[str, Any]) -> None:
|
|
state["continuous_scraping"] = {
|
|
"enabled": bool(config.get("enabled", True)),
|
|
"interval_minutes": max(1, int(config.get("interval_minutes", 1) or 1)),
|
|
"channels": [
|
|
str(item).strip()
|
|
for item in config.get("channels", [])
|
|
if str(item).strip()
|
|
],
|
|
"run_all_tracked": bool(config.get("run_all_tracked", True)),
|
|
}
|
|
|
|
return self.update(mutate)["continuous_scraping"]
|
|
|
|
def _default_state(self) -> Dict[str, Any]:
|
|
return json.loads(json.dumps(DEFAULT_STATE))
|
|
|
|
def _merge_defaults(self, state: Dict[str, Any]) -> Dict[str, Any]:
|
|
merged = self._default_state()
|
|
for key, value in state.items():
|
|
if key == "continuous_scraping" and isinstance(value, dict):
|
|
merged[key].update(value)
|
|
else:
|
|
merged[key] = value
|
|
return merged
|