SMANAGE/Main.py

import os
import sys
import ssl
import json
import time
import logging
import logging.handlers
import datetime
import traceback
import signal
import socket
import subprocess
from pathlib import Path

import colorama
import gevent
from gevent import monkey
from gevent.pywsgi import WSGIServer
from gevent.pool import Pool
from gevent.threadpool import ThreadPool

monkey.patch_all()

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'gvsds.settings')

import django
django.setup()

CPU_CORES = os.cpu_count() or 4
MAX_COROUTINES = CPU_CORES * 50
MAX_THREADS = CPU_CORES * 2

# 单实例控制端口（仅回环）：全局只允许一个 Main.py 实例，同时作为 --reload 的通知通道
# 行协议：PING\n → PONG\n ；RELOAD [json-argv]\n → OK\n（收到后由既有实例按该参数自重启）
CONTROL_HOST = '127.0.0.1'
# ⚠ 默认端口必须避开 Windows「保留端口段」（Hyper-V / WSL / Docker 会动态保留若干区间）：
# 落在保留段内时，即使没有任何进程监听，bind 也会失败（WinError 10013 权限禁止），
# 看起来像"该端口被占用"。查看保留段：netsh int ipv4 show excludedportrange protocol=tcp
# 同时取 < 49152（动态端口范围起点）的静态区，避免被其它程序临时占用。
# 本机实测：57321 落在保留段 57269-57368 → 必失败；45731 可正常绑定。
CONTROL_PORT = int(os.environ.get('MAIN_CONTROL_PORT') or 45731)
# 抢绑端口的宽限时间：覆盖"上一个实例刚 execv、端口尚未释放"的瞬间
INSTANCE_TAKEOVER_SECONDS = 5.0
# 抢绑重试间隔 / 探测既有实例的连接与应答超时（都要短，避免启动慢）
INSTANCE_RETRY_INTERVAL = 0.2
CONTROL_TIMEOUT = 0.5

# WSGI 环境的 SERVER_NAME：必须预置，否则 gevent 的 WSGIServer.init_socket() 会对监听地址
# 做反向 DNS（socket.getfqdn('0.0.0.0')）—— 实测本机要 1.3s 起，网络抖动时更久，期间服务器
# 一直没 started（表现为"启动很慢"甚至误判成启动失败）。用本机名预置即可跳过该查询（纯本地调用，不走 DNS）。
SERVER_NAME = socket.gethostname()

from gvsds.secrets import get as _secret_get
SQL_IP = _secret_get('database', 'host', fallback='127.0.0.1')

WHITE_HOSTS = [
    "www.gvsds.com", "gvsds.com", "api.gvsds.com", "127.0.0.1", "localhost", SQL_IP,
    "gvsds.dev", "yc8z.org", "ycbz.org", "gvsds.org", "gvsds.studio", "114.66.59.25",
    "librok.gvsds.com", "rollcall.gvsds.com",
    "viper.gvsds.com", "venom.gvsds.com",
]
BAN_IPS = [
    "www.kanglaohui.com", "www.best-loving.cn", "11.4.5.14", "185.223.152.189"
]

PIRACY_DIR = Path("TempLate") / "Errors" / "Piracy"
PIRACY_LIST = os.listdir(PIRACY_DIR) if PIRACY_DIR.exists() else []

# 静态站点：host -> 站点根目录
# viper.gvsds.com 走 Django SubdomainView（TempLate/Viper，Vue 官网产物）
STATIC_SITES = {
    "librok.gvsds.com": Path("TempLate") / "rok" / "lib",
    "rollcall.gvsds.com": Path("TempLate") / "rollcall",
    "venom.gvsds.com": Path("TempLate") / "Venom",
}

# 单页应用站点：命中不到文件时回退 index.html，交由前端路由处理
# （如 venom 官网的 /project/<包名>/ 详情页；带扩展名的资源请求仍按 404 处理）
SPA_STATIC_SITES = {"venom.gvsds.com"}

LOG_CONFIG = {
    'disable_static_success': False,
    'disable_favicon': False,
    'disable_all_static': False,
    'color_whole_line': True,
    'static_extensions': ['.js', '.css', '.png', '.jpg', '.jpeg', '.gif', '.ico', '.svg', '.woff', '.woff2', '.ttf', '.eot'],
    'log_level': 'INFO',
    'show_domain': True,
    'show_full_host': True,
}

STATUS_COLOR_MAP = {
    200: colorama.Fore.GREEN,
    301: colorama.Fore.CYAN,
    302: colorama.Fore.CYAN,
    304: colorama.Fore.CYAN,
    400: colorama.Fore.YELLOW,
    403: colorama.Fore.RED,
    404: colorama.Fore.MAGENTA,
    500: colorama.Fore.RED + colorama.Style.BRIGHT,
    502: colorama.Fore.RED + colorama.Style.BRIGHT,
    503: colorama.Fore.RED + colorama.Style.BRIGHT,
}

