gotelegram_pro/admin-web/server.py

#!/usr/bin/env python3
"""
goTelegram Pro local web admin.

The service is intentionally bound to 127.0.0.1:1984. Operators reach it
through an SSH tunnel; it must never be exposed directly on the public network.
"""

from __future__ import annotations

import csv
import fcntl
import hashlib
import json
import mimetypes
import os
import re
import secrets
import socket
import subprocess
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any


ADMIN_DIR = Path(os.getenv("GOTELEGRAM_ADMIN_DIR", "/opt/gotelegram-admin"))
STATIC_DIR = Path(os.getenv("GOTELEGRAM_ADMIN_STATIC", str(ADMIN_DIR / "static")))

GOTELEGRAM_CONFIG = Path(os.getenv("GOTELEGRAM_CONFIG", "/opt/gotelegram/config.json"))
TELEMT_CONFIG = Path(os.getenv("TELEMT_CONFIG", "/etc/telemt/config.toml"))
HISTORY_FILE = Path(os.getenv("GOTELEGRAM_STATS_HISTORY", "/opt/gotelegram/stats_history.csv"))
USER_HISTORY_FILE = Path(os.getenv("GOTELEGRAM_USER_STATS_HISTORY", "/opt/gotelegram/user_stats_history.csv"))
CURRENT_STATS = Path(os.getenv("GOTELEGRAM_STATS_CURRENT", "/run/gotelegram/stats_current.json"))
BACKUP_DIR = Path(os.getenv("GOTELEGRAM_BACKUP_DIR", "/opt/gotelegram/backups"))
INSTALL_DIR = Path(os.getenv("GOTELEGRAM_DIR", "/opt/gotelegram"))
BOT_DIR = Path(os.getenv("GOTELEGRAM_BOT_DIR", "/opt/gotelegram-bot"))
DISABLED_USERS_FILE = Path(os.getenv("GOTELEGRAM_DISABLED_USERS", "/opt/gotelegram/disabled_users.json"))
USER_LOCK_FILE = Path(os.getenv("GOTELEGRAM_USER_LOCK", "/run/gotelegram/admin-users.lock"))
SHARED_443_CONFIG = Path(os.getenv("GOTELEGRAM_SHARED_443", "/opt/gotelegram/shared-443.json"))

HOST = os.getenv("GOTELEGRAM_ADMIN_HOST", "127.0.0.1")
PORT = int(os.getenv("GOTELEGRAM_ADMIN_PORT", "1984"))
VERSION = "2.5.0"
USER_RE = re.compile(r"^[A-Za-z0-9_.-]{1,48}$")
LANG_RE = re.compile(r"^(en|ru)$")
SENSITIVE_CONFIG_KEYS = {"secret"}
TRAFFIC_WINDOWS = {
    "15m": 15 * 60,
    "1h": 60 * 60,
    "24h": 24 * 60 * 60,
    "month": 30 * 24 * 60 * 60,
}


def utc_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def run(cmd: list[str], timeout: int = 8) -> tuple[int, str, str]:
    try:
        proc = subprocess.run(
            cmd,
            text=True,
            capture_output=True,
            timeout=timeout,
            check=False,
        )
        return proc.returncode, proc.stdout, proc.stderr
    except Exception as exc:  # pragma: no cover - system dependent
        return 125, "", str(exc)


class FileLock:
    def __init__(self, path: Path):
        self.path = path
        self.handle: Any = None

    def __enter__(self) -> "FileLock":
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.handle = self.path.open("w", encoding="utf-8")
        fcntl.flock(self.handle.fileno(), fcntl.LOCK_EX)
        return self

    def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
        if self.handle:
            fcntl.flock(self.handle.fileno(), fcntl.LOCK_UN)
            self.handle.close()


def load_json(path: Path, fallback: Any = None) -> Any:
    try:
        with path.open("r", encoding="utf-8") as fh:
            return json.load(fh)
    except Exception:
        return fallback


def save_json(path: Path, data: Any, mode: int = 0o600) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    tmp.write_text(json.dumps(data, ensure_ascii=False, indent=4) + "\n", encoding="utf-8")
    os.chmod(tmp, mode)
    tmp.replace(path)


def read_language(config: dict[str, Any] | None = None) -> str:
    config = config or load_json(GOTELEGRAM_CONFIG, {}) or {}
    lang = str(config.get("language") or config.get("lang") or "").strip().lower()
    marker = INSTALL_DIR / ".language"
    if lang not in {"en", "ru"} and marker.exists():
        try:
            lang = marker.read_text(encoding="utf-8", errors="ignore").strip().lower()[:2]
        except OSError:
            lang = ""
    return lang if lang in {"en", "ru"} else "en"


def public_config(config: dict[str, Any]) -> dict[str, Any]:
    return {key: value for key, value in config.items() if key not in SENSITIVE_CONFIG_KEYS}


def write_language(lang: str) -> dict[str, Any]:
    lang = str(lang or "").strip().lower()
    if not LANG_RE.match(lang):
        raise ValueError("unsupported language")
    config = load_json(GOTELEGRAM_CONFIG, {}) or {}
    if not isinstance(config, dict):
        config = {}
    config["language"] = lang
    config["updated_at"] = utc_now()
    save_json(GOTELEGRAM_CONFIG, config)
    INSTALL_DIR.mkdir(parents=True, exist_ok=True)
    (INSTALL_DIR / ".language").write_text(lang + "\n", encoding="utf-8")
    bot_env = BOT_DIR / ".env"
    if bot_env.exists():
        lines = bot_env.read_text(encoding="utf-8", errors="ignore").splitlines()
        found = False
        out = []
        for line in lines:
            if line.startswith("BOT_LANG="):
                out.append(f"BOT_LANG={lang}")
                found = True
            else:
                out.append(line)
        if not found:
            out.append(f"BOT_LANG={lang}")
        bot_env.write_text("\n".join(out).rstrip() + "\n", encoding="utf-8")
        os.chmod(bot_env, 0o600)
    return {"language": lang}


