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