dev.py

import os
import platform
import re
import subprocess
import sys
import threading
import time

# Windows 中文 locale 默认 stdout 为 GBK，无法直接打印子进程返回的 UTF-8 诊断
# （如 asyncpg 的 WinError 1225 中文说明），会抛 UnicodeEncodeError 直接崩 dev.py。
# 在启动时统一将自己的 stdout / stderr reconfigure 为 utf-8 + replace 堆叠策略，
# 比在每个 log() 中重复防御更干净。
for _stream in (sys.stdout, sys.stderr):
    _reconfigure = getattr(_stream, "reconfigure", None)
    _reconfigure and _reconfigure(encoding="utf-8", errors="replace")

# 定义项目路径
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
BACKEND_DIR = os.path.join(BASE_DIR, "backend")
FRONTEND_DIR = os.path.join(BASE_DIR, "frontend")
ADMIN_DIR = os.path.join(BACKEND_DIR, "admin")

# 颜色代码（跨平台可能需要 colorama，这里为了简单只做基本处理或不加）
PROCESSES = []

# AgentScope 2.0 要求 Python >= 3.11
_MIN_PYTHON = (3, 11)

def log(message, prefix="[SYSTEM]"):
    print(f"{prefix} {message}")


def _check_python_version() -> None:
    """启动前 fail-fast：Python 版本 < 3.11 直接报错退出。

    AgentScope 2.0 在 PyPI 元数据里强制 Requires-Python >= 3.11，
    pip 会静默跳过不兼容版本，最后报 "Could not find a version”。
    提前检测能避免开发者走 5+ 分钟弯路。
    """
    if sys.version_info >= _MIN_PYTHON:
        return
    current = platform.python_version()
    required = ".".join(map(str, _MIN_PYTHON))
    log(
        f"Python {current} 不满足要求；AgentScope 2.0 需 Python >= {required}",
        "[BACKEND]",
    )
    log("Windows  : winget install -e --id Python.Python.3.12", "[BACKEND]")
    log("macOS    : brew install python@3.12", "[BACKEND]")
    log("Linux    : 参考 https://www.python.org/downloads/ 或包管理器", "[BACKEND]")
    log("安装完后请删除 backend/venv 后重新运行本脚本", "[BACKEND]")
    sys.exit(1)


def get_python_exec():
    """获取虚拟环境中的 python 解释器路径"""
    if sys.platform == "win32":
        return os.path.join(BACKEND_DIR, "venv", "Scripts", "python.exe")
    return os.path.join(BACKEND_DIR, "venv", "bin", "python")


def _venv_python_version(python_exec: str) -> tuple[int, int] | None:
    """读取现有 venv 里的 python 版本，以 (major, minor) 元组返回；读不到则 None。"""
    if not os.path.exists(python_exec):
        return None
    try:
        out = subprocess.check_output(
            [python_exec, "-c", "import sys; print(sys.version_info[0], sys.version_info[1])"],
            stderr=subprocess.STDOUT,
            text=True,
            timeout=10,
        ).strip().split()
        return int(out[0]), int(out[1])
    except Exception:  # noqa: BLE001 — 检测失败不该阻断主流程
        return None


def setup_backend():
    """检查并安装后端依赖"""
    log("Checking backend environment...", "[BACKEND]")

    _check_python_version()

    venv_path = os.path.join(BACKEND_DIR, "venv")
    python_exec = get_python_exec()

    # 如果已存在的 venv 是 Python < 3.11（旧 1.0 遗留），提示手动重建
    existing_ver = _venv_python_version(python_exec)
    if existing_ver is not None and existing_ver < _MIN_PYTHON:
        required = ".".join(map(str, _MIN_PYTHON))
        log(
            f"当前 venv 是 Python {existing_ver[0]}.{existing_ver[1]}，不满足 AgentScope 2.0 (>= {required})",
            "[BACKEND]",
        )
        log(
            f"请删除后重建：\n"
            f"  Remove-Item -Recurse -Force \"{venv_path}\"   # PowerShell\n"
            f"  rm -rf \"{venv_path}\"                        # bash\n"
            f"随后使用 Python >= {required} 重新运行本脚本即可",
            "[BACKEND]",
        )
        sys.exit(1)

    if not os.path.exists(venv_path):
        log("Creating virtual environment...", "[BACKEND]")
        subprocess.check_call([sys.executable, "-m", "venv", "venv"], cwd=BACKEND_DIR)

    # 强制 pip 以 UTF-8 读取 requirements.txt，避免 Windows 中文系统默认 GBK 解码失败
    # （requirements.txt 含中文注释，旧版 pip 在非 UTF-8 locale 下会报 UnicodeDecodeError）
    pip_env = os.environ.copy()
    pip_env["PYTHONUTF8"] = "1"
    pip_env["PYTHONIOENCODING"] = "utf-8"

    # 先升级 pip 自身（旧版 pip 23.x 在 Windows 中文环境读取含非 ASCII 的 requirements 会失败）
    log("Upgrading pip...", "[BACKEND]")
    try:
        subprocess.check_call(
            [python_exec, "-m", "pip", "install", "--upgrade", "pip"],
            cwd=BACKEND_DIR,
            env=pip_env,
        )
    except subprocess.CalledProcessError:
        log("Warning: pip upgrade failed, continuing with current version.", "[BACKEND]")

    log("Installing/Updating dependencies...", "[BACKEND]")
    try:
        subprocess.check_call(
            [python_exec, "-m", "pip", "install", "-r", "requirements.txt"],
            cwd=BACKEND_DIR,
            env=pip_env,
        )
    except subprocess.CalledProcessError:
        log("Failed to install backend dependencies. Please check requirements.txt.", "[BACKEND]")
        sys.exit(1)
        
    return python_exec

