修改配置

app/api/qgis_map_export.py

@@ -20,7 +20,6 @@ from app.config.paths import get_logger
 from app.config.qgis_mappings import build_static_layers_config, get_gpkg_dir
 from app.repositories import qgis_repository
 from app.schemas.api_schemas import QgisMapExportResponse, QgisMapExportRequest
-from app.utils.api_deps import get_prediction_semaphore
 from app.utils.db_helper import db_helper
 from config import settings
 
@@ -28,16 +27,7 @@ logger = get_logger("api.qgis")
 
 router = APIRouter(prefix="/qgis", tags=["专题图导出"])
 
-# 线程池（按配置初始化 worker 数量）
-_worker_threads = getattr(settings, "QGIS_WORKER_THREADS", 4)
-_thread_pool = concurrent.futures.ThreadPoolExecutor(
-    max_workers=_worker_threads,
-    thread_name_prefix="qgis-worker",
-)
-logger.info(f"QGIS 线程池初始化: {_worker_threads} workers")
-
-# 去重锁：in_progress_locks[disaster_time] = asyncio.Lock
-# 同一灾害时间的并发请求会排队，避免重复产图
+# 去重锁
 _in_progress_locks: dict[str, asyncio.Lock] = {}
 _locks_lock = asyncio.Lock()
 
@@ -45,20 +35,6 @@ _locks_lock = asyncio.Lock()
 xian_center = getattr( settings, "XIAN_CENTER", [108.948024, 34.263161])
 
 
-# ============================================================
-# 时间戳格式化
-# ============================================================
-
-def format_disaster_time(occurred_time) -> str:
-    """将 occurred_time 格式化为时间戳字符串（YYYYMMDDHHmmss），作为文件夹名"""
-    if isinstance(occurred_time, datetime):
-        return occurred_time.strftime("%Y%m%d%H%M%S")
-    elif occurred_time:
-        return str(occurred_time).replace("-", "").replace(":", "").replace(" ", "")[:14]
-    else:
-        return datetime.now().strftime("%Y%m%d%H%M%S")
-
-
 # ============================================================
 # region_code → 区县名称映射
 # ============================================================
@@ -67,7 +43,6 @@ REGION_CODE_MAP = settings.area.to_dict()
 
 
 def _resolve_district(condition: dict) -> str:
-    """从 condition.region_code 映射区县名称"""
     code = condition.get("region_code")
     if code and str(code) in REGION_CODE_MAP:
         return REGION_CODE_MAP[str(code)]
@@ -75,43 +50,23 @@ def _resolve_district(condition: dict) -> str:
 
 
 def _build_map_title(event_type: str, condition: dict, template_name: str) -> str:
-    """
-    构建地图标题。
-
-    格式：陕西西安{区县名称}{震级/降雨量}{模板名称}
-    - 地震：陕西西安临潼区5.1级地震专题图
-    - 暴雨（有降雨量）：陕西西安长安区120mm降雨专题图
-    - 暴雨（无降雨量）：陕西西安降雨专题图
-    """
     district = _resolve_district(condition)
     prefix = f"陕西西安{district}" if district else "陕西西安"
-
     if event_type == "earthquake":
         magnitude = condition.get("magnitude")
         if magnitude is not None:
             return f"{prefix}{float(magnitude)}级{template_name}"
         return f"{prefix}{template_name}"
-
     elif event_type == "rainfall":
         rainfall = condition.get("rainfall")
         if rainfall is not None and rainfall != "":
             return f"{prefix}{float(rainfall)}mm{template_name}"
         return f"{prefix}{template_name}"
-
     return f"{prefix}{template_name}"
 
 
 def _build_info_text(event_type: str, condition: dict, occurred_time) -> str:
-    """
-    构建信息面板文本（左上角橙色矩形区域）。
-
-    暴雨：时间 + 累计降雨量（如有）+ 已持续（如有）
-    地震：时间 + 震级（如有）+ 位置（如有）
-    不显示灾害类型标签（暴雨/地震）。
-    """
     lines = []
-
-    # 时间
     if isinstance(occurred_time, datetime):
         time_str = f"{occurred_time.year}年{occurred_time.month:02d}月{occurred_time.day:02d}日{occurred_time.hour:02d}时{occurred_time.minute:02d}分"
     elif occurred_time:
@@ -119,26 +74,21 @@ def _build_info_text(event_type: str, condition: dict, occurred_time) -> str:
     else:
         time_str = datetime.now().strftime("%Y年%m月%d日%H时%M分")
     lines.append(f"时间：{time_str}")
-
     if event_type == "rainfall":
         rainfall = condition.get("rainfall")
         if rainfall is not None and rainfall != "":
             lines.append(f"累计降雨量：{float(rainfall)}mm")
-
         duration = condition.get("duration")
         if duration is not None and duration != "":
             lines.append(f"已持续：{duration}")
-
     elif event_type == "earthquake":
         magnitude = condition.get("magnitude")
         if magnitude is not None and magnitude != "":
             lines.append(f"震级：{float(magnitude)}级")
-
         lon = condition.get("epicenter_lon")
         lat = condition.get("epicenter_lat")
         if lon is not None and lat is not None:
             lines.append(f"位置：经度{float(lon)}°, 纬度{float(lat)}°")
-
     return "\n".join(lines)
 
 
@@ -219,23 +169,6 @@ def _build_qgis_config(batch_folder: str) -> dict:
         "qgis": {
             "exportDpi": getattr(settings, "QGIS_EXPORT_DPI", 300),
         },
-        "template_override": {
-            "enabled": getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ENABLED", False),
-            "original": {
-                "host": getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ORIGINAL_HOST", "localhost"),
-                "port": getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ORIGINAL_PORT", 5432),
-                "dbname": getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ORIGINAL_DB_NAME", "yjzyk_xian"),
-                "schema": getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ORIGINAL_SCHEMA", "base"),
-            },
-            "actual": {
-                "host": getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ACTUAL_HOST", ""),
-                "port": getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ACTUAL_PORT", ""),
-                "dbname": getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ACTUAL_DB_NAME", "xian_new"),
-                "schema": getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ACTUAL_SCHEMA", "qgis"),
-                "username": getattr(settings, "DB_USER", "postgres"),
-                "password": getattr(settings, "DB_PASSWORD", ""),
-            },
-        },
         "static_layers": build_static_layers_config(gpkg_dir),
         "batch_folder": batch_folder,
     }
@@ -245,82 +178,54 @@ def _build_qgis_config(batch_folder: str) -> dict:
 # 接口实现
 # ============================================================
 
+# 全局并发限制
+import asyncio as _asyncio
+_concurrent = getattr(settings, "QGIS_MAX_CONCURRENT", 2)
+_qgis_semaphore = _asyncio.Semaphore(_concurrent)
+
+
 @router.post("/export/map", response_model=QgisMapExportResponse, summary="QGIS 批量专题图导出")
 async def export_map(req: QgisMapExportRequest):
     """
-    根据模拟ID批量导出专题图。同一 occurred_time 视为同一场灾害，共享文件夹
+    根据模拟ID批量导出专题图。同一 inferenceId 共享文件夹，增量产出缺失图片。
     """
     from app.services.qgis.qgis_env import is_qgis_available
     if not is_qgis_available():
-        raise HTTPException(status_code=503, detail="QGIS 环境不可用（未找到 QGIS Python 3.12 解释器）")
+        raise HTTPException(status_code=503, detail="QGIS 环境不可用")
 
-    semaphore = get_prediction_semaphore()
-
-    async with semaphore:
+    async with _qgis_semaphore:
         inference_id = req.inferenceId
         loop = asyncio.get_event_loop()
 
         try:
-            # 查询推理记录，获取 occurred_time
             inference = await loop.run_in_executor(
                 None, qgis_repository.query_inference_result, inference_id
             )
 
-            # 将 occurred_time 格式化为时间戳作为文件夹名
-            disaster_time = format_disaster_time(inference["occurred_time"])
             event_type = inference["event_type"]
+            batch_key = str(inference_id)
 
-            # 构建批次文件夹路径
-            file_store = getattr(settings, "FILE_STORE_DIR", "G:/files")
-            output_tmpl = getattr(settings, "QGIS_OUTPUT_DIR", "xian/qgis/map/:eventType/:disasterTime")
-            output_dir = output_tmpl.replace(":eventType", event_type).replace(":disasterTime", disaster_time)
+            # 构建批次文件夹路径（用 inferenceId 作为唯一文件夹名）
+            file_store = getattr(settings, "FILE_STORE_DIR", "G:/files").replace("\\", "/")
+            output_tmpl = getattr(settings, "QGIS_OUTPUT_DIR", "xian/qgis/map/:eventType/:inferenceId")
+            output_dir = output_tmpl.replace(":eventType", event_type).replace(":inferenceId", batch_key)
             batch_folder = os.path.join(file_store, output_dir).replace("\\", "/")
 
-            # 去重检查：同一 disasterTime 只产图一次
-            if os.path.isdir(batch_folder):
-                existing_files = [
-                    f for f in os.listdir(batch_folder) if f.endswith(".jpg")
-                ]
-                if existing_files and disaster_time not in _in_progress_locks:
-                    logger.info(
-                        f"[去重] disasterTime={disaster_time} 已产图，直接返回 "
-                        f"({len(existing_files)} 张)"
-                    )
-                    return QgisMapExportResponse(
-                        code=200,
-                        message=f"已存在，共{len(existing_files)}张",
-                        data=disaster_time,
-                    )
-
-            # 去重锁：同一灾害时间排队等待
+            # 去重锁
             async with _locks_lock:
-                if disaster_time not in _in_progress_locks:
-                    _in_progress_locks[disaster_time] = asyncio.Lock()
-                task_lock = _in_progress_locks[disaster_time]
+                if batch_key not in _in_progress_locks:
+                    _in_progress_locks[batch_key] = asyncio.Lock()
+                task_lock = _in_progress_locks[batch_key]
 
             async with task_lock:
-                if os.path.isdir(batch_folder):
-                    existing_files = [
-                        f for f in os.listdir(batch_folder) if f.endswith(".jpg")
-                    ]
-                    if existing_files:
-                        logger.info(
-                            f"[去重] disasterTime={disaster_time} 已被并发请求产图完成"
-                        )
-                        return QgisMapExportResponse(
-                            code=200,
-                            message=f"已存在，共{len(existing_files)}张",
-                            data=disaster_time,
-                        )
+                # 精确判断在模板扫描后（比较文件数 vs 模板数）
 
                 logger.info(
                     f"推理结果查询成功: id={inference['id']}, "
-                    f"type={inference['event_type']}, "
-                    f"occurred_time={inference['occurred_time']}, "
-                    f"disasterTime={disaster_time}"
+                    f"type={event_type}, "
+                    f"occurred_time={inference['occurred_time']}"
                 )
 
-                # 推导参数 + 构建配置
                 os.makedirs(batch_folder, exist_ok=True)
 
                 config = _build_qgis_config(batch_folder)
@@ -335,10 +240,49 @@ async def export_map(req: QgisMapExportRequest):
                     f for f in os.listdir(template_dir)
                     if f.endswith(".qgz") and not f.startswith("tmp")
                 ])
+                # 优先模板排到前面
+                priority_keywords = getattr(settings, "QGIS_PRIORITY_TEMPLATES", [])
+                if priority_keywords:
+                    def _priority_order(name: str) -> tuple:
+                        for i, kw in enumerate(priority_keywords):
+                            if kw in name:
+                                return (0, i)  # 优先组，按配置顺序
+                        return (1, name)  # 非优先组，按文件名
+                    template_files.sort(key=_priority_order)
+                    priority_count = sum(
+                        1 for f in template_files
+                        if any(kw in f for kw in priority_keywords)
+                    )
+                    logger.info(f"优先模板: {priority_count} 张排前面")
 
                 if not template_files:
                     raise FileNotFoundError(f"模板文件夹为空: {template_dir}")
 
