WonderableAI/shaguabijia-app-server
10
0
Fork 0
Code Issues Pull Requests 8 Actions Packages Projects Releases Wiki Activity
Files
7b6756f93658be79ac424055f8f8941f72a60ab6
shaguabijia-app-server/app/utils/meituan_city.py
T
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

449 lines
16 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""美团城市词典 + reverse_geocoder 离线反查。
从 feed 入参的 latitude/longitude 计算出美团城市 ID，
用于后续美团 CPS 接口的 cityId 参数。
⚠️ 跨系统耦合:本模块返回的 city_id 取自 data/city_dict.txt,而离线库
`meituan_coupon.city_id` 由 ETL(另一套系统)灌入。二者必须用同一份城市 ID 口径,
否则 `WHERE city_id == <本模块结果>` 会静默查到 0 行 → 接口永久降级返空。
改动 city_dict.txt 或 ETL 的城市 ID 来源时,务必同步两侧。
"""
from __future__ import annotations
import logging
import re
from functools import lru_cache
from pathlib import Path
from app.utils.geo import get_city as _get_geo_city
logger = logging.getLogger("shagua.meituan_city")
# city_dict.txt 作为运行时数据随包分发(见 pyproject [tool.setuptools.package-data])
_CITY_DICT_PATH = Path(__file__).resolve().parent / "data" / "city_dict.txt"
# ─────────── 反向地理编码 admin1 → 中文省份名 ───────────
# reverse_geocoder 的 admin1 格式不统一:
# 直辖市: "Beijing" / "Shanghai Shi" / "Tianjin Shi" / "Chongqing Shi"
# 省份: "Guangdong" / "Jiangsu Sheng" / "Hubei" ...
# 自治区: "Xinjiang Uygur Zizhiqu" / "Tibet Autonomous Region" ...
# 下面用前缀匹配，去掉了 Sheng/Shi/Zizhiqu/Autonomous Region 等后缀。
_PROVINCE_EN_PREFIX: list[tuple[str, str]] = [
# 直辖市 — admin1 即城市名
("Beijing", "北京市"),
("Shanghai", "上海市"),
("Tianjin", "天津市"),
("Chongqing", "重庆市"),
# 省
("Hebei", "河北省"),
("Shanxi", "山西省"), # 注意: 指山西省,不是陕西
("Liaoning", "辽宁省"),
("Jilin", "吉林省"),
("Heilongjiang", "黑龙江省"),
("Jiangsu", "江苏省"),
("Zhejiang", "浙江省"),
("Anhui", "安徽省"),
("Fujian", "福建省"),
("Jiangxi", "江西省"),
("Shandong", "山东省"),
("Henan", "河南省"),
("Hubei", "湖北省"),
("Hunan", "湖南省"),
("Guangdong", "广东省"),
("Hainan", "海南省"),
("Sichuan", "四川省"),
("Guizhou", "贵州省"),
("Yunnan", "云南省"),
("Shaanxi", "陕西省"), # 双写 a 是官方拼音
("Gansu", "甘肃省"),
("Qinghai", "青海省"),
# 自治区 — 注意匹配顺序, Xinjiang 要在 Guangxi 前面(Guangxi 也是 Xi 开头但先匹配 Xin 不会误判)
("Guangxi", "广西壮族自治区"),
("Inner Mongolia", "内蒙古自治区"),
("Nei Mongol", "内蒙古自治区"),
("Tibet", "西藏自治区"),
("Xizang", "西藏自治区"),
("Ningxia", "宁夏回族自治区"),
("Xinjiang", "新疆维吾尔自治区"),
# 特别行政区
("Hong Kong", "香港特别行政区"),
("Macau", "澳门特别行政区"),
("Macao", "澳门特别行政区"),
# 台湾（city_dict 里省份名为 "台湾"，没有省/自治区后缀）
("Taiwan", "台湾"),
]
# ─────────── 常见城市名 英文→中文 映射 ───────────
# 覆盖所有直辖市 + 省会 + 一线城市 + 部分 reverse_geocoder 只能命中到区/县的城市。
# key 全小写，匹配时做小写比较。
_CITY_EN_TO_CN: dict[str, str] = {
# 直辖市
"beijing": "北京市",
"shanghai": "上海市",
"tianjin": "天津市",
"chongqing": "重庆市",
# 省会 / 副省级
"guangzhou": "广州市",
"shenzhen": "深圳市",
"chengdu": "成都市",
"hangzhou": "杭州市",
"wuhan": "武汉市",
"xi'an": "西安市",
"nanjing": "南京市",
"changsha": "长沙市",
"zhengzhou": "郑州市",
"jinan": "济南市",
"kunming": "昆明市",
"fuzhou": "福州市",
"harbin": "哈尔滨市",
"lanzhou": "兰州市",
"guiyang": "贵阳市",
"nanning": "南宁市",
"shijiazhuang": "石家庄市",
"taiyuan": "太原市",
"shenyang": "沈阳市",
"changchun": "长春市",
"hefei": "合肥市",
"nanchang": "南昌市",
"haikou": "海口市",
"hohhot": "呼和浩特市",
"huhehaote": "呼和浩特市",
"urumqi": "乌鲁木齐市",
"wulumuqi": "乌鲁木齐市",
"lhasa": "拉萨市",
"yinchuan": "银川市",
"xining": "西宁市",
# 其他常见城市
"xiamen": "厦门市",
"suzhou": "苏州市",
"qingdao": "青岛市",
"dalian": "大连市",
"ningbo": "宁波市",
"wuxi": "无锡市",
"foshan": "佛山市",
"dongguan": "东莞市",
"zhuhai": "珠海市",
"zhongshan": "中山市",
"wenzhou": "温州市",
"shaoxing": "绍兴市",
"jiaxing": "嘉兴市",
"jinhua": "金华市",
"taizhou": "台州市",
"yangzhou": "扬州市",
"nantong": "南通市",
"changzhou": "常州市",
"xuzhou": "徐州市",
"zhengjiang": "镇江市",
"yantai": "烟台市",
"weifang": "潍坊市",
"zibo": "淄博市",
"linyi": "临沂市",
"weihai": "威海市",
"rizhao": "日照市",
"luoyang": "洛阳市",
"kaifeng": "开封市",
"xinxiang": "新乡市",
"nanyang": "南阳市",
"yichang": "宜昌市",
"xiangyang": "襄阳市",
"huangshi": "黄石市",
"zhuzhou": "株洲市",
"xiangtan": "湘潭市",
"yueyang": "岳阳市",
"hengyang": "衡阳市",
"mianyang": "绵阳市",
"luzhou": "泸州市",
"yibin": "宜宾市",
"nanchong": "南充市",
"zigong": "自贡市",
"qujing": "曲靖市",
"yuxi": "玉溪市",
"zunyi": "遵义市",
"guilin": "桂林市",
"liuzhou": "柳州市",
"sanya": "三亚市",
"tangshan": "唐山市",
"baoding": "保定市",
"handan": "邯郸市",
"qinhuangdao": "秦皇岛市",
"langfang": "廊坊市",
"datong": "大同市",
"changzhi": "长治市",
"linfen": "临汾市",
"baotou": "包头市",
"ordos": "鄂尔多斯市",
"eerduosi": "鄂尔多斯市",
"daqing": "大庆市",
"qiqihar": "齐齐哈尔市",
"jilin_city": "吉林市",
"anshan": "鞍山市",
"fushun": "抚顺市",
"benxi": "本溪市",
"jinzhou": "锦州市",
"yingkou": "营口市",
"dandong": "丹东市",
"huizhou": "惠州市",
"jiangmen": "江门市",
"zhanjiang": "湛江市",
"maoming": "茂名市",
"zhaoqing": "肇庆市",
"chaozhou": "潮州市",
"shantou": "汕头市",
"shaoguan": "韶关市",
"meizhou": "梅州市",
"jieyang": "揭阳市",
"qingyuan": "清远市",
"heyuan": "河源市",
"yangjiang": "阳江市",
"shanwei": "汕尾市",
"yunfu": "云浮市",
}
# ─────────── 省会映射（城市匹配失败时回退） ───────────
# city_dict.txt 内省份的第一个城市不一定是省会，故显式维护。
_PROVINCE_CAPITAL: dict[str, str] = {
"安徽省": "合肥市",
"澳门特别行政区": "澳门",
"北京市": "北京市",
"福建省": "福州市",
"甘肃省": "兰州市",
"广东省": "广州市",
"广西壮族自治区": "南宁市",
"贵州省": "贵阳市",
"海南省": "海口市",
"河北省": "石家庄市",
"河南省": "郑州市",
"黑龙江省": "哈尔滨市",
"湖北省": "武汉市",
"湖南省": "长沙市",
"吉林省": "长春市",
"江苏省": "南京市",
"江西省": "南昌市",
"辽宁省": "沈阳市",
"内蒙古自治区": "呼和浩特市",
"宁夏回族自治区": "银川市",
"青海省": "西宁市",
"山东省": "济南市",
"山西省": "太原市",
"陕西省": "西安市",
"上海市": "上海市",
"四川省": "成都市",
"台湾": "台北市",
"天津市": "天津市",
"西藏自治区": "拉萨市",
"香港特别行政区": "香港",
"新疆维吾尔自治区": "乌鲁木齐市",
"云南省": "昆明市",
"浙江省": "杭州市",
"重庆市": "重庆市",
}
# ─────────── 城市字典加载 ───────────
def _parse_city_dict(path: str | Path) -> list[dict[str, str]]:
"""解析 city_dict.txt，返回 [{city_id, city_name, province_name}, ...]。
city_dict.txt 格式（TSV）:
城市ID\t城市名称\t省份名称
示例行:
3NUYJKKJXPHVNZUHFK3HWUDHNM\t宣城市\t安徽省
"""
data: list[dict[str, str]] = []
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
parts = line.split("\t")
if len(parts) < 3:
continue
city_id, city_name, province_name = parts[0], parts[1], parts[2]
if city_id == "城市ID":
continue # 跳过表头
if city_id and city_name and province_name:
data.append({
"city_id": city_id,
"city_name": city_name,
"province_name": province_name,
})
return data
# 模块加载时一次解析
try:
_CITY_DICT: list[dict[str, str]] = _parse_city_dict(_CITY_DICT_PATH)
except Exception:
logger.exception("加载 city_dict.txt 失败,美团城市反查将不可用")
_CITY_DICT = []
def _build_province_index() -> dict[str, list[dict[str, str]]]:
"""构建 省份名 → 该省全部城市列表 的索引。"""
idx: dict[str, list[dict[str, str]]] = {}
for entry in _CITY_DICT:
idx.setdefault(entry["province_name"], []).append(entry)
return idx
_PROVINCE_INDEX: dict[str, list[dict[str, str]]] | None = None
def _get_province_index() -> dict[str, list[dict[str, str]]]:
global _PROVINCE_INDEX
if _PROVINCE_INDEX is None:
_PROVINCE_INDEX = _build_province_index()
return _PROVINCE_INDEX
# ─────────── 查询 ───────────
def _map_admin1_to_cn_province(admin1: str) -> str:
"""将 reverse_geocoder 的 admin1 映射到 city_dict 中的中文省份名。"""
if not admin1:
return ""
normalized = admin1.strip()
# 多级匹配:先精确、再前缀
for en_prefix, cn_name in _PROVINCE_EN_PREFIX:
if normalized == en_prefix or normalized.startswith(en_prefix):
return cn_name
return ""
def _lookup_city_in_province(city_en_lower: str, province_cn: str) -> str:
"""在指定省份内查找匹配的城市名（EN→CN 映射）。"""
if not province_cn:
return ""
index = _get_province_index()
candidates = index.get(province_cn, [])
if not candidates:
return ""
# 1) 精确映射
if city_en_lower in _CITY_EN_TO_CN:
cn_city = _CITY_EN_TO_CN[city_en_lower]
for c in candidates:
if c["city_name"] == cn_city:
return cn_city
# 2) 前缀/包含匹配（处理 admin1 直辖市场景:行政区 → 直辖市本身）
for c in candidates:
# 去掉"市"后缀比较
city_core = c["city_name"].rstrip("市")
if city_en_lower.startswith(city_core.lower()) or city_core.lower().startswith(city_en_lower):
return c["city_name"]
# city_en_lower 可能是拼音，city_core 是中文，尝试从 EN→CN 映射反向匹配
for en_k, cn_v in _CITY_EN_TO_CN.items():
if cn_v == c["city_name"] and (city_en_lower in en_k or en_k in city_en_lower):
return cn_v
# 3) 匹配不到 → 返回省会
capital = _PROVINCE_CAPITAL.get(province_cn, "")
if capital:
for c in candidates:
if c["city_name"] == capital:
return capital
return candidates[0]["city_name"] # 终极兜底
def _sanitize_city_name(name: str) -> str:
"""去除 reverse_geocoder name 中常见的行政后缀使匹配更鲁棒。"""
# 去掉 " District" / " Qu" / " Shi" 等英文后缀
for suffix in ("District", "Qu", "Shi", "Sheng", "Xian", "Cun", "Zhen", "Xiang",
"Zizhiqu", "Autonomous Region", "Special Administrative Region"):
name = re.sub(rf"\s+{suffix}$", "", name, flags=re.IGNORECASE)
return name.strip()
@lru_cache(maxsize=512)
def _resolve_meituan_city(latitude: float, longitude: float) -> dict[str, str]:
"""反查实现;入参已量化(见 get_meituan_city),故 lru_cache 命中率高。
返回的 dict 被缓存复用 —— 调用方勿原地修改(get_meituan_city 已返回副本)。
"""
if not _CITY_DICT:
return {"city_id": "", "city_name": "", "province_name": ""}
logger.debug("resolve_meituan_city: lat=%.2f lon=%.2f", latitude, longitude)
geo = _get_geo_city(latitude, longitude)
name_en = _sanitize_city_name(geo.get("name", ""))
admin1 = geo.get("admin1", "")
country = geo.get("country", "")
if country != "CN":
logger.debug("resolve_meituan_city: 坐标(%.2f,%.2f)不在中国境内(country=%s)", latitude, longitude, country)
return {"city_id": "", "city_name": "", "province_name": ""}
# 1) 映射省份
province_cn = _map_admin1_to_cn_province(admin1)
if not province_cn:
logger.warning("get_meituan_city: admin1=%r 无法映射到中文省份", admin1)
return {"city_id": "", "city_name": "", "province_name": ""}
# 2) 查找城市
name_lower = name_en.lower()
city_cn = _lookup_city_in_province(name_lower, province_cn)
# 3) 按省份+城市匹配 city_dict 中的城市 ID
index = _get_province_index()
candidates = index.get(province_cn, [])
for c in candidates:
if city_cn and c["city_name"] == city_cn:
return {
"city_id": c["city_id"],
"city_name": c["city_name"],
"province_name": province_cn,
}
# 4) 最终回退:返回该省省会
if candidates:
capital = _PROVINCE_CAPITAL.get(province_cn, "")
if capital:
for c in candidates:
if c["city_name"] == capital:
logger.info("get_meituan_city: 城市匹配失败 name_en=%r, 回退到省会 %s", name_en, capital)
return {
"city_id": c["city_id"],
"city_name": capital,
"province_name": province_cn,
}
# 终极兜底:第一个城市
fallback = candidates[0]
logger.info("get_meituan_city: 城市匹配失败 name_en=%r, 回退到 %s", name_en, fallback["city_name"])
return {
"city_id": fallback["city_id"],
"city_name": fallback["city_name"],
"province_name": province_cn,
}
return {"city_id": "", "city_name": "", "province_name": ""}
def get_meituan_city(latitude: float, longitude: float) -> dict[str, str]:
"""根据经纬度反查美团城市 ID + 城市名 + 省份名(对外入口)。
返回:
- city_id: 美团城市 ID(如 3NUYJKKJXPHVNZUHFK3HWUDHNM);
匹配失败时返回 ""
- city_name: 中文城市名(如 "北京市")
- province_name: 中文省份名(如 "北京市")
原理:
1. reverse_geocoder 根据经纬度查出英文地名 + 省份
2. 英文省份→中文省份映射(前缀匹配)
3. 英文地名→中文城市名映射(精确映射 + 省内候选回退)
4. 在 city_dict.txt 中按省份+城市名匹配城市 ID
城市名匹配失败的策略:
- 直辖市(京沪津渝): admin1 本身即城市名,直接取
- 省会: 回退到该省第一个城市(city_dict.txt 中每个省的省会通常排第一位)
实现说明:先把坐标量化到 ~1km(round 到 2 位小数)再进 lru_cache —— 原始 GPS 坐标
每次抖动到小数点后 5~6 位,直接做缓存 key 几乎不命中;城市级解析对 1km 误差不敏感,
量化后"同一地点反复请求"可命中缓存。返回缓存 dict 的副本,调用方可安全读写。
"""
return dict(_resolve_meituan_city(round(latitude, 2), round(longitude, 2)))
Reference in New Issue View Git Blame Copy Permalink