重构启动方式，更方便管理维护，实现生产开发一键切换

apps/jingrow/jingrow/app.py

@@ -0,0 +1,35 @@
+"""
+应用启动入口：读取 Config 并启动 Uvicorn。
+开发与生产均使用该单一角色入口，由外层脚本/编排并行管理。
+"""
+
+from pathlib import Path
+import uvicorn
+
+from jingrow.config import Config
+
+
+def main() -> None:
+    env = str(getattr(Config, "environment", "development")).lower()
+    is_production = env == "production"
+
+    host = getattr(Config, "backend_host", "0.0.0.0")
+    port = int(getattr(Config, "backend_port", 9001))
+    reload_enabled = (not is_production) and bool(getattr(Config, "backend_reload", True))
+
+    apps_dir = Path(__file__).resolve().parents[2]
+    reload_dirs = [str(apps_dir)] if reload_enabled else None
+
+    uvicorn.run(
+        "jingrow.main:app",
+        host=host,
+        port=port,
+        reload=reload_enabled,
+        reload_dirs=reload_dirs,
+    )
+
+
+if __name__ == "__main__":
+    main()
+
+

apps/jingrow/jingrow/config.py

@@ -33,6 +33,8 @@ class Settings(BaseSettings):
     
     # 运行模式
     run_mode: str = 'api'
+    # 环境：development/production（控制启动模式、热重载等）
+    environment: str = 'development'
     
     # 本地后端主机配置
     backend_host: str = '0.0.0.0'

apps/jingrow/jingrow/dramatiq.py

@@ -0,0 +1,29 @@
+"""
+队列启动入口：读取 Config 并以 exec 方式启动 Dramatiq CLI。
+开发与生产均使用该单一角色入口，由外层脚本/编排并行管理。
+"""
+
+import os
+
+from jingrow.config import Config
+
+
+def main() -> None:
+    processes = int(getattr(Config, "worker_processes", 1))
+    threads = int(getattr(Config, "worker_threads", 1))
+
+    args = [
+        "dramatiq",
+        "jingrow.services.queue",
+        "--processes",
+        str(processes),
+        "--threads",
+        str(threads),
+    ]
+    os.execvp("dramatiq", args)
+
+
+if __name__ == "__main__":
+    main()
+
+

apps/jingrow/jingrow/run.py

@@ -1,121 +0,0 @@
-"""
-统一启动入口：`python -m jingrow.run [app|dramatiq|all]`
-app      - 读取 Config 启动 Uvicorn
-dramatiq - 读取 Config 以 exec 方式启动 Dramatiq CLI
-all      - 默认；同时以子进程启动 app 与 dramatiq，并转发信号
-"""
-
-import argparse
-import os
-import signal
-import subprocess
-import sys
-from pathlib import Path
-import uvicorn
-
-from jingrow.config import Config
-
-
-def run_app() -> None:
-    host = getattr(Config, "backend_host", "0.0.0.0")
-    port = int(getattr(Config, "backend_port", 9001))
-    reload_enabled = bool(getattr(Config, "backend_reload", True))
-
-    apps_dir = Path(__file__).resolve().parents[2]
-    reload_dirs = [str(apps_dir)] if reload_enabled else None
-
-    # 使用导入字符串以启用 uvicorn 的 reload 能力并标准化启动行为
-    uvicorn.run(
-        "jingrow.main:app",
-        host=host,
-        port=port,
-        reload=reload_enabled,
-        reload_dirs=reload_dirs,
-    )
-
-
-def run_dramatiq() -> None:
-    processes = int(getattr(Config, "worker_processes", 1))
-    threads = int(getattr(Config, "worker_threads", 1))
-
-    args = [
-        "dramatiq",
-        "jingrow.services.queue",
-        "--processes",
-        str(processes),
-        "--threads",
-        str(threads),
-    ]
-    os.execvp("dramatiq", args)
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(prog="jingrow.run", description="Jingrow 统一运行入口")
-    parser.add_argument("role", nargs="?", choices=["app", "dramatiq", "all"], default="all", help="运行角色：app/dramatiq/all，默认 all")
-    args = parser.parse_args()
-
-    if args.role == "app":
-        run_app()
-        return
-    if args.role == "dramatiq":
-        run_dramatiq()
-        return
-
-    # all：并行启动 app 与 dramatiq（作为子进程，便于优雅退出）
-    children: list[subprocess.Popen] = []
-
-    def spawn(cmd: list[str]) -> subprocess.Popen:
-        return subprocess.Popen(cmd)
-
-    api_cmd = [sys.executable, "-m", "jingrow.run", "app"]
-    dramatiq_cmd = [sys.executable, "-m", "jingrow.run", "dramatiq"]
-
-    children.append(spawn(api_cmd))
-    children.append(spawn(dramatiq_cmd))
-
-    def handle_signal(signum, frame):
-        for p in children:
-            if p.poll() is None:
-                try:
-                    p.terminate()
-                except Exception:
-                    pass
-        # 二次强杀
-        try:
-            for p in children:
-                try:
-                    p.wait(timeout=5)
-                except Exception:
-                    if p.poll() is None:
-                        p.kill()
-        finally:
-            # 以第一个子进程退出码作为整体退出码
-            code = next((p.returncode for p in children if p.returncode is not None), 0)
-            os._exit(code)
-
-    signal.signal(signal.SIGINT, handle_signal)
-    signal.signal(signal.SIGTERM, handle_signal)
-
-    # 等待任一子进程退出，随后清理另一个
-    exit_code = 0
-    try:
-        while True:
-            any_running = False
-            for p in children:
-                rc = p.poll()
-                if rc is None:
-                    any_running = True
-                else:
-                    exit_code = rc
-                    raise SystemExit
-            if not any_running:
-                break
-            signal.pause()
-    except SystemExit:
-        handle_signal(signal.SIGTERM, None)
-    finally:
-        sys.exit(exit_code)
-
-
-if __name__ == "__main__":
-    main()

dev.sh

@@ -125,9 +125,10 @@ start_all() {
     info "启动所有服务..."
     # 使用延迟启动，让后端先启动3秒
     npx concurrently \
-        --names "BACKEND,FRONTEND" \
-        --prefix-colors "blue,green" \
-        "cd apps/jingrow && python3 -m jingrow.run" \
+        --names "BACKEND,WORKER,FRONTEND" \
+        --prefix-colors "blue,yellow,green" \
+        "cd apps/jingrow && python3 -m jingrow.app" \
+        "cd apps/jingrow && python3 -m jingrow.dramatiq" \
         "sleep 3 && cd apps/jingrow/frontend && npm run dev"
 }
 
@@ -154,9 +155,10 @@ start_backend() {
     RELOAD_DIRS=$(build_reload_dirs)
     
     npx concurrently \
-        --names "BACKEND" \
-        --prefix-colors "blue" \
-        "cd apps/jingrow && python3 -m jingrow.run"
+        --names "BACKEND,WORKER" \
+        --prefix-colors "blue,yellow" \
+        "cd apps/jingrow && python3 -m jingrow.app" \
+        "cd apps/jingrow && python3 -m jingrow.dramatiq"
 }
 
 # 停止服务