专题图修改虚拟环境

app/api/qgis_map_export.py

@@ -379,7 +379,7 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
     import json
     import subprocess
     import tempfile
-    from app.services.qgis.qgis_env import build_qgis_command
+    from app.services.qgis.qgis_env import build_qgis_command, build_clean_subprocess_env
 
     try:
         logger.info(f"[批量产图] 开始: {len(models)} 张专题图, batch={disaster_time}")
@@ -403,12 +403,16 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
             cmd = build_qgis_command(qgis_root)
             cmd.append(tmp_json.name)
 
-            logger.info(f"[批量产图] 启动 QGIS 子进程: {' '.join(cmd[:3])}...")
+            logger.info(f"[批量产图] 启动 QGIS 子进程: {' '.join(cmd[:2])}...")
 
             result = subprocess.run(
                 cmd,
                 capture_output=True,
-                timeout=300,  # 5 分钟超时（批量处理多个模板）
+                timeout=600,  # 10 分钟超时（15 张模板 × ~40s/张 ≈ 600s）
+                env=build_clean_subprocess_env(),
+                # 使用干净的环境变量：移除 venv 的 PYTHONPATH/VIRTUAL_ENV/PATH 污染，
+                # 避免 QGIS Python 3.12 的 DLL 加载被干扰导致 0xC0000005 崩溃。
+                # QGIS Python 3.12 仅通过 sys.path 即可正确加载所有模块和 DLL。
             )
         finally:
             try:
@@ -416,7 +420,38 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
             except OSError:
                 pass
 
-        if result.returncode != 0:
+        # 解析子进程输出 —— 优先检查 stdout 是否有有效结果
+        stdout_text = result.stdout.decode("utf-8", errors="replace").strip()
+        parsed_output = None
+        if stdout_text:
+            for line in reversed(stdout_text.split("\n")):
+                line = line.strip()
+                if line.startswith("{"):
+                    try:
+                        parsed_output = json.loads(line)
+                        break
+                    except json.JSONDecodeError:
+                        continue
+
+        if parsed_output is not None:
+            batch_results = parsed_output.get("results", [])
+            success_count = sum(1 for r in batch_results if "error" not in r)
+            fail_count = len(batch_results) - success_count
+            logger.info(
+                f"[批量产图] 完成: 成功={success_count}, 失败={fail_count}"
+            )
+            for r in batch_results:
+                if "error" not in r:
+                    logger.info(f"  OK {r.get('output', 'N/A')}")
+                else:
+                    logger.error(f"  FAIL {r.get('error', 'unknown')}")
+            if success_count == 0 and fail_count > 0:
+                first_err = batch_results[0].get("error", "unknown")
+                raise RuntimeError(
+                    f"QGIS 子进程所有模型均失败 ({fail_count}张): {first_err}"
+                )
+        elif result.returncode != 0:
+            # stdout 没有有效结果且退出码异常，才报错
             stderr_text = result.stderr.decode("utf-8", errors="replace").strip()
             logger.error(f"[批量产图] QGIS 子进程失败 (exit={result.returncode}):")
             for line in stderr_text.split("\n"):
@@ -424,35 +459,15 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
             raise RuntimeError(
                 f"QGIS 子进程失败: {stderr_text[:300]}"
             )
-
-        # 解析子进程输出
-        stdout_text = result.stdout.decode("utf-8", errors="replace").strip()
-        if stdout_text:
-            for line in reversed(stdout_text.split("\n")):
-                line = line.strip()
-                if line.startswith("{"):
-                    output = json.loads(line)
-                    batch_results = output.get("results", [])
-                    success_count = sum(1 for r in batch_results if "error" not in r)
-                    fail_count = len(batch_results) - success_count
-                    logger.info(
-                        f"[批量产图] 完成: 成功={success_count}, 失败={fail_count}"
-                    )
-                    for r in batch_results:
-                        if "error" not in r:
-                            logger.info(f"  OK {r.get('output', 'N/A')}")
-                        else:
-                            logger.error(f"  FAIL {r.get('error', 'unknown')}")
-                    break
-            else:
-                logger.warning("[批量产图] 子进程输出中未找到 JSON 结果")
         else:
