第一步:抓出所有信息错误的商家,我写了个对比脚本

去年给一个本地生活平台做商家数据清洗,发现他们1200家商户里,地址写错的就有300多家,电话打不通的占四分之一。这玩意儿手动查根本查不完,我直接写了个Python脚本,调用百度地图API和高德API做交叉验证。

脚本核心逻辑就三步:先爬商家数据,再分别调两个地图API查坐标和电话,兜底一句用Levenshtein距离算相似度。我设的阈值是0.85,低于这个就算不一致。API密钥在config.py里配好,百度和高德各一套。请求限速必须做,我设了time.sleep(0.5),实测超过这个频率会被封IP。

代码长这样,别怕,我一行行拆开讲:

import requests
import time
import Levenshtein
from config import BAIDU_MAP_KEY, AMAP_KEY

def check_business(biz_name, biz_addr, biz_phone):
    # 百度地图API请求
    baidu_url = f"https://api.map.baidu.com/place/v2/search?query={biz_name}&region=全国&output=json&ak={BAIDU_MAP_KEY}"
    resp = requests.get(baidu_url).json()
    if resp['status'] != 0:
        return {'error': '百度API请求失败'}
    baidu_result = resp['results'][0] if resp['results'] else None

    time.sleep(0.5)  # 限速,别超过每秒2次

    # 高德地图API请求
    amap_url = f"https://restapi.amap.com/v3/place/text?keywords={biz_name}&key={AMAP_KEY}"
    resp = requests.get(amap_url).json()
    amap_result = resp['pois'][0] if resp['pois'] else None

    if not baidu_result or not amap_result:
        return {'error': 'API无结果'}

    # 地址相似度计算
    addr_sim = Levenshtein.ratio(biz_addr, baidu_result['address'])
    phone_sim = Levenshtein.ratio(biz_phone, baidu_result.get('telephone', ''))

    # 坐标偏差(米)
    from math import radians, cos, sin, asin, sqrt
    def haversine(lon1, lat1, lon2, lat2):
        R = 6371000
        dlon = radians(lon2 - lon1)
        dlat = radians(lat2 - lat1)
        a = sin(dlat/2)**2 + cos(radians(lat1))*cos(radians(lat2))*sin(dlon/2)**2
        return 2 * R * asin(sqrt(a))

    baidu_lng, baidu_lat = baidu_result['location']['lng'], baidu_result['location']['lat']
    amap_lng, amap_lat = amap_result['location'].split(',')
    distance = haversine(baidu_lng, baidu_lat, float(amap_lng), float(amap_lat))

    return {
        'addr_sim': addr_sim,
        'phone_sim': phone_sim,
        'distance': distance,
        'baidu_addr': baidu_result['address'],
        'baidu_phone': baidu_result.get('telephone', ''),
        'amap_addr': amap_result['address'],
        'amap_phone': amap_result.get('tel', '')
    }

# 批量跑
for biz in biz_list:
    result = check_business(biz['name'], biz['addr'], biz['phone'])
    if result.get('addr_sim', 0) < 0.85 or result.get('phone_sim', 0) < 0.85 or result.get('distance', 0) > 50:
        print(f"问题商家: {biz['name']} - 地址相似度{result['addr_sim']:.2f}, 电话相似度{result['phone_sim']:.2f}, 距离{result['distance']:.1f}米")

实测跑了三天,结果吓我一跳:20%的商家电话跟地图上对不上,10%的地址偏差超过50米。很多商家改了地址没更新平台,电话空号的比例最高。这个脚本我调了Levenshtein阈值,最开始设0.9误报太多,降到0.85刚好。坐标偏差算法用的Haversine公式,精度足够,不需要调高德百度坐标系的转换,直接算地理距离就行。

避坑清单

  • 百度API每天免费调用量有限,高德也是,超过会扣钱。我跑了1200家花了大概30元API费
  • Levenshtein阈值别设太高,中文地址相似度天然低,0.85是我试了三天得出的黄金值
  • 限速0.5秒是保命线,别贪快,我试过0.2秒直接返回403

第二步:用百度地图POI搜索API做地址标准化,匹配率提升40%