+                # 检查已产出的图片，只生成缺失的
+                existing = set()
+                if os.path.isdir(batch_folder):
+                    existing = {f for f in os.listdir(batch_folder) if f.endswith(".jpg")}
+
+                if len(existing) == len(template_files):
+                    logger.info(f"[跳过] {len(existing)}/{len(template_files)} 张已全部产出")
+                    return QgisMapExportResponse(
+                        code=200,
+                        message=f"已存在，共{len(existing)}张",
+                        data=batch_key,
+                    )
+
+                missing_templates = []
+                for tpl_file in template_files:
+                    out_name = os.path.splitext(tpl_file)[0] + ".jpg"
+                    if out_name not in existing:
+                        missing_templates.append(tpl_file)
+
+                if len(missing_templates) < len(template_files):
+                    logger.info(
+                        f"[增量] 已有{len(existing)}张, 缺{len(missing_templates)}张"
+                    )
+                template_files = missing_templates
+
                 # 构建所有模型参数（批量模式）
                 models = []
                 for tpl_file in template_files:
@@ -352,13 +296,13 @@ async def export_map(req: QgisMapExportRequest):
                     )
                     models.append(model)
 
-                # 一次性提交所有模型到 QGIS 子进程（单次 DLL 加载）
-                _generate_batch_maps(models, config, disaster_time)
+                # 一次性提交所有模型到 QGIS 子进程（并行多进程，实时写进度）
+                _generate_batch_maps(models, config, batch_key, inference_id, file_store)
 
                 return QgisMapExportResponse(
                     code=200,
                     message=f"任务已完成，共{len(models)}张专题图",
-                    data=disaster_time,
+                    data=batch_key,
                 )
 
         except ValueError as e:
@@ -374,92 +318,192 @@ async def export_map(req: QgisMapExportRequest):
             pass
 
 
-def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None:
-    """通过 QGIS Python 3.12 子进程批量生成专题图（单次 DLL 加载）"""
+def _generate_batch_maps(models: list, config: dict, batch_key: str,
+                        inference_id: int = None, file_store: str = None) -> None:
+    """并行启动多个 QGIS 子进程，实时读取每张图进度并写 DB"""
+    import json, math, concurrent.futures, subprocess, tempfile, threading
+    from app.services.qgis.qgis_env import (
+        get_qgis_python_path, get_runner_script, build_qgis_subprocess_env,
+    )
+
+    max_workers = getattr(settings, "QGIS_PARALLEL_WORKERS", 4)
+    workers = min(max_workers, len(models))
+    chunk_size = math.ceil(len(models) / workers)
+
+    chunks = []
+    for i in range(0, len(models), chunk_size):
+        chunks.append(models[i:i + chunk_size])
+
+    logger.info(
+        f"[批量产图] {len(models)} 张图 → {len(chunks)} 个并行子进程 "
+        f"(每进程 {chunk_size} 张)"
+    )
+
+    errors = []
+    all_results = []
+    db_lock = threading.Lock()  # 保护 DB 写入
+
+    def _run_chunk(chunk_models: list, chunk_idx: int):
+        """单个子进程，逐张读取进度并实时写 DB"""
+        request = json.dumps({"config": config, "models": chunk_models}, ensure_ascii=False)
+        tmp = tempfile.NamedTemporaryFile(suffix=".json", delete=False, mode="w", encoding="utf-8")
+        tmp.write(request)
+        tmp.close()
+
+        qgis_root = getattr(settings, "QGIS_ROOT", "D:/QGIS")
+        python_path = get_qgis_python_path(qgis_root)
+        runner = get_runner_script()
+        cmd = [python_path, runner, tmp.name]
+
+        proc = subprocess.Popen(
+            cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
+            env=build_qgis_subprocess_env(qgis_root),
+            text=True, encoding="utf-8", errors="replace",
+        )
+
+        chunk_results = []
+        for line in proc.stdout:
+            line = line.strip()
+            if line.startswith("PROGRESS:"):
+                try:
+                    r = json.loads(line[len("PROGRESS:"):])
+                    chunk_results.append(r)
+                    # ★ 每张图产出后立即写 DB
+                    if inference_id and file_store and "error" not in r:
+                        _write_single_path(inference_id, r.get("output", ""), file_store, db_lock)
+                except json.JSONDecodeError:
+                    pass
+
+        proc.wait()
+        try:
+            os.remove(tmp.name)
+        except OSError:
+            pass
+
+        if proc.returncode != 0:
+            raise RuntimeError(f"[子进程{chunk_idx}] 失败 (exit={proc.returncode})")
+
+        return chunk_results
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=len(chunks)) as executor:
+        futures = {executor.submit(_run_chunk, c, i): i for i, c in enumerate(chunks)}
+        for future in concurrent.futures.as_completed(futures):
+            chunk_idx = futures[future]
+            try:
+                results = future.result()
+                all_results.extend(results)
+            except Exception as e:
+                logger.error(f"[批量产图] 子进程 {chunk_idx} 失败: {e}")
+                errors.append(str(e))
+
+    if errors and not all_results:
+        raise RuntimeError(f"所有子进程均失败: {'; '.join(errors[:2])}")
+
+    success_count = sum(1 for r in all_results if "error" not in r)
+    fail_count = len(all_results) - success_count
+    logger.info(f"[批量产图] 完成: 成功={success_count}, 失败={fail_count}")
+    for r in all_results:
+        if "error" not in r:
+            logger.info(f"  OK {r.get('output', 'N/A')}")
+        else:
+            logger.error(f"  FAIL {r['name']}: {r.get('error', 'unknown')}")
+    if success_count == 0 and fail_count > 0:
+        first_err = all_results[0].get("error", "unknown")
+        raise RuntimeError(f"所有模型均失败 ({fail_count}张): {first_err}")
+
+
+def _generate_maps_subprocess(chunk_models: list, config: dict, chunk_idx: int) -> list:
+    """单个 QGIS 子进程，处理一批模板，返回结果列表"""
     import json
     import subprocess
     import tempfile
-    from app.services.qgis.qgis_env import build_qgis_command
+    from app.services.qgis.qgis_env import (
+        get_qgis_python_path, get_runner_script, build_qgis_subprocess_env,
+    )
+
+    request_data = json.dumps(
+        {"config": config, "models": chunk_models},
+        ensure_ascii=False,
+    )
+
+    tmp_json = tempfile.NamedTemporaryFile(
+        suffix=".json", delete=False, mode="w", encoding="utf-8"
+    )
+    tmp_json.write(request_data)
+    tmp_json.close()
 
     try:
-        logger.info(f"[批量产图] 开始: {len(models)} 张专题图, batch={disaster_time}")
+        from config import settings
+        qgis_root = getattr(settings, "QGIS_ROOT", "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()
+        cmd = [python_path, runner_script, tmp_json.name]
 
-        # 构建子进程请求 JSON —— 批量格式
-        request_data = json.dumps(
-            {"config": config, "models": models},
-            ensure_ascii=False,
+        logger.info(f"[子进程{chunk_idx}] 启动: {len(chunk_models)} 张图")
+
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            timeout=600,
+            env=build_qgis_subprocess_env(qgis_root),
         )
-
-        # 将请求 JSON 写入临时文件（避免 stdin 管道问题）
-        tmp_json = tempfile.NamedTemporaryFile(
-            suffix=".json", delete=False, mode="w", encoding="utf-8"
-        )
-        tmp_json.write(request_data)
-        tmp_json.close()
-
+    finally:
         try:
-            from config import settings
-            qgis_root = getattr(settings, "QGIS_ROOT", "D:/QGIS")
-            cmd = build_qgis_command(qgis_root)
-            cmd.append(tmp_json.name)
+            os.remove(tmp_json.name)
+        except OSError:
+            pass
 
-            logger.info(f"[批量产图] 启动 QGIS 子进程: {' '.join(cmd[:3])}...")
-
-            result = subprocess.run(
-                cmd,
-                capture_output=True,
-                timeout=300,  # 5 分钟超时（批量处理多个模板）
-            )
-        finally:
-            try:
-                os.remove(tmp_json.name)
-            except OSError:
-                pass
-
-        if result.returncode != 0:
-            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()
-        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')}")
+    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
-            else:
-                logger.warning("[批量产图] 子进程输出中未找到 JSON 结果")
-        else:
-            logger.warning("[批量产图] 子进程无输出，但 exit code = 0")
+                except json.JSONDecodeError:
+                    continue
 
-    except subprocess.TimeoutExpired:
-        logger.error(f"[批量产图] QGIS 子进程超时 (300s)")
+    if parsed_output is not None:
+        results = parsed_output.get("results", [])
+        ok = sum(1 for r in results if "error" not in r)
+        logger.info(f"[子进程{chunk_idx}] 完成: {ok}/{len(results)}")
+        return results
+
+    if result.returncode != 0:
+        stderr_text = result.stderr.decode("utf-8", errors="replace").strip()
+        raise RuntimeError(f"[子进程{chunk_idx}] 失败: {stderr_text[:200]}")
+
+    logger.warning(f"[子进程{chunk_idx}] 无输出")
+    return []
+
+
+# ============================================================
+# file_path 数据库记录
+# ============================================================
+
+def _relative_path(absolute_path: str, file_store: str) -> str:
+    """绝对路径 → 相对于 FILE_STORE_DIR 的路径"""
+    fs = file_store.replace("\\", "/").rstrip("/")
+    ap = absolute_path.replace("\\", "/")
+    if ap.startswith(fs + "/"):
+        return ap[len(fs) + 1:]
+    return ap
+
+
+def _write_single_path(inference_id: int, out_path: str, file_store: str, lock) -> None:
+    """单张图产出后立即写入进度表"""
+    if not out_path or not os.path.isfile(out_path):
+        return
+    rel = _relative_path(out_path, file_store)
+    if not rel:
+        return
+    try:
+        from app.repositories.qgis_repository import qgis_repository
+        with lock:
+            qgis_repository.insert_file_paths(inference_id, [rel])
     except Exception as e:
-        logger.error(f"[批量产图] 产图失败: {e}", exc_info=True)
-
-
-# ============================================================
-# 清理函数
-# ============================================================
-
-def shutdown_thread_pool() -> None:
-    """关闭线程池（在 server.py lifespan 关闭阶段调用）"""
-    _thread_pool.shutdown(wait=False)
-    logger.info("QGIS 线程池已关闭")
+        logger.error(f"[实时写入] 失败: {e}")

app/config/qgis_mappings.py

@@ -29,6 +29,12 @@ STATIC_LAYERS = {
     "traffic_railway":    {"table": "base.traffic_railway",            "gpkg": "traffic_railway.gpkg"},
     "traffic_township":   {"table": "base.traffic_township",           "gpkg": "traffic_township.gpkg"},
     "traffic_trunk_line": {"table": "base.traffic_trunk_line",         "gpkg": "traffic_trunk_line.gpkg"},
+    # ── 新增静态层 ──
+    "积水点":             {"table": "base.hazard_hydrops",             "gpkg": "hazard_hydrops.gpkg"},
+    "排水口":             {"table": "base.lifeline_outfall",           "gpkg": "lifeline_outfall.gpkg"},
+    "供水管网":           {"table": "base.lifeline_pipe",              "gpkg": "lifeline_pipe.gpkg"},
+    "risk_census_population": {"table": "base.risk_census_population", "gpkg": "risk_census_population.gpkg"},
+    "西安乡镇":           {"table": "base.sx_xa_towns",                "gpkg": "sx_xa_towns.gpkg"},
 }
 
 # ============================================================

app/core/server.py

@@ -16,33 +16,26 @@ logger = get_logger("api")
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    """应用生命周期：启动时预加载模型"""
+    """应用生命周期：启动时预加载 DBN 模型"""
     logger.info("正在预加载DBN模型...")
     get_rainfall_model()
     get_earthquake_model()
     logger.info("DBN模型预加载完成")
 
-    # 检测 QGIS 子进程环境
+    # 检测 QGIS 子进程环境（产图时按需启动子进程）
     qgis_root = getattr(settings, "QGIS_ROOT", None)
     if qgis_root:
         try:
             from app.services.qgis.qgis_env import is_qgis_available
             if is_qgis_available(qgis_root):
-                logger.info("QGIS 环境检测通过（子进程模式）")
+                logger.info("QGIS 环境检测通过")
             else:
-                logger.warning("QGIS 环境不可用（未找到 Python 3.12 解释器），专题图功能降级")
+                logger.warning("QGIS 环境不可用，专题图功能降级")
         except Exception as e:
             logger.error(f"QGIS 环境检测失败: {e}")
 
     yield
 
-    # 清理资源
-    try:
-        from app.api.qgis_map_export import shutdown_thread_pool
-        shutdown_thread_pool()
-    except Exception:
-        pass
-
     logger.info("应用关闭")
 
 def create_app() -> FastAPI:

app/repositories/qgis_repository.py

@@ -27,5 +27,31 @@ class QgisRepository:
             "condition": row["condition"] if isinstance(row["condition"], dict) else {},
         }
 