LEVEL_COLOR_MAP = {
    'DEBUG': colorama.Fore.CYAN,
    'INFO': colorama.Fore.GREEN,
    'WARNING': colorama.Fore.YELLOW,
    'ERROR': colorama.Fore.RED,
    'CRITICAL': colorama.Fore.RED + colorama.Style.BRIGHT,
}

logger = None


class NoOpLog:
    @staticmethod
    def write(msg):
        pass
    @staticmethod
    def flush():
        pass


class CustomWSGIServer(WSGIServer):
    def handle_error(self, request, client_address):
        exc_type, exc_value, exc_traceback = self.get_exc_info()
        ignored_exceptions = (
            ssl.SSLEOFError, ConnectionResetError, BrokenPipeError, OSError, EOFError
        )
        if isinstance(exc_value, ignored_exceptions):
            logger.debug(f"客户端 {client_address} 连接异常断开: {exc_value}",
                         extra={'domain': 'system', 'status_code': 0})
            return
        super().handle_error(request, client_address)


class WholeLineColorFormatter(logging.Formatter):
    def format(self, record):
        timestamp = self.formatTime(record, datefmt='%Y-%m-%d %H:%M:%S')
        levelname = record.levelname
        message = record.getMessage()
        status_code = getattr(record, 'status_code', 0)
        domain = getattr(record, 'domain', '')
        host = getattr(record, 'host', '')

        if LOG_CONFIG['color_whole_line'] and status_code:
            color = STATUS_COLOR_MAP.get(status_code, LEVEL_COLOR_MAP.get(levelname, colorama.Fore.WHITE))
        else:
            color = LEVEL_COLOR_MAP.get(levelname, colorama.Fore.WHITE)

        log_parts = [f"[{timestamp}]", f"[{levelname:7s}]"]
        if LOG_CONFIG['show_domain']:
            if domain:
                log_parts.append(f"[域:{domain:20s}]")
            elif host:
                log_parts.append(f"[Host:{host:20s}]")

        log_parts.append(message)
        log_line = " ".join(log_parts)

        if LOG_CONFIG['color_whole_line']:
            log_line = f"{color}{log_line}{colorama.Style.RESET_ALL}"
        return log_line


class RequestFilter:
    @staticmethod
    def is_favicon(path):
        return path.lower() == '/favicon.ico'

    @staticmethod
    def is_static_file(path):
        ext = os.path.splitext(path)[1].lower()
        return ext in LOG_CONFIG['static_extensions']

    @staticmethod
    def should_skip_log(path, status_code):
        if LOG_CONFIG['disable_favicon'] and RequestFilter.is_favicon(path):
            return True
        if RequestFilter.is_static_file(path):
            if LOG_CONFIG['disable_all_static']:
                return True
            if LOG_CONFIG['disable_static_success'] and status_code < 400:
                return True
        return False

    @staticmethod
    def extract_domain(host):
        if not host:
            return ''
        return host.split(':')[0] if ':' in host else host

def GetIP(environ: dict = {}):
    x_forwarded_for_raw = environ.get('HTTP_X_FORWARDED_FOR', '')
    x_forwarded_for_list = [x.strip() for x in x_forwarded_for_raw.split(',') if x.strip()]
    x_forwarded_for = x_forwarded_for_list[0] if x_forwarded_for_list else ''
    return x_forwarded_for

class FilteredLoggingMiddleware:
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        path = environ.get('PATH_INFO', '')
        query_string = environ.get('QUERY_STRING', '')
        full_path = path + (f"?{query_string}" if query_string else "")
        method = environ.get('REQUEST_METHOD', '')
        ip = environ.get('REMOTE_ADDR', '')
        host = environ.get('HTTP_HOST', '')
        domain = RequestFilter.extract_domain(host)
        start_time = datetime.datetime.now()

        real_ip = GetIP(environ)
        real_ip = real_ip if real_ip else ip


        if RequestFilter.should_skip_log(full_path, 200):
            def wrapped_start_response(status, headers, exc_info=None):
                return start_response(status, headers, exc_info)
            return self.app(environ, wrapped_start_response)

        def wrapped_start_response(status, headers, exc_info=None):
            try:
                status_code = int(status.split()[0]) if status else 0

                if RequestFilter.should_skip_log(full_path, status_code):
                    return start_response(status, headers, exc_info)

                duration = (datetime.datetime.now() - start_time).total_seconds()
                log_message = f"[{status_code:3d}] [{method.ljust(4)[:4]}] [{real_ip:15}] [{duration:.6f}s] {full_path}"

                extra = {
                    'status_code': status_code,
                    'domain': domain,
                    'host': host if LOG_CONFIG['show_full_host'] else ''
                }

                if status_code >= 500:
                    logger.error(log_message, extra=extra, exc_info=exc_info)
                elif status_code >= 400:
                    logger.warning(log_message, extra=extra)
                else:
                    logger.info(log_message, extra=extra)
            except Exception as e:
                logger.error(f"日志记录失败: {e}", exc_info=True)

            return start_response(status, headers, exc_info)

        return self.app(environ, wrapped_start_response)

import threading
import time
import re