def read_telemt_users() -> dict[str, str]:
    if not TELEMT_CONFIG.exists():
        return {}
    users: dict[str, str] = {}
    in_users = False
    for raw in TELEMT_CONFIG.read_text(encoding="utf-8", errors="ignore").splitlines():
        line = raw.strip()
        if line == "[access.users]":
            in_users = True
            continue
        if in_users and line.startswith("["):
            break
        if not in_users or not line or line.startswith("#") or "=" not in line:
            continue
        name, value = line.split("=", 1)
        name = parse_toml_key(name)
        value = value.strip().split("#", 1)[0].strip()
        if value.startswith('"') and '"' in value[1:]:
            value = value[1:].split('"', 1)[0]
        elif value.startswith("'") and "'" in value[1:]:
            value = value[1:].split("'", 1)[0]
        if USER_RE.match(name) and value:
            users[name] = value
    return users


def read_disabled_users() -> dict[str, str]:
    raw = load_json(DISABLED_USERS_FILE, {}) or {}
    if not isinstance(raw, dict):
        return {}
    users = raw.get("users") if isinstance(raw.get("users"), dict) else raw
    if not isinstance(users, dict):
        return {}
    clean: dict[str, str] = {}
    for name, secret in users.items():
        if name in {"version", "updated_at"}:
            continue
        name_s = str(name).strip()
        secret_s = str(secret or "").strip()
        if USER_RE.match(name_s) and secret_s:
            clean[name_s] = secret_s
    return clean


def write_disabled_users(users: dict[str, str]) -> None:
    payload = {
        "version": 1,
        "updated_at": utc_now(),
        "users": {name: users[name] for name in sorted(users)},
    }
    save_json(DISABLED_USERS_FILE, payload)


def read_user_records() -> dict[str, dict[str, Any]]:
    active = read_telemt_users()
    disabled = read_disabled_users()
    records: dict[str, dict[str, Any]] = {}
    for name, secret in disabled.items():
        records[name] = {"secret": secret, "enabled": False}
    for name, secret in active.items():
        records[name] = {"secret": secret, "enabled": True}
    return records


def _ordered_user_lines(users: dict[str, str]) -> list[str]:
    names = []
    if "main" in users:
        names.append("main")
    names.extend(sorted(n for n in users if n != "main"))
    return [f'{quote_toml_key(name)} = "{users[name]}"' for name in names]


def parse_toml_key(raw: str) -> str:
    key = raw.strip()
    if len(key) >= 2 and key[0] == key[-1] == '"':
        try:
            return json.loads(key)
        except json.JSONDecodeError:
            return key[1:-1].replace('\\"', '"').replace("\\\\", "\\")
    if len(key) >= 2 and key[0] == key[-1] == "'":
        return key[1:-1]
    return key


def quote_toml_key(name: str) -> str:
    escaped = name.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'


def write_telemt_users(users: dict[str, str]) -> None:
    TELEMT_CONFIG.parent.mkdir(parents=True, exist_ok=True)
    lines = TELEMT_CONFIG.read_text(encoding="utf-8", errors="ignore").splitlines() if TELEMT_CONFIG.exists() else []
    rendered = _ordered_user_lines(users)
    out: list[str] = []
    in_users = False
    found = False

    for raw in lines:
        if raw.strip() == "[access.users]":
            found = True
            in_users = True
            out.append(raw)
            out.extend(rendered)
            continue
        if in_users and raw.strip().startswith("["):
            in_users = False
        if in_users:
            continue
        out.append(raw)

    if not found:
        if out and out[-1].strip():
            out.append("")
        out.append("[access.users]")
        out.extend(rendered)

    tmp = TELEMT_CONFIG.with_name(TELEMT_CONFIG.name + ".tmp")
    tmp.write_text("\n".join(out).rstrip() + "\n", encoding="utf-8")
    os.chmod(tmp, 0o600)
    tmp.replace(TELEMT_CONFIG)


def restart_service(name: str) -> bool:
    code, _, _ = run(["systemctl", "restart", name], timeout=25)
    if code != 0:
        return False
    if name == "telemt":
        return wait_tcp_port(read_telemt_port(), timeout=90)
    return True


def request_service_restart(name: str) -> bool:
    code, _, _ = run(["systemctl", "--no-block", "restart", name], timeout=5)
    return code == 0


def service_status(name: str) -> str:
    code, stdout, _ = run(["systemctl", "is-active", name], timeout=3)
    value = stdout.strip()
    if code == 0 and value == "active":
        return "running"
    code, stdout, _ = run(["systemctl", "list-unit-files", f"{name}.service", "--no-legend"], timeout=3)
    if code != 0 or not stdout.strip():
        return "not_installed"
    if value in {"failed", "inactive", "activating", "deactivating"}:
        return value
    return "stopped"


def read_telemt_port() -> int:
    if not TELEMT_CONFIG.exists():
        return 443
    in_server = False
    for raw in TELEMT_CONFIG.read_text(encoding="utf-8", errors="ignore").splitlines():
        line = raw.strip()
        if line == "[server]":
            in_server = True
            continue
        if in_server and line.startswith("["):
            break
        if in_server and line.startswith("port") and "=" in line:
            try:
                return int(line.split("=", 1)[1].strip().split("#", 1)[0])
            except ValueError:
                return 443
    return 443


