jiedian/scripts/render-server.py

#!/usr/bin/env python3
"""根据 data/nodes.db 与 .env 生成 sing-box 服务端配置。"""
from __future__ import annotations

import json
import os
import sqlite3
import subprocess
import sys
from pathlib import Path

ROOT = Path(os.environ.get("JIEDIAN_ROOT", Path(__file__).resolve().parents[1]))
ENV_FILE = ROOT / ".env"
DB_FILE = ROOT / "data" / "nodes.db"
OUT_FILE = Path("/etc/sing-box/config.json")


def load_env(path: Path) -> dict[str, str]:
    env: dict[str, str] = {}
    if not path.exists():
        raise SystemExit(f"缺少 .env: {path}")
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        env[key.strip()] = value.strip()
    return env


def load_nodes(db_path: Path) -> list[dict]:
    if not db_path.exists():
        raise SystemExit(f"缺少节点数据库: {db_path}，请先运行 install.sh")
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    rows = conn.execute(
        "SELECT id, name, uuid, hy2_password FROM nodes WHERE enabled = 1 ORDER BY id"
    ).fetchall()
    conn.close()
    if not rows:
        raise SystemExit("没有可用节点，请在管理面板中添加节点")
    return [dict(row) for row in rows]


def build_config(env: dict[str, str], nodes: list[dict]) -> dict:
    required = [
        "REALITY_PRIVATE_KEY",
        "REALITY_SHORT_ID",
        "REALITY_SERVER_NAME",
        "DOMAIN",
    ]
    for key in required:
        if not env.get(key):
            raise SystemExit(f".env 缺少 {key}")

    vless_users = [{"uuid": n["uuid"], "flow": "xtls-rprx-vision"} for n in nodes]
    hy2_base_port = 8443

    inbounds: list[dict] = [
        {
            "type": "vless",
            "tag": "vless-reality-in",
            "listen": "0.0.0.0",
            "listen_port": 443,
            "users": vless_users,
            "tls": {
                "enabled": True,
                "server_name": env["REALITY_SERVER_NAME"],
                "reality": {
                    "enabled": True,
                    "handshake": {
                        "server": env["REALITY_SERVER_NAME"],
                        "server_port": 443,
                    },
                    "private_key": env["REALITY_PRIVATE_KEY"],
                    "short_id": ["", env["REALITY_SHORT_ID"]],
                    "max_time_difference": "1m",
                },
            },
        },
    ]
    for index, node in enumerate(nodes):
        inbounds.append(
            {
                "type": "hysteria2",
                "tag": f"hy2-in-{node['id']}",
                "listen": "0.0.0.0",
                "listen_port": hy2_base_port + index,
                "users": [
                    {"name": node["uuid"], "password": node["hy2_password"]},
                ],
                "tls": {
                    "enabled": True,
                    "server_name": env["DOMAIN"],
                    "certificate_path": "/etc/sing-box/certs/fullchain.pem",
                    "key_path": "/etc/sing-box/certs/privkey.pem",
                },
            }
        )

    clash_secret = env.get("CLASH_API_SECRET", "")

    config = {
        "log": {"level": "info", "timestamp": True},
        "inbounds": inbounds,
        "outbounds": [{"type": "direct", "tag": "direct"}],
        "route": {
            "rules": [{"ip_is_private": True, "action": "reject"}],
            "final": "direct",
        },
    }

    experimental: dict = {
        "clash_api": {
            "external_controller": "127.0.0.1:9090",
        },
    }
    if clash_secret:
        experimental["clash_api"]["secret"] = clash_secret
    config["experimental"] = experimental
    return config


def main() -> None:
    env = load_env(ENV_FILE)
    nodes = load_nodes(DB_FILE)
    config = build_config(env, nodes)

    OUT_FILE.parent.mkdir(parents=True, exist_ok=True)
    OUT_FILE.write_text(json.dumps(config, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")

    check = subprocess.run(
        ["sing-box", "check", "-c", str(OUT_FILE)],
        capture_output=True,
        text=True,
    )
    if check.returncode != 0:
        sys.stderr.write(check.stderr or check.stdout)
        raise SystemExit(check.returncode)

    print(f"已生成 {OUT_FILE}（{len(nodes)} 个节点）")


if __name__ == "__main__":
    main()