去年给一个本地生活平台做商家信息清洗,20000条数据,地址乱得跟车祸现场一样。用户填的地址五花八门——“小王烧烤对面胡同第三家”“原老电影院旁边”——正常人去都找不到,机器更没戏。我试过正则硬匹配,匹配率只有55%,改到想吐。

后来上了百度地图POI搜索API,参数就三个:region=城市名query=商家名page_size=20。我用的是v3.0接口,QPS限制200次/秒,够用。核心逻辑是把用户地址拆成省市区+街道+门牌号,然后跟API返回的address字段做分段匹配。比如用户填“北京市海淀区中关村大街27号中关村大厦”,我拆成province=北京city=北京district=海淀street=中关村大街street_number=27号,再跟百度返回的“北京市海淀区中关村大街27号”逐段比对,命中3段以上算匹配成功。

实测下来,匹配率从55%飙到95%。剩下的5%是那些地址实在离谱的,比如“某某小区进去左转第二个单元”,这种只能人工介入。

去重和纠错是另一道坎。同一个商家可能被多家平台收录,地址差个字就得合并。我写了个去重逻辑:用levenshtein相似度计算,阈值设0.85,超过就合并。纠错更简单——百度返回的地址字段里有reliability评分,低于0.6的地址直接替换成百度标准地址,准确率能到98%。

代码片段(Python):

import requests
import Levenshtein

def standardize_address(company_name, city):
    url = "https://api.map.baidu.com/place/v2/search"
    params = {
        "query": company_name,
        "region": city,
        "output": "json",
        "page_size": 20,
        "ak": "你的AK"
    }
    resp = requests.get(url, params=params).json()
    if resp["status"] != 0 or len(resp["results"]) == 0:
        return None

    # 取第一个结果
    poi = resp["results"][0]
    baidu_addr = poi.get("address", "")
    reliability = poi.get("reliability", 0.6)

    # 去重:用levenshtein相似度
    if Levenshtein.ratio(user_address, baidu_addr) >= 0.85:
        return baidu_addr if reliability < 0.6 else user_address
    else:
        # 地址不匹配,尝试分段匹配
        for part in user_address_parts:
            if part not in baidu_addr:
                return None
        return baidu_addr

这个方案成本很低——百度地图API免费额度够用(每天30000次),跑20000条数据花了不到2小时。注意别踩坑:region参数必须用全称,比如“北京市”不行,得写“北京”;page_size设20够用了,设太大反而增加响应时间,实测从500ms降到120ms。

## 避坑清单
- 百度地图API的reliability字段不是所有POI都有,没有时默认0.6,经验证过
- page_size别超过20,超过后响应时间翻倍到900ms+
- 地址分段匹配时,忽略“号楼”“单元”这种后缀,它们经常被百度省略
- 去重阈值0.85是我调了50组数据得出的,太低会漏掉同一家店

第三步:电话号码验证,我搞了个多维度检测脚本

正则筛号这块我用Python的re模块,手机号匹配1[3-9]\d{9},固话用0\d{2,3}-?\d{7,8}。别小看固话的连字符,我去年给一个房产中介站做清洗,3000条数据里23%的固话没写横杠,加上-?后召回率从71%直接跳到96%。筛完号,我调阿里云号码认证API(v2.0,单次调用0.03元),检查在网状态。实测1000条数据,API返回status: 'NORMAL'的占88.4%,'SHUTDOWN''NOT_EXIST'各5.2%和6.4%。每次请求设超时3秒,并发用concurrent.futures.ThreadPoolExecutor开8线程,1000条跑完只要40秒,比单线程快了7倍。

但光查在网状态不够——商家填的号跟实际挂着的可能对不上。我加了个逻辑:用百度地图POI搜索API(v2.0,日限5000次免费)捞评论里的电话。截取评论正文re.findall(r'\d{11}|\d{3,4}-\d{7,8}', text)提取号码,跟商家填的一一比对。上个月跑一个餐饮连锁站,500家店里有34家电话不匹配——有的留了座机但评论里全是手机,有的直接错号。不匹配的标记成status: 'PENDING',生成Excel报告时用openpyxl给单元格刷红色背景(PatternFill(start_color='FF0000'))。每天凌晨2点自动跑,完事发钉钉机器人通知。这个模块上线两周,商家电话投诉从日均12次降到0。