def _is_port_addr(value: str, port: int) -> bool:
    token = value.strip()
    if token.startswith("[") and "]:" in token:
        return token.rsplit(":", 1)[-1] == str(port)
    return token.rsplit(":", 1)[-1] == str(port) if ":" in token else False


def _process_role(process: str) -> str:
    lowered = process.lower()
    if "telemt" in lowered or "mtproto" in lowered:
        return "mtproxy"
    if "nginx" in lowered or "apache" in lowered or "caddy" in lowered:
        return "site"
    if "xray" in lowered or "x-ui" in lowered or "3x-ui" in lowered or "xui" in lowered:
        return "xray"
    if "amnezia" in lowered or "awg" in lowered or "wireguard" in lowered or re.search(r"\bwg\b", lowered):
        return "amneziawg"
    return "other"


def parse_ss_listeners(output: str, proto: str, port: int = 443) -> list[dict[str, Any]]:
    listeners: list[dict[str, Any]] = []
    seen: set[tuple[str, str, str]] = set()
    for line in output.splitlines():
        parts = line.split()
        address = next((part for part in parts if _is_port_addr(part, port)), "")
        if not address:
            continue
        matches = re.findall(r'\("([^"]+)",pid=(\d+)', line)
        if matches:
            process_names = []
            pids = []
            for proc, pid in matches:
                if proc not in process_names:
                    process_names.append(proc)
                if pid not in pids:
                    pids.append(pid)
            process = ", ".join(process_names)
            pid_text = ", ".join(pids)
        else:
            process = "unknown"
            pid_text = ""
        key = (proto, address, process)
        if key in seen:
            continue
        seen.add(key)
        listeners.append({
            "proto": proto.upper(),
            "address": address,
            "process": process,
            "pid": pid_text,
            "role": _process_role(process),
        })
    return listeners


def collect_port_listeners(port: int) -> tuple[list[dict[str, Any]], list[str]]:
    listeners: list[dict[str, Any]] = []
    errors: list[str] = []
    for proto, args in {
        "tcp": ["ss", "-H", "-ltnp"],
        "udp": ["ss", "-H", "-lunp"],
    }.items():
        code, stdout, stderr = run(args, timeout=2)
        if code == 0:
            listeners.extend(parse_ss_listeners(stdout, proto, port))
        elif stderr.strip():
            errors.append(stderr.strip())
    listeners.sort(key=lambda item: (item["proto"], item["address"], item["process"]))
    return listeners, errors


def read_telemt_edge_settings() -> dict[str, Any]:
    settings: dict[str, Any] = {"tls_domain": "", "mask_port": 0, "dns_overrides": []}
    if not TELEMT_CONFIG.exists():
        return settings
    section = ""
    for raw in TELEMT_CONFIG.read_text(encoding="utf-8", errors="ignore").splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("[") and line.endswith("]"):
            section = line.strip("[]")
            continue
        if "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip().split("#", 1)[0].strip()
        if section == "censorship" and key == "tls_domain":
            settings["tls_domain"] = value.strip('"').strip("'")
        elif section == "censorship" and key == "mask_port":
            try:
                settings["mask_port"] = int(value)
            except ValueError:
                settings["mask_port"] = 0
        elif section == "network" and key == "dns_overrides":
            settings["dns_overrides"] = re.findall(r'"([^"]+)"', value)
    return settings


def load_shared443_config() -> dict[str, Any]:
    raw = load_json(SHARED_443_CONFIG, {}) or {}
    if not isinstance(raw, dict):
        return {}
    routes = raw.get("xray_routes") if isinstance(raw.get("xray_routes"), list) else []
    clean_routes = []
    for item in routes:
        if not isinstance(item, dict):
            continue
        public = str(item.get("public") or item.get("domain") or "").strip()
        target = str(item.get("target") or "").strip()
        if public and target:
            clean_routes.append({"public": public, "target": target})
    return {
        "enabled": bool(raw.get("enabled")),
        "dispatcher": str(raw.get("dispatcher") or "nginx-stream"),
        "public_port": _int_value(raw.get("public_port") or 443) or 443,
        "telemt_target": str(raw.get("telemt_target") or "127.0.0.1:7443"),
        "site_target": str(raw.get("site_target") or ""),
        "xray_routes": clean_routes,
        "updated_at": str(raw.get("updated_at") or ""),
    }


def listener_for_target(target: str) -> dict[str, Any] | None:
    try:
        port = int(target.rsplit(":", 1)[-1])
    except ValueError:
        return None
    listeners, _ = collect_port_listeners(port)
    return listeners[0] if listeners else None


def routed_behind_443() -> list[dict[str, Any]]:
    config = load_json(GOTELEGRAM_CONFIG, {}) or {}
    mode = str(config.get("mode") or "")
    domain = str(config.get("domain") or "")
    settings = read_telemt_edge_settings()
    shared = load_shared443_config()
    mask_port = int(settings.get("mask_port") or 0)
    tls_domain = str(settings.get("tls_domain") or domain)
    routes: list[dict[str, Any]] = []
    if shared.get("enabled"):
        telemt_target = str(shared.get("telemt_target") or "127.0.0.1:7443")
        telemt_listener = listener_for_target(telemt_target)
        routes.append({
            "role": "mtproxy",
            "proto": "MTProxy",
            "public": f"{domain or tls_domain or 'default'}:443",
            "target": telemt_target,
            "process": (telemt_listener or {}).get("process") or "telemt",
            "pid": (telemt_listener or {}).get("pid") or "",
            "status": service_status("telemt"),
            "via": "nginx stream ssl_preread",
            "tls_domain": tls_domain,
            "details": ["default -> telemt"] if not shared.get("xray_routes") else [],
        })
        for item in shared.get("xray_routes", []):
            target = item.get("target", "")
            listener = listener_for_target(target)
            public = item.get("public", "")
            if public and ":" not in public:
                public = f"{public}:443"
            routes.append({
                "role": "xray",
                "proto": "VLESS",
                "public": public or "xray:443",
                "target": target,
                "process": (listener or {}).get("process") or "xray",
                "pid": (listener or {}).get("pid") or "",
                "status": "running" if listener else "not_installed",
                "via": "nginx stream ssl_preread",
                "tls_domain": public.split(":", 1)[0] if public else "",
                "details": [],
            })
    if mode == "pro" and domain and mask_port and mask_port != 443:
        internal, _ = collect_port_listeners(mask_port)
        site_listener = next((item for item in internal if item.get("role") == "site"), None)
        routes.append({
            "role": "site",
            "proto": "HTTPS",
            "public": f"{domain}:443",
            "target": f"127.0.0.1:{mask_port}",
            "process": (site_listener or {}).get("process") or "nginx",
            "pid": (site_listener or {}).get("pid") or "",
            "status": service_status("nginx"),
            "via": "telemt dns_overrides",
            "tls_domain": tls_domain,
            "details": settings.get("dns_overrides") or [],
        })
    return routes


