Files
xian_algorithm_new/app/services/qgis/qgis_env.py
T
2026-06-19 17:04:03 +08:00

266 lines
10 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
QGIS 环境初始化模块(跨平台:Windows / Linux)。
在 server.py lifespan 启动阶段调用 init_qgis_env()
完成共享库注入、环境变量设置、QgsApplication 初始化。
"""
import os
import sys
import platform
from pathlib import Path
from app.config.paths import get_logger
logger = get_logger("qgis.env")
_qgs_app = None
_initialized = False
_IS_WINDOWS = platform.system() == "Windows"
def init_qgis_env(qgis_root: str) -> None:
"""
初始化 QGIS 运行环境。整个应用生命周期只需调用一次。
Args:
qgis_root: QGIS 安装根目录
Windows: "D:/QGIS"
Linux: "/usr""/opt/QGIS"
"""
global _qgs_app, _initialized
if _initialized:
return
qgis_app_dir = _find_qgis_app_dir(qgis_root)
# 共享库搜索路径(平台相关)
# Windows: os.add_dll_directory() 显式注册 DLL 目录
# Linux: LD_LIBRARY_PATH 追加 .so 搜索目录
_add_library_paths(qgis_root, qgis_app_dir)
# 环境变量
_set_environment(qgis_root, qgis_app_dir)
# Python 模块路径
_add_python_paths(qgis_root, qgis_app_dir)
# 初始化 QgsApplication
_init_qgs_application(qgis_app_dir)
_initialized = True
logger.info(f"QGIS 环境初始化完成 ({platform.system()})")
def cleanup_qgis_env() -> None:
"""清理 QGIS 资源(应用退出时调用)"""
global _qgs_app, _initialized
if _qgs_app is not None:
_qgs_app.exitQgis()
_qgs_app = None
_initialized = False
logger.info("QGIS 资源已清理")
def is_qgis_ready() -> bool:
"""检查 QGIS 是否已初始化"""
return _initialized
# ─────────────────────────────────────────────────────────────
# 平台检测
# ─────────────────────────────────────────────────────────────
def _find_qgis_app_dir(root: str) -> str:
"""
自动检测 QGIS 应用目录。
WindowsOSGeo4W 安装): {root}/apps/qgis-ltr/
Linux(包管理器安装): {root}/share/qgis/Python 在 {root}/share/qgis/python
"""
if _IS_WINDOWS:
return _find_qgis_app_dir_windows(root)
else:
return _find_qgis_app_dir_linux(root)
def _find_qgis_app_dir_windows(root: str) -> str:
"""Windows: 在 {root}/apps/ 下查找 qgis* 目录"""
apps_dir = Path(root) / "apps"
if apps_dir.is_dir():
for name in sorted(apps_dir.iterdir()):
if name.name.startswith("qgis") and name.is_dir():
logger.info(f"检测到 QGIS 应用目录: {name}")
return str(name)
fallback = str(apps_dir / "qgis")
logger.warning(f"未检测到 qgis* 目录,使用默认路径: {fallback}")
return fallback
def _find_qgis_app_dir_linux(root: str) -> str:
"""
Linux: 返回 QGIS_PREFIX_PATH。
包管理器安装的标准路径:
Debian/Ubuntu: /usr qgis-core 在 /usr/lib/qgis/
RHEL/CentOS: /usr
QGIS.org 官方: /usr 或 /opt/QGIS
"""
candidates = [
Path(root) / "share" / "qgis", # /usr/share/qgis
Path(root) / "lib" / "qgis", # /usr/lib/qgis
Path(root) / "share" / "qgis-ltr", # qgis-ltr 变体
]
for c in candidates:
if c.is_dir():
logger.info(f"检测到 QGIS 应用目录: {c}")
return str(c)
# 回退:用 root 本身(/usr),由 QgsApplication 自行探测
logger.warning(f"未找到 QGIS 标准目录,使用 root: {root}")
return str(root)
# ─────────────────────────────────────────────────────────────
# 共享库注入
# ─────────────────────────────────────────────────────────────
def _add_library_paths(root: str, qgis_app: str) -> None:
"""
注入共享库搜索路径。
Windows: 调用 os.add_dll_directory() 显式注册每个 DLL 目录,
Python 3.8+ 不再自动搜索 PATH 中的 DLL。
Linux: 追加 LD_LIBRARY_PATH,让动态链接器 (ld-linux) 找到 .so。
"""
if _IS_WINDOWS:
_add_dll_directories_windows(root, qgis_app)
else:
_add_ld_library_path_linux(root, qgis_app)
def _add_dll_directories_windows(root: str, qgis_app: str) -> None:
"""Windows: 显式注册 DLL 搜索目录"""
if not hasattr(os, "add_dll_directory"):
return # Python < 3.8
dll_dirs = [
os.path.join(root, "apps", "Qt5", "bin"),
os.path.join(qgis_app, "bin"),
os.path.join(root, "bin"),
os.path.join(root, "apps", "gdal", "bin"),
os.path.join(root, "apps", "gdal", "lib"),
]
for d in dll_dirs:
if os.path.isdir(d):
os.add_dll_directory(d)
logger.debug(f"注册 DLL 目录: {d}")
def _add_ld_library_path_linux(root: str, qgis_app: str) -> None:
"""
Linux: 追加 LD_LIBRARY_PATH,使动态链接器找到 QGIS/GDAL 的 .so。
QGIS 包管理器安装时的标准 .so 路径:
/usr/lib/ — libqgis_core.so, libqgis_analysis.so
/usr/lib/qgis/ — 插件 .so
/usr/lib/qgis/plugins/ — provider .so
/usr/share/qgis/python/ — Python 模块
"""
ld_dirs = [
os.path.join(root, "lib"), # /usr/lib
os.path.join(root, "lib", "qgis"), # /usr/lib/qgis
os.path.join(root, "lib", "qgis", "plugins"), # /usr/lib/qgis/plugins
os.path.join(root, "share", "qgis", "lib"), # 某些安装方式
os.path.join(root, "apps", "gdal", "lib"), # OSGeo4W 风格
]
existing = os.environ.get("LD_LIBRARY_PATH", "")
new_dirs = [d for d in ld_dirs if os.path.isdir(d) and d not in existing]
if new_dirs:
joined = ":".join(new_dirs)
os.environ["LD_LIBRARY_PATH"] = f"{joined}:{existing}" if existing else joined
logger.info(f"LD_LIBRARY_PATH 追加: {new_dirs}")
# ─────────────────────────────────────────────────────────────
# 环境变量
# ─────────────────────────────────────────────────────────────
def _set_environment(root: str, qgis_app: str) -> None:
"""设置 QGIS 和 GDAL 相关环境变量(平台自适应)"""
env_vars = {
"QGIS_PREFIX_PATH": qgis_app,
"PYTHONUTF8": "1",
"GDAL_FILENAME_IS_UTF8": "YES",
"VSI_CACHE": "TRUE",
"VSI_CACHE_SIZE": "1000000",
}
if _IS_WINDOWS:
env_vars["QT_PLUGIN_PATH"] = (
f"{os.path.join(qgis_app, 'qtplugins')};"
f"{os.path.join(root, 'apps', 'Qt5', 'plugins')}"
)
env_vars["GDAL_DATA"] = os.path.join(root, "apps", "gdal", "share", "gdal")
else:
# Linux: GDAL 数据通常在系统路径 /usr/share/gdal/ 下
gdal_candidates = [
os.path.join(root, "share", "gdal"),
os.path.join(root, "share", "qgis", "resources"),
]
for p in gdal_candidates:
if os.path.isdir(p):
env_vars["GDAL_DATA"] = p
break
# QGIS 插件路径(Linux 标准)
qt_plugin = os.path.join(root, "lib", "qt", "plugins")
if os.path.isdir(qt_plugin):
env_vars["QT_PLUGIN_PATH"] = qt_plugin
for key, value in env_vars.items():
os.environ[key] = value
logger.debug(f"环境变量: {key}={value}")
# ─────────────────────────────────────────────────────────────
# Python 模块路径
# ─────────────────────────────────────────────────────────────
def _add_python_paths(root: str, qgis_app: str) -> None:
"""将 QGIS Python 模块路径加入 sys.path(平台自适应)"""
if _IS_WINDOWS:
python_paths = [
os.path.join(qgis_app, "python"),
os.path.join(root, "apps", "Python312", "Lib", "site-packages"),
]
else:
python_paths = [
os.path.join(qgis_app, "python"), # /usr/share/qgis/python
os.path.join(root, "lib", "python3", "dist-packages"), # Debian/Ubuntu
os.path.join(root, "lib", "python3.10", "site-packages"), # 通用
]
for p in python_paths:
if os.path.isdir(p) and p not in sys.path:
sys.path.insert(0, p)
logger.info(f"添加 Python 路径: {p}")
# ─────────────────────────────────────────────────────────────
# QgsApplication 初始化
# ─────────────────────────────────────────────────────────────
def _init_qgs_application(qgis_app: str) -> None:
"""创建并初始化 QgsApplication 实例"""
global _qgs_app
from qgis.core import QgsApplication, QgsSettings
QgsApplication.setPrefixPath(qgis_app, True)
_qgs_app = QgsApplication([], False) # False = 不启动 GUI
settings = QgsSettings()
settings.setValue("/qgis/render_decorations", False)
settings.setValue("/qgis/parallel_rendering", True)
settings.setValue("/qgis/use_spatial_index", True)
_qgs_app.initQgis()
logger.info("QgsApplication 初始化完成")