-            logger.warning("[批量产图] 子进程无输出，但 exit code = 0")
+            logger.warning("[批量产图] 子进程无有效输出，exit code = 0")
 
     except subprocess.TimeoutExpired:
         logger.error(f"[批量产图] QGIS 子进程超时 (300s)")
+        raise
     except Exception as e:
         logger.error(f"[批量产图] 产图失败: {e}", exc_info=True)
+        raise
 
 
 # ============================================================

app/services/qgis/qgis_env.py

@@ -74,16 +74,9 @@ def get_qgis_python_path(qgis_root: str = None) -> str | None:
 
 def build_qgis_command(qgis_root: str = None) -> list[str]:
     """
-    构建通过 .bat 包装器启动 QGIS 子进程的命令列表。
-    设置环境变量并启动 QGIS Python 3.12。
+    构建 QGIS 子进程启动命令（直接调用 Python，不经过 bat 包装器）。
+    环境变量通过 build_qgis_env() 构建后传给 subprocess.run(env=...)。
     """
-    import platform
-    if platform.system() != "Windows":
-        python_path = get_qgis_python_path(qgis_root)
-        if not python_path:
-            raise RuntimeError("未找到 QGIS Python 解释器")
-        return [python_path, get_runner_script()]
-
     if qgis_root is None:
         qgis_root = _detect_qgis_root() or "D:/QGIS"
 
@@ -95,8 +88,137 @@ def build_qgis_command(qgis_root: str = None) -> list[str]:
     if not os.path.isfile(runner_script):
         raise RuntimeError(f"QGIS Runner 脚本不存在: {runner_script}")
 
