#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from __future__ import annotations

import argparse
import asyncio
import contextlib
import dataclasses
import json
import re
import socket
import sqlite3
import struct
import time
import traceback
import urllib.error
import urllib.parse
import urllib.request
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any

APP_DIR = Path(__file__).resolve().parent
DEFAULT_CONFIG = APP_DIR / "config.json"
DEFAULT_DB = APP_DIR / "data" / "monitor.db"


def now_ts() -> int:
    return int(time.time())


def iso(ts: int | None = None) -> str:
    return datetime.fromtimestamp(ts or now_ts()).strftime("%Y-%m-%d %H:%M:%S")


def load_json(path: Path) -> dict[str, Any]:
    if not path.exists():
        return {}
    return json.loads(path.read_text(encoding="utf-8"))


def save_json(path: Path, data: dict[str, Any]) -> 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=2) + "\n", encoding="utf-8")
    tmp.replace(path)


def deep_copy(obj: Any) -> Any:
    return json.loads(json.dumps(obj, ensure_ascii=False))


def default_config() -> dict[str, Any]:
    return {
        "interval_seconds": 300,
        "timeout_seconds": 8,
        "target_host": "www.baidu.com",
        "target_port": 443,
        "fail_threshold": 3,
        "recover_threshold": 1,
        "max_checks_per_proxy": 500,
        "db_path": str(DEFAULT_DB),
        "proxies": [],
        "notifications": {"enabled": True, "timeout_seconds": 10, "channels": []},
    }


def merge_defaults(cfg: dict[str, Any]) -> dict[str, Any]:
    base = default_config()
    base.update(cfg or {})
    notifications = base.get("notifications") or {}
    base["notifications"] = {"enabled": True, "timeout_seconds": 10, "channels": [], **notifications}
    if base.get("target_host") in (None, "", "1.1.1.1", "8.8.8.8", "google.com"):
        base["target_host"] = "www.baidu.com"
    base["target_port"] = int(base.get("target_port") or 443)
    base["interval_seconds"] = int(base.get("interval_seconds") or 300)
    base["max_checks_per_proxy"] = max(1, int(base.get("max_checks_per_proxy") or 500))
    base.setdefault("proxies", [])
    for proxy in base["proxies"]:
        if proxy.get("target_host") in (None, "", "1.1.1.1", "8.8.8.8", "google.com"):
            proxy["target_host"] = base["target_host"]
        proxy["target_port"] = int(proxy.get("target_port") or base["target_port"])
        proxy["interval_seconds"] = int(proxy.get("interval_seconds") or base["interval_seconds"])
    base["notifications"].setdefault("channels", [])
    base["notifications"]["channels"] = [normalize_channel(c) for c in base["notifications"].get("channels", [])]
    return base


def public_proxy(proxy: dict[str, Any]) -> dict[str, Any]:
    p = deep_copy(proxy)
    if p.get("password"):
        p["password_masked"] = "********"
        p["password"] = ""
    return p


def normalize_channel(channel: dict[str, Any]) -> dict[str, Any]:
    """Return canonical notification channel shape: common fields + type-specific config.

    Backward compatible with the old flat schema, but output never mixes fields
    across Bark / PushPlus / Autman / Webhook.
    """
    c = dict(channel or {})
    typ = c.get("type") or "webhook"
    cfg = dict(c.get("config") or {})
    if typ == "bark":
        cfg = {
            "server": cfg.get("server") or c.get("server") or "https://api.day.app",
            "key": cfg.get("key") or c.get("key") or "",
            "group": cfg.get("group") or c.get("group") or "SOCKS5监控",
        }
    elif typ == "pushplus":
        cfg = {
            "token": cfg.get("token") or c.get("token") or c.get("key") or "",
            "template": cfg.get("template") or c.get("template") or "txt",
            "topic": cfg.get("topic") or c.get("topic") or c.get("group") or "",
        }
    elif typ == "autman":
        cfg = {
            "url": cfg.get("url") or c.get("url") or c.get("webhook") or "",
            "access_token": cfg.get("access_token") or cfg.get("token") or c.get("access_token") or c.get("token") or "",
        }
    elif typ == "webhook":
        cfg = {
            "url": cfg.get("url") or c.get("url") or "",
            "headers": cfg.get("headers") if isinstance(cfg.get("headers"), dict) else (c.get("headers") if isinstance(c.get("headers"), dict) else {}),
            "payload": cfg.get("payload") if isinstance(cfg.get("payload"), dict) else (c.get("payload") if isinstance(c.get("payload"), dict) else {}),
        }
    return {
        "id": c.get("id") or f"ch-{uuid.uuid4().hex[:8]}",
        "type": typ,
        "name": c.get("name") or typ,
        "enabled": bool(c.get("enabled", True)),
        "config": cfg,
    }


