专题图修改虚拟环境

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,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/config/paths.py

@@ -27,5 +27,5 @@ def get_logger(name: str = "algorithm"):
     Returns:
         logging.Logger 实例
     """
-    from app.utils.logger import get_logger as _get_logger
+    from app.utils.logger import LoggerManager
-    return _get_logger(name, str(LOG_DIR))
+    return LoggerManager.get_logger(name, str(LOG_DIR))

app/script/run_qgis.bat

@@ -30,6 +30,8 @@ set "GDAL_FILENAME_IS_UTF8=YES"
 set "VSI_CACHE=TRUE"
 set "VSI_CACHE_SIZE=1000000"
 set "PATH=%QGIS_APP%\bin;%QGIS_ROOT%\apps\Qt5\bin;%QGIS_ROOT%\apps\gdal\lib;%PATH%"
+REM 强制使用 QGIS 自带的 PROJ，避免 PostgreSQL/PostGIS 的旧 proj.db 干扰
+if exist "%QGIS_ROOT%\share\proj" set "PROJ_DATA=%QGIS_ROOT%\share\proj"
 
 REM --- 启动 ---
 "%PY%\python3.exe" %*

app/services/qgis/map_service.py

@@ -48,16 +48,10 @@ class MapService:
             template_cache.reset_project_state(project, texts, extent)
         else:
             logger.info(f"[首次加载] {os.path.basename(template_path)}")
-            modifier = TemplateModifier(self.config)
+            # 直接读原始模板，不做文件级修改（避免 ZIP 兼容性问题）
-            actual_path = modifier.modify(template_path)
+            project.read(template_path)
-            template_cache.load_template(actual_path, layout_name=model.get("mapLayout", "A4"))
-            if actual_path != template_path:
-                try:
-                    os.remove(actual_path)
-                except OSError:
-                    pass
 
-        # 更新 PostgreSQL 图层连接（仅首次加载）
+        # 更新图层连接 + GPKG/SRID/表名修正（仅首次加载）
         if not is_cache_hit:
             self._update_db_connections(project)
 
@@ -94,20 +88,50 @@ class MapService:
         return model["name"]
 
     def _update_db_connections(self, project: QgsProject) -> None:
-        """更新所有 PostgreSQL 图层的数据库连接参数"""
+        """更新图层连接 + SRID修正 + GPKG静态层替换"""
         db_config = self.config["db"]
         override = self.config.get("template_override", {})
         actual_schema = override.get("actual", {}).get("schema", "qgis")
+        static_config = self.config.get("static_layers", {})
+        static_enabled = static_config.get("enabled", False)
+        gpkg_dir = static_config.get("gpkg_dir", "")
+        static_layers_map = static_config.get("layers", {})
+
         static_count = 0
 
         for layer in project.mapLayers().values():
-            if layer.providerType() == "ogr":
+            provider = layer.providerType()
+
+            # GPKG 静态层替换：postgres 表 → 本地 GPKG 文件
+            if provider == "postgres" and static_enabled:
+                uri_str = layer.dataProvider().uri().uri()
+                for name, info in static_layers_map.items():
+                    table_key = info["table"]
+                    schema, table = table_key.split(".", 1)
+                    if f'table="{schema}"."{table}"' in uri_str:
+                        gpkg_path = os.path.join(gpkg_dir, info["file"]).replace("\\", "/")
+                        layer.setDataSource(gpkg_path, name, "ogr")
+                        static_count += 1
+                        logger.debug(f"静态图层 {name} → GPKG")
+                        break
+                else:
+                    # 没匹配到静态层，继续处理为 postgres
+                    self._fix_postgres_layer(layer, db_config, actual_schema)
+                continue
+
+            if provider == "ogr":
                 static_count += 1
                 continue
 
-            if layer.providerType() != "postgres":
+            if provider == "postgres":
-                continue
+                self._fix_postgres_layer(layer, db_config, actual_schema)
 
+        if static_count:
+            logger.info(f"静态底图已本地化: {static_count} 个图层")
+
+    @staticmethod
+    def _fix_postgres_layer(layer, db_config, actual_schema):
+        """修正单个 PostgreSQL 图层的连接参数"""
         try:
             uri = layer.dataProvider().uri()
             uri.setConnection(
@@ -117,19 +141,17 @@ class MapService:
                 db_config["username"],
                 db_config["password"],
             )
-                # 更新 schema
+            # Schema 替换
-                old_uri = uri.uri()
+            uri_str = uri.uri()
-                if f'table="{actual_schema}".' not in old_uri:
             for old_schema in SCHEMA_REPLACEMENTS:
-                        new_uri = old_uri.replace(
+                if f'table="{old_schema}".' in uri_str:
+                    uri_str = uri_str.replace(
                         f'table="{old_schema}".',
                         f'table="{actual_schema}".',
                     )
-                        if new_uri != old_uri:
+                    uri = QgsDataSourceUri(uri_str)
-                            uri = QgsDataSourceUri(new_uri)
                     break
-
+            # 表名映射
-                # 表名映射（模板表名 ≠ 目标库表名）
             uri_str = uri.uri()
             for old_name, new_name in TABLE_RENAMES.items():
                 full_old = f'table="{actual_schema}"."{old_name}"'
@@ -137,6 +159,11 @@ class MapService:
                 if full_old in uri_str:
                     uri_str = uri_str.replace(full_old, full_new)
                     uri = QgsDataSourceUri(uri_str)
+            # SRID 修正
+            uri_str = uri.uri()
+            if " srid=0 " in uri_str:
+                uri_str = uri_str.replace(" srid=0 ", " srid=4326 ")
+                uri = QgsDataSourceUri(uri_str)
 
             layer.setDataSource(uri.uri(), layer.name(), "postgres")
 
@@ -147,6 +174,3 @@ class MapService:
                 logger.error(f"图层 {layer.name()} 更新后仍无效")
         except Exception as e:
             logger.error(f"更新图层 {layer.name()} 连接失败: {e}")
-
-        if static_count:
-            logger.info(f"静态底图已本地化: {static_count} 个 GPKG 图层跳过远程连接")

app/services/qgis/qgis_env.py

@@ -11,6 +11,7 @@ QGIS 环境检测与子进程配置模块。
   - get_runner_script():    获取 qgis_runner.py 的路径
 """
 import os