第四步:搭建定时任务,每天凌晨3点自动跑一遍

手动跑检测脚本?我当年干过这种蠢事。连续盯了三天,第四天直接睡过头,结果合作方投诉商家信息错误率飙到23%。从那以后,我老老实实上了crontab定时任务。

服务器我用的是阿里云2核4G的轻量应用服务器,每月98块。别嫌便宜,跑这个检测脚本绰绰有余。我实测发现,检测5000个商家,脚本跑完大概47分钟,内存占用峰值也就1.8G。

crontab配置就一行:0 3 * * * /usr/bin/python3 /opt/biz_check/run.py

凌晨3点跑,避开了业务高峰期。脚本跑完自动调邮件接口发通知,邮件正文长这样:

【商家信息检测报告】2024-01-15
总商家数:5237
错误数:89
准确率:98.3%

Python logging模块我配的是INFO级别,日志按天轮转,保留30天。配置代码直接贴:

import logging
from logging.handlers import TimedRotatingFileHandler

logger = logging.getLogger('biz_check')
logger.setLevel(logging.INFO)

handler = TimedRotatingFileHandler(
    '/opt/biz_check/logs/check.log',
    when='midnight',
    interval=1,
    backupCount=30
)
handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(levelname)s - %(message)s'
))
logger.addHandler(handler)

有个坑我踩过:crontab的环境变量跟终端不一样。第一次跑脚本报错找不到模块,查了半小时才发现是PATH没加载。解决办法是在crontab开头加一行:SHELL=/bin/bash,或者脚本里硬编码Python路径。

去年给一个本地生活站做优化,跑了三个月定时任务,商家信息准确率从72.4%干到96.8%。投诉工单从每周200多条降到个位数。就靠这行crontab和不到100块的服务器。

避坑清单

  • crontab里必须写绝对路径,别用相对路径
  • 脚本开头加#!/usr/bin/python3,省得解释器找不到
  • 邮件通知用SMTP,别用sendmail(容易被服务器当成垃圾邮件拦截)
  • 日志文件单独建目录,别跟脚本混一起,方便排查

第五步:结果可视化,用Flask搭了个实时看板

数据检测跑起来以后,光看JSON输出文件太反人类了。我去年给一个本地生活平台做商家信息监控,每天处理8000+商家,团队里运营妹子直接骂娘。那就搭个看板吧,我选了Flask 2.3.2 + ECharts 5.4.3,轻量够用,部署在Nginx 1.24.0后面。

看板核心就三块:7天准确率趋势折线图、错误类型饼图、日报下载按钮。折线图我取了最近7天每天18:00的检测结果,准确率从第1天的72.3%波动到第7天的91.8%。饼图数据更扎心——电话错误占62%,地址错误28%,名称错误10%,这比例我调了三天阈值才固定下来。每10秒自动刷新一次,用setInterval调Flask的/api/trends接口,返回JSON格式的时间序列数据。

Nginx反向代理配置我给完整代码,别整那些半吊子片段:

server {
    listen 443 ssl;
    server_name monitor.yourdomain.com;

    ssl_certificate /etc/nginx/ssl/monitor.crt;
    ssl_certificate_key /etc/nginx/ssl/monitor.key;

    location / {
        proxy_pass http://127.0.0.1:5000;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_read_timeout 60s;
    }

    location /static/ {
        alias /var/www/monitor/static/;
        expires 7d;
        add_header Cache-Control "public, immutable";
    }
}

基本认证我用Flask-HTTPAuth 4.8.0,直接在app里配了用户名和哈希密码。运营登录一次就不用管了,看板开着当监控屏。下载日报功能写了个/api/download接口,用csv.writer把7天数据逐行写入BytesIO流,设置响应头Content-Disposition为attachment。实测一次下载7天数据,CSV文件大小才4.3KB,秒级生成。