class Store:
    def __init__(self, path: Path):
        self.path = path
        self.path.parent.mkdir(parents=True, exist_ok=True)
        self.conn = sqlite3.connect(self.path, check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        self.lock = asyncio.Lock()
        self.init()

    def init(self) -> None:
        self.conn.executescript(
            """
            create table if not exists checks(
              id integer primary key autoincrement,
              proxy_id text not null,
              ts integer not null,
              ok integer not null,
              latency_ms integer,
              error text,
              exit_ip text
            );
            create index if not exists idx_checks_proxy_ts on checks(proxy_id, ts desc);
            create table if not exists events(
              id integer primary key autoincrement,
              proxy_id text not null,
              ts integer not null,
              old_status text,
              new_status text,
              message text
            );
            create index if not exists idx_events_ts on events(ts desc);
            create table if not exists notification_logs(
              id integer primary key autoincrement,
              ts integer not null,
              channel_id text,
              channel_name text,
              ok integer not null,
              title text,
              message text,
              result text
            );
            create index if not exists idx_notification_logs_ts on notification_logs(ts desc);
            """
        )
        self.conn.commit()

    async def add_check(self, proxy_id: str, ok: bool, latency_ms: int | None, error: str | None, exit_ip: str | None, max_per_proxy: int | None = None) -> None:
        async with self.lock:
            self.conn.execute(
                "insert into checks(proxy_id,ts,ok,latency_ms,error,exit_ip) values(?,?,?,?,?,?)",
                (proxy_id, now_ts(), 1 if ok else 0, latency_ms, error, exit_ip),
            )
            if max_per_proxy:
                self.conn.execute(
                    """
                    delete from checks
                    where proxy_id=? and id not in (
                      select id from checks where proxy_id=? order by ts desc, id desc limit ?
                    )
                    """,
                    (proxy_id, proxy_id, max(1, int(max_per_proxy))),
                )
            self.conn.commit()

    async def add_event(self, proxy_id: str, old_status: str | None, new_status: str, message: str) -> None:
        async with self.lock:
            self.conn.execute(
                "insert into events(proxy_id,ts,old_status,new_status,message) values(?,?,?,?,?)",
                (proxy_id, now_ts(), old_status, new_status, message),
            )
            self.conn.commit()

    async def add_notification(self, channel: dict[str, Any], ok: bool, title: str, message: str, result: str) -> None:
        async with self.lock:
            self.conn.execute(
                "insert into notification_logs(ts,channel_id,channel_name,ok,title,message,result) values(?,?,?,?,?,?,?)",
                (now_ts(), channel.get("id"), channel.get("name") or channel.get("type"), 1 if ok else 0, title, message, result),
            )
            self.conn.commit()

    def recent_checks(self, proxy_id: str, limit: int = 50) -> list[dict[str, Any]]:
        rows = self.conn.execute("select * from checks where proxy_id=? order by ts desc, id desc limit ?", (proxy_id, limit)).fetchall()
        return [dict(r) for r in rows]

    def recent_events(self, limit: int = 50) -> list[dict[str, Any]]:
        rows = self.conn.execute("select * from events order by ts desc limit ?", (limit,)).fetchall()
        return [dict(r) for r in rows]

    def recent_notifications(self, limit: int = 50) -> list[dict[str, Any]]:
        rows = self.conn.execute("select * from notification_logs order by ts desc limit ?", (limit,)).fetchall()
        return [dict(r) for r in rows]

    async def clear_logs(self) -> dict[str, int]:
        async with self.lock:
            removed = {}
            for table in ("checks", "events", "notification_logs"):
                cur = self.conn.execute(f"delete from {table}")
                removed[table] = cur.rowcount if cur.rowcount is not None else 0
            self.conn.commit()
            return removed


@dataclasses.dataclass
class RuntimeState:
    status: str = "unknown"
    last_ok: bool | None = None
    last_ts: int | None = None
    latency_ms: int | None = None
    error: str | None = None
    exit_ip: str | None = None
    fail_count: int = 0
    ok_count: int = 0
    notified_down: bool = False


class Notifier:
    def __init__(self, cfg: dict[str, Any], store: Store | None = None):
        self.cfg = cfg or {}
        self.store = store
        self.timeout = int(self.cfg.get("timeout_seconds", 10))

    async def send_all(self, title: str, body: str) -> list[dict[str, Any]]:
        if not self.cfg.get("enabled", True):
            return [{"channel": "all", "ok": False, "error": "notifications disabled"}]
        channels = [c for c in self.cfg.get("channels", []) if c.get("enabled", True)]
        return await self.send_channels(channels, title, body)

    async def send_channel_id(self, channel_id: str, title: str, body: str) -> dict[str, Any]:
        ch = next((c for c in self.cfg.get("channels", []) if c.get("id") == channel_id), None)
        if not ch:
            return {"channel": channel_id, "ok": False, "error": "channel not found"}
        # 单通道测试应可验证配置本身，即使该渠道平时处于停用状态。
        test_ch = dict(ch)
        test_ch["enabled"] = True
        res = await self.send_channels([test_ch], title, body)
        return res[0] if res else {"channel": channel_id, "ok": False, "error": "not sent"}

    async def send_channels(self, channels: list[dict[str, Any]], title: str, body: str) -> list[dict[str, Any]]:
        results = []
        for ch in channels:
            res = await asyncio.to_thread(self._send_one, ch, title, body)
            results.append(res)
            if self.store:
                await self.store.add_notification(ch, bool(res.get("ok")), title, body, json.dumps(res, ensure_ascii=False))
        return results

    def _http_get(self, url: str) -> tuple[bool, str]:
        try:
            with urllib.request.urlopen(url, timeout=self.timeout) as r:
                text = r.read(4096).decode("utf-8", "replace")
                return 200 <= r.status < 300, f"HTTP {r.status} {text[:300]}"
        except Exception as e:
            return False, repr(e)

    def _http_post_json(self, url: str, payload: dict[str, Any], headers: dict[str, str] | None = None) -> tuple[bool, str]:
        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json", **(headers or {})}, method="POST")
        try:
            with urllib.request.urlopen(req, timeout=self.timeout) as r:
                text = r.read(4096).decode("utf-8", "replace")
                return 200 <= r.status < 300, f"HTTP {r.status} {text[:300]}"
        except Exception as e:
            return False, repr(e)

    def _send_one(self, ch: dict[str, Any], title: str, body: str) -> dict[str, Any]:
        ch = normalize_channel(ch)
        typ = ch.get("type")
        cfg = ch.get("config") or {}
        name = ch.get("name") or typ
        if not ch.get("enabled", True):
            return {"channel": name, "ok": False, "error": "disabled"}
        try:
            if typ == "bark":
                base = (cfg.get("server") or "https://api.day.app").rstrip("/")
                key = cfg.get("key") or ""
                group = cfg.get("group") or "SOCKS5监控"
                if not key:
                    return {"channel": name, "ok": False, "error": "missing Bark key"}
                url = f"{base}/{urllib.parse.quote(key)}/{urllib.parse.quote(title)}/{urllib.parse.quote(body)}?group={urllib.parse.quote(group)}"
                ok, msg = self._http_get(url)
            elif typ == "pushplus":
                token = cfg.get("token") or ""
                if not token:
                    return {"channel": name, "ok": False, "error": "missing PushPlus token"}
                payload = {"token": token, "title": title, "content": body, "template": cfg.get("template", "txt")}
                if cfg.get("topic"):
                    payload["topic"] = cfg.get("topic")
                ok, msg = self._http_post_json("https://www.pushplus.plus/send", payload)
            elif typ == "autman":
                url = cfg.get("url") or ""
                if not url:
                    return {"channel": name, "ok": False, "error": "missing autman url"}
                message = f"标题:{title}\n内容:{body}"
                payload = {"message": message, "access_token": cfg.get("access_token") or ""}
                ok, msg = self._http_post_json(url, payload, {})
            elif typ == "webhook":
                url = cfg.get("url") or ""
                if not url:
                    return {"channel": name, "ok": False, "error": "missing webhook url"}
                payload = dict(cfg.get("payload", {}) if isinstance(cfg.get("payload"), dict) else {})
                payload.update({"title": title, "content": body, "ts": now_ts()})
                ok, msg = self._http_post_json(url, payload, cfg.get("headers") or {})
            else:
                return {"channel": name, "ok": False, "error": f"unknown type {typ}"}
            return {"channel": name, "ok": ok, "message": msg}
        except Exception as e:
            return {"channel": name, "ok": False, "error": repr(e)}


async def socks5_open_connection(proxy: dict[str, Any], timeout: float, target_host: str | None = None, target_port: int | None = None) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
    host = proxy["host"]
    port = int(proxy["port"])
    username = proxy.get("username") or ""
    password = proxy.get("password") or ""
    target_host = str(target_host or proxy.get("target_host") or "www.baidu.com")
    target_port = int(target_port or proxy.get("target_port", 443))
    reader, writer = await asyncio.wait_for(asyncio.open_connection(host, port), timeout=timeout)
    try:
        methods = [0x00]
        if username or password:
            methods.append(0x02)
        writer.write(bytes([0x05, len(methods), *methods])); await writer.drain()
        resp = await asyncio.wait_for(reader.readexactly(2), timeout=timeout)
        if resp[0] != 0x05: raise RuntimeError(f"invalid socks version {resp[0]}")
        method = resp[1]
        if method == 0xFF: raise RuntimeError("socks auth method rejected")
        if method == 0x02:
            ub, pb = username.encode(), password.encode()
            writer.write(bytes([0x01, len(ub)]) + ub + bytes([len(pb)]) + pb); await writer.drain()
            auth = await asyncio.wait_for(reader.readexactly(2), timeout=timeout)
            if auth[1] != 0x00: raise RuntimeError("socks username/password auth failed")
        elif method != 0x00:
            raise RuntimeError(f"unsupported socks method {method}")
        try:
            addr = socket.inet_aton(target_host); atyp = 0x01; body = addr
        except OSError:
            hb = target_host.encode(); atyp = 0x03; body = bytes([len(hb)]) + hb
        writer.write(bytes([0x05, 0x01, 0x00, atyp]) + body + struct.pack("!H", target_port)); await writer.drain()
        head = await asyncio.wait_for(reader.readexactly(4), timeout=timeout)
        if head[1] != 0x00: raise RuntimeError(f"socks connect failed rep=0x{head[1]:02x}")
        atyp = head[3]
        if atyp == 0x01: await reader.readexactly(4)
        elif atyp == 0x03: await reader.readexactly((await reader.readexactly(1))[0])
        elif atyp == 0x04: await reader.readexactly(16)
        await reader.readexactly(2)
        return reader, writer
    except Exception:
        writer.close()
        with contextlib.suppress(Exception): await writer.wait_closed()
        raise


async def socks5_connect(proxy: dict[str, Any], timeout: float) -> None:
    reader, writer = await socks5_open_connection(proxy, timeout)
    writer.close()
    with contextlib.suppress(Exception): await writer.wait_closed()


async def http_get_via_socks(proxy: dict[str, Any], timeout: float, host: str, path: str = "/") -> str:
    reader, writer = await socks5_open_connection(proxy, timeout, host, 80)
    try:
        req = (
            f"GET {path} HTTP/1.1\r\n"
            f"Host: {host}\r\n"
            "User-Agent: socks5-monitor/1.0\r\n"
            "Accept: text/plain, application/json;q=0.9, */*;q=0.1\r\n"
            "Connection: close\r\n\r\n"
        ).encode("ascii")
        writer.write(req); await writer.drain()
        chunks = []
        deadline = asyncio.get_running_loop().time() + timeout
        while True:
            remaining = max(0.1, deadline - asyncio.get_running_loop().time())
            if remaining <= 0.1 and chunks:
                break
            try:
                chunk = await asyncio.wait_for(reader.read(4096), timeout=remaining)
            except asyncio.TimeoutError:
                break
            if not chunk:
                break
            chunks.append(chunk)
            if sum(len(x) for x in chunks) >= 65536:
                break
        raw = b"".join(chunks)
        return raw.decode("utf-8", "replace")
    finally:
        writer.close()
        with contextlib.suppress(Exception): await writer.wait_closed()


def extract_ip(text: str) -> str | None:
    body = text.split("\r\n\r\n", 1)[-1].strip()
    m = re.search(r"\b(?:(?:25[0-5]|2[0-4]\d|1?\d?\d)\.){3}(?:25[0-5]|2[0-4]\d|1?\d?\d)\b", body)
    if m:
        return m.group(0)
    # Very small IPv6 matcher, enough for plain text IP services.
    m = re.search(r"\b(?:[0-9a-fA-F]{1,4}:){2,}[0-9a-fA-F]{0,4}\b", body)
    return m.group(0) if m else None


async def get_exit_ip_via_socks(proxy: dict[str, Any], timeout: float) -> str | None:
    """Fetch the public egress IP through the SOCKS5 proxy without failing health checks."""
    endpoints = [
        ("api.ipify.org", "/?format=text"),
        ("ipv4.icanhazip.com", "/"),
        ("ifconfig.me", "/ip"),
        ("ident.me", "/"),
        ("checkip.amazonaws.com", "/"),
    ]
    errors = []
    per_endpoint_timeout = max(2.0, min(float(timeout), 6.0))
    for host, path in endpoints:
        try:
            text = await http_get_via_socks(proxy, per_endpoint_timeout, host, path)
            ip = extract_ip(text)
            if ip:
                return ip
            errors.append(f"{host}: no ip in response")
        except Exception as e:
            errors.append(f"{host}: {e}")
    proxy["last_exit_ip_error"] = "; ".join(errors[-3:]) or "exit ip lookup failed: empty response from all providers"
    return None


async def check_proxy(proxy: dict[str, Any]) -> tuple[bool, int | None, str | None, str | None]:
    timeout = float(proxy.get("timeout_seconds", 8))
    start = time.perf_counter()
    try:
        await socks5_connect(proxy, timeout)
        latency = int((time.perf_counter() - start) * 1000)
        exit_ip = await get_exit_ip_via_socks(proxy, timeout)
        err = None if exit_ip else (proxy.get("last_exit_ip_error") or "exit ip lookup failed")
        return True, latency, err, exit_ip
    except Exception as e:
        return False, int((time.perf_counter() - start) * 1000), str(e), None


class App:
    def __init__(self, config_path: Path):
        self.config_path = config_path
        self.config = merge_defaults(load_json(config_path))
        if not config_path.exists(): save_json(config_path, self.config)
        self.store = Store(Path(self.config.get("db_path") or DEFAULT_DB))
        self.notifier = Notifier(self.config.get("notifications") or {}, self.store)
        self.states: dict[str, RuntimeState] = {}
        self.monitor_tasks: dict[str, asyncio.Task] = {}
        self.config_lock = asyncio.Lock()

    @property
    def proxies(self) -> list[dict[str, Any]]:
        return self.config.setdefault("proxies", [])

    @property
    def channels(self) -> list[dict[str, Any]]:
        return self.config.setdefault("notifications", {}).setdefault("channels", [])

    def state_for(self, proxy_id: str) -> RuntimeState:
        return self.states.setdefault(proxy_id, RuntimeState())

    async def save_config(self) -> None:
        async with self.config_lock:
            save_json(self.config_path, self.config)
            self.notifier = Notifier(self.config.get("notifications") or {}, self.store)

    def get_proxy(self, proxy_id: str) -> dict[str, Any] | None:
        return next((p for p in self.proxies if p.get("id") == proxy_id), None)

    def get_channel(self, channel_id: str) -> dict[str, Any] | None:
        return next((c for c in self.channels if c.get("id") == channel_id), None)

    def validate_proxy(self, data: dict[str, Any], existing: dict[str, Any] | None = None) -> dict[str, Any]:
        p = dict(existing or {})
        p.update(data)
        p.setdefault("id", re.sub(r"[^a-zA-Z0-9_-]+", "-", (p.get("name") or "proxy").strip()).strip("-") or f"proxy-{uuid.uuid4().hex[:8]}")
        p.setdefault("name", p["id"])
        if not p.get("host"): raise ValueError("host required")
        p["port"] = int(p.get("port") or 1080)
        p["target_host"] = p.get("target_host") or self.config.get("target_host") or "www.baidu.com"
        p["target_port"] = int(p.get("target_port") or self.config.get("target_port") or 443)
        p["timeout_seconds"] = int(p.get("timeout_seconds") or self.config.get("timeout_seconds") or 8)
        p["interval_seconds"] = max(5, int(p.get("interval_seconds") or self.config.get("interval_seconds") or 300))
        p["fail_threshold"] = int(p.get("fail_threshold") or self.config.get("fail_threshold") or 3)
        p["recover_threshold"] = int(p.get("recover_threshold") or self.config.get("recover_threshold") or 1)
        p["enabled"] = bool(p.get("enabled", True))
        return p

    def validate_channel(self, data: dict[str, Any], existing: dict[str, Any] | None = None) -> dict[str, Any]:
        incoming = normalize_channel(data or {})
        if existing:
            incoming["id"] = existing.get("id")
        typ = incoming.get("type")
        if typ not in {"bark", "pushplus", "autman", "webhook"}:
            raise ValueError("type must be bark/pushplus/autman/webhook")
        incoming.setdefault("name", typ)
        incoming["enabled"] = bool(incoming.get("enabled", True))
        cfg = incoming["config"]
        if typ == "bark" and not cfg.get("server"):
            cfg["server"] = "https://api.day.app"
        if typ == "pushplus" and not cfg.get("template"):
            cfg["template"] = "txt"
        return incoming

    def validate_settings(self, data: dict[str, Any]) -> dict[str, Any]:
        out = {}
        if "interval_seconds" in data:
            out["interval_seconds"] = max(5, int(data.get("interval_seconds") or 300))
        if "timeout_seconds" in data:
            out["timeout_seconds"] = max(1, int(data.get("timeout_seconds") or 8))
        if "target_host" in data:
            host = str(data.get("target_host") or "").strip()
            if not host:
                raise ValueError("target_host required")
            out["target_host"] = host
        if "target_port" in data:
            port = int(data.get("target_port") or 443)
            if port < 1 or port > 65535:
                raise ValueError("target_port must be 1..65535")
            out["target_port"] = port
        if "fail_threshold" in data:
            out["fail_threshold"] = max(1, int(data.get("fail_threshold") or 3))
        if "recover_threshold" in data:
            out["recover_threshold"] = max(1, int(data.get("recover_threshold") or 1))
        if "max_checks_per_proxy" in data:
            out["max_checks_per_proxy"] = max(1, int(data.get("max_checks_per_proxy") or 500))
        return out

    async def update_settings(self, data: dict[str, Any]) -> dict[str, Any]:
        settings = self.validate_settings(data)
        self.config.update(settings)
        # Apply changed global defaults to nodes that were still using old built-in defaults.
        for p in self.proxies:
            if settings.get("target_host") and p.get("target_host") in (None, "", "1.1.1.1", "8.8.8.8", "google.com"):
                p["target_host"] = settings["target_host"]
            if settings.get("target_port") and not p.get("target_port"):
                p["target_port"] = settings["target_port"]
            if settings.get("interval_seconds") and not p.get("interval_seconds"):
                p["interval_seconds"] = settings["interval_seconds"]
        await self.save_config()
        for p in list(self.proxies):
            await self.restart_proxy_task(p["id"])
        return {k: self.config.get(k) for k in ["interval_seconds", "timeout_seconds", "target_host", "target_port", "fail_threshold", "recover_threshold", "max_checks_per_proxy"]}

    async def upsert_proxy(self, data: dict[str, Any], proxy_id: str | None = None) -> dict[str, Any]:
        existing = self.get_proxy(proxy_id) if proxy_id else None
        p = self.validate_proxy(data, existing)
        if proxy_id and existing:
            idx = self.proxies.index(existing); self.proxies[idx] = p
        else:
            if self.get_proxy(p["id"]): raise ValueError("proxy id already exists")
            self.proxies.append(p)
        await self.save_config(); await self.restart_proxy_task(p["id"])
        return public_proxy(p)

    async def delete_proxy(self, proxy_id: str) -> bool:
        p = self.get_proxy(proxy_id)
        if not p: return False
        self.proxies.remove(p)
        task = self.monitor_tasks.pop(proxy_id, None)
        if task: task.cancel()
        self.states.pop(proxy_id, None)
        await self.save_config()
        return True

    async def upsert_channel(self, data: dict[str, Any], channel_id: str | None = None) -> dict[str, Any]:
        existing = self.get_channel(channel_id) if channel_id else None
        c = self.validate_channel(data, existing)
        if channel_id and existing:
            idx = self.channels.index(existing); self.channels[idx] = c
        else:
            if self.get_channel(c["id"]): raise ValueError("channel id already exists")
            self.channels.append(c)
        await self.save_config()
        return c

    async def delete_channel(self, channel_id: str) -> bool:
        c = self.get_channel(channel_id)
        if not c: return False
        self.channels.remove(c)
        await self.save_config()
        return True

    async def run_once(self, proxy: dict[str, Any]) -> None:
        if not proxy.get("enabled", True): return
        proxy_id = proxy["id"]
        st = self.state_for(proxy_id)
        ok, latency, err, exit_ip = await check_proxy(proxy)
        await self.store.add_check(proxy_id, ok, latency, err, exit_ip, int(self.config.get("max_checks_per_proxy") or 500))
        st.last_ts = now_ts(); st.last_ok = ok; st.latency_ms = latency; st.error = err; st.exit_ip = exit_ip
        fail_threshold = int(proxy.get("fail_threshold", self.config.get("fail_threshold", 3)))
        recover_threshold = int(proxy.get("recover_threshold", self.config.get("recover_threshold", 1)))
        old = st.status
        if ok:
            st.ok_count += 1; st.fail_count = 0
            if st.status == "down" and st.ok_count >= recover_threshold:
                st.status = "up"; st.notified_down = False
                msg = f"{proxy.get('name', proxy_id)} 已恢复。延迟 {latency}ms，时间 {iso()}"
                await self.store.add_event(proxy_id, old, st.status, msg)
                await self.notifier.send_all(f"SOCKS5恢复：{proxy.get('name', proxy_id)}", msg)
            elif st.status == "unknown":
                st.status = "up"
                await self.store.add_event(proxy_id, old, st.status, f"{proxy.get('name', proxy_id)} 首次检测正常")
        else:
            st.fail_count += 1; st.ok_count = 0
            if st.fail_count >= fail_threshold:
                st.status = "down"
                if not st.notified_down:
                    msg = f"{proxy.get('name', proxy_id)} 不通，连续失败 {st.fail_count} 次。错误：{err}。时间 {iso()}"
                    await self.store.add_event(proxy_id, old, st.status, msg)
                    await self.notifier.send_all(f"SOCKS5故障：{proxy.get('name', proxy_id)}", msg)
                    st.notified_down = True

    async def monitor_loop(self, proxy_id: str) -> None:
        while True:
            proxy = self.get_proxy(proxy_id)
            if not proxy: return
            if proxy.get("enabled", True):
                await self.run_once(proxy)
            await asyncio.sleep(int(proxy.get("interval_seconds", self.config.get("interval_seconds", 60))))

    async def restart_proxy_task(self, proxy_id: str) -> None:
        old = self.monitor_tasks.pop(proxy_id, None)
        if old: old.cancel()
        if self.get_proxy(proxy_id):
            self.monitor_tasks[proxy_id] = asyncio.create_task(self.monitor_loop(proxy_id))

    async def start_tasks(self) -> None:
        for p in self.proxies:
            await self.restart_proxy_task(p["id"])

    def status_payload(self) -> dict[str, Any]:
        items = []
        for p in self.proxies:
            sid = p["id"]; st = self.state_for(sid)
            items.append({"proxy": public_proxy(p), "state": dataclasses.asdict(st), "recent": self.store.recent_checks(sid, 20)})
        return {
            "ok": True, "now": now_ts(), "now_text": iso(),
            "settings": {k: self.config.get(k) for k in ["interval_seconds", "timeout_seconds", "target_host", "target_port", "fail_threshold", "recover_threshold", "max_checks_per_proxy"]},
            "proxies": items,
            "notifications": self.config.get("notifications", {}),
            "events": self.store.recent_events(50),
            "notification_logs": self.store.recent_notifications(30),
        }


def get_index_html() -> str:
    static_index = APP_DIR / "static" / "index.html"
    if static_index.exists():
        return static_index.read_text(encoding="utf-8")
    return "<h1>SOCKS5 Sentinel</h1><p>static/index.html missing</p>"



def json_response(obj: Any, status: int = 200) -> tuple[int, str, bytes]:
    return status, "application/json; charset=utf-8", json.dumps(obj, ensure_ascii=False).encode("utf-8")


async def read_request(reader: asyncio.StreamReader) -> tuple[str, str, dict[str, str], bytes]:
    head = await reader.readuntil(b"\r\n\r\n")
    text = head.decode("iso-8859-1")
    lines = text.split("\r\n")
    method, path, _ = lines[0].split(" ", 2)
    headers = {}
    for line in lines[1:]:
        if ":" in line:
            k, v = line.split(":", 1); headers[k.lower()] = v.strip()
    length = int(headers.get("content-length") or "0")
    body = await reader.readexactly(length) if length else b""
    return method, path, headers, body


async def handle_http(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, app: App) -> None:
    try:
        method, path, headers, body = await read_request(reader)
        parsed = urllib.parse.urlparse(path); p = parsed.path
        status, ctype, resp = 200, "text/plain; charset=utf-8", b""
        data = json.loads(body.decode("utf-8")) if body else {}
        if method == "GET" and p == "/":
            ctype = "text/html; charset=utf-8"; resp = get_index_html().encode("utf-8")
        elif method == "GET" and p == "/api/status":
            status, ctype, resp = json_response(app.status_payload())
        elif method == "PUT" and p == "/api/settings":
            status, ctype, resp = json_response({"ok": True, "settings": await app.update_settings(data)})
        elif method == "DELETE" and p == "/api/logs":
            status, ctype, resp = json_response({"ok": True, "removed": await app.store.clear_logs()})
        elif method == "POST" and p == "/api/proxies":
            status, ctype, resp = json_response({"ok": True, "proxy": await app.upsert_proxy(data)}, 201)
        elif method == "PUT" and p.startswith("/api/proxies/"):
            pid = urllib.parse.unquote(p.rsplit("/", 1)[-1])
            status, ctype, resp = json_response({"ok": True, "proxy": await app.upsert_proxy(data, pid)})
        elif method == "DELETE" and p.startswith("/api/proxies/"):
            pid = urllib.parse.unquote(p.rsplit("/", 1)[-1])
            status, ctype, resp = json_response({"ok": await app.delete_proxy(pid)})
        elif method == "POST" and p.startswith("/api/check/"):
            pid = urllib.parse.unquote(p.rsplit("/", 1)[-1]); proxy = app.get_proxy(pid)
            if not proxy: status, ctype, resp = json_response({"ok": False, "error": "proxy not found"}, 404)
            else:
                await app.run_once(proxy)
                status, ctype, resp = json_response({"ok": True, "state": dataclasses.asdict(app.state_for(pid))})
        elif method == "POST" and p == "/api/channels":
            status, ctype, resp = json_response({"ok": True, "channel": await app.upsert_channel(data)}, 201)
        elif method == "PUT" and p.startswith("/api/channels/"):
            cid = urllib.parse.unquote(p.rsplit("/", 1)[-1])
            status, ctype, resp = json_response({"ok": True, "channel": await app.upsert_channel(data, cid)})
        elif method == "DELETE" and p.startswith("/api/channels/"):
            cid = urllib.parse.unquote(p.rsplit("/", 1)[-1])
            status, ctype, resp = json_response({"ok": await app.delete_channel(cid)})
        elif method == "POST" and p.startswith("/api/channels/") and p.endswith("/test"):
            cid = urllib.parse.unquote(p.split("/")[-2])
            res = await app.notifier.send_channel_id(cid, "SOCKS5监控测试", f"测试通知 {iso()}")
            status, ctype, resp = json_response({"ok": bool(res.get("ok")), "result": res})
        elif method == "POST" and p == "/api/notify/test":
            res = await app.notifier.send_all("SOCKS5监控测试", f"测试通知 {iso()}")
            status, ctype, resp = json_response({"ok": True, "results": res})
        else:
            status, ctype, resp = json_response({"ok": False, "error": "not found"}, 404)
    except Exception as e:
        traceback.print_exc()
        status, ctype, resp = json_response({"ok": False, "error": str(e)}, 500)
    try:
        reason = {200:"OK",201:"Created",400:"Bad Request",404:"Not Found",500:"Internal Server Error"}.get(status,"OK")
        writer.write(f"HTTP/1.1 {status} {reason}\r\nContent-Type: {ctype}\r\nContent-Length: {len(resp)}\r\nConnection: close\r\n\r\n".encode()+resp)
        await writer.drain()
    finally:
        writer.close()
        with contextlib.suppress(Exception): await writer.wait_closed()


async def main_async(args: argparse.Namespace) -> None:
    app = App(Path(args.config))
    await app.start_tasks()
    server = await asyncio.start_server(lambda r,w: handle_http(r,w,app), args.host, args.port)
    print(f"SOCKS5 monitor panel: http://{args.host}:{args.port}", flush=True)
    print(f"Config: {args.config}", flush=True)
    async with server:
        await server.serve_forever()


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--config", default=str(DEFAULT_CONFIG))
    ap.add_argument("--host", default="0.0.0.0")
    ap.add_argument("--port", type=int, default=8787)
    args = ap.parse_args()
    asyncio.run(main_async(args))


if __name__ == "__main__":
    main()