快递平台怎么识别盲发诈骗和刷单？用 IP 归属地给寄件端做风控（Python 实现）

先说一个常见误解：用 IP 归属地查不出你的快递现在在哪。 IP 归属地描述的是「操作这套系统的设备接入网络的位置」，而包裹的位置来自扫描巴枪的 GPS 和物流轨迹，两者是两套数据。你在北京打开 App 查件，IP 显示北京，但包裹可能正躺在郑州的分拣中心。

那 IP 归属地在快递行业还有什么用？答案在下单和揽收这两个环节。收件人被骗的「盲发快递」、商家冲销量用的「空包刷单」、黄牛批量代发的「代购件」，这些异常在操作日志里都会留下 IP 痕迹。这篇用 Python 写一套寄件端的风险研判脚本，重点讲清哪些信号真的有用、哪些看起来很吓人其实是误报。

一、场景问题：三类需要拦的寄件流量

盲发快递。 你没买东西，却收到一条「您的包裹已发出，到付 39 元」的短信。这类包裹里可能是几块钱的小商品，也可能是空的，赚的就是那笔到付运费。寄件人往往是在异地批量下单，收件地址是从某些渠道买来的个人信息。

空包刷单 / 代购刷单。 商家为冲销量，寄一个空信封或者一块砖头。特征很明确：单件货值极低、寄件人集中在少数几个出口、收件地址却铺满全国好几个省。代购的黄牛也类似，人在 A 城，接单后从全国各个城市发货。

工作室批量操作。 一个出口 IP 关联大量寄件手机号，每个号只发几单。这种结构和电商刷单、秒杀薅羊毛是同一套玩法，只是落到了快递场景里。

还有一个必须单独处理的坑：快递员的巴枪和自助快递柜用的是物联网卡，出口 IP 常年显示在运营商总部所在的城市（北京、广州），跟网点实际所在地完全无关。如果风控脚本把这类 IP 也拿来比对城市，一线员工的扫描记录会变成海量误报。下面代码里用 source 字段把这类单据直接排除在地域判定之外。

二、方案设计

查询层。 一个带缓存和限速的 lookup(ip)：成功结果缓存 7 天，失败结果只缓存 60 秒，请求间隔 1 秒（免费版 60 次/分钟）。所有网络异常都在函数内部消化，返回带 err 字段的记录，绝不往上抛。

规则层。 只判断「命中/不命中 + 分值」，处置交给业务。当前规则：

规则	条件	分值
R1	下单 IP 城市 ≠ 揽收城市（境内）	25
R2	下单 IP 归属境外	30
R3	同一出口 IP 关联 ≥4 个寄件手机号	30
R4	同一寄件手机号当天出现 ≥3 个城市	25
R5	货值 ≤20 元 + 到付 + 收件地址分布在 ≥3 个省份	20

豁免层。 三类单据直接跳过地域判定：source=device 的设备单据（巴枪、自助柜）、内网/保留地址、IP 无法解析。豁免只写日志，不加分也不减分。

分档：≥60 分建议暂缓上门揽收并转人工；有分但不到 60 交给客服抽检；0 分正常放行。

三、Python 实现

#!/usr/bin/env python3
"""寄件风控：用 IP 归属地识别盲发诈骗、代购刷单和工作室批量寄件。"""
import json
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from collections import defaultdict

API = "https://ip9.com.cn/get"
CACHE_FILE = "ship_cache.json"
TTL_OK, TTL_FAIL, MIN_INTERVAL = 7 * 86400, 60, 1.0

try:
    with open(CACHE_FILE, "r", encoding="utf-8") as fh:
        CACHE = json.load(fh)
except (OSError, ValueError):
    CACHE = {}

_last_at = [0.0]


