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.database 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. }