+import tempfile
 from pathlib import Path
 
 from app.config.paths import get_logger
@@ -73,34 +74,153 @@ def get_qgis_python_path(qgis_root: str = None) -> str | None:
 
 def build_qgis_command(qgis_root: str = None) -> list[str]:
     """
-    构建通过静态启动脚本运行 QGIS 子进程的命令列表。
+    构建 QGIS 子进程启动命令（直接调用 Python，不经过 bat 包装器）。
-
+    环境变量通过 build_qgis_env() 构建后传给 subprocess.run(env=...)。
-    Windows: 调用 app/script/run_qgis.bat（通过 QGIS_ROOT 环境变量定位）
-    Linux:   直接调用 QGIS Python 解释器
     """
-    import platform
     if qgis_root is None:
         qgis_root = _detect_qgis_root() or "D:/QGIS"
 
+    python_path = get_qgis_python_path(qgis_root)
+    if not python_path:
+        raise RuntimeError("未找到 QGIS Python 3.12 解释器")
+
     runner_script = get_runner_script()
     if not os.path.isfile(runner_script):
         raise RuntimeError(f"QGIS Runner 脚本不存在: {runner_script}")
 
-    if platform.system() == "Windows":
-        script_dir = Path(__file__).parent.parent.parent / "script"
-        bat_path = str(script_dir / "run_qgis.bat")
-        if not os.path.isfile(bat_path):
-            raise RuntimeError(f"QGIS 启动脚本不存在: {bat_path}")
-        # 设置环境变量让 .bat 能找到 QGIS
-        os.environ["QGIS_ROOT"] = qgis_root
-        return ["cmd.exe", "/c", bat_path, runner_script]
-    else:
-        python_path = get_qgis_python_path(qgis_root)
-        if not python_path:
-            raise RuntimeError("未找到 QGIS Python 解释器")
     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:
     """检查 QGIS 环境是否可用"""
     return get_qgis_python_path(qgis_root) is not None
@@ -111,6 +231,60 @@ def get_runner_script() -> str:
     return str(Path(__file__).parent / "qgis_runner.py")
 
 