-    bat_path = _generate_bat_wrapper(qgis_root, python_path, runner_script)
-    return ["cmd.exe", "/c", bat_path]
+    return [python_path, runner_script]
+
+
+def build_qgis_env(qgis_root: str = None) -> dict:
+    """
+    构建 QGIS 子进程所需的完整环境变量字典。
+
+    基于当前进程环境继承，设置 QGIS 所需的 PYTHONPATH、PATH、
+    GDAL_DATA、PROJ_DATA 等变量。调用方直接传给 subprocess.run(env=...)。
+    """
+    import platform
+
+    if qgis_root is None:
+        qgis_root = _detect_qgis_root() or "D:/QGIS"
+
+    env = dict(os.environ)
+
+    if platform.system() != "Windows":
+        return env
+
+    # --- 检测 QGIS 应用目录 ---
+    qgis_app_dir = os.path.join(qgis_root, "apps", "qgis-ltr")
+    if not os.path.isdir(qgis_app_dir):
+        qgis_app_dir = os.path.join(qgis_root, "apps", "qgis")
+    if not os.path.isdir(qgis_app_dir):
+        raise RuntimeError(f"未找到 QGIS 应用目录: {qgis_root}\\apps\\qgis-ltr 或 qgis")
+
+    # --- 检测 Python 目录 ---
+    python_dir = os.path.join(qgis_root, "apps", "Python312")
+    if not os.path.isdir(python_dir):
+        for name in ("Python39", "Python310", "Python311"):
+            candidate = os.path.join(qgis_root, "apps", name)
+            if os.path.isdir(candidate):
+                python_dir = candidate
+                break
+
+    # --- 核心路径 ---
+    qtplugins = os.path.join(qgis_app_dir, "qtplugins")
+    qt5_plugins = os.path.join(qgis_root, "apps", "Qt5", "plugins")
+    gdal_data = os.path.join(qgis_root, "apps", "gdal", "share", "gdal")
+    qgis_python = os.path.join(qgis_app_dir, "python")
+    qgis_bin = os.path.join(qgis_app_dir, "bin")
+    qt5_bin = os.path.join(qgis_root, "apps", "Qt5", "bin")
+    gdal_lib = os.path.join(qgis_root, "apps", "gdal", "lib")
+
+    env["PYTHONPATH"] = qgis_python
+    env["QGIS_PREFIX_PATH"] = qgis_app_dir
+    env["QT_PLUGIN_PATH"] = f"{qtplugins};{qt5_plugins}"
+    env["GDAL_DATA"] = gdal_data
+    env["PYTHONUTF8"] = "1"
+    env["GDAL_FILENAME_IS_UTF8"] = "YES"
+    env["VSI_CACHE"] = "TRUE"
+    env["VSI_CACHE_SIZE"] = "1000000"
+
+    # --- PROJ 数据目录 ---
+    for pd in [
+        os.path.join(qgis_app_dir, "share", "proj"),
+        os.path.join(qgis_root, "share", "proj"),
+        os.path.join(qgis_root, "apps", "gdal", "share", "proj"),
+    ]:
+        if os.path.isfile(os.path.join(pd, "proj.db")):
+            env["PROJ_DATA"] = pd
+            break
+
+    # --- PATH：前置 QGIS 二进制目录 ---
+    prepend = f"{qgis_bin};{qt5_bin};{gdal_lib}"
+    env["PATH"] = f"{prepend};{env.get('PATH', '')}"
+
+    logger.debug(f"QGIS env built: QGIS_ROOT={qgis_root}, QGIS_APP={qgis_app_dir}")
+    return env
+
+
+def build_clean_subprocess_env() -> dict:
+    """
+    为 QGIS Python 3.12 子进程构建干净的环境变量。
+
+    问题背景：
+      主进程运行在 venv Python 3.10 中，继承了 venv 的环境变量
+      （PYTHONPATH 指向 venv site-packages、VIRTUAL_ENV、PATH 含 venv Scripts 等）。
+      QGIS Python 3.12 子进程如果继承这些变量，DLL 加载会被干扰，
+      导致 0xC0000005 (ACCESS_VIOLATION) 崩溃。
+
+      同时，QGIS_PREFIX_PATH、QT_PLUGIN_PATH 等变量也会与
+      QGIS 内置 DLL 加载机制冲突，同样导致崩溃。
+
+    策略：
+      从 os.environ 中移除所有 Python/venv 相关的污染变量，
+      只保留操作系统正常运行所需的基础变量（SystemRoot、TEMP 等）。
+      QGIS Python 3.12 仅通过 sys.path 即可正确加载所有模块和 DLL。
+    """
+    env = dict(os.environ)
+
+    # 先记录 venv 路径（后续清理 PATH 时需要用）
+    venv_root = env.get("VIRTUAL_ENV", "").lower()
+
+    # 移除 Python/venv 相关变量 —— 这些会污染 QGIS Python 3.12 的 DLL 加载
+    for key in [
+        "PYTHONPATH",      # venv 设置，指向 site-packages → DLL 冲突
+        "PYTHONHOME",      # 如果存在，改变 Python 查找路径
+        "VIRTUAL_ENV",     # venv 标识，不必要
+        "PYTHONDONTWRITEBYTECODE",
+        "QGIS_PREFIX_PATH",  # 与 QGIS 内置 DLL 加载冲突
+        "QT_PLUGIN_PATH",    # 与 Qt DLL 加载冲突
+        "GDAL_DATA",         # build_qgis_env 设置，可能导致路径冲突
+        "PROJ_DATA",         # build_qgis_env 设置，可能导致路径冲突
+    ]:
+        env.pop(key, None)
+
+    # 清理 PATH：仅移除 venv 相关路径（避免 DLL 搜索顺序冲突）
+    # 保留 QGIS/系统路径（测试证明这些是安全的）
+    if venv_root:
+        path_parts = env.get("PATH", "").split(";")
+        clean_parts = []
+        for p in path_parts:
+            pl = p.lower().strip()
+            if not pl:
+                continue
+            # 跳过 venv 相关路径（.venv/Scripts, .venv/Lib 等）
+            if venv_root in pl:
+                continue
+            clean_parts.append(p)
+        env["PATH"] = ";".join(clean_parts)
+
+    # 移除 VIRTUAL_ENV 本身（已从 PATH 清理中使用过）
+    env.pop("VIRTUAL_ENV", None)
+
+    logger.debug(
+        f"Clean env built: removed PYTHONPATH/VIRTUAL_ENV/venv-PATH, "
+        f"kept {len(env)} vars"
+    )
+    return env
 
 
 def is_qgis_available(qgis_root: str = None) -> bool:

app/services/qgis/qgis_runner.py

@@ -43,61 +43,21 @@ def _setup_python_path():
 
 
 def _setup_environment():