def port_443_status() -> dict[str, Any]:
    listeners, errors = collect_port_listeners(443)
    shared = load_shared443_config()
    if shared.get("enabled"):
        for item in listeners:
            if item.get("role") == "site" and "nginx" in str(item.get("process", "")).lower():
                item["role"] = "edge"
                item["details"] = "nginx stream ssl_preread"
    return {
        "checked_at": int(time.time()),
        "configured_port": read_telemt_port(),
        "listeners": listeners,
        "routes": routed_behind_443(),
        "shared_443": shared,
        "ok": not errors,
        "error": "; ".join(errors[:2]),
    }


def wait_tcp_port(port: int, timeout: int = 90) -> bool:
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if service_status("telemt") not in {"running", "activating"}:
            return False
        try:
            with socket.create_connection(("127.0.0.1", port), timeout=0.6):
                return True
        except OSError:
            time.sleep(1)
    return False


def public_ip() -> str:
    code, stdout, _ = run(["curl", "-s", "-4", "--max-time", "3", "https://api.ipify.org"], timeout=5)
    ip = stdout.strip()
    if code == 0 and re.match(r"^\d{1,3}(\.\d{1,3}){3}$", ip):
        return ip
    code, stdout, _ = run(["hostname", "-I"], timeout=3)
    return stdout.split()[0] if code == 0 and stdout.split() else "0.0.0.0"


def proxy_link(secret: str) -> str:
    config = load_json(GOTELEGRAM_CONFIG, {}) or {}
    mode = str(config.get("mode", "lite"))
    port = int(config.get("port", 443) or 443)
    domain = str(config.get("domain", "") or "")
    mask_host = str(config.get("mask_host", "") or "")

    if mode == "pro" and domain:
        host_hex = domain.encode().hex()
        return f"tg://proxy?server={domain}&port={port}&secret=ee{secret}{host_hex}"

    server = public_ip()
    if mask_host:
        host_hex = mask_host.encode().hex()
        return f"tg://proxy?server={server}&port={port}&secret=ee{secret}{host_hex}"
    return f"tg://proxy?server={server}&port={port}&secret={secret}"


def telemt_api(path: str) -> Any:
    try:
        with urllib.request.urlopen(f"http://127.0.0.1:9091{path}", timeout=1.8) as resp:
            payload = resp.read(256 * 1024)
        return json.loads(payload.decode("utf-8"))
    except Exception:
        return None


def site_status(config: dict[str, Any] | None = None) -> dict[str, Any]:
    config = config or load_json(GOTELEGRAM_CONFIG, {}) or {}
    host = str(config.get("domain") or "").strip()
    if not host:
        return {"host": "", "url": "", "http_code": 0, "ok": False, "checked": False, "error": "domain_missing"}
    if not re.match(r"^[A-Za-z0-9.-]{1,253}$", host) or ".." in host or host.startswith(".") or host.endswith("."):
        return {"host": host, "url": "", "http_code": 0, "ok": False, "checked": False, "error": "invalid_domain"}
    url = f"https://{host}/"
    code, stdout, stderr = run(["curl", "-k", "-L", "-sS", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "8", url], timeout=10)
    raw_code = stdout.strip()
    try:
        http_code = int(raw_code)
    except ValueError:
        http_code = 0
    return {
        "host": host,
        "url": url,
        "http_code": http_code,
        "ok": code == 0 and http_code == 200,
        "checked": True,
        "error": "" if code == 0 else (stderr.strip() or f"curl exit {code}"),
        "checked_at": int(time.time()),
    }


def load_stats_history(limit: int | None = 240) -> list[dict[str, int]]:
    if not HISTORY_FILE.exists():
        return []
    rows: list[dict[str, int]] = []
    try:
        with HISTORY_FILE.open("r", encoding="utf-8", newline="") as fh:
            for row in csv.DictReader(fh):
                try:
                    rows.append({
                        "epoch": int(row.get("epoch") or 0),
                        "proxy_bytes": int(row.get("proxy_bytes") or 0),
                        "site_bytes": int(row.get("site_bytes") or 0),
                    })
                except ValueError:
                    continue
    except OSError:
        return []
    if limit:
        rows = rows[-limit:]
    previous = None
    enriched: list[dict[str, int]] = []
    for row in rows:
        item = dict(row)
        if previous:
            item["proxy_delta"] = max(0, row["proxy_bytes"] - previous["proxy_bytes"])
            item["site_delta"] = max(0, row["site_bytes"] - previous["site_bytes"])
        else:
            item["proxy_delta"] = 0
            item["site_delta"] = 0
        enriched.append(item)
        previous = row
    return enriched