class SecurityMiddleware:
    def __init__(self, app):
        self.app = app

        # 恶意参数正则
        self.bad_reg = re.compile(r'^(?=.{8,})(\d+[A-Za-z&]{1,}|[A-Za-z]\d+|&\d+)', re.I)

        # ==================== 文件配置 ====================
        self.ip_file = "./ips.txt"                # 黑名单IP
        self.ip_times_file = "./iptimes.txt"      # IP:次数
        self.total_times_file = "./times.txt"     # 总拦截次数

        # ==================== 内存存储 ====================
        self.seen_ips = self.load_file_to_set(self.ip_file)
        self.ip_counter = self.load_file_to_dict(self.ip_times_file)  # {ip: count}
        self.total_count = self.load_total_count()

        # ==================== 线程队列 & 异步写入 ====================
        self.queue = []
        self.queue_lock = threading.Lock()
        self.sync_interval = 5  # 每5秒批量落地一次
        self.start_async_saver()

    # ==================== 加载工具 ====================
    def load_file_to_set(self, path):
        s = set()
        if os.path.exists(path):
            with open(path, "r", encoding="utf-8") as f:
                for line in f:
                    ip = line.strip()
                    if ip:
                        s.add(ip)
        return s

    def load_file_to_dict(self, path):
        d = {}
        if os.path.exists(path):
            with open(path, "r", encoding="utf-8") as f:
                for line in f:
                    line = line.strip()
                    if not line or ":" not in line:
                        continue
                    ip, cnt = line.split(":", 1)
                    try:
                        d[ip.strip()] = int(cnt.strip())
                    except:
                        continue
        return d

    def load_total_count(self):
        if os.path.exists(self.total_times_file):
            try:
                with open(self.total_times_file, "r", encoding="utf-8") as f:
                    return int(f.read().strip())
            except:
                return 0
        return 0

    # ==================== 异步落地方案（核心） ====================
    def start_async_saver(self):
        def saver():
            while True:
                time.sleep(self.sync_interval)
                self.sync_to_disk()

        t = threading.Thread(target=saver, daemon=True)
        t.start()

    def sync_to_disk(self):
        with self.queue_lock:
            if not self.queue:
                return
            queue_copy = self.queue.copy()
            self.queue.clear()

        # 批量更新
        for ip in queue_copy:
            self.ip_counter[ip] = self.ip_counter.get(ip, 0) + 1
            self.total_count += 1
            self.seen_ips.add(ip)

        # 写入磁盘
        with open(self.ip_times_file, "w", encoding="utf-8") as f:
            for ip, cnt in self.ip_counter.items():
                f.write(f"{ip}:{cnt}\n")

        with open(self.ip_file, "w", encoding="utf-8") as f:
            for ip in self.seen_ips:
                f.write(f"{ip}\n")

        with open(self.total_times_file, "w", encoding="utf-8") as f:
            f.write(str(self.total_count))

    # ==================== 记录IP（入队列，不阻塞） ====================
    def add_ip_to_queue(self, ip):
        with self.queue_lock:
            self.queue.append(ip)

    # ==================== 获取客户端IP ====================
    def get_ip(self, environ):
        forward = environ.get('HTTP_X_FORWARDED_FOR', '')
        if forward:
            ip = forward.split(',')[0].strip()
            if ip:
                return ip
        return environ.get('REMOTE_ADDR', '127.0.0.1')

    # ==================== 请求入口 ====================
    def __call__(self, environ, start_response):
        query = environ.get('QUERY_STRING', '')

        # ==================== 拦截恶意参数 ====================
        if query and self.bad_reg.search(query):
            ip = self.get_ip(environ)
            self.add_ip_to_queue(ip)

            # 直接关闭连接，不产生日志
            start_response('444 No Response', [])
            return []

        # ==================== 原有逻辑保持不变 ====================
        host = environ.get('HTTP_HOST', '').lower()
        path = environ.get('PATH_INFO', '')

        # 跳转到 www.gvsds.com
        if host == "gvsds.com":
            location = f'https://www.gvsds.com{path}'
            if query:
                location += f'?{query}'
            start_response('301 Moved Permanently', [('Location', location), ('Content-Type', 'text/plain')])
            return [b'Redirecting to www.gvsds.com...']

        # robots.txt
        if path == "/robots.txt":
            start_response('200 OK', [('Content-Type', 'text/plain')])
            return [b'User-agent: *\nDisallow: /assets/']

        # 盗版域名拦截
        domain_suffix = ".".join(host.split(".")[-2:]) if "." in host else host
        if domain_suffix in PIRACY_LIST:
            try:
                content = open(PIRACY_DIR / domain_suffix, encoding="utf-8").read()
            except Exception:
                content = "Access Denied"
            start_response('403 Forbidden', [('Content-Type', 'text/html')])
            return [content.encode('utf-8')]

        # 非法HOST + 黑名单IP
        x_forwarded_for = self.get_ip(environ)
        host_no_port = host.split(':')[0] if ':' in host else host
        if (host not in WHITE_HOSTS and host_no_port not in WHITE_HOSTS
                and domain_suffix not in WHITE_HOSTS) or x_forwarded_for in BAN_IPS:
            try:
                content = open(Path("TempLate") / "Errors" / "_403_HOST.html", encoding="utf-8").read()
            except Exception:
                content = "Forbidden"
            start_response('403 Forbidden', [('Content-Type', 'text/html')])
            return [content.encode('utf-8')]

        # 拦截脚本后缀
        lower_path = path.lower().split("?")[0]
        if "." in lower_path and lower_path.split(".")[-1] in ["php", "asp", "aspx", "cgi"]:
            start_response('403 Forbidden', [('Content-Type', 'text/plain')])
            return [b'Server does not support this type of page.']

        # 静态站点路由：将指定 host 的请求直接映射到本地目录；
        # /API/ 前缀（如 rollcall 数据持久化接口）转交 Django 处理
        if host in STATIC_SITES:
            if path.startswith('/API/'):
                return self.app(environ, start_response)
            return self._serve_static_site(environ, start_response, host, path)

        return self.app(environ, start_response)

    def _serve_static_site(self, environ, start_response, host, path):
        """为静态站点（host -> 目录）提供文件服务
        - 默认入口：目录请求 / 返回 index.html
        - 仅允许访问站点根目录内的文件（防止 ../ 越权）
        """
        import mimetypes
        mimetypes.add_type('application/javascript', '.js')
        mimetypes.add_type('text/css', '.css')
        mimetypes.add_type('image/svg+xml', '.svg')

        site_root = STATIC_SITES[host].resolve()
        # 清理路径：去掉前导 /，URL 反解码，过滤 ..
        rel_path = path.lstrip('/')
        if not rel_path:
            rel_path = 'index.html'
        try:
            from urllib.parse import unquote
            rel_path = unquote(rel_path)
        except Exception:
            pass
        # 防止目录越权
        if '..' in rel_path.split('/'):
            start_response('403 Forbidden', [('Content-Type', 'text/plain')])
            return [b'Forbidden']

        target = (site_root / rel_path).resolve()
        try:
            target.relative_to(site_root)
        except ValueError:
            start_response('403 Forbidden', [('Content-Type', 'text/plain')])
            return [b'Forbidden']

        # 目录则尝试 index.html
        if target.is_dir():
            target = target / 'index.html'
        # SPA 兜底：客户端路由（无扩展名，如 /project/<包名>/）回退站点入口
        if not target.is_file() and host in SPA_STATIC_SITES \
                and '.' not in os.path.basename(rel_path):
            spa_index = site_root / 'index.html'
            if spa_index.is_file():
                target = spa_index
        if not target.is_file():
            start_response('404 Not Found', [('Content-Type', 'text/html')])
            try:
                content = open(Path("TempLate") / "Errors" / "404.html", encoding="utf-8").read().encode('utf-8')
            except Exception:
                content = b'404 Not Found'
            return [content]

        try:
            data = target.read_bytes()
        except Exception as e:
            start_response('500 Internal Server Error', [('Content-Type', 'text/plain')])
            return [f'Read error: {e}'.encode('utf-8')]

        ctype, _ = mimetypes.guess_type(str(target))
        if not ctype:
            ctype = 'application/octet-stream'
        start_response('200 OK', [
            ('Content-Type', ctype),
            ('Content-Length', str(len(data))),
            ('Cache-Control', 'no-cache'),
        ])
        return [data]


