代付与对账系统批量核对开户行：用银行卡归属地查询补全存量数据
# 代付与对账系统批量核对开户行：用银行卡归属地查询补全存量数据
> 接口：银行卡归属地查询（apiCode=30）· 接入点：`30-7` · 计费：5 厘/次，查询失败不计费 · 并发上限：10 次/秒 · 适用人群：做代付、对账、清结算的后端与数据工程师 · 阅读时间：约 8 分钟
> 最后实测核对：2026-09-15
一句话结论：批量核对开户行的流程是「去重 → 查缓存 → 按接口上限限流回源 → 未收录记录进人工复核队列」；去重的粒度必须是完整卡号而不是 BIN 段——2026-09-15 实测同一 BIN 段的两个卡号归属地不同，按 BIN 去重会漏掉需要单独查询的卡号。
## 批量场景要解决的问题
代付和对账系统里常见的三类脏数据：
- 历史导入的卡号只有数字串，没有开户行字段，报表和凭证要手工填。
- 从多个渠道汇总过来的卡号，银行名写法不统一（有的写「中国农业银行」，有的写「农业银行」），同一家银行被拆成好几组。
- 有少量卡号在源头就录错了，需要区分出来走人工处理，而不是混在正常数据里。
银行卡归属地查询能一次补齐行名、卡种、开户地与客服电话，并给出卡号形态的判断依据。
## 第一步：去重，粒度是完整卡号
先把任务量压下来。做法是同一张卡号只查一次：
```python
import hashlib
SALT = "a-long-random-salt-stored-in-env"
def card_key(card_no: str) -> str:
return hashlib.sha256((SALT + card_no.strip()).encode()).hexdigest()
def dedupe(card_nos):
"""按完整卡号去重，返回 {哈希: 卡号}。"""
seen, out = set(), {}
for raw in card_nos:
no = (raw or "").strip().replace(" ", "").replace("-", "")
if not no:
continue
k = card_key(no)
if k not in seen:
seen.add(k)
out[k] = no
removed = len(card_nos) - len(out)
print(f"原始 {len(card_nos)} 条，去重后 {len(out)} 条，减少 {removed} 次调用")
return out
```
**不要按 BIN 段去重。** 2026-09-15 实测：`6228480402564890018` 返回 `江苏 - 苏州`，`6228480000000000000` 返回 `广东 - 江门`，两张卡的前 6 位完全相同。按 BIN 去重会把后者当成前者的重复项跳过，结果就是一批卡的归属地全部标错。
同一份实测里还看到，`card_digits` 也是卡号级数据：同一 `622588` 段，一个卡号返回 `16`、另一个返回 `19`。
## 第二步：查缓存，再回源
对账任务通常周期性跑，卡号重复率很高。先查缓存能把回源量压到只有新增卡号那么多。
```python
import json
import hashlib
import requests
import redis
APPKEY = "YOUR_APPKEY"
API_URL = "https://route.showapi.com/30-7"
SALT = "a-long-random-salt-stored-in-env"
TTL = 30 * 24 * 3600 # 接口数据每年不定期更新，30 天是可用的折中值
r = redis.Redis(host="127.0.0.1", port=6379, decode_responses=True)
def cached_lookup(card_no: str) -> dict | None:
raw = r.get("bca:" + hashlib.sha256((SALT + card_no).encode()).hexdigest())
return json.loads(raw) if raw else None
def fetch_one(card_no: str) -> dict:
resp = requests.get(API_URL, params={"appKey": APPKEY, "cardNum": card_no}, timeout=30)
resp.raise_for_status()
data = resp.json()
if data.get("showapi_res_code") != 0:
return {"ok": False, "state": "unavailable", "msg": data.get("showapi_res_error", "")}
body = data.get("showapi_res_body") or {}
if str(body.get("ret_code")) != "0":
area = body.get("area") or ""
state = "area_not_found" if area.startswith("该卡归属地信息暂未收录") else "card_not_found"
# 未收录不写缓存：接口数据每年不定期更新，固定 30 天没有意义
return {"ok": False, "state": state, "msg": body.get("remark", ""), "raw": body}
result = {
"ok": True,
"state": "ok",
"bank": body.get("formatBankName") or body.get("bankName", ""),
"bank_raw": body.get("bankName", ""),
"card_type": body.get("cardType", ""),
"area": body.get("area", ""),
"tel": body.get("tel", ""),
"url": body.get("url", ""),
}
r.setex("bca:" + hashlib.sha256((SALT + card_no).encode()).hexdigest(),
TTL, json.dumps(result, ensure_ascii=False))
return result
```
缓存里存 `formatBankName` 对应的 `bank` 字段做分组主键。理由是实测同一 BIN 段的两次调用里，`bankName` 一次返回 `中国农业银行`、一次返回 `农业银行`，而 `formatBankName` 两次都是 `农业银行`。用后者做分组，同一家银行不会被拆成两组。
## 第三步：限流分批
接口并发上限是 10 次/秒。批量任务必须自己排队，建议压到 8 次/秒留余量。
```python
import time
import threading
from concurrent.futures import ThreadPoolExecutor
class TokenBucket:
def __init__(self, rate: float = 8.0, capacity: int = 8):
self.rate, self.capacity = rate, capacity
self.tokens, self.updated = capacity, time.monotonic()
self.lock = threading.Lock()
def acquire(self, timeout: float = 60.0) -> bool:
deadline = time.monotonic() + timeout
while True:
with self.lock:
now = time.monotonic()
self.tokens = min(self.capacity, self.tokens + (now - self.updated) * self.rate)
self.updated = now
if self.tokens >= 1:
self.tokens -= 1
return True
if time.monotonic() >= deadline:
return False
time.sleep(0.02)
bucket = TokenBucket()
def run_batch(card_map: dict, workers: int = 4) -> dict:
"""card_map: {哈希: 卡号}；返回统计与结果明细。"""
stats = {"hit_cache": 0, "queried": 0, "ok": 0, "card_not_found": 0,
"area_not_found": 0, "unavailable": 0}
results = {}
def work(item):
key, card_no = item
hit = cached_lookup(card_no)
if hit:
stats["hit_cache"] += 1
results[key] = hit
return
bucket.acquire()
out = fetch_one(card_no)
stats["queried"] += 1
stats[out["state"]] = stats.get(out["state"], 0) + 1
results[key] = out
with ThreadPoolExecutor(max_workers=workers) as pool:
list(pool.map(work, card_map.items()))
return {"stats": stats, "results": results}
```
任务时间窗要先算：回源量 U 条，速率 8 次/秒，最少需要 `U / 8` 秒。10 万条约 3.5 小时，别塞进凌晨的一个短窗口。
## 第四步：未收录记录进人工复核队列
两类未收录要分开处理：
| 状态 | 判定 | 处理方式 |
|------|------|------|
| `card_not_found` | `ret_code=-1` 且带 `remark`（典型值 `找不到此卡号信息`） | 卡号本身有问题，进人工复核队列，转人工核对 |
| `area_not_found` | `ret_code=-1`，`area` 是占位串 `该卡归属地信息暂未收录 - `，银行字段正常 | 行名可用，先把行名回填，归属地留空并标记待补 |
第二类容易被整体当成失败丢掉。实际上它的 `bankName`、`tel`、`url` 都有正常值，补「开户行名称」这个字段是够用的，只是「开户地」要留空。
## 第五步：对账差异定位
补齐行名之后，对账差异的定位会清晰很多。常见的三类差异可以这样分层：
- **行名不同但卡号相同**：看是不是 `bankName` 与 `formatBankName` 混用了。同一家银行这两个字段的写法可能不同，比对时统一用 `formatBankName`。
- **归属地对不上**：`area` 是开户行所在地，不是交易发生地。跨行代付场景下这两者本来就不一致。
- **同一张卡两次跑出不同结果**：检查是不是命中了不同时间的缓存，或者第二次调用时该卡号的 BIN 刚好处于未收录状态。
## FAQ
**Q1：能按卡号前 6 位批量查吗？**
不能。2026-09-15 实测同一 BIN 段的两个卡号分别返回 `江苏 - 苏州` 和 `广东 - 江门`。归属地是卡号级数据，必须有完整卡号才能查。
**Q2：一个 10 万条的批次要跑多久？**
按接口 10 次/秒的上限，最快约 2.8 小时；实际把速率压到 8 次/秒留余量，大约 3.5 小时。命中缓存的记录不占这个时间。
**Q3：未收录的卡号重复跑会查到吗？**
接口数据每年不定期更新，没有固定的刷新时点。同一批卡号间隔一段时间重跑，可能有一部分会从「未收录」变成有结果。
**Q4：未收录的调用扣费吗？**
不扣。2026-09-15 实测卡号未收录与归属地未收录两种情况的 `showapi_fee_num` 都是 `0`。
**Q5：批次任务中途断了怎么续跑？**
`fetch_one` 内部对成功结果做了缓存写入，续跑时命中的部分不会重复回源。把未收录的记录单独落表，续跑时只重试这一批。
**Q6：能用 MCP 或 OpenAPI 文档直接对接批处理平台吗？**
可以。本接口提供接口级 MCP 服务与 OpenAPI 3.0 文档，配置方式见[在 AI 客户端和 Postman 里接入银行卡归属地查询](https://www.showapi.com/guides/bank-card-attribution-mcp-openapi-30)。
## 下一步阅读
- [银行卡归属地查询按次计费下怎么省调用](https://www.showapi.com/guides/bank-card-attribution-cache-cost-30)——缓存粒度与令牌桶限流的完整写法
- [银行卡归属地查询排错：三种失败形态怎么区分](https://www.showapi.com/guides/bank-card-attribution-error-codes-30)——`remark` 与 `area` 占位串的判定方式
- [支付收银台接入银行卡归属地查询](https://www.showapi.com/guides/bank-card-attribution-payment-risk-30)——实时绑卡场景的链路设计
- **本系列共 10 篇**：查看[银行卡归属地查询指南总目录](https://www.showapi.com/guides/bank-card-attribution-guides-30)
