telegram-scraper/webui_server.py

import asyncio
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
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, Dict, List, Optional

import qrcode
import qrcode.image.svg
from app_state import (
    StateStore,
    account_exists,
    account_data_dir,
    account_session_path,
    get_account_store,
    get_global_store,
    load_account,
    list_accounts,
)
from health import health_payload
from scraper_jobs import ScraperJobService
from telethon import TelegramClient
from telethon.errors import SessionPasswordNeededError
from telegram_scraper_with_forwarding import _ensure_session_wal


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"))
START_CONTINUOUS = os.environ.get("TELEGRAM_SCRAPER_START_CONTINUOUS", "1") != "0"
SESSION_DIR = BASE_DIR / "session"
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()


def load_state() -> Dict[str, Any]:
    return STATE_STORE.load()


def save_state(state: Dict[str, Any]) -> None:
    STATE_STORE.save(state)


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 channel_db_path(account_id: Optional[str], channel_id: str) -> Path:
    if account_id:
        return account_data_dir(DATA_DIR, account_id) / channel_id / f"{channel_id}.db"
    return DATA_DIR / channel_id / f"{channel_id}.db"


def database_summary(account_id: Optional[str], channel_id: str) -> Dict[str, Any]:
    db_path = channel_db_path(account_id, channel_id)
    summary: Dict[str, Any] = {
        "message_count": 0,
        "last_date": None,
        "first_date": None,
        "media_count": 0,
        "has_database": db_path.exists(),
        "last_message_preview": "",
        "last_sender": "",
    }
    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,
            }
        )
        try:
            cursor.execute(
                "SELECT message, media_type, first_name, last_name, username, post_author "
                "FROM messages ORDER BY message_id DESC LIMIT 1"
            )
            latest = cursor.fetchone()
        except sqlite3.OperationalError:
            latest = None
        if latest:
            text = (latest[0] or "").strip()
            if not text and latest[1]:
                text = f"[{latest[1]}]"
            sender = " ".join(part for part in [latest[2], latest[3]] if part).strip()
            summary["last_message_preview"] = text[:180]
            summary["last_sender"] = sender or latest[4] or latest[5] or ""
        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(account_id: Optional[str] = None) -> List[Dict[str, Any]]:
    if account_id:
        state = load_account(DATA_DIR, account_id)
    else:
        state = load_state()
    channels: List[Dict[str, Any]] = []
    for channel_id, last_message_id in state.get("channels", {}).items():
        summary = database_summary(account_id, 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 account_state_for_legacy() -> Optional[Dict[str, Any]]:
    legacy = load_state()
    if legacy.get("api_id") and legacy.get("api_hash"):
        return legacy
    return None


def load_messages(
    account_id: Optional[str],
    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():
        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] = []
        where: List[str] = []
        if before_message_id is not None:
            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()
    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"],
            }
        )

    reply_ids = list({m["reply_to"] for m in messages if m["reply_to"]})
    if reply_ids:
        placeholders = ",".join("?" for _ in reply_ids)
        try:
            conn2 = sqlite3.connect(str(db_path))
            conn2.row_factory = sqlite3.Row
            try:
                rows2 = conn2.execute(
                    f"SELECT message_id, message, first_name, last_name, username "
                    f"FROM messages WHERE message_id IN ({placeholders})",
                    reply_ids,
                ).fetchall()
                reply_map: Dict[int, Dict[str, str]] = {}
                for r in rows2:
                    r_sender = " ".join(
                        part for part in [r["first_name"], r["last_name"]] if part
                    ).strip() or (r["username"] or "")
                    reply_map[r["message_id"]] = {
                        "text": (r["message"] or "")[:300],
                        "sender_name": r_sender,
                    }
                for m in messages:
                    rt = m.get("reply_to")
                    if rt and rt in reply_map:
                        m["reply_to_message"] = reply_map[rt]
            finally:
                conn2.close()
        except sqlite3.Error:
            pass

    return messages


# ── Job ──────────────────────────────────────────────────────────────────

@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
    account_id: 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,
            "account_id": self.account_id,
        }


# ── JobRunner ────────────────────────────────────────────────────────────

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._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")
        account_id = payload.get("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]
        self.queue.put(job)
        return job

    def recent_jobs(self, account_id: Optional[str] = None) -> List[Dict[str, Any]]:
        with self.lock:
            result = [self.jobs[job_id].to_dict() for job_id in self.job_order]
        if account_id is not None:
            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)
            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 not self._shutdown_flag:
            try:
                job = self.queue.get(timeout=1)
            except queue.Empty:
                continue
            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()


# ── TelegramAuthManager (per-account) ────────────────────────────────────