def _int_value(value: Any) -> int:
    try:
        return int(value or 0)
    except (TypeError, ValueError):
        return 0


def load_user_stats_history(name: str | None = None, limit: int | None = 240) -> list[dict[str, Any]]:
    if not USER_HISTORY_FILE.exists():
        return []
    rows: list[dict[str, Any]] = []
    try:
        with USER_HISTORY_FILE.open("r", encoding="utf-8", newline="") as fh:
            for row in csv.DictReader(fh):
                user = str(row.get("user") or "").strip()
                if name is not None and user != name:
                    continue
                if not USER_RE.match(user):
                    continue
                rows.append({
                    "epoch": _int_value(row.get("epoch")),
                    "user": user,
                    "total_octets": _int_value(row.get("total_octets")),
                    "current_connections": _int_value(row.get("current_connections")),
                    "active_unique_ips": _int_value(row.get("active_unique_ips")),
                    "recent_unique_ips": _int_value(row.get("recent_unique_ips")),
                })
    except OSError:
        return []
    rows.sort(key=lambda item: (item["user"], item["epoch"]))
    if limit and name is not None:
        rows = rows[-limit:]

    previous_by_user: dict[str, dict[str, Any]] = {}
    enriched: list[dict[str, Any]] = []
    for row in rows:
        item = dict(row)
        previous = previous_by_user.get(row["user"])
        item["total_delta"] = max(0, row["total_octets"] - previous["total_octets"]) if previous else 0
        enriched.append(item)
        previous_by_user[row["user"]] = row
    if limit and name is None:
        enriched = enriched[-limit:]
    return enriched


def latest_user_stats() -> dict[str, dict[str, Any]]:
    latest: dict[str, dict[str, Any]] = {}
    for row in load_user_stats_history(limit=None):
        if row["epoch"] >= latest.get(row["user"], {}).get("epoch", 0):
            latest[row["user"]] = row
    return latest


def runtime_user_traffic(name: str, enabled: bool = True) -> dict[str, Any]:
    if not enabled:
        return {"ok": False, "enabled": False, "total_octets": 0, "current_connections": 0, "active_unique_ips": 0, "recent_unique_ips": 0}
    payload = telemt_api(f"/v1/users/{urllib.parse.quote(name, safe='')}")
    data = payload.get("data", payload) if isinstance(payload, dict) else {}
    if not isinstance(data, dict):
        data = {}
    return {
        "ok": bool(payload),
        "enabled": True,
        "total_octets": _int_value(data.get("total_octets")),
        "current_connections": _int_value(data.get("current_connections")),
        "active_unique_ips": _int_value(data.get("active_unique_ips")),
        "recent_unique_ips": _int_value(data.get("recent_unique_ips")),
        "in_runtime": bool(data.get("in_runtime")) if data else False,
    }


def history_limit_for_range(range_key: str) -> int:
    return {
        "15m": 180,
        "1h": 240,
        "24h": 1800,
        "month": 50000,
    }.get(range_key, 240)


def normalize_range(range_key: str) -> str:
    return range_key if range_key in TRAFFIC_WINDOWS else "1h"


def filter_history_by_range(rows: list[dict[str, int]], range_key: str) -> list[dict[str, int]]:
    if not rows:
        return []
    seconds = TRAFFIC_WINDOWS[normalize_range(range_key)]
    latest = max(row.get("epoch", 0) for row in rows)
    cutoff = latest - seconds
    return [row for row in rows if row.get("epoch", 0) >= cutoff]


def traffic_interval_summaries(rows: list[dict[str, int]]) -> list[dict[str, Any]]:
    if not rows:
        return [
            {"range": key, "points": 0, "from": 0, "to": 0, "proxy_delta": 0, "site_delta": 0, "proxy_total": 0, "site_total": 0}
            for key in TRAFFIC_WINDOWS
        ]
    latest = max(row.get("epoch", 0) for row in rows)
    summaries = []
    for key, seconds in TRAFFIC_WINDOWS.items():
        window = [row for row in rows if row.get("epoch", 0) >= latest - seconds]
        if not window:
            summaries.append({"range": key, "points": 0, "from": 0, "to": latest, "proxy_delta": 0, "site_delta": 0, "proxy_total": 0, "site_total": 0})
            continue
        first = window[0]
        last = window[-1]
        summaries.append({
            "range": key,
            "points": len(window),
            "from": first.get("epoch", 0),
            "to": last.get("epoch", 0),
            "proxy_delta": sum(max(0, int(item.get("proxy_delta", 0))) for item in window),
            "site_delta": sum(max(0, int(item.get("site_delta", 0))) for item in window),
            "proxy_total": int(last.get("proxy_bytes", 0)),
            "site_total": int(last.get("site_bytes", 0)),
        })
    return summaries


