chongming-core/public/main.py

# main.py - 修复 reload 问题
#!/usr/bin/env python3
"""
FastAPI chongming 示例
主入口文件，适配 chongming 打包系统

服务器选择（APP_SERVER 环境变量，默认见 DEFAULT_SERVER）：
- gunicorn：现状保留（Linux 默认，uvloop worker + preload 多进程）
- uvicorn：Windows 默认（gunicorn 依赖 fork，Windows 不可用）
- granian：Rust HTTP 内核，多进程

启动参数保持 config.toml `[<env>.server]` 现状（host / port / workers /
reload / access_log / timeout_keep_alive），由 `_to_uvicorn`、`_to_granian`、
`_to_gunicorn_args` 转换为对应服务器的参数。
"""
import asyncio  # noqa: F401
import os
import platform
import sys
from multiprocessing import freeze_support
from pathlib import Path

import server  # noqa: F401  # 预加载：master 内完成模块银行装载，fork 出的 worker 共享
import uvloop  # noqa: F401  # 显式导入保证打包时收集 uvloop（uvicorn/granian loop 依赖）
from utils.launch import launch
from uvicorn.workers import UvicornWorker

use_shell = platform.system() == "Windows"

# 未设置 APP_SERVER 时的平台默认（保持现状）
DEFAULT_SERVER = "uvicorn" if use_shell else "gunicorn"
SUPPORTED_SERVERS = ("gunicorn", "uvicorn", "granian")

# 热重载（reload=true）监听的源码目录，避免 granian 默认扫 cwd 时被 logs/ 写日志触发
RELOAD_PATHS = [Path("src/chongming"), Path("src/plugins")]


class UvloopUvicornWorker(UvicornWorker):
    CONFIG_KWARGS = {"loop": "uvloop"}


def _server_params(app_config: dict) -> dict:
    """把 `[<env>.server]` 的现状参数整理为各服务器共用的启动参数。

    只做缺省兜底与运行环境覆盖（打包环境强制禁 reload、APP_WORKERS 覆盖副本数），
    键名保持 config.toml 现状，改名交给下面的转换函数。
    """
    server_config = app_config.get("server", {})
    # 打包环境强制禁用重载
    is_frozen = getattr(sys, "frozen", False)
    # 支持通过环境变量 APP_WORKERS 覆盖 worker 数量（方便嵌入式设备调优）
    default_workers = server_config.get("workers", 4)
    workers = int(os.environ.get("APP_WORKERS", str(default_workers)))
    return {
        "host": server_config.get("host", "0.0.0.0"),
        "port": server_config.get("port", 8000),
        "workers": workers,
        "reload": False if is_frozen else server_config.get("reload", False),
        "access_log": server_config.get("access_log", False),
        "timeout_keep_alive": server_config.get("timeout_keep_alive", 5),
        "log_level": "debug" if app_config.get("debug", False) else "info",
    }


def _to_uvicorn(params: dict) -> dict:
    """共用参数 → uvicorn.run 关键字参数（键名一致）。"""
    return {
        "host": params["host"],
        "port": params["port"],
        "reload": params["reload"],
        "workers": params["workers"],
        "log_level": params["log_level"],
        "access_log": params["access_log"],
        "timeout_keep_alive": params["timeout_keep_alive"],
    }


def _to_granian(params: dict) -> dict:
    """共用参数 → granian.Granian 关键字参数。

    仅两处改名：host → address、access_log → log_access；
    timeout_keep_alive 在 granian 无对应项（HTTP/1 keep-alive 只有开关、无空闲
    超时配置），忽略。
    """
    return {
        "address": params["host"],
        "port": params["port"],
        "interface": "asgi",
        "workers": params["workers"],
        "reload": params["reload"],
        "log_level": params["log_level"],
        "log_access": params["access_log"],
    }


def _to_gunicorn_args(params: dict) -> list:
    """共用参数 → gunicorn 命令行参数（保持现状：uvloop worker + preload）。"""
    return [
        "gunicorn",
        "server:app",
        "-k",
        "__main__.UvloopUvicornWorker",
        "--preload",
        "-w",
        str(params["workers"]),
        "-b",
        f"{params['host']}:{params['port']}",
    ]


def _select_server(logger) -> str:
    """按 APP_SERVER 环境变量选择服务器，非法值/平台不兼容时回退默认。"""
    server_kind = (os.environ.get("APP_SERVER") or DEFAULT_SERVER).strip().lower()
    if server_kind not in SUPPORTED_SERVERS:
        logger.warning(
            f"未知 APP_SERVER={server_kind!r}（可选: {', '.join(SUPPORTED_SERVERS)}），"
            f"回退 {DEFAULT_SERVER}"
        )
        server_kind = DEFAULT_SERVER
    if server_kind == "gunicorn" and use_shell:
        logger.warning("Windows 不支持 gunicorn（依赖 fork），回退 uvicorn")
        server_kind = "uvicorn"
    return server_kind


def app_run(app_config: dict, default_config: dict):
    """
    应用运行函数 - 适配模块银行环境

    Args:
        app_config: 应用配置字典，包含服务器配置、环境设置等
        default_config: 默认配置字典，包含应用名称、版本等基础信息

    Returns:
        None: 此函数不返回值，直接启动服务器并阻塞运行
    """
    from app.core.logger import get_logger  # type: ignore

    logger = get_logger("app")

    params = _server_params(app_config)
    server_kind = _select_server(logger)

    # 显示启动信息
    env = app_config.get("env", "development")
    logger.info("=" * 50)
    logger.info(f"🚀 启动 {default_config.get('app.name', 'FastAPI 应用')}")
    logger.info(f"📦 版本: {default_config.get('app.version', '1.0.0')}")
    logger.info(f"🌍 环境: {env}")
    logger.info(f"🖥️ 服务器: {server_kind}")
    logger.info(f"📍 地址: http://{params['host']}:{params['port']}")
    logger.info(f"📚 文档: http://{params['host']}:{params['port']}/docs")
    logger.info(f"🔧 调试: {'启用' if app_config.get('debug', False) else '禁用'}")
    logger.info(f"🔄 重载: {'启用' if params['reload'] else '禁用'}")
    logger.info(f"👥 Workers: {params['workers']}")
    logger.info("=" * 50)

    # 根据所选服务器启动（默认平台分支保持现状）
    if server_kind == "granian":
        import granian

        granian.Granian(
            target="server:app",
            **_to_granian(params),
            reload_paths=RELOAD_PATHS,
        ).serve()
    elif server_kind == "uvicorn":
        import uvicorn

        uvicorn.run("server:app", **_to_uvicorn(params))
    else:
        from gunicorn.app.wsgiapp import run

        sys.argv = _to_gunicorn_args(params)
        run()


def main():
    """
    主函数入口 - 初始化多进程支持并启动应用

    Returns:
        None: 此函数不返回值，调用launch函数启动应用
    """
    freeze_support()
    launch(app_run, "production")


if __name__ == "__main__":
    main()