class TelegramAuthManager:
    def __init__(self) -> None:
        self.lock = threading.Lock()
        self.loop = asyncio.new_event_loop()
        self.thread = threading.Thread(target=self._run_loop, daemon=True)
        self.thread.start()
        self.clients: Dict[str, TelegramClient] = {}
        self.auth_data: Dict[str, Dict[str, Any]] = {}

    def _run_loop(self) -> None:
        asyncio.set_event_loop(self.loop)
        self.loop.run_forever()

    def _run(self, coro):
        future = asyncio.run_coroutine_threadsafe(coro, self.loop)
        return future.result()

    def _get_auth_data(self, account_id: str) -> Dict[str, Any]:
        if account_id not in self.auth_data:
            self.auth_data[account_id] = {
                "phase": "idle",
                "status": "unknown",
                "details": "",
                "qr_url": None,
                "qr_image": None,
                "phone": None,
                "phone_code_hash": None,
                "qr_login": None,
                "qr_wait_task": None,
                "user_id": None,
                "updated_at": utc_now_iso(),
            }
        return self.auth_data[account_id]

    def _set_state(self, account_id: str, **updates: Any) -> None:
        data = self._get_auth_data(account_id)
        data.update(updates)
        data["updated_at"] = utc_now_iso()

    def _make_qr_image(self, qr_url: str) -> str:
        qr = qrcode.QRCode(border=1, box_size=8)
        qr.add_data(qr_url)
        qr.make(fit=True)
        image = qr.make_image(image_factory=qrcode.image.svg.SvgPathImage)
        buffer = io.BytesIO()
        image.save(buffer)
        encoded = base64.b64encode(buffer.getvalue()).decode("ascii")
        return f"data:image/svg+xml;base64,{encoded}"

    async def _get_client(self, account_id: str) -> TelegramClient:
        acc_state = load_account(DATA_DIR, account_id)
        api_id = acc_state.get("api_id")
        api_hash = acc_state.get("api_hash")
        if not api_id or not api_hash:
            raise RuntimeError("Save api_id and api_hash first for this account.")
        if account_id not in self.clients or self.clients[account_id] is None:
            _ensure_session_wal(account_session_path(SESSION_DIR, account_id))
            self.clients[account_id] = TelegramClient(
                account_session_path(SESSION_DIR, account_id),
                api_id,
                api_hash,
            )
        client = self.clients[account_id]
        if not client.is_connected():
            await client.connect()
        data = self._get_auth_data(account_id)
        if data.get("user_id") is None and await client.is_user_authorized():
            try:
                me = await client.get_me()
                data["user_id"] = me.id
            except Exception:
                pass
        return client

    def auth_state(self, account_id: str) -> Dict[str, Any]:
        data = self._get_auth_data(account_id)
        acc_state = load_account(DATA_DIR, account_id)
        snapshot = dict(data)
        snapshot.pop("qr_login", None)
        snapshot.pop("qr_wait_task", None)
        snapshot.pop("phone_code_hash", None)
        snapshot["saved_credentials"] = {
            "api_id": acc_state.get("api_id"),
            "api_hash_present": bool(acc_state.get("api_hash")),
        }
        snapshot["auth_status"] = auth_status_for(account_id)
        return snapshot

    def all_auth_states(self) -> Dict[str, Dict[str, Any]]:
        result: Dict[str, Dict[str, Any]] = {}
        for account_id in list_accounts(DATA_DIR):
            result[account_id] = self.auth_state(account_id)
        return result

    def save_credentials(self, account_id: str, api_id: int, api_hash: str) -> Dict[str, Any]:
        def mutate(state: Dict[str, Any]) -> None:
            state["api_id"] = int(api_id)
            state["api_hash"] = api_hash.strip()

        store = get_account_store(DATA_DIR, account_id)
        store.update(mutate)
        self._set_state(
            account_id,
            phase="credentials_saved",
            status="ready_for_auth",
            details="Credentials saved. You can login with QR or phone code.",
        )
        return self.auth_state(account_id)

    async def _start_qr_login(self, account_id: str) -> Dict[str, Any]:
        client = await self._get_client(account_id)
        data = self._get_auth_data(account_id)
        if await client.is_user_authorized():
            self._set_state(
                account_id,
                phase="authorized",
                status="authorized",
                details="Telegram session is already authorized.",
            )
            return self.auth_state(account_id)
        qr_login = await client.qr_login()
        qr_url = qr_login.url
        data["qr_login"] = qr_login
        self._set_state(
            account_id,
            phase="qr_waiting",
            status="qr_waiting",
            details="Scan the QR code in Telegram: Settings -> Devices -> Scan QR.",
            qr_url=qr_url,
            qr_image=self._make_qr_image(qr_url),
        )
        data["qr_wait_task"] = self.loop.create_task(
            self._wait_for_qr_login(account_id)
        )
        return self.auth_state(account_id)

    async def _wait_for_qr_login(self, account_id: str) -> None:
        data = self._get_auth_data(account_id)
        qr_login = data.get("qr_login")
        if not qr_login:
            return
        try:
            await qr_login.wait()
            self._set_state(
                account_id,
                phase="authorized",
                status="authorized",
                details="Telegram session is authorized.",
                qr_url=None,
                qr_image=None,
                phone=None,
            )
        except SessionPasswordNeededError:
            self._set_state(
                account_id,
                phase="password_required",
                status="password_required",
                details="Two-factor authentication is enabled. Enter your Telegram password.",
            )
        except Exception as exc:
            self._set_state(
                account_id,
                phase="error",
                status="error",
                details=f"QR login failed: {exc}",
                qr_url=None,
                qr_image=None,
            )

    def start_qr_login(self, account_id: str) -> Dict[str, Any]:
        return self._run(self._start_qr_login(account_id))

    async def _request_phone_code(self, account_id: str, phone: str) -> Dict[str, Any]:
        client = await self._get_client(account_id)
        data = self._get_auth_data(account_id)
        if await client.is_user_authorized():
            self._set_state(
                account_id,
                phase="authorized",
                status="authorized",
                details="Telegram session is already authorized.",
            )
            return self.auth_state(account_id)
        sent = await client.send_code_request(phone)
        data["phone"] = phone
        data["phone_code_hash"] = sent.phone_code_hash
        self._set_state(
            account_id,
            phase="code_required",
            status="code_required",
            details=f"Code sent to {phone}. Enter it below.",
            phone=phone,
        )
        return self.auth_state(account_id)

    def request_phone_code(self, account_id: str, phone: str) -> Dict[str, Any]:
        return self._run(self._request_phone_code(account_id, phone))

    async def _submit_phone_code(self, account_id: str, code: str) -> Dict[str, Any]:
        client = await self._get_client(account_id)
        data = self._get_auth_data(account_id)
        if not data.get("phone") or not data.get("phone_code_hash"):
            raise RuntimeError("Request a phone code first.")
        try:
            await client.sign_in(
                phone=data["phone"],
                code=code,
                phone_code_hash=data["phone_code_hash"],
            )
            self._set_state(
                account_id,
                phase="authorized",
                status="authorized",
                details="Telegram session is authorized.",
                qr_url=None,
                qr_image=None,
            )
        except SessionPasswordNeededError:
            self._set_state(
                account_id,
                phase="password_required",
                status="password_required",
                details="Two-factor authentication is enabled. Enter your Telegram password.",
            )
        return self.auth_state(account_id)

    def submit_phone_code(self, account_id: str, code: str) -> Dict[str, Any]:
        return self._run(self._submit_phone_code(account_id, code))

    async def _submit_password(self, account_id: str, password: str) -> Dict[str, Any]:
        client = await self._get_client(account_id)
        await client.sign_in(password=password)
        self._set_state(
            account_id,
            phase="authorized",
            status="authorized",
            details="Telegram session is authorized.",
            qr_url=None,
            qr_image=None,
        )
        return self.auth_state(account_id)

    def submit_password(self, account_id: str, password: str) -> Dict[str, Any]:
        return self._run(self._submit_password(account_id, password))

    def delete_account(self, account_id: str) -> None:
        if account_id in self.clients:
            client = self.clients[account_id]
            try:
                future = asyncio.run_coroutine_threadsafe(
                    self._disconnect_client(client), self.loop
                )
                future.result(timeout=5)
            except Exception:
                pass
            del self.clients[account_id]
        self.auth_data.pop(account_id, None)

    async def _disconnect_client(self, client: TelegramClient) -> None:
        if client and client.is_connected():
            await client.disconnect()

    def shutdown(self, timeout: float = 5.0) -> None:
        async def _disconnect_all():
            for client in self.clients.values():
                try:
                    if client and client.is_connected():
                        await client.disconnect()
                except Exception:
                    pass

        if self.clients:
            try:
                future = asyncio.run_coroutine_threadsafe(_disconnect_all(), self.loop)
                future.result(timeout=timeout)
            except Exception:
                pass
        self.loop.call_soon_threadsafe(self.loop.stop)
        self.thread.join(timeout=timeout)


# ── PerAccountContinuousScrapeManager ────────────────────────────────────

