Appearance
网课账号被多人共用怎么查？用 IP 归属地做登录地序列检测（Python 实现）
做在线教育的团队大概率都遇到过这个场面：某个账号的课程完成率很奇怪，视频看完了但作业全是错的；客服查后台发现账号一天里在三个城市登录过。高客单价的长期班、机构买断的账号最容易成为共享对象——一个学员买课，宿舍里五个人轮流看，甚至还有挂在二手平台上转卖的。
防账号共享这件事，技术上有几档方案：设备指纹、人脸核验、并发设备数管控、水印溯源。前两种成本高、体验差，通常只留给最贵的课。IP 归属地属于成本最低的那一档前置信号：不用申请任何权限、不用改造客户端，只要登录和播放心跳带上 IP，就能拿到「这个账号的地理动线」，而且几乎所有方案都会把它当作交叉验证的一环。
这篇文章讲怎么用 IP 归属地搭一个能跑的检测器，重点是把误伤控制住——学生放假回家、宿舍共用一个校园网出口、4G 出口跨省，这些都会被粗糙的规则判成共享。
一、在线教育场景里，什么样的行为值得怀疑
先明确一点：单次登录地不能说明任何问题，有意义的是同一个账号在一段时间里的地理轨迹。
短时间内跨城跳变。 上午 10 点在南京看直播回放，10 点 40 分从西安登录继续看。两地直线距离 800 多公里，坐高铁要三小时，40 分钟到不了。这种「时间上不可能」的跳变，是最硬的共享或转卖证据。
一段时间内城市集合过大。 正常学员的地理范围基本固定在一到两个城市（学校 + 家）。如果 7 天内出现四个以上不同的城市，即使每次都间隔很久，也值得看一眼——除非账号是机构采购的多点共用账号。
同一时段多地在线。 两个人同时用一个账号，一个在郑州做题、一个在成都看课，两个会话的心跳 IP 落在不同城市。这条规则对「同时在线」的判定最直接，但要小心 4G 出口：同一个人手机流量切到 WiFi，出口城市可能变，但不会变成相邻省份以外的城市。
出口来自云厂商机房。 网课的黑产链条里，课件批量下载和录屏转卖常用云服务器跑脚本。免费版的返回字段里没有 IP 类型，但 isp 字段会把腾讯云、阿里云、华为云这类机房出口标出来，命中机房 ISP 且大量拉取视频分片的账号直接进观察名单。更准的做法是 VIP 版的 ip_type（IDC 机房 / ISP 家宽 / BUS 企业），命中 IDC 基本可以定性。
反过来说，下面这些不该当成异常：
- 假期回老家，城市从「武汉」变成「黄冈」——同省相邻，属正常。
- 学校宿舍和教学楼出口是两个不同城市？不太可能，但校园网出口经常显示成学校所在城市甚至是省会，同城范围波动都正常。
- 学生用运营商流量，出口城市和实际城市不一致，这在移动网络里是常态。
二、方案设计
整体分四层，从上到下依次是：采集 → 属地化（带缓存与限速）→ 规则打分 → 处置。
登录 / 每 N 分钟播放心跳
→ 拿到 IP（服务端从请求头里取，别信前端传的）
→ GeoResolver：本地缓存命中就直接用，未命中才调 ip9.com.cn/get（令牌桶限速）
→ 落库：account_id + ip + 省 + 市 + 经纬度 + 时间戳
→ 规则引擎：速度跳变 / 城市集合 / 同时段多地 / 机房出口 → 风险分
→ 处置：低分放行，中分弹二次验证，高分限制并发设备数并通知教务三个工程上的取舍：
属地化一定要有本地缓存。 一个平台的日活用户可能几万，但活跃的 IP 段其实很集中：校园网、三大运营商的地市出口、企业专线。把 IP → 属地缓存 7 天（IP 段变动没那么快），命中率通常能到 70% 以上，剩下的才消耗接口额度。免费版 60 次/分钟的量级，配上缓存足够支撑几千日活的测试期。
限速要做在服务端，不要靠 try/except 硬扛。 用令牌桶把出站查询控制在 55 次/分钟，请求排队而不是失败——超时就当「本次没有位置信号」，绝不能因此拦用户登录。
规则输出的是分数和理由，不是结论。 每条规则给不同的权重，最后按总分给处置建议，理由字符串要照着能解释给客服听的方式写（「10:00 南京 → 10:40 西安，速度 1200km/h」这种），这样教务复核的时候不用回来翻日志。
三、Python 实现
下面这份脚本只依赖标准库（requests 都不需要），可以直接 python3 course_risk.py 跑起来看效果。核心是三个类：限速器、属地解析器、检测器。
python
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""网课账号共享检测：IP 归属地 + 登录地序列规则"""
import json
import math
import sqlite3
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
IP9_API = "https://ip9.com.cn/get"
GEO_TTL = 7 * 86400 # IP → 属地缓存 7 天
API_TIMEOUT = 2.0
RATE_PER_MIN = 55 # 免费版上限 60 次/分钟，留点余量
MAX_SPEED = 900.0 # km/h，超过就认为物理上不可达
IDC_KEYWORDS = ("腾讯云", "阿里云", "华为云", "亚马逊", "数据中心", "IDC")
class TokenBucket:
"""令牌桶：把出站查询压在 RATE_PER_MIN 以内，超出的排队等待"""
def __init__(self, rate_per_min: int):
self.capacity = float(rate_per_min)
self.rate = rate_per_min / 60.0
self.tokens = float(rate_per_min)
self.ts = time.monotonic()
self.lock = threading.Lock()
def take(self) -> None:
while True:
with self.lock:
now = time.monotonic()
self.tokens = min(self.capacity, self.tokens + (now - self.ts) * self.rate)
self.ts = now
if self.tokens >= 1:
self.tokens -= 1
return
wait = (1 - self.tokens) / self.rate
time.sleep(min(wait, 1.0))
class GeoResolver:
"""IP 归属地解析：SQLite 缓存 + 限速，失败返回 None 而不是抛异常"""
def __init__(self, db_path: str = "course_risk.db"):
self.db = sqlite3.connect(db_path, check_same_thread=False)
self.db.execute("CREATE TABLE IF NOT EXISTS ip_geo (ip TEXT PRIMARY KEY, payload TEXT, ts INTEGER)")
self.db.execute(
"CREATE TABLE IF NOT EXISTS login_events ("
" id INTEGER PRIMARY KEY AUTOINCREMENT, account TEXT, ip TEXT,"
" prov TEXT, city TEXT, isp TEXT, lat REAL, lng REAL, event_ts INTEGER)"
)
self.db.commit()
self.bucket = TokenBucket(RATE_PER_MIN)
self.api_calls = 0
def _cached(self, ip: str):
row = self.db.execute("SELECT payload, ts FROM ip_geo WHERE ip = ?", (ip,)).fetchone()
if row and time.time() - row[1] < GEO_TTL:
return json.loads(row[0])
return None
def lookup(self, ip: str):
cached = self._cached(ip)
if cached is not None:
return cached or None # 缓存里的空对象 = 当时没查到
self.bucket.take()
self.api_calls += 1
url = IP9_API + "?ip=" + urllib.parse.quote(ip)
try:
req = urllib.request.Request(url, headers={"User-Agent": "course-risk/1.0"})
with urllib.request.urlopen(req, timeout=API_TIMEOUT) as resp:
body = json.loads(resp.read().decode("utf-8"))
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, OSError):
return None # 超时/网络错误：不写缓存，下次重试
data = body.get("data") or {}
if body.get("ret") != 200 or not data:
# ret=400 IP 非法，ret=429 限速：缓存 10 分钟，避免同一批坏数据反复打接口
self.db.execute("INSERT OR REPLACE INTO ip_geo VALUES (?,?,?)",
(ip, "{}", int(time.time()) - GEO_TTL + 600))
self.db.commit()
return None
geo = {
"country": data.get("country", ""),
"prov": data.get("prov", ""),
"city": data.get("city", ""),
"isp": data.get("isp", ""),
"lat": float(data.get("lat") or 0),
"lng": float(data.get("lng") or 0),
}
self.db.execute("INSERT OR REPLACE INTO ip_geo VALUES (?,?,?)",
(ip, json.dumps(geo, ensure_ascii=False), int(time.time())))
self.db.commit()
return geo
def record(self, account: str, ip: str, event_ts: int = None) -> dict:
"""记录一次登录/心跳，返回带上属地的行数据"""
event_ts = event_ts or int(time.time())
geo = self.lookup(ip) or {}
row = {
"account": account, "ip": ip, "event_ts": event_ts,
"prov": geo.get("prov", ""), "city": geo.get("city", ""),
"isp": geo.get("isp", ""), "lat": geo.get("lat", 0.0), "lng": geo.get("lng", 0.0),
}
self.db.execute(
"INSERT INTO login_events (account, ip, prov, city, isp, lat, lng, event_ts)"
" VALUES (:account,:ip,:prov,:city,:isp,:lat,:lng,:event_ts)", row)
self.db.commit()
return row
def haversine(lat1: float, lng1: float, lat2: float, lng2: float) -> float:
"""两地城市中心点之间的距离（公里）"""
R = 6371.0
p1, p2 = math.radians(lat1), math.radians(lat2)
dp = math.radians(lat2 - lat1)
dl = math.radians(lng2 - lng1)
a = math.sin(dp / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dl / 2) ** 2
return 2 * R * math.asin(math.sqrt(a))
class SharedAccountDetector:
"""规则引擎：全部是「有理由的怀疑」，不直接给封号结论"""
def __init__(self, resolver: GeoResolver, window_days: int = 7):
self.resolver = resolver
self.window_days = window_days
def analyze(self, account: str) -> dict:
since = int(time.time()) - self.window_days * 86400
rows = self.resolver.db.execute(
"SELECT ip, prov, city, isp, lat, lng, event_ts FROM login_events"
" WHERE account = ? AND event_ts >= ? ORDER BY event_ts", (account, since)
).fetchall()
events = [{"ip": r[0], "prov": r[1], "city": r[2], "isp": r[3],
"lat": r[4], "lng": r[5], "ts": r[6]} for r in rows]
score, reasons = 0, []
located = [e for e in events if e["city"]]
# 规则 1：相邻两次登录的地理跳变速度
for prev, cur in zip(located, located[1:]):
if prev["city"] == cur["city"]:
continue
dt_hours = (cur["ts"] - prev["ts"]) / 3600.0
if dt_hours <= 0:
continue
dist = haversine(prev["lat"], prev["lng"], cur["lat"], cur["lng"])
speed = dist / dt_hours
if speed > MAX_SPEED:
score += 40
reasons.append(
f"{time.strftime('%m-%d %H:%M', time.localtime(prev['ts']))} {prev['city']}"
f" → {time.strftime('%H:%M', time.localtime(cur['ts']))} {cur['city']}"
f"，相距 {dist:.0f}km / {dt_hours:.2f}h ≈ {speed:.0f}km/h，物理上不可达"
)
# 规则 2：时间窗内城市集合过大
cities = {e["city"] for e in located}
if len(cities) >= 4:
score += 25
reasons.append(f"{self.window_days} 天内出现 {len(cities)} 个不同城市：{'、'.join(sorted(cities))}")
elif len(cities) == 3:
score += 10
reasons.append(f"{self.window_days} 天内出现 3 个不同城市：{'、'.join(sorted(cities))}")
# 规则 3：同一时段多地在线（10 分钟内两个城市）
for i, e in enumerate(located):
near = [x for x in located[i + 1:] if x["ts"] - e["ts"] <= 600]
other = {x["city"] for x in near} - {e["city"]}
if other:
score += 30
reasons.append(
f"{time.strftime('%m-%d %H:%M', time.localtime(e['ts']))} 前后 10 分钟内"
f"同时出现 {e['city']} 与 {'、'.join(sorted(other))}，疑似多地同时在线"
)
break
# 规则 4：出口是云厂商 / 机房（免费版看 isp 文本，VIP 版可用 ip_type=IDC）
idc = [e for e in events if any(k.lower() in e["isp"].lower() for k in IDC_KEYWORDS)]
if idc:
score += 20
reasons.append(f"{len(idc)} 次会话出口为机房/云厂商（{idc[0]['isp']}），常见于批量拉流脚本")
if len(located) == 0:
reasons.append("该账号近期没有可用的位置信号（接口失败或 IP 无法归属）")
score = min(score, 100)
if score >= 70:
level = "high" # 限制并发设备 + 通知教务人工复核
elif score >= 35:
level = "medium" # 下次登录要求短信二次验证
else:
level = "low"
return {"account": account, "score": score, "level": level,
"events": len(events), "reasons": reasons}
if __name__ == "__main__":
resolver = GeoResolver()
det = SharedAccountDetector(resolver)
# 模拟一个账号的登录序列：南京 → 20 分钟后广州，另有一次云机房出口
now = int(time.time())
demo = [
("stu_10086", "114.114.114.114", now - 3600), # 江苏南京
("stu_10086", "202.96.128.86", now - 2400), # 广东广州
("stu_10086", "119.29.29.29", now - 1900), # 广州，腾讯云出口
]
for account, ip, ts in demo:
row = resolver.record(account, ip, ts)
print(f"记录 {ip:<16} → {row['prov']}{row['city']} / {row['isp']}")
result = det.analyze("stu_10086")
print("\n检测结果：")
print(json.dumps(result, ensure_ascii=False, indent=2))
print(f"\n本次实际调用接口 {resolver.api_calls} 次（其余命中缓存）")跑起来大概是这个输出（实测）：
记录 114.114.114.114 → 江苏南京 / 114DNS
记录 202.96.128.86 → 广东广州 / 中国电信/DNS
记录 119.29.29.29 → 广东广州 / 腾讯云/DNSPod DNS+
检测结果：
{
"account": "stu_10086",
"score": 60,
"level": "medium",
"events": 3,
"reasons": [
"09-15 07:04 南京 → 07:24 广州，相距 1128km / 0.33h ≈ 3385km/h，物理上不可达",
"1 次会话出口为机房/云厂商（腾讯云/DNSPod DNS+），常见于批量拉流脚本"
]
}
本次实际调用接口 3 次（其余命中缓存）在这条轨迹里，南京 → 广州那一跳就吃到了 40 分，机房出口再补 20 分，总分 60 落到「下次登录要短信二次验证」这一档。真实业务里这类样本会更多，因为同一个账号在多地的会话不止两条。
四、落地时最容易被忽略的几件事
阈值必须按自己的用户结构调。 900km/h 这个速度阈值是通用起点，但平台如果有大量留学生（跨境网络出口经常在国内城市和海外之间跳），跨境那一跳会稳定命中规则 1，需要单独给海外账号另设一套阈值，或者对「跨境跳变」只记分不告警。
同省之间的移动不计分。 上面规则只对 city 不同计分，实际上应该再放宽一层：同一个 prov 内的城市变化（武汉 ↔ 黄冈、广州 ↔ 佛山）在业务上就是同城通勤，加进去只会制造噪音。真要计分也从「跨大区」开始，big_area 字段（华东、华南这类）拿来判断跨大区更省事。
共享 ≠ 盗号，处置层级要分开。 账号共享大部分时候是学员自己的行为，盗号是另一回事。共享的处置顺序应该是：先提示、再要求二次验证、最后才是限制并发设备数。一上来就封号，投诉量会很可观。
学校和企业出口 IP 要建白名单。 校园网、培训机构机房、企业专线经常是一个出口后面挂着几百个学生。这类 IP 的属地本来就不精确，规则命中时先查白名单，命中就降权。站内《IP 风控总误伤正常用户》那篇讲共享出口豁免的做法，可以直接套。
合规上要有告知。 用 IP 属地做账号风控，属于个人信息处理，用户协议或隐私政策里写明采集范围（IP、归属地省市）和用途（账号安全），不要写到「可能用于任何目的」这种。同时把属地定位只用到市级：免费接口给的就是省市、运营商、城市中心经纬度这一档，area（区县）在免费版是空的，VIP 版才有。指望用它精确定位到宿舍楼是不现实的，也没必要。
接口失败要有兜底。 上面 lookup() 返回 None 时，record() 仍然会记一行，只是属地字段为空。不要出现「查不到归属地就拦登录」这种逻辑——归属地接口是辅助信号，不是身份认证。
总结
用 IP 归属地抓网课账号共享，核心不是规则多复杂，而是三件事做对：属地化带缓存和限速（省额度也省时间）、规则只输出分数和可解释的理由（教务能复核）、处置分级（提示 → 二次验证 → 限制并发设备，不直接封）。跑通之后可以先只记录、不处置，看一周数据统计各条规则的命中率，再决定打开哪几条。
接口用 IP9 免费版：https://ip9.com.cn/get?ip=<IP> 查指定 IP（IPv4/IPv6 都支持），不带参数返回请求方自己的归属地，返回省市、邮编、区号、运营商、城市中心经纬度、big_area 大区等字段，免注册无需鉴权、60 次/分钟。需要区县、IP 类型（ip_type，判断机房最直接）和 AS 号时再上 VIP 版。官网：https://www.ip9.com.cn