最近刚把一个“人车核验加强版”对接项目完整推上线,整个过程踩了不少坑,也沉淀出一些值得记录的方案。这个需求背景不算复杂:平台要做更严格的合规审查,要求司机、车辆、车主三者的绑定关系不能只看表面一致,还要能经得起追溯。技术选型上我们最终确定用 Python 做整套集成,一方面因为风控引擎本身就在 Python 生态里,另一方面这类对接第三方核验服务的活,Python 做胶水集成的效率确实高。
这篇内容更适合正在做人车核验、实名认证类风控对接的开发者看,也可以给准备做合规审查的同学当一份落地参考。核心围绕三件事:核验链路怎么搭、结果怎么翻译成风控决策、以及合规审查必须考虑的留痕与复核机制。
1. 为什么风控系统需要“人车核验”这一环
1.1 先看看谁在做人车核验,他们要解决什么问题
人车核验并不是一个新概念,但它在风控体系里的角色经常被低估。做出行平台的司机准入,需要确认注册司机和车辆登记人是否一致;做货运物流,要验证承运车辆和实际运营者的绑定关系;汽车租赁、二手车交易、甚至车辆抵押贷款,都有类似诉求。
这些场景的共同痛点是:平台的运营主体和运营车辆到底是不是真实、匹配的关系,以及背后有没有隐藏风险。比如一辆车是不是抵押车、锁定车、过户异常车,单纯靠司机上传一张行驶证照片,根本看不出来。传统做法是人工审核,但效率和准确性都跟不上业务量。
人车核验的作用,是在准入或交易环节,通过权威数据源对“人”和“车”做一次事实校验,把真实状态在风控链路里显性化。它解决的不只是“这个人开的车是不是自己的”这个问题,更是为后续的风控决策提供一份可靠的事实底座。
1.2 “加强版”到底强在哪里
市面上一般的个人车辆核验,大多只返回“该车牌是否与某人存在绑定关系”这个结论,粒度很粗。而“加强版”核验通常会在一次调用里返回更多维度,我这次接的加强版主要包括:
- 车主身份信息与车辆登记信息是否一致;
- 车辆当前状态:正常、抵押、锁定、注销等;
- 驾驶证状态和有效性校验;
- 手机号实名信息与车主是否一致;
- 历史绑定变更频率,以及风险标签(比如事故车、泡水车等)。
普通版和加强版的差别,可以用下面这个表格直观感受下:
| 对比维度 | 普通版 | 加强版 |
|---|---|---|
| 核验内容 | 车牌与车主是否绑定 | 多维度组合校验 |
| 返回结果 | 一致/不一致 | 一致/不一致/查无记录/异常状态 |
| 车辆状态 | 不返回 | 正常/抵押/锁定/注销等 |
| 手机号核验 | 不包含 | 可选,校验实名一致性 |
| 风险标签 | 不包含 | 可返回事故、泡水、过户异常等 |
| 风控适用性 | 只能做硬性准入 | 可支撑分层决策 |
加强版对于风控而言价值很大:它不再是一个“是与否”的判断题,而是一个多维度的画像输入,能让风控系统做更细的决策,比如“车辆状态正常但手机号不一致”和“车辆状态抵押且车主不一致”,风险等级完全不同,处理方式也应该不同。
1.3 为什么是 Python 来做这套集成
选型的时候也考虑过让 Java 服务直接对接,毕竟核心交易系统是 Java。但最后集成层仍然放在 Python,有几点原因:
- 风控规则引擎和特征计算服务本身是 Python 写的,人车核验结果直接进特征池,不用跨语言做一次序列化转换;
- 对接第三方核验接口本质上是 HTTP 客户端开发,Python 的 requests 生态和错误的可读性比 Java 的 HttpClient 更顺手;
- 后面要做批量存量复核、清洗核验结果做数据分析,pandas 和各类调度工具链都是现成的;
- Python 在这类“中间对接层”项目里的开发效率,对于人力有限的团队来说优势很大。
2. 集成前的准备工作:依赖选型与数据流设计
2.1 接口对接方式:requests 裸调还是自封装客户端
刚开始很容易图省事,直接在业务代码里用 requests.post 调一把了事,但真这么做,后面会被各种细节磨疯。第三方核验服务通常要求签名、时间戳、请求去重,不同供应商的鉴权方式还不一样,裸调的代码会散落在各个调用方,改一个签名规则要找半天。
我这边的选择是:基于 requests 封装一个轻量客户端,把签名、超时、重试、会话连接池统一收敛到一个类里。用到的第三方库非常少,核心就下面几个:
- requests:HTTP 客户端;
- urllib3 的 Retry:配合 HTTPAdapter 做连接级重试;
- pydantic:对返回结果做结构校验和规范化;
- loguru:日志输出,方便排查接口问题。
依赖少的好处是环境部署不会出幺蛾子,尤其是在容器化的场景里,镜像体积和启动速度都更好控制。
2.2 数据流设计:实时校验和批量复查要分开
人车核验在业务里有两种典型触发方式,必须要分开设计数据流。
一种是实时触发,比如司机首次注册、执行大额订单前、车辆信息变更时。这类场景对延迟敏感,通常需要在请求链路上同步等待核验结果,我的超时控制设在 3 秒,超过就进入兜底逻辑,不能因为核验服务波动把主业务流程拖死。
另一种是批量触发,比如对存量司机池做周期性复核、清理长期未重核的失效数据。这种场景用同步调用就很不划算,我直接做成异步任务:从数据库捞出一批待核验记录,投进队列,由 worker 逐个调核验接口,再批量写回结果。异步任务还方便做限流,避免核验服务方那边因为请求量突增把我们限流掉。
两个链路共用同一个核验客户端,但超时和重试参数不一样。实时链路重试次数更少,异步链路可以多给几次机会,毕竟实时链路用户体验优先,异步链路成功率优先。
2.3 核验记录的落库设计
核验结果不能只拿来即用即弃,必须落库。我设计了一张核验记录总表,核心字段包括业务单号、车牌号、车主身份证哈希、车主姓名、手机号、核验类型、供应商编码、核验状态、原始返回 JSON、规范化结果 JSON、风险得分、最终决策、核验时间。
建表时有几个细节值得注意。身份证号绝不允许明文落库,我存的是哈希值加掩码;车牌号建议单独存一个字段,格式校验在写入前完成,新能源车牌的 6 位字符不能截断;原始返回 JSON 全量保留,方便将来审计定位问题。
表结构大致如下:
CREATE TABLE vehicle_owner_verify_record ( id BIGINT PRIMARY KEY AUTO_INCREMENT, biz_order_no VARCHAR(64) NOT NULL COMMENT '业务单号', plate_no VARCHAR(10) NOT NULL COMMENT '车牌号', owner_id_card_hash CHAR(64) NOT NULL COMMENT '身份证SHA256', owner_name VARCHAR(32) NOT NULL COMMENT '车主姓名', phone_no VARCHAR(11) DEFAULT NULL COMMENT '手机号', verify_type TINYINT NOT NULL COMMENT '1实时/2批量', provider_code VARCHAR(16) NOT NULL COMMENT '供应商编码', verify_status TINYINT NOT NULL COMMENT '1成功/2失败/3超时/4兜底', raw_result JSON DEFAULT NULL COMMENT '供应商原始返回', normalized_result JSON DEFAULT NULL COMMENT '规范化结果', risk_score TINYINT DEFAULT NULL COMMENT '风险得分', final_decision VARCHAR(16) DEFAULT NULL COMMENT 'PASS/REVIEW/REJECT', verify_time DATETIME NOT NULL COMMENT '核验时间', create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_biz_order_no (biz_order_no), KEY idx_plate_no (plate_no), KEY idx_verify_time (verify_time) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='人车核验记录表';3. 核心链路实现:从核验请求到结果解析
3.1 核验客户端封装:签名、超时、重试一个都不能少
核验客户端最核心的是三件事:签名、超时、重试。签名规则各家不一样,但基本都逃不出“参数排序 + 拼接密钥 + 摘要算法”的套路。我遇到的这家要求 MD5 大写,参数按字典序排列,非空参数参与签名。
完整实现如下:
import hashlib import time import logging import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry logger = logging.getLogger(__name__) class PersonVehicleVerifyClient: def __init__(self, app_key, app_secret, base_url, timeout=3): self.app_key = app_key self.app_secret = app_secret self.base_url = base_url.rstrip("/") self.timeout = timeout self.session = requests.Session() retry = Retry( total=2, connect=1, read=1, backoff_factor=0.5, status_forcelist=[500, 502, 503, 504], ) adapter = HTTPAdapter(max_retries=retry) self.session.mount("http://", adapter) self.session.mount("https://", adapter) def _sign(self, params: dict) -> str: sorted_keys = sorted(params.keys()) raw_str = "&".join( f"{k}={params[k]}" for k in sorted_keys if params[k] != "" ) raw_str += self.app_secret return hashlib.md5(raw_str.encode("utf-8")).hexdigest().upper() def verify( self, plate_no: str, owner_name: str, id_card: str, phone_no: str | None = None, vin: str | None = None, ) -> dict: params = { "app_key": self.app_key, "timestamp": str(int(time.time())), "plate_no": plate_no, "owner_name": owner_name, "id_card": id_card, "phone_no": phone_no or "", "vin": vin or "", } params["sign"] = self._sign(params) resp = self.session.post( f"{self.base_url}/api/v1/person-vehicle/verify", data=params, timeout=self.timeout, ) if resp.status_code >= 400: logger.error( "核验请求异常 status=%s body=%s", resp.status_code, resp.text ) resp.raise_for_status() return resp.json()这里面有几个细节我想专门说一下。
第一,签名参数必须用排序后的键名拼接,有些同事图省事直接按参数传入顺序拼,密钥不对排查半天,最后发现是排序问题。第二,超时一定要传,不要依赖默认值,默认是不超时的,一旦供应商接口卡住,整个业务线程池都会被占满。第三,重试次数我设的是 total=2,不要更多,因为核验接口不是所有请求都幂等,重试太多会把一个不该重放的请求重复提交,反而产生脏数据。
3.2 各家供应商返回格式混乱?统一抽象来兜底
做这种对接,最烦的是不同供应商的返回结构五花八门。同样是“核验成功”,有的返回 success=true,有的返回 code=“0000”,有的返回 status=1;车主姓名字段,有的叫 owner_name,有的叫 vehicle_owner;车辆状态有的叫 car_status,有的叫 vehicle_state。
如果业务代码直接消费原始返回,后面接第二家供应商时,所有判断逻辑都要再写一遍。我的做法是定义了一个规范化结果模型,所有供应商的返回在进入业务层之前,统一转换为这个结构。
from pydantic import BaseModel class NormalizedVerifyResult(BaseModel): code: str # SUCCESS / FAIL / UNKNOWN owner_match: str # MATCH / MISMATCH / NO_RECORD vehicle_status: str # NORMAL / LOAN / LOCKED / DELETED owner_phone_match: str | None = None # MATCH / MISMATCH / NO_RECORD risk_tags: list[str] = [] raw_payload: dict = {} @property def is_pass(self) -> bool: return self.code == "SUCCESS" and self.owner_match == "MATCH"为什么单独定义成 pydantic 模型,而不是用一个字典解决?因为字典的字段拼写错误只有运行时才发现,pydantic 在赋值阶段就能校验类型,而且后面接新的供应商时,映射逻辑的好坏可以直接用单元测试覆盖。
3.3 官方字段和业务字段是两码事
对接过程中很容易犯一个错误:把供应商返回字段直接当业务结论用。
举个例子,供应商返回 owner_match=MISMATCH,从业务上看是不是就要拒绝这个司机?不一定。如果这是一家租车公司的司机,车辆本来就是公司登记的,车主不一致反而是正常的业务形态。再比如车辆状态是 LOAN(抵押),对于网约车平台这种重资产公司反而常见,需要看产品策略,而不是一刀切。
所以我坚持把“核验事实”和“风控决策”两层拆开:核验客户端只负责拿到事实,并且确保事实准确;至于这个事实意味着什么风险、该不该拒绝,交给后面的风控规则引擎去判断。这样的分层让代码职责清晰,也让规则可以灵活调整,而不用每次改规则都动底层对接代码。
4. 把核验结果翻译成风控语言:规则引擎设计
4.1 为什么核验结果不能直接用
如果把核验结果直接拿来做通过/拒绝,等于把风控决策权交给了供应商。但供应商返回的只是一个事实快照,它不会考虑你的业务模式、客户分层、历史违约率这些信息。
更现实的问题是,很多核验结果存在“灰色地带”。比如查无记录(NO_RECORD),是查不到这辆车,还是查不到绑定关系,还是数据源还没更新?直接拒绝会误杀一部分真用户,直接通过又放过了潜在风险。这时候风控系统需要的是分层决策,而不是二值判断。
4.2 设计一套可配置的评分规则
我设计的评分方案是加权打分制。把加强版返回的几个核心维度拆开,每个维度根据风险严重程度设置不同分值,然后汇总得到 0 到 100 的风险分。
| 核验维度 | 结果值 | 分值 |
|---|---|---|
| 车主一致性 | MATCH | 0 |
| 车主一致性 | NO_RECORD | 50 |
| 车主一致性 | MISMATCH | 80 |
| 车辆状态 | NORMAL | 0 |
| 车辆状态 | LOAN | 20 |
| 车辆状态 | LOCKED | 40 |
| 车辆状态 | DELETED | 100 |
| 手机号一致性 | MATCH | 0 |
| 手机号一致性 | NO_RECORD | 20 |
| 手机号一致性 | MISMATCH | 40 |
| 风险标签 | 每命中一个 | 15 |
实现上直接用配置字典,不写死在代码里:
VERIFY_RULE_CONFIG = { "owner_match": { "MATCH": 0, "MISMATCH": 80, "NO_RECORD": 50, }, "vehicle_status": { "NORMAL": 0, "LOAN": 20, "LOCKED": 40, "DELETED": 100, }, "owner_phone_match": { "MATCH": 0, "MISMATCH": 40, "NO_RECORD": 20, }, "risk_tag": 15, } def calculate_risk_score(res: NormalizedVerifyResult) -> int: score = 0 score += VERIFY_RULE_CONFIG["owner_match"].get(res.owner_match, 50) score += VERIFY_RULE_CONFIG["vehicle_status"].get(res.vehicle_status, 0) if res.owner_phone_match: score += VERIFY_RULE_CONFIG["owner_phone_match"].get( res.owner_phone_match, 20 ) score += VERIFY_RULE_CONFIG["risk_tag"] * len(res.risk_tags) return min(score, 100) def make_decision(score: int) -> str: if score < 30: return "PASS" if score < 70: return "REVIEW" return "REJECT"分值设计是有些讲究的。车主不一致是最高危的信号,直接给 80 分,基本一票否决;查无记录没有直接判死,给 50 分,进入人工复核;车辆抵押这种状态给 20 分,提示风险但留有余地,因为不同业务对抵押车的容忍度不一样;风险标签单个只给 15 分,但可以累加,目的是体现“多风险标签叠加比单一标签更值得关注”的考量。
决策阈值我定的是 30 分以下通过,30 到 70 分转人工,70 分以上拒绝。这套阈值不是拍脑袋定的,我拿历史三个月的核验数据回放了一遍,调整权重和阈值,让通过率、人工复核率、拒绝率的分布贴近业务预期,才算定下来。
4.3 分数不是万能的:名单、场景例外与二次校验
评分规则解决了大部分常规情况,但业务里总有规则覆盖不到的地方。
我加了一个特殊名单机制。比如平台自己的企业车队,车辆全是公司统一登记的,核验结果是车主不一致,但这种情况其实是合规的,直接拒绝会影响运力。我这边在规则引擎之前先查一遍白名单:车牌号加归属企业 ID 匹配,命中白名单就直接放行,不再走评分。
还有一类是夫妻双方、直系亲属之间过户但未及时更新的情况,车主不一致但实际是家庭共用车辆。这种不能一棍子打死,我把它留在人工复核区,让审核人员结合其他材料判断。
评分是固定的,但决策可以是灵活的,这也是我把规则配置化的原因。权重和阈值随时可以调,不需要改代码、发版,改个配置就能生效。
5. 高可用与异常兜底:真实业务中的那点事
5.1 服务方没返回时,风控流程不能跟着停摆
人车核验属于强依赖外部服务,供应商那边的机房抖动、网络波动、限流策略,都不是我们能控制的。上线前我专门设计了兜底逻辑,目标很简单:核验服务挂了,业务不能被完全卡死。
同步链路里,请求一旦超时或返回 5xx,我不会直接抛异常给上游,而是返回一个 UNKNOWN 状态的规范化结果,让风控决策走 REVIEW 分支,转人工审核队列。为什么不是 REJECT?因为很多情况下核验失败只是暂时的技术问题,不代表用户真的有风险,直接拒绝会造成大量误杀,回头还要处理客诉。
异步批量任务则不一样,失败可以重试。重试用的是带退避的重试策略,退避因子设成 0.5,最多重试 3 次,如果还是失败,把记录标记为“待人工介入”,同时发告警通知到值班群。
5.2 请求、响应、异常日志要分表分存
做合规项目,日志和证据留痕是我最上心的地方。我强烈建议把核验请求、核验响应、核验异常分三类存储,而不是堆在一张表里。
请求表记录发出去的是什么参数,响应表记录供应商返回的是什么内容,异常表记录的是超时、网络错误、解析失败这些非正常情况。分表的好处是:查询审计时互不干扰,定位异常时能直接看清是请求问题、响应问题还是网络问题,而且每类数据的保留策略可以不同。
实际存储时,请求参数里的敏感信息一样要做脱敏处理。身份证号只存哈希和掩码,姓名可以做拼音化或者只保留姓氏加星号,手机号保留前三位后四位。日常排查问题时,这些信息已经够用,不需要明文。
5.3 几个踩过的坑:身份证掩码、车牌格式、时区
列几个我实际踩过的坑,估计你们也会遇到。
身份证号落库,一开始我只做了 SHA256 哈希,但业务上需要展示部分信息,后来加了一个掩码字段,不然每次都要现算。车牌号格式校验别偷懒,新能源车牌是 6 位字符,普通蓝牌是 5 位,我踩过截断的坑,入库之后再去补数,非常痛苦。时间统一用 Asia/Shanghai 时区,供应商返回的时间有的是 UTC,有的是时间戳,不统一的话,审计的时候时间线对不上,是要出大问题的。
另外,生产环境一定要对核验客户端做调用量监控。我加了两个指标:核验接口平均响应时间、核验失败率,一旦失败率超过 5% 持续 5 分钟,触发告警。这套监控在上线第二周就真的抓到过一次供应商接口异常,比业务方先发现,赢得了很宝贵的处理时间。
6. 合规审查不是查一次就完了:留痕与周期性复核
6.1 留痕是合规审查的生命线
合规审查和普通风控最大的区别是,审查结果要能“被追溯”。三个月后、半年后,监管或内部审计来问:这个司机当时为什么被拒绝,依据是什么?你得能完整地还原当时的核验场景。
我的做法是每次核验都记录一个快照,包含三类信息:
- 核验输入:车牌、姓名、身份证哈希、手机号;
- 核验输出:供应商原始返回、规范化结果、风险得分、最终决策;
- 规则版本:当时生效的评分规则版本号。
规则版本号这一点很容易被忽略。风控规则是会迭代的,今天还是 80 分拒绝,下个月可能就改成 70 分拒绝。如果查询审计时只看到“拒绝”这个结果,没有版本信息,根本无法判断这个决策是否合理。我这边每条决策记录都会带上规则版本号,配置修改时版本号递增,这样审计回溯时能准确复现当时的判断逻辑。
6.2 数据会过期,复核机制要跟上
人车核验的数据不是永久有效的。行驶证会过期,驾驶证会被扣分暂扣,车辆可能过户,车主手机号可能换绑。如果平台拿三个月前的核验结果当现在的事实用,合规上是站不住脚的。
我的做法是建立周期性复核机制,对存量司机池每 90 天扫一遍:
- 距离上次核验超过 90 天的记录,批量捞出来重新核验;
- 重核结果会更新到最新状态,同时保留历史核验记录不做覆盖;
- 重核失败不立即清退,先进入人工复核,给用户一个宽限期去补充材料。
周期性复核刚开始跑的时候需要注意节奏,别一次性把全部存量都丢进队列,核验服务方承受不住。我按业务线分批,每天固定处理一批,限流控制在供应商允许的 QPS 范围内,跑了两周多才完成第一轮全量复核。
复核中发现的最常见问题就是司机换了手机号,但平台里的联系方式没更新,导致手机号核验一直不通过。这类情况我单独抽出来,在业务端触发用户更新联系方式的消息提醒,相当于是用数据反向驱动了一轮用户信息治理。
这套人车核验加强版上线到现在跑了一个多月,全链路累计调用量过了百万次。做之前我也以为只是个普通的接口对接,真正做下来才发现,人车核验这类需求,难点从来不在调通接口,而在把一次核验变成一条完整、可追溯的证据链,并且让这个证据链能够支撑风控规则的持续迭代。如果你也在做类似的合规审查需求,希望这篇内容能帮你少走一些弯路。