province.py 2.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687
  1. from sqlalchemy import Column, Integer, String, Text, Index, JSON
  2. from sqlalchemy.dialects.postgresql import JSONB
  3. from geoalchemy2 import Geometry
  4. from app.models.orm_models import Base
  5. from shapely.geometry import shape, MultiPolygon, Polygon
  6. import json
  7. class Province(Base):
  8. """省级行政区划地理数据表"""
  9. __tablename__ = "provinces"
  10. # 字段映射字典
  11. FIELD_MAPPING = {
  12. '省名': 'name',
  13. '省代码': 'code'
  14. }
  15. # 反向映射字典
  16. REVERSE_FIELD_MAPPING = {v: k for k, v in FIELD_MAPPING.items()}
  17. id = Column(Integer, primary_key=True, index=True)
  18. name = Column(String(100), nullable=False, comment="省名")
  19. code = Column(Integer, nullable=False, unique=True, comment="省代码")
  20. # 使用PostGIS的几何类型来存储多边形数据
  21. geometry = Column(Geometry('MULTIPOLYGON', srid=4326), nullable=False, comment="省级行政区划的几何数据")
  22. # 存储完整的GeoJSON数据
  23. geojson = Column(JSON, nullable=False, comment="完整的GeoJSON数据")
  24. # 显式定义索引
  25. __table_args__ = (
  26. Index('idx_provinces_name', 'name'),
  27. Index('idx_provinces_code', 'code'),
  28. #Index('idx_provinces_geometry', 'geometry', postgresql_using='gist'),
  29. )
  30. @classmethod
  31. def from_geojson_feature(cls, feature):
  32. """从GeoJSON Feature创建Province实例
  33. Args:
  34. feature: GeoJSON Feature对象
  35. Returns:
  36. Province: 新创建的Province实例
  37. """
  38. properties = feature['properties']
  39. # 转换中文字段名为英文
  40. mapped_properties = {
  41. cls.FIELD_MAPPING.get(k, k): v
  42. for k, v in properties.items()
  43. }
  44. # 将GeoJSON geometry转换为EWKT格式(SRID=4326)
  45. geometry = feature['geometry']
  46. geom = shape(geometry)
  47. # 统一为 MultiPolygon 以匹配列类型
  48. if isinstance(geom, Polygon):
  49. geom = MultiPolygon([geom])
  50. wkt = f"SRID=4326;{geom.wkt}"
  51. # 创建实例
  52. province = cls(
  53. **mapped_properties,
  54. geometry=wkt,
  55. geojson=feature
  56. )
  57. return province
  58. def to_geojson_feature(self):
  59. """将Province实例转换为GeoJSON Feature
  60. Returns:
  61. dict: GeoJSON Feature对象
  62. """
  63. properties = {}
  64. # 转换英文字段名为中文
  65. for eng_field, cn_field in self.REVERSE_FIELD_MAPPING.items():
  66. if hasattr(self, eng_field):
  67. properties[cn_field] = getattr(self, eng_field)
  68. return {
  69. 'type': 'Feature',
  70. 'properties': properties,
  71. 'geometry': self.geometry
  72. }