+    @staticmethod
+    def insert_file_paths(inference_id: int, paths: list[str]) -> None:
+        """实时写入文件路径到进度表（唯一索引自动去重）"""
+        if not paths:
+            return
+        import os
+        sql = """
+            INSERT INTO xian_inference_result_file (inference_id, file_path, file_name)
+            VALUES (%s, %s, %s)
+            ON CONFLICT DO NOTHING
+        """
+        for p in paths:
+            name = os.path.basename(p)
+            db_helper.execute_update(sql, (inference_id, p, name))
+
+    @staticmethod
+    def get_file_paths(inference_id: int) -> list[str]:
+        """获取已产出的文件路径列表（从进度表）"""
+        sql = """
+            SELECT file_path FROM xian_inference_result_file
+            WHERE inference_id = %s AND is_delete = 0
+            ORDER BY create_time
+        """
+        rows = db_helper.execute_query(sql, (inference_id,))
+        return [r["file_path"] for r in rows]
+
 
 qgis_repository = QgisRepository()

app/script/export_static_layers.py

@@ -43,6 +43,11 @@ STATIC_LAYERS = [
     ("traffic_railway",    "qgis.traffic_railway",         "traffic_railway.gpkg"),
     ("traffic_township",   "qgis.traffic_township",        "traffic_township.gpkg"),
     ("traffic_trunk_line", "qgis.traffic_trunk_line",      "traffic_trunk_line.gpkg"),
+    ("积水点",             "qgis.hazard_hydrops",          "hazard_hydrops.gpkg"),
+    ("排水口",             "qgis.lifeline_outfall",        "lifeline_outfall.gpkg"),
+    ("供水管网",           "qgis.lifeline_pipe",           "lifeline_pipe.gpkg"),
+    ("风险人口",           "qgis.risk_census_population",  "risk_census_population.gpkg"),
+    ("西安乡镇",           "qgis.sx_xa_towns",             "sx_xa_towns.gpkg"),
 ]
 
 
@@ -117,12 +122,9 @@ def main():
     # 从 config.settings 读取数据库配置
     try:
         from config import settings
-        host = getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ACTUAL_HOST",
-                       getattr(settings, "DB_HOST", "47.92.216.173"))
-        port = str(getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ACTUAL_PORT",
-                           getattr(settings, "DB_PORT", 7654)))
-        dbname = getattr(settings, "QGIS_TEMPLATE_OVERRIDE_ACTUAL_DB_NAME",
-                         getattr(settings, "DB_NAME", "xian_new"))
+        host = getattr(settings, "DB_HOST", "47.92.216.173")
+        port = str(getattr(settings, "DB_PORT", 7654))
+        dbname = getattr(settings, "DB_NAME", "xian_new")
         user = getattr(settings, "DB_USER", "postgres")
         password = getattr(settings, "DB_PASSWORD", "zhangsan")
     except Exception:

app/script/run_qgis.bat

@@ -1,54 +0,0 @@
-@echo off
-REM ============================================================
-REM QGIS 子进程启动脚本（Windows）
-REM 自动检测 QGIS 安装结构，支持 qgis-ltr/qgis + Python312/Python39
-REM ============================================================
-
-if "%QGIS_ROOT%"=="" goto :no_root
-if not exist "%QGIS_ROOT%" goto :no_root_dir
-
-REM --- 检测 QGIS 应用目录 ---
-set "QGIS_APP=%QGIS_ROOT%\apps\qgis-ltr"
-if not exist "%QGIS_APP%" set "QGIS_APP=%QGIS_ROOT%\apps\qgis"
-if not exist "%QGIS_APP%" goto :no_qgis_app
-
-REM --- 检测 Python 目录 ---
-set "PY=%QGIS_ROOT%\apps\Python312"
-if not exist "%PY%" set "PY=%QGIS_ROOT%\apps\Python39"
-if not exist "%PY%" set "PY=%QGIS_ROOT%\apps\Python310"
-if not exist "%PY%" set "PY=%QGIS_ROOT%\apps\Python311"
-if not exist "%PY%" goto :no_python
-
-REM --- 设置环境 ---
-set "PYTHONHOME=%PY%"
-set "PYTHONPATH=%QGIS_APP%\python"
-set "QGIS_PREFIX_PATH=%QGIS_APP%"
-set "QT_PLUGIN_PATH=%QGIS_APP%\qtplugins;%QGIS_ROOT%\apps\Qt5\plugins"
-set "GDAL_DATA=%QGIS_ROOT%\apps\gdal\share\gdal"
-set "PYTHONUTF8=1"
-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" %*
-exit /b %ERRORLEVEL%
-
-:no_root
-echo [ERROR] QGIS_ROOT 环境变量未设置
-exit /b 1
-
-:no_root_dir
-echo [ERROR] QGIS_ROOT 目录不存在: %QGIS_ROOT%
-exit /b 1
-
-:no_qgis_app
-echo [ERROR] 未找到 QGIS 应用目录
-exit /b 1
-
-:no_python
-echo [ERROR] 未找到 QGIS Python 目录
-exit /b 1

app/script/scan_template_tables.py

@@ -0,0 +1,82 @@
+"""
+扫描所有模板，找出所有引用的 PostgreSQL 表，列出哪些已导出 GPKG、哪些还没。
+"""
+import zipfile, re, os, sys
+
+TEMPLATE_BASE = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+    "app", "data", "template"
+)
+
+# 当前已导出的 GPKG 表（从 qgis_mappings.py / export_static_layers.py）
+EXPORTED = {
+    "rivers", "sx_capital", "river", "active_fault", "sx",
+    "sx_street", "sx_xa_county", "sx_xa_county_boundary",
+    "sx_zb_county_boundary", "sx_zb_city", "sx_zb_county",
+    "traffic_expressway", "traffic_provincial", "traffic_railway",
+    "traffic_township", "traffic_trunk_line",
+}
+
+# 避难场所 OGR→PG 映射的表（已由 TemplateModifier 转换）
+OGR_TABLES = {
+    "shelter_park", "shelter_school", "shelter_cultural",
+    "shelter_defence", "shelter_gymnasium", "shelter_square", "shelter_stay",
+}
+
+all_tables = {}  # {table_name: set of templates}
+
+for event_type in ["rainfall", "earthquake"]:
+    tdir = os.path.join(TEMPLATE_BASE, event_type)
+    if not os.path.isdir(tdir):
+        continue
+    for tf in sorted(os.listdir(tdir)):
+        if not tf.endswith(".qgz") or tf.startswith("tmp"):
+            continue
+        tpath = os.path.join(tdir, tf)
+        with zipfile.ZipFile(tpath, "r") as z:
+            for item in z.infolist():
+                if item.filename.endswith(".qgs"):
+                    content = z.read(item.filename).decode("utf-8")
+                    # 匹配 datasource 中的 table="schema"."name"
+                    for m in re.finditer(r'table="(\w+)"\."(\w+)"', content):
+                        tbl = m.group(2)
+                        all_tables.setdefault(tbl, set()).add(tf)
+                    # 也检查 provider=postgres 但没有 table= 的（如震中）
+                    # 匹配 dbname='...' 格式
+                    for m in re.finditer(r"dbname='([^']+)'", content):
+                        db = m.group(1)
+                        if db == 'xxgx_client_ya':
+                            all_tables.setdefault(f"[外部DB: {db}]", set()).add(tf)
+
+print("=" * 60)
+print("已导出 GPKG (16):")
+for t in sorted(EXPORTED):
+    if t in all_tables:
+        print(f"  ✅ {t}")
+print()
+
+print("避难场所 OGR→PG (7):")
+for t in sorted(OGR_TABLES):
+    print(f"  ✅ {t}")
+print()
+
+print("未导出，需要新增:")
+need_export = []
+for tbl, tmps in sorted(all_tables.items()):
+    if tbl.startswith("[外部DB:"):
+        continue
+    if tbl in EXPORTED or tbl in OGR_TABLES:
+        continue
+    need_export.append(tbl)
+    print(f"  ❌ {tbl} ← {', '.join(sorted(tmps))}")
+print()
+
+print("外部数据库（无法静态化）:")
+for tbl, tmps in sorted(all_tables.items()):
+    if tbl.startswith("[外部DB:"):
+        print(f"  🔗 {tbl} ← {', '.join(sorted(tmps))}")
+print()
+
+print(f"总结: 需要新增导出 {len(need_export)} 张表")
+if need_export:
+    print(f"  {', '.join(sorted(need_export))}")

app/services/qgis/map_service.py

@@ -1,6 +1,6 @@
 """
 地图生成主流程控制器。
-协调模板加载、图层过滤、缩放、文本更新、导出。
+模板加载 → 图层过滤 → 缩放 → 文本更新 → 导出。
 """
 import os
 import time
@@ -9,125 +9,173 @@ from qgis.core import QgsProject, QgsDataSourceUri
 
 from app.config.paths import get_logger
 from app.config.qgis_mappings import TABLE_RENAMES, SCHEMA_REPLACEMENTS
-from .template_cache import TemplateCache
 from .template_modifier import TemplateModifier
 from .layer_filter import LayerFilter
 from .map_exporter import MapExporter
 from app.utils.map_zoom import MapZoom
-from app.config.paths import get_logger
 
 logger = get_logger("qgis.service")
 
-# 全局模板缓存（跨请求复用）
-template_cache = TemplateCache()
-
 
 class MapService:
     def __init__(self, config: dict):
         self.config = config
 
     def generate(self, model: dict) -> str:
-        """
-        执行完整的地图生成流程。
+        _timing = {}
+        t_total = time.time()
 
-        Args:
-            model: 包含地图参数的字典
-
-        Returns:
-            地图名称
-        """
-        t_start = time.time()
         template_path = model["path"]
+        template_name = os.path.basename(template_path)
         project = QgsProject.instance()
-        is_cache_hit = template_cache.is_loaded(template_path)
 
