专题图修改

app/api/qgis_map_export.py

@@ -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 —— 让子进程继承父进程环境，
+                # runner 内部 _setup_environment() 会设置 QGIS 所需变量。
+                # build_qgis_env() 设置的 PYTHONPATH/PATH/QGIS_PREFIX_PATH
+                # 与 QGIS DLL 加载冲突，导致 0xC0000005 崩溃。
             )
         finally:
             try:
@@ -416,23 +420,21 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
             except OSError:
                 pass
 
-        if 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"):
-                logger.error(f"  {line}")
-            raise RuntimeError(
-                f"QGIS 子进程失败: {stderr_text[:300]}"
-            )
-
-        # 解析子进程输出
         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("{"):
-                    output = json.loads(line)
+                    try:
-                    batch_results = output.get("results", [])
+                        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(
@@ -443,16 +445,29 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
                     logger.info(f"  OK {r.get('output', 'N/A')}")
                 else:
                     logger.error(f"  FAIL {r.get('error', 'unknown')}")
-                    break
+            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"):
+                logger.error(f"  {line}")
+            raise RuntimeError(
+                f"QGIS 子进程失败: {stderr_text[:300]}"
+            )
         else:
-                logger.warning("[批量产图] 子进程输出中未找到 JSON 结果")
+            logger.warning("[批量产图] 子进程无有效输出，exit code = 0")
-        else:
-            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，不经过 bat 包装器）。
-    设置环境变量并启动 QGIS Python 3.12。
+    环境变量通过 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,76 @@ 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 [python_path, runner_script]
-    return ["cmd.exe", "/c", bat_path]
+
+
+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 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 运行所需的环境变量。
-    # 自动检测 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")
 
-    # ★ 必须在任何 DLL 加载前设置 PROJ_DATA，防止 PostgreSQL 的旧 proj.db 干扰
+    注意：QGIS Python 3.12 自带完整的 DLL 配置，不应设置
-    for proj_dir in [
+    QGIS_PREFIX_PATH、QT_PLUGIN_PATH、PATH 等变量，
-        os.path.join(qgis_app_dir, "share", "proj"),
+    否则会与 QGIS 内置的 DLL 加载机制冲突，
-        os.path.join(QGIS_ROOT, "share", "proj"),
+    导致 0xC0000005 (ACCESS_VIOLATION) 崩溃。
-        os.path.join(QGIS_ROOT, "apps", "gdal", "share", "proj"),
-    ]:
-        if os.path.isdir(proj_dir):
-            os.environ["PROJ_DATA"] = proj_dir
-            break
 
-    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__":
+    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)