refactor: 重构项目结构,将geo_tools重命名为app并更新相关引用
- 将主包名从geo_tools改为app - 更新所有模块中的引用路径 - 迁移并更新测试用例 - 添加项目规则文档 - 保持原有功能不变,仅进行结构调整
This commit is contained in:
183
app/utils/validators.py
Normal file
183
app/utils/validators.py
Normal file
@@ -0,0 +1,183 @@
|
||||
"""
|
||||
geo_tools.utils.validators
|
||||
~~~~~~~~~~~~~~~~~~~~~~~~~~
|
||||
数据验证工具:CRS 合法性、几何有效性、文件格式等。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import geopandas as gpd
|
||||
from shapely.geometry.base import BaseGeometry
|
||||
|
||||
|
||||
# ── CRS 校验 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def is_valid_crs(crs_input: str | int) -> bool:
|
||||
"""检查 CRS 是否可以被 pyproj 正常解析。
|
||||
|
||||
Parameters
|
||||
----------
|
||||
crs_input:
|
||||
EPSG 代码(整数或 ``"EPSG:4326"`` 字符串)或 proj 字符串。
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
"""
|
||||
try:
|
||||
from pyproj import CRS
|
||||
CRS.from_user_input(crs_input)
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def validate_crs(crs_input: str | int) -> str:
|
||||
"""校验并标准化 CRS,返回 EPSG 代码字符串。
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
如果 CRS 无法被 pyproj 解析。
|
||||
"""
|
||||
from pyproj import CRS
|
||||
try:
|
||||
crs_obj = CRS.from_user_input(crs_input)
|
||||
# 尝试返回简洁的 EPSG 字符串
|
||||
epsg = crs_obj.to_epsg()
|
||||
if epsg:
|
||||
return f"EPSG:{epsg}"
|
||||
return crs_obj.to_string()
|
||||
except Exception as exc:
|
||||
raise ValueError(f"无效的 CRS:{crs_input!r}。原因:{exc}") from exc
|
||||
|
||||
|
||||
# ── 几何校验 ──────────────────────────────────────────────────────────────────
|
||||
|
||||
def validate_geometry(gdf: "gpd.GeoDataFrame", *, raise_on_invalid: bool = False) -> dict[str, int]:
|
||||
"""检查 GeoDataFrame 中几何对象的有效性。
|
||||
|
||||
Parameters
|
||||
----------
|
||||
gdf:
|
||||
待检查的 GeoDataFrame。
|
||||
raise_on_invalid:
|
||||
若为 True,当存在无效几何时抛出 ``ValueError``。
|
||||
|
||||
Returns
|
||||
-------
|
||||
dict
|
||||
包含 ``total``、``valid``、``invalid``、``null`` 计数。
|
||||
"""
|
||||
null_count = gdf.geometry.isna().sum()
|
||||
non_null = gdf.geometry.dropna()
|
||||
invalid_mask = ~non_null.is_valid # type: ignore
|
||||
invalid_count = int(invalid_mask.sum())
|
||||
valid_count = len(non_null) - invalid_count
|
||||
|
||||
result = {
|
||||
"total": len(gdf),
|
||||
"valid": valid_count,
|
||||
"invalid": invalid_count,
|
||||
"null": int(null_count),
|
||||
}
|
||||
|
||||
if raise_on_invalid and (invalid_count > 0 or null_count > 0):
|
||||
raise ValueError(
|
||||
f"GeoDataFrame 存在 {invalid_count} 个无效几何、{null_count} 个空几何。"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
# ── 文件格式校验 ───────────────────────────────────────────────────────────────
|
||||
|
||||
#: 支持读取的矢量文件扩展名(fiona 驱动映射)
|
||||
SUPPORTED_VECTOR_EXTENSIONS: dict[str, str] = {
|
||||
".shp": "ESRI Shapefile",
|
||||
".geojson": "GeoJSON",
|
||||
".json": "GeoJSON",
|
||||
".gpkg": "GPKG",
|
||||
".gdb": "OpenFileGDB",
|
||||
".kml": "KML",
|
||||
".kmz": "KML",
|
||||
".csv": "CSV",
|
||||
".gml": "GML",
|
||||
".dxf": "DXF",
|
||||
".fgb": "FlatGeobuf",
|
||||
}
|
||||
|
||||
|
||||
def is_supported_vector_format(path: str | Path) -> bool:
|
||||
"""判断路径是否为已知的矢量格式。
|
||||
|
||||
Parameters
|
||||
----------
|
||||
path:
|
||||
输入路径,可以是字符串或 Path 对象。
|
||||
|
||||
Returns
|
||||
-------
|
||||
bool
|
||||
如果路径是支持的矢量格式,返回 True;否则返回 False。
|
||||
"""
|
||||
path = Path(path)
|
||||
suffix = path.suffix.lower()
|
||||
# .gdb 可能是目录(FileGDB)
|
||||
if path.is_dir() and suffix == ".gdb":
|
||||
return True
|
||||
return suffix in SUPPORTED_VECTOR_EXTENSIONS
|
||||
|
||||
|
||||
def validate_vector_path(path: str | Path) -> Path:
|
||||
"""校验矢量数据路径,返回 Path 对象。
|
||||
|
||||
Raises
|
||||
------
|
||||
FileNotFoundError
|
||||
文件或目录不存在。
|
||||
ValueError
|
||||
文件格式不受支持。
|
||||
"""
|
||||
path = Path(path)
|
||||
# GDB 是目录
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"路径不存在:{path}")
|
||||
if not is_supported_vector_format(path):
|
||||
raise ValueError(
|
||||
f"不支持的矢量格式:{path.suffix!r}。"
|
||||
f"支持的格式:{list(SUPPORTED_VECTOR_EXTENSIONS.keys())}"
|
||||
)
|
||||
return path
|
||||
|
||||
|
||||
def ensure_valid_geometry(geom: 'BaseGeometry', verbose: bool = False) -> 'BaseGeometry':
|
||||
"""确保几何对象有效,无效时尝试修复。
|
||||
|
||||
Parameters
|
||||
----------
|
||||
geom:
|
||||
输入几何。
|
||||
verbose:
|
||||
是否输出修复几何的警告信息。
|
||||
|
||||
Returns
|
||||
-------
|
||||
BaseGeometry
|
||||
有效的几何对象(可能是修复后的)。
|
||||
"""
|
||||
from app.core.geometry import fix_geometry
|
||||
from app.utils.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
if not geom.is_valid:
|
||||
fixed_geom = fix_geometry(geom)
|
||||
if fixed_geom is not None:
|
||||
if verbose:
|
||||
logger.warning("几何对象无效,已自动修复")
|
||||
return fixed_geom
|
||||
return geom
|
||||
Reference in New Issue
Block a user