def lookup(ip: str) -> dict:
    """查一个 IP 的归属地，带缓存与限速；任何异常都返回 err 字段而不是抛出去。"""
    hit = CACHE.get(ip)
    if hit and time.time() - hit["ts"] < (TTL_OK if hit["ok"] else TTL_FAIL):
        return dict(hit, cached=True)

    rec = {"ts": time.time(), "cached": False, "ok": False, "country": "", "prov": "",
           "city": "", "isp": "", "err": ""}
    try:
        now = time.monotonic()
        if now - _last_at[0] < MIN_INTERVAL:
            time.sleep(MIN_INTERVAL - (now - _last_at[0]))
        _last_at[0] = time.monotonic()

        url = f"{API}?ip={urllib.parse.quote(ip, safe='')}"
        req = urllib.request.Request(url, headers={"User-Agent": "ship-risk/1.0",
                                                   "Accept": "application/json"})
        try:
            with urllib.request.urlopen(req, timeout=6) as resp:
                body = resp.read().decode("utf-8", "replace")
        except urllib.error.HTTPError as http_err:
            # 非法 IP 会直接返回 HTTP 400，body 里还有 ret，要读出来才知道原因
            body = http_err.read().decode("utf-8", "replace")
        data = json.loads(body)
        if data.get("ret") != 200:
            rec["err"] = f"ret={data.get('ret')}"        # 400 非法 IP / 429 限速
        else:
            d = data.get("data") or {}
            rec.update(ok=True, country=d.get("country", ""), prov=d.get("prov", ""),
                       city=d.get("city", ""), isp=d.get("isp", ""))
    except Exception as exc:                              # 超时、DNS、JSON 解析失败都在这里兜住
        rec["err"] = f"{type(exc).__name__}: {exc}"

    CACHE[ip] = rec
    with open(CACHE_FILE, "w", encoding="utf-8") as fh:
        json.dump(CACHE, fh, ensure_ascii=False)
    return dict(rec)


def is_internal(g: dict) -> bool:
    """内网/保留地址：不能拿来做地域校验。"""
    return g["ok"] and (g["country"] == "保留" or g["isp"] == "内网地址")


def is_overseas(g: dict) -> bool:
    return g["ok"] and g["country"] not in ("中国", "保留", "")


def judge(ship: dict, geo: dict, ip_phones: dict, phone_cities: dict) -> dict:
    """规则只回答「命中什么 + 多少分」，怎么处置交给业务。"""
    hits, score = [], 0

    # 巴枪/自助柜等设备发起的单据：物联网卡出口 IP 常年显示运营商总部所在地，
    # 与网点实际城市无关，直接跳过地域判定，只留日志。
    if ship.get("source") == "device":
        return {"score": 0, "hits": [f"设备单据（{geo['isp'] or '未知出口'}），不做地域校验"], "auto": False}

    if is_internal(geo):
        return {"score": 0, "hits": [f"内网/保留地址（{geo['isp']}），不做地域校验"], "auto": False}
    if not geo["ok"]:
        return {"score": 0, "hits": [f"IP 无法定位：{geo['err']}（不参与研判）"], "auto": False}

    # R1 寄件人下单 IP 城市与揽收城市不一致
    if geo["country"] == "中国" and geo["city"] and geo["city"] != ship["pickup_city"]:
        score += 25
        hits.append(f"下单 IP 在{geo['prov']}{geo['city']}，揽收地{ship['pickup_city']}")

    # R2 境外 IP 下境内单
    if is_overseas(geo):
        score += 30
        hits.append(f"下单 IP 归属{geo['country']}，非境内")

    # R3 同一出口 IP 关联多个寄件手机号（工作室/群控）
    n = ip_phones.get(ship["sender_ip"], 1)
    if n >= 4:
        score += 30
        hits.append(f"同一出口 IP 关联 {n} 个寄件手机号")

    # R4 同一寄件手机号短时间出现在多个城市（跨城批量下单）
    cities = phone_cities.get(ship["sender_phone"], set())
    if len(cities) >= 3:
        score += 25
        hits.append(f"手机号 {ship['sender_phone'][-4:]}**** 当天出现 {len(cities)} 个城市")

    # R5 低值 + 到付 + 收件地址分散：盲发件/刷单的典型组合
    if ship["goods_value"] <= 20 and ship["collect"] and ship["dest_spread"] >= 3:
        score += 20
        hits.append(f"低值({ship['goods_value']}元)到付，收件地址分布在 {ship['dest_spread']} 个省份")

    return {"score": score, "hits": hits, "auto": score >= 60}

主流程负责把「同一 IP 关联多少个手机号」「同一手机号出现在几个城市」这两个聚合指标算出来，再逐单研判：