-        # 加载/恢复模板
-        if is_cache_hit:
-            logger.info(f"[缓存命中] {os.path.basename(template_path)}")
-            project, texts, extent = template_cache.restore_template(template_path)
-            template_cache.reset_project_state(project, texts, extent)
-        else:
-            logger.info(f"[首次加载] {os.path.basename(template_path)}")
-            # 直接读原始模板，不做文件级修改（避免 ZIP 兼容性问题）
-            project.read(template_path)
+        # ── 步骤 1：加载模板 ──
+        t0 = time.time()
+        project.clear()
+        modifier = TemplateModifier(self.config)
+        actual_path = modifier.modify(template_path)
+        project.read(actual_path)
+        if actual_path != template_path:
+            try:
+                os.remove(actual_path)
+            except OSError:
+                pass
+        _timing["1.load"] = time.time() - t0
 
-        # 更新图层连接 + GPKG/SRID/表名修正（仅首次加载）
-        if not is_cache_hit:
-            self._update_db_connections(project)
+        # ── 步骤 2：图层连接修正 ──
+        t0 = time.time()
+        self._fix_invalid_layers(project)
+        _timing["2.connections"] = time.time() - t0
 
-        # 图层过滤
+        # ── 步骤 3：图层过滤 ──
+        t0 = time.time()
         LayerFilter().apply(project, model)
+        _timing["3.filter"] = time.time() - t0
 
-        # 地图缩放
+        # ── 步骤 4：地图缩放 ──
+        t0 = time.time()
         layout = project.layoutManager().layoutByName(model["mapLayout"])
         if layout is None:
             available = [l.name() for l in project.layoutManager().layouts()]
-            raise RuntimeError(
-                f"模板中未找到布局 '{model['mapLayout']}'，可用布局：{available}"
-            )
-
+            raise RuntimeError(f"模板中未找到布局 '{model['mapLayout']}'，可用布局：{available}")
         map_item = layout.itemById("Map")
         zoom = MapZoom(project, layout, map_item)
         zoom.execute(model["zoomRule"], {
-            "X": model["centerX"],
-            "Y": model["centerY"],
-            "value": model["zoomValue"],
+            "X": model["centerX"], "Y": model["centerY"], "value": model["zoomValue"],
         })
+        _timing["4.zoom"] = time.time() - t0
 
-        # 文本更新 + 比例尺 + 导出
+        # ── 步骤 5：文本 + 比例尺 + 导出 ──
+        t0 = time.time()
         exporter = MapExporter(self.config, layout)
         exporter.update_texts(model)
         exporter.update_scale_bar()
         exporter.export(model["outFile"])
+        _timing["5.export"] = time.time() - t0
 
-        elapsed = time.time() - t_start
-        logger.info(
-            f"{'[缓存命中]' if is_cache_hit else '[首次加载]'} "
-            f"导出完成: {model['name']}，耗时 {elapsed:.1f}s"
-        )
+        total = time.time() - t_total
+        steps = ", ".join(f"{k}={v:.1f}s" for k, v in _timing.items())
+        logger.info(f"{template_name} → {total:.1f}s ({steps})")
         return model["name"]
 
-    def _update_db_connections(self, project: QgsProject) -> None:
-        """更新图层连接 + SRID修正 + GPKG静态层替换"""
+    def _fix_invalid_layers(self, project: QgsProject) -> None:
+        """修复 TemplateModifier 处理后仍无效的 PostgreSQL 图层。
+
+        批量查询 DB 确认表存在性，只对存在的表执行 setDataSource 修复。
+        """
+        t0 = time.time()
         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
+        actual_schema = "qgis"
 
+        invalid_pg = []
+        valid_count = 0
+        total = 0
         for layer in project.mapLayers().values():
-            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)
+            if layer.providerType() != "postgres":
                 continue
-
-            if provider == "ogr":
-                static_count += 1
+            total += 1
+            if layer.isValid():
+                valid_count += 1
                 continue
+            invalid_pg.append(layer)
 
-            if provider == "postgres":
+        if not invalid_pg:
+            elapsed = time.time() - t0
+            logger.info(f"  图层修正: 总计{total}个PG层, 全部有效, 耗时{elapsed:.3f}s")
+            return
+
+        # 批量 DB 查询确认表存在性
+        existing = self._batch_check_tables(db_config, actual_schema, invalid_pg)
+
+        fixed = 0
+        skipped_missing = 0
+        failed = 0
+        for layer in invalid_pg:
+            uri = layer.dataProvider().uri()
+            key = (uri.schema() or actual_schema, uri.table())
+            if key not in existing:
+                logger.debug(f"  跳过不存在的表: {key[0]}.{key[1]} (图层: {layer.name()})")
+                skipped_missing += 1
+                continue
+            try:
                 self._fix_postgres_layer(layer, db_config, actual_schema)
+                if layer.isValid():
+                    fixed += 1
+                else:
+                    failed += 1
+            except Exception as e:
+                logger.error(f"  修复图层 {layer.name()} 失败: {e}")
+                failed += 1
 
-        if static_count:
-            logger.info(f"静态底图已本地化: {static_count} 个图层")
+        elapsed = time.time() - t0
+        if total:
+            logger.info(
+                f"  图层修正: 总计{total}个PG层, 有效{valid_count}, "
+                f"跳过{skipped_missing}(表不存在), "
+                f"修复{fixed}, 仍失败{failed}, 耗时{elapsed:.3f}s"
+            )
+
+    @staticmethod
+    def _batch_check_tables(db_config, schema, layers):
+        """批量检查表是否存在（单次 DB 查询），返回存在的表集合"""
+        tables = set()
+        for layer in layers:
+            uri = layer.dataProvider().uri()
+            tables.add((uri.schema() or schema, uri.table()))
+
+        if not tables:
+            return set()
+
+        try:
+            import psycopg2
+            conn = psycopg2.connect(
+                host=db_config["host"],
+                port=int(db_config["port"]),
+                dbname=db_config["database"],
+                user=db_config["username"],
+                password=db_config["password"],
+                connect_timeout=3,
+            )
+            cur = conn.cursor()
+            conditions = " OR ".join(
+                "(table_schema = %s AND table_name = %s)"
+                for _ in tables
+            )
+            params = []
+            for s, t in tables:
+                params.extend([s, t])
+            cur.execute(
+                f"SELECT table_schema, table_name FROM information_schema.tables "
+                f"WHERE {conditions}",
+                params,
+            )
+            existing = {(row[0], row[1]) for row in cur.fetchall()}
+            cur.close()
+            conn.close()
+            return existing
+        except Exception as e:
+            logger.warning(f"  表存在性检查失败，将尝试修复所有图层: {e}")
+            return set(tables)
 
     @staticmethod
     def _fix_postgres_layer(layer, db_config, actual_schema):
@@ -135,23 +183,15 @@ class MapService:
         try:
             uri = layer.dataProvider().uri()
             uri.setConnection(
-                db_config["host"],
-                str(db_config["port"]),
-                db_config["database"],
-                db_config["username"],
-                db_config["password"],
+                db_config["host"], str(db_config["port"]),
+                db_config["database"], db_config["username"], db_config["password"],
             )
-            # Schema 替换
             uri_str = uri.uri()
             for old_schema in SCHEMA_REPLACEMENTS:
                 if f'table="{old_schema}".' in uri_str:
-                    uri_str = uri_str.replace(
-                        f'table="{old_schema}".',
-                        f'table="{actual_schema}".',
-                    )
+                    uri_str = uri_str.replace(f'table="{old_schema}".', f'table="{actual_schema}".')
                     uri = QgsDataSourceUri(uri_str)
                     break
-            # 表名映射
             uri_str = uri.uri()
             for old_name, new_name in TABLE_RENAMES.items():
                 full_old = f'table="{actual_schema}"."{old_name}"'
@@ -159,18 +199,14 @@ 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")
-
             if layer.isValid():
-                fc = layer.featureCount()
-                logger.info(f"图层 {layer.name()} 连接更新成功 ({fc} features)")
+                logger.debug(f"  图层 {layer.name()} 连接更新成功")
             else:
-                logger.error(f"图层 {layer.name()} 更新后仍无效")
+                logger.error(f"  图层 {layer.name()} 更新后仍无效")
         except Exception as e:
-            logger.error(f"更新图层 {layer.name()} 连接失败: {e}")
+            logger.error(f"  更新图层 {layer.name()} 连接失败: {e}")

app/services/qgis/qgis_daemon.py

@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+"""
+QGIS 常驻 Daemon 进程。
+
+由主进程在 FastAPI lifespan 中启动，通过 stdin/stdout JSON 行协议通信。
+
+启动时预加载所有模板到内存缓存，后续请求秒级响应。
+
+协议（每行一个完整 JSON）：
+  输入（stdin）: {"config": {...}, "models": [...], "result_file": "path/to/result.json"}
+  输出（result_file）: {"results": [...]}
+  控制:          {"action": "shutdown"}
+
+生命周期：
+  1. FastAPI 启动 → 启动 daemon 子进程
+  2. daemon 初始化 QGIS，加载磁盘缓存
+  3. 收到请求 → 处理模板 → 结果写入 result_file → stdout 输出完成标记
+  4. FastAPI 关闭 → 发送 shutdown → daemon 退出
+"""
+import json
+import os
+import sys
+import time
+
+# ============================================================
+# 环境初始化（必须在 QGIS import 之前）
+# ============================================================
+
+QGIS_ROOT = os.environ.get("QGIS_ROOT", "D:/QGIS")
+
+
+def _detect_qgis_app_dir():
+    for name in ("qgis-ltr", "qgis"):
+        d = os.path.join(QGIS_ROOT, "apps", name)
+        if os.path.isdir(d):
+            return d
+    raise RuntimeError(f"未找到 QGIS 应用目录: {QGIS_ROOT}\\apps\\qgis-ltr 或 qgis")
+
+
+def _setup_environment():
+    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":
+        import ctypes
+        qgis_app_dir = _detect_qgis_app_dir()
+        for dll_dir in [
+            os.path.join(qgis_app_dir, "bin"),
+            os.path.join(QGIS_ROOT, "apps", "Qt5", "bin"),
+            os.path.join(QGIS_ROOT, "apps", "gdal", "lib"),
+        ]:
+            if os.path.isdir(dll_dir):
+                os.add_dll_directory(dll_dir)
+
+        _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 [
+            os.path.join(qgis_app_dir, "bin"),
+            os.path.join(QGIS_ROOT, "apps", "Qt5", "bin"),
+            os.path.join(QGIS_ROOT, "apps", "gdal", "lib"),
+        ]:
+            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
+
+
+def _setup_python_path():
+    project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+    if project_root not in sys.path:
+        sys.path.insert(0, project_root)
+    qgis_app_dir = _detect_qgis_app_dir()
+    qgis_python = os.path.join(qgis_app_dir, "python")
+    if os.path.isdir(qgis_python) and qgis_python not in sys.path:
+        sys.path.insert(0, qgis_python)
+
+
+def _scan_templates() -> list[str]:
+    """扫描所有模板文件，返回绝对路径列表"""
+    project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+    template_base = os.path.join(project_root, "app", "data", "template")
+    templates = []
+    for event_type in ["rainfall", "earthquake"]:
+        tdir = os.path.join(template_base, event_type)
+        if not os.path.isdir(tdir):
+            continue
+        for tf in sorted(os.listdir(tdir)):
+            if tf.endswith(".qgz") and not tf.startswith("tmp"):
+                templates.append(os.path.join(tdir, tf).replace("\\", "/"))
+    return templates
+
+
+# ============================================================
+# 主逻辑
+# ============================================================
+
+def main():
+    _setup_environment()
+    _setup_python_path()
+
+    from qgis.core import QgsApplication, QgsProject
+
+    # 初始化 QGIS（只做一次）
+    qgis_app_dir = _detect_qgis_app_dir()
+    QgsApplication.setPrefixPath(qgis_app_dir, True)
+    qgs_app = QgsApplication([], False)
+    qgs_app.initQgis()
+
+    from app.services.qgis.map_service import MapService
+    from app.config.qgis_mappings import build_static_layers_config, get_gpkg_dir
+
+    print("[qgis_daemon] QGIS 已启动", file=sys.stderr, flush=True)
+
+    # ── 请求处理循环 ──
+    print("[qgis_daemon] 等待请求...", file=sys.stderr, flush=True)
+
+    for line in sys.stdin:
+        line = line.strip()
+        if not line:
+            continue
+
+        try:
+            request = json.loads(line)
+        except json.JSONDecodeError as e:
+            resp = json.dumps({"status": "error", "message": f"JSON 解析失败: {e}"})
+            sys.stdout.write(resp + "\n")
+            sys.stdout.flush()
+            continue
+
+        action = request.get("action", "map")
+
+        if action == "shutdown":
+            print("[qgis_daemon] 收到 shutdown，退出", file=sys.stderr, flush=True)
+            break
+
+        if action == "map":
+            config = request["config"]
+            models = request["models"]
+            result_file = request.get("result_file", "")
+            results = []
+
+            t_batch = time.time()
+            if "static_layers" not in config or not config["static_layers"].get("gpkg_dir"):
+                config["static_layers"] = build_static_layers_config(get_gpkg_dir())
+
+            service = MapService(config)
+
+            for i, model in enumerate(models):
+                t_model = time.time()
+                try:
+                    name = service.generate(model)
+                    results.append({"name": name, "output": model["outFile"]})
+                    elapsed = time.time() - t_model
+                    print(f"[qgis_daemon] [{i+1}/{len(models)}] {name} ({elapsed:.1f}s)",
+                          file=sys.stderr, flush=True)
+                except Exception as e:
+                    elapsed = time.time() - t_model
+                    results.append({"name": model.get("name", ""), "output": "", "error": str(e)})
+                    print(f"[qgis_daemon] [{i+1}/{len(models)}] 失败: {e}",
+                          file=sys.stderr, flush=True)
+
+            total = time.time() - t_batch
+            ok = sum(1 for r in results if "error" not in r)
+            print(f"[qgis_daemon] 完成: {ok}/{len(models)}, {total:.1f}s",
+                  file=sys.stderr, flush=True)
+
+            resp = {"results": results}
+
+            # 写入结果文件（可靠），然后 stdout 发完成信号
+            if result_file:
+                try:
+                    with open(result_file, "w", encoding="utf-8") as f:
+                        json.dump(resp, f, ensure_ascii=False)
+                except Exception as e:
+                    print(f"[qgis_daemon] 写结果文件失败: {e}", file=sys.stderr, flush=True)
+
+            sys.stdout.write("OK\n")
+            sys.stdout.flush()
+
+    # 清理
+    project = QgsProject.instance()
+    project.clear()
+    print("[qgis_daemon] 已退出", file=sys.stderr, flush=True)
+
+
+if __name__ == "__main__":
+    main()

