from sqlalchemy import Column, Integer, String, Text, Index, JSON from sqlalchemy.dialects.postgresql import JSONB from geoalchemy2 import Geometry from app.database import Base import json class County(Base): """县级行政区划地理数据表""" __tablename__ = "counties" # 字段映射字典 FIELD_MAPPING = { '县名': 'name', '县代码': 'code', '市名': 'city_name', '市代码': 'city_code', '省名': 'province_name', '省代码': 'province_code' } # 反向映射字典 REVERSE_FIELD_MAPPING = {v: k for k, v in FIELD_MAPPING.items()} id = Column(Integer, primary_key=True, index=True) name = Column(String(100), nullable=False, comment="县名") code = Column(Integer, comment="县代码") city_name = Column(String(100), comment="市名") city_code = Column(Integer, comment="市代码") province_name = Column(String(100), comment="省名") province_code = Column(Integer, comment="省代码") # 使用PostGIS的几何类型来存储多边形数据 geometry = Column(Geometry('MULTIPOLYGON', srid=4326), nullable=False, comment="县级行政区划的几何数据") # 存储完整的GeoJSON数据 geojson = Column(JSON, nullable=False, comment="完整的GeoJSON数据") # 显式定义索引 __table_args__ = ( Index('idx_counties_name', 'name'), Index('idx_counties_code', 'code'), Index('idx_counties_geometry', 'geometry', postgresql_using='gist'), ) @classmethod def from_geojson_feature(cls, feature): """从GeoJSON Feature创建County实例 Args: feature: GeoJSON Feature对象 Returns: County: 新创建的County实例 """ properties = feature['properties'] # 转换中文字段名为英文 mapped_properties = { cls.FIELD_MAPPING.get(k, k): v for k, v in properties.items() } # 将GeoJSON geometry转换为WKT格式 geometry = feature['geometry'] wkt = f"SRID=4326;{json.dumps(geometry)}" # 创建实例 county = cls( **mapped_properties, geometry=wkt, geojson=feature ) return county def to_geojson_feature(self): """将County实例转换为GeoJSON Feature Returns: dict: GeoJSON Feature对象 """ properties = {} # 转换英文字段名为中文 for eng_field, cn_field in self.REVERSE_FIELD_MAPPING.items(): if hasattr(self, eng_field): properties[cn_field] = getattr(self, eng_field) return { 'type': 'Feature', 'properties': properties, 'geometry': self.geometry }