def SetupLogging():
    global logger
    log_dir = "./logs"
    os.makedirs(log_dir, exist_ok=True)
    log_filename = os.path.join(log_dir, f"server_{datetime.datetime.now().strftime('%Y-%m-%d')}.log")

    log_level = getattr(logging, LOG_CONFIG['log_level'].upper(), logging.INFO)
    formatter = WholeLineColorFormatter()

    file_handler = logging.handlers.RotatingFileHandler(
        log_filename, maxBytes=10*1024*1024, backupCount=5, encoding='utf-8'
    )
    file_handler.setFormatter(formatter)

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)

    logging.basicConfig(level=log_level, handlers=[file_handler, console_handler], force=True)

    for lib in ['django', 'django.request', 'django.server', 'django.db.backends']:
        logging.getLogger(lib).setLevel(logging.WARNING)

    logger = logging.getLogger('server')
    logger.setLevel(log_level)
    logger.propagate = False

    if not logger.handlers:
        logger.addHandler(file_handler)
        logger.addHandler(console_handler)

    return logger


class ConnectionPoolManager:
    _instance = None
    _lock = gevent.lock.BoundedSemaphore(1)

    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
                    cls._instance._init_pools()
        return cls._instance

    def _init_pools(self):
        self.coroutine_pool = Pool(MAX_COROUTINES)
        self.thread_pool = ThreadPool(MAX_THREADS)


def CreateSslContext(cert_path, key_path):
    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    context.load_cert_chain(certfile=cert_path, keyfile=key_path)
    context.minimum_version = ssl.TLSVersion.TLSv1_2
    context.maximum_version = ssl.TLSVersion.TLSv1_3
    context.set_ciphers('ECDHE-ECDSA-AES128-GCM-SHA256:ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-CHACHA20-POLY1305:ECDHE-RSA-CHACHA20-POLY1305')
    context.options |= ssl.OP_CIPHER_SERVER_PREFERENCE
    return context