# 数据库不可达时向开发者展示的 fallback 指引（映射表，避免 if 分支堆叠）
# body 的每一行统一由 _emit_db_fallback_hints 控制缩进，不在字面量里预填空格。
_DB_FALLBACK_HINTS = (
    ("1", "Docker 一键启动（推荐）", [
        "docker compose -f deploy/docker-compose.dev.yml up -d",
    ]),
    ("2", "本机装 PostgreSQL 18", [
        "Windows : winget install -e --id PostgreSQL.PostgreSQL.18",
        "macOS   : brew install postgresql@18 && brew services start postgresql@18",
        "Linux   : 按发行版包管理器安装",
        "随后建库：createdb -U postgres kunflix_db",
    ]),
    ("3", "手动降级到 SQLite", [
        "在 backend/.env 里设置：",
        "DATABASE_URL=sqlite+aiosqlite:///./kunflix.db",
    ]),
    ("4", "自动降级 SQLite（免安装兜底）", [
        "选择下方交互菜单的 [1]，dev.py 会自动写入 backend/.env",
        "并以 SQLite 完成完整启动（仅适合本地开发/快速体验）",
    ]),
)

# SQLite 兜底连接串：写入 backend/.env 后由 config.py 加载，
# database.py 的 SQLite 方言映射（PRAGMA/写锁/池调优）自动生效。
_SQLITE_FALLBACK_URL = "sqlite+aiosqlite:///./kunflix.db"


def _check_database_connectivity(python_exec):
    """启动前探测 DATABASE_URL 是否可达；不可达时 fail-fast 并给出修复引导。

    通过独立子进程调用 backend/scripts/probe_db.py，让 backend venv 决定用哪个
    驱动（asyncpg / aiosqlite）。返回码语义见 probe_db.main。
    """
    log("Probing database connectivity...", "[DATABASE]")
    probe_script = os.path.join(BACKEND_DIR, "scripts", "probe_db.py")
    try:
        result = subprocess.run(
            [python_exec, probe_script],
            cwd=BACKEND_DIR,
            capture_output=True,
            text=True,
            timeout=15,
            encoding="utf-8",
            errors="replace",
        )
    except subprocess.TimeoutExpired:
        log("Probe timed out after 15s (network / DNS issue?)", "[DATABASE]")
        _emit_db_fallback_hints()
        sys.exit(1)

    # 返回码 -> 处理策略（映射表：0 通过 / 2 跳过 / 其余进入兜底交互）
    handlers = {
        0: lambda: log("Database is reachable.", "[DATABASE]"),
        2: lambda: log(
            "Probe skipped (backend deps not ready); will rely on Alembic to surface errors.",
            "[DATABASE]",
        ),
    }
    action = handlers.get(result.returncode)
    if action is not None:
        action()
        return

    # 不可达：把 probe 的诊断行透传出来，给出 fallback 指引 + 交互兜底
    diagnostic = (result.stderr or "").strip() or "(no diagnostic output)"
    for line in diagnostic.splitlines():
        log(line, "[DATABASE]")
    _emit_db_fallback_hints()
    _offer_db_fallback_choice(python_exec)


