QGIS的docker管理

This commit is contained in:
wzy-warehouse
2026-06-21 22:30:04 +08:00
parent 4e459fc203
commit 154f0a968e
11 changed files with 761 additions and 377 deletions
+2 -2
View File
@@ -27,5 +27,5 @@ __all__ = [
# 不在顶层导入 QGIS 依赖模块,避免主进程崩溃
# 使用方式:
# from app.services.qgis.map_service import MapService (仅在子进程中)
# from app.services.qgis.qgis_env import is_qgis_available (主进程安全)
# from app.services.qgis.map_service import MapService (仅在容器子进程中)
# from app.services.qgis.qgis_env import get_docker_container (主进程检查容器状态)
+103 -33
View File
@@ -26,62 +26,132 @@ import time
# 环境初始化(必须在 QGIS import 之前)
# ============================================================
QGIS_ROOT = os.environ.get("QGIS_ROOT", "D:/QGIS")
IS_WINDOWS = sys.platform == "win32"
def _detect_qgis_root():
"""自动检测 QGIS 安装根目录(跨平台)"""
root = os.environ.get("QGIS_ROOT")
if root and os.path.isdir(root):
return root
if IS_WINDOWS:
for candidate in ["D:/QGIS", "C:/OSGeo4W", "C:/QGIS"]:
if os.path.isdir(candidate):
return candidate
else:
for candidate in ["/usr", "/opt/QGIS", "/home/QGIS", "/usr/local"]:
if _is_valid_qgis_root_linux(candidate):
return candidate
return None
def _is_valid_qgis_root_linux(path: str) -> bool:
"""验证 Linux 路径是否为有效的 QGIS 安装根目录"""
if not os.path.isdir(path):
return False
return (
os.path.isdir(os.path.join(path, "share", "qgis"))
or any(os.path.isdir(os.path.join(path, "apps", n)) for n in ("qgis-ltr", "qgis"))
or os.path.isfile(os.path.join(path, "bin", "qgis"))
)
QGIS_ROOT = _detect_qgis_root()
def _detect_qgis_app_dir():
"""自动检测 QGIS 应用目录(跨平台)"""
if QGIS_ROOT is None:
raise RuntimeError("未检测到 QGIS 安装目录,请设置 QGIS_ROOT 环境变量")
if IS_WINDOWS:
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"
)
# Linux: 先检查独立安装器的 app 目录
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")
# Linux apt 安装:返回 prefix 目录
for prefix in ("/usr", "/opt/QGIS"):
if os.path.isdir(os.path.join(prefix, "share", "qgis")):
return prefix
raise RuntimeError(
f"未找到 QGIS 应用目录: {QGIS_ROOT} 下无 apps/qgis-ltr 或 share/qgis"
)
def _setup_environment():
"""设置 QGIS 运行所需的环境变量和 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":
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)
if not IS_WINDOWS:
return
_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
# ── Windows: DLL 预加载 ──
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():
"""将项目根目录和 QGIS Python 路径加入 sys.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)
if IS_WINDOWS:
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)
else:
for p in [
"/usr/lib/python3/dist-packages",
"/usr/lib/python3.10/dist-packages",
"/usr/lib/python3.11/dist-packages",
"/usr/lib/python3.12/dist-packages",
]:
if os.path.isdir(p) and p not in sys.path:
sys.path.insert(0, p)
def _scan_templates() -> list[str]:
+203 -169
View File
@@ -1,202 +1,236 @@
"""
QGIS 环境检测与子进程配置模块。
QGIS 环境配置模块 — Docker 模式
主进程运行在 Python 3.10,无法直接加载 QGIS 的 Python 3.12 C 扩展。
所有 QGIS 操作通过 subprocess 调用 QGIS Python 3.12 执行
所有 QGIS 操作通过 docker exec 在容器内执行,
主进程(Python 3.10)无需安装 QGIS 或处理 DLL 加载
容器内使用 qgis/qgis 官方镜像(QGIS 3.x + Python 3)。
本模块提供:
- get_qgis_python_path(): 检测 QGIS Python 3.12 解释器路径
- build_qgis_subprocess_env(): 构建子进程完整环境变量(替代 bat 包装器)
- is_qgis_available(): 检查 QGIS 是否可用
- get_runner_script(): 获取 qgis_runner.py 的路径
- get_docker_container(): 获取 Docker 容器名称
- get_docker_project_dir(): 获取容器内项目根目录
- build_docker_exec_cmd(): 构建 docker exec 命令
- build_docker_volume_mounts(): 构建 docker run 卷挂载参数
- get_runner_script(): 获取 qgis_runner.py 的绝对路径
- get_container_python_path(): 获取容器内 Python 解释器路径
"""
import os
import platform
from pathlib import Path
from app.config.paths import get_logger
logger = get_logger("qgis.env")
IS_WINDOWS = platform.system() == "Windows"
def get_qgis_python_path(qgis_root: str = None) -> str | None:
# ============================================================
# Docker 容器配置
# ============================================================
def get_docker_container() -> str:
"""获取 Docker 容器名称/ID"""
try:
from config import settings
container = getattr(settings, "QGIS_DOCKER_CONTAINER", "") or ""
if container.strip():
return container.strip()
except Exception:
pass
# 环境变量覆盖
env_container = os.environ.get("QGIS_DOCKER_CONTAINER", "")
if env_container.strip():
return env_container.strip()
return "qgis-server"
def get_docker_project_dir() -> str:
"""获取容器内项目根目录(代码挂载目标路径)"""
try:
from config import settings
project_dir = getattr(settings, "QGIS_DOCKER_PROJECT_DIR", "") or ""
if project_dir.strip():
return project_dir.strip()
except Exception:
pass
env_dir = os.environ.get("QGIS_DOCKER_PROJECT_DIR", "")
if env_dir.strip():
return env_dir.strip()
return "/app"
def get_container_python_path() -> str:
"""获取容器内 Python 解释器路径"""
try:
from config import settings
python_path = getattr(settings, "QGIS_DOCKER_PYTHON", "") or ""
if python_path.strip():
return python_path.strip()
except Exception:
pass
env_path = os.environ.get("QGIS_DOCKER_PYTHON", "")
if env_path.strip():
return env_path.strip()
# qgis/qgis 官方镜像默认路径
return "/usr/bin/python3"
# ============================================================
# docker exec 命令构建
# ============================================================
def build_docker_exec_cmd(python_path: str, runner_script: str, json_file: str) -> list:
"""
检测 QGIS 自带的 Python 3.12 解释器路径
构建 docker exec 命令
Windows: {QGIS_ROOT}/apps/Python312/python3.exe
Args:
python_path: 容器内 Python 路径(如 /usr/bin/python3
runner_script: 容器内 runner 脚本路径
json_file: 容器内 JSON 请求文件路径
Returns:
解释器绝对路径,不存在则返回 None
docker exec 命令列表
"""
if qgis_root is None:
qgis_root = _detect_qgis_root()
if qgis_root is None:
return None
container = get_docker_container()
project_dir = get_docker_project_dir()
import platform
if platform.system() == "Windows":
for name in ("python3.exe", "python.exe"):
candidate = os.path.join(qgis_root, "apps", "Python312", name)
if os.path.isfile(candidate):
logger.info(f"检测到 QGIS Python: {candidate}")
return candidate
logger.warning(f"未找到 QGIS Python 3.12")
return None
else:
import shutil
# Qt 平台插件(从配置读取,避免硬编码)
try:
from config import settings
qt_platform = getattr(settings, "QGIS_DOCKER_QT_PLATFORM", "offscreen")
except Exception:
qt_platform = "offscreen"
# 优先检查环境变量
env_python = os.environ.get("QGIS_PYTHON_PATH")
if env_python and os.path.isfile(env_python):
logger.info(f"QGIS_PYTHON_PATH 指定: {env_python}")
return env_python
# 搜索标准 Linux QGIS 安装路径
linux_paths = [
"/usr/bin/qgis", # Ubuntu/Debian apt 安装
"/usr/bin/qgis.bin",
"/opt/QGIS/apps/Python312/bin/python3", # 独立安装器
"/opt/QGIS/apps/Python3/bin/python3",
"/usr/libexec/qgis/python3", # Fedora/RHEL
]
for candidate in linux_paths:
if os.path.isfile(candidate):
logger.info(f"检测到 QGIS 可执行文件: {candidate}")
return candidate
# 回退到系统 Python(如果 qgis.core 可导入)
sys_python = shutil.which("python3") or shutil.which("python")
if sys_python:
logger.info(f"Linux 环境,使用系统 Python: {sys_python}")
return sys_python
return None
cmd = [
"docker", "exec",
"-w", project_dir,
"-e", f"QT_QPA_PLATFORM={qt_platform}",
container,
python_path, runner_script, json_file,
]
return cmd
def build_qgis_subprocess_env(qgis_root: str = None) -> dict:
# ============================================================
# docker run 卷挂载(首次启动容器用)
# ============================================================
def get_host_file_store() -> str:
"""获取主机端文件输出目录"""
try:
from config import settings
fs = getattr(settings, "FILE_STORE_DIR", "") or ""
if fs.strip():
return fs.strip().replace("\\", "/")
except Exception:
pass
return "G:/files"
def get_container_file_store() -> str:
"""获取容器内文件输出目录(Linux 路径)"""
# 挂载时主机 G:/files → 容器内 /files
host = get_host_file_store()
# 取最后一段目录名作为容器内路径
tail = host.rstrip("/").rsplit("/", 1)[-1]
return f"/{tail}"
def map_host_to_container(path: str) -> str:
"""将主机路径映射为容器内路径。
优先匹配 file_storeG:/files → /files),其次匹配项目根目录(F:/project/... → /app)。
"""
构建 QGIS Python 3.12 子进程的完整环境变量(替代 bat 包装器)。
normalized = path.replace("\\", "/")
策略:
1. 从 os.environ 继承,移除 venv 污染(PYTHONPATH/VIRTUAL_ENV 等)
2. 将 QGIS 的 bin 目录前置到 PATHQt5→qgis→gdal 顺序,与原 bat 一致)
3. 设置 PYTHONPATH / QGIS_PREFIX_PATH / QT_PLUGIN_PATH / GDAL_DATA / PROJ_DATA
4. 子进程 _setup_environment() 负责 os.add_dll_directory + ctypes 预加载
# 1. file_store 路径映射
host_fs = get_host_file_store()
container_fs = get_container_file_store()
if normalized.lower().startswith(host_fs.lower()):
return container_fs + normalized[len(host_fs):]
# 2. 项目根目录映射
project_root = str(Path(__file__).parent.parent.parent.parent).replace("\\", "/")
project_dir = get_docker_project_dir()
if normalized.lower().startswith(project_root.lower()):
return project_dir + normalized[len(project_root):]
return normalized
def map_container_to_host(path: str) -> str:
"""将容器内路径映射回主机路径"""
host = get_host_file_store()
container = get_container_file_store()
if path.startswith(container):
return host + path[len(container):]
return path
def build_docker_volume_mounts() -> list:
"""
import platform
构建 Docker 卷挂载参数列表(用于 docker run)。
# 继承当前环境,清理 venv 污染
env = dict(os.environ)
venv_root = env.get("VIRTUAL_ENV", "").lower()
挂载内容:
1. 项目代码目录 → 容器内 /app(只读)
2. 输出文件目录 → 容器内 /files(读写)
"""
project_root = str(Path(__file__).parent.parent.parent.parent)
host_file_store = get_host_file_store()
container_file_store = get_container_file_store()
project_dir = get_docker_project_dir()
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"
env["QGIS_ROOT"] = qgis_root
if platform.system() != "Windows":
return env
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 目录到 PATHQt5 必须在最前面,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
mounts = [
f"{project_root}:{project_dir}:ro",
f"{host_file_store}:{container_file_store}",
]
return mounts
def is_qgis_available(qgis_root: str = None) -> bool:
"""检查 QGIS 环境是否可用"""
return get_qgis_python_path(qgis_root) is not None
def build_docker_run_cmd(image: str = None) -> list:
"""
构建 docker run 启动命令(首次部署用)。
Returns:
docker run 命令列表
"""
if image is None:
try:
from config import settings
image = getattr(settings, "QGIS_DOCKER_IMAGE", "") or ""
if not image.strip():
image = "qgis/qgis:latest"
except Exception:
image = "qgis/qgis:latest"
# 容器保活命令(从配置读取)
try:
from config import settings
keep_alive = getattr(settings, "QGIS_DOCKER_KEEP_ALIVE", "sleep infinity")
except Exception:
keep_alive = "sleep infinity"
container = get_docker_container()
mounts = build_docker_volume_mounts()
cmd = [
"docker", "run", "-d",
"--name", container,
"--restart", "unless-stopped",
]
for m in mounts:
cmd.extend(["-v", m])
cmd.append(image)
# 保活命令
cmd.extend(keep_alive.split())
return cmd
# ============================================================
# Runner 脚本路径
# ============================================================
def get_runner_script() -> str:
"""获取 qgis_runner.py 的绝对路径"""
"""获取 qgis_runner.py 的绝对路径(主机端)"""
return str(Path(__file__).parent / "qgis_runner.py")
def _detect_qgis_root() -> str | None:
"""
自动检测 QGIS 安装根目录。
优先级:
1. 环境变量 QGIS_ROOT
2. Windows 默认路径 D:/QGIS
3. Linux 常见路径
"""
env_root = os.environ.get("QGIS_ROOT")
if env_root and os.path.isdir(env_root):
return env_root
import platform
if platform.system() == "Windows":
for candidate in ["D:/QGIS", "C:/OSGeo4W", "C:/QGIS"]:
if os.path.isdir(candidate):
logger.info(f"检测到 QGIS 根目录: {candidate}")
return candidate
else:
for candidate in ["/usr", "/opt/QGIS", "/home/QGIS"]:
if os.path.isdir(candidate):
logger.info(f"检测到 QGIS 根目录: {candidate}")
return candidate
logger.warning("未检测到 QGIS 安装目录")
return None
+32 -63
View File
@@ -1,9 +1,8 @@
#!/usr/bin/env python3
"""
QGIS 专题图生成子进程入口。
QGIS 专题图生成子进程入口 — Docker 容器内运行
由主进程 (Python 3.10) 通过 subprocess 调用,
运行在 QGIS 自带的 Python 3.12 环境中。
由主进程通过 docker exec 调用,运行在 QGIS Docker 容器内。
支持两种模式:
- 批量模式(推荐):单次启动 QgsApplication,顺序处理多个模板
@@ -23,102 +22,70 @@ import sys
import time
# ============================================================
# 环境初始化(必须在任何 QGIS/Qt import 之前
# 环境初始化(Docker 容器内,QGIS 通过 apt 安装在 /usr
# ============================================================
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"""
"""将项目根目录和 QGIS dist-packages 加入 sys.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)
# QGIS Python 包路径(从配置读取,避免硬编码)
try:
from config import settings
pythonpath = getattr(settings, "QGIS_DOCKER_PYTHONPATH", None) or []
except Exception:
pythonpath = []
for p in pythonpath:
if os.path.isdir(p) and p not in sys.path:
sys.path.insert(0, p)
def _setup_environment():
"""设置 QGIS 运行所需的环境变量和 DLL 搜索路径。
QGIS_PREFIX_PATH / QT_PLUGIN_PATH / GDAL_DATA / PROJ_DATA
已由主进程通过 build_qgis_subprocess_env() 设置,子进程不重复设。
这里注册 DLL 搜索目录并预加载核心 DLL 以确保正确加载顺序。
"""
"""设置 QGIS 运行所需的环境变量"""
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()
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:
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 _init_qgis():
"""初始化 QgsApplication只做一次"""
"""初始化 QgsApplication从配置读取 prefixPath,避免硬编码"""
from qgis.core import QgsApplication
qgis_app_dir = _detect_qgis_app_dir()
QgsApplication.setPrefixPath(qgis_app_dir, True)
try:
from config import settings
prefix_path = getattr(settings, "QGIS_DOCKER_PREFIX_PATH", "/usr")
except Exception:
prefix_path = "/usr"
QgsApplication.setPrefixPath(prefix_path, True)
qgs_app = QgsApplication([], False)
qgs_app.initQgis()
return qgs_app
def _process_single(service, model):
"""处理单个模板,返回结果 dict"""
"""处理单个模板,返回结果 dict。成功/失败均输出 PROGRESS 行。"""
name = service.generate(model)
result = {"name": name, "output": model["outFile"]}
# ★ 实时进度:每完成一张图就输出到 stdout
print(f"PROGRESS:{json.dumps(result, ensure_ascii=False)}", flush=True)
return result
def _emit_progress(result: dict):
"""输出 PROGRESS 行(成功或失败均调用)"""
print(f"PROGRESS:{json.dumps(result, ensure_ascii=False)}", flush=True)
def main():
t_start = time.time()
@@ -166,7 +133,9 @@ def main():
f"耗时 {elapsed:.1f}s — {error_msg}",
file=sys.stderr,
)
results.append({"name": model.get("name", ""), "output": "", "error": error_msg})
fail_result = {"name": model.get("name", ""), "output": "", "error": error_msg}
_emit_progress(fail_result)
results.append(fail_result)
# 输出结果
if len(models) == 1 and not request.get("models"):
+6 -2
View File
@@ -39,12 +39,13 @@ class TemplateModifier:
"""
将 GPKG 文件路径对应的 <provider ...>postgres</provider> 改为 ogr。
策略:按 <maplayer> 块处理,若块内 datasource 是文件路径(盘符开头),
策略:按 <maplayer> 块处理,若块内 datasource 是文件路径(盘符开头或 Linux 绝对路径),
则该块内的 provider 改为 ogr。避免跨层误改。
"""
maplayer_re = re.compile(r'(<maplayer[^>]*>.*?</maplayer>)', re.DOTALL)
provider_re = re.compile(r'(<provider[^>]*>)postgres(</provider>)')
file_ds_re = re.compile(r'<datasource>([A-Za-z]:/[^<]+)</datasource>')
# 匹配 Windows 盘符路径 (G:/...) 或 Linux 绝对路径 (/app/...)
file_ds_re = re.compile(r'<datasource>(?:[A-Za-z]:/|/)[^<]*</datasource>')
def _fix_layer(m):
layer_xml = m.group(1)
@@ -60,8 +61,11 @@ class TemplateModifier:
has_static = bool(self._static_map)
if not override and not has_static:
logger.info(f"模板无需修改(无 override 且 static_map 为空),直接返回: {template_path}")
return template_path
logger.info(f"模板修改: static_map有{len(self._static_map)}条, override={'' if override else ''}")
orig = override.get("original") if override else None
actual = override.get("actual") if override else None