forust/telegram-scraper
1
0 Fork 0
Code Issues Pull Requests Actions Packages Activity

Harden account import/continuous behavior and finish multi-account UI polish

Browse Source
This commit is contained in:
forust committed 2026-06-27 21:26:18 +02:00
1 parent 59cdad220a
commit 881e7ddfca
8 files changed
+572 -35
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

webui_server.py
+252 -16
View File
@@ -194,6 +194,7 @@ def load_messages(
channel_id: str,
limit: int = 120,
before_message_id: Optional[int] = None,
search: Optional[str] = None,
) -> List[Dict[str, Any]]:
db_path = channel_db_path(account_id, channel_id)
if not db_path.exists():
@@ -207,9 +208,15 @@ def load_messages(
"FROM messages "
)
params: List[Any] = []
where: List[str] = []
if before_message_id is not None:
query += "WHERE message_id < ? "
where.append("message_id < ?")
params.append(before_message_id)
if search:
where.append("message LIKE ?")
params.append(f"%{search}%")
if where:
query += "WHERE " + " AND ".join(where) + " "
query += "ORDER BY message_id DESC LIMIT ?"
params.append(limit)
rows = conn.execute(query, params).fetchall()
@@ -323,16 +330,29 @@ class JobRunner:
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)}"
account_id = payload.get("account_id")
job = Job(
job_id=job_id,
job_type=job_type,
title=title,
payload=payload,
account_id=account_id,
)
with self.lock:
if account_id:
for job_id in self.job_order:
existing = self.jobs[job_id]
if (
existing.account_id == account_id
and existing.status in {"queued", "running"}
):
existing.logs = (
existing.logs
+ "\n"
+ f"[{datetime.now().strftime('%H:%M:%S')}] Reused existing active job for this account."
).strip()
return existing
job_id = f"job-{int(time.time() * 1000)}"
job = Job(
job_id=job_id,
job_type=job_type,
title=title,
payload=payload,
account_id=account_id,
)
self.jobs[job_id] = job
self.job_order.insert(0, job_id)
self.job_order = self.job_order[:20]
@@ -346,6 +366,15 @@ class JobRunner:
result = [j for j in result if j.get("account_id") == account_id]
return result
def active_jobs_by_account(self) -> Dict[str, Dict[str, Any]]:
with self.lock:
active = {}
for job_id in self.job_order:
job = self.jobs[job_id]
if job.account_id and job.status in {"queued", "running"}:
active[job.account_id] = job.to_dict()
return active
def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
with self.lock:
job = self.jobs.get(job_id)
@@ -811,6 +840,20 @@ class PerAccountContinuousScrapeManager:
def _run_loop(self) -> None:
while not self.stop_event.is_set():
# Check auth — don't iterate if account isn't authorized
auth_info = auth_status_for(self.account_id)
if auth_info.get("status") not in ("ready", "authorized"):
self._log(
f"Account not authorized (status={auth_info.get('status')}), "
"skipping iteration",
"warn",
)
sleep_seconds = max(5, 60)
interrupted = self.stop_event.wait(timeout=sleep_seconds)
if interrupted:
break
continue
channels = self._resolve_channels()
cfg = self.snapshot()["config"]
interval_minutes = cfg.get("interval_minutes", 1)
@@ -904,10 +947,10 @@ class ContinuousScrapeOrchestrator:
mgr = self._get_or_create(account_id)
return mgr.update(enabled, interval_minutes, channels, run_all_tracked)
def add_account(self, account_id: str) -> None:
def add_account(self, account_id: str, auto_start: bool = True) -> None:
acc_state = load_account(DATA_DIR, account_id)
cfg = acc_state.get("continuous_scraping", {})
if cfg.get("enabled", True) and START_CONTINUOUS:
if auto_start and cfg.get("enabled", False) and START_CONTINUOUS:
self.start_account(account_id)
def remove_account(self, account_id: str) -> None:
@@ -924,7 +967,7 @@ class ContinuousScrapeOrchestrator:
for account_id in list_accounts(DATA_DIR):
acc_state = load_account(DATA_DIR, account_id)
cfg = acc_state.get("continuous_scraping", {})
if cfg.get("enabled", True):
if cfg.get("enabled", False) and START_CONTINUOUS:
self.start_account(account_id)
@@ -973,6 +1016,27 @@ def auth_status_for(account_id: Optional[str] = None) -> Dict[str, Any]:
return status
def account_health_summary(account_id: str, job_runner: Optional[JobRunner] = None) -> Dict[str, Any]:
acc_state = load_account(DATA_DIR, account_id)
acc_dir = account_data_dir(DATA_DIR, account_id)
session_file = Path(account_session_path(SESSION_DIR, account_id))
channels = list_channels_snapshot(account_id)
active_job = (job_runner.active_jobs_by_account().get(account_id) if job_runner else None)
last_scrape = next((item.get("last_date") for item in channels if item.get("last_date")), None)
return {
"account_id": account_id,
"label": acc_state.get("label", ""),
"data_dir_exists": acc_dir.exists(),
"session_ready": session_file.exists(),
"api_credentials": bool(acc_state.get("api_id") and acc_state.get("api_hash")),
"channel_count": len(channels),
"message_count": sum(int(item.get("message_count") or 0) for item in channels),
"media_count": sum(int(item.get("media_count") or 0) for item in channels),
"last_scrape": last_scrape,
"active_job": active_job,
}
def auth_status() -> Dict[str, Any]:
return auth_status_for(account_id=None)
@@ -1166,6 +1230,12 @@ def openapi_payload() -> Dict[str, Any]:
"in": "query",
"schema": {"type": "integer"},
},
{
"name": "search",
"in": "query",
"schema": {"type": "string"},
"description": "Full-text search in message text",
},
],
"responses": json_response,
}
@@ -1229,6 +1299,31 @@ def openapi_payload() -> Dict[str, Any]:
},
}
},
"/api/jobs/{job_id}/events": {
"get": {
"summary": "Job SSE event stream",
"description": "Server-Sent Events stream of job status updates. Closes when job completes or fails.",
"parameters": [
{
"name": "job_id",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"responses": {
"200": {
"description": "SSE stream of job status objects",
"content": {
"text/event-stream": {
"schema": {"type": "string"}
}
},
},
"404": {"description": "Job not found"},
},
}
},
"/api/jobs/scrape": {
"post": {
"summary": "Scrape one channel or all tracked channels",
@@ -1318,6 +1413,23 @@ def openapi_payload() -> Dict[str, Any]:
"responses": {**json_response, **error_response},
},
},
"/api/accounts/import": {
"post": {
"summary": "Import account from exported JSON",
"requestBody": json_body(
{
"account_id": {"type": "string", "example": "work"},
"state": {
"type": "object",
"description": "Exported account state object",
"example": {"label": "Work", "api_id": 123456, "api_hash": "abc", "channels": {}, "continuous_scraping": {"enabled": False}},
},
},
["account_id", "state"],
),
"responses": {**json_response, **error_response},
}
},
"/api/accounts/{id}": {
"get": {
"summary": "Account dashboard",
@@ -1344,6 +1456,34 @@ def openapi_payload() -> Dict[str, Any]:
"responses": json_response,
},
},
"/api/accounts/{id}/export": {
"get": {
"summary": "Export account settings as JSON",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"responses": json_response,
}
},
"/api/accounts/{id}/health": {
"get": {
"summary": "Account health summary",
"parameters": [
{
"name": "id",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"responses": json_response,
}
},
"/api/accounts/{id}/auth": {
"get": {
"summary": "Account auth snapshot",
@@ -1452,6 +1592,12 @@ def openapi_payload() -> Dict[str, Any]:
"in": "query",
"schema": {"type": "integer"},
},
{
"name": "search",
"in": "query",
"schema": {"type": "string"},
"description": "Full-text search in message text",
},
],
"responses": json_response,
}
@@ -1634,6 +1780,9 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
)
if path == "/api/jobs":
return self.send_json(self.app.job_runner.recent_jobs())
if path.startswith("/api/jobs/") and path.endswith("/events"):
job_id = path.split("/")[-2]
return self.stream_job_events(job_id)
if path.startswith("/api/jobs/"):
job_id = path.rsplit("/", 1)[-1]
job = self.app.job_runner.get_job(job_id)
@@ -1703,6 +1852,7 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
"auth": auth_info,
"status": auth_info.get("status", "unknown"),
"continuous_running": cs_info.get("status", {}).get("running", False),
"health": account_health_summary(account_id, self.app.job_runner),
})
return self.send_json({"accounts": accounts})
@@ -1719,6 +1869,15 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
return self._handle_get_account_dashboard(account_id)
if sub == ["auth"]:
return self.send_json(self.app.auth_manager.auth_state(account_id))
if sub == ["health"]:
return self.send_json(account_health_summary(account_id, self.app.job_runner))
if sub == ["export"]:
payload = {
"version": 1,
"account_id": account_id,
"state": load_account(DATA_DIR, account_id),
}
return self.send_json(payload)
if sub == ["channels"]:
return self._handle_get_account_channels(account_id)
if len(sub) >= 3 and sub[0] == "channels" and sub[-1] == "messages":
@@ -1744,9 +1903,37 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
"jobs": self.app.job_runner.recent_jobs(account_id=account_id),
"continuous": cs_info,
"scrape_media": bool(acc_state.get("scrape_media", True)),
"health": account_health_summary(account_id, self.app.job_runner),
}
return self.send_json(payload)
def stream_job_events(self, job_id: str) -> None:
if not self.app.job_runner.get_job(job_id):
return self.send_error_json(HTTPStatus.NOT_FOUND, "Job not found")
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache")
self.send_header("Connection", "keep-alive")
self.end_headers()
last_payload = None
deadline = time.time() + 60 * 30
while time.time() < deadline:
job = self.app.job_runner.get_job(job_id)
if not job:
break
payload = json.dumps(job, ensure_ascii=False)
if payload != last_payload:
try:
self.wfile.write(f"data: {payload}\n\n".encode("utf-8"))
self.wfile.flush()
except (BrokenPipeError, ConnectionResetError, OSError):
break
last_payload = payload
if job.get("status") in {"completed", "failed"}:
break
time.sleep(1)
def _handle_get_account_channels(self, account_id: str) -> None:
return self.send_json(list_channels_snapshot(account_id))
@@ -1756,10 +1943,12 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
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
search = (query.get("search") or query.get("q") or [""])[0].strip()
payload = {
"channel_id": channel_id,
"messages": load_messages(
account_id, channel_id, limit=limit, before_message_id=before_message_id
, search=search or None
),
"channel": next(
(
@@ -1997,6 +2186,8 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
# ── /api/accounts/* POST routes ─────────────────────────────────
if path == "/api/accounts":
return self._handle_post_accounts_create(body)
if path == "/api/accounts/import":
return self._handle_post_accounts_import(body)
if path.startswith("/api/accounts/"):
return self._handle_post_account(path, body)
@@ -2034,14 +2225,52 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
"scrape_media": True,
"forwarding_rules": [],
"continuous_scraping": {
"enabled": True,
"enabled": False,
"interval_minutes": 1,
"channels": [],
"run_all_tracked": True,
},
}
store.save(init_state)
self.app.continuous_orchestrator.add_account(account_id)
self.app.continuous_orchestrator.add_account(account_id, auto_start=False)
return self.send_json({"ok": True, "account_id": account_id})
def _handle_post_accounts_import(self, body: Dict[str, Any]) -> None:
account_id = str(body.get("account_id") or body.get("id") or "").strip()
state = body.get("state")
if not account_id:
return self.send_error_json(HTTPStatus.BAD_REQUEST, "account_id is required")
if not account_id.replace("-", "").replace("_", "").isalnum():
return self.send_error_json(HTTPStatus.BAD_REQUEST, "account_id must be alphanumeric (dashes and underscores allowed)")
if account_exists(DATA_DIR, account_id):
return self.send_error_json(HTTPStatus.BAD_REQUEST, f"Account '{account_id}' already exists")
if not isinstance(state, dict):
return self.send_error_json(HTTPStatus.BAD_REQUEST, "state object is required")
imported_state = {
"label": str(body.get("label") or state.get("label") or account_id).strip(),
"api_id": state.get("api_id"),
"api_hash": state.get("api_hash"),
"channels": state.get("channels") if isinstance(state.get("channels"), dict) else {},
"channel_names": state.get("channel_names") if isinstance(state.get("channel_names"), dict) else {},
"scrape_media": bool(state.get("scrape_media", True)),
"forwarding_rules": state.get("forwarding_rules") if isinstance(state.get("forwarding_rules"), list) else [],
"continuous_scraping": state.get("continuous_scraping") if isinstance(state.get("continuous_scraping"), dict) else {
"enabled": False,
"interval_minutes": 1,
"channels": [],
"run_all_tracked": True,
},
}
def mutate_global(global_state: Dict[str, Any]) -> None:
accounts = global_state.setdefault("accounts", [])
if account_id not in accounts:
accounts.append(account_id)
get_global_store(DATA_DIR).update(mutate_global)
get_account_store(DATA_DIR, account_id).save(imported_state)
self.app.continuous_orchestrator.add_account(account_id, auto_start=False)
return self.send_json({"ok": True, "account_id": account_id})
def _handle_post_account(self, path: str, body: Dict[str, Any]) -> None:
@@ -2300,9 +2529,16 @@ class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
if acc_dir.exists():
import shutil
shutil.rmtree(str(acc_dir), ignore_errors=True)
if session_file.exists():
for session_sidecar in (
session_file,
Path(str(session_file) + "-wal"),
Path(str(session_file) + "-shm"),
session_file.with_suffix(session_file.suffix + "-journal"),
):
if not session_sidecar.exists():
continue
try:
session_file.unlink()
session_sidecar.unlink()
except OSError:
pass