重要提示: 当项目发生以下变化时,开发人员和AI助手必须同步更新本规则文件:
AcidMap/
├── app/ # 应用核心代码
│ ├── api/ # API路由层
│ │ ├── raster.py # 栅格数据接口
│ │ ├── vector.py # 矢量数据接口
│ │ └── cd_prediction.py # Cd预测模型接口
│ ├── services/ # 业务逻辑层
│ │ ├── raster_service.py
│ │ ├── vector_service.py
│ │ └── cd_prediction_service.py # Cd预测业务服务
│ ├── models/ # 数据模型层
│ │ ├── base.py # 基础模型
│ │ ├── orm_models.py # ORM模型
│ │ ├── raster.py # 栅格数据模型
│ │ ├── vector.py # 矢量数据模型
│ │ ├── county.py # 县域数据模型
│ │ └── farmland.py # 农田样点数据模型
│ ├── utils/ # 工具函数
│ │ ├── file_validators.py
│ │ └── cd_prediction_wrapper.py # Cd预测系统包装器
│ ├── config/ # 配置管理
│ │ ├── __init__.py
│ │ └── cd_prediction_config.py # Cd预测配置管理
│ ├── static/ # 静态文件
│ │ └── cd_predictions/ # Cd预测输出文件
│ │ ├── figures/ # 地图和直方图
│ │ ├── raster/ # 栅格文件
│ │ ├── reports/ # 报告文件
│ │ └── data/ # 数据文件
│ ├── logs/ # 日志文件
│ ├── scripts/ # 脚本文件
│ ├── database.py # 数据库配置
│ └── main.py # FastAPI应用入口
├── Cd_Prediction_Integrated_System/ # Cd预测系统
│ ├── models/ # 预测模型
│ │ ├── crop_cd_model/
│ │ └── effective_cd_model/
│ ├── analysis/ # 数据分析模块
│ ├── utils/ # 工具函数
│ ├── data/ # 数据文件
│ ├── output/ # 输出文件
│ ├── main.py # 主执行脚本
│ └── config.py # 配置文件
├── data/ # 数据文件
│ └── counties.geojson # 县域地理数据
├── migrations/ # 数据库迁移文件
├── scripts/ # 项目脚本
├── ssl/ # SSL证书文件
├── config.env # 环境配置
├── environment.yml # Conda环境依赖
├── soilgd.sql # 数据库备份文件
├── db_migrate.py # 数据库迁移脚本
├── reset_db.py # 数据库重置脚本
├── test_cd_integration.py # Cd集成测试脚本
├── INTEGRATION_GUIDE.md # 集成使用指南
├── CD_INTEGRATION_SUMMARY.md # 集成总结文档
└── main.py # 应用启动入口
必须使用JSDoc风格注释:
def process_raster_data(file_path: str, options: dict) -> dict:
"""
处理栅格数据文件
@param {str} file_path - 栅格文件路径
@param {dict} options - 处理选项配置
@returns {dict} 处理结果,包含状态和数据信息
@throws {ValueError} 当文件路径无效时抛出
@example
>>> result = process_raster_data('/path/to/file.tif', {'format': 'GeoTIFF'})
>>> print(result['status'])
"""
pass
# 标准库导入
import os
import logging
from typing import List, Dict, Optional
# 第三方库导入
from fastapi import FastAPI, HTTPException
from sqlalchemy import create_engine
from geoalchemy2 import Geometry
# 本地模块导入
from .models import Base
from .services import RasterService
/api/raster/*
/api/vector/*
/api/cd-prediction/*
{
"success": true,
"message": "操作成功",
"data": {...},
"error": null
}
high_res=True
参数启用600 DPI输出config.VISUALIZATION_CONFIG["dpi"] = 300
plt.rcParams['savefig.dpi'] = 300
high_res
参数可动态控制输出DPIcreate_raster_map(high_res=False)
- 默认300 DPI标准分辨率输出create_histogram(high_res=False)
- 默认300 DPI标准分辨率输出figure_size
和figsize
参数调整# 核心配置项
- SQLALCHEMY_DATABASE_URL: 数据库连接字符串
- SessionLocal: 会话工厂
- get_db(): 依赖注入函数
- execute_sql(): 原生SQL执行
第一步: 创建/修改 ORM 模型
app/models/
目录下创建或编辑模型文件确保在 app/models/__init__.py
中导入新模型
# 实际案例: 农田数据模型
# app/models/farmland.py
from sqlalchemy import Column, Integer, Float
from geoalchemy2 import Geometry
from app.database import Base # 统一的Base导入
class FarmlandData(Base):
"""
农田样点空间位置与索引数据模型
@param {int} farmland_id - 区域农业用地矢量点编号(主键)
@param {int} sample_id - 采样自增的ID(主键+自增)
@param {float} lon - 经度
@param {float} lan - 纬度
@param {float} type - 用地类型:旱地(0)、水田(1)、水浇地(2)
"""
__tablename__ = 'Farmland_data'
farmland_id = Column('Farmland_ID', Integer, primary_key=True)
sample_id = Column('Sample_ID', Integer, primary_key=True, autoincrement=True)
lon = Column('lon', Float, nullable=True)
lan = Column('lan', Float, nullable=True)
type = Column('Type', Float, nullable=True)
geom = Column(Geometry('POINT', srid=4326))
第二步: 使用迁移脚本生成迁移文件
# 使用项目自带的迁移脚本
python db_migrate.py create "添加新数据表"
# 或直接使用alembic命令
alembic revision --autogenerate -m "添加新数据表"
第三步: 检查生成的迁移文件
migrations/versions/
目录下新生成的迁移文件第四步: 执行数据库迁移
# 使用项目迁移脚本
python db_migrate.py upgrade
# 或直接使用alembic命令
alembic upgrade head
第五步: 验证迁移结果
# 查看当前数据库版本
python db_migrate.py current
# 查看迁移历史
python db_migrate.py history
当需要从现有数据库表生成ORM模型时:
# 反向生成所有表的模型
sqlacodegen postgresql://postgres:123456789Qq@localhost/soilgd > models.py
# 生成特定表的模型
sqlacodegen postgresql://postgres:123456789Qq@localhost/soilgd --tables table1,table2 > specific_models.py
db_migrate.py 脚本命令:
python db_migrate.py create "描述"
- 创建新迁移python db_migrate.py upgrade
- 升级到最新版本python db_migrate.py downgrade
- 降级到上一版本python db_migrate.py current
- 显示当前版本python db_migrate.py history
- 显示迁移历史python db_migrate.py stamp head
- 标记数据库版本其他数据库操作:
python reset_db.py
- 重置数据库psql -U postgres -d soilgd -f soilgd.sql
- 从备份文件恢复数据库scripts/import_counties.py
(GeoJSON导入) 和 scripts/import_farmland_data.py
(Excel导入)"""
数据导入脚本模板
@description: [描述导入的数据类型和来源]
"""
import os
import sys
import logging
# 添加项目根目录到Python路径
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.database import SessionLocal
from app.models.[model_file] import [ModelClass]
# 设置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class [DataType]Importer:
"""
[数据类型]导入器
"""
def __init__(self, file_path: str):
self.file_path = file_path
def read_data(self):
"""读取数据文件"""
pass
def validate_data(self, data):
"""验证数据格式和完整性"""
pass
def import_data(self, data):
"""将数据导入到数据库"""
db = SessionLocal()
try:
# 批量处理逻辑
# 事务提交
db.commit()
logger.info("数据导入完成")
except Exception as e:
db.rollback()
logger.error(f"数据导入失败: {str(e)}")
raise
finally:
db.close()
def run_import(self):
"""执行完整的导入流程"""
try:
data = self.read_data()
validated_data = self.validate_data(data)
self.import_data(validated_data)
except Exception as e:
logger.error(f"导入流程失败: {str(e)}")
raise
def main():
file_path = "path/to/data/file"
importer = [DataType]Importer(file_path)
importer.run_import()
if __name__ == "__main__":
main()
参考脚本: scripts/import_counties.py
主要特点:
from_geojson_feature
方法处理几何数据事务安全保证
# 核心导入逻辑示例
def import_counties_from_geojson(file_path: str, db: Session):
with open(file_path, 'r', encoding='utf-8') as f:
geojson_data = json.load(f)
for feature in geojson_data['features']:
try:
county = County.from_geojson_feature(feature)
db.add(county)
logger.info(f"成功导入: {county.name}")
except Exception as e:
logger.error(f"导入失败: {str(e)}")
continue
db.commit()
参考脚本: scripts/import_farmland_data.py
主要特点:
数据验证和清洗
# 核心导入逻辑示例
class FarmlandDataImporter:
def __init__(self, excel_path, sheet_name='Farmland'):
self.excel_path = excel_path
self.sheet_name = sheet_name
self.type_mapping = {'旱': 0.0, '水田': 1.0, '水浇地': 2.0}
def read_excel_data(self):
df = pd.read_excel(self.excel_path, sheet_name=self.sheet_name)
return df
def validate_data(self, df):
# 数据类型转换
df['Type_Numeric'] = df['Type'].map(self.type_mapping)
return df
def create_geometry(self, lon, lat):
return WKTElement(f'POINT({lon} {lat})', srid=4326)
def import_data(self, df):
# 批量导入逻辑
batch_size = 1000
for i in range(0, len(df), batch_size):
batch_objects = []
# ... 创建批次对象
db.add_all(batch_objects)
db.commit()
import_counties.py
)import_farmland_data.py
)app.database
中的同一个Base实例
from .base import Base
(会创建独立的Base)from app.database import Base
(使用统一的Base)config.env
: 数据库连接配置environment.yml
: Conda环境依赖# 开发环境
uvicorn app.main:app --reload
# 生产环境
uvicorn app.main:app --host 0.0.0.0 --port 8000
# Cd集成测试
python test_cd_integration.py
POST /api/cd-prediction/crop-cd/generate-and-get-map
POST /api/cd-prediction/effective-cd/generate-and-get-map
POST /api/cd-prediction/crop-cd/generate-and-get-histogram
POST /api/cd-prediction/effective-cd/generate-and-get-histogram
POST /api/cd-prediction/crop-cd/generate-map
POST /api/cd-prediction/effective-cd/generate-map
GET /api/cd-prediction/crop-cd/download-map
GET /api/cd-prediction/effective-cd/download-map
GET /api/cd-prediction/crop-cd/download-histogram
GET /api/cd-prediction/effective-cd/download-histogram
FileResponse
图片文件Content-Type: image/jpeg
Content-Disposition: attachment; filename="crop_cd_prediction_map.jpg"
[图片文件二进制数据]
{
"success": true,
"message": "作物Cd预测地图生成成功",
"data": {
"map_path": "string",
"histogram_path": "string",
"raster_path": "string",
"prediction_stats": {}
},
"error": null
}
/docs
)本规范文档旨在为AcidMap项目的开发提供统一的标准和指导。所有开发人员和AI助手在进行功能开发、代码维护和系统扩展时,都应严格遵循本规范,以确保代码质量、系统稳定性和开发效率。
重要提醒: 本规范文件应与项目功能同步更新。任何新功能集成、架构调整或开发流程变更都必须及时反映到相应的规范章节中。
遇到规范未覆盖的情况时,应优先考虑:
最后更新: 2025-06-14 (新增数据库导入脚本开发规范和农田数据模型)