gis/xian_algorithm_new
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重置QGIS实现

Browse Source
This commit is contained in:
wzy-warehouse
2026-06-21 12:47:43 +08:00
parent 480e793ff8
commit fe3ccd005d
4 changed files with 160 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/api/qgis_map_export.py
+46 -30
View File
@@ -379,7 +379,9 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
import json
import subprocess
import tempfile
from app.services.qgis.qgis_env import build_qgis_command
from app.services.qgis.qgis_env import (
get_qgis_python_path, get_runner_script, build_qgis_subprocess_env,
)
try:
logger.info(f"[批量产图] 开始: {len(models)} 张专题图, batch={disaster_time}")
@@ -400,15 +402,19 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
try:
from config import settings
qgis_root = getattr(settings, "QGIS_ROOT", "D:/QGIS")
cmd = build_qgis_command(qgis_root)
cmd.append(tmp_json.name)
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]
logger.info(f"[批量产图] 启动 QGIS 子进程: {' '.join(cmd[:3])}...")
logger.info(f"[批量产图] 启动 QGIS 子进程: {python_path}...")
result = subprocess.run(
cmd,
capture_output=True,
timeout=300, # 5 分钟超时（批量处理多个模板）
timeout=600, # 10 分钟超时（15 张模板 × ~40s/张）
env=build_qgis_subprocess_env(qgis_root),
)
finally:
try:
@@ -416,7 +422,37 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
except OSError:
pass
if result.returncode != 0:
# 解析子进程输出 —— 优先检查 stdout 是否有有效结果
stdout_text = result.stdout.decode("utf-8", errors="replace").strip()
parsed_output = None
if stdout_text:
for line in reversed(stdout_text.split("\n")):
line = line.strip()
if line.startswith("{"):
try:
parsed_output = json.loads(line)
break
except json.JSONDecodeError:
continue
if parsed_output is not None:
batch_results = parsed_output.get("results", [])
success_count = sum(1 for r in batch_results if "error" not in r)
fail_count = len(batch_results) - success_count
logger.info(
f"[批量产图] 完成: 成功={success_count}, 失败={fail_count}"
)
for r in batch_results:
if "error" not in r:
logger.info(f" OK {r.get('output', 'N/A')}")
else:
logger.error(f" FAIL {r.get('error', 'unknown')}")
if success_count == 0 and fail_count > 0:
first_err = batch_results[0].get("error", "unknown")
raise RuntimeError(
f"QGIS 子进程所有模型均失败 ({fail_count}张): {first_err}"
)
elif result.returncode != 0:
stderr_text = result.stderr.decode("utf-8", errors="replace").strip()
logger.error(f"[批量产图] QGIS 子进程失败 (exit={result.returncode}):")
for line in stderr_text.split("\n"):
@@ -424,35 +460,15 @@ def _generate_batch_maps(models: list, config: dict, disaster_time: str) -> None
raise RuntimeError(
f"QGIS 子进程失败: {stderr_text[:300]}"
)
# 解析子进程输出
stdout_text = result.stdout.decode("utf-8", errors="replace").strip()
if stdout_text:
for line in reversed(stdout_text.split("\n")):
line = line.strip()
if line.startswith("{"):
output = json.loads(line)
batch_results = output.get("results", [])
success_count = sum(1 for r in batch_results if "error" not in r)
fail_count = len(batch_results) - success_count
logger.info(
f"[批量产图] 完成: 成功={success_count}, 失败={fail_count}"
)
for r in batch_results:
if "error" not in r:
logger.info(f" OK {r.get('output', 'N/A')}")
else:
logger.error(f" FAIL {r.get('error', 'unknown')}")
break
else:
logger.warning("[批量产图] 子进程输出中未找到 JSON 结果")
else:
logger.warning("[批量产图] 子进程无输出，但 exit code = 0")
logger.warning("[批量产图] 子进程无有效输出，exit code = 0")
except subprocess.TimeoutExpired:
logger.error(f"[批量产图] QGIS 子进程超时 (300s)")
logger.error(f"[批量产图] QGIS 子进程超时 (600s)")
raise
except Exception as e:
logger.error(f"[批量产图] 产图失败: {e}", exc_info=True)
raise
# ============================================================
app/script/run_qgis.bat
-54
View File
@@ -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/services/qgis/qgis_env.py
+85 -75
View File
@@ -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
+29 -33
View File
@@ -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()