def _emit_db_fallback_hints():
    log("", "[DATABASE]")
    log("Cannot connect to the database. Pick one of the fixes below:", "[DATABASE]")
    for tag, title, lines in _DB_FALLBACK_HINTS:
        log(f"  {tag}) {title}", "[DATABASE]")
        for body_line in lines:
            log(f"       {body_line}", "[DATABASE]")
    log("", "[DATABASE]")


def _offer_db_fallback_choice(python_exec) -> None:
    """探测失败后的交互兜底：自动降级 SQLite / 重试探测 / 退出。

    新手开发者未装 PostgreSQL 时选 [1] 即可零依赖完整启动项目；
    非交互终端（CI / 管道）input() 抛 EOFError，按退出处理，
    保持原有的 fail-fast 行为不变。
    """
    log("[1] 自动降级 SQLite 继续启动（兜底，仅本地开发）", "[DATABASE]")
    log("[2] 我已修复，重新探测", "[DATABASE]")
    log("[3] 退出，手动处理后再运行 dev.py", "[DATABASE]")
    try:
        choice = input("[DATABASE] 请选择 [1/2/3]（默认 3）: ").strip() or "3"
    except (EOFError, KeyboardInterrupt):
        log("非交互终端或已取消，按退出处理。", "[DATABASE]")
        sys.exit(1)
    actions = {
        "1": lambda: _apply_sqlite_fallback(python_exec),
        "2": lambda: _check_database_connectivity(python_exec),
    }
    chosen = actions.get(choice)
    chosen and chosen()
    # 非 1/2 的选择视为退出
    chosen or sys.exit(1)


def _apply_sqlite_fallback(python_exec) -> None:
    """把 backend/.env 的 DATABASE_URL 改写为 SQLite（不存在则追加），
    随后重新探测连通性，通过即继续完整启动流程。"""
    env_path = os.path.join(BACKEND_DIR, ".env")
    sqlite_line = f"DATABASE_URL={_SQLITE_FALLBACK_URL}"
    try:
        with open(env_path, "r", encoding="utf-8") as fh:
            content = fh.read()
    except FileNotFoundError:
        content = ""
    # 已有 DATABASE_URL 行就替换，否则追加到文件末尾
    replaced, count = re.subn(r"^DATABASE_URL=.*$", sqlite_line, content, flags=re.MULTILINE)
    updated = f"{replaced.rstrip()}\n{sqlite_line}\n" if count else (content.rstrip() + "\n" if content else "") + f"{sqlite_line}\n"
    with open(env_path, "w", encoding="utf-8") as fh:
        fh.write(updated)
    log(f"已将 DATABASE_URL 降级为 SQLite（{env_path}）", "[DATABASE]")
    log("提示：SQLite 仅供本地开发/快速体验，生产与多人协作请切回 PostgreSQL。", "[DATABASE]")
    _check_database_connectivity(python_exec)


def init_database(python_exec):
    """初始化数据库（执行迁移和种子数据），幂等操作"""
    log("Initializing database...", "[DATABASE]")
    
    seed_script = os.path.join(BACKEND_DIR, "scripts", "seed_db.py")
    try:
        subprocess.check_call([python_exec, seed_script], cwd=BACKEND_DIR)
        log("Database initialization completed.", "[DATABASE]")
    except subprocess.CalledProcessError as e:
        log(f"Database initialization failed: {e}", "[DATABASE]")
        sys.exit(1)

def _ensure_node_deps(project_dir, tag):
    """幂等安装前端依赖：仅在 node_modules 缺失时跳 npm ci，避免每次启动污染 package-lock.json。

    - node_modules 存在：跳过，直接用现有依赖启动。
    - node_modules 缺失：用 npm ci（严格按 package-lock.json 安装，不修改 lock）；
      如果 lock 与 package.json 不一致导致 npm ci 失败，提示开发者手动处理，
      不自动 fallback 到 npm install（避免污染跨平台 lock，参见 PR #121 教训）。
    """
    node_modules = os.path.join(project_dir, "node_modules")
    if os.path.exists(node_modules):
        log("node_modules already present, skip install (use `npm ci` manually if you pulled lock changes).", tag)
        return

    log("Installing dependencies via `npm ci` (first run)...", tag)
    try:
        subprocess.check_call("npm ci --no-audit --no-fund", shell=True, cwd=project_dir)
    except subprocess.CalledProcessError:
        log(
            "`npm ci` failed. This usually means package.json is out of sync with package-lock.json.\n"
            "       Run `npm install` manually in this directory to update the lock file,\n"
            "       then review the lock diff before committing (preserve cross-platform optional deps).",
            tag,
        )
        sys.exit(1)