def RedirectApp(environ, start_response):
    query = environ.get('QUERY_STRING', '')
    path = environ.get('PATH_INFO', '')
    if query:
        path += '?' + query
    host = environ.get('HTTP_HOST', '')
    location = f'https://{host}{path}'

    domain = RequestFilter.extract_domain(host)
    logger.info(f"[301] [GET] {path} - {environ.get('REMOTE_ADDR', '')} - 0.000000s",
                extra={'status_code': 301, 'domain': domain, 'host': host})

    start_response('301 Moved Permanently', [
        ('Location', location),
        ('Content-Type', 'text/plain'),
        ('Access-Control-Allow-Origin', f'https://{host}'),
        ('Access-Control-Allow-Credentials', 'true')
    ])
    return [b'Redirecting to HTTPS...']


def get_django_wsgi_application():
    from django.core.wsgi import get_wsgi_application
    app = get_wsgi_application()
    app = FilteredLoggingMiddleware(app)
    app = SecurityMiddleware(app)
    from gvsdsdk.msync.sio import create_sio_app
    app = create_sio_app(app)
    return app


class HotReloader:
    """文件变更探测（轮询 .py 的 mtime/size，**不用 watchdog**）

    **为什么不用 watchdog**：`monkey.patch_all()` 把 threading 一起打了补丁，watchdog 的
    Observer/BaseThread 于是变成 gevent 协程（实测 MRO：BaseThread → gevent.threading.Thread）；
    它内部阻塞的 ReadDirectoryChangesW 是 Win32 阻塞调用、不会让出 hub，一个协程就能把整个
    事件循环卡死 —— 端口全在监听却不再 accept，80/443 与控制端口一起静默失联。
    轮询是无副作用的纯检查，代价可忽略（约 400 个文件），且不引入任何线程/dependency。
    """

    def __init__(self, watch_dirs=None, extensions=None):
        self.watch_dirs = watch_dirs or ['./gvsds']
        self.extensions = extensions or {'.py'}
        self._changed = gevent.event.Event()
        self._snapshot = {}

    def _ShouldWatch(self, path):
        ext = os.path.splitext(path)[1].lower()
        if ext not in self.extensions:
            return False
        parts = os.path.normpath(path).split(os.sep)
        return '__pycache__' not in parts and '.git' not in parts

    def _Scan(self):
        """扫描监视目标，返回 {路径: (mtime, size)}（目录递归、单文件直取）"""
        snapshot = {}
        for target in self.watch_dirs:
            node = Path(target)
            if node.is_dir():
                candidates = node.rglob('*')
            elif node.is_file():
                candidates = [node]
            else:
                continue
            for item in candidates:
                if not self._ShouldWatch(str(item)):
                    continue
                try:
                    stat = item.stat()
                except OSError:
                    continue
                snapshot[str(item)] = (stat.st_mtime, stat.st_size)
        return snapshot

    def start(self):
        """记录初始快照作为基线（之后只对比变化）"""
        self._snapshot = self._Scan()
        logger.info(
            f"[热重启] 文件监听已启动（轮询 mtime）: {self.watch_dirs}，共 {len(self._snapshot)} 个文件",
            extra={'domain': 'system', 'status_code': 0},
        )

    def _Poll(self):
        snapshot = self._Scan()
        detail = None
        for path, fingerprint in snapshot.items():
            if self._snapshot.get(path) != fingerprint:
                detail = os.path.relpath(path, '.')
                break
        if detail is None and len(snapshot) != len(self._snapshot):
            detail = '文件新增或删除'
        self._snapshot = snapshot
        if detail is None:
            return
        logger.warning(f"[热重启] 检测到文件变更: {detail}", extra={'domain': 'system', 'status_code': 0})
        self._changed.set()

    def wait_for_change(self, timeout=None):
        """等一个轮询周期后检查一次变更；返回是否已置位"""
        self._changed.clear()
        gevent.sleep(timeout if timeout is not None else 1.0)
        self._Poll()
        return self._changed.is_set()

    @property
    def is_changed(self):
        return self._changed.is_set()