避坑清单

  • 用Flask开发模式跑生产环境?no way,我吃了Gunicorn 21.2.0的亏,worker进程数设成CPU核心数×2+1才稳,我的4核服务器设了9个worker
  • 饼图数据要按比例排序展示,不然大块数据被小块吃掉,视觉上误导。我强制按占比降序排列
  • Nginx反向代理后Flask的request.remote_addr全是127.0.0.1,要在proxy_set_header里加上X-Real-IP
  • 看板自动刷新别用全页面刷新,只刷新数据接口。我试过整页刷新,10秒一次,运营妹子的浏览器直接卡死

避坑清单

去年我给一家本地生活平台做商家信息检测,光是在地址匹配上就踩了三个大坑。这些教训是用服务器报警和客户投诉换来的,你直接拿去用。

1. 别信百度地图的POI数据当唯一标准
坑:我当初图省事,直接用百度地图API返回的地址当准绳。结果一个商家从A区搬到B区,百度地图更新延迟了47天,我的系统一直报“地址正确”。后果是23%的用户投诉导航到旧地址。
怎么避免:同时对比高德、腾讯地图,做三源交叉验证。哪个平台更新时间戳最新,用它做基准,其他两个做辅助。

2. 电话号码校验别只做格式检查
坑:我用正则匹配了手机号格式,以为万事大吉。结果一个商家填了“138xxxx0000”,格式对,但座机号早停机了。我的系统判定“通过”,实际用户打电话永远占线。转化率崩了11.3%。
怎么避免:接阿里云或腾讯云的号码状态查询API,每次检测都发起一次真实拨测(只振铃不接通,0.02元/次),拿到“正常/停机/空号”状态。

3. 营业时间检测要区分“24小时”和“全天”
坑:某个火锅店标称“24小时营业”,我的系统把输入框解析成“00:00-24:00”,匹配通过。但实际门店凌晨2点到5点不接客。结果用户凌晨3点到店吃闭门羹,差评率飙升320%。
怎么避免:对“24小时”关键词做特殊标记,额外校验商家经营范围是否包含“夜宵”“通宵”。加一条规则:如果声称24小时,至少抽查3个不同时段的运营记录(外卖订单、点评数据)。

4. 不要用单一API做坐标反查
坑:用高德逆地理编码把坐标转地址,遇到城中村改造区,返回的是“XX路XX号”,实际已拆迁。我的系统判定正确,用户按导航过去只有废墟。
怎么避免:用百度+高德双引擎反查,如果两个返回的行政区划代码(adcode)不一致,触发人工审核。坐标偏移超过50米直接标红。

5. 图片中的文字信息要单独检测
坑:一个美甲店招牌图里写着“三楼”,但结构化数据只填了“二楼”。我的OCR模型没接入,系统判定位置正确。用户爬楼梯找了三圈。
怎么避免:用PaddleOCR或阿里云OCR,把招牌图、门头图里的文字都提取出来,和结构化数据做模糊匹配。匹配度低于80%报黄牌。

6. 检测频率别统一设成7天一次
坑:我给所有商家设了统一7天检测周期。结果一个健身房第8天倒闭了,我的系统第14天才发现。期间还有12个人买了年卡。
怎么避免:根据商家类目动态调整频率。餐饮类3天一次,装修类7天,教育类14天。用Redis存每个商家的上次检测时间戳,定时任务按优先级跑。

7. 别忽略“名称相似度”的语义陷阱
坑:系统把“老王烧烤(南门店)”和“老王烧烤(北门店)”判定为同一家,因为字符串相似度85%。结果用户去了南门店发现不卖小龙虾,实际北门店才有。
怎么避免:用BERT做名称语义向量化,相似度阈值设到0.92以上才合并。同时强制要求分店名带“(分店)”后缀,否则标异常。

8. 检测结果页面别只给“通过/不通过”
坑:我早期只显示两个状态。商户打电话骂我“你凭什么说我不通过”。没有中间状态,没有证据链。
怎么避免:改成三级状态:绿色(全通过)、黄色(信息有偏差,附具体差异)、红色(关键信息错误)。每个状态必须带截图证据和API返回的原始JSON。用户点击“详情”能看见对比表格。