app/services/qgis/qgis_env.py

@@ -5,13 +5,12 @@ QGIS 环境检测与子进程配置模块。
 所有 QGIS 操作通过 subprocess 调用 QGIS Python 3.12 执行。
 
 本模块提供：
-  - get_qgis_python_path(): 检测 QGIS Python 3.12 解释器路径
-  - build_qgis_command():   构建通过 .bat 启动 QGIS 子进程的命令
-  - is_qgis_available():    检查 QGIS 是否可用
-  - get_runner_script():    获取 qgis_runner.py 的路径
+  - get_qgis_python_path():       检测 QGIS Python 3.12 解释器路径
+  - build_qgis_subprocess_env():  构建子进程完整环境变量（替代 bat 包装器）
+  - is_qgis_available():          检查 QGIS 是否可用
+  - get_runner_script():          获取 qgis_runner.py 的路径
 """
 import os
-import tempfile
 from pathlib import Path
 
 from app.config.paths import get_logger
@@ -72,31 +71,96 @@ def get_qgis_python_path(qgis_root: str = None) -> str | None:
         return None
 
 
-def build_qgis_command(qgis_root: str = None) -> list[str]:
+def build_qgis_subprocess_env(qgis_root: str = None) -> dict:
     """
-    构建通过 .bat 包装器启动 QGIS 子进程的命令列表。
-    设置环境变量并启动 QGIS Python 3.12。
+    构建 QGIS Python 3.12 子进程的完整环境变量（替代 bat 包装器）。
+
+    策略：
+      1. 从 os.environ 继承，移除 venv 污染（PYTHONPATH/VIRTUAL_ENV 等）
+      2. 将 QGIS 的 bin 目录前置到 PATH（Qt5→qgis→gdal 顺序，与原 bat 一致）
+      3. 设置 PYTHONPATH / QGIS_PREFIX_PATH / QT_PLUGIN_PATH / GDAL_DATA / PROJ_DATA
+      4. 子进程 _setup_environment() 负责 os.add_dll_directory + ctypes 预加载
     """
     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()]
 
+    # 继承当前环境，清理 venv 污染
+    env = dict(os.environ)
+    venv_root = env.get("VIRTUAL_ENV", "").lower()
+
+    for key in (
+        "PYTHONPATH",
+        "PYTHONHOME",
+        "VIRTUAL_ENV",
+        "PYTHONDONTWRITEBYTECODE",
+    ):
+        env.pop(key, None)
+
+    if venv_root:
+        path_parts = env.get("PATH", "").split(";")
+        clean = []
+        for p in path_parts:
+            if not p or venv_root in p.lower():
+                continue
+            clean.append(p)
+        env["PATH"] = ";".join(clean)
+
+    # 检测 QGIS 安装路径
     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 解释器")
+    env["QGIS_ROOT"] = qgis_root
 
-    runner_script = get_runner_script()
-    if not os.path.isfile(runner_script):
-        raise RuntimeError(f"QGIS Runner 脚本不存在: {runner_script}")
+    if platform.system() != "Windows":
+        return env
 
-    bat_path = _generate_bat_wrapper(qgis_root, python_path, runner_script)
-    return ["cmd.exe", "/c", bat_path]
+    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"
+        )
+
+    # 前置 QGIS bin 目录到 PATH（Qt5 必须在最前面，QGIS 依赖它）
+    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")
+    qt5_plugins = os.path.join(qgis_root, "apps", "Qt5", "plugins")
+    qtplugins = os.path.join(qgis_app_dir, "qtplugins")
+    qgis_python_dir = os.path.join(qgis_app_dir, "python")
+
+    env["PATH"] = f"{qt5_bin};{qgis_bin};{gdal_lib};{env.get('PATH', '')}"
+
+    # QGIS 核心环境变量（与 bat 包装器保持一致）
+    env["PYTHONPATH"] = qgis_python_dir
+    env["QGIS_PREFIX_PATH"] = qgis_app_dir
+    env["QT_PLUGIN_PATH"] = f"{qtplugins};{qt5_plugins}"
+
+    # GDAL / PROJ 数据目录（避免系统旧版 proj.db 干扰）
+    gdal_data = os.path.join(qgis_root, "apps", "gdal", "share", "gdal")
+    if os.path.isdir(gdal_data):
+        env["GDAL_DATA"] = gdal_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")):
+            env["PROJ_DATA"] = pd
+            break
+
+    # UTF-8 / GDAL 编码辅助变量
+    env["PYTHONUTF8"] = "1"
+    env["GDAL_FILENAME_IS_UTF8"] = "YES"
+    env["VSI_CACHE"] = "TRUE"
+    env["VSI_CACHE_SIZE"] = "1000000"
+
+    logger.debug(
+        f"QGIS subprocess env built: root={qgis_root}, app={qgis_app_dir}, "
+        f"PATH prefixed with qgis_bin/qt5_bin/gdal_lib"
+    )
+    return env
 
 
 def is_qgis_available(qgis_root: str = None) -> bool:
@@ -109,60 +173,6 @@ 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

@@ -23,12 +23,23 @@ import sys
 import time
 
 # ============================================================
-# 1. 环境初始化（必须在任何 QGIS/Qt import 之前）
+# 环境初始化（必须在任何 QGIS/Qt import 之前）
 # ============================================================
 
 QGIS_ROOT = os.environ.get("QGIS_ROOT", "D:/QGIS")
 
 
+def _detect_qgis_app_dir():
+    """自动检测 QGIS 应用目录（qgis-ltr 或 qgis）"""
+    for name in ("qgis-ltr", "qgis"):
+        d = os.path.join(QGIS_ROOT, "apps", name)
+        if os.path.isdir(d):
+            return d
+    raise RuntimeError(
+        f"未找到 QGIS 应用目录: {QGIS_ROOT}\\apps\\qgis-ltr 或 qgis"
+    )
+
+
 def _setup_python_path():
     """将项目根目录和 QGIS Python 路径加入 sys.path"""
     project_root = os.path.dirname(
@@ -37,59 +48,44 @@ def _setup_python_path():
     if project_root not in sys.path:
         sys.path.insert(0, project_root)
 
-    qgis_python = os.path.join(QGIS_ROOT, "apps", "qgis-ltr", "python")
+    qgis_app_dir = _detect_qgis_app_dir()
+    qgis_python = os.path.join(qgis_app_dir, "python")
     if os.path.isdir(qgis_python) and qgis_python not in sys.path:
         sys.path.insert(0, qgis_python)
 
 
 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 搜索路径。
 
-    # ★ 必须在任何 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
-
-    os.environ["QGIS_PREFIX_PATH"] = qgis_app_dir
+    QGIS_PREFIX_PATH / QT_PLUGIN_PATH / GDAL_DATA / PROJ_DATA
+    已由主进程通过 build_qgis_subprocess_env() 设置，子进程不重复设。
+    这里注册 DLL 搜索目录并预加载核心 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 = [
+        qgis_app_dir = _detect_qgis_app_dir()
+        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"),
         ]
+        for dll_dir in dll_dirs:
+            if os.path.isdir(dll_dir):
+                os.add_dll_directory(dll_dir)
+
+        # 预加载核心 DLL —— 强制从 QGIS 目录加载，防止系统 PATH 中同名 DLL 干扰
         _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_dir in dll_dirs:
             for dll_name in _preload_dlls:
                 dll_path = os.path.join(dll_dir, dll_name)
                 if os.path.isfile(dll_path):
@@ -100,14 +96,14 @@ def _setup_environment():
 
 
 # ============================================================
-# 2. 主逻辑
+# 主逻辑
 # ============================================================
 
 def _init_qgis():
     """初始化 QgsApplication（只做一次）"""
     from qgis.core import QgsApplication
 
-    qgis_app_dir = os.path.join(QGIS_ROOT, "apps", "qgis-ltr")
+    qgis_app_dir = _detect_qgis_app_dir()
     QgsApplication.setPrefixPath(qgis_app_dir, True)
     qgs_app = QgsApplication([], False)
     qgs_app.initQgis()
@@ -117,7 +113,10 @@ def _init_qgis():
 def _process_single(service, model):
     """处理单个模板，返回结果 dict"""
     name = service.generate(model)
-    return {"name": name, "output": model["outFile"]}
+    result = {"name": name, "output": model["outFile"]}
+    # ★ 实时进度：每完成一张图就输出到 stdout
+    print(f"PROGRESS:{json.dumps(result, ensure_ascii=False)}", flush=True)
+    return result
 
 
 def main():

app/services/qgis/template_cache.py

@@ -1,120 +0,0 @@
-"""
-模板缓存引擎。解决 QgsProject 单例问题。
-
-流程：
-1. 首次请求：project.read() 加载模板（慢，仅一次）
-2. 加载后 project.write() 保存到临时文件
-3. 后续同模板请求：从临时文件恢复（快，连接复用）
-4. 手动恢复文本/过滤/缩放（毫秒级）
-"""
-import os
-import time
-import tempfile
-
-from qgis.core import QgsProject, QgsRectangle
-
-from app.config.paths import get_logger
-
-logger = get_logger("qgis.cache")
-
-
-class TemplateCache:
-    def __init__(self):
-        self._cache: dict[str, dict] = {}
-
-    def is_loaded(self, template_path: str) -> bool:
-        return template_path in self._cache
-
-    def load_template(self, template_path: str, layout_name: str = "A4") -> None:
-        """首次加载模板"""
-        start = time.time()
-        project = QgsProject.instance()
-
-        logger.info(f"首次加载: {os.path.basename(template_path)}")
-        project.read(template_path)
-        logger.info(f"project.read() 耗时: {time.time() - start:.1f}s")
-
-        # 保存到临时文件
-        tmp_file = tempfile.NamedTemporaryFile(
-            suffix=".qgz", delete=False,
-            dir=tempfile.gettempdir(),
-        )
-        tmp_path = tmp_file.name
-        tmp_file.close()
-
-        t_save = time.time()
-        project.write(tmp_path)
-        logger.info(f"项目保存耗时: {time.time() - t_save:.1f}s")
-
-        # 记录初始状态
-        texts = {}
-        extent = None
-        layout = project.layoutManager().layoutByName(layout_name)
-        if layout:
-            for item_id in ["mapTitle", "mapTime", "mapUint", "info"]:
-                item = layout.itemById(item_id)
-                if item:
-                    texts[item_id] = item.text()
-            map_item = layout.itemById("Map")
-            if map_item:
-                extent = QgsRectangle(map_item.extent())
-
-        self._cache[template_path] = {
-            "file": tmp_path,
-            "texts": texts,
-            "extent": extent,
-            "layout": layout_name,
-        }
-        logger.info(f"模板加载完成，总耗时: {time.time() - start:.1f}s")
-
-    def restore_template(self, template_path: str) -> tuple:
-        """从缓存恢复模板"""
-        cached = self._cache.get(template_path)
-        if not cached:
-            raise RuntimeError(f"模板未缓存: {template_path}")
-
-        start = time.time()
-        project = QgsProject.instance()
-
-        logger.info(f"恢复模板: {os.path.basename(template_path)}")
-        project.read(cached["file"])
-        logger.info(f"project.read() 耗时: {time.time() - start:.1f}s")
-
-        return project, cached["texts"], cached["extent"]
-
-    def reset_project_state(self, project: QgsProject, texts: dict, extent) -> None:
-        """重置项目到干净状态"""
-        start = time.time()
-
-        for layer in project.mapLayers().values():
-            if layer.subsetString():
-                layer.setSubsetString("")
-
-        # 获取 layout 名称（从缓存中）
-        layout_name = "A4"
-        for cached in self._cache.values():
-            layout_name = cached.get("layout", "A4")
-            break
-
-        layout = project.layoutManager().layoutByName(layout_name)
-        if layout:
-            for item_id, text in texts.items():
-                item = layout.itemById(item_id)
-                if item:
-                    item.setText(text)
-            if extent:
-                map_item = layout.itemById("Map")
-                if map_item:
-                    map_item.zoomToExtent(extent)
-
-        logger.info(f"状态重置耗时: {time.time() - start:.3f}s")
-
-    def cleanup(self) -> None:
-        """清理所有临时文件"""
-        for cached in self._cache.values():
-            try:
-                os.remove(cached["file"])
-            except OSError:
-                pass
-        self._cache.clear()
-        logger.info("已清理所有缓存")

app/services/qgis/template_modifier.py

@@ -57,14 +57,13 @@ class TemplateModifier:
     def modify(self, template_path: str) -> str:
         """修改模板文件，返回修改后的临时 .qgz 路径"""
         override = self.config.get("template_override")
-        has_override = override and override.get("enabled", False)
         has_static = bool(self._static_map)
 
-        if not has_override and not has_static:
+        if not override and not has_static:
             return template_path
 
-        orig = override["original"] if has_override else None
-        actual = override["actual"] if has_override else None
+        orig = override.get("original") if override else None
+        actual = override.get("actual") if override else None
 
         try:
             tmp = tempfile.NamedTemporaryFile(
@@ -75,7 +74,7 @@ class TemplateModifier:
             tmp.close()
 
             datasource_re = re.compile(r"(<datasource>)(.*?)(</datasource>)", re.DOTALL)
-            table_re = re.compile(r'table="(\w+)"\."(\w+)"')
+            table_re = re.compile(r'table=(?:"|&quot;)(\w+)(?:"|&quot;)\.(?:"|&quot;)(\w+)(?:"|&quot;)')
 
             with zipfile.ZipFile(template_path, "r") as zin, \
                  zipfile.ZipFile(tmp_path, "w", zipfile.ZIP_DEFLATED) as zout:

settings.toml

@@ -11,16 +11,20 @@ PREDICT_PROBABILITY_THRESHOLD = 50
 # 静态底图 GeoPackage 目录（相对于项目根目录）
 QGIS_GPKG_DIR = "app/data/gpkg"
 # 专题图输出子目录（相对于 FILE_STORE_DIR）
-QGIS_OUTPUT_DIR = "xian/qgis/map/:eventType/:disasterTime"
+QGIS_OUTPUT_DIR = "xian/qgis/map/:eventType/:inferenceId"
 # 专题图默认参数
-QGIS_DEFAULTS_MAP_LAYOUT = "A3"
+QGIS_DEFAULTS_MAP_LAYOUT = "A4"
 QGIS_DEFAULTS_ZOOM_RULE = "11"
-QGIS_DEFAULTS_ZOOM_VALUE = "50"
+QGIS_DEFAULTS_ZOOM_VALUE = "5"
 QGIS_DEFAULTS_MAP_UNIT = "制图单位：西安市应急管理局"
 # 专题图DPI
-QGIS_EXPORT_DPI = 300
-# 批量产图线程池
-QGIS_WORKER_THREADS = 4
+QGIS_EXPORT_DPI = 200
+# 并行子进程数（每进程独立 QGIS 实例）
+QGIS_PARALLEL_WORKERS = 4
+# 最大并发请求数（防止多人同时触发资源耗尽）
+QGIS_MAX_CONCURRENT = 2
+# 优先产出模板
+QGIS_PRIORITY_TEMPLATES = ["暴雨地质灾害风险区分布图", "暴雨滑坡潜在隐患点及人口分布图", "暴雨山洪潜在隐患点及人口分布图", "暴雨泥石流潜在隐患点及人口分布图", "暴雨内涝潜在隐患点及人口分布图", "暴雨避难场所分布图"]
 
 # 西安市中心经纬度
 XIAN_CENTER = [108.948024, 34.263161]
@@ -81,18 +85,8 @@ FILE_STORE_DIR = "G:/files"
 # ============================================================
 QGIS_ROOT = "D:/QGIS"
 # 专题图输出子目录
-QGIS_OUTPUT_DIR = "xian/qgis/map/:eventType/:disasterTime"
+QGIS_OUTPUT_DIR = "xian/qgis/map/:eventType/:inferenceId"
 QGIS_DEFAULTS_MAP_UNIT = "制图单位：西安市应急管理局"
-# 模板数据库覆盖：将模板中硬编码的连接替换为实际环境连接
-QGIS_TEMPLATE_OVERRIDE_ENABLED = true
-QGIS_TEMPLATE_OVERRIDE_ORIGINAL_HOST = "localhost"
-QGIS_TEMPLATE_OVERRIDE_ORIGINAL_PORT = 5432
-QGIS_TEMPLATE_OVERRIDE_ORIGINAL_DB_NAME = "yjzyk_xian"
-QGIS_TEMPLATE_OVERRIDE_ORIGINAL_SCHEMA = "base"
-QGIS_TEMPLATE_OVERRIDE_ACTUAL_HOST = "47.92.216.173"
-QGIS_TEMPLATE_OVERRIDE_ACTUAL_PORT = 7654
-QGIS_TEMPLATE_OVERRIDE_ACTUAL_DB_NAME = "xian_new"
-QGIS_TEMPLATE_OVERRIDE_ACTUAL_SCHEMA = "qgis"
 
 # ============================================================
 # 生产环境
@@ -131,13 +125,3 @@ FILE_STORE_DIR = "/data"
 # QGIS 配置
 # ============================================================
 QGIS_ROOT = "/home/QGIS"
-# 模板数据库覆盖：将模板中硬编码的连接替换为实际环境连接
-QGIS_TEMPLATE_OVERRIDE_ENABLED = true
-QGIS_TEMPLATE_OVERRIDE_ORIGINAL_HOST = "localhost"
-QGIS_TEMPLATE_OVERRIDE_ORIGINAL_PORT = 5432
-QGIS_TEMPLATE_OVERRIDE_ORIGINAL_DB_NAME = "yjzyk_xian"
-QGIS_TEMPLATE_OVERRIDE_ORIGINAL_SCHEMA = "base"
-QGIS_TEMPLATE_OVERRIDE_ACTUAL_HOST = "10.22.245.138"
-QGIS_TEMPLATE_OVERRIDE_ACTUAL_PORT = 54321
-QGIS_TEMPLATE_OVERRIDE_ACTUAL_DB_NAME = "xian_new"
-QGIS_TEMPLATE_OVERRIDE_ACTUAL_SCHEMA = "qgis"

tools/check_dynamic_layers.py

@@ -1,72 +0,0 @@
-"""
-批量检查所有 rainfall 模板的图层，看哪些动态图层需要显示、表是否存在
-"""
-import zipfile, re, os, psycopg2
-
-TEMPLATE_DIR = r"F:\project\xian\xian_algorithm_new\app\data\template\rainfall"
-
-# 收集所有非GPKG动态图层的表名
-all_dynamic = set()
-for fname in sorted(os.listdir(TEMPLATE_DIR)):
-    if not fname.endswith('.qgz') or fname.startswith('tmp'):
-        continue
-    path = os.path.join(TEMPLATE_DIR, fname)
-    z = zipfile.ZipFile(path)
-    qgs_name = [n for n in z.namelist() if n.endswith('.qgs')][0]
-    content = z.read(qgs_name).decode('utf-8')
-    
-    maplayer_re = re.compile(r'<maplayer[^>]*>(.*?)</maplayer>', re.DOTALL)
-    for m in maplayer_re.finditer(content):
-        block = m.group(1)
-        name_m = re.search(r'<layername>([^<]+)</layername>', block)
-        provider_m = re.search(r'<provider[^>]*>(\w+)</provider>', block)
-        ds_m = re.search(r'<datasource>(.*?)</datasource>', block, re.DOTALL)
-        
-        provider = provider_m.group(1) if provider_m else '?'
-        if provider != 'postgres':
-            continue
-        
-        ds = ds_m.group(1).strip() if ds_m else ''
-        table_m = re.search(r'table="(\w+)"\."(\w+)"', ds)
-        layer_name = name_m.group(1) if name_m else '?'
-        
-        if table_m:
-            schema = table_m.group(1)
-            table = table_m.group(2)
-            key = f"{schema}.{table}"
-            all_dynamic.add((layer_name, key))
-
-# 已知的GPKG静态层（会被替换为ogr）
-static_tables = {
-    'base.rivers', 'base.river', 'base.sx', 'base.sx_capital',
-    'base.sx_street', 'base.sx_xa_county', 'base.sx_xa_county_boundary',
-    'base.sx_zb_county_boundary', 'base.sx_zb_city', 'base.sx_zb_county',
-    'base.active_fault', 'base.traffic_expressway', 'base.traffic_provincial',
-    'base.traffic_railway', 'base.traffic_township', 'base.traffic_trunk_line',
-}
-
-# DB检查
-c = psycopg2.connect(host='47.92.216.173', port=7654, user='postgres', password='zhangsan', database='xian_new')
-c.autocommit = True
-cur = c.cursor()
-
-print(f"{'图层名':20s} {'原表':35s} {'qgis中存在':12s} {'行数':>8s}")
-print("-" * 80)
-
-for layer_name, table_key in sorted(all_dynamic):
-    if table_key in static_tables:
-        continue  # 会被GPKG替换，跳过
-    
-    schema, table = table_key.split('.', 1)
-    try:
-        cur.execute(f'SELECT count(*) FROM qgis.{table}')
-        count = cur.fetchone()[0]
-        exists = "YES"
-    except:
-        exists = "NO"
-        count = 0
-    
-    marker = " !!!" if exists == "NO" else ""
-    print(f"{layer_name:20s} {table_key:35s} {exists:12s} {count:>8,d}{marker}")
-
-c.close()

tools/check_layer_order.py

@@ -1,54 +0,0 @@
-"""检查模板中所有图层的渲染顺序（z-order）"""
-import zipfile, re, os
-
-TEMPLATE_DIR = r"F:\project\xian\xian_algorithm_new\app\data\template\rainfall"
-fname = "暴雨内涝潜在隐患点及人口分布图.qgz"
-path = os.path.join(TEMPLATE_DIR, fname)
-
-z = zipfile.ZipFile(path)
-qgs_name = [n for n in z.namelist() if n.endswith('.qgs')][0]
-content = z.read(qgs_name).decode('utf-8')
-
-# 提取所有 maplayer 块（保持顺序 — 这就是渲染顺序）
-maplayer_re = re.compile(r'<maplayer[^>]*>(.*?)</maplayer>', re.DOTALL)
-
-print(f"模板: {fname}")
-print(f"{'#':>3s} {'图层名':20s} {'类型':10s} {'可见':4s} {'表名'}")
-print("-" * 80)
-
-for i, m in enumerate(maplayer_re.finditer(content)):
-    block = m.group(1)
-    
-    # 图层名
-    name_m = re.search(r'<layername>([^<]+)</layername>', block)
-    name = name_m.group(1) if name_m else '?'
-    
-    # provider
-    provider_m = re.search(r'<provider[^>]*>(\w+)</provider>', block)
-    provider = provider_m.group(1) if provider_m else '?'
-    
-    # datasource table
-    ds_m = re.search(r'<datasource>(.*?)</datasource>', block, re.DOTALL)
-    ds = ds_m.group(1).strip() if ds_m else ''
-    table_m = re.search(r'table="(\w+)"\."(\w+)"', ds)
-    table = f"{table_m.group(1)}.{table_m.group(2)}" if table_m else '?'
-    
-    # 可见性
-    visible = 'Y' if 'visible="1"' in m.group(0) or 'visible="1"' in block else 'N'
-    
-    # 判断图层类型
-    is_static = table in [
-        'base.rivers', 'base.river', 'base.sx', 'base.sx_capital',
-        'base.sx_street', 'base.sx_xa_county', 'base.sx_xa_county_boundary',
-        'base.sx_zb_county_boundary', 'base.sx_zb_city', 'base.sx_zb_county',
-        'base.active_fault', 'base.traffic_expressway', 'base.traffic_provincial',
-        'base.traffic_railway', 'base.traffic_township', 'base.traffic_trunk_line',
-    ]
-    layer_type = "底图" if is_static else "动态"
-    
-    marker = " <<<" if not is_static and provider == 'postgres' else ""
-    print(f"{i+1:>3d} {name:20s} {provider:10s} {visible:4s} {table:35s} [{layer_type}]{marker}")
-
-print()
-print("提示: 图层按从上到下的顺序渲染（序号小的在底层，序号大的在顶层）")
-print("动态图层如果在底图下方，会被底图完全遮盖")

tools/check_layout.py

@@ -1,65 +0,0 @@
-"""检查布局 Map 项的图层配置"""
-import zipfile, re, os
-
-TEMPLATE_DIR = r"F:\project\xian\xian_algorithm_new\app\data\template\rainfall"
-fname = "暴雨内涝潜在隐患点及人口分布图.qgz"
-path = os.path.join(TEMPLATE_DIR, fname)
-
-z = zipfile.ZipFile(path)
-qgs_name = [n for n in z.namelist() if n.endswith('.qgs')][0]
-content = z.read(qgs_name).decode('utf-8')
-
-# 提取 Layout 元素（找 Map 项）
-# QGIS 有两种布局格式：<Composer> (QGIS 2.x style) 和 <Layout> (QGIS 3.x style)
-layout_re = re.compile(r'<(?:Composer|Layout)[^>]*name="([^"]*)"[^>]*>(.*?)</(?:Composer|Layout)>', re.DOTALL)
-
-for m in layout_re.finditer(content):
-    layout_name = m.group(1)
-    layout_content = m.group(2)
-    print(f"=== 布局: {layout_name} ===")
-    
-    # 检查 Map 项
-    map_item_re = re.compile(r'<ComposerMap[^>]*>(.*?)</ComposerMap>', re.DOTALL)
-    for mm in map_item_re.finditer(layout_content):
-        map_content = mm.group(1)
-        
-        # 检查是否有 lockedLayers
-        locked = re.findall(r'<lockedLayers>(.*?)</lockedLayers>', map_content, re.DOTALL)
-        if locked:
-            print(f"  lockedLayers: {locked[0][:500]}")
-        
-        # 检查 keepLayerSet
-        keep_set = re.findall(r'keepLayerSet="([^"]*)"', mm.group(0))
-        if keep_set:
-            print(f"  keepLayerSet: {keep_set[0]}")
-        
-        # 检查 followPreset
-        preset = re.findall(r'followPreset="([^"]*)"', mm.group(0))
-        if preset:
-            print(f"  followPreset: {preset[0]}")
-        
-        # 检查 followPresetName
-        preset_name = re.findall(r'followPresetName="([^"]*)"', mm.group(0))
-        if preset_name:
-            print(f"  followPresetName: {preset_name[0]}")
-        
-        # 检查 <layerSet>
-        layer_set = re.findall(r'<layerSet>(.*?)</layerSet>', map_content, re.DOTALL)
-        if layer_set:
-            print(f"  layerSet: {layer_set[0][:500]}")
-        
-        # 检查 <ComposerMapGrid>
-        grids = re.findall(r'<ComposerMapGrid[^>]*/>', map_content)
-        print(f"  grids: {len(grids)}")
-
-    # 也检查 <Layout> 格式 (QGIS 3.x)
-    map_item2_re = re.compile(r'<LayoutItem[^>]*type="[^"]*map[^"]*"[^>]*>(.*?)</LayoutItem>', re.DOTALL | re.IGNORECASE)
-    for mm in map_item2_re.finditer(layout_content):
-        map_content = mm.group(1)
-        print(f"  [LayoutItem Map] attributes: {mm.group(0)[:300]}")
-
-# 打印模板中所有的 map theme / visibility preset
-presets = re.findall(r'<(?:visibility-presets|map-theme-collection).*?</(?:visibility-presets|map-theme-collection)>', content, re.DOTALL)
-if presets:
-    print("\n=== 可见性预设 ===")
-    print(presets[0][:500])

tools/check_shelter.py

@@ -1,71 +0,0 @@
-"""检查暴雨避难场所分布图的所有动态图层是否有数据"""
-import zipfile, re, os, psycopg2
-
-TEMPLATE_PATH = r"F:\project\xian\xian_algorithm_new\app\data\template\rainfall\暴雨避难场所分布图.qgz"
-
-z = zipfile.ZipFile(TEMPLATE_PATH)
-qgs_name = [n for n in z.namelist() if n.endswith('.qgs')][0]
-content = z.read(qgs_name).decode('utf-8')
-
-maplayer_re = re.compile(r'<maplayer[^>]*>(.*?)</maplayer>', re.DOTALL)
-
-# 收集所有 PostgreSQL 动态图层
-layers = []
-for m in maplayer_re.finditer(content):
-    block = m.group(1)
-    name_m = re.search(r'<layername>([^<]+)</layername>', block)
-    provider_m = re.search(r'<provider[^>]*>(\w+)</provider>', block)
-    ds_m = re.search(r'<datasource>(.*?)</datasource>', block, re.DOTALL)
-    
-    provider = provider_m.group(1) if provider_m else '?'
-    if provider != 'postgres':
-        continue
-    
-    name = name_m.group(1) if name_m else '?'
-    ds = ds_m.group(1).strip() if ds_m else ''
-    table_m = re.search(r'table="(\w+)"\."(\w+)"', ds)
-    table_key = f"{table_m.group(1)}.{table_m.group(2)}" if table_m else '?'
-
-    layers.append((name, table_key))
-
-# 已知 GPKG 静态层
-static_tables = {
-    'base.rivers', 'base.river', 'base.sx', 'base.sx_capital',
-    'base.sx_street', 'base.sx_xa_county', 'base.sx_xa_county_boundary',
-    'base.sx_zb_county_boundary', 'base.sx_zb_city', 'base.sx_zb_county',
-    'base.active_fault', 'base.traffic_expressway', 'base.traffic_provincial',
-    'base.traffic_railway', 'base.traffic_township', 'base.traffic_trunk_line',
-}
-
-c = psycopg2.connect(host='47.92.216.173', port=7654, user='postgres', password='zhangsan', database='xian_new')
-c.autocommit = True
-cur = c.cursor()
-
-print(f"模板: 暴雨避难场所分布图")
-print(f"{'图层名':20s} {'原表':40s} {'qgis有数据':10s} {'行数':>6s}")
-print("-" * 85)
-
-for name, table_key in layers:
-    if table_key in static_tables:
-        continue
-    
-    schema, table = table_key.split('.', 1)
-    mapped_table = 'hazard_hydrops' if table == 'hazard_waterlogging' else table
-    
-    try:
-        cur.execute(f'SELECT count(*) FROM qgis.{mapped_table}')
-        count = cur.fetchone()[0]
-        exists = "YES" if count > 0 else "EMPTY"
-    except:
-        try:
-            cur.execute(f'SELECT count(*) FROM {schema}.{table}')
-            count = cur.fetchone()[0]
-            exists = f"在{schema}"
-        except:
-            exists = "NO"
-            count = 0
-    
-    marker = " <-- 无数据!" if count == 0 else ""
-    print(f"{name:20s} {table_key:40s} {exists:10s} {count:>6,d}{marker}")
-
-c.close()

tools/list_shelter_layers.py

@@ -1,16 +0,0 @@
-import zipfile, re
-
-z = zipfile.ZipFile(r"F:\project\xian\xian_algorithm_new\app\data\template\rainfall\暴雨避难场所分布图.qgz")
-c = z.read([n for n in z.namelist() if n.endswith('.qgs')][0]).decode()
-layers = re.findall(r'<maplayer[^>]*>(.*?)</maplayer>', c, re.DOTALL)
-
-for i, l in enumerate(layers):
-    nm = re.search(r'<layername>([^<]+)</layername>', l)
-    pv = re.search(r'<provider[^>]*>(\w+)</provider>', l)
-    ds = re.search(r'<datasource>(.*?)</datasource>', l, re.DOTALL)
-    name = nm.group(1) if nm else '?'
-    prov = pv.group(1) if pv else '?'
-    ds_text = ds.group(1).strip()[:120] if ds else ''
-    print(f"{i:2d}. [{prov:10s}] {name}")
-    if ds_text:
-        print(f"    ds: {ds_text}")

tools/test_single_map.py

@@ -1,295 +0,0 @@
-"""
-单张专题图测试脚本 - 独立版
-直接构建 model/config 并通过 subprocess 调用 qgis_runner.py
-"""
-import json, os, sys, tempfile, subprocess, re
-from datetime import datetime
-
-# ============================================================
-# 配置（和 settings.toml 一致）
-# ============================================================
-DB_CONFIG = {
-    "host": "47.92.216.173",
-    "port": 7654,
-    "username": "postgres",
-    "password": "zhangsan",
-    "database": "xian_new",
-}
-QGIS_ROOT = "D:/QGIS"
-XIAN_CENTER = [108.948024, 34.263161]
-GPKG_DIR = r"F:\project\xian\xian_algorithm_new\app\data\gpkg".replace("\\", "/")
-TEMPLATE_BASE = r"F:\project\xian\xian_algorithm_new\app\data\template"
-
-# ============================================================
-# 模拟推理结果（inference_id=50）
-# ============================================================
-# 从实际数据库查询得知:
-# event_type='rainfall', condition={'region_code':'610116','rainfall':'120'},
-# occurred_time='2025-09-16 20:00:00'
-inference = {
-    "id": 50,
-    "event_type": "rainfall",
-    "condition": {"region_code": "610116", "rainfall": "120"},
-    "occurred_time": datetime(2025, 9, 16, 20, 0, 0),
-}
-
-# ============================================================
-# 区域映射（从 settings.toml）
-# ============================================================
-AREA_MAP = {
-    "610102": "新城区", "610103": "碑林区", "610104": "莲湖区",
-    "610111": "灞桥区", "610112": "未央区", "610113": "雁塔区",
-    "610114": "阎良区", "610115": "临潼区", "610116": "长安区",
-    "610117": "高陵区", "610118": "鄠邑区", "610122": "蓝田县",
-    "610124": "周至县",
-}
-
-STATIC_LAYERS = {
-    "水库": ("base.rivers", "rivers.gpkg"),
-    "市州驻地": ("base.sx_capital", "sx_capital.gpkg"),
-    "河流": ("base.river", "river.gpkg"),
-    "active_fault": ("base.active_fault", "active_fault.gpkg"),
-    "陕西省": ("base.sx", "sx.gpkg"),
-    "乡镇驻地": ("base.sx_street", "sx_street.gpkg"),
-    "区县驻地": ("base.sx_xa_county", "sx_xa_county.gpkg"),
-    "县界": ("base.sx_xa_county_boundary", "sx_xa_county_boundary.gpkg"),
-    "周边区县": ("base.sx_zb_county_boundary", "sx_zb_county_boundary.gpkg"),
-    "周边市州": ("base.sx_zb_city", "sx_zb_city.gpkg"),
-    "周边县区": ("base.sx_zb_county", "sx_zb_county.gpkg"),
-    "traffic_expressway": ("base.traffic_expressway", "traffic_expressway.gpkg"),
-    "traffic_provincial": ("base.traffic_provincial", "traffic_provincial.gpkg"),
-    "traffic_railway": ("base.traffic_railway", "traffic_railway.gpkg"),
-    "traffic_township": ("base.traffic_township", "traffic_township.gpkg"),
-    "traffic_trunk_line": ("base.traffic_trunk_line", "traffic_trunk_line.gpkg"),
-}
-
-# ============================================================
-# 辅助函数（从 qgis_map_export.py 直接复制）
-# ============================================================
-def format_disaster_time(occurred_time):
-    if isinstance(occurred_time, datetime):
-        return occurred_time.strftime("%Y%m%d%H%M%S")
-    return str(occurred_time).replace("-", "").replace(":", "").replace(" ", "")[:14]
-
-def resolve_district(condition):
-    code = condition.get("region_code")
-    if code and str(code) in AREA_MAP:
-        return AREA_MAP[str(code)]
-    return ""
-
-def build_map_title(event_type, condition, template_name):
-    district = resolve_district(condition)
-    prefix = f"陕西西安{district}" if district else "陕西西安"
-    if event_type == "rainfall":
-        rainfall = condition.get("rainfall")
-        if rainfall is not None and rainfall != "":
-            return f"{prefix}{float(rainfall)}mm{template_name}"
-        return f"{prefix}{template_name}"
-    return f"{prefix}{template_name}"
-
-def build_info_text(event_type, condition, occurred_time):
-    """构建信息面板文本（不显示灾害类型标签）"""
-    lines = []
-    if isinstance(occurred_time, datetime):
-        time_str = f"{occurred_time.year}年{occurred_time.month:02d}月{occurred_time.day:02d}日{occurred_time.hour:02d}时{occurred_time.minute:02d}分"
-    elif occurred_time:
-        time_str = str(occurred_time)
-    else:
-        time_str = ""
-    lines.append(f"时间：{time_str}")
-    
-    if event_type == "rainfall":
-        rainfall = condition.get("rainfall")
-        if rainfall is not None and rainfall != "":
-            lines.append(f"累计降雨量：{float(rainfall)}mm")
-        duration = condition.get("duration")
-        if duration is not None and duration != "":
-            lines.append(f"已持续：{duration}")
-    elif event_type == "earthquake":
-        magnitude = condition.get("magnitude")
-        if magnitude is not None and magnitude != "":
-            lines.append(f"震级：{float(magnitude)}级")
-        lon = condition.get("epicenter_lon")
-        lat = condition.get("epicenter_lat")
-        if lon is not None and lat is not None:
-            lines.append(f"位置：经度{float(lon)}°, 纬度{float(lat)}°")
-    
-    return "\n".join(lines)
-
-def derive_model_params(inference, batch_folder, template_path):
-    event_type = inference["event_type"]
-    condition = inference["condition"]
-    occurred_time = inference["occurred_time"]
-    
-    template_name = os.path.splitext(os.path.basename(template_path))[0]
-    map_title = build_map_title(event_type, condition, template_name)
-    safe_name = re.sub(r'[\\/:*?"<>|]', '_', template_name)
-    out_file = os.path.join(batch_folder, f"{safe_name}.jpg").replace("\\", "/")
-    
-    if isinstance(occurred_time, datetime):
-        map_time = occurred_time.strftime("%Y-%m-%d %H:%M")
-    else:
-        map_time = str(occurred_time)
-    
-    info_text = build_info_text(event_type, condition, occurred_time)
-    
-    center_x, center_y = XIAN_CENTER
-    if event_type == "earthquake":
-        lon = condition.get("epicenter_lon", XIAN_CENTER[0])
-        lat = condition.get("epicenter_lat", XIAN_CENTER[1])
-        center_x, center_y = float(lon), float(lat)
-    
-    return {
-        "name": f"test_{inference['id']}_{safe_name}",
-        "path": template_path,
-        "outFile": out_file,
-        "mapLayout": "A3",
-        "mapTitle": map_title,
-        "mapTime": map_time,
-        "mapUnit": "西安市应急管理局",
-        "info": info_text,
-        "centerX": center_x,
-        "centerY": center_y,
-        "event": str(inference["id"]),
-        "queueId": str(inference["id"]),
-        "zoomRule": "11",
-        "zoomValue": "50",
-    }
-
-def build_qgis_config(batch_folder):
-    static_layers_config = {}
-    for name, (table, gpkg_file) in STATIC_LAYERS.items():
-        static_layers_config[name] = {"file": gpkg_file, "table": table}
-    
-    return {
-        "db": DB_CONFIG,
-        "qgis": {"exportDpi": 300},
-        "template_override": {
-            "enabled": True,
-            "original": {
-                "host": "localhost",
-                "port": 5432,
-                "dbname": "yjzyk_xian",
-                "schema": "base",
-            },
-            "actual": {
-                "host": DB_CONFIG["host"],
-                "port": DB_CONFIG["port"],
-                "dbname": DB_CONFIG["database"],
-                "schema": "qgis",
-                "username": DB_CONFIG["username"],
-                "password": DB_CONFIG["password"],
-            },
-        },
-        "static_layers": {
-            "enabled": True,
-            "gpkg_dir": GPKG_DIR,
-            "layers": static_layers_config,
-        },
-        "batch_folder": batch_folder,
-    }
-
-# ============================================================
-# 主逻辑
-# ============================================================
-event_type = inference["event_type"]
-disaster_time = format_disaster_time(inference["occurred_time"])
-batch_folder = os.path.join("G:/files", "xian/qgis/map", disaster_time).replace("\\", "/")
-os.makedirs(batch_folder, exist_ok=True)
-
-template_dir = os.path.join(TEMPLATE_BASE, event_type)
-template_path = os.path.join(template_dir, "暴雨内涝潜在隐患点及人口分布图.qgz").replace("\\", "/")
-
-model = derive_model_params(inference, batch_folder, template_path)
-config = build_qgis_config(batch_folder)
-
-print("=" * 60)
-print(f"模板: {template_path}")
-print(f"输出: {model['outFile']}")
-print(f"标题: {model['mapTitle']}")
-print(f"时间: {model['mapTime']}")
-print(f"单位: {model['mapUnit']}")
-print(f"info:\n{model['info']}")
-print(f"中心: ({model['centerX']}, {model['centerY']})")
-print(f"event: {model['event']}, queueId: {model['queueId']}")
-print("=" * 60)
-
-request_data = json.dumps(
-    {"config": config, "model": model},
-    ensure_ascii=False,
-)
-
-tmp_json = tempfile.NamedTemporaryFile(
-    suffix=".json", delete=False, mode="w", encoding="utf-8"
-)
-tmp_json.write(request_data)
-tmp_json.close()
-
-runner_script = os.path.join(
-    os.path.dirname(__file__),
-    r"F:\project\xian\xian_algorithm_new\app\services\qgis\qgis_runner.py"
-).replace("\\", "/")
-# Fix: use absolute path
-runner_script = r"F:\project\xian\xian_algorithm_new\app\services\qgis\qgis_runner.py"
-
-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_test.bat")
-
-qgis_app_dir = os.path.join(QGIS_ROOT, "apps", "qgis-ltr").replace("/", "\\")
-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("/", "\\")
-python_exe = os.path.join(python_dir, "python3.exe").replace("/", "\\")
-
-bat_content = f"""@echo off
-set "PYTHONHOME={python_dir}"
-set "PYTHONPATH={qgis_python_dir}"
-set "QGIS_PREFIX_PATH={qgis_app_dir}"
-set "QT_PLUGIN_PATH={qtplugins};{qt5_plugins}"
-set "GDAL_DATA={gdal_data}"
-set "PYTHONUTF8=1"
-set "GDAL_FILENAME_IS_UTF8=YES"
-set "VSI_CACHE=TRUE"
-set "VSI_CACHE_SIZE=1000000"
-set "PATH={qgis_bin};{qt5_bin};{gdal_lib};%PATH%"
-"{python_exe}" "{runner_script}" "{tmp_json.name}"
-"""
-
-with open(bat_path, "w", encoding="utf-8") as f:
-    f.write(bat_content)
-
-cmd = ["cmd.exe", "/c", bat_path]
-print(f"执行: {' '.join(cmd[:3])} ...")
-print()
-
-try:
-    result = subprocess.run(cmd, capture_output=True, timeout=300)
-    stderr_text = result.stderr.decode("utf-8", errors="replace")
-    stdout_text = result.stdout.decode("utf-8", errors="replace")
-
-    print("=== stderr ===")
-    for line in stderr_text.split("\n"):
-        print(f"  {line}")
-    print()
-    print("=== stdout ===")
-    print(stdout_text[:500] if stdout_text else "(empty)")
-
-    if result.returncode != 0:
-        print(f"\n!!! FAIL: exit={result.returncode}")
-    else:
-        print(f"\n=== SUCCESS ===")
-        if os.path.isfile(model["outFile"]):
-            print(f"文件: {model['outFile']} ({os.path.getsize(model['outFile'])/1024:.1f} KB)")
-        else:
-            print(f"文件不存在: {model['outFile']}")
-finally:
-    try:
-        os.remove(tmp_json.name)
-    except OSError:
-        pass