def main(path: str) -> None:
    with open(path, "r", encoding="utf-8") as fh:
        ships = json.load(fh)

    ip_phones = defaultdict(set)
    phone_cities = defaultdict(set)
    for s in ships:
        ip_phones[s["sender_ip"]].add(s["sender_phone"])
        phone_cities[s["sender_phone"]].add(s["pickup_city"])
    ip_phones = {k: len(v) for k, v in ip_phones.items()}

    print(f"待研判运单 {len(ships)} 张，涉及 {len(ip_phones)} 个寄件出口 IP\n")
    flagged = auto = 0
    for s in ships:
        geo = lookup(s["sender_ip"])
        r = judge(s, geo, ip_phones, phone_cities)
        tag = "[缓存]" if geo["cached"] else "[接口]"
        if geo["ok"] and geo["country"] == "中国":
            where = f"{geo['prov']}{geo['city']}"
        elif is_internal(geo):
            where = "内网"
        elif geo["ok"]:
            where = geo["country"]
        else:
            where = "（未定位）"
        print(f"{s['waybill']} {tag} 寄件IP {s['sender_ip']:<16} -> {where:<8} "
              f"揽收 {s['pickup_city']} 寄往 {s['dest_city']} 风险分 {r['score']}")
        for h in r["hits"]:
            print(f"      · {h}")
        if r["score"]:
            flagged += 1
            print("      => 建议人工复核并暂缓上门揽收" if r["auto"]
                  else "      => 转人工抽检（不自动拒单）")
        if r["auto"]:
            auto += 1
        print()
    print(f"命中规则的运单 {flagged} 张，其中建议重点复核 {auto} 张")

运单样例（shipments.json，source 区分「用户自助下单」和「设备/巴枪录入」，dest_spread 是这张单所属批次覆盖的收件省份数）：

[
  {"waybill":"WB-8801","sender_phone":"13800001111","sender_ip":"58.213.1.1","pickup_city":"南京","dest_city":"苏州","goods_value":120,"collect":false,"dest_spread":1,"source":"user"},
  {"waybill":"WB-8802","sender_phone":"18600002222","sender_ip":"45.153.160.2","pickup_city":"广州","dest_city":"成都","goods_value":9.9,"collect":true,"dest_spread":4,"source":"user"},
  {"waybill":"WB-8809","sender_phone":"17700004444","sender_ip":"119.29.29.29","pickup_city":"南京","dest_city":"无锡","goods_value":500,"collect":false,"dest_spread":1,"source":"device"}
]

四、实测输出

13 张单、6 个出口 IP 跑下来（摘录了有代表性的几条）：

待研判运单 13 张，涉及 6 个寄件出口 IP

WB-8801 [接口] 寄件IP 58.213.1.1       -> 江苏南京     揽收 南京 寄往 苏州 风险分 0

WB-8802 [接口] 寄件IP 45.153.160.2     -> 捷克       揽收 广州 寄往 成都 风险分 80
      · 下单 IP 归属捷克，非境内
      · 同一出口 IP 关联 4 个寄件手机号
      · 低值(9.9元)到付，收件地址分布在 4 个省份
      => 建议人工复核并暂缓上门揽收

WB-8805 [缓存] 寄件IP 218.30.1.1       -> 北京北京     揽收 上海 寄往 杭州 风险分 55
      · 下单 IP 在北京北京，揽收地上海
      · 同一出口 IP 关联 5 个寄件手机号
      => 转人工抽检（不自动拒单）

WB-8809 [接口] 寄件IP 119.29.29.29     -> 广东广州     揽收 南京 寄往 无锡 风险分 0
      · 设备单据（腾讯云/DNSPod DNS+），不做地域校验

WB-8810 [接口] 寄件IP 192.168.1.10     -> 内网       揽收 南京 寄往 镇江 风险分 0
      · 内网/保留地址（内网地址），不做地域校验

WB-8811 [接口] 寄件IP 999.1.1.1        -> （未定位）    揽收 南京 寄往 常州 风险分 0
      · IP 无法定位：ret=400（不参与研判）

命中规则的运单 9 张，其中建议重点复核 4 张