class PerAccountContinuousScrapeManager:
    def __init__(self, account_id: str) -> None:
        self.account_id = account_id
        self.lock = threading.RLock()
        self.thread: Optional[threading.Thread] = None
        self.stop_event = threading.Event()
        self.config: Dict[str, Any] = self._load_config()
        self.status: Dict[str, Any] = {
            "running": False,
            "last_started_at": None,
            "last_finished_at": None,
            "last_iteration_at": None,
            "last_error": None,
            "logs": [],
            "log_entries": [],
        }

    def _load_config(self) -> Dict[str, Any]:
        acc_state = load_account(DATA_DIR, self.account_id)
        return dict(acc_state.get("continuous_scraping", {
            "enabled": True,
            "interval_minutes": 1,
            "channels": [],
            "run_all_tracked": True,
        }))

    def _save_config(self) -> None:
        store = get_account_store(DATA_DIR, self.account_id)
        def mutate(state: Dict[str, Any]) -> None:
            state["continuous_scraping"] = {
                "enabled": bool(self.config.get("enabled", True)),
                "interval_minutes": max(1, int(self.config.get("interval_minutes", 1) or 1)),
                "channels": [
                    str(item).strip()
                    for item in self.config.get("channels", [])
                    if str(item).strip()
                ],
                "run_all_tracked": bool(self.config.get("run_all_tracked", True)),
            }
        store.update(mutate)

    def refresh_config(self) -> None:
        """
        Re-read account state from disk and sync the in-memory config.
        Call this after external state changes (import, channel add/remove)
        so the manager picks up new channels or config updates without a restart.
        """
        disk_state = load_account(DATA_DIR, self.account_id)
        disk_cfg = disk_state.get("continuous_scraping") or {}
        with self.lock:
            # Merge disk config into memory, preserving our current running state
            self.config["enabled"] = bool(disk_cfg.get("enabled", False))
            self.config["interval_minutes"] = max(1, int(disk_cfg.get("interval_minutes", 1) or 1))
            self.config["channels"] = [
                str(item).strip()
                for item in disk_cfg.get("channels", [])
                if str(item).strip()
            ]
            self.config["run_all_tracked"] = bool(disk_cfg.get("run_all_tracked", True))
            # Sync running state with desired enabled state
            if self.config["enabled"] and not self.status["running"]:
                pass  # don't auto-start — user must call start()
            elif not self.config["enabled"] and self.status["running"]:
                self._log("Continuous disabled via external state change, stopping.", "warn")
                self.stop_event.set()
                self.status["running"] = False
        self._log("Config refreshed from disk.", "debug")

    def _log(self, message: str, level: str = "debug") -> None:
        timestamp = utc_now_iso()
        entry = {"timestamp": timestamp, "level": level, "message": message}
        with self.lock:
            logs = self.status.setdefault("logs", [])
            log_entries = self.status.setdefault("log_entries", [])
            logs.append(f"[{datetime.now().strftime('%H:%M:%S')}] {message}")
            log_entries.append(entry)
            if len(logs) > 300:
                del logs[:-300]
            if len(log_entries) > 300:
                del log_entries[:-300]

    def snapshot(self) -> Dict[str, Any]:
        with self.lock:
            return {
                "config": dict(self.config),
                "status": {
                    **self.status,
                    "logs": list(self.status.get("logs", [])),
                    "log_entries": list(self.status.get("log_entries", [])),
                },
            }

    def update(
        self,
        enabled: bool,
        interval_minutes: int,
        channels: List[str],
        run_all_tracked: bool,
    ) -> Dict[str, Any]:
        interval_minutes = max(1, int(interval_minutes))
        with self.lock:
            self.config = {
                "enabled": bool(enabled),
                "interval_minutes": interval_minutes,
                "channels": channels,
                "run_all_tracked": bool(run_all_tracked),
            }
            self._save_config()
        self.config = self._load_config()
        if enabled:
            self.start()
        else:
            self.stop()
        return self.snapshot()

    def start(self) -> None:
        with self.lock:
            if self.status["running"]:
                self._log("Continuous scraping is already running.", "warn")
                return
            self.stop_event.clear()
            self.status["running"] = True
            self.status["last_started_at"] = utc_now_iso()
            self.status["last_error"] = None
        self.thread = threading.Thread(target=self._run_loop, daemon=True)
        self.thread.start()
        self._log("Continuous scraping started.", "info")

    def stop(self) -> None:
        self.stop_event.set()
        with self.lock:
            was_running = self.status["running"]
            self.status["running"] = False
        if was_running:
            self._log("Continuous scraping stop requested.", "warn")

    def _resolve_channels(self) -> List[str]:
        acc_state = load_account(DATA_DIR, self.account_id)
        with self.lock:
            run_all_tracked = self.config.get("run_all_tracked", True)
            configured = list(self.config.get("channels", []))
        if run_all_tracked:
            return list(acc_state.get("channels", {}).keys())
        tracked = set(acc_state.get("channels", {}).keys())
        return [channel for channel in configured if channel in tracked]

    def _run_loop(self) -> None:
        while not self.stop_event.is_set():
            # Refresh config from disk so channel / setting changes take effect
            self.refresh_config()

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

            if not channels:
                self._log("No channels configured for continuous scraping.", "warn")
            else:
                self._log(f"Starting iteration for {len(channels)} channel(s).", "info")
                buffer = io.StringIO()
                try:
                    with contextlib.redirect_stdout(buffer), contextlib.redirect_stderr(buffer):
                        run_job("scrape_selected", {
                            "channels": channels,
                            "account_id": self.account_id,
                        })
                    output = buffer.getvalue().strip()
                    if output:
                        for line in output.splitlines():
                            self._log(line)
                    with self.lock:
                        self.status["last_iteration_at"] = utc_now_iso()
                        self.status["last_finished_at"] = utc_now_iso()
                        self.status["last_error"] = None
                    self._log("Iteration finished.", "success")
                except Exception as exc:
                    output = buffer.getvalue().strip()
                    if output:
                        for line in output.splitlines():
                            self._log(line)
                    self._log(f"Iteration failed: {exc}", "error")
                    with self.lock:
                        self.status["last_error"] = str(exc)

            sleep_seconds = max(5, interval_minutes * 60)
            self._log(f"Sleeping for {interval_minutes} minute(s).", "debug")
            interrupted = self.stop_event.wait(timeout=sleep_seconds)
            if interrupted:
                break

        with self.lock:
            self.status["running"] = False
        self._log("Continuous scraping stopped.", "warn")


# ── ContinuousScrapeOrchestrator ─────────────────────────────────────────

class ContinuousScrapeOrchestrator:
    def __init__(self) -> None:
        self.lock = threading.Lock()
        self.managers: Dict[str, PerAccountContinuousScrapeManager] = {}

    def _get_or_create(self, account_id: str) -> PerAccountContinuousScrapeManager:
        with self.lock:
            if account_id not in self.managers:
                self.managers[account_id] = PerAccountContinuousScrapeManager(account_id)
            return self.managers[account_id]

    def start_account(self, account_id: str) -> None:
        mgr = self._get_or_create(account_id)
        mgr.start()

    def stop_account(self, account_id: str) -> None:
        with self.lock:
            mgr = self.managers.get(account_id)
        if mgr:
            mgr.stop()

    def snapshot_for(self, account_id: str) -> Dict[str, Any]:
        mgr = self._get_or_create(account_id)
        return mgr.snapshot()

    def snapshot(self) -> Dict[str, Dict[str, Any]]:
        result: Dict[str, Dict[str, Any]] = {}
        with self.lock:
            for account_id, mgr in self.managers.items():
                result[account_id] = dict(mgr.snapshot())
        missing = set(list_accounts(DATA_DIR)) - set(result.keys())
        for account_id in missing:
            mgr = self._get_or_create(account_id)
            result[account_id] = dict(mgr.snapshot())
        return result

    def refresh_account(self, account_id: str) -> None:
        """Re-read account state from disk and sync the continuous manager config."""
        mgr = self._get_or_create(account_id)
        mgr.refresh_config()

    def update_for(
        self,
        account_id: str,
        enabled: bool,
        interval_minutes: int,
        channels: List[str],
        run_all_tracked: bool,
    ) -> Dict[str, Any]:
        mgr = self._get_or_create(account_id)
        result = mgr.update(enabled, interval_minutes, channels, run_all_tracked)
        # After explicit update, sync state to disk is already done by mgr.update
        return result

    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 auto_start and cfg.get("enabled", False) and START_CONTINUOUS:
            self.start_account(account_id)

    def remove_account(self, account_id: str) -> None:
        self.stop_account(account_id)
        with self.lock:
            self.managers.pop(account_id, None)

    def stop_all(self) -> None:
        with self.lock:
            for account_id in list(self.managers.keys()):
                self.managers[account_id].stop()

    def start_all(self) -> None:
        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", False) and START_CONTINUOUS:
                self.start_account(account_id)


