import requests
from typing import Literal
RISK_API = "https://api.ipdatacloud.com/v2/query"
API_KEY = "YOUR_RISK_API_KEY" # ← IP风险画像产品专属 key
RiskAction = Literal["ALLOW", "CHALLENGE", "BLOCK"]
def query_ip_risk(ip: str) -> dict:
"""调 IP数据云风险画像，返回 data.risk 子树；失败抛 RuntimeError"""
resp = requests.get(RISK_API, params={"ip": ip, "key": API_KEY}, timeout=2)
resp.raise_for_status()
body = resp.json()
if body.get("code") != 200:
raise RuntimeError(f"IP风险查询失败: {body.get('message')}")
return body["data"]["risk"]
def risk_decision(risk: dict) -> tuple[RiskAction, str]:
"""
把多维风险数据转成一条风控动作，对应字段含义：
proxy_type -> 是否为代理
risk_score -> 0-100 综合评分
risk_level -> 低/中/高风险
risk_tag[] -> 垃圾注册 / 黄牛 / 薅羊毛 / 高危设备等
mb_rate -> 秒拨概率 0%-100%
real -> 真人概率 0%-100%，越近 0 越像机器
"""
score = int(risk.get("risk_score", 0))
proxy = risk.get("proxy_type", "")
tags = [t["label"] for t in (risk.get("risk_tag") or [])]
mb = float(risk.get("mb_rate", "0").rstrip("%"))
real_p = float(risk.get("real", "100").rstrip("%"))
—— 阻拦档 ——
if score >= 70 or mb > 80 or proxy in ("是",):
return "BLOCK", f"高风险(score={score}, proxy={proxy}, mb={mb}%)"
—— 二次验证档 ——
if proxy or score >= 40 or real_p < 30
or any(t in tags for t in ("scalping", "wool", "fraudulentSignup")):
return "CHALLENGE", f"可疑(proxy={proxy}, tags={tags}, real={real_p}%)"
—— 放行档 ——
return "ALLOW", f"低风险(score={score})"
===== 业务侧调用示例（比如登录 / 下单 / 领券入口）=====
def on_user_login(ip: str, user_id: str):
risk = query_ip_risk(ip)
action, reason = risk_decision(risk)