Appearance
网站境外访问占比怎么统计?用 IP 归属地做一份数据出境自查报告(Python 实现)
做数据出境合规的活儿,最常被问到的一句话是"我们到底有多少境外访问?"——不是问法规条款,是问数字。合规同事手上有一堆文件(安全评估办法、标准合同、负面清单),唯独缺一张能放进材料的统计表:境外访问占全部访问的比例、来源集中在哪些国家、这些访问主要打的是哪些接口。
这些问题的原料其实早就在服务器上躺着:Web 访问日志里的客户端 IP。缺的只是把它变成数字的那一步。
这篇写的不是合规解读,是怎么从日志里算出一份能交差的统计。全文一个 Python 脚本,只用标准库,跑完输出地域分布、境外占比和 CSV 明细。
一、先说口径,口径错了数字全废
写脚本之前有四件事必须定下来,否则算出来的数字自己都不敢用。
第一,内网和保留地址要单独剔除。 日志里混着 Kubernetes 探针(10.0.0.5)、负载均衡回源(192.168.1.1)、文档示例段(203.0.113.9)这类地址。如果脚本只判断"这行有没有 IP",它们会被当成正常访客去查归属地,查出来 country=保留、isp=内网地址,然后混进地域统计。这类地址占比看着不高,但在合规材料里出现"保留"两个字很难解释。
第二,多 IP 的日志行该取哪一个。 很多站点记的是 $http_x_forwarded_for $remote_addr,一行开头就是 1.2.3.4, 10.0.0.1。这时候要取的是第一个公网地址,而不是简单取第一个字段——第一个字段可能是逗号拼的一串,也可能是内网跳板。脚本里的做法是按空白切开、取每个 token 逗号前的部分逐个校验,第一个通过公网校验的才算。
第三,代理和机房 IP 会让境外占比虚高。 实测里有一条 45.153.160.2,查出来是"捷克 / AT&T",看起来像境外用户,实际更像 VPN 或者云主机出口。免费版的 isp 字段能看出些端倪(写着 Google Cloud、Cloudflare 的基本都是机房),但要精确判断是不是 IDC 机房,得用 VIP 版的 ip_type 字段。报告里最好加一句"含代理/机房出口,未剔除",别让审阅的人误读。
第四,境外 IP 通常只到国家。 8.8.8.8 返回 country=美国,prov、city 都是空的,经纬度是美国地理中心点。港澳台不一样:203.198.0.1 会返回 prov=香港、big_area=港澳台。所以聚合的 key 不能写死成"省份",得按"有省取省、没省取国家"的规则回退,否则境外数据全挤进一个"未知"。
还有一个容易说错的边界:境外访问占比统计的是"访问来源",不是"数据是否出境"。数据出境的判定看的是数据内容和接收方,IP 统计只是评估工作里的一个输入项。这句话写进报告开头,能省掉后面不少扯皮。
二、方案设计:五步,全靠去重把额度压下来
流程不复杂,关键在中间的去重 + 落盘缓存:
- 抽 IP:逐行读日志,取第一个公网 IP,跳过私网/保留地址;
- 去重:日志几百万行,独立 IP 通常只有几千个。先收集全部 IP 去重,再批量查,请求数直接从百万级降到几千级;
- 查归属地:循环调
https://ip9.com.cn/get?ip=<IP>,两次请求之间隔 1.1 秒(免费版 60 次/分钟,留点余量);结果写进本地 JSON 缓存,下次跑同一批日志直接命中; - 聚合:按"国家 + 省份"分组,统计请求数、占比、独立 IP 数,境外部分单独汇总;
- 出报告:打印 Markdown 表格 + 写
chujing_report.csv,境外占比超过阈值时用退出码 1 报警,方便挂进定时任务。
为什么不用离线库一次性搞定?两种都行,取舍在别处:离线库没有网络调用、吞吐高,但要自己维护数据更新,且算出来的口径和你给客户看的在线接口可能对不上;在线接口的好处是结果可复现——报告里的每个数字都能用同一个接口当场复核。合规场景里这一点比性能重要。
三、Python 实现
只用标准库:ipaddress 做公网判断,urllib 发请求,collections 做聚合,不需要 pip 装任何东西。
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""跨境访问占比统计:从 Web 日志里算出「境外访问占比」与按国家/地区的分布。
用法: python3 chujing_report.py /var/log/nginx/access.log [--limit 2000] [--alert 0.05]
"""
import argparse, csv, ipaddress, json, os, re, sys, time
from collections import defaultdict
from urllib.parse import urlencode
from urllib.request import Request, urlopen
API = "https://ip9.com.cn/get"
UA = "Mozilla/5.0 (X11; Linux x86_64) geo-report/1.0" # 不带 UA 可能被前置防护挡掉
CACHE_FILE = os.path.expanduser("~/.ip9_geo_cache.json")
IP_RE = re.compile(r"[0-9a-fA-F:.]{3,45}")
CACHE = {}
def pick_ip(line):
"""一行日志里挑第一个公网 IP:兼容 remote_addr 在前,
也兼容 'X-Forwarded-For 在前 + remote_addr 在后' 的格式"""
for tok in line.split()[:4]:
cand = tok.split(",")[0].strip()
if IP_RE.fullmatch(cand) and is_public(cand):
return cand
return None
def is_public(ip):
"""内网、回环、链路本地、保留地址一律不查,免得混进地域统计"""
try:
a = ipaddress.ip_address(ip)
except ValueError:
return False
return not (a.is_private or a.is_loopback or a.is_link_local
or a.is_multicast or a.is_reserved or a.is_unspecified)
def geo(ip, limiter):
"""查归属地:内存缓存 + 落盘缓存 + 限速 + 重试"""
if ip in CACHE:
return CACHE[ip]
limiter()
url = API + "?" + urlencode({"ip": ip})
req = Request(url, headers={"User-Agent": UA})
for attempt in range(3):
try:
with urlopen(req, timeout=5) as r:
body = json.loads(r.read().decode("utf-8"))
break
except Exception as e:
if attempt == 2: # 查不到不阻断整轮统计
CACHE[ip] = {"error": str(e)}
save_cache()
return CACHE[ip]
time.sleep(2 * (attempt + 1))
if body.get("ret") != 200: # 非法 IP 返回 400
info = {"error": "ret=%s" % body.get("ret")}
else:
d = body.get("data") or {}
info = {
"country": d.get("country") or "",
"country_code": (d.get("country_code") or "").lower(),
"prov": d.get("prov") or "",
"isp": d.get("isp") or "",
}
CACHE[ip] = info
save_cache()
return info
def main():
ap = argparse.ArgumentParser()
ap.add_argument("logfile")
ap.add_argument("--limit", type=int, default=0, help="只读日志前 N 行")
ap.add_argument("--alert", type=float, default=0.0, help="境外占比超过该阈值时退出码为 1")
args = ap.parse_args()
load_cache()
last = [0.0]
def limiter(): # 免费版 60 次/分钟
gap = time.time() - last[0]
if gap < 1.1:
time.sleep(1.1 - gap)
last[0] = time.time()
total_hits = skipped = unknown_hits = 0
hits = defaultdict(int) # 地域 -> 请求数
ips_by_region = defaultdict(set) # 地域 -> 独立 IP
isp_counter = defaultdict(int)
with open(args.logfile, errors="ignore") as f:
for i, line in enumerate(f):
if args.limit and i >= args.limit:
break
ip = pick_ip(line)
if not ip:
skipped += 1
continue
total_hits += 1
g = geo(ip, limiter)
if g.get("error"):
unknown_hits += 1
continue
if g["prov"]: # 有省取省
region = "%s %s" % (g["country"], g["prov"])
elif g["country"]: # 境外通常只有国家
region = g["country"]
else:
region = "未知"
hits[region] += 1
ips_by_region[region].add(ip)
if g["isp"]:
isp_counter[(g["country"], g["isp"])] += 1
known = sum(hits.values())
overseas = sum(v for k, v in hits.items() if not k.startswith("中国"))
print("有效请求行 %d,跳过(无公网 IP) %d,查询失败 %d"
% (total_hits, skipped, unknown_hits))
print("%-18s %10s %10s %8s" % ("地域", "请求数", "占比", "独立IP"))
for region, n in sorted(hits.items(), key=lambda x: -x[1]):
print("%-18s %10d %9.2f%% %8d" % (region, n, n / max(known, 1) * 100,
len(ips_by_region[region])))
if known:
print("\n境外访问请求占比: %.2f%%(%d/%d)" % (overseas / known * 100, overseas, known))
with open("chujing_report.csv", "w", newline="") as f:
w = csv.writer(f)
w.writerow(["region", "hits", "share", "unique_ips"])
for region, n in sorted(hits.items(), key=lambda x: -x[1]):
w.writerow([region, n, round(n / max(known, 1), 4), len(ips_by_region[region])])
if args.alert and known and overseas / known > args.alert:
print("!! 境外占比 %.2f%% 超过阈值 %.2f%%" % (overseas / known * 100, args.alert * 100))
sys.exit(1)load_cache / save_cache 两个函数就是往 ~/.ip9_geo_cache.json 读写一份 {ip: {...}},代码略掉了,就是普通的 json.load / json.dump。这份缓存的实际价值在第二次跑批:同一批 IP 不再发请求,几百个 IP 的统计从十分钟缩到两三秒。
四、拿一份构造日志跑一遍
真实生产日志不能贴出来,我用一份 15 行的样例日志跑(含内网 IP、IPv6、文档示例段、重复 IP,故意把边界情况都放进去):
text
101.226.4.6 - - [19/Sep/2026:09:01:02 +0800] "GET /api/ip?ip=1.2.3.4 HTTP/1.1" 200 331 "-" "curl/8.5.0"
10.0.0.5 - - [19/Sep/2026:09:01:07 +0800] "GET /healthz HTTP/1.1" 200 2 "-" "kube-probe/1.29"
8.8.8.8 - - [19/Sep/2026:09:02:11 +0800] "POST /api/ip HTTP/1.1" 200 331 "-" "python-requests/2.32"
203.0.113.9 - - [19/Sep/2026:09:08:41 +0800] "GET /api/ip HTTP/1.1" 200 331 "-" "curl/8.5.0"
240e:ff:e02c:1:0:ff:b0e4:20f - - [19/Sep/2026:09:07:33 +0800] "GET /blog/ HTTP/1.1" 200 22190 "-" "Mozilla/5.0"
45.153.160.2 - - [19/Sep/2026:09:10:19 +0800] "POST /api/ip HTTP/1.1" 200 331 "-" "python-requests/2.32"python3 chujing_report.py sample_access.log --alert 0.3 的输出:
text
有效请求行 12,跳过(无公网 IP) 3,查询失败 0
地域 请求数 占比 独立IP
中国 江苏 3 25.00% 1
中国 上海 2 16.67% 1
美国 1 8.33% 1
中国 河南 1 8.33% 1
澳大利亚 1 8.33% 1
中国 香港 1 8.33% 1
中国 广东 1 8.33% 1
捷克 1 8.33% 1
美国 加利福尼亚州 1 8.33% 1
境外访问请求占比: 33.33%(4/12)
!! 境外占比 33.33% 超过阈值 30.00%整个跑批 10.5 秒,其中 9 个不同的 IP 各花 1.1 秒限速等待。几个能说明问题的细节:
- 15 行日志里 3 行被跳过:
10.0.0.5和192.168.1.1是内网,203.0.113.9落在文档示例段(ipaddress也把它算作保留段)。这些行如果不去掉,报告里会多出两行"保留"。 114.114.114.114出现 3 次,只查了一次,后面两次命中缓存——日志里 IP 的重复率就是这么回事。240e:...这条 IPv6 正常查出来了(广东 / 中国电信),日志里混着 IPv6 不用特殊处理,接口两种协议都支持。8.8.8.8是"美国"没有省份,45.153.160.2是"捷克 / AT&T"——这两个就是报告里要备注"含代理/机房出口"的部分,境外占比 33.33% 这个数字里它们贡献了 2 个点,真实境外用户比例大概率比这低。
五、落地时的几个建议
跑成定时任务,留档做同比。 单次占比的数字意义有限,能看出趋势才有用。每周一凌晨跑一次,CSV 按日期归档(chujing_report_20260919.csv),季度做同比。新增的境外来源国家、突然翻倍的某国访问量,都是更值得关注的变化。
阈值报警别设太紧。 境外占比受爬虫、CDN 回源、公司海外节点访问影响,波动是常态。设一个比基线高一倍的值做报警,比设 5% 这种死线实用。
日志收不过来就先抽样。 一天千万行日志没必要全解析,按行号抽样 1%(或只统计带 POST 的行)足够反映比例,--limit 参数就是干这个的。
报告里把"访问来源"和"数据出境"分开写。 IP 归属地只能说明请求从哪个地区发出,城市级精度、可能被代理污染;真正判断数据是否出境,要看数据内容、接收方和存储位置。统计表放在"评估输入"章节,不要放在"结论"里。
额度不够就升 VIP。 免费版 60 次/分钟够单个网站周报用(去重之后每次都只有几千个 IP)。如果要给整个业务线做日常统计,或者需要区县、ip_type(判断是不是 IDC 机房出口)、ip_asn 这些维度来剔除机房流量,就上 VIP 版——按次计费 0.05 元/千次起,18 万次/分钟的额度,统计类任务基本用不完。
脚本里查归属地用的是 IP9 的免费接口 https://ip9.com.cn/get?ip=<IP>,不传 ip 参数就返回调用方自己的归属地,IPv4/IPv6 都支持,免费版 60 次/分钟、无需注册;返回的 country/country_code/prov/isp/big_area 几个字段正好够做这份统计。字段说明和 VIP 能力在官网 https://www.ip9.com.cn 上。如果关心"记录 IP 本身合不合规",可以看本站《IP 地址是个人信息吗》;要看日志侧的地域分布可视化,看《用IP查询分析网站日志》。