def setup_frontend():
    """检查并安装前端依赖"""
    log("Checking frontend environment...", "[FRONTEND]")
    _ensure_node_deps(FRONTEND_DIR, "[FRONTEND]")

def run_process(command, cwd, prefix):
    """运行一个子进程并实时打印输出"""
    process = None
    try:
        # Use shell=True for node commands, but for uvicorn/python it's better to pass it as a list 
        # or use CREATE_NEW_PROCESS_GROUP on windows to avoid signal propagation issues during reload.
        kwargs = {
            "cwd": cwd,
            "shell": True,
            "stdout": subprocess.PIPE,
            "stderr": subprocess.STDOUT,
            "text": True,
            "bufsize": 1,
            "encoding": 'utf-8',
            "errors": 'replace'
        }
        
        process = subprocess.Popen(command, **kwargs)
        PROCESSES.append(process)
        
        # 实时读取输出
        for line in process.stdout:
            print(f"{prefix} {line.strip()}")
            
        process.wait()
    except Exception as e:
        log(f"Error running process: {e}", prefix)
    finally:
        if process and process in PROCESSES:
            PROCESSES.remove(process)

def main():
    log("Starting development environment setup...")

    # 1. Setup Phase (Blocking)
    python_exec = setup_backend()
    setup_frontend()
    
    # Setup Admin Dashboard
    log("Checking admin dashboard environment...", "[ADMIN]")
    _ensure_node_deps(ADMIN_DIR, "[ADMIN]")

    # 在建库前先探测连通性——连不上就 fail-fast 给出 3 条修复命令，
    # 避免 seed_db.py 内部 asyncpg 抛出难以定位的深层堆栈。
    _check_database_connectivity(python_exec)

    # Initialize database (migrations + seed data), idempotent operation
    init_database(python_exec)

    log("Setup complete. Starting servers...")

    # 2. Start Phase (Parallel)
    # Backend Command
    # Use asyncio loop for Windows compatibility with asyncpg
    # Exclude skills/active_skills from watchfiles to prevent reload loop when toggling skills
    # Must use absolute path: uvicorn FileFilter compares exclude_dir against absolute paths from watchfiles
    active_skills_abs = os.path.join(BACKEND_DIR, "skills", "active_skills")
    # SQLite WAL 模式下 kunflix.db-wal/-shm 会随读写频繁变动；uvicorn 层
    # 通过 FileFilter 兜底排除，watchfiles 层刷屏问题另在 main.py 抑制日志
    reload_excludes = [
        active_skills_abs,
        "*.db",
        "*.db-wal",
        "*.db-shm",
    ]
    exclude_flags = " ".join(f'--reload-exclude "{pat}"' for pat in reload_excludes)
    backend_cmd = f'"{python_exec}" -m uvicorn main:app --reload {exclude_flags} --host 127.0.0.1 --port 8000 --loop asyncio'
    
    # Frontend Command
    frontend_cmd = "npm run dev"
    
    # Admin Dashboard Command
    admin_cmd = "npm run dev"

    # 使用线程并发运行
    # 注意：这里的 run_process 是阻塞的，所以需要放在线程里
    t1 = threading.Thread(target=run_process, args=(backend_cmd, BACKEND_DIR, "[BACKEND]"))
    t2 = threading.Thread(target=run_process, args=(frontend_cmd, FRONTEND_DIR, "[FRONTEND]"))
    t3 = threading.Thread(target=run_process, args=(admin_cmd, ADMIN_DIR, "[ADMIN]"))

    t1.daemon = True
    t2.daemon = True
    t3.daemon = True

    t1.start()
    t2.start()
    t3.start()

    try:
        # Give processes a moment to start and be added to the PROCESSES list
        time.sleep(2)
        
        # Keep the main thread alive.
        # Check if all processes have exited instead of just threads.
        # In Windows, Uvicorn reload might cause the thread to finish or the process to restart.
        while True:
            # Re-fetch the length to ensure we don't prematurely exit 
            # if processes are still initializing.
            if len(PROCESSES) > 0:
                active_processes = [p for p in PROCESSES if p.poll() is None]
                if not active_processes:
                    log("All processes exited. Stopping dev environment.", "[SYSTEM]")
                    break
            time.sleep(1)
    except KeyboardInterrupt:
        log("Stopping servers...", "[SYSTEM]")
        for p in list(PROCESSES):
            try:
                if sys.platform == "win32":
                    subprocess.run(["taskkill", "/F", "/T", "/PID", str(p.pid)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
                else:
                    p.terminate()
            except Exception:
                pass
        sys.exit(0)

if __name__ == "__main__":
    main()