由于前面的工作全部脱离于队伍与框架,只着眼于功能实现,这期博客任务主要是下载并理解团队提供的后端框架(基于 FastAPI),在现有架构中新增 map 模块,并完成路由(Router)的注册与接入。此外与爬虫板块队友交流后,地图通勤功能还要具备地理编码解析(地址转坐标)和周边设施全部直接提取的能力。

一、框架认识以及首要准备

  1. 框架落地:下载项目源码,理解混合技术栈结构。拿到项目代码后,我首先分析了 house-recommend 目录下的结构。这是一个典型的前后端混合工程:根目录:包含 package.json 和 src/,说明前端或 Agent 逻辑基于 TypeScript。

  2. 环境构建:配置 nodejs、httpx、Python 依赖与敏感信息环境变量。

  3. 模块开发api/ 目录:这是 Python 后端的核心。我注意到框架已经预设了 crawler(爬虫)模块,采用了标准的 MVC 分层设计:routers/:负责定义 URL 路由,接收前端请求。services/:负责具体的业务逻辑处理。现在在 api/map 目录下实现地图服务的业务逻辑与接口路由。

二、中文地址转经纬度

由于队友的爬虫板块只能返回中文地址,而地图api的所需参数是经纬度,所以要加一个转化过程。

关键代码实现
在 MapService 类中,我编写了 get_location 方法。这里有几个关键的处理细节:

  • 环境变量安全调用:使用 os.getenv("AMAP_API_KEY") 动态读取密钥,确保 Key 不硬编码在代码中。
  • 异常处理机制:地址解析可能会失败(如地址不存在),必须捕获异常并抛出友好的提示,而不是让程序崩溃。
  • 数据清洗:高德返回的经纬度是字符串(如 "117.117,36.65"),我们需要将其拆分为 float 类型的元组。
# 关键代码片段:地址解析逻辑
async def get_location(self, address: str) -> Tuple[float, float]:
    url = f"{self.base_url}/geocode/geo" # 地理编码接口
    params = {
        "address": address,
        "city": "" # 留空表示全国搜索,不局限于特定城市
    }
    
    data = await self._request(url, params) # 调用内部通用请求方法
    
    if data.get("geocodes"):
        # 1. 原始数据处理:从JSON中提取 location 字段
        location_str = data["geocodes"][0]["location"] 
        # 2. 类型转换:字符串 "117.117,36.65" -> 元组 (117.117, 36.65)
        lng, lat = map(float, location_str.split(","))
        return lng, lat
    else:
        raise ValueError(f"无法解析地址: {address}")

三、地图模块的路由设计与实现

在 FastAPI 框架中,路由(Router)是连接前端请求与后端逻辑的桥梁。为了实现地图功能,我在 api/map/routers/ 目录下创建了 map_router.py

  • 定义前缀:将所有地图相关的接口统一挂载在 /map 路径下,避免与爬虫接口冲突。
  • 逻辑分发:接收请求后,调用 services 层处理业务,最后返回 JSON 数据。

关键代码片段:map_router.py

router = APIRouter(prefix="/map", tags=["地图分析服务"])
map_service = MapService()

@router.post("/analyze")
async def analyze_location(request: AddressRequest):
    """
    接收地址,返回周边配套和通勤分析
    """
    try:
        result = await map_service.analyze_house_location(request.address)
        if "error" in result:
            raise HTTPException(status_code=400, detail=result["error"])
        return result
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

通过这段代码,我深刻理解了 FastAPI 的 解耦思想routers 层只负责“传话”(接收参数、返回结果),具体的 API 密钥调用、HTTP 请求发送等复杂逻辑,全部被封装在 services 层中。这种设计使得代码维护性极强。

四、框架入口:在 app.py 中注册模块

api/app.py 是整个后端应用的总指挥。它负责初始化 FastAPI 实例,并将各个模块(Router)“挂载”到应用中。

# 导入刚才写的 map_router
from map.routers.map_router import router as map_router

# 注册地图路由
app.include_router(map_router) 

五、周边配套设施自动查询以及通勤分析功能的逻辑转变

在上次博客工作对于周边设施查询方面只做了简单的关键词手动查找,并不符合项目功能自动获取信息并储存返回的要求,于是改为一次性查出所以信息,包括目标地址3km以内的所有"地铁站","公交站","医院","超市"的数量

async def analyze_house_location(self, address: str) -> Dict[str, Any]:
        """
        主流程:一次性查出所有信息
        """
        try:
            # 1. 获取房源坐标
            house_lng, house_lat = await self.get_location(address)
            house_loc = (house_lng, house_lat)

            # 2. 并行搜索周边 (地铁、医院、超市)
            # 注意:高德搜索关键词:地铁站用 '地铁站',公交站用 '公交站'
            metro_pois = await self.search_nearby(house_lng, house_lat, "地铁站")
            hospital_pois = await self.search_nearby(house_lng, house_lat, "医院")
            supermarket_pois = await self.search_nearby(house_lng, house_lat, "超市")
            bus_pois = await self.search_nearby(house_lng, house_lat, "公交站")

然后组装

此外通勤分析方面也是类似逻辑,这里原本准备将所有周边设施的时间计算出但是实际上高德api对于短时间内大量计算请求很可能会出现问题,所以从实际角度出发改为计算步行到最近公共交通地点的时间,符合“通勤能力”的体现

nearest_metro = {"name": "无", "distance": "未知", "duration": "未知"}
            nearest_bus = {"name": "无", "distance": "未知", "duration": "未知"}

            # 计算最近地铁
            if metro_pois:
                metro_loc = tuple(map(float, metro_pois[0]["location"].split(",")))
                walking_info = await self.get_walking_info(house_loc, metro_loc)
                nearest_metro = {
                    "name": metro_pois[0]["name"],
                    "distance": walking_info["distance"],
                    "duration": walking_info["duration"]
                }

            # 计算最近公交
            if bus_pois:
                bus_loc = tuple(map(float, bus_pois[0]["location"].split(",")))
                walking_info = await self.get_walking_info(house_loc, bus_loc)
                nearest_bus = {
                    "name": bus_pois[0]["name"],
                    "distance": walking_info["distance"],
                    "duration": walking_info["duration"]
                }

六、总结

本次博客是真正意义上的对于任务2.3地图POI以及通勤分析的基本完成,后续主要考虑修复,优化和升级工作,下次博客将继续在框架的基础之上开始进行对任务2.4评分功能的开发。

更多推荐