Files
guke 7b6756f936 feat(meituan-cps): 经纬度→城市离线反查 + rec/销量最高按城市过滤 (#116)
## 主要功能
新增离线「经纬度 → 美团 cityId」反查,让 `rec`(智能推荐)与 `top-sales`(销量最高)从离线库只返**同城券**(此前会混返异地券)。

- `app/utils/geo.py` + `app/utils/meituan_city.py`:坐标 → 美团 `city_id`(reverse_geocoder 离线反查,零网络)。
- `feed?tab=rec` / `/top-sales`:按 `city_id` 过滤;解析不出 / 老客户端不带坐标 → 降级返空。
- `top-sales` 与 `rec` 一致置空库内距离(相对城市默认点、对用户无意义)。

---------

Co-authored-by: guke <guke@autohome.com.cn>
Reviewed-on: #116
Co-authored-by: guke <guke@wonderable.ai>
Co-committed-by: guke <guke@wonderable.ai>
2026-07-05 09:31:53 +08:00

138 lines
6.0 KiB
Python

"""reverse_geocoder 经纬度→城市 测试。
验证离线库对国内主要城市的匹配准确性。注意:gazetteer 的中国数据粒度不一致——
直辖市/省会通常直接命中城市名,部分城市可能命中到区/街道级(如天津→Erwangzhuang、
西安→Zhangjiabao),此时 admin1 为省级行政区。测试以 admin1(省级)匹配为主。
"""
from __future__ import annotations
import pytest
from app.utils.geo import get_city
# ─────────────── 国内主要城市 ───────────────
# (城市, 纬度, 经度, 期望 admin1 包含字串)
_CITY_CASES = [
# 直辖市 — admin1 即城市名(可能带 Shi 后缀)
("北京", 39.9042, 116.4074, "Beijing"),
("上海", 31.2304, 121.4737, "Shanghai"),
("重庆", 29.4316, 106.9123, "Chongqing"),
("天津", 39.3434, 117.3616, "Tianjin"),
# 省会 / 一线 — admin1 为省份
("广州", 23.1291, 113.2644, "Guangdong"),
("深圳", 22.5431, 114.0579, "Guangdong"),
("成都", 30.5728, 104.0668, "Sichuan"),
("杭州", 30.2741, 120.1551, "Zhejiang"),
("武汉", 30.5928, 114.3055, "Hubei"),
("西安", 34.3416, 108.9398, "Shaanxi"),
("南京", 32.0603, 118.7969, "Jiangsu"),
("长沙", 28.2282, 112.9388, "Hunan"),
("郑州", 34.7466, 113.6253, "Henan"),
("济南", 36.6512, 116.9946, "Shandong"),
("昆明", 25.0389, 102.7183, "Yunnan"),
("福州", 26.0745, 119.2965, "Fujian"),
("哈尔滨", 45.8038, 126.5350, "Heilongjiang"),
("乌鲁木齐", 43.8256, 87.6168, "Xinjiang"),
("拉萨", 29.6500, 91.1000, "Tibet"),
# 非省会
("厦门", 24.4798, 118.0894, "Fujian"),
("苏州", 31.2990, 120.5853, "Jiangsu"),
("青岛", 36.0671, 120.3826, "Shandong"),
]
@pytest.mark.parametrize("label,lat,lon,expected_admin1", _CITY_CASES)
def test_city_admin1_match(label: str, lat: float, lon: float, expected_admin1: str) -> None:
"""所有城市经纬度应能匹配到正确的省级行政区 (admin1)。"""
r = get_city(lat, lon)
assert r["name"] != "", f"{label}: name should not be empty"
assert r["country"] == "CN", f"{label}: expected country=CN, got={r['country']}"
assert expected_admin1 in r["admin1"], \
f"{label}: expected admin1 to contain '{expected_admin1}', got={r['admin1']!r}"
# ─────────────── 直辖市 / 省会直接命中城市名 ───────────────
# 这些城市在 gazetteer 中的坐标恰好命中城市级条目(而非区/街道级),
# 验证 name 字段也正确。
_DIRECT_HIT_CASES = [
("北京", 39.9042, 116.4074, "Beijing"),
("上海", 31.2304, 121.4737, "Shanghai"),
("广州", 23.1291, 113.2644, "Guangzhou"),
("深圳", 22.5431, 114.0579, "Shenzhen"),
("成都", 30.5728, 104.0668, "Chengdu"),
("杭州", 30.2741, 120.1551, "Hangzhou"),
("郑州", 34.7466, 113.6253, "Zhengzhou"),
("济南", 36.6512, 116.9946, "Jinan"),
("昆明", 25.0389, 102.7183, "Kunming"),
("哈尔滨", 45.8038, 126.5350, "Harbin"),
("厦门", 24.4798, 118.0894, "Xiamen"),
("苏州", 31.2990, 120.5853, "Suzhou"),
("青岛", 36.0671, 120.3826, "Qingdao"),
("拉萨", 29.6500, 91.1000, "Lhasa"),
]
@pytest.mark.parametrize("label,lat,lon,expected_name", _DIRECT_HIT_CASES)
def test_city_name_direct_hit(label: str, lat: float, lon: float, expected_name: str) -> None:
"""直辖市/省会等主要城市坐标应直接命中城市名(而非区/街道级)。"""
r = get_city(lat, lon)
assert r["name"] == expected_name, \
f"{label}: expected name={expected_name}, got={r['name']!r}"
# ─────────────── 边界情况 ───────────────
def test_ocean_not_china() -> None:
"""远洋坐标不应误判为国内城市。"""
# 太平洋中部 → 可能匹配到最近有人岛(如法属波利尼西亚 Taiohae),但绝不应是 CN
r = get_city(0.0, -140.0)
assert r["country"] != "CN", f"mid-Pacific should not be CN, got {r}"
# 南大西洋
r2 = get_city(-30.0, -20.0)
assert r2["country"] != "CN", f"South Atlantic should not be CN, got {r2}"
def test_return_keys_and_types() -> None:
"""返回 dict 应包含全部五个字段且类型为 str。"""
r = get_city(39.9042, 116.4074)
for key in ("name", "admin1", "country", "latitude", "longitude"):
assert key in r, f"missing key: {key}"
assert isinstance(r[key], str), f"key {key} should be str, got {type(r[key])}"
def test_empty_result_keys() -> None:
"""结果始终应包含完整字段且全为 str 类型(即使匹配到偏远地)。"""
# reverse_geocoder KDTree 总找最近聚居点;业务侧如需判定"是否有效城市"
# 应自行按 country / admin1 做二次校验,而非依赖空字符串。
r = get_city(0.0, -140.0)
assert r["name"] != ""
assert isinstance(r["name"], str)
assert isinstance(r["admin1"], str)
assert isinstance(r["country"], str)
assert isinstance(r["latitude"], str)
assert isinstance(r["longitude"], str)
def test_same_coords_consistent() -> None:
"""同一坐标两次查询应返回相同结果(幂等)。"""
r1 = get_city(31.2304, 121.4737)
r2 = get_city(31.2304, 121.4737)
assert r1 == r2
def test_near_border_has_result() -> None:
"""省界附近的坐标应返回结果(非空 + 国内)。"""
# 苏鲁豫皖交界区域(徐州/商丘/宿州附近)
r = get_city(34.2, 116.8)
assert r["name"] != "", "border region should find a nearby populated place"
assert r["country"] == "CN"
def test_extreme_lat_lon_no_crash() -> None:
"""极值经纬度不应抛异常。"""
r1 = get_city(90.0, 0.0) # 北极
r2 = get_city(-90.0, 0.0) # 南极
assert isinstance(r1, dict)
assert isinstance(r2, dict)