# ── Legacy helpers ───────────────────────────────────────────────────────

def import_scraper_class():
    from telegram_scraper_with_forwarding import OptimizedTelegramScraper
    return OptimizedTelegramScraper


def run_job(job_type: str, payload: Dict[str, Any]) -> None:
    SCRAPER_JOBS.run(job_type, payload)


def auth_status_for(account_id: Optional[str] = None) -> Dict[str, Any]:
    if account_id:
        acc_state = load_account(DATA_DIR, account_id)
        session_file = account_session_path(SESSION_DIR, account_id)
    else:
        acc_state = load_state()
        session_file = str(SESSION_DIR / "session.session")
    status: Dict[str, Any] = {
        "has_api_credentials": bool(acc_state.get("api_id") and acc_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

    status["session_ready"] = Path(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."
    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)


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


def openapi_payload() -> Dict[str, Any]:
    json_response = {
        "200": {
            "description": "JSON response",
            "content": {"application/json": {"schema": {"type": "object"}}},
        }
    }
    accepted_response = {
        "202": {
            "description": "Job accepted",
            "content": {"application/json": {"schema": {"type": "object"}}},
        }
    }
    error_response = {"400": {"description": "Bad request"}}

    def json_body(properties: Dict[str, Any], required: Optional[List[str]] = None):
        return {
            "required": True,
            "content": {
                "application/json": {
                    "schema": {
                        "type": "object",
                        "properties": properties,
                        "required": required or [],
                        "example": {
                            key: value.get("example")
                            for key, value in properties.items()
                            if "example" in value
                        },
                    }
                }
            },
        }

    return {
        "openapi": "3.0.3",
        "info": {
            "title": "Telegram Scraper Web UI API",
            "version": "0.1.0",
            "description": "Local endpoints used by the Telegram Scraper web interface.",
        },
        "servers": [{"url": "/"}],
        "paths": {
            "/api/dashboard": {
                "get": {
                    "summary": "Dashboard snapshot",
                    "description": "Returns state summary, channels, auth status, and recent jobs.",
                    "responses": json_response,
                }
            },
            "/api/auth": {
                "get": {
                    "summary": "Authentication status",
                    "description": "Returns saved credential flags, login phase, and QR login data.",
                    "responses": json_response,
                }
            },
            "/api/auth/credentials": {
                "post": {
                    "summary": "Save Telegram API credentials",
                    "requestBody": json_body(
                        {
                            "api_id": {"type": "integer", "example": 123456},
                            "api_hash": {
                                "type": "string",
                                "example": "0123456789abcdef",
                            },
                        },
                        ["api_id", "api_hash"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/auth/qr/start": {
                "post": {
                    "summary": "Start QR login",
                    "requestBody": json_body({}),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/auth/phone/request": {
                "post": {
                    "summary": "Request phone login code",
                    "requestBody": json_body(
                        {"phone": {"type": "string", "example": "+1234567890"}},
                        ["phone"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/auth/phone/submit": {
                "post": {
                    "summary": "Submit phone login code",
                    "requestBody": json_body(
                        {"code": {"type": "string", "example": "12345"}},
                        ["code"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/auth/password": {
                "post": {
                    "summary": "Submit 2FA password",
                    "requestBody": json_body(
                        {"password": {"type": "string", "example": "password"}},
                        ["password"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/continuous": {
                "get": {
                    "summary": "Continuous scraping status",
                    "responses": json_response,
                },
                "post": {
                    "summary": "Update continuous scraping",
                    "requestBody": json_body(
                        {
                            "enabled": {"type": "boolean", "example": True},
                            "interval_minutes": {"type": "integer", "example": 5},
                            "run_all_tracked": {"type": "boolean", "example": False},
                            "channels": {
                                "type": "array",
                                "items": {"type": "string"},
                                "example": ["-1001234567890"],
                            },
                        }
                    ),
                    "responses": {**json_response, **error_response},
                },
            },
            "/health/continuous": {
                "get": {
                    "summary": "Continuous scraping health",
                    "responses": json_response,
                }
            },
            "/health": {
                "get": {
                    "summary": "Application health",
                    "responses": json_response,
                }
            },
            "/api/channels": {
                "get": {
                    "summary": "List tracked channels",
                    "responses": json_response,
                }
            },
            "/api/channels/{channel_id}/messages": {
                "get": {
                    "summary": "List channel messages",
                    "parameters": [
                        {
                            "name": "channel_id",
                            "in": "path",
                            "required": True,
                            "schema": {"type": "string"},
                        },
                        {
                            "name": "limit",
                            "in": "query",
                            "schema": {
                                "type": "integer",
                                "default": 120,
                                "maximum": 300,
                            },
                        },
                        {
                            "name": "before",
                            "in": "query",
                            "schema": {"type": "integer"},
                        },
                        {
                            "name": "search",
                            "in": "query",
                            "schema": {"type": "string"},
                            "description": "Full-text search in message text",
                        },
                    ],
                    "responses": json_response,
                }
            },
            "/api/channels/add": {
                "post": {
                    "summary": "Track a channel",
                    "requestBody": json_body(
                        {
                            "channel_id": {
                                "type": "string",
                                "example": "-1001234567890",
                            },
                            "name": {"type": "string", "example": "Research feed"},
                        },
                        ["channel_id"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/channels/remove": {
                "post": {
                    "summary": "Remove a tracked channel",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}},
                        ["channel_id"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/settings/media": {
                "post": {
                    "summary": "Toggle media scraping",
                    "requestBody": json_body(
                        {"value": {"type": "boolean", "example": True}},
                        ["value"],
                    ),
                    "responses": {**accepted_response, **error_response},
                }
            },
            "/api/jobs": {
                "get": {
                    "summary": "Recent jobs",
                    "responses": json_response,
                }
            },
            "/api/jobs/{job_id}": {
                "get": {
                    "summary": "Job details",
                    "parameters": [
                        {
                            "name": "job_id",
                            "in": "path",
                            "required": True,
                            "schema": {"type": "string"},
                        }
                    ],
                    "responses": {
                        **json_response,
                        "404": {"description": "Job not found"},
                    },
                }
            },
            "/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",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}}
                    ),
                    "responses": accepted_response,
                }
            },
            "/api/jobs/export": {
                "post": {
                    "summary": "Export all tracked channels",
                    "requestBody": json_body({}),
                    "responses": accepted_response,
                }
            },
            "/api/jobs/export-channel": {
                "post": {
                    "summary": "Export one channel",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}},
                        ["channel_id"],
                    ),
                    "responses": {**accepted_response, **error_response},
                }
            },
            "/api/jobs/rescrape-media": {
                "post": {
                    "summary": "Rescrape media for a channel",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}},
                        ["channel_id"],
                    ),
                    "responses": {**accepted_response, **error_response},
                }
            },
            "/api/jobs/fix-media": {
                "post": {
                    "summary": "Fix missing media for a channel",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}},
                        ["channel_id"],
                    ),
                    "responses": {**accepted_response, **error_response},
                }
            },
            "/api/jobs/refresh-dialogs": {
                "post": {
                    "summary": "Refresh Telegram dialogs",
                    "requestBody": json_body({}),
                    "responses": accepted_response,
                }
            },
            "/openapi.json": {
                "get": {
                    "summary": "OpenAPI document",
                    "responses": json_response,
                }
            },
            "/api/accounts": {
                "get": {
                    "summary": "List all accounts",
                    "responses": json_response,
                },
                "post": {
                    "summary": "Create a new account",
                    "requestBody": json_body(
                        {
                            "account_id": {
                                "type": "string",
                                "example": "work",
                            },
                            "label": {
                                "type": "string",
                                "example": "Work Account",
                            },
                            "api_id": {
                                "type": "integer",
                                "example": 123456,
                            },
                            "api_hash": {
                                "type": "string",
                                "example": "0123456789abcdef",
                            },
                        }
                    ),
                    "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",
                    "parameters": [
                        {
                            "name": "id",
                            "in": "path",
                            "required": True,
                            "schema": {"type": "string"},
                        }
                    ],
                    "responses": json_response,
                },
                "delete": {
                    "summary": "Delete account",
                    "parameters": [
                        {
                            "name": "id",
                            "in": "path",
                            "required": True,
                            "schema": {"type": "string"},
                        }
                    ],
                    "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",
                    "parameters": [
                        {
                            "name": "id",
                            "in": "path",
                            "required": True,
                            "schema": {"type": "string"},
                        }
                    ],
                    "responses": json_response,
                },
            },
            "/api/accounts/{id}/auth/credentials": {
                "post": {
                    "summary": "Save credentials for account",
                    "requestBody": json_body(
                        {
                            "api_id": {"type": "integer", "example": 123456},
                            "api_hash": {"type": "string", "example": "0123456789abcdef"},
                        },
                        ["api_id", "api_hash"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/accounts/{id}/auth/qr/start": {
                "post": {
                    "summary": "Start QR login for account",
                    "requestBody": json_body({}),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/accounts/{id}/auth/phone/request": {
                "post": {
                    "summary": "Request phone code for account",
                    "requestBody": json_body(
                        {"phone": {"type": "string", "example": "+1234567890"}},
                        ["phone"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/accounts/{id}/auth/phone/submit": {
                "post": {
                    "summary": "Submit phone code for account",
                    "requestBody": json_body(
                        {"code": {"type": "string", "example": "12345"}},
                        ["code"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/accounts/{id}/auth/password": {
                "post": {
                    "summary": "Submit 2FA password for account",
                    "requestBody": json_body(
                        {"password": {"type": "string", "example": "password"}},
                        ["password"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/accounts/{id}/channels": {
                "get": {
                    "summary": "List channels for account",
                    "parameters": [
                        {
                            "name": "id",
                            "in": "path",
                            "required": True,
                            "schema": {"type": "string"},
                        }
                    ],
                    "responses": json_response,
                }
            },
            "/api/accounts/{id}/channels/{channel_id}/messages": {
                "get": {
                    "summary": "List messages for account channel",
                    "parameters": [
                        {
                            "name": "id",
                            "in": "path",
                            "required": True,
                            "schema": {"type": "string"},
                        },
                        {
                            "name": "channel_id",
                            "in": "path",
                            "required": True,
                            "schema": {"type": "string"},
                        },
                        {
                            "name": "limit",
                            "in": "query",
                            "schema": {
                                "type": "integer",
                                "default": 120,
                                "maximum": 300,
                            },
                        },
                        {
                            "name": "before",
                            "in": "query",
                            "schema": {"type": "integer"},
                        },
                        {
                            "name": "search",
                            "in": "query",
                            "schema": {"type": "string"},
                            "description": "Full-text search in message text",
                        },
                    ],
                    "responses": json_response,
                }
            },
            "/api/accounts/{id}/channels/add": {
                "post": {
                    "summary": "Add channel to account",
                    "requestBody": json_body(
                        {
                            "channel_id": {"type": "string", "example": "-1001234567890"},
                            "name": {"type": "string", "example": "Research feed"},
                        },
                        ["channel_id"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/accounts/{id}/channels/remove": {
                "post": {
                    "summary": "Remove channel from account",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}},
                        ["channel_id"],
                    ),
                    "responses": {**json_response, **error_response},
                }
            },
            "/api/accounts/{id}/jobs/scrape": {
                "post": {
                    "summary": "Scrape channels for account",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}}
                    ),
                    "responses": accepted_response,
                }
            },
            "/api/accounts/{id}/jobs/export": {
                "post": {
                    "summary": "Export all channels for account",
                    "requestBody": json_body({}),
                    "responses": accepted_response,
                }
            },
            "/api/accounts/{id}/jobs/export-channel": {
                "post": {
                    "summary": "Export one channel for account",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}},
                        ["channel_id"],
                    ),
                    "responses": {**accepted_response, **error_response},
                }
            },
            "/api/accounts/{id}/jobs/rescrape-media": {
                "post": {
                    "summary": "Rescrape media for account channel",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}},
                        ["channel_id"],
                    ),
                    "responses": {**accepted_response, **error_response},
                }
            },
            "/api/accounts/{id}/jobs/fix-media": {
                "post": {
                    "summary": "Fix missing media for account channel",
                    "requestBody": json_body(
                        {"channel_id": {"type": "string", "example": "-1001234567890"}},
                        ["channel_id"],
                    ),
                    "responses": {**accepted_response, **error_response},
                }
            },
            "/api/accounts/{id}/jobs/refresh-dialogs": {
                "post": {
                    "summary": "Refresh dialogs for account",
                    "requestBody": json_body({}),
                    "responses": accepted_response,
                }
            },
            "/api/accounts/{id}/settings/media": {
                "post": {
                    "summary": "Toggle media scraping for account",
                    "requestBody": json_body(
                        {"value": {"type": "boolean", "example": True}},
                        ["value"],
                    ),
                    "responses": {**accepted_response, **error_response},
                }
            },
            "/api/accounts/{id}/continuous": {
                "get": {
                    "summary": "Continuous scraping status for account",
                    "parameters": [
                        {
                            "name": "id",
                            "in": "path",
                            "required": True,
                            "schema": {"type": "string"},
                        }
                    ],
                    "responses": json_response,
                },
                "post": {
                    "summary": "Update continuous scraping for account",
                    "requestBody": json_body(
                        {
                            "enabled": {"type": "boolean", "example": True},
                            "interval_minutes": {"type": "integer", "example": 5},
                            "run_all_tracked": {"type": "boolean", "example": False},
                            "channels": {
                                "type": "array",
                                "items": {"type": "string"},
                                "example": ["-1001234567890"],
                            },
                        }
                    ),
                    "responses": {**json_response, **error_response},
                },
            },
        },
    }


# ── Request Handler ──────────────────────────────────────────────────────

class TelegramScraperRequestHandler(BaseHTTPRequestHandler):
    server_version = "TelegramScraperWebUI/0.1"

    @property
    def app(self) -> "TelegramScraperWebServer":
        return self.server  # type: ignore[return-value]

    # ── GET ──────────────────────────────────────────────────────────────

    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 in {"/swagger", "/swigger", "/docs"}:
            return self.serve_file(WEBUI_DIR / "swagger.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 == "/openapi.json":
            return self.send_json(openapi_payload())
        if path == "/api/dashboard":
            return self.send_json(dashboard_payload(self.app.job_runner))
        if path == "/api/auth":
            if self.app.legacy_account_id:
                return self.send_json(self.app.auth_manager.auth_state(self.app.legacy_account_id))
            return self.send_json(self._legacy_auth_snapshot())
        if path == "/api/continuous":
            if self.app.legacy_account_id:
                return self.send_json(self.app.continuous_orchestrator.snapshot_for(self.app.legacy_account_id))
            return self.send_json({"config": {}, "status": {"running": False, "logs": [], "log_entries": []}})
        if path == "/health/continuous":
            if self.app.legacy_account_id:
                return self.send_json(self.app.continuous_orchestrator.snapshot_for(self.app.legacy_account_id))
            return self.send_json({"config": {}, "status": {"running": False, "logs": [], "log_entries": []}})
        if path == "/health":
            cs_snapshot = self.app.continuous_orchestrator.snapshot()
            return self.send_json(
                health_payload(
                    DATA_DIR,
                    SESSION_DIR,
                    STATE_STORE,
                    cs_snapshot,
                    self.app.job_runner.queue.qsize(),
                    list_accounts(DATA_DIR),
                )
            )
        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)
            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 = urllib.parse.unquote(parts[3]).lstrip("@")
            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(
                    None, channel_id, limit=limit, before_message_id=before_message_id,
                    search=search or None
                ),
                "channel": next(
                    (
                        item
                        for item in list_channels_snapshot()
                        if item["channel_id"] == channel_id
                    ),
                    None,
                ),
            }
            return self.send_json(payload)

        # ── /api/accounts/* routes ──────────────────────────────────────
        if path == "/api/accounts":
            return self._handle_get_accounts_list()
        if path.startswith("/api/accounts/"):
            return self._handle_get_account(path, query)

        return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found")

    def _legacy_auth_snapshot(self) -> Dict[str, Any]:
        state = load_state()
        return {
            "phase": "unknown",
            "status": "unknown",
            "details": "",
            "qr_url": None,
            "qr_image": None,
            "phone": None,
            "user_id": None,
            "saved_credentials": {
                "api_id": state.get("api_id"),
                "api_hash_present": bool(state.get("api_hash")),
            },
            "auth_status": auth_status(),
        }

    def _handle_get_accounts_list(self) -> None:
        ids = list_accounts(DATA_DIR)
        accounts = []
        for account_id in ids:
            acc_state = load_account(DATA_DIR, account_id)
            auth_info = auth_status_for(account_id)
            cs_info = self.app.continuous_orchestrator.snapshot_for(account_id)
            accounts.append({
                "id": account_id,
                "label": acc_state.get("label", ""),
                "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})

    def _handle_get_account(self, path: str, query: Dict[str, List[str]]) -> None:
        rest = path[len("/api/accounts/"):]
        parts = rest.split("/")
        account_id = urllib.parse.unquote(parts[0])
        sub = parts[1:]

        if not self._account_exists_or_404(account_id):
            return

        if not sub:
            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":
            channel_id = urllib.parse.unquote(sub[1]).lstrip("@")
            return self._handle_get_account_channel_messages(account_id, channel_id, query)
        if sub == ["continuous"]:
            return self.send_json(
                self.app.continuous_orchestrator.snapshot_for(account_id)
            )
        return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found")

    def _handle_get_account_dashboard(self, account_id: str) -> None:
        acc_state = load_account(DATA_DIR, account_id)
        channels = list_channels_snapshot(account_id)
        auth_info = auth_status_for(account_id)
        cs_info = self.app.continuous_orchestrator.snapshot_for(account_id)
        payload = {
            "account_id": account_id,
            "label": acc_state.get("label", ""),
            "auth": auth_info,
            "auth_detail": self.app.auth_manager.auth_state(account_id),
            "channels": channels,
            "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))

    def _handle_get_account_channel_messages(
        self, account_id: str, channel_id: str, query: Dict[str, List[str]]
    ) -> None:
        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(
                (
                    item
                    for item in list_channels_snapshot(account_id)
                    if item["channel_id"] == channel_id
                ),
                None,
            ),
        }
        return self.send_json(payload)

    # ── HEAD ─────────────────────────────────────────────────────────────

    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 in {"/swagger", "/swigger", "/docs"}:
            return self.serve_file(WEBUI_DIR / "swagger.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/auth",
                "/api/continuous",
                "/api/jobs",
                "/api/channels",
                "/api/accounts",
                "/health",
                "/health/continuous",
                "/openapi.json",
            }
            or path.startswith("/api/jobs/")
            or (path.startswith("/api/channels/") and path.endswith("/messages"))
            or path.startswith("/api/accounts/")
        ):
            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")

    # ── POST ─────────────────────────────────────────────────────────────

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

        # ── Legacy endpoints ────────────────────────────────────────────
        if path == "/api/settings/media":
            value = bool(body.get("value"))
            if self.app.legacy_account_id:
                return self._handle_account_settings_media(self.app.legacy_account_id, 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/continuous":
            try:
                enabled = bool(body.get("enabled"))
                interval_minutes = int(body.get("interval_minutes", 1))
                channels = [
                    str(item).strip()
                    for item in body.get("channels", [])
                    if str(item).strip()
                ]
                run_all_tracked = bool(body.get("run_all_tracked", True))
                if self.app.legacy_account_id:
                    payload = self.app.continuous_orchestrator.update_for(
                        account_id=self.app.legacy_account_id,
                        enabled=enabled,
                        interval_minutes=interval_minutes,
                        channels=channels,
                        run_all_tracked=run_all_tracked,
                    )
                else:
                    payload = {"config": {}, "status": {"running": False, "logs": [], "log_entries": []}}
            except Exception as exc:
                return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
            return self.send_json(payload)

        if path == "/api/auth/credentials":
            api_id = body.get("api_id")
            api_hash = str(body.get("api_hash", "")).strip()
            if not api_id or not api_hash:
                return self.send_error_json(HTTPStatus.BAD_REQUEST, "api_id and api_hash are required")
            try:
                state = load_state()
                state["api_id"] = int(api_id)
                state["api_hash"] = api_hash
                save_state(state)
                payload = self._legacy_auth_snapshot()
            except Exception as exc:
                return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
            return self.send_json(payload)

        if path == "/api/auth/qr/start":
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "Legacy auth is not available in multi-account mode. Use /api/accounts/{id}/auth/qr/start instead.")

        if path == "/api/auth/phone/request":
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "Legacy auth is not available in multi-account mode. Use /api/accounts/{id}/auth/phone/request instead.")

        if path == "/api/auth/phone/submit":
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "Legacy auth is not available in multi-account mode. Use /api/accounts/{id}/auth/phone/submit instead.")

        if path == "/api/auth/password":
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "Legacy auth is not available in multi-account mode. Use /api/accounts/{id}/auth/password instead.")

        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")
            if self.app.legacy_account_id:
                return self._handle_account_channel_add(self.app.legacy_account_id, channel_id, body.get("name"))
            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()
            if self.app.legacy_account_id:
                return self._handle_account_channel_remove(self.app.legacy_account_id, channel_id)
            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")
            payload = {}
            account_id = self.app.legacy_account_id
            if account_id:
                payload["account_id"] = account_id
            if channel_id:
                payload["channel_id"] = str(channel_id)
                job = self.app.job_runner.create_job(
                    "scrape_channel",
                    f"Scrape channel {channel_id}" + (f" [{account_id}]" if account_id else ""),
                    payload,
                )
            else:
                job = self.app.job_runner.create_job(
                    "scrape_all",
                    "Scrape all tracked channels" + (f" [{account_id}]" if account_id else ""),
                    payload,
                )
            return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

        if path == "/api/jobs/export":
            payload = {}
            if self.app.legacy_account_id:
                payload["account_id"] = self.app.legacy_account_id
            job = self.app.job_runner.create_job(
                "export_all",
                "Export all tracked channels",
                payload,
            )
            return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

        if path == "/api/jobs/export-channel":
            channel_id = str(body.get("channel_id", "")).strip()
            if not channel_id:
                return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required")
            payload = {"channel_id": channel_id}
            if self.app.legacy_account_id:
                payload["account_id"] = self.app.legacy_account_id
            job = self.app.job_runner.create_job(
                "export_channel",
                f"Export channel {channel_id}",
                payload,
            )
            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")
            payload = {"channel_id": channel_id}
            if self.app.legacy_account_id:
                payload["account_id"] = self.app.legacy_account_id
            job = self.app.job_runner.create_job(
                "rescrape_media",
                f"Rescrape media for {channel_id}",
                payload,
            )
            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")
            payload = {"channel_id": channel_id}
            if self.app.legacy_account_id:
                payload["account_id"] = self.app.legacy_account_id
            job = self.app.job_runner.create_job(
                "fix_missing_media",
                f"Fix missing media for {channel_id}",
                payload,
            )
            return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

        if path == "/api/jobs/refresh-dialogs":
            payload = {}
            if self.app.legacy_account_id:
                payload["account_id"] = self.app.legacy_account_id
            job = self.app.job_runner.create_job(
                "refresh_dialogs",
                "Refresh Telegram dialogs list",
                payload,
            )
            return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

        # ── /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)

        return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found")

    def _handle_post_accounts_create(self, body: Dict[str, Any]) -> None:
        account_id = str(body.get("account_id", "")).strip()
        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)")

        existing = list_accounts(DATA_DIR)
        if account_id in existing:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, f"Account '{account_id}' already exists")

        label = str(body.get("label", account_id)).strip()
        api_id = body.get("api_id")
        api_hash = str(body.get("api_hash", "")).strip()

        def mutate_global(state: Dict[str, Any]) -> None:
            accounts = state.setdefault("accounts", [])
            if account_id not in accounts:
                accounts.append(account_id)

        get_global_store(DATA_DIR).update(mutate_global)

        store = get_account_store(DATA_DIR, account_id)
        init_state = {
            "label": label,
            "api_id": int(api_id) if api_id else None,
            "api_hash": api_hash if api_id else None,
            "channels": {},
            "channel_names": {},
            "scrape_media": True,
            "forwarding_rules": [],
            "continuous_scraping": {
                "enabled": False,
                "interval_minutes": 1,
                "channels": [],
                "run_all_tracked": True,
            },
        }
        store.save(init_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_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)
        self.app.continuous_orchestrator.refresh_account(account_id)
        return self.send_json({"ok": True, "account_id": account_id})

    def _handle_post_account(self, path: str, body: Dict[str, Any]) -> None:
        rest = path[len("/api/accounts/"):]
        parts = rest.split("/")
        account_id = urllib.parse.unquote(parts[0])
        sub = parts[1:]

        if not sub:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "Missing endpoint")
        if not self._account_exists_or_404(account_id):
            return
        if sub == ["auth", "credentials"]:
            return self._handle_post_account_auth_credentials(account_id, body)
        if sub == ["auth", "qr", "start"]:
            return self._handle_post_account_auth_qr_start(account_id)
        if sub == ["auth", "phone", "request"]:
            return self._handle_post_account_auth_phone_request(account_id, body)
        if sub == ["auth", "phone", "submit"]:
            return self._handle_post_account_auth_phone_submit(account_id, body)
        if sub == ["auth", "password"]:
            return self._handle_post_account_auth_password(account_id, body)
        if sub == ["channels", "add"]:
            return self._handle_account_channel_add(account_id, str(body.get("channel_id", "")).strip(), body.get("name"))
        if sub == ["channels", "remove"]:
            return self._handle_account_channel_remove(account_id, str(body.get("channel_id", "")).strip())
        if sub == ["jobs", "scrape"]:
            return self._handle_account_job_scrape(account_id, body)
        if sub == ["jobs", "export"]:
            return self._handle_account_job_export(account_id)
        if sub == ["jobs", "export-channel"]:
            return self._handle_account_job_export_channel(account_id, body)
        if sub == ["jobs", "rescrape-media"]:
            return self._handle_account_job_rescrape_media(account_id, body)
        if sub == ["jobs", "fix-media"]:
            return self._handle_account_job_fix_media(account_id, body)
        if sub == ["jobs", "refresh-dialogs"]:
            return self._handle_account_job_refresh_dialogs(account_id)
        if sub == ["settings", "media"]:
            return self._handle_account_settings_media(account_id, bool(body.get("value")))
        if sub == ["continuous"]:
            return self._handle_account_continuous(account_id, body)
        return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found")

    def _account_exists_or_404(self, account_id: str) -> bool:
        if not account_id or not account_exists(DATA_DIR, account_id):
            self.send_error_json(HTTPStatus.NOT_FOUND, f"Account '{account_id}' not found")
            return False
        return True

    def _handle_post_account_auth_credentials(self, account_id: str, body: Dict[str, Any]) -> None:
        api_id = body.get("api_id")
        api_hash = str(body.get("api_hash", "")).strip()
        if not api_id or not api_hash:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "api_id and api_hash are required")
        try:
            payload = self.app.auth_manager.save_credentials(account_id, int(api_id), api_hash)
        except Exception as exc:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
        return self.send_json(payload)

    def _handle_post_account_auth_qr_start(self, account_id: str) -> None:
        try:
            payload = self.app.auth_manager.start_qr_login(account_id)
        except Exception as exc:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
        return self.send_json(payload)

    def _handle_post_account_auth_phone_request(self, account_id: str, body: Dict[str, Any]) -> None:
        phone = str(body.get("phone", "")).strip()
        if not phone:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "phone is required")
        try:
            payload = self.app.auth_manager.request_phone_code(account_id, phone)
        except Exception as exc:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
        return self.send_json(payload)

    def _handle_post_account_auth_phone_submit(self, account_id: str, body: Dict[str, Any]) -> None:
        code = str(body.get("code", "")).strip()
        if not code:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "code is required")
        try:
            payload = self.app.auth_manager.submit_phone_code(account_id, code)
        except Exception as exc:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
        return self.send_json(payload)

    def _handle_post_account_auth_password(self, account_id: str, body: Dict[str, Any]) -> None:
        password = str(body.get("password", "")).strip()
        if not password:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "password is required")
        try:
            payload = self.app.auth_manager.submit_password(account_id, password)
        except Exception as exc:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
        return self.send_json(payload)

    def _handle_account_channel_add(self, account_id: str, channel_id: str, name: Optional[str]) -> None:
        channel_id = channel_id.lstrip("@")
        if not channel_id:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required")
        store = get_account_store(DATA_DIR, account_id)
        def mutate(state: Dict[str, Any]) -> None:
            if channel_id not in state.setdefault("channels", {}):
                state["channels"][channel_id] = 0
            if name:
                state.setdefault("channel_names", {})[channel_id] = str(name).strip()
        store.update(mutate)
        self.app.continuous_orchestrator.refresh_account(account_id)
        return self.send_json({"ok": True, "channel_id": channel_id})

    def _handle_account_channel_remove(self, account_id: str, channel_id: str) -> None:
        if not channel_id:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "channel_id is required")
        store = get_account_store(DATA_DIR, account_id)
        existed = [False]
        def mutate(state: Dict[str, Any]) -> None:
            chans = state.get("channels", {})
            if channel_id in chans:
                existed[0] = True
                del chans[channel_id]
            state.get("channel_names", {}).pop(channel_id, None)
        store.update(mutate)
        self.app.continuous_orchestrator.refresh_account(account_id)
        return self.send_json({"ok": existed[0], "channel_id": channel_id})

    def _handle_account_job_scrape(self, account_id: str, body: Dict[str, Any]) -> None:
        channel_id = body.get("channel_id")
        if channel_id:
            job = self.app.job_runner.create_job(
                "scrape_channel",
                f"Scrape channel {channel_id} [{account_id}]",
                {"channel_id": str(channel_id), "account_id": account_id},
            )
        else:
            job = self.app.job_runner.create_job(
                "scrape_all",
                f"Scrape all tracked channels [{account_id}]",
                {"account_id": account_id},
            )
        return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

    def _handle_account_job_export(self, account_id: str) -> None:
        job = self.app.job_runner.create_job(
            "export_all",
            f"Export all tracked channels [{account_id}]",
            {"account_id": account_id},
        )
        return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

    def _handle_account_job_export_channel(self, account_id: str, body: Dict[str, Any]) -> None:
        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(
            "export_channel",
            f"Export channel {channel_id} [{account_id}]",
            {"channel_id": channel_id, "account_id": account_id},
        )
        return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

    def _handle_account_job_rescrape_media(self, account_id: str, body: Dict[str, Any]) -> None:
        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} [{account_id}]",
            {"channel_id": channel_id, "account_id": account_id},
        )
        return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

    def _handle_account_job_fix_media(self, account_id: str, body: Dict[str, Any]) -> None:
        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} [{account_id}]",
            {"channel_id": channel_id, "account_id": account_id},
        )
        return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

    def _handle_account_job_refresh_dialogs(self, account_id: str) -> None:
        job = self.app.job_runner.create_job(
            "refresh_dialogs",
            f"Refresh Telegram dialogs [{account_id}]",
            {"account_id": account_id},
        )
        return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

    def _handle_account_settings_media(self, account_id: str, value: bool) -> None:
        store = get_account_store(DATA_DIR, account_id)
        def mutate(state: Dict[str, Any]) -> None:
            state["scrape_media"] = value
        store.update(mutate)
        job = self.app.job_runner.create_job(
            "set_scrape_media",
            f"Update media scraping setting [{account_id}]",
            {"value": value, "account_id": account_id},
        )
        return self.send_json(job.to_dict(), status=HTTPStatus.ACCEPTED)

    def _handle_account_continuous(self, account_id: str, body: Dict[str, Any]) -> None:
        try:
            enabled = bool(body.get("enabled"))
            interval_minutes = int(body.get("interval_minutes", 1))
            channels = [
                str(item).strip()
                for item in body.get("channels", [])
                if str(item).strip()
            ]
            run_all_tracked = bool(body.get("run_all_tracked", True))
            payload = self.app.continuous_orchestrator.update_for(
                account_id=account_id,
                enabled=enabled,
                interval_minutes=interval_minutes,
                channels=channels,
                run_all_tracked=run_all_tracked,
            )
        except Exception as exc:
            return self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
        return self.send_json(payload)

    # ── DELETE ───────────────────────────────────────────────────────────

    def do_DELETE(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        path = parsed.path

        if path.startswith("/api/accounts/"):
            rest = path[len("/api/accounts/"):]
            parts = rest.split("/")
            account_id = urllib.parse.unquote(parts[0])
            if len(parts) == 1:
                return self._handle_delete_account(account_id)
            return self.send_error_json(HTTPStatus.BAD_REQUEST, "Invalid account path")

        return self.send_error_json(HTTPStatus.NOT_FOUND, "Not found")

    def _handle_delete_account(self, account_id: str) -> None:
        ids = list_accounts(DATA_DIR)
        if account_id not in ids:
            return self.send_error_json(HTTPStatus.NOT_FOUND, f"Account '{account_id}' not found")

        self.app.continuous_orchestrator.remove_account(account_id)
        self.app.auth_manager.delete_account(account_id)

        def mutate_global(state: Dict[str, Any]) -> None:
            accounts = state.setdefault("accounts", [])
            if account_id in accounts:
                accounts.remove(account_id)

        get_global_store(DATA_DIR).update(mutate_global)

        acc_dir = account_data_dir(DATA_DIR, account_id)
        session_file = Path(account_session_path(SESSION_DIR, account_id))
        if acc_dir.exists():
            import shutil
            shutil.rmtree(str(acc_dir), ignore_errors=True)
        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_sidecar.unlink()
            except OSError:
                pass

        return self.send_json({"ok": True, "account_id": account_id})

    # ── Helpers ─────────────────────────────────────────────────────────

    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:
        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")
        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": int(status)}, status=status)

    def log_message(self, format: str, *args: Any) -> None:
        logger.info(
            "%s %s — %s",
            self.command,
            self.path,
            args[1] if len(args) > 1 else "-",
        )


# ── Server ───────────────────────────────────────────────────────────────

class TelegramScraperWebServer(ThreadingHTTPServer):
    def __init__(self, server_address: tuple[str, int]):
        super().__init__(server_address, TelegramScraperRequestHandler)
        self.job_runner = JobRunner()
        self.auth_manager = TelegramAuthManager()
        self.continuous_orchestrator = ContinuousScrapeOrchestrator()
        self.legacy_account_id: Optional[str] = None
        self._detect_legacy_account()

        if START_CONTINUOUS:
            self.continuous_orchestrator.start_all()

    def _detect_legacy_account(self) -> None:
        ids = list_accounts(DATA_DIR)
        if not ids:
            legacy = load_state()
            if legacy.get("api_id") and legacy.get("api_hash"):
                import app_state as as_mod
                as_mod.migrate_legacy_state(DATA_DIR, SESSION_DIR)
                ids = list_accounts(DATA_DIR)
        if ids and len(ids) == 1:
            self.legacy_account_id = ids[0]


def run_server(host: str = DEFAULT_HOST, port: int = DEFAULT_PORT) -> None:
    server = TelegramScraperWebServer((host, port))
    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_orchestrator.stop_all()
        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:
        while not shutdown_event.is_set():
            server.handle_request()
    except KeyboardInterrupt:
        _shutdown()
    finally:
        server.server_close()
        logger.info("Server stopped.")


if __name__ == "__main__":
    run_server()