在线问诊的患者在哪？用 IP 归属地给互联网医院做属地留痕和辅助校验（Python 实现）

今年 2 月，互联网诊疗首诊试点在北京的两家医院启动，新华网的报道里有一句话被同行反复转发：监管要求「执业信息、人员资质、诊疗数据和活动过程全程留痕、可追溯」。加上《互联网诊疗监管细则（试行）》本来就写明的实名制、复诊条件、病历资料留存，互联网医院的技术团队今年普遍在被问同一个问题 —— 患者到底在哪，我们记录得下来吗？

「在哪」这件事，病历里有患者自填的常住地，实名认证有身份证，按说不缺数据。缺的是请求当时的那份客观记录：这条问诊是从哪个网络位置发起的。医生排班调度、代问诊识别、异地就医说明，都要用到它。IP 归属地正好补这一块，它便宜、不需要用户授权弹窗、也不涉及定位权限。

先说清边界：IP 属地只能做辅助留痕，不能当实名认证，更不能拿来拒绝患者。 出差、开流量、用公司代理，都会让属地看起来「不对」，这是正常的。下面这套代码全程按这个定位写。

一、场景：三条线要交叉看

一条问诊记录进来，我们手上有三份信息：患者填的常住地、接诊机构的所在城市、以及这次请求的 IP 属地。把它们摆在一起，能分出四种情况：

• 城市一致 —— 最正常的一类，留痕记一下就行。
• 同省不同市 —— 省内异地就医也很常见，不告警，但记录里要能看出来。
• 不同省 —— 值得在就诊记录上挂一个提示，让医生或客服有机会问一句，而不是直接拦。
• 没有省市 —— 海外 IP 只有国家字段，内网 IP 干脆什么都没有，这两类要分开处理，不能混进「异常」里。

另外还有一类信号是 IP 属地本身给不出来的：同一个 IP 在短时间内被多个患者账号用来问诊。代问诊、黄牛抢号、脚本挂号都有这个特征。这个不用额外接口，把自己库里的 IP 拿出来数一数就行 —— 但前提是每条记录都存了 IP，所以采集这一步必须在请求入口就做掉。

二、方案设计

代码干四件事，全部落在一个脚本里：

1. 采集：请求进来时把客户端 IP 记下来（Nginx 里就是 $remote_addr，前面有代理的话取 X-Forwarded-For 最左边那个可信地址 —— 这是另一个坑，别直接把 XFF 第一段当真）。
2. 查询：调 https://ip9.com.cn/get?ip=<ip>，拿 prov、city、isp、经纬度。
3. 比对：按上一节那四条线给结论，输出 属地一致 / 属地存疑 / 境外或无法定位 / 无法判定，不放行也不拦截，只给结论。
4. 留痕：写成 JSONL，一行一条。这份文件的用途是「事后能查」，所以字段要能自解释：结论、依据、采集时间、数据来源都写上。

几个实现上的取舍：

• 非公网地址和写错的 IP 分开处理。192.168.1.1 是内网，不该报错；999.1.1.1 是用户输入错误，值得单独计数。
• 同一次运行内的 IP 去重。一次批处理跑几千条问诊记录，重复 IP 很常见，免费版 60 次/分钟经不起重复查。
• 限速只在成功后等待。失败请求本来就没占额度，全等一遍纯属浪费。
• 失败一律降级成「无法判定」，绝不因为接口抖动让问诊流程中断。校验只是旁路。

三、完整代码

#!/usr/bin/env python3
"""互联网医院 / 在线问诊的「患者所在地」留痕与辅助校验。

用法:
    python3 asksite_check.py
    python3 asksite_check.py records.jsonl      # 每行一条问诊记录
"""
import json
import sys
import time
import ipaddress
from datetime import datetime, timezone, timedelta
from collections import Counter

import requests

API = "https://ip9.com.cn/get"
CST = timezone(timedelta(hours=8))
SESSION = requests.Session()
_cache = {}          # 同一次运行里同 IP 只查一次
_delay = 1.1         # 免费版 60 次/分钟，留出余量


def now_cst():
    return datetime.now(CST).strftime("%Y-%m-%d %H:%M:%S")


def lookup(ip):
    """查 IP 归属地。返回 (数据字典, 失败原因)；成功时原因为空。"""
    if ip in _cache:
        data, err = _cache[ip]
        return data, err or "cached"

    reason = _classify(ip)
    if reason:
        _cache[ip] = (None, reason)
        return None, reason

    for attempt in range(2):
        try:
            r = SESSION.get(API, params={"ip": ip}, timeout=6,
                            headers={"User-Agent": "asksite-check/1.0"})
        except requests.RequestException as exc:
            return None, "网络异常: %s" % exc

        if r.status_code == 400:            # 非法 IP：接口 HTTP 也是 400
            _cache[ip] = (None, "非法 IP")
            return None, "非法 IP"
        try:
            body = r.json()
        except ValueError:
            return None, "响应不是 JSON"

        ret = body.get("ret")
        if ret == 200:
            data = body.get("data") or {}
            _cache[ip] = (data, "")
            time.sleep(_delay)              # 成功才需要限速
            return data, ""
        if ret == 429:                      # 被限速：退避后重试
            time.sleep(2 * (attempt + 1))
            continue
        _cache[ip] = (None, "接口返回 ret=%s" % ret)
        return None, "接口返回 ret=%s" % ret
    return None, "连续被限速"