def user_traffic_interval_summaries(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    if not rows:
        return [
            {"range": key, "points": 0, "from": 0, "to": 0, "total_delta": 0, "total_octets": 0}
            for key in TRAFFIC_WINDOWS
        ]
    latest = max(row.get("epoch", 0) for row in rows)
    summaries = []
    for key, seconds in TRAFFIC_WINDOWS.items():
        window = [row for row in rows if row.get("epoch", 0) >= latest - seconds]
        if not window:
            summaries.append({"range": key, "points": 0, "from": 0, "to": latest, "total_delta": 0, "total_octets": 0})
            continue
        first = window[0]
        last = window[-1]
        summaries.append({
            "range": key,
            "points": len(window),
            "from": first.get("epoch", 0),
            "to": last.get("epoch", 0),
            "total_delta": sum(max(0, int(item.get("total_delta", 0))) for item in window),
            "total_octets": int(last.get("total_octets", 0)),
        })
    return summaries


def count_history_rows() -> int:
    if not HISTORY_FILE.exists():
        return 0
    try:
        with HISTORY_FILE.open("r", encoding="utf-8", errors="ignore") as fh:
            return sum(1 for line in fh if line and line[0].isdigit())
    except OSError:
        return 0


def count_user_history_rows(name: str | None = None) -> int:
    if not USER_HISTORY_FILE.exists():
        return 0
    try:
        with USER_HISTORY_FILE.open("r", encoding="utf-8", errors="ignore") as fh:
            if name is None:
                return sum(1 for line in fh if line and line[0].isdigit())
            return sum(1 for line in fh if line.startswith(tuple(str(d) for d in range(10))) and f",{name}," in line)
    except OSError:
        return 0


def stats_status(current: dict[str, Any] | None = None, history: list[dict[str, int]] | None = None) -> dict[str, Any]:
    current = current if current is not None else (load_json(CURRENT_STATS, {}) or {})
    history = history if history is not None else load_stats_history(limit=2)
    service = service_status("gotelegram-stats")
    now = int(time.time())
    ts = int(current.get("ts") or 0) if isinstance(current, dict) else 0
    age = max(0, now - ts) if ts else None
    error = str(current.get("error") or "") if isinstance(current, dict) else ""
    history_rows = count_history_rows()
    if error:
        health = "error"
    elif service == "running" and current and age is not None and age <= 180:
        health = "ok"
    elif service == "running":
        health = "stale"
    elif service == "not_installed":
        health = "not_installed"
    else:
        health = "stopped"
    return {
        "health": health,
        "service": service,
        "current_exists": CURRENT_STATS.exists(),
        "history_exists": HISTORY_FILE.exists(),
        "history_rows": history_rows,
        "history_points": len(history or []),
        "last_ts": ts,
        "age_seconds": age,
        "error": error,
    }


def run_stats_action(action: str) -> tuple[bool, str, dict[str, Any]]:
    if action == "repair":
        body = (
            "source /opt/gotelegram/lib/common.sh; "
            "source /opt/gotelegram/lib/i18n.sh; "
            "source /opt/gotelegram/lib/stats.sh; "
            "load_language \"$(detect_language 2>/dev/null || echo en)\"; "
            "install_stats_collector; "
            "stats_collect"
        )
        timeout = 180
    else:
        body = (
            "source /opt/gotelegram/lib/common.sh; "
            "source /opt/gotelegram/lib/stats.sh; "
            "stats_init >/dev/null 2>&1 || true; "
            "stats_collect"
        )
        timeout = 30
    code, stdout, stderr = run(["bash", "-lc", body], timeout=timeout)
    message = (stdout.strip().splitlines()[-1:] or stderr.strip().splitlines()[-1:] or [""])[0]
    current = load_json(CURRENT_STATS, {}) or {}
    history = load_stats_history()
    return code == 0, message, {"current": current, "history": history, "status": stats_status(current, history)}


def list_backups() -> list[dict[str, Any]]:
    if not BACKUP_DIR.exists():
        return []
    items = []
    for path in sorted(BACKUP_DIR.glob("*.tar.gz*"), key=lambda p: p.stat().st_mtime, reverse=True):
        if path.name.endswith(".sha256"):
            continue
        try:
            st = path.stat()
        except OSError:
            continue
        items.append({
            "name": path.name,
            "path": str(path),
            "size": st.st_size,
            "mtime": int(st.st_mtime),
            "encrypted": path.name.endswith(".enc"),
        })
    return items[:30]


def create_backup() -> tuple[bool, str]:
    script = (
        "source /opt/gotelegram/lib/common.sh; "
        "source /opt/gotelegram/lib/i18n.sh; "
        "source /opt/gotelegram/lib/telemt.sh; "
        "source /opt/gotelegram/lib/website.sh; "
        "source /opt/gotelegram/lib/backup.sh; "
        "load_language \"$(detect_language 2>/dev/null || echo en)\"; "
        "create_backup \"\""
    )
    code, stdout, stderr = run(["bash", "-lc", script], timeout=180)
    text = (stdout.strip().splitlines()[-1:] or stderr.strip().splitlines()[-1:] or [""])[0]
    return code == 0, text


def read_log_payload(service: str) -> dict[str, Any]:
    allowed = {"telemt", "nginx", "gotelegram-bot", "gotelegram-stats", "gotelegram-admin"}
    if service not in allowed:
        raise ValueError("unsupported service")
    code, stdout, stderr = run(["journalctl", "-u", service, "-n", "180", "--no-pager", "-o", "short-iso"], timeout=10)
    text = stdout if code == 0 else stderr
    lines = text.splitlines()
    if code == 0 and not lines:
        text = f"No journal entries for {service}."
        lines = [text]
    return {
        "service": service,
        "ok": code == 0,
        "exit_code": code,
        "line_count": len(lines),
        "text": text,
    }


def user_payload(
    name: str,
    secret: str,
    enabled: bool = True,
    include_runtime: bool = False,
    traffic_snapshot: dict[str, Any] | None = None,
) -> dict[str, Any]:
    item: dict[str, Any] = {
        "name": name,
        "secret": secret,
        "link": proxy_link(secret),
        "main": name == "main",
        "enabled": bool(enabled),
    }
    if traffic_snapshot:
        item["traffic"] = {
            "epoch": traffic_snapshot.get("epoch", 0),
            "total_octets": traffic_snapshot.get("total_octets", 0),
            "current_connections": traffic_snapshot.get("current_connections", 0),
            "active_unique_ips": traffic_snapshot.get("active_unique_ips", 0),
            "recent_unique_ips": traffic_snapshot.get("recent_unique_ips", 0),
        }
    if include_runtime and enabled:
        item["runtime"] = telemt_api(f"/v1/users/{urllib.parse.quote(name, safe='')}")
    return item


def overview_payload() -> dict[str, Any]:
    config = load_json(GOTELEGRAM_CONFIG, {}) or {}
    language = read_language(config)
    users = read_user_records()
    current = load_json(CURRENT_STATS, {}) or {}
    history = load_stats_history()
    summary = telemt_api("/v1/stats/summary")
    services = {
        "telemt": service_status("telemt"),
        "nginx": service_status("nginx"),
        "bot": service_status("gotelegram-bot"),
        "stats": service_status("gotelegram-stats"),
        "admin": service_status("gotelegram-admin"),
    }
    return {
        "version": VERSION,
        "time": utc_now(),
        "language": language,
        "admin_bind": {"host": HOST, "port": PORT},
        "config": public_config(config),
        "site_status": site_status(config),
        "users_count": len(users),
        "services": services,
        "port_443": port_443_status(),
        "stats_current": current,
        "stats_history": history,
        "stats_status": stats_status(current, history),
        "runtime_summary": summary,
        "backups": list_backups(),
    }


class AdminHandler(BaseHTTPRequestHandler):
    server_version = "goTelegramProAdmin/2.5.0"

    def log_message(self, fmt: str, *args: Any) -> None:
        print("%s - %s" % (self.address_string(), fmt % args))

    def send_json(self, payload: Any, status: int = 200) -> None:
        body = 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("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def send_error_json(self, status: int, message: str) -> None:
        self.send_json({"ok": False, "error": message}, status)

    def read_json_body(self) -> Any:
        length = int(self.headers.get("Content-Length", "0") or 0)
        if length > 1024 * 1024:
            raise ValueError("request body too large")
        if length <= 0:
            return {}
        return json.loads(self.rfile.read(length).decode("utf-8"))

    def require_write_guard(self) -> bool:
        if self.command in {"POST", "PUT", "PATCH", "DELETE"} and self.headers.get("X-GoTelegram-Admin") != "1":
            self.send_error_json(403, "missing write guard")
            return False
        return True

    def route_get_api(self, parsed: urllib.parse.ParseResult) -> None:
        path = parsed.path
        if path == "/api/overview":
            self.send_json({"ok": True, "data": overview_payload()})
        elif path == "/api/users":
            users = read_user_records()
            latest = latest_user_stats()
            items = []
            for name in sorted(users, key=lambda item: (item != "main", item)):
                record = users[name]
                items.append(user_payload(name, record["secret"], record["enabled"], traffic_snapshot=latest.get(name)))
            self.send_json({"ok": True, "data": items})
        elif path.startswith("/api/users/") and path.endswith("/traffic"):
            name = urllib.parse.unquote(path[len("/api/users/"):-len("/traffic")])
            users = read_user_records()
            if name not in users:
                self.send_error_json(404, "user not found")
                return
            qs = urllib.parse.parse_qs(parsed.query)
            range_key = normalize_range(qs.get("range", ["1h"])[0])
            all_history = load_user_stats_history(name, limit=history_limit_for_range("month"))
            history = filter_history_by_range(all_history[-history_limit_for_range(range_key):], range_key)
            current = runtime_user_traffic(name, bool(users[name].get("enabled")))
            self.send_json({
                "ok": True,
                "data": {
                    "name": name,
                    "range": range_key,
                    "current": current,
                    "history": history,
                    "summary_rows": user_traffic_interval_summaries(all_history),
                    "status": {
                        "history_exists": USER_HISTORY_FILE.exists(),
                        "history_rows": count_user_history_rows(name),
                        "history_points": len(history),
                        "last_ts": history[-1]["epoch"] if history else 0,
                        "runtime_ok": current.get("ok", False),
                    },
                },
            })
        elif path.startswith("/api/users/"):
            name = urllib.parse.unquote(path[len("/api/users/"):])
            users = read_user_records()
            if name not in users:
                self.send_error_json(404, "user not found")
                return
            record = users[name]
            self.send_json({"ok": True, "data": user_payload(name, record["secret"], record["enabled"], include_runtime=True, traffic_snapshot=latest_user_stats().get(name))})
        elif path == "/api/backups":
            self.send_json({"ok": True, "data": list_backups()})
        elif path == "/api/stats":
            qs = urllib.parse.parse_qs(parsed.query)
            range_key = normalize_range(qs.get("range", ["1h"])[0])
            current = load_json(CURRENT_STATS, {}) or {}
            all_history = load_stats_history(limit=history_limit_for_range("month"))
            history = filter_history_by_range(all_history[-history_limit_for_range(range_key):], range_key)
            self.send_json({
                "ok": True,
                "data": {
                    "range": range_key,
                    "current": current,
                    "history": history,
                    "summary_rows": traffic_interval_summaries(all_history),
                    "status": stats_status(current, history),
                },
            })
        elif path == "/api/site/check":
            self.send_json({"ok": True, "data": site_status()})
        elif path == "/api/logs":
            qs = urllib.parse.parse_qs(parsed.query)
            service = qs.get("service", ["telemt"])[0]
            try:
                payload = read_log_payload(service)
            except ValueError:
                self.send_error_json(400, "unsupported service")
                return
            self.send_json({"ok": True, "data": payload})
        else:
            self.send_error_json(404, "not found")

    def route_post_api(self, parsed: urllib.parse.ParseResult) -> None:
        if not self.require_write_guard():
            return
        path = parsed.path
        try:
            body = self.read_json_body()
        except Exception as exc:
            self.send_error_json(400, str(exc))
            return

        if path == "/api/users":
            name = str(body.get("name", "")).strip()
            if not USER_RE.match(name):
                self.send_error_json(400, "invalid user name")
                return
            try:
                with FileLock(USER_LOCK_FILE):
                    records = read_user_records()
                    if name in records:
                        self.send_error_json(409, "user already exists")
                        return
                    users = read_telemt_users()
                    seed = f"{name}:{time.time()}:{secrets.token_hex(32)}".encode()
                    secret = hashlib.sha256(seed).hexdigest()[:32]
                    users[name] = secret
                    write_telemt_users(users)
            except Exception as exc:
                self.send_error_json(500, f"failed to save config: {exc}")
                return
            restart_requested = request_service_restart("telemt")
            self.send_json({"ok": True, "data": user_payload(name, secret, True), "restart": {"mode": "async", "requested": restart_requested}})
        elif path.startswith("/api/users/") and path.endswith("/enabled"):
            name = urllib.parse.unquote(path[len("/api/users/"):-len("/enabled")])
            if name == "main":
                self.send_error_json(400, "main user cannot be disabled")
                return
            enabled = bool(body.get("enabled"))
            try:
                with FileLock(USER_LOCK_FILE):
                    active = read_telemt_users()
                    disabled = read_disabled_users()
                    records = read_user_records()
                    if name not in records:
                        self.send_error_json(404, "user not found")
                        return
                    if enabled:
                        secret = disabled.pop(name, records[name]["secret"])
                        active[name] = secret
                    else:
                        secret = active.pop(name, records[name]["secret"])
                        disabled[name] = secret
                    if enabled:
                        write_telemt_users(active)
                        write_disabled_users(disabled)
                    else:
                        write_disabled_users(disabled)
                        write_telemt_users(active)
            except Exception as exc:
                self.send_error_json(500, f"failed to save config: {exc}")
                return
            restart_requested = request_service_restart("telemt")
            self.send_json({"ok": True, "data": user_payload(name, secret, enabled), "restart": {"mode": "async", "requested": restart_requested}})
        elif path == "/api/backups":
            ok, result = create_backup()
            self.send_json({"ok": ok, "data": {"path": result, "backups": list_backups()}}, 200 if ok else 500)
        elif path == "/api/stats/collect":
            ok, message, payload = run_stats_action("collect")
            payload["message"] = message
            self.send_json({"ok": ok, "data": payload}, 200 if ok else 500)
        elif path == "/api/stats/repair":
            ok, message, payload = run_stats_action("repair")
            payload["message"] = message
            self.send_json({"ok": ok, "data": payload}, 200 if ok else 500)
        elif path == "/api/settings/language":
            try:
                lang_payload = write_language(str(body.get("language", "")))
            except Exception as exc:
                self.send_error_json(400, str(exc))
                return
            self.send_json({"ok": True, "data": lang_payload})
        elif path.startswith("/api/services/") and path.endswith("/restart"):
            service = path[len("/api/services/"):-len("/restart")]
            allowed = {"telemt", "nginx", "gotelegram-bot", "gotelegram-stats"}
            if service not in allowed:
                self.send_error_json(400, "unsupported service")
                return
            ok = restart_service(service)
            self.send_json({"ok": ok, "status": service_status(service)}, 200 if ok else 500)
        else:
            self.send_error_json(404, "not found")

    def route_delete_api(self, parsed: urllib.parse.ParseResult) -> None:
        if not self.require_write_guard():
            return
        path = parsed.path
        if not path.startswith("/api/users/"):
            self.send_error_json(404, "not found")
            return
        name = urllib.parse.unquote(path[len("/api/users/"):])
        if name == "main":
            self.send_error_json(400, "main user cannot be deleted")
            return
        try:
            with FileLock(USER_LOCK_FILE):
                active = read_telemt_users()
                disabled = read_disabled_users()
                records = read_user_records()
                if name not in records:
                    self.send_error_json(404, "user not found")
                    return
                active.pop(name, None)
                disabled.pop(name, None)
                write_telemt_users(active)
                write_disabled_users(disabled)
        except Exception as exc:
            self.send_error_json(500, f"failed to save config: {exc}")
            return
        restart_requested = request_service_restart("telemt")
        self.send_json({"ok": True, "restart": {"mode": "async", "requested": restart_requested}})

    def send_static(self, parsed: urllib.parse.ParseResult) -> None:
        rel = parsed.path.lstrip("/") or "index.html"
        if rel.startswith("api/") or ".." in rel.split("/"):
            self.send_error(404)
            return
        path = STATIC_DIR / rel
        if path.is_dir():
            path = path / "index.html"
        if not path.exists():
            path = STATIC_DIR / "index.html"
        try:
            body = path.read_bytes()
        except OSError:
            self.send_error(404)
            return
        mime = mimetypes.guess_type(str(path))[0] or "application/octet-stream"
        self.send_response(200)
        self.send_header("Content-Type", mime)
        self.send_header("Cache-Control", "no-store")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path.startswith("/api/"):
            self.route_get_api(parsed)
        else:
            self.send_static(parsed)

    def do_POST(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path.startswith("/api/"):
            self.route_post_api(parsed)
        else:
            self.send_error(404)

    def do_DELETE(self) -> None:
        parsed = urllib.parse.urlparse(self.path)
        if parsed.path.startswith("/api/"):
            self.route_delete_api(parsed)
        else:
            self.send_error(404)


def main() -> None:
    if not STATIC_DIR.exists():
        raise SystemExit(f"static dir not found: {STATIC_DIR}")
    httpd = ThreadingHTTPServer((HOST, PORT), AdminHandler)
    print(f"goTelegram Pro admin listening on http://{HOST}:{PORT}")
    httpd.serve_forever()


if __name__ == "__main__":
    main()