county.py 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990
  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. import json
  6. class County(Base):
  7. """县级行政区划地理数据表"""
  8. __tablename__ = "counties"
  9. # 字段映射字典
  10. FIELD_MAPPING = {
  11. '县名': 'name',
  12. '县代码': 'code',
  13. '市名': 'city_name',
  14. '市代码': 'city_code',
  15. '省名': 'province_name',
  16. '省代码': 'province_code'
  17. }
  18. # 反向映射字典
  19. REVERSE_FIELD_MAPPING = {v: k for k, v in FIELD_MAPPING.items()}
  20. id = Column(Integer, primary_key=True, index=True)
  21. name = Column(String(100), nullable=False, comment="县名")
  22. code = Column(Integer, comment="县代码")
  23. city_name = Column(String(100), comment="市名")
  24. city_code = Column(Integer, comment="市代码")
  25. province_name = Column(String(100), comment="省名")
  26. province_code = Column(Integer, comment="省代码")
  27. # 使用PostGIS的几何类型来存储多边形数据
  28. geometry = Column(Geometry('MULTIPOLYGON', srid=4326), nullable=False, comment="县级行政区划的几何数据")
  29. # 存储完整的GeoJSON数据
  30. geojson = Column(JSON, nullable=False, comment="完整的GeoJSON数据")
  31. # 显式定义索引
  32. __table_args__ = (
  33. Index('idx_counties_name', 'name'),
  34. Index('idx_counties_code', 'code'),
  35. #Index('idx_counties_geometry', 'geometry', postgresql_using='gist'),
  36. )
  37. @classmethod
  38. def from_geojson_feature(cls, feature):
  39. """从GeoJSON Feature创建County实例
  40. Args:
  41. feature: GeoJSON Feature对象
  42. Returns:
  43. County: 新创建的County实例
  44. """
  45. properties = feature['properties']
  46. # 转换中文字段名为英文
  47. mapped_properties = {
  48. cls.FIELD_MAPPING.get(k, k): v
  49. for k, v in properties.items()
  50. }
  51. # 将GeoJSON geometry转换为WKT格式
  52. geometry = feature['geometry']
  53. wkt = f"SRID=4326;{json.dumps(geometry)}"
  54. # 创建实例
  55. county = cls(
  56. **mapped_properties,
  57. geometry=wkt,
  58. geojson=feature
  59. )
  60. return county
  61. def to_geojson_feature(self):
  62. """将County实例转换为GeoJSON Feature
  63. Returns:
  64. dict: GeoJSON Feature对象
  65. """
  66. properties = {}
  67. # 转换英文字段名为中文
  68. for eng_field, cn_field in self.REVERSE_FIELD_MAPPING.items():
  69. if hasattr(self, eng_field):
  70. properties[cn_field] = getattr(self, eng_field)
  71. return {
  72. 'type': 'Feature',
  73. 'properties': properties,
  74. 'geometry': self.geometry
  75. }