def _classify(ip):
    """区分「写错了的 IP」和「内网/保留地址」，两者的处理方式不一样。"""
    try:
        a = ipaddress.ip_address(ip)
    except ValueError:
        return "非法 IP"
    if a.is_private or a.is_loopback or a.is_reserved or a.is_multicast:
        return "非公网地址"
    return ""


def check(rec):
    """一条问诊记录 => 一条留痕结果（字典）。"""
    data, err = lookup(rec.get("ip", ""))
    out = {
        "ts": now_cst(),
        "consult_id": rec.get("consult_id"),
        "patient_id": rec.get("patient_id"),
        "ip": rec.get("ip"),
        "declared_city": rec.get("declared_city"),      # 患者填写的常驻地
        "clinic_city": rec.get("clinic_city"),          # 接诊机构所在城市
        "source": "ip9-free",
    }
    if data is None:
        out.update({"ip_prov": "", "ip_city": "", "isp": "",
                    "verdict": "无法判定", "note": err})
        return out

    prov, city, isp = data.get("prov", ""), data.get("city", ""), data.get("isp", "")
    out.update({
        "ip_prov": prov,
        "ip_city": city,
        "isp": isp,
        "lng": data.get("lng", ""),
        "lat": data.get("lat", ""),
        "big_area": data.get("big_area", ""),
    })

    declared = (rec.get("declared_city") or "").replace("市", "")
    decl_prov = (rec.get("declared_prov") or "").strip()
    same_city = bool(declared) and bool(city) and declared in city
    same_prov = bool(decl_prov) and bool(prov) and decl_prov == prov

    if not prov and not city:
        # 海外 IP 或内网：只有国家，没有省市
        out.update({"verdict": "境外或无法定位",
                    "note": "国家=%s，省市字段为空" % data.get("country", "")})
    elif same_city:
        out.update({"verdict": "属地一致", "note": "IP 属地与患者填写城市一致"})
    elif same_prov:
        out.update({"verdict": "属地一致", "note": "同省不同市，符合就近复诊特征"})
    elif declared:
        out.update({"verdict": "属地存疑",
                    "note": "患者填写 %s，IP 属地 %s %s，需人工确认是否出差/异地就医" % (declared, prov, city)})
    else:
        out.update({"verdict": "仅留痕", "note": "患者未填写常驻地，本次只做记录"})

    if rec.get("clinic_city") and city and rec["clinic_city"].replace("市", "") not in city:
        out["note"] = (out.get("note", "") + "；接诊机构在%s，患者请求来自%s" %
                       (rec["clinic_city"], city)).strip("；")
    return out


def main(records):
    results = [check(r) for r in records]

    print("== 本次问诊请求留痕（%d 条）==" % len(results))
    for r in results:
        print("%-10s %-16s %-8s %-10s %s" % (
            r["patient_id"], r["ip"], r["ip_prov"] + r["ip_city"], r["verdict"], r["note"]))

    # 同一 IP 在短时间内被多个账号使用：代问诊 / 黄牛抢号的常见特征
    ip_count = Counter(r["ip"] for r in results if r["ip"])
    multi = {ip: n for ip, n in ip_count.items() if n > 1}
    print("\n== 同 IP 多次问诊 ==")
    print("无" if not multi else json.dumps(multi, ensure_ascii=False))

    # 落盘：留痕记录本身就是监管检查时要能拿出来的东西
    with open("asksite_audit.jsonl", "w", encoding="utf-8") as f:
        for r in results:
            f.write(json.dumps(r, ensure_ascii=False) + "\n")
    print("\n留痕已写入 asksite_audit.jsonl")

    stat = Counter(r["verdict"] for r in results)
    print("结果分布：%s" % json.dumps(dict(stat), ensure_ascii=False))