首跑 5.4 秒（6 次真实请求，每次间隔 1 秒），全缓存二跑 0.073 秒。几条判定值得解释：

WB-8802 这一组（共 4 张单）拿了 80 分。 4 张单都是境外 IP 下单、4 个不同手机号共用同一个出口、货值都是 9.9 元的到付件、收件地址散在 4 个省。单看任何一条都能解释（有人给海外亲戚寄件、有人在公司网络下单），凑在一起就基本能定性成批量盲发。注意脚本打的是「暂缓上门揽收 + 转人工」，不是自动取消 —— 一旦自动取消的是真实订单，赔付和投诉成本比放过去一单高得多。

WB-8805 到 WB-8808 都是 55 分，卡在阈值下面。 这四张单来自 218.30.1.1 这个北京出口，分别从上海、杭州、武汉、成都揽收，是典型的「一个人替多地下单」结构。但它们的货值正常、不需要到付、也没有到极端分散的收件地址，所以规则只让它们进抽检队列。这条边界就是刻意的：跨城下单在代寄、礼品、异地办公场景里太常见。

WB-8809 是设备单据，被直接豁免。 它的 IP 是广州的云服务出口，网点在南京，按地域规则应该扣 25 分 —— 但快递员的巴枪就长这样。豁免逻辑写在这里以后，一线员工的扫描操作不会再污染风控队列。

WB-8810 和 WB-8811 分别是内网地址和非法 IP。 前者的接口响应是 country=保留、isp=内网地址，后者直接返回 HTTP 400 加 {"ret":400}。这类结果一律不计分、不降级，只记原因。

五、上线前需要定下来的几件事

先跑观察期，再决定要不要拦。 建议前两周只记录不处置，把命中规则的单量、人工复核的确认率统计出来。如果 55 分段里 90% 都是误报，说明阈值太低；如果 80 分段里也有一半是真实订单，说明某条规则本身有问题。

豁免名单比规则更重要。 除了设备单据，至少还要豁免这几类：企业客户批量寄件（合同客户往往在总部统一下单、全国发货）、代客下单的电商 ERP（出口 IP 是服务商机房）、跨境包裹的境内代理。豁免要用白名单机制管理，别硬编码在规则里。

机房 IP 的识别要靠 ip_type，这是 VIP 版字段。 免费版能告诉你 IP 在哪个城市、哪家运营商，但分辨不了家庭宽带和 IDC 机房。工作室批量操作基本都跑在云主机上，出口是机房 IP —— 这个判断需要 ip_type（ISP 家庭 / BUS 企业 / IDC 机房）和 ip_asn，只有 VIP 版提供。先用免费版把行为类规则（多手机号、多城市）跑起来，确认业务里真有这类流量，再考虑上 VIP。

缓存和限速不只是省钱。 快递下单有明显的波峰（晚上八点到十点、大促当天），如果每单都实时查询，免费版 60 次/分钟瞬间打满触发 429，风控就变成了误报机器。缓存 7 天 + 1 秒 1 次的限速能把波峰摊平；单量再大，就把查询挪到下单后的异步任务里，主链路只存 IP 不算分。

合规别忽略。 风控结论存「命中哪条规则、分多少」就够了，原始 IP 属于个人信息，日志里该掩码就掩码（114.114.114.*），研判记录定期清理。这套做法在做数据出境自查、个人信息保护审计时都能直接对上。

最后一句实话：IP 归属地是「便宜、能快速覆盖全量」的那一层筛子，它拦不住精心伪装的惯犯。真要压住盲发和刷单，得把它跟运单数据（收件地址分散度、货值分布）、账号行为（注册时间、历史单量）和举报数据叠在一起看。接口 GET https://ip9.com.cn/get?ip=<ip> 免费版免注册、60 次/分钟/IP，返回 prov、city、isp、big_area、经纬度（城市中心点）等字段；区县、ip_type、ip_asn 是 VIP 版能力，具体口径以官网 https://www.ip9.com.cn 为准。

上一篇Ruby on Rails 怎么获取访客 IP 归属地？用 middleware 给请求打地域标签（含缓存与限流）下一篇二手交易平台怎么防「假同城」？用 IP 归属地校验发布地和聊天属地（PHP 实现）