-    """设置 QGIS 运行所需的环境变量"""
-    # 自动检测 QGIS 应用目录（与 run_qgis.bat 保持一致）
-    qgis_app_dir = os.path.join(QGIS_ROOT, "apps", "qgis-ltr")
-    if not os.path.isdir(qgis_app_dir):
-        qgis_app_dir = os.path.join(QGIS_ROOT, "apps", "qgis")
+    """设置 QGIS 运行所需的环境变量。
 
-    # ★ 必须在任何 DLL 加载前设置 PROJ_DATA，防止 PostgreSQL 的旧 proj.db 干扰
-    for proj_dir in [
-        os.path.join(qgis_app_dir, "share", "proj"),
-        os.path.join(QGIS_ROOT, "share", "proj"),
-        os.path.join(QGIS_ROOT, "apps", "gdal", "share", "proj"),
-    ]:
-        if os.path.isdir(proj_dir):
-            os.environ["PROJ_DATA"] = proj_dir
-            break
+    注意：QGIS Python 3.12 自带完整的 DLL 配置，不应设置
+    QGIS_PREFIX_PATH、QT_PLUGIN_PATH、PATH 等变量，
+    否则会与 QGIS 内置的 DLL 加载机制冲突，
+    导致 0xC0000005 (ACCESS_VIOLATION) 崩溃。
 
-    os.environ["QGIS_PREFIX_PATH"] = qgis_app_dir
+    仅设置不影响 DLL 加载的辅助变量。
+    """
+    # 仅设置 UTF-8 和 GDAL 相关的编码变量（不影响 DLL 加载）
     os.environ["PYTHONUTF8"] = "1"
     os.environ["GDAL_FILENAME_IS_UTF8"] = "YES"
     os.environ["VSI_CACHE"] = "TRUE"
     os.environ["VSI_CACHE_SIZE"] = "1000000"
 
-    if sys.platform == "win32":
-        os.environ["QT_PLUGIN_PATH"] = (
-            f"{os.path.join(qgis_app_dir, 'qtplugins')};"
-            f"{os.path.join(QGIS_ROOT, 'apps', 'Qt5', 'plugins')}"
-        )
-        gdal_data = os.path.join(QGIS_ROOT, "apps", "gdal", "share", "gdal")
-        if os.path.isdir(gdal_data):
-            os.environ["GDAL_DATA"] = gdal_data
-
-        import ctypes
-        _dll_dirs = [
-            os.path.join(qgis_app_dir, "bin"),
-            os.path.join(QGIS_ROOT, "apps", "Qt5", "bin"),
-            os.path.join(QGIS_ROOT, "apps", "gdal", "lib"),
-        ]
-        _preload_dlls = [
-            "qgis_core.dll", "qgispython.dll",
-            "Qt5Core.dll", "Qt5Gui.dll", "Qt5Widgets.dll",
-            "Qt5Network.dll", "Qt5Svg.dll", "Qt5Xml.dll",
-            "Qt5Concurrent.dll", "Qt5PrintSupport.dll",
-        ]
-        for dll_dir in _dll_dirs:
-            if not os.path.isdir(dll_dir):
-                continue
-            os.add_dll_directory(dll_dir)
-            for dll_name in _preload_dlls:
-                dll_path = os.path.join(dll_dir, dll_name)
-                if os.path.isfile(dll_path):
-                    try:
-                        ctypes.WinDLL(dll_path)
-                    except OSError:
-                        pass
-
 
 # ============================================================
 # 2. 主逻辑
@@ -189,8 +149,22 @@ def main():
         print(f"[qgis_runner] 致命错误 ({elapsed:.1f}s): {e}", file=sys.stderr)
         sys.exit(1)
     finally:
-        qgs_app.exitQgis()
+        # 跳过 qgs_app.exitQgis() — 已知 Qt DLL 在 exitQgis() 时会触发
+        # STATUS_ACCESS_VIOLATION (0xC0000005) 崩溃，进程通过 os._exit 终止，
+        # 操作系统会回收所有资源。
+        pass
 
 
 if __name__ == "__main__":
-    main()
+    try:
+        main()
+    except Exception:
+        pass
+    # 强制终止进程，避免 Qt/QGIS 后台线程在 ExitProcess 等待期间 DLL 崩溃
+    try:
+        import ctypes
+        ctypes.windll.kernel32.TerminateProcess(
+            ctypes.windll.kernel32.GetCurrentProcess(), 0
+        )
+    except Exception:
+        os._exit(0)