if __name__ == "__main__":
    if len(sys.argv) > 1:
        with open(sys.argv[1], encoding="utf-8") as f:
            recs = [json.loads(line) for line in f if line.strip()]
    else:                       # 内置样例：覆盖一致 / 跨省 / 境外 / 内网 / 非法 IP
        recs = [
            {"consult_id": "A1001", "patient_id": "P001", "ip": "114.114.114.114",
             "declared_city": "南京", "declared_prov": "江苏", "clinic_city": "南京"},
            {"consult_id": "A1002", "patient_id": "P002", "ip": "183.204.111.150",
             "declared_city": "郑州", "declared_prov": "河南", "clinic_city": "郑州"},
            {"consult_id": "A1003", "patient_id": "P003", "ip": "220.181.38.148",
             "declared_city": "南京", "declared_prov": "江苏", "clinic_city": "南京"},
            {"consult_id": "A1004", "patient_id": "P004", "ip": "45.153.160.2",
             "declared_city": "南京", "declared_prov": "江苏", "clinic_city": "南京"},
            {"consult_id": "A1005", "patient_id": "P005", "ip": "192.168.1.1",
             "declared_city": "南京", "declared_prov": "江苏", "clinic_city": "南京"},
            {"consult_id": "A1006", "patient_id": "P006", "ip": "999.1.1.1",
             "declared_city": "南京", "declared_prov": "江苏", "clinic_city": "南京"},
            {"consult_id": "A1007", "patient_id": "P007", "ip": "114.114.114.114",
             "declared_city": "南京", "declared_prov": "江苏", "clinic_city": "南京"},
        ]
    main(recs)

四、实测输出

拿七个样例记录跑一遍（覆盖同城一致、同省不同市、跨省、境外、内网、写错的 IP、重复 IP）：

$ python3 asksite_check.py
== 本次问诊请求留痕（7 条）==
P001       114.114.114.114  江苏南京     属地一致       IP 属地与患者填写城市一致
P002       183.204.111.150  河南新乡     属地一致       同省不同市，符合就近复诊特征；接诊机构在郑州，患者请求来自新乡
P003       220.181.38.148   北京北京     属地存疑       患者填写 南京，IP 属地 北京 北京，需人工确认是否出差/异地就医；接诊机构在南京，患者请求来自北京
P004       45.153.160.2              境外或无法定位    国家=捷克，省市字段为空
P005       192.168.1.1               无法判定       非公网地址
P006       999.1.1.1                 无法判定       非法 IP
P007       114.114.114.114  江苏南京     属地一致       IP 属地与患者填写城市一致

== 同 IP 多次问诊 ==
{"114.114.114.114": 2}

留痕已写入 asksite_audit.jsonl
结果分布：{"属地一致": 3, "属地存疑": 1, "境外或无法定位": 1, "无法判定": 2}

real    0m5.484s

七条记录跑完 5.5 秒，去掉限速等待，实际网络时间不到一秒。几个观察：

• 北京那条被判「属地存疑」，不是「异常」。这就是刻意的 —— 结论用词要让运营同事能读、也不会被自动化流程拿去当成拦截理由。真要拦，得有人看一眼再说。
• 捷克那个 IP 返回的是 国家=捷克，prov、city 全空。海外 IP 就是这个形态，所以「没有省市」和「内网地址」必须走两条分支，否则境外患者全被标成内网异常。
• 999.1.1.1 走的是 HTTP 400：接口在 HTTP 状态码上就是 400，body 里 ret=400 且 data 是空数组。代码里先判 r.status_code == 400 就返回，简单可靠。
• 重复 IP 被缓存命中，没有第二次请求，所以七条记录只花了六次调用。

五、上线前要过的几道关

告知与最小化。 留痕这件事要么写进隐私政策，要么在问诊前那页的知情同意里带一句。建议只存「省市 + 结论」，不存原始 IP 的完整日志副本 —— 除非你的风控确实需要，那就把保存期限写清楚。这一条不是可选项，个人信息保护相关的检查会看。

别让校验挡住业务。 接口超时、限速、返回异常，结论一律降级成「无法判定」，问诊流程照走。代码里所有失败分支都返回结果而不是抛异常，就是为了这个。

异地就医是常态，提醒要看对象。 同一个「属地存疑」结论，推给医生（提醒核对病历）和推给患者（要求解释为什么在外地）是完全不同的体验。前一种没人有意见，后一种会被投诉。

要区县和 IP 类型，那是 VIP 字段。 免费版给的是国家/省市/邮编/区号/运营商/经纬度（城市中心点）、long_ip、大区；如果风控上想区分「这个 IP 是家庭宽带还是 IDC 机房」（批量脚本问诊基本都走机房），需要 ip_type，那属于 VIP 版的能力。免费版先把「省市留痕 + 同 IP 多账号」这两件事做扎实，已经能覆盖大多数检查场景。

留痕文件本身要能被查。 asksite_audit.jsonl 这种按天切分的文件，建议至少保留到病历的保存期限，并且记录一份「谁查过这批数据」。监管说的「可追溯」包含这一层。接口和字段说明在官网 https://www.ip9.com.cn，免费版不用注册、不限调用总量，只有 60 次/分钟的单 IP 频率限制，按问诊量算，只要加了缓存基本够用。

上一篇Flink 实时任务怎么关联 IP 归属地维表？异步 IO + 本地缓存给日志打地域标签（Java 实现）下一篇出口 IP 换了城市，平台就提示异地登录？用 Go 写一个出口 IP 归属地漂移监控（单文件、无第三方依赖）