尧图网页设计 
前阵子做乡村走访类实践时去了一趟鹿泉一带的北白砂村。这类“村村通探访”的场景并不只是拍几张照片、记录几条路况那么简单。真正麻烦的是把走访结果整理成结构化数据比如村庄之间是否通硬化路、路况等级如何、公共交通是否覆盖、手机信号是否稳定这些信息如果没有统一分类后续很难对比分析。本文把这些调研需求抽象成一个可复用的“村庄连通性调查记录系统”。项目以鹿泉村村通探访中采集的数据为例使用 FastAPI SQLAlchemy SQLite 搭建后端接口再用 Leaflet 做地图可视化。整个项目适合新手学习 Web 开发与 GIS 数据录入也可以作为毕业设计或数字乡村类项目的基础版本。1. 项目背景与业务分析1.1 村村通探访记录为什么要系统化“村村通”在很多场景下指的是通村道路、公共交通和通信基础设施的延伸。以往实地调研北白砂这类村庄通常会用纸质问卷或者 Excel 表格记录。这种方式有几个问题一是数据分散。每个走访小组记录的字段可能都不一样有人记录道路宽度有人只记录是否硬化后面合并数据非常痛苦。二是无法观察空间分布。Excel 只能告诉你某个村“有几条通村路”但很难回答“从北白砂去南白砂是否必须绕路”“哪个村信号薄弱点比较集中”。三是缺少时间维度。村庄的道路状态会变化今年走访和明年走访的结果如果都放在一张表里没有访问时间和人员信息就很难追踪变化。所以我们需要把一次“探访北白砂”这样的走访事件转化为一条条有坐标、有时间、有类型的结构化记录。1.2 系统核心对象定义本系统围绕调查任务设计了两个核心对象村庄点Village。用来存放行政村或自然村的基础信息和地图坐标是地图上最基本的标注单位。走访记录VisitRecord。每次到村里开展的连通性调查都作为一条记录。一条记录包含路况评价、公交便利度、信号强度、走访人、走访时间等字段。这样设计的好处是一个村庄可以被走访多次比如前半年路况良好后半年因为施工导致道路变差系统里可以保留多条记录便于做时序对比。本文最终会得到一个可以在浏览器中打开的地图页面。地图上有村庄标注点击后可以看到走访记录列表和详情。2. 需求分析与技术选型2.1 功能点梳理围绕鹿泉区村庄探访的实际场景第一期版本先实现以下功能村庄档案维护添加、编辑、查询村庄基本信息。走访记录录入记录道路类型、道路状况、公交可达性、网络信号、走访人、走访时间、备注等。地图点位展示所有村庄以经纬度在地图上展示点击后展示走访记录。基础统计按道路状况、信号等级统计走访数量方便判断问题集中区域。第一期不做复杂的权限管理默认走访人可以添加记录后续正式落地时需要加上用户登录和审核流程。2.2 技术栈选择选择技术栈时主要考虑新手友好度和部署成本。后端使用 FastAPI。它是 Python 生态中比较现代的异步 Web 框架自带接口文档适合快速开发数据接口。数据库使用 SQLite项目本地运行无需安装额外数据库服务适合调研演示正式生产可以平滑切换为 PostgreSQL。前端地图使用 Leaflet。它是一个轻量级开源 JavaScript 地图库相比 OpenLayers、Mapbox GL 等专业 GIS 库Leaflet 对新手更友好几行代码就能加载地图。具体依赖如下Python 3.10 及以上版本FastAPIUvicornASGI 服务器SQLAlchemy 2.xJinja2模板渲染Leaflet 1.9.x通过 CDN 加载下面开始搭建项目。3. 环境准备与项目结构3.1 创建项目目录在本地新建目录mkdir luanquan-village-survey cd luanquan-village-survey目录结构如下luanquan-village-survey/ ├── app/ │ ├── __init__.py │ ├── database.py │ ├── main.py │ ├── models.py │ └── schemas.py ├── templates/ │ └── index.html ├── requirements.txt └── init_db.py由于不使用复杂的前端工程化代码页面直接由后端渲染 HTMLLeaflet 通过 CDN 引入。3.2 安装依赖创建requirements.txtfastapi0.110.0 uvicorn[standard]0.29.0 sqlalchemy2.0.20 jinja23.1.0 pydantic2.5.0执行安装pip install -r requirements.txt各依赖版本以安装时实际解析为准。核心思路是先跑通再根据线上环境锁定版本。4. 数据库模型设计4.1 村庄表设计村庄表存储每个自然村或行政村的固定信息。在走访北白砂之前可以把北白砂预录入到村庄表后续所有记录都关联到这个村庄点。这里使用的字段如下id自增主键name村庄名称town所属乡镇或片区longitude经度latitude纬度description简介或备注经纬度建议统一使用 WGS84 坐标这是一个重要的约定。手机 GPS 采集的原始坐标通常是 WGS84但在国内部分地图软件中显示的是 GCJ-02 加密后的坐标。如果不做转换标记点可能会出现几百米甚至更远的偏移。本项目中现场采集记录统一以设备 GPS 原始坐标为准。表结构示例可以理解为以下 SQLCREATE TABLE villages ( id INTEGER PRIMARY KEY AUTOINCREMENT, name VARCHAR(50) NOT NULL, town VARCHAR(100), longitude FLOAT, latitude FLOAT, description TEXT );在实际代码中我们使用 SQLAlchemy ORM不需要手写建表 SQL。4.2 走访记录表设计走访记录表是系统的核心业务表记录每一次调查访问结果CREATE TABLE visit_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, village_id INTEGER NOT NULL, visit_date DATE NOT NULL, investigator VARCHAR(50), road_type VARCHAR(20), road_condition VARCHAR(20), bus_access BOOLEAN, network_signal VARCHAR(20), remark TEXT, FOREIGN KEY (village_id) REFERENCES villages (id) );各字段含义road_type道路类型示例值包括沥青路、水泥路、碎石路、土路。road_condition路况评价示例值包括良好、一般、较差、无法通行。bus_access是否通公交或客运班车。network_signal手机信号评价示例值包括强、中、弱、无信号。remark补充备注比如某路段正在施工。通过village_id外键一条走访记录对应一个村庄。一个村庄可以有多条走访记录这符合多次回访的业务需要。4.3 模型代码实现在app/models.py中编写模型from datetime import date from sqlalchemy import Boolean, Date, Float, ForeignKey, String, Text from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship class Base(DeclarativeBase): pass class Village(Base): __tablename__ villages id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(50), nullableFalse) town: Mapped[str] mapped_column(String(100), default) longitude: Mapped[float] mapped_column(Float, nullableFalse) latitude: Mapped[float] mapped_column(Float, nullableFalse) description: Mapped[str] mapped_column(Text, default) records: Mapped[list[VisitRecord]] relationship( back_populatesvillage, cascadeall, delete-orphan, ) class VisitRecord(Base): __tablename__ visit_records id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) village_id: Mapped[int] mapped_column(ForeignKey(villages.id), nullableFalse) visit_date: Mapped[date] mapped_column(Date, nullableFalse) investigator: Mapped[str] mapped_column(String(50), default) road_type: Mapped[str] mapped_column(String(20), default) road_condition: Mapped[str] mapped_column(String(20), default) bus_access: Mapped[bool] mapped_column(Boolean, defaultFalse) network_signal: Mapped[str] mapped_column(String(20), default) remark: Mapped[str] mapped_column(Text, default) village: Mapped[Village] relationship(back_populatesrecords)代码中字段和设计中完全对应。由于 SQLite 的数据类型有限布尔值在实际存储时会保存为整数 0 或 1读取时 ORM 会自动转换成True或False不影响使用。4.4 数据库连接封装在app/database.py中配置数据库连接from sqlalchemy import create_engine from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker DATABASE_URL sqlite:///./village_survey.db engine create_engine( DATABASE_URL, connect_args{check_same_thread: False}, ) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) from app.models import Base def init_db(): Base.metadata.create_all(bindengine) def get_db(): db SessionLocal() try: yield db finally: db.close()check_same_threadFalse是 SQLite 与 FastAPI 协作时的常见配置因为 FastAPI 可能在不同线程中访问同一个连接。4.5 初始化示例数据为了演示效果在init_db.py中插入一条北白砂的示例村庄记录以及一条走访记录。经纬度先用一个近似位置正式使用时一定要以现场实测坐标为准。from datetime import date from app.database import SessionLocal, init_db from app.models import Village, VisitRecord def seed_data(): db SessionLocal() village Village( name北白砂, town鹿泉区示例镇, longitude114.34, latitude38.10, description本次村村通探访的实地样本村演示数据请以实测为准。, ) db.add(village) db.flush() record VisitRecord( village_idvillage.id, visit_datedate(2025, 3, 20), investigator张三, road_type水泥路, road_condition良好, bus_accessTrue, network_signal中, remark村内主干道硬化完成部分巷道宽度较窄。, ) db.add(record) db.commit() db.close() if __name__ __main__: init_db() seed_data() print(数据库初始化完成)初始化后执行python init_db.py再查看当前目录会发现多出一个village_survey.db文件。5. FastAPI 后端接口开发5.1 定义请求与响应模型Pydantic 模型用于接口的请求参数校验和响应格式化。在app/schemas.py中定义from datetime import date from pydantic import BaseModel, ConfigDict class VillageOut(BaseModel): model_config ConfigDict(from_attributesTrue) id: int name: str town: str longitude: float latitude: float description: str class VisitRecordCreate(BaseModel): village_id: int visit_date: date investigator: str road_type: str road_condition: str bus_access: bool False network_signal: str remark: str class VisitRecordOut(VillageOut, VisitRecordCreate): id: int其中VisitRecordOut让它既包含村庄名称也包含走访记录字段。这里直接继承VillageOut是一种简洁写法字段名重复时需要注意实际生产项目推荐改为显式声明village: VillageOut。5.2 编写接口在app/main.py中实现接口和页面渲染from fastapi import Depends, FastAPI, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from sqlalchemy import select from sqlalchemy.orm import Session from app.database import get_db, init_db from app.models import VisitRecord, Village from app.schemas import VisitRecordCreate, VisitRecordOut, VillageOut app FastAPI(title村庄连通性调查记录系统) templates Jinja2Templates(directorytemplates) app.on_event(startup) def on_startup(): init_db() app.get(/, response_classHTMLResponse) def index(request: Request): return templates.TemplateResponse(request, index.html) app.get(/api/villages, response_modellist[VillageOut]) def get_villages(db: Session Depends(get_db)): return db.scalars(select(Village).order_by(Village.id)).all() app.get(/api/records, response_modellist[VisitRecordOut]) def get_records(db: Session Depends(get_db)): result db.execute( select(VisitRecord, Village) .join(Village, VisitRecord.village_id Village.id) .order_by(VisitRecord.visit_date.desc()) ) data [] for record, village in result.all(): data.append( VisitRecordOut( idrecord.id, village_idrecord.village_id, visit_daterecord.visit_date, investigatorrecord.investigator, road_typerecord.road_type, road_conditionrecord.road_condition, bus_accessrecord.bus_access, network_signalrecord.network_signal, remarkrecord.remark, namevillage.name, townvillage.town, longitudevillage.longitude, latitudevillage.latitude, descriptionvillage.description, ) ) return data app.post(/api/records, response_modelVisitRecordOut) def create_record( payload: VisitRecordCreate, db: Session Depends(get_db), ): record VisitRecord(**payload.model_dump()) db.add(record) db.commit() db.refresh(record) village db.get(Village, record.village_id) return VisitRecordOut( idrecord.id, village_idrecord.village_id, visit_daterecord.visit_date, investigatorrecord.investigator, road_typerecord.road_type, road_conditionrecord.road_condition, bus_accessrecord.bus_access, network_signalrecord.network_signal, remarkrecord.remark, namevillage.name, townvillage.town, longitudevillage.longitude, latitudevillage.latitude, descriptionvillage.description, )当前接口逻辑虽然能工作但create_record中缺少对village_id是否存在的判断。如果前端传入了不存在的村庄 ID系统仍然会创建一条外键无效的记录。虽然 SQLite 默认开启了外键约束但 SQLAlchemy 的会话不一定实时检查因此最佳实践是在插入前显式检查village db.get(Village, payload.village_id) if not village: raise HTTPException(status_code404, detail村庄不存在)5.3 启动项目在项目根目录执行uvicorn app.main:app --reload --port 8000启动后访问接口文档http://127.0.0.1:8000/docs页面http://127.0.0.1:8000FastAPI 会自动生成 Swagger 风格文档在浏览器中可以直接测试接口。6. Leaflet 地图可视化6.1 页面加载村庄标记下面编写templates/index.html使用 Leaflet 把村庄显示在地图上并把走访记录渲染到标记点的弹窗中。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title鹿泉村村通探访记录/title link relstylesheet hrefhttps://unpkg.com/leaflet1.9.4/dist/leaflet.css script srchttps://unpkg.com/leaflet1.9.4/dist/leaflet.js/script style body { margin: 0; font-family: Microsoft YaHei, sans-serif; } #map { height: 100vh; width: 100%; } /style /head body div idmap/div script var map L.map(map).setView([38.10, 114.34], 12); L.tileLayer(https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png, { maxZoom: 19, attribution: copy; OpenStreetMap contributors }).addTo(map); var villageLayer L.layerGroup().addTo(map); var currentBounds []; function loadData() { Promise.all([ fetch(/api/villages).then(res res.json()), fetch(/api/records).then(res res.json()) ]).then(function ([villages, records]) { villageLayer.clearLayers(); records.forEach(function (r) { var marker L.marker([r.latitude, r.longitude]) .addTo(villageLayer) .bindPopup(buildPopup(r)); currentBounds.push([r.latitude, r.longitude]); }); villages.forEach(function (v) { var exist records.some(r r.village_id v.id); if (!exist) { var marker L.marker([v.latitude, v.longitude]) .addTo(villageLayer) .bindPopup(b v.name /bbr暂无走访记录); currentBounds.push([v.latitude, v.longitude]); } }); if (currentBounds.length 0) { map.fitBounds(currentBounds); } }); } function buildPopup(r) { var signalMap {1: 强, 2: 中, 3: 弱, 4: 无信号}; var busText r.bus_access ? 是 : 否; return b r.name /bbr 走访日期 r.visit_date br 走访人 r.investigator br 道路类型 r.road_type br 路况 r.road_condition br 是否通公交 busText br 信号评价 signalMap[r.network_signal] || r.network_signal br 备注 (r.remark || 无); } loadData(); /script /body /html在示例数据的弹窗中能直观看到北白砂这条走访记录。实际开发时network_signal字段如果直接用“中”“强”等文字存储代码中的signalMap就不需要了。这里为了展示前后端字段对齐的思路保留汉字存储更简洁。6.2 新增走访记录的交互建议如果希望直接在页面上录入走访记录把networkSignal改成下拉框并由fetch提交即可。核心请求代码如下fetch(/api/records, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ village_id: 1, visit_date: 2025-03-20, investigator: document.getElementById(investigator).value, road_type: document.getElementById(roadType).value, road_condition: document.getElementById(roadCondition).value, bus_access: document.getElementById(busAccess).checked, network_signal: document.getElementById(networkSignal).value, remark: document.getElementById(remark).value }) }).then(function (res) { return res.json(); }).then(function (data) { loadData(); });这个 POST 接口的核心作用是“提交一条走访结果”。每次走访生成一条新数据而不是覆盖之前的记录。这种设计可以防止误操作导致历史数据丢失。7. 实地走访数据采集的关键注意点在真正的“探访北白砂”过程中硬件设备、坐标体系和数据习惯会直接影响采集数据质量。7.1 GPS 坐标与地图坐标偏移手机定位得到的经纬度通常是 WGS84 坐标。但国内大部分互联网地图为了提高安全性会给坐标加上偏移算法结果得到 GCJ-02 坐标。常见的现象是手机中某个地图 App 显示的位置与导入到 Leaflet 标记的位置不一致。解决思路有两种使用手机端原生定位 SDK 获取原始坐标。使用地图厂商提供的坐标转换接口转换后再入库。本项目为了简洁直接按 WGS84 处理。如果你的坐标来自高德或腾讯地图需要先转成 WGS84否则地图点会偏移。转发时建议使用专门工具核对位置。7.2 走访照片的处理走访记录中经常要附带现场照片。数据库设计时没有直接存储图片文件而是预留了备注字段。如果要加照片一个简单的做法是增加photo_url字段图片上传到 OSS 或服务器指定目录。考虑到本系统可能运行在没有外部云服务的服务器上建议先约定统一的文件命名规则访问日期_村庄名_序号.jpg例如20250320_北白砂_01.jpg后期使用照片时可以通过文件名中的日期和村庄名快速定位现场图片。7.3 现场记录习惯这类调查系统最容易出现的问题是字段取值不统一。比如有人在备注里写“路不错”有人写“路面好还行”数据统计时就很难归类。建议在走访前先约定字典表。以路况评价为例统一使用良好、一般、较差、无法通行。如果录入时发现选项不够不要去数据库里临时加一段文字而是扩充字典项保持数据规范。7.4 常见问题排查本地运行或者部署过程中常见报错和解决方式如下。问题现象常见原因解决思路运行 init_db.py 提示找不到 app 模块没有在项目根目录执行先cd到项目根目录再运行Leaflet 地图显示空白CDN 无法访问替换可访问的地图 CDN 地址接口返回 422 错误Pydantic 校验失败日期或字段类型不对查看 FastAPI 文档页中的 Schema 提示POST 记录后村庄名称是空返回结果没有关联查询在create_record中二次查询 Village 信息点位偏移明显坐标系混用检查坐标来源是否经过加密偏移8. 工程化改进与最佳实践8.1 数据库层面的改进SQLite 适合本地演示。当系统要支持多人同时录入、后台管理、统计报表时建议迁移到 PostgreSQL。迁移时需要改动的主要是database.py中的连接信息DATABASE_URL postgresql://user:passwordlocalhost:5432/village_survey如果已经使用 SQLAlchemy ORM模型代码几乎不需要改动只需要调整连接字符串再使用alembic生成迁移脚本。这也是在项目一开始就使用 ORM 而不是裸 SQL 的价值所在。8.2 接口层增加权限与校验当前系统的走访记录接口是公开的。生产环境中走访记录可能涉及内部调查资料和村民信息必须做访问控制。一个最小方案是增加简单的 Token 认证登录后获取 Token在请求头中携带 Token 才能新增记录。FastAPI 中可以基于OAuth2PasswordBearer实现相关的依赖注入方式如下from fastapi.security import OAuth2PasswordBearer oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) def get_current_user(token: str Depends(oauth2_scheme)): # 校验 token 并返回当前用户 pass在以后的系统版本中建议把村庄档案维护和走访记录修改分开管理。普通走访人员只允许新增记录管理员可以编辑或删除历史记录。8.3 数据备份与导出如果走访数据通过手机输入上报后台数据库需要每天做备份。SQLite 的备份比较简单在服务停止时直接复制数据库文件即可。生产环境建议写入定时任务cp village_survey.db village_survey_$(date %Y%m%d).db然后保留最近 30 份备份文件。导出功能也很实用。走访记录最终往往要交给其他部门处理提供一个 CSV 导出接口可以降低人工拷贝成本。8.4 前期验证与回滚策略在向正式系统写入数据之前务必先使用测试数据和测试库跑通流程。例如先录入北白砂的一条测试走访记录检查地图点是否显示正确再决定是否批量导入历史数据。如果发现需要回滚优先使用备份文件恢复而不是直接删除表中记录。对于需要修改多条历史数据的场景例如统一替换村庄名称执行更新前先使用SELECT查看会受影响的记录数量确认无误后再执行UPDATE。保留一份更新前的 SQL 还原脚本是更稳妥的做法。9. 后续扩展方向这个系统目前已经具备一个调研类 GIS 应用的基本形态。如果继续深化可以从下面几个方向扩展把数据库从 SQLite 改为 PostgreSQL PostGIS支持空间查询。PostGIS 是 PostgreSQL 的空间扩展能够做“查询某一公里范围内的村庄”“沿着道路线计算可达性”等空间分析而这些查询用普通表格字段很难实现。增加走访任务管理模块。每次探访前先创建任务任务下再绑定具体村庄和走访人。这样后续数据统计可以按任务分组减少录入错误。增加历史趋势对比。系统保留了多条记录的时间维度可以画出某个村庄从春到秋的道路与信号变化折线图为基层调研提供更直观的数据支撑。如果愿意继续深入可以先研究 Flask、Django 或 FastAPI 中更完善的前后端分离方案。当前模板渲染方式有利于快速理解流程但前后端分离的方案在多人协作和移动端适配方面可维护性更好。项目完整代码可以直接按照本文步骤搭建。建议先复制代码跑通一条记录再按自己的思路补充“附近公交站点”“道路宽度”“是否错车困难”等字段最终打磨成符合真实调研场景的系统。这样整个项目做出来之后不仅能记录一次鹿泉探访也能复用到其他乡镇的村庄调查工作中。 用FastAPI和Leaflet搭建村庄连通性调查记录系统
前阵子做乡村走访类实践时去了一趟鹿泉一带的北白砂村。这类“村村通探访”的场景并不只是拍几张照片、记录几条路况那么简单。真正麻烦的是把走访结果整理成结构化数据比如村庄之间是否通硬化路、路况等级如何、公共交通是否覆盖、手机信号是否稳定这些信息如果没有统一分类后续很难对比分析。本文把这些调研需求抽象成一个可复用的“村庄连通性调查记录系统”。项目以鹿泉村村通探访中采集的数据为例使用 FastAPI SQLAlchemy SQLite 搭建后端接口再用 Leaflet 做地图可视化。整个项目适合新手学习 Web 开发与 GIS 数据录入也可以作为毕业设计或数字乡村类项目的基础版本。1. 项目背景与业务分析1.1 村村通探访记录为什么要系统化“村村通”在很多场景下指的是通村道路、公共交通和通信基础设施的延伸。以往实地调研北白砂这类村庄通常会用纸质问卷或者 Excel 表格记录。这种方式有几个问题一是数据分散。每个走访小组记录的字段可能都不一样有人记录道路宽度有人只记录是否硬化后面合并数据非常痛苦。二是无法观察空间分布。Excel 只能告诉你某个村“有几条通村路”但很难回答“从北白砂去南白砂是否必须绕路”“哪个村信号薄弱点比较集中”。三是缺少时间维度。村庄的道路状态会变化今年走访和明年走访的结果如果都放在一张表里没有访问时间和人员信息就很难追踪变化。所以我们需要把一次“探访北白砂”这样的走访事件转化为一条条有坐标、有时间、有类型的结构化记录。1.2 系统核心对象定义本系统围绕调查任务设计了两个核心对象村庄点Village。用来存放行政村或自然村的基础信息和地图坐标是地图上最基本的标注单位。走访记录VisitRecord。每次到村里开展的连通性调查都作为一条记录。一条记录包含路况评价、公交便利度、信号强度、走访人、走访时间等字段。这样设计的好处是一个村庄可以被走访多次比如前半年路况良好后半年因为施工导致道路变差系统里可以保留多条记录便于做时序对比。本文最终会得到一个可以在浏览器中打开的地图页面。地图上有村庄标注点击后可以看到走访记录列表和详情。2. 需求分析与技术选型2.1 功能点梳理围绕鹿泉区村庄探访的实际场景第一期版本先实现以下功能村庄档案维护添加、编辑、查询村庄基本信息。走访记录录入记录道路类型、道路状况、公交可达性、网络信号、走访人、走访时间、备注等。地图点位展示所有村庄以经纬度在地图上展示点击后展示走访记录。基础统计按道路状况、信号等级统计走访数量方便判断问题集中区域。第一期不做复杂的权限管理默认走访人可以添加记录后续正式落地时需要加上用户登录和审核流程。2.2 技术栈选择选择技术栈时主要考虑新手友好度和部署成本。后端使用 FastAPI。它是 Python 生态中比较现代的异步 Web 框架自带接口文档适合快速开发数据接口。数据库使用 SQLite项目本地运行无需安装额外数据库服务适合调研演示正式生产可以平滑切换为 PostgreSQL。前端地图使用 Leaflet。它是一个轻量级开源 JavaScript 地图库相比 OpenLayers、Mapbox GL 等专业 GIS 库Leaflet 对新手更友好几行代码就能加载地图。具体依赖如下Python 3.10 及以上版本FastAPIUvicornASGI 服务器SQLAlchemy 2.xJinja2模板渲染Leaflet 1.9.x通过 CDN 加载下面开始搭建项目。3. 环境准备与项目结构3.1 创建项目目录在本地新建目录mkdir luanquan-village-survey cd luanquan-village-survey目录结构如下luanquan-village-survey/ ├── app/ │ ├── __init__.py │ ├── database.py │ ├── main.py │ ├── models.py │ └── schemas.py ├── templates/ │ └── index.html ├── requirements.txt └── init_db.py由于不使用复杂的前端工程化代码页面直接由后端渲染 HTMLLeaflet 通过 CDN 引入。3.2 安装依赖创建requirements.txtfastapi0.110.0 uvicorn[standard]0.29.0 sqlalchemy2.0.20 jinja23.1.0 pydantic2.5.0执行安装pip install -r requirements.txt各依赖版本以安装时实际解析为准。核心思路是先跑通再根据线上环境锁定版本。4. 数据库模型设计4.1 村庄表设计村庄表存储每个自然村或行政村的固定信息。在走访北白砂之前可以把北白砂预录入到村庄表后续所有记录都关联到这个村庄点。这里使用的字段如下id自增主键name村庄名称town所属乡镇或片区longitude经度latitude纬度description简介或备注经纬度建议统一使用 WGS84 坐标这是一个重要的约定。手机 GPS 采集的原始坐标通常是 WGS84但在国内部分地图软件中显示的是 GCJ-02 加密后的坐标。如果不做转换标记点可能会出现几百米甚至更远的偏移。本项目中现场采集记录统一以设备 GPS 原始坐标为准。表结构示例可以理解为以下 SQLCREATE TABLE villages ( id INTEGER PRIMARY KEY AUTOINCREMENT, name VARCHAR(50) NOT NULL, town VARCHAR(100), longitude FLOAT, latitude FLOAT, description TEXT );在实际代码中我们使用 SQLAlchemy ORM不需要手写建表 SQL。4.2 走访记录表设计走访记录表是系统的核心业务表记录每一次调查访问结果CREATE TABLE visit_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, village_id INTEGER NOT NULL, visit_date DATE NOT NULL, investigator VARCHAR(50), road_type VARCHAR(20), road_condition VARCHAR(20), bus_access BOOLEAN, network_signal VARCHAR(20), remark TEXT, FOREIGN KEY (village_id) REFERENCES villages (id) );各字段含义road_type道路类型示例值包括沥青路、水泥路、碎石路、土路。road_condition路况评价示例值包括良好、一般、较差、无法通行。bus_access是否通公交或客运班车。network_signal手机信号评价示例值包括强、中、弱、无信号。remark补充备注比如某路段正在施工。通过village_id外键一条走访记录对应一个村庄。一个村庄可以有多条走访记录这符合多次回访的业务需要。4.3 模型代码实现在app/models.py中编写模型from datetime import date from sqlalchemy import Boolean, Date, Float, ForeignKey, String, Text from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship class Base(DeclarativeBase): pass class Village(Base): __tablename__ villages id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) name: Mapped[str] mapped_column(String(50), nullableFalse) town: Mapped[str] mapped_column(String(100), default) longitude: Mapped[float] mapped_column(Float, nullableFalse) latitude: Mapped[float] mapped_column(Float, nullableFalse) description: Mapped[str] mapped_column(Text, default) records: Mapped[list[VisitRecord]] relationship( back_populatesvillage, cascadeall, delete-orphan, ) class VisitRecord(Base): __tablename__ visit_records id: Mapped[int] mapped_column(primary_keyTrue, autoincrementTrue) village_id: Mapped[int] mapped_column(ForeignKey(villages.id), nullableFalse) visit_date: Mapped[date] mapped_column(Date, nullableFalse) investigator: Mapped[str] mapped_column(String(50), default) road_type: Mapped[str] mapped_column(String(20), default) road_condition: Mapped[str] mapped_column(String(20), default) bus_access: Mapped[bool] mapped_column(Boolean, defaultFalse) network_signal: Mapped[str] mapped_column(String(20), default) remark: Mapped[str] mapped_column(Text, default) village: Mapped[Village] relationship(back_populatesrecords)代码中字段和设计中完全对应。由于 SQLite 的数据类型有限布尔值在实际存储时会保存为整数 0 或 1读取时 ORM 会自动转换成True或False不影响使用。4.4 数据库连接封装在app/database.py中配置数据库连接from sqlalchemy import create_engine from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker DATABASE_URL sqlite:///./village_survey.db engine create_engine( DATABASE_URL, connect_args{check_same_thread: False}, ) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) from app.models import Base def init_db(): Base.metadata.create_all(bindengine) def get_db(): db SessionLocal() try: yield db finally: db.close()check_same_threadFalse是 SQLite 与 FastAPI 协作时的常见配置因为 FastAPI 可能在不同线程中访问同一个连接。4.5 初始化示例数据为了演示效果在init_db.py中插入一条北白砂的示例村庄记录以及一条走访记录。经纬度先用一个近似位置正式使用时一定要以现场实测坐标为准。from datetime import date from app.database import SessionLocal, init_db from app.models import Village, VisitRecord def seed_data(): db SessionLocal() village Village( name北白砂, town鹿泉区示例镇, longitude114.34, latitude38.10, description本次村村通探访的实地样本村演示数据请以实测为准。, ) db.add(village) db.flush() record VisitRecord( village_idvillage.id, visit_datedate(2025, 3, 20), investigator张三, road_type水泥路, road_condition良好, bus_accessTrue, network_signal中, remark村内主干道硬化完成部分巷道宽度较窄。, ) db.add(record) db.commit() db.close() if __name__ __main__: init_db() seed_data() print(数据库初始化完成)初始化后执行python init_db.py再查看当前目录会发现多出一个village_survey.db文件。5. FastAPI 后端接口开发5.1 定义请求与响应模型Pydantic 模型用于接口的请求参数校验和响应格式化。在app/schemas.py中定义from datetime import date from pydantic import BaseModel, ConfigDict class VillageOut(BaseModel): model_config ConfigDict(from_attributesTrue) id: int name: str town: str longitude: float latitude: float description: str class VisitRecordCreate(BaseModel): village_id: int visit_date: date investigator: str road_type: str road_condition: str bus_access: bool False network_signal: str remark: str class VisitRecordOut(VillageOut, VisitRecordCreate): id: int其中VisitRecordOut让它既包含村庄名称也包含走访记录字段。这里直接继承VillageOut是一种简洁写法字段名重复时需要注意实际生产项目推荐改为显式声明village: VillageOut。5.2 编写接口在app/main.py中实现接口和页面渲染from fastapi import Depends, FastAPI, Request from fastapi.responses import HTMLResponse from fastapi.templating import Jinja2Templates from sqlalchemy import select from sqlalchemy.orm import Session from app.database import get_db, init_db from app.models import VisitRecord, Village from app.schemas import VisitRecordCreate, VisitRecordOut, VillageOut app FastAPI(title村庄连通性调查记录系统) templates Jinja2Templates(directorytemplates) app.on_event(startup) def on_startup(): init_db() app.get(/, response_classHTMLResponse) def index(request: Request): return templates.TemplateResponse(request, index.html) app.get(/api/villages, response_modellist[VillageOut]) def get_villages(db: Session Depends(get_db)): return db.scalars(select(Village).order_by(Village.id)).all() app.get(/api/records, response_modellist[VisitRecordOut]) def get_records(db: Session Depends(get_db)): result db.execute( select(VisitRecord, Village) .join(Village, VisitRecord.village_id Village.id) .order_by(VisitRecord.visit_date.desc()) ) data [] for record, village in result.all(): data.append( VisitRecordOut( idrecord.id, village_idrecord.village_id, visit_daterecord.visit_date, investigatorrecord.investigator, road_typerecord.road_type, road_conditionrecord.road_condition, bus_accessrecord.bus_access, network_signalrecord.network_signal, remarkrecord.remark, namevillage.name, townvillage.town, longitudevillage.longitude, latitudevillage.latitude, descriptionvillage.description, ) ) return data app.post(/api/records, response_modelVisitRecordOut) def create_record( payload: VisitRecordCreate, db: Session Depends(get_db), ): record VisitRecord(**payload.model_dump()) db.add(record) db.commit() db.refresh(record) village db.get(Village, record.village_id) return VisitRecordOut( idrecord.id, village_idrecord.village_id, visit_daterecord.visit_date, investigatorrecord.investigator, road_typerecord.road_type, road_conditionrecord.road_condition, bus_accessrecord.bus_access, network_signalrecord.network_signal, remarkrecord.remark, namevillage.name, townvillage.town, longitudevillage.longitude, latitudevillage.latitude, descriptionvillage.description, )当前接口逻辑虽然能工作但create_record中缺少对village_id是否存在的判断。如果前端传入了不存在的村庄 ID系统仍然会创建一条外键无效的记录。虽然 SQLite 默认开启了外键约束但 SQLAlchemy 的会话不一定实时检查因此最佳实践是在插入前显式检查village db.get(Village, payload.village_id) if not village: raise HTTPException(status_code404, detail村庄不存在)5.3 启动项目在项目根目录执行uvicorn app.main:app --reload --port 8000启动后访问接口文档http://127.0.0.1:8000/docs页面http://127.0.0.1:8000FastAPI 会自动生成 Swagger 风格文档在浏览器中可以直接测试接口。6. Leaflet 地图可视化6.1 页面加载村庄标记下面编写templates/index.html使用 Leaflet 把村庄显示在地图上并把走访记录渲染到标记点的弹窗中。!DOCTYPE html html langzh-CN head meta charsetUTF-8 meta nameviewport contentwidthdevice-width, initial-scale1.0 title鹿泉村村通探访记录/title link relstylesheet hrefhttps://unpkg.com/leaflet1.9.4/dist/leaflet.css script srchttps://unpkg.com/leaflet1.9.4/dist/leaflet.js/script style body { margin: 0; font-family: Microsoft YaHei, sans-serif; } #map { height: 100vh; width: 100%; } /style /head body div idmap/div script var map L.map(map).setView([38.10, 114.34], 12); L.tileLayer(https://{s}.tile.openstreetmap.org/{z}/{x}/{y}.png, { maxZoom: 19, attribution: copy; OpenStreetMap contributors }).addTo(map); var villageLayer L.layerGroup().addTo(map); var currentBounds []; function loadData() { Promise.all([ fetch(/api/villages).then(res res.json()), fetch(/api/records).then(res res.json()) ]).then(function ([villages, records]) { villageLayer.clearLayers(); records.forEach(function (r) { var marker L.marker([r.latitude, r.longitude]) .addTo(villageLayer) .bindPopup(buildPopup(r)); currentBounds.push([r.latitude, r.longitude]); }); villages.forEach(function (v) { var exist records.some(r r.village_id v.id); if (!exist) { var marker L.marker([v.latitude, v.longitude]) .addTo(villageLayer) .bindPopup(b v.name /bbr暂无走访记录); currentBounds.push([v.latitude, v.longitude]); } }); if (currentBounds.length 0) { map.fitBounds(currentBounds); } }); } function buildPopup(r) { var signalMap {1: 强, 2: 中, 3: 弱, 4: 无信号}; var busText r.bus_access ? 是 : 否; return b r.name /bbr 走访日期 r.visit_date br 走访人 r.investigator br 道路类型 r.road_type br 路况 r.road_condition br 是否通公交 busText br 信号评价 signalMap[r.network_signal] || r.network_signal br 备注 (r.remark || 无); } loadData(); /script /body /html在示例数据的弹窗中能直观看到北白砂这条走访记录。实际开发时network_signal字段如果直接用“中”“强”等文字存储代码中的signalMap就不需要了。这里为了展示前后端字段对齐的思路保留汉字存储更简洁。6.2 新增走访记录的交互建议如果希望直接在页面上录入走访记录把networkSignal改成下拉框并由fetch提交即可。核心请求代码如下fetch(/api/records, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ village_id: 1, visit_date: 2025-03-20, investigator: document.getElementById(investigator).value, road_type: document.getElementById(roadType).value, road_condition: document.getElementById(roadCondition).value, bus_access: document.getElementById(busAccess).checked, network_signal: document.getElementById(networkSignal).value, remark: document.getElementById(remark).value }) }).then(function (res) { return res.json(); }).then(function (data) { loadData(); });这个 POST 接口的核心作用是“提交一条走访结果”。每次走访生成一条新数据而不是覆盖之前的记录。这种设计可以防止误操作导致历史数据丢失。7. 实地走访数据采集的关键注意点在真正的“探访北白砂”过程中硬件设备、坐标体系和数据习惯会直接影响采集数据质量。7.1 GPS 坐标与地图坐标偏移手机定位得到的经纬度通常是 WGS84 坐标。但国内大部分互联网地图为了提高安全性会给坐标加上偏移算法结果得到 GCJ-02 坐标。常见的现象是手机中某个地图 App 显示的位置与导入到 Leaflet 标记的位置不一致。解决思路有两种使用手机端原生定位 SDK 获取原始坐标。使用地图厂商提供的坐标转换接口转换后再入库。本项目为了简洁直接按 WGS84 处理。如果你的坐标来自高德或腾讯地图需要先转成 WGS84否则地图点会偏移。转发时建议使用专门工具核对位置。7.2 走访照片的处理走访记录中经常要附带现场照片。数据库设计时没有直接存储图片文件而是预留了备注字段。如果要加照片一个简单的做法是增加photo_url字段图片上传到 OSS 或服务器指定目录。考虑到本系统可能运行在没有外部云服务的服务器上建议先约定统一的文件命名规则访问日期_村庄名_序号.jpg例如20250320_北白砂_01.jpg后期使用照片时可以通过文件名中的日期和村庄名快速定位现场图片。7.3 现场记录习惯这类调查系统最容易出现的问题是字段取值不统一。比如有人在备注里写“路不错”有人写“路面好还行”数据统计时就很难归类。建议在走访前先约定字典表。以路况评价为例统一使用良好、一般、较差、无法通行。如果录入时发现选项不够不要去数据库里临时加一段文字而是扩充字典项保持数据规范。7.4 常见问题排查本地运行或者部署过程中常见报错和解决方式如下。问题现象常见原因解决思路运行 init_db.py 提示找不到 app 模块没有在项目根目录执行先cd到项目根目录再运行Leaflet 地图显示空白CDN 无法访问替换可访问的地图 CDN 地址接口返回 422 错误Pydantic 校验失败日期或字段类型不对查看 FastAPI 文档页中的 Schema 提示POST 记录后村庄名称是空返回结果没有关联查询在create_record中二次查询 Village 信息点位偏移明显坐标系混用检查坐标来源是否经过加密偏移8. 工程化改进与最佳实践8.1 数据库层面的改进SQLite 适合本地演示。当系统要支持多人同时录入、后台管理、统计报表时建议迁移到 PostgreSQL。迁移时需要改动的主要是database.py中的连接信息DATABASE_URL postgresql://user:passwordlocalhost:5432/village_survey如果已经使用 SQLAlchemy ORM模型代码几乎不需要改动只需要调整连接字符串再使用alembic生成迁移脚本。这也是在项目一开始就使用 ORM 而不是裸 SQL 的价值所在。8.2 接口层增加权限与校验当前系统的走访记录接口是公开的。生产环境中走访记录可能涉及内部调查资料和村民信息必须做访问控制。一个最小方案是增加简单的 Token 认证登录后获取 Token在请求头中携带 Token 才能新增记录。FastAPI 中可以基于OAuth2PasswordBearer实现相关的依赖注入方式如下from fastapi.security import OAuth2PasswordBearer oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) def get_current_user(token: str Depends(oauth2_scheme)): # 校验 token 并返回当前用户 pass在以后的系统版本中建议把村庄档案维护和走访记录修改分开管理。普通走访人员只允许新增记录管理员可以编辑或删除历史记录。8.3 数据备份与导出如果走访数据通过手机输入上报后台数据库需要每天做备份。SQLite 的备份比较简单在服务停止时直接复制数据库文件即可。生产环境建议写入定时任务cp village_survey.db village_survey_$(date %Y%m%d).db然后保留最近 30 份备份文件。导出功能也很实用。走访记录最终往往要交给其他部门处理提供一个 CSV 导出接口可以降低人工拷贝成本。8.4 前期验证与回滚策略在向正式系统写入数据之前务必先使用测试数据和测试库跑通流程。例如先录入北白砂的一条测试走访记录检查地图点是否显示正确再决定是否批量导入历史数据。如果发现需要回滚优先使用备份文件恢复而不是直接删除表中记录。对于需要修改多条历史数据的场景例如统一替换村庄名称执行更新前先使用SELECT查看会受影响的记录数量确认无误后再执行UPDATE。保留一份更新前的 SQL 还原脚本是更稳妥的做法。9. 后续扩展方向这个系统目前已经具备一个调研类 GIS 应用的基本形态。如果继续深化可以从下面几个方向扩展把数据库从 SQLite 改为 PostgreSQL PostGIS支持空间查询。PostGIS 是 PostgreSQL 的空间扩展能够做“查询某一公里范围内的村庄”“沿着道路线计算可达性”等空间分析而这些查询用普通表格字段很难实现。增加走访任务管理模块。每次探访前先创建任务任务下再绑定具体村庄和走访人。这样后续数据统计可以按任务分组减少录入错误。增加历史趋势对比。系统保留了多条记录的时间维度可以画出某个村庄从春到秋的道路与信号变化折线图为基层调研提供更直观的数据支撑。如果愿意继续深入可以先研究 Flask、Django 或 FastAPI 中更完善的前后端分离方案。当前模板渲染方式有利于快速理解流程但前后端分离的方案在多人协作和移动端适配方面可维护性更好。项目完整代码可以直接按照本文步骤搭建。建议先复制代码跑通一条记录再按自己的思路补充“附近公交站点”“道路宽度”“是否错车困难”等字段最终打磨成符合真实调研场景的系统。这样整个项目做出来之后不仅能记录一次鹿泉探访也能复用到其他乡镇的村庄调查工作中。