class ControlServer:
    """单实例控制端口（仅 127.0.0.1）

    两个职责：
    1. **单实例闸门** —— 绑定该端口即"我是唯一实例"，后来者绑不上就有据可查；
    2. **热重载通道** —— 后来者先 PING 探测，再发 RELOAD 让既有实例自重启，
       从而做到"全局只有一个 Main.py 实例，重启是通知而不是新起一个抢 80/443"。

    行协议（客户端发一行，服务端回一行）：
        PING\\n    → PONG\\n
        RELOAD\\n  → OK\\n（随后由主协程按本进程自己的启动参数重启，不在本协程里做）

    两个必须遵守的实现约束：
    - **不能设 SO_REUSEADDR** —— Windows 下它允许重复绑定同一端口，单实例闸门会失效；
    - **连接就地处理、不另开协程** —— 消息只有几十字节，直接 accept 后同步收发即可；
      一旦改成 spawn，处理协程若因任何原因没被调度，连接就会堆在 CloseWait 里占满
      backlog，后来者被"连接拒绝"，闸门与通知通道一起失效（实测踩过）。
    """

    def __init__(self):
        self.host = CONTROL_HOST
        self.port = CONTROL_PORT
        self.on_reload = None
        self.socket = None
        self.greenlet = None
        self.last_error = None

    def SetReloadHandler(self, handler):
        self.on_reload = handler
        return self

    def TryBind(self):
        """尝试绑定控制端口；成功返回 True，被占用返回 False（原因存 last_error）"""
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            sock.bind((self.host, self.port))
            sock.listen(8)
        except OSError as e:
            self.last_error = e
            try:
                sock.close()
            except Exception:
                pass
            return False
        sock.settimeout(1)
        self.socket = sock
        return True

    def start(self):
        self.greenlet = gevent.spawn(self._Serve)
        return self

    def _Serve(self):
        while True:
            try:
                conn, addr = self.socket.accept()
            except socket.timeout:
                continue
            except (OSError, gevent.GreenletExit):
                return
            self._Handle(conn, addr)

    def _Handle(self, conn, addr):
        """逐行收发（一个连接可以连发多条命令：PING 探活 + RELOAD 通知）"""
        try:
            conn.settimeout(CONTROL_TIMEOUT)
            buffer = b''
            while True:
                try:
                    chunk = conn.recv(256)
                except socket.timeout:
                    return
                if not chunk:
                    return
                buffer += chunk
                while b'\n' in buffer:
                    line, buffer = buffer.split(b'\n', 1)
                    if not self._Dispatch(conn, addr, line.decode('utf-8', 'replace').strip()):
                        return
        except Exception:
            pass
        finally:
            try:
                conn.close()
            except Exception:
                pass

    def _Dispatch(self, conn, addr, line):
        """处理一条命令；返回 False 表示可以收尾（收到重启通知）"""
        command = line.split(' ', 1)[0].upper()
        if command == 'PING':
            conn.sendall(b'PONG\n')
            return True
        if command in ('RELOAD', 'RESTART'):
            logger.warning(f"[单实例] 收到来自 {addr[0]} 的重启通知", extra={'domain': 'system', 'status_code': 0})
            conn.sendall(b'OK\n')
            # 只置标志，真正重启交给主协程（本协程若在此 stop_all 会被连带停掉）
            if self.on_reload:
                self.on_reload()
            return False
        conn.sendall(b'ERR\n')
        return True

    def stop(self):
        greenlet, self.greenlet = self.greenlet, None
        if greenlet is not None:
            try:
                greenlet.kill()
            except Exception:
                pass
        sock, self.socket = self.socket, None
        if sock is not None:
            try:
                sock.close()
            except Exception:
                pass


def ProbeRunningInstance():
    """PING 探测是否已有 Main.py 实例（**不发重启指令**，只用于给出准确报错信息）"""
    try:
        with socket.create_connection((CONTROL_HOST, CONTROL_PORT), timeout=CONTROL_TIMEOUT) as conn:
            conn.settimeout(CONTROL_TIMEOUT)
            conn.sendall(b'PING\n')
            return conn.recv(64).strip().upper().startswith(b'PONG')
    except OSError:
        return False


def NotifyRunningInstance():
    """探测既有 Main.py 实例并通知它重启；通知成功返回 True

    连不上 / 不应答 / 回的既不是 PONG，都说明"没有既有实例"，一律返回 False，
    由调用方决定怎么办（`--reload` 直接退出，不带参数则继续抢绑端口）。
    """
    try:
        with socket.create_connection((CONTROL_HOST, CONTROL_PORT), timeout=CONTROL_TIMEOUT) as conn:
            conn.settimeout(CONTROL_TIMEOUT)
            conn.sendall(b'PING\n')
            if not conn.recv(64).strip().upper().startswith(b'PONG'):
                return False
            conn.sendall(b'RELOAD\n')
            conn.recv(64)
            return True
    except OSError:
        return False


def WaitForPortFree(port, host='0.0.0.0', timeout=INSTANCE_TAKEOVER_SECONDS):
    """等待端口真正可绑（execv 重启时旧进程会短暂占着 80/443）

    **探测用的临时 socket 不能设 SO_REUSEADDR**：Windows 下带该选项的 bind 在端口被占用时
    也会"成功"，于是新进程会在旧进程尚未退出时抢绑 80/443 —— 表面绑上了，实际没有任何
    连接进来（静默停服）。必须用裸 bind 才能如实反映"是否仍被占用"。
    """
    deadline = time.time() + timeout
    while True:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        try:
            sock.bind((host, port))
            return True
        except OSError:
            pass
        finally:
            try:
                sock.close()
            except Exception:
                pass
        if time.time() >= deadline:
            return False
        time.sleep(INSTANCE_RETRY_INTERVAL)


