ui: simplify configuration forms
This commit is contained in:
@@ -0,0 +1,684 @@
|
||||
#!/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,
|
||||
"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.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) -> 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),
|
||||
)
|
||||
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 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]
|
||||
|
||||
|
||||
@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_connect(proxy: dict[str, Any], timeout: float) -> None:
|
||||
host = proxy["host"]
|
||||
port = int(proxy["port"])
|
||||
username = proxy.get("username") or ""
|
||||
password = proxy.get("password") or ""
|
||||
target_host = proxy.get("target_host", "1.1.1.1")
|
||||
target_port = int(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)
|
||||
finally:
|
||||
writer.close()
|
||||
with contextlib.suppress(Exception): await writer.wait_closed()
|
||||
|
||||
|
||||
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)
|
||||
return True, int((time.perf_counter() - start) * 1000), None, None
|
||||
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))
|
||||
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"]}
|
||||
|
||||
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)
|
||||
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"]},
|
||||
"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 == "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()
|
||||
Reference in New Issue
Block a user