亿企赢客户与商品大类采集：分页、断点续跑与本地数据归档
补充虚构分页数据、JSONL 断点续跑 Demo、两次运行的预期输出和项目阅读顺序。
亿企赢客户与商品大类采集：分页、断点续跑与本地数据归档
这次整理的是我做的一个客户与商品大类采集项目。项目的目标很明确：在已经获得授权的账号和业务范围内，读取客户主表，再按客户关联读取商品管理数据，最后把结果保存成本地 JSON 和 JSONL 文件，方便后续检索、统计和二次处理。
项目中我把客户列表和商品明细拆成两个层次处理。客户是主表，一条客户记录对应一个 customerId；商品是明细表，同一个 customerId 可以对应多条商品记录。这样设计以后，既可以只导出客户，也可以根据客户继续拉取商品，而且中途失败时只需要从未完成的客户继续。
本文示例只适用于本人或所在组织已经获得明确授权的数据。登录验证、访问控制和平台的安全策略应当正常执行；遇到验证码或二次验证时，程序暂停并由人工完成验证，不自动识别或绕过验证。
一、项目文件和最终结果
我把项目拆成了两个主要脚本：
亿企赢大类采集/
├── yqy_kehu.py # 客户列表和客户关联商品
├── yqy_spgl.py # 单个客户、公司组和商品管理流程
├── kehu_all.json # 客户主表
├── kehu_names.json # 客户名称和 customerId 的轻量索引
├── kehu_spgl_all.json # 按客户聚合后的商品结果
├── spgl_progress.jsonl # 逐客户追加的断点文件
└── spgl_<customerId>.json # 单个客户的商品明细
当前本地结果中，客户主表约有 3.3 万条记录，客户名称索引也保持了相同的关联关系。商品结果按客户保存，每条记录包含客户标识、商品名称、税收分类编码、单位、单价、税率和状态等字段。
客户主表中的字段大致如下：
{
"customerId": "客户唯一标识",
"customerName": "客户名称",
"customerNo": "客户编号",
"fullName": "客户全称",
"status": "客户状态",
"levelName": "客户等级",
"industryCategoryName": "行业分类",
"areaName": "所属区域",
"managerName": "负责人",
"address": "地址"
}
商品明细的结构大致如下：
{
"customerId": "所属客户",
"spfwjc": "商品或服务分类",
"xmmc": "项目名称",
"sphfwssflhbbm": "税收分类编码",
"ggxh": "规格型号",
"dw": "单位",
"dj": "不含税单价",
"hsdj": "含税单价",
"slv": "税率",
"used": "是否使用",
"createDate": 0,
"modifyDate": 0
}
二、先理清客户和商品的关系
采集前我先确认了数据关系，而不是把所有返回内容直接拼在一起。客户主表负责保存客户基本信息，商品管理表负责保存该客户下的商品或服务项目，二者通过 customerId 关联。
from collections import defaultdict
customer_map = {
item["customerId"]: item
for item in customers
if item.get("customerId")
}
products_by_customer = defaultdict(list)
for item in products:
customer_id = item.get("customerId")
if customer_id:
products_by_customer[customer_id].append(item)
for customer_id, customer in customer_map.items():
customer["products"] = products_by_customer.get(customer_id, [])
这样做有三个好处：
- 客户没有商品时，也能保留客户记录。
- 商品明细可以单独统计分类、税率和单位。
- 后面重新采集某个客户时，不需要重写整份客户主表。
三、登录态和业务上下文
脚本使用一个长生命周期的 HTTP Session 保存同一次授权访问所需要的状态。Session 的作用是复用连接、保存服务端正常返回的会话信息，并统一设置超时和有限重试。
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
def create_session():
session = requests.Session()
retry = Retry(
total=3,
backoff_factor=1,
status_forcelist=[502, 503, 504],
allowed_methods=frozenset({"GET", "POST"})
)
adapter = HTTPAdapter(max_retries=retry)
session.mount("https://", adapter)
session.mount("http://", adapter)
return session
我把登录、公司组选择、客户列表、商品列表分成不同方法，采集方法不直接处理密码，也不把会话信息写入 JSON。实际使用时，账号信息应该通过交互输入、系统密钥环或环境变量提供，输出文件中只保存业务数据。
如果平台返回验证码、二次确认或风控拦截，我的处理方式是停止当前任务并提示人工验证：
def require_manual_verification(response):
if response.status_code in {401, 403, 412}:
raise RuntimeError("需要人工完成平台验证后再继续")
body = response.json()
if body.get("needVerify") is True:
raise RuntimeError("平台要求人工验证，任务已暂停")
这里不把验证码识别、验证接口调用、Cookie 拼接和设备指纹生成写进公开教程。它们属于登录安全边界，公开后容易被直接复用，也不利于账号和数据安全。
四、Base64、MD5 和真正的加密不是一回事
项目里有两类容易混淆的处理。
第一类是 Base64。它只是把字节转换成文本，方便放进 JSON 或请求字段中，任何人拿到字符串都可以解码。它不提供保密性。
import base64
import json
payload = {
"language": "zh-CN",
"screen": "desktop"
}
encoded = base64.b64encode(
json.dumps(payload, ensure_ascii=False).encode("utf-8")
).decode("ascii")
decoded = json.loads(
base64.b64decode(encoded).decode("utf-8")
)
第二类是 MD5 摘要。项目中对输入口令做 MD5，只代表提交的是摘要值，不代表这是安全的密码加密方案。MD5 是单向摘要，不能还原原文，但它速度很快，也没有现代密码存储需要的盐值和成本控制。真正的账号系统应使用服务端规定的安全认证流程，密码存储应使用带盐的慢哈希算法。
import hashlib
def md5_digest(text: str) -> str:
return hashlib.md5(text.encode("utf-8")).hexdigest()
我在这个项目里没有发现 AES、RSA、WASM 解密或自定义加密算法。商品和客户数据主要是 JSON 返回，字段值也没有再套一层业务加密。因此，分析重点是数据关系、分页边界、状态处理和本地归档，而不是寻找不存在的“解密算法”。
五、客户列表分页
客户列表采用页码和页大小查询。我设置客户页大小为 100，每次请求后检查返回记录数和服务端总数，满足任意一个结束条件就停止：
- 当前页为空；
- 累计数量达到服务端返回的总数；
- 当前页数量小于页大小。
下面是我抽象后的分页代码，具体业务路径由项目配置提供：
def fetch_all_pages(session, request_page, page_size=100):
page_no = 1
result = []
while True:
data = request_page(
session=session,
page_no=page_no,
page_size=page_size
)
records = data.get("records") or data.get("list") or []
total = int(data.get("total") or 0)
if not records:
break
result.extend(records)
print(f"第 {page_no} 页：{len(records)} 条，累计 {len(result)}/{total}")
if total and len(result) >= total:
break
if len(records) < page_size:
break
page_no += 1
return result
实际项目中，客户数据会优先读取已经保存的 kehu_all.json，避免每次运行都重新拉取完整客户主表：
from pathlib import Path
import json
def load_json_if_exists(path):
file_path = Path(path)
if not file_path.exists():
return None
with file_path.open("r", encoding="utf-8") as file:
return json.load(file)
六、商品明细分页
商品管理是客户维度的明细查询。我在项目中将商品页大小设为 200，通过 customerId 查询某一个客户的商品记录。
def fetch_customer_products(request_page, customer_id, page_size=200):
page_index = 1
products = []
while True:
body = request_page(
customer_id=customer_id,
page_index=page_index,
page_size=page_size
)
records = body.get("records") or body.get("list") or []
if not records:
break
products.extend(records)
print(
f"客户 {customer_id} 第 {page_index} 页："
f"{len(records)} 条，累计 {len(products)} 条"
)
if len(records) < page_size:
break
page_index += 1
return products
这里不能只请求第一页。部分客户的商品数量超过一页，如果看到第一页有数据就直接保存，会得到不完整的结果。结束条件必须同时考虑空页和短页；如果服务端提供总数，也要用总数做二次校验。
七、断点续跑：JSONL 比一次性 JSON 更合适
商品采集通常比客户列表耗时更长。为了避免运行到一半因为网络、服务端限流或本机重启而全部重来，我采用 JSONL 格式，每完成一个客户就追加一行。
import json
PROGRESS_FILE = "spgl_progress.jsonl"
def load_done_ids(path=PROGRESS_FILE):
done = set()
try:
with open(path, "r", encoding="utf-8") as file:
for line in file:
line = line.strip()
if not line:
continue
try:
item = json.loads(line)
except json.JSONDecodeError:
continue
customer_id = item.get("customerId")
if customer_id:
done.add(customer_id)
except FileNotFoundError:
pass
return done
def append_result(record, path=PROGRESS_FILE):
with open(path, "a", encoding="utf-8") as file:
file.write(
json.dumps(record, ensure_ascii=False, separators=(",", ":"))
+ "\n"
)
主循环只处理未完成的客户：
done_ids = load_done_ids()
for customer in customers:
customer_id = customer.get("customerId")
if not customer_id or customer_id in done_ids:
continue
try:
products = fetch_customer_products(
request_page=load_product_page,
customer_id=customer_id
)
append_result({
"customerId": customer_id,
"customerName": customer.get("customerName", ""),
"spgl_count": len(products),
"spgl": products
})
done_ids.add(customer_id)
except Exception as exc:
print(f"{customer_id} 处理失败：{exc}")
我选择“一个客户完成后立即写入”，而不是把所有数据先放在内存中。这样即使任务中途退出，已经完成的客户也不会丢失。重新运行时读取 spgl_progress.jsonl，根据已完成的 customerId 跳过对应记录即可。
八、公司组和账套的业务上下文
项目的另一个特点是一个账号可能对应多个公司组。公司组决定当前客户检索范围，部分客户还需要进入对应的账套上下文后才能读取商品管理数据。
我把这一过程抽象成四步：
获取公司组列表
↓
选择或切换当前公司组
↓
按企业名称查询客户
↓
进入对应账套，再读取商品明细
切换公司组时，我会检查服务端返回的业务状态，而不是只看 HTTP 状态码。HTTP 200 只说明请求被服务器接收，业务失败仍然可能通过 JSON 中的错误码或消息体现。
def ensure_success(data):
head = data.get("head") or {}
code = head.get("code")
message = head.get("msg")
if code not in (None, "00000000") and message != "成功":
raise RuntimeError(message or f"业务错误：{code}")
return data.get("body") or {}
企业名称搜索也要考虑同名、简称和分页问题。我的做法是先在当前公司组查询，找不到时再按照允许访问的公司组范围逐组查找；最终保存 customerId、企业全称和账套标识，后续商品查询全部使用唯一标识，不使用名称作为关联键。
九、数据清洗、去重和统计
分页结果保存以后，我还会做一次本地校验。商品明细里最稳定的主键通常是商品记录 ID；如果某些历史数据没有 ID，再使用客户、商品名称、分类编码和单位组成临时键。
def deduplicate_products(products):
unique = {}
for item in products:
record_id = item.get("id") or item.get("cargoId")
if record_id:
key = f"id:{record_id}"
else:
key = "|".join([
str(item.get("customerId") or ""),
str(item.get("xmmc") or ""),
str(item.get("sphfwjc") or ""),
str(item.get("sphfwssflhbbm") or ""),
str(item.get("dw") or "")
])
unique[key] = item
return list(unique.values())
我还会统计每个客户的商品数量，以及商品分类和税率的分布：
from collections import Counter
def product_statistics(products):
category_count = Counter(
item.get("spfwjc") or "未分类"
for item in products
)
tax_rate_count = Counter(
str(item.get("slv") or "未填写")
for item in products
)
return {
"total": len(products),
"category_count": dict(category_count),
"tax_rate_count": dict(tax_rate_count)
}
十、输出文件的选择
我保留三种输出形式。
kehu_all.json 适合保存完整客户主表，便于一次性加载和查看；kehu_names.json 只保留名称和 ID，适合作为搜索索引；spgl_progress.jsonl 适合长任务和断点恢复。
如果后续数据量继续增加，我会把 JSONL 作为原始归档，把清洗后的结果再导入 SQLite。导入时以 customerId 和商品记录 ID 建立索引：
CREATE TABLE IF NOT EXISTS customer (
customer_id TEXT PRIMARY KEY,
customer_name TEXT,
full_name TEXT,
status TEXT,
manager_name TEXT
);
CREATE TABLE IF NOT EXISTS product (
product_id TEXT PRIMARY KEY,
customer_id TEXT NOT NULL,
category_name TEXT,
item_name TEXT,
tax_code TEXT,
unit TEXT,
price TEXT,
tax_rate TEXT
);
CREATE INDEX IF NOT EXISTS idx_product_customer
ON product(customer_id);
十一、运行顺序
我实际运行时按照下面的顺序检查：
- 确认当前账号拥有对应客户数据的访问权限。
- 通过正常登录流程建立会话，遇到人工验证就暂停处理。
- 获取并确认当前公司组。
- 优先加载已有的客户主表，没有文件时再分页获取。
- 读取
spgl_progress.jsonl，得到已经完成的客户集合。 - 逐个客户分页读取商品明细，每完成一个客户立即追加保存。
- 对结果做数量校验、主键去重和字段统计。
- 将 JSONL 原始结果转换为聚合 JSON 或 SQLite。
十二、常见问题
1. 为什么客户数量和商品客户数量不一致？
客户主表是全量客户，部分客户本身没有商品记录，或者当前授权范围不包含对应商品。因此不能用“有商品的客户数”反推客户总数。
2. 为什么最后一页不一定等于页大小？
最后一页通常是短页，也可能因为筛选条件变化提前结束。程序应当接受“短页结束”，同时保留服务端返回总数时的数量校验。
3. JSONL 文件中间有一行损坏怎么办？
读取时对每行单独捕获 JSONDecodeError，跳过损坏行并记录日志，不要因为一行损坏导致整个断点文件无法使用。正式归档前再把有效行重新整理成规范 JSON。
4. 为什么程序收到 200 仍然算失败？
因为 HTTP 层和业务层是两回事。接口可能返回 200，但业务状态码表示权限不足、会话失效、需要人工验证或参数不完整，所以必须同时检查 HTTP 状态和 JSON 业务状态。
5. 为什么不把会话信息保存到结果文件？
Cookie、Token、设备信息和账号状态都属于敏感数据。结果文件只保存客户与商品业务字段，运行日志也不打印完整会话值。任务结束后应清理临时认证文件，并限制输出目录权限。
总结
这个项目的核心不是把请求数量堆上去，而是把客户主表、商品明细、公司组上下文和本地归档拆清楚。客户列表使用页码分页，商品明细使用客户 ID 分页，长任务使用 JSONL 逐条落盘，再通过已完成 ID 实现断点续跑。
在数据量较大的情况下，分页边界、业务状态校验、主键去重和结果持久化比单纯“请求成功”更重要。至于登录验证和访问控制，我只让程序在授权范围内正常工作，遇到平台要求验证时暂停并交给人工处理。
十三、小白从零开始：先跑本地分页 Demo
第一次接触这个项目时，不需要先登录，也不需要请求线上接口。先用虚构数据理解分页和断点文件。
进入项目目录：
cd "/Users/liuweifeng/bnballobj/web_nixiang/亿企赢大类采集"
python3 --version
python3 beginner_local_demo.py
第一次运行会生成 beginner_progress.jsonl，正常输出类似：
完成： c001 1
完成： c002 1
完成： c003 0
完成： c004 1
完成： c005 1
再次运行同一个命令，输出会变成：
跳过已完成： c001
跳过已完成： c002
跳过已完成： c003
跳过已完成： c004
跳过已完成： c005
这就说明断点续跑生效了。打开 beginner_progress.jsonl，每一行都是一个已经完成的客户结果：
{"customerId": "c001", "spgl_count": 1,
"spgl": [{"id": "p001", "xmmc": "测试商品", "slv": 0.01}]}
这个 Demo 使用纯 Python 标准库，不需要安装第三方包。它用 page() 模拟服务端分页，用 page_no 和 page_size 切片，用 JSONL 追加写入结果。
十四、从本地 Demo 过渡到真实项目
本地 Demo 跑通后，我会按下面顺序阅读：
beginner_local_demo.py → 分页、短页和 JSONL
yqy_kehu.py → 客户主表和批量商品
yqy_spgl.py → 单客户、公司组和商品管理
kehu_all.json → 客户字段
kehu_spgl_all.json → 客户与商品关联结果
spgl_progress.jsonl → 长任务断点
小白最容易混淆的是“页码”和“客户 ID”：
page_no或page_index只控制当前客户的第几页；customerId决定当前查询的是哪一个客户；spgl_count是某个客户的商品数量；- JSONL 一行通常对应一个已经完成的客户。
真实项目运行前，先复制一份输出目录并确认授权范围，再运行客户列表读取和商品明细采集。遇到登录验证、权限错误或服务端限流时，停止任务并按平台正常流程处理，不把会话信息写进文章、日志或结果文件。
评论 0