import os import platform import shutil import subprocess import sys import asyncio from pathlib import Path from module_bank import PythonToSQLite from module_bank.encryption import Encryption from sqlite_vfs.core import SQLiteVFS from sqlite_vfs.folder_packer import FolderPacker from ..app.core.config import get_config PYINSTALLER = "pyinstaller" NUITKA = "nuitka" is_windows = platform.system() == "Windows" build_mode = PYINSTALLER if len(sys.argv) == 2: build_mode = sys.argv[1] if build_mode not in [PYINSTALLER, NUITKA]: raise ValueError(f"不支持{build_mode}打包模式") def build_vue_web(): """ 打包 Vue Web 前端项目 该函数执行以下操作: 1. 运行 npm build 命令构建 Vue 项目 2. 将构建产物打包到 SQLite VFS 虚拟文件系统中 Returns: None: 无返回值,通过打印信息反馈执行结果 """ # 构建 Vue Web 命令 dist_dir = "./src/chongming-web/dist" output_db = "./build/static.svfs" fs_name = "frontend" cmd = [ "npm", "run", "build", ] result = subprocess.run( cmd, cwd="./src/chongming-web", capture_output=True, text=True, shell=is_windows, encoding="utf-8", errors="replace", ) if result.returncode != 0: print("打包 Vue Web 失败") print("标准输出:", result.stdout) print("错误输出:", result.stderr) return else: print("Vue Web 打包成功") if result.stdout: print(result.stdout) # 检查 dist 目录是否存在 if not os.path.exists(dist_dir): print(f"错误: dist 目录不存在: {dist_dir}") return vfs = SQLiteVFS(output_db, compress=False) packer = FolderPacker( sqlite_vfs=vfs, exclude_patterns=[], fs_name=fs_name, ) try: packer.pack_folder(dist_dir) print(f"✅ 打包完成,文件保存为 {output_db}") finally: vfs.close() def run_pyarmor_obfuscate(): """ 使用 PyArmor 对关键 Python 文件进行代码混淆 该函数遍历预定义的文件列表,对每个文件执行 pyarmor gen 命令进行混淆处理, 以保护源代码不被轻易反编译。 处理的文件包括: - main.py (主入口文件) - server.py (服务器文件) - utils/launch.py (启动工具) - utils/config.py (配置工具) Returns: None: 无返回值,遇到错误时提前返回 """ # pyarmor obfuscate --output=obfuscated_script.py original_script.py obfuscate_files = [ ("main.py", "./"), ("server.py", "./"), ("utils/launch.py", "./utils"), ("utils/config.py", "./utils"), ] for file, output in obfuscate_files: cmd = [ "pyarmor", "gen", f"--output={output}", file, ] result = subprocess.run(cmd, cwd="./build", capture_output=True, text=True) if result.returncode != 0: print(f"打包文件: {file} 失败") print(result.stderr) return else: print(result.stdout) print(f"打包文件: {file}") def run_nuitka(): """ 兼容旧版 Nuitka 的打包函数（不使用任何可能不支持的参数） 编译后手动 strip + upx 压缩 """ import shutil config = get_config() build_import_base = config \ .get("pyinstaller_build_import", {}) \ .get("base", []) linux_extra = config \ .get("pyinstaller_build_import", {}) \ .get("linux", []) windows_extra = config \ .get("pyinstaller_build_import", {}) \ .get("windows", []) all_imports = list(build_import_base) if is_windows: all_imports.extend(windows_extra) else: all_imports.extend(linux_extra) packages = [ "module_bank", "sqlite_vfs", "fastapi", "uvicorn", "gunicorn", "diskcache", "bcrypt", "apscheduler", "aiofiles", "aiosqlite", "sqlmodel", "jwt", "python_multipart", "tomli", "typing_extensions" ] # 最基础的参数（兼容所有 Nuitka 版本） cmd = [ sys.executable, "-m", "nuitka", "--standalone", "--onefile", "--follow-imports", "--static-libpython=no", "--remove-output", "--assume-yes-for-downloads", ] # 尝试添加 LTO（如果不支持，会在此阶段失败，我们在下方会用 try/except 回退） cmd.append("--lto=yes") # 尝试启用 UPX 插件（旧版可能不支持） if shutil.which("upx"): cmd.append("--enable-plugin=upx") for pkg in packages: cmd.append(f"--include-package={pkg}") for mod in all_imports: cmd.append(f"--include-module={mod}") cmd.append("--disable-plugin=tk-inter") cmd.append("--disable-plugin=numpy") cmd += ["--output-dir=.", "--output-filename=chongming", "main.py"] cmd.append("--jobs=2") if is_windows: icon_path = Path("../../public/chongming.ico") if icon_path.exists(): cmd.append(f"--windows-icon-from-ico={icon_path}") print("正在运行 Nuitka 编译...") try: subprocess.run(cmd, cwd="./build", check=True, capture_output=False, text=True) except subprocess.CalledProcessError: # 如果因为 --lto=yes 或 --enable-plugin=upx 失败，则用更基础的命令重试 print("检测到参数不支持，移除 --lto=yes 和 --enable-plugin=upx 重新编译...") fallback_cmd = [arg for arg in cmd if arg not in ("--lto=yes", "--enable-plugin=upx")] subprocess.run(fallback_cmd, cwd="./build", check=True, capture_output=False, text=True) print("Nuitka 编译完成！开始后处理瘦身...") exe_path = Path("./build/chongming") if not is_windows and exe_path.exists(): # 手动 strip 去掉符号表 subprocess.run(["strip", str(exe_path)], check=False) print("已执行 strip，去除调试符号") if shutil.which("upx") and exe_path.exists(): # 手动 UPX 极限压缩 subprocess.run(["upx", "--best", str(exe_path)], check=False) print("已执行 UPX 压缩") # 清理临时入口文件 shutil.rmtree("build/utils", ignore_errors=True) os.remove("build/main.py") os.remove("build/server.py") def run_pyinstaller(): """ 使用 PyInstaller 将应用打包成单文件可执行程序 该函数根据操作系统平台(Windows/非Windows)构建不同的 PyInstaller 命令, 包含所有必要的隐藏导入以确保依赖库被正确打包。 Windows 和非 Windows 平台的主要区别: - 非 Windows 平台额外包含 gunicorn 相关模块 Returns: None: 无返回值,通过打印信息反馈执行结果 """ config = get_config() try: # 构建 PyInstaller 命令 build_import_base = config \ .get("pyinstaller_build_import", {}) \ .get("base", []) cmd = [ sys.executable, "-m", "PyInstaller", "--onefile", "--noconfirm", "--optimize", "2", "--strip", "--name", "chongming", "--distpath", ".", "--workpath", "./build_temp", "--specpath", "./specs", "--clean", # 其他选项 "--icon", "../../public/chongming.ico", "main.py", ] if is_windows: windows_build_import = config \ .get("pyinstaller_build_import", {}) \ .get("windows", []) + build_import_base for build_import in windows_build_import: cmd.append("--hidden-import") cmd.append(build_import) else: linux_build_import = config \ .get("pyinstaller_build_import", {}) \ .get("linux", []) + build_import_base for build_import in linux_build_import: cmd.append("--hidden-import") cmd.append(build_import) print(f"正在运行 PyInstaller: {' '.join(cmd)}") result = subprocess.run(cmd, cwd="./build", capture_output=True, text=True) if result.returncode == 0: print("PyInstaller 打包成功!") print(result.stdout) else: print("PyInstaller 打包失败!") print("错误信息:", result.stderr) except FileNotFoundError: print("错误:未找到 PyInstaller,请先安装 PyInstaller:pip install pyinstaller") except Exception as e: print(f"PyInstaller 执行出错: {e}") async def init_database(): """ 异步初始化数据库 该函数执行以下操作: 1. 从配置中读取数据库连接信息 2. 创建异步数据库引擎和会话工厂 3. 设置 SQLite 优化参数(WAL 模式和同步策略) 4. 创建所有 SQLModel 定义的数据库表 5. 初始化权限系统和默认管理员账户 Raises: ValueError: 当数据库配置不存在时抛出 OperationalError: 当数据库操作出现异常时抛出(已存在表的情况除外) Returns: None: 无返回值 """ from sqlalchemy.ext.asyncio import create_async_engine from sqlmodel import SQLModel, text from sqlalchemy.exc import OperationalError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from ..app.task.dev_init_db import init_permission, init_default_roles, dev_init_admin database = get_config()["database"] database_type = database["type"] database_config = database.get(database_type, None) if database_config is None: raise ValueError("配置不存在") os.makedirs("./build/data", exist_ok=True) engine = create_async_engine( "sqlite+aiosqlite:///./build/data/database.db", **database_config ) async_session_maker = async_sessionmaker( engine, class_=AsyncSession, expire_on_commit=False ) async with engine.begin() as conn: if database_type == "sqlite": await conn.execute(text("PRAGMA journal_mode=WAL")) await conn.execute(text("PRAGMA synchronous=NORMAL")) try: await conn.run_sync( lambda conn: SQLModel.metadata.create_all(bind=conn, checkfirst=True) ) except OperationalError as e: if "already exists" in str(e): raise # 添加管理员用户 async with async_session_maker() as session: # await UserService.create_user(session, "admin", "admin", is_superuser=True) # type: ignore await init_permission(session) # 再初始化默认角色及权限关联 await init_default_roles(session) # 最后初始化管理员用户 await dev_init_admin(session) async def main(): """ 主构建流程函数 该函数协调整个应用的构建过程,按顺序执行以下步骤: 1. 加密打包 Python 模块(app 和 plugins)到 SQLite 数据库文件 2. 创建并准备 build 目录结构 3. 复制必要的源文件和配置文件到 build 目录 4. 注入动态生成的加密密钥到配置文件 5. 对关键文件进行代码混淆 6. 使用 PyInstaller 打包成可执行文件 7. 清理临时构建文件 8. 初始化数据库 9. 构建 Vue 前端项目 Returns: None: 无返回值 """ # 打包模块 resources_path = Path("resources") if not resources_path.exists(): resources_path.mkdir() mbank_list = get_config()["production"].get("module_system", {}).get("path", []) for mbank in mbank_list: key = Encryption.generate_key() app_packer = PythonToSQLite(mbank[0], key) app_packer.pack_directory(mbank[1], mbank[2]) app_packer.verify_package_structure() app_packer.packer_key(key) app_packer.delete_source_code(None) # 创建 build 目录 build_path = Path("build") if build_path.exists() and build_path.is_dir(): shutil.rmtree("build") os.makedirs("build", exist_ok=True) # 复制 utils 文件夹到 build 文件夹 utils_path = Path("utils") if utils_path.exists(): target_utils_path = Path("build") / "utils" shutil.copytree(utils_path, target_utils_path, dirs_exist_ok=True) print(f"已复制 utils 文件夹到 {target_utils_path}") # 复制 resources 文件夹到 build 文件夹 resources_path = Path("resources") if resources_path.exists(): target_resources_path = Path("build") / "resources" shutil.copytree(resources_path, target_resources_path, dirs_exist_ok=True) print(f"已复制 resources 文件夹到 {target_resources_path}") print("模块打包完成") # 复制 main.py 到 build 文件夹 shutil.copy("public/main.py", "build") shutil.copy("public/server.py", "build") # 复制 config.toml 到 build 文件夹 shutil.copy("public/config.toml", "build") # 复制 utils 文件夹到 build 文件夹 shutil.copytree("public/utils", "build/utils") jwt_key = Encryption.generate_key() with open("./build/config.toml", "r", encoding="utf-8") as f: context = f.read() context = context.replace("your-secret-key-change-in-production", jwt_key) with open("build/config.toml", "w", encoding="utf-8") as f: f.write(context) if build_mode == PYINSTALLER: # 混淆 run_pyarmor_obfuscate() # 运行 PyInstaller run_pyinstaller() # 清理临时文件 shutil.rmtree(r"build/build_temp", ignore_errors=True) shutil.rmtree(r"build/specs", ignore_errors=True) shutil.rmtree(r"build/utils", ignore_errors=True) shutil.rmtree(r"build/pyarmor_runtime_000000", ignore_errors=True) os.remove(r"build/main.py") os.remove(r"build/server.py") elif build_mode == NUITKA: run_nuitka() # Nuitka 编译后清理临时文件 shutil.rmtree(r"build/utils", ignore_errors=True) # os.remove(r"build/main.py") # os.remove(r"build/server.py") print("构建完成") # 初始化数据库 await init_database() # 构建 vue 项目 build_vue_web() def build(): """ 构建入口函数 该函数作为整个构建流程的入口点,通过 asyncio.run() 启动异步主构建流程。 Returns: None: 无返回值 """ asyncio.run(main())