+def _generate_bat_wrapper(qgis_root: str, python_path: str, runner_script: str) -> str:
+    """生成 .bat 包装脚本，设置 QGIS 环境变量并启动 runner"""
+    # 自动检测 QGIS 应用目录（qgis-ltr 或 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")
+    qgis_app_dir = qgis_app_dir.replace("/", "\\")
+
+    # 自动检测 PROJ 目录
+    proj_data = ""
+    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")):
+            proj_data = pd.replace("/", "\\")
+            break
+
+    python_dir = os.path.join(qgis_root, "apps", "Python312").replace("/", "\\")
+    qt5_plugins = os.path.join(qgis_root, "apps", "Qt5", "plugins").replace("/", "\\")
+    qtplugins = os.path.join(qgis_app_dir, "qtplugins").replace("/", "\\")
+    gdal_data = os.path.join(qgis_root, "apps", "gdal", "share", "gdal").replace("/", "\\")
+    qgis_python_dir = os.path.join(qgis_app_dir, "python").replace("/", "\\")
+    qgis_bin = os.path.join(qgis_app_dir, "bin").replace("/", "\\")
+    qt5_bin = os.path.join(qgis_root, "apps", "Qt5", "bin").replace("/", "\\")
+    gdal_lib = os.path.join(qgis_root, "apps", "gdal", "lib").replace("/", "\\")
+
+    proj_line = f'set "PROJ_DATA={proj_data}"' if proj_data else "REM PROJ_DATA not found"
+
+    bat_content = f"""@echo off
+chcp 65001 >nul
+set "PYTHONPATH={qgis_python_dir}"
+set "QGIS_PREFIX_PATH={qgis_app_dir}"
+set "QT_PLUGIN_PATH={qtplugins};{qt5_plugins}"
+set "GDAL_DATA={gdal_data}"
+{proj_line}
+set "PYTHONUTF8=1"
+set "GDAL_FILENAME_IS_UTF8=YES"
+set "VSI_CACHE=TRUE"
+set "VSI_CACHE_SIZE=1000000"
+set "PATH={qt5_bin};{qgis_bin};{gdal_lib};%PATH%"
+"{python_path}" "{runner_script}" %*
+"""
+
+    bat_dir = os.path.join(tempfile.gettempdir(), "qgis_runner")
+    os.makedirs(bat_dir, exist_ok=True)
+    bat_path = os.path.join(bat_dir, "run_qgis.bat")
+    with open(bat_path, "w", encoding="utf-8") as f:
+        f.write(bat_content)
+    logger.debug(f"生成 QGIS 包装脚本: {bat_path}")
+    return bat_path
+
+
 def _detect_qgis_root() -> str | None:
     """
     自动检测 QGIS 安装根目录。

app/services/qgis/qgis_runner.py

@@ -43,47 +43,21 @@ def _setup_python_path():
 
 
 def _setup_environment():
-    """设置 QGIS 运行所需的环境变量"""
+    """设置 QGIS 运行所需的环境变量。
-    qgis_app_dir = os.path.join(QGIS_ROOT, "apps", "qgis-ltr")
+
-    os.environ["QGIS_PREFIX_PATH"] = qgis_app_dir
+    注意：QGIS Python 3.12 自带完整的 DLL 配置，不应设置
+    QGIS_PREFIX_PATH、QT_PLUGIN_PATH、PATH 等变量，
+    否则会与 QGIS 内置的 DLL 加载机制冲突，
+    导致 0xC0000005 (ACCESS_VIOLATION) 崩溃。
+
+    仅设置不影响 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. 主逻辑
@@ -175,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)

app/services/qgis/template_cache.py

@@ -37,7 +37,7 @@ class TemplateCache:
         # 保存到临时文件
         tmp_file = tempfile.NamedTemporaryFile(
             suffix=".qgz", delete=False,
-            dir=os.path.dirname(template_path),
+            dir=tempfile.gettempdir(),
         )
         tmp_path = tmp_file.name
         tmp_file.close()

app/services/qgis/template_modifier.py

@@ -69,7 +69,7 @@ class TemplateModifier:
         try:
             tmp = tempfile.NamedTemporaryFile(
                 suffix=".qgz", delete=False,
-                dir=os.path.dirname(template_path),
+                dir=tempfile.gettempdir(),
             )
             tmp_path = tmp.name
             tmp.close()

app/utils/logger.py

@@ -1,35 +1,45 @@
 """
 日志工具类
-使用 loguru 提供增强的日志功能，支持按天分割、自动清理过期日志
+主进程使用 loguru，QGIS 子进程（无 loguru）回退到标准 logging。
 """
 import sys
 from pathlib import Path
-from loguru import logger
+
+try:
+    from loguru import logger
+    _HAS_LOGURU = True
+except ImportError:
+    _HAS_LOGURU = False
+    import logging
 
 
 class LoggerManager:
-    """日志管理器 - 基于 loguru"""
+    """日志管理器"""
 
     _configured = False
 
     @classmethod
     def get_logger(cls, name: str = "algorithm", log_dir: str = "logs"):
-        """
+        if not _HAS_LOGURU:
-        获取日志记录器（loguru 不需要传统意义上的 logger 实例）
+            return cls._get_stdlib_logger(name)
-        
-        Args:
-            name: 日志名称（用于文件命名）
-            log_dir: 日志目录
-            
-        Returns:
-            loguru.logger 实例
-        """
         if not cls._configured:
-            cls._configure_logger(name, log_dir)
+            cls._configure_loguru(name, log_dir)
         return logger
 
     @classmethod
-    def _configure_logger(cls, name: str, log_dir: str):
+    def _get_stdlib_logger(cls, name: str):
+        lg = logging.getLogger(name)
+        if not lg.handlers:
+            lg.setLevel(logging.INFO)
+            h = logging.StreamHandler(sys.stderr)
+            h.setFormatter(logging.Formatter(
+                "%(asctime)s [%(threadName)s] %(levelname)-5s %(name)s - %(message)s"
+            ))
+            lg.addHandler(h)
+        return lg
+
+    @classmethod
+    def _configure_loguru(cls, name: str, log_dir: str):
         """
         配置 loguru 日志处理器