def AcquireInstanceControl():
    """抢绑单实例控制端口（仅"启动服务器"的路径调用）：返回 (control, status)

    status 取值：
        'primary'  —— 本进程拿到控制端口，是唯一实例（control 已绑定，尚未 start）
        'occupied' —— 端口被占（既有实例或别的程序），调用方应报错退出

    **顺序很关键：先抢绑端口（瞬时），绑不上再 PING 看是不是既有实例（≤0.5s）。**
    反着来（先 PING 再绑）会在"上一个实例刚 execv、端口半开"时白等十几秒 —— 启动慢的根源。
    """
    control = ControlServer()
    deadline = time.time() + INSTANCE_TAKEOVER_SECONDS
    while True:
        if control.TryBind():
            return control, 'primary'
        if time.time() >= deadline:
            return None, 'occupied'
        time.sleep(INSTANCE_RETRY_INTERVAL)


class ServerManager:
    def __init__(self, control=None):
        global logger
        if logger is None:
            SetupLogging()

        self.pool_manager = ConnectionPoolManager()
        self.servers = []
        # 控制端口（单实例闸门 + --reload 通知通道）：由 __main__ 抢绑后传入
        self.control_server = control
        # 控制端口请求的重启信号：由控制协程置位，主协程执行重启
        self.reload_signal = gevent.event.Event()
        self.ssl_context = CreateSslContext(
            os.environ.get('SSL_CERT_PATH', './SSL/gvsds.com_public.crt'),
            os.environ.get('SSL_KEY_PATH', './SSL/gvsds.com.key')
        )

        self.django_app = get_django_wsgi_application()

        logger.info("Django 应用初始化完成", extra={'domain': 'system', 'status_code': 0})

    def add_http_redirect_server(self, port=80):
        server = CustomWSGIServer(
            ('0.0.0.0', port), RedirectApp,
            spawn=self.pool_manager.coroutine_pool,
            log=NoOpLog(), error_log=NoOpLog(),
            environ={'SERVER_NAME': SERVER_NAME, 'SERVER_PORT': str(port)},
        )
        self.servers.append(('HTTP Redirect', port, server, None))
        return self

    def add_django_server(self, port, is_ssl=True, app_name="Django App"):
        kwargs = {
            'listener': ('0.0.0.0', port),
            'application': self.django_app,
            'spawn': self.pool_manager.coroutine_pool,
            'log': NoOpLog(),
            'error_log': NoOpLog(),
            'suppress_ragged_eofs': True,
            'do_handshake_on_connect': False,
            'environ': {'SERVER_NAME': SERVER_NAME, 'SERVER_PORT': str(port)},
        }
        if is_ssl:
            kwargs['ssl_context'] = self.ssl_context

        server = CustomWSGIServer(**kwargs)
        self.servers.append((app_name, port, server, None))
        return self

    def RequestReload(self):
        """控制端口收到重启通知：置位重启信号（真正的重启由主协程按本进程自身参数执行）"""
        self.reload_signal.set()

    def start_all(self):
        logger.info("=== 启动 Django 服务器集群 ===", extra={'domain': 'system', 'status_code': 0})
        logger.info(f"协程池: {MAX_COROUTINES} | 线程池: {MAX_THREADS}", extra={'domain': 'system', 'status_code': 0})

        # 启动单实例控制端口：绑定动作已在 __main__ 的 AcquireInstanceControl() 完成，
        # 这里只挂上重启回调并开接受循环（80/443 之前启动，闸门才算真正生效）
        if self.control_server is None:
            self.control_server = ControlServer()
            if not self.control_server.TryBind():
                logger.critical(
                    f"控制端口 {CONTROL_HOST}:{CONTROL_PORT} 绑定失败"
                    f"（{self.control_server.last_error}）：已有 Main.py 实例或该端口被占用，本进程退出",
                    extra={'domain': 'system', 'status_code': 0},
                )
                sys.exit(1)
        self.control_server.SetReloadHandler(self.RequestReload).start()
        logger.info(f"单实例控制端口: {CONTROL_HOST}:{CONTROL_PORT}", extra={'domain': 'system', 'status_code': 0})

        from django.urls import get_resolver
        resolver = get_resolver()
        logger.info("=== Django URL 路由已加载 ===", extra={'domain': 'system', 'status_code': 0})

        running_servers = []
        for name, port, server, _ in self.servers:
            # 等端口真正空闲再绑：execv 重启时旧进程可能还没退完，抢绑会变成"静默停服"
            if not WaitForPortFree(port):
                logger.critical(
                    f"{name} 端口 {port} 仍被占用，无法监听，本进程退出",
                    extra={'domain': 'system', 'status_code': 0},
                )
                sys.exit(1)
            logger.info(f"启动 {name} 服务器 on 0.0.0.0:{port}", extra={'domain': 'system', 'status_code': 0})
            gevent.sleep(0.1)
            g = gevent.spawn(server.serve_forever)
            running_servers.append((name, port, server, g))

        self.servers = running_servers
        # 绑成功与否只有 gevent 内部知道（异常发生在协程里、不进日志），这里复核一次。
        # 用有界轮询而不是固定 sleep：固定等待在机器繁忙时会把"正在启动"误判成"启动失败"（实测踩过）。
        deadline = time.time() + INSTANCE_TAKEOVER_SECONDS
        while True:
            pending = [item for item in self.servers if not getattr(item[2], 'started', False)]
            if not pending:
                break
            if time.time() >= deadline:
                for name, port, _server, _g in pending:
                    logger.critical(
                        f"{name} 服务器未成功监听 0.0.0.0:{port}，本进程退出",
                        extra={'domain': 'system', 'status_code': 0},
                    )
                sys.exit(1)
            gevent.sleep(0.1)

        # 文件热重载常开：监视 gvsds（业务代码）+ gvsdsdk（SDK，业务大量逻辑在这里）
        # + Main.py 自身，任一 .py 变更即 os.execv 全量重启。
        reloader = HotReloader(watch_dirs=['./gvsds', './gvsdsdk', './Main.py'])
        reloader.start()

        try:
            while True:
                # 每轮最多等 1s：既轮询文件变更，也等控制端口（单实例通知）的重启信号
                reloader.wait_for_change(timeout=1)

                if self.reload_signal.is_set():
                    logger.warning("[单实例] 收到 TCP 热重载通知，正在重启服务器...", extra={'domain': 'system', 'status_code': 0})
                elif reloader.is_changed:
                    logger.warning("[热重启] 检测到文件变更，正在重启服务器...", extra={'domain': 'system', 'status_code': 0})
                else:
                    continue

                self.stop_all()
                # 重启沿用本进程自己的启动参数（--reload 那条路径不会走到这里，它只负责通知）
                os.execv(sys.executable, [sys.executable] + sys.argv)
        except KeyboardInterrupt:
            self.stop_all()

    def stop_all(self):
        logger.info("=== 停止所有服务器 ===", extra={'domain': 'system', 'status_code': 0})
        # 先释放单实例控制端口：execv 重启时新进程要能立刻绑上它
        if self.control_server is not None:
            try:
                self.control_server.stop()
            except:
                pass
            self.control_server = None
        for item in self.servers:
            if len(item) == 4:
                _, _, server, g = item
                try:
                    server.stop()
                except:
                    pass
        try:
            self.pool_manager.coroutine_pool.kill()
            self.pool_manager.thread_pool.kill()
        except:
            pass
        logger.info("所有服务器已停止", extra={'domain': 'system', 'status_code': 0})


if __name__ == "__main__":
    colorama.init(autoreset=True)
    os.environ['GEVENT_THREADPOOL_SIZE'] = str(MAX_THREADS)

    server_manager = None
    try:
        logger = SetupLogging()

        # `--reload` 是**纯指令**：只通知正在运行的实例重启，自己不新开服务器。
        # 有实例 → 通知并退出；没有实例 → 直接退出（要启动服务器请不带参数运行）。
        if '--reload' in sys.argv:
            if NotifyRunningInstance():
                logger.warning(
                    f"[单实例] 已通知正在运行的 Main.py 实例重启，本进程退出"
                    f"（控制端口 {CONTROL_HOST}:{CONTROL_PORT}）",
                    extra={'domain': 'system', 'status_code': 0},
                )
            else:
                logger.warning(
                    "[单实例] 没有正在运行的 Main.py 实例，--reload 无事可做，本进程退出"
                    "（要启动服务器请直接运行 python Main.py）",
                    extra={'domain': 'system', 'status_code': 0},
                )
            sys.exit(0)

        # 单实例：全局只允许一个 Main.py。先抢绑控制端口（瞬时），抢不到按占用报错退出。
        control, status = AcquireInstanceControl()
        if status != 'primary':
            if ProbeRunningInstance():
                detail = "已有 Main.py 实例在运行（要让它重启请执行 python Main.py --reload）"
            else:
                detail = (
                    "该端口被其它程序占用，或落入 Windows 保留端口段"
                    "（保留段内即使无进程监听也会 bind 失败并报 WinError 10013；"
                    "查看：netsh int ipv4 show excludedportrange protocol=tcp；"
                    "换端口：设置环境变量 MAIN_CONTROL_PORT）"
                )
            logger.critical(
                f"控制端口 {CONTROL_HOST}:{CONTROL_PORT} 绑定失败：{detail}，本进程退出",
                extra={'domain': 'system', 'status_code': 0},
            )
            sys.exit(1)
        logger.info("[热重启] 文件监听已启用（改动 gvsds / gvsdsdk / Main.py 自动重启）", extra={'domain': 'system', 'status_code': 0})
        server_manager = ServerManager(control)
        server_manager.add_http_redirect_server(80)
        server_manager.add_django_server(443, is_ssl=True, app_name="Main Django App")
        server_manager.start_all()
    except Exception as e:
        traceback.print_exc()
        if logger:
            logger.critical(f"服务器启动失败: {e}", exc_info=True, extra={'domain': 'system', 'status_code': 500})
        if server_manager:
            server_manager.stop_all()
    finally:
        if server_manager:
            server_manager.stop_all()
        gevent.killall([], timeout=5)