1. 数据清洗里最烦的那类脏数据含汉字的记录怎么批量删做数据清洗的人大概都遇到过这种场景从业务库导出一张 CSV或者从日志表里拉一批记录里面混着一堆本该是纯英文/纯数字的字段结果夹了汉字。比如用户表里 MemberName 字段正常应该是拼音或英文 ID偏偏有人填了「超级女生」「测试账号」这种中文又比如设备日志的 device_id 列本该是SN-2024-001这种格式结果混进了「张三的机器」。这类含汉字的记录如果不清理下游做匹配、做聚合、做导入第三方系统时就会各种报错。手工在 Excel 里筛几千行还行几十万行直接卡死。写 SQL 游标逐行判断我见过用find_regular_expression配合游标删除的写法逻辑没错但性能差、可移植性差换个数据库就得重写。这篇要解决的就是用 Python 写一个可复制的过滤脚本把 CSV 或数据库里含中文字符的记录批量剔除并且用 TaoToken 的统一 Key 把「判断是否含汉字」这一步做得更稳、更可核对。适合谁适合做数据工程、ETL、日志清洗以及需要批量处理表格但不想装一堆重型依赖的同学。核心检索词就是「删除包含汉字的记录」和「Python 过滤脚本」下面直接给能跑的东西。先说清楚思路避免你走弯路。判断一个字符串是否含汉字最朴素的是正则[\u4e00-\u9fa5]这个范围覆盖了绝大多数常用汉字。但实际数据里往往还混着全角标点、日文假名、韩文等所以更稳的写法是把范围放宽到[\u4e00-\u9fff\u3400-\u4dbf\uf900-\ufaff]再叠加全角字符区间[\uff00-\uffef]。这一步用 Python 的re模块就能搞定不需要任何外部服务。那 TaoToken 在这里扮演什么角色它不是用来做正则匹配的而是用来统一管理你在清洗流程里调用大模型能力的 Key。比如你想让模型帮你判断「这条记录到底算不算脏数据」或者清洗完让模型生成一份核对报告这些调用都需要一个稳定的 API 入口。TaoToken 提供统一 Key一个 Key 走通对话、编码、Agent 等场景省得你在多个平台之间来回切换配置。官网入口在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 地址是 https://taotoken.net/api 注意 API 地址不带 UTM 参数。我试过把「正则过滤 模型复核」串成一条流水线先用正则快速剔除明显含汉字的行再把边界情况比如只含一个全角符号的行丢给模型判断。这样既快又准而且整个流程只需要一个 Key。下面从环境准备开始一步步来。2. 前置准备TaoToken 统一 Key 与 Python 环境在写脚本之前先把两件事搞定Python 环境和 TaoToken 的 Key。这两步都不复杂但顺序别搞反否则后面调试会浪费时间。2.1 Python 环境与依赖Python 版本建议 3.9 以上3.10/3.11 都行。依赖只有两个pandas用来读写 CSV 和做行数统计requests用来调 TaoToken 的 API 做模型复核。如果你只做正则过滤其实标准库csvre就够了但 pandas 在处理大表格和输出对比时更省事。安装命令pip install pandas requests如果你用的是虚拟环境先激活再装python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install pandas requests装完可以验证一下python -c import pandas, requests; print(pandas.__version__, requests.__version__)能打印出版本号就说明环境没问题。这里提醒一句pandas 在大文件上内存占用较高如果你的 CSV 超过 500MB建议用chunksize分块读后面脚本里我会给分块版本。2.2 获取 TaoToken 统一 Key打开 https://taotoken.net/api-keys 登录后创建一个 API Key。这个 Key 就是你后面所有模型调用的凭证复制下来保存好别直接写进代码提交到 Git。创建完 Key你还需要知道两件事Base URL 和 Model ID。TaoToken 的 Base URL 是https://taotoken.net/apiModel ID 根据你用的模型填比如做文本判断可以用通用的对话模型。这三个要素——Base URL、Key、Model ID——是后面配置的核心缺一不可。如果你用的是 Claude Code 或 Cline 这类工具配置方式略有不同但本质都是填这三样。注意Key 只显示一次创建后立刻复制。如果丢了就重新生成一个旧的作废。2.3 目录结构建议为了避免路径混乱建议建一个干净的工作目录clean_demo/ ├── data/ │ ├── input.csv │ └── output.csv ├── clean.py └── config.jsonconfig.json放 Key 和 Base URLclean.py放主逻辑data/放输入输出。这样后面改配置不用动代码。3. 可复制配置settings 片段与过滤脚本这一节是核心直接给能复制粘贴的东西。先给配置文件再给 Python 脚本最后给一个分块处理的大文件版本。3.1 config.json 配置片段把下面内容存成config.json路径就是项目根目录下的config.json{ base_url: https://taotoken.net/api, api_key: sk-你的Key粘贴在这里, model_id: 你的模型ID, timeout: 30 }如果你用的是 Claude Code 的 settings 风格可以写成 TOML[taotoken] base_url https://taotoken.net/api api_key sk-你的Key粘贴在这里 model_id 你的模型ID timeout 30两种格式选一种就行脚本里我按 JSON 读。注意base_url结尾不要带斜杠api_key不要有多余空格这两个是后面 401 报错的高频原因。3.2 主过滤脚本 clean.py这个脚本做三件事读 CSV、用正则标记含汉字的行、输出过滤后的文件和行数对比。import json import re import pandas as pd # 读取配置 with open(config.json, r, encodingutf-8) as f: cfg json.load(f) # 汉字 全角字符正则 HANZI_PATTERN re.compile( r[\u4e00-\u9fff\u3400-\u4dbf\uf900-\ufaff\uff00-\uffef] ) def has_hanzi(value) - bool: 判断单元格是否含汉字或全角字符 if value is None: return False return bool(HANZI_PATTERN.search(str(value))) def clean_csv(input_path: str, output_path: str, check_colsNone): df pd.read_csv(input_path, dtypestr, keep_default_naFalse) before len(df) # 没指定列就检查所有列 cols check_cols or df.columns.tolist() # 任意一列命中就标记为脏数据 mask df[cols].apply( lambda row: any(has_hanzi(v) for v in row), axis1 ) dirty df[mask] clean df[~mask] clean.to_csv(output_path, indexFalse, encodingutf-8-sig) print(f过滤前总行数: {before}) print(f命中含汉字行数: {len(dirty)}) print(f过滤后剩余行数: {len(clean)}) print(f输出文件: {output_path}) return before, len(dirty), len(clean) if __name__ __main__: clean_csv(data/input.csv, data/output.csv)几个关键点解释一下。dtypestr保证所有列按字符串读避免数字列被自动转成 int 导致正则匹配异常。keep_default_naFalse让空值保持空字符串不然 pandas 会把空值变成 NaNstr(NaN)会得到nan虽然不含汉字但逻辑上不干净。encodingutf-8-sig是为了 Excel 打开不乱码。check_cols参数允许你只检查特定列。比如你只想删 MemberName 含汉字的行就传check_cols[MemberName]。不传就全列检查适合不确定哪列脏的情况。3.3 大文件分块版本如果 CSV 很大用下面这个分块版本边读边写内存友好import json import re import pandas as pd with open(config.json, r, encodingutf-8) as f: cfg json.load(f) HANZI_PATTERN re.compile( r[\u4e00-\u9fff\u3400-\u4dbf\uf900-\ufaff\uff00-\uffef] ) def has_hanzi(value) - bool: if value is None: return False return bool(HANZI_PATTERN.search(str(value))) def clean_csv_chunked(input_path, output_path, chunksize50000): before dirty_count clean_count 0 first_chunk True for chunk in pd.read_csv( input_path, dtypestr, keep_default_naFalse, chunksizechunksize ): before len(chunk) mask chunk.apply( lambda row: any(has_hanzi(v) for v in row), axis1 ) dirty_count int(mask.sum()) clean_chunk chunk[~mask] clean_count len(clean_chunk) clean_chunk.to_csv( output_path, indexFalse, modew if first_chunk else a, headerfirst_chunk, encodingutf-8-sig, ) first_chunk False print(f过滤前总行数: {before}) print(f命中含汉字行数: {dirty_count}) print(f过滤后剩余行数: {clean_count}) if __name__ __main__: clean_csv_chunked(data/input.csv, data/output.csv)分块版本用modew写第一块之后modea追加header只在第一块写。这样输出文件和一次性读写的效果一致。3.4 数据库场景从 SQL 结果过滤如果你的数据来自数据库先用 SQL 查出结果存成 CSV再跑上面的脚本。或者直接在 Python 里连库读import pandas as pd from sqlalchemy import create_engine engine create_engine(mysqlpymysql://user:passhost:3306/dbname) df pd.read_sql(SELECT MemberID, MemberName FROM Member WHERE MemberID 100000, engine)拿到 df 后复用has_hanzi逻辑即可。注意别在生产库上直接跑 DELETE先导出、过滤、核对再决定是否回写。这是数据清洗的铁律。4. 验证请求跑一次看行数对比配置和脚本都有了现在跑一次确认结果可核对。这一步很重要因为「删了多少行」必须和「预期命中多少行」对得上否则说明正则或列选择有问题。4.1 准备测试数据建一个data/input.csv故意混入含汉字的行MemberID,MemberName,City 100001,alice,Beijing 100002,超级女生,Shanghai 100003,bob,Guangzhou 100004,测试账号,Shenzhen 100005,carol,Hangzhou 100006,张三的机器,Chengdu 100007,dave,Wuhan7 行数据其中第 2、4、6 行含汉字预期过滤后剩 4 行。4.2 运行脚本python clean.py预期输出过滤前总行数: 7 命中含汉字行数: 3 过滤后剩余行数: 4 输出文件: data/output.csv打开data/output.csv应该只剩 alice、bob、carol、dave 四行。如果行数对不上先检查正则是否被转义、列是否选对。4.3 用 TaoToken 做模型复核正则能覆盖绝大多数情况但边界数据比如只含一个全角逗号的行可能误判。这时候可以用 TaoToken 调模型复核。下面是一个复核函数import json import requests with open(config.json, r, encodingutf-8) as f: cfg json.load(f) def model_check(text: str) - str: url f{cfg[base_url]}/v1/chat/completions headers { Authorization: fBearer {cfg[api_key]}, Content-Type: application/json, } payload { model: cfg[model_id], messages: [ {role: system, content: 你是一个数据清洗助手只回答 yes 或 no。}, {role: user, content: f下面这段文本是否包含中文字符只回答 yes 或 no\n{text}}, ], temperature: 0, } resp requests.post(url, headersheaders, jsonpayload, timeoutcfg[timeout]) resp.raise_for_status() data resp.json() return data[choices][0][message][content].strip().lower() if __name__ __main__: print(model_check(超级女生)) print(model_check(alice))跑一下预期输出yes和no。这一步验证了你的 Key、Base URL、Model ID 三件套配置正确。如果这里报错先别急着改过滤脚本先把 API 调通。4.4 结果核对表把正则结果和模型结果对照做个简单表格原始文本正则判断模型判断是否一致超级女生含汉字yes一致alice不含no一致测试账号含汉字yes一致张三的机器含汉字yes一致一致率高说明正则够用不一致的样本单独拎出来分析决定是放宽正则还是加模型复核。这样整个清洗流程就是可解释、可核对的。5. 常见报错排查401、local proxy failed、reading choices跑脚本时最容易在这几个地方卡住我按真实报错逐个说。5.1 401 Unauthorized报错长这样requests.exceptions.HTTPError: 401 Client Error: Unauthorized for url: https://taotoken.net/api/v1/chat/completions原因基本是 Key 不对。检查三处config.json里api_key是否有多余空格或换行Key 是否已过期或被删除请求头是不是Bearer sk-xxx格式Bearer和 Key 之间有一个空格。改完重跑401 一般就没了。5.2 local proxy failed报错类似requests.exceptions.ProxyError: HTTPSConnectionPool(hosttaotoken.net, port443): Max retries exceeded ... local proxy failed这是本地网络代理配置导致的。检查环境变量HTTP_PROXY、HTTPS_PROXY是否指向了一个不可用的地址。临时清掉再跑unset HTTP_PROXY HTTPS_PROXY python clean.pyWindows 上用set HTTP_PROXY清空。注意这里说的是本地环境变量清理不是让你去配什么网络工具纯粹是排除干扰项。5.3 reading choices 报错报错类似KeyError: choices或者TypeError: NoneType object is not subscriptable说明返回的 JSON 里没有choices字段。先打印完整响应看看print(resp.status_code) print(resp.text)常见原因是 Model ID 填错了服务端返回了错误信息而不是正常结果。对照 https://taotoken.net/api-keys 页面确认 Model ID或者用模型对话页面先手动测一次确认模型可用。另外temperature设成 0 有时某些模型不支持改成 0.1 试试。5.4 OAuth 相关报错如果你用的是 Claude Code 或 Cline 这类工具可能遇到 OAuth 报错。这类工具通常需要填 Base URL、Key、Model ID 三件套。以 Claude Code 为例配置里要写Base URL: https://taotoken.net/api API Key: sk-你的Key Model ID: 你的模型ID三个都填对OAuth 流程才能走通。缺一个就会报认证失败。Cline 的 MCP 配置同理Base URL 和 Key 必须成对出现。5.5 行数对不上脚本跑完发现过滤后行数和预期不符。先检查是不是有空行被算进去了keep_default_naFalse下空字符串不含汉字不会被误删。再检查是不是某些列被 pandas 自动转成了数字加dtypestr强制字符串。最后检查正则里的范围是不是写错了\u4e00-\u9fff别写成\u4e00-\u9fa5之外的范围。6. 把清洗流程固定下来统一 Key 的长期用法脚本跑通一次不难难的是长期稳定地用。这里说几个我踩过的坑和固定下来的做法。第一把config.json加进.gitignoreKey 永远不进版本库。团队协作时用环境变量注入import os api_key os.environ.get(TAOTOKEN_API_KEY, cfg[api_key])第二清洗前后都留一份行数日志追加到clean.logimport datetime with open(clean.log, a, encodingutf-8) as f: f.write(f{datetime.datetime.now()} before{before} dirty{dirty} clean{clean}\n)这样出问题能回溯是哪一批数据、哪个时间点清洗的。第三如果清洗任务要长期跑、还要接 Agent 做自动复核建议用 Coding Plan把模型调用额度固定下来避免临时 Key 过期导致流水线中断。入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。第四模型复核只用在边界样本上别全量调模型成本和延迟都扛不住。正则先筛一遍把「疑似但不确认」的行单独存成review.csv再批量调模型。这样既快又省。第五验证动作要固化。每次清洗完自动对比before - dirty clean不相等就报警。这个断言能挡住 90% 的配置错误assert before - dirty clean, 行数核对失败请检查过滤逻辑最后给一个完整的调用示例把正则过滤和模型复核串起来import json import re import requests import pandas as pd with open(config.json, r, encodingutf-8) as f: cfg json.load(f) HANZI_PATTERN re.compile(r[\u4e00-\u9fff\u3400-\u4dbf\uf900-\ufaff\uff00-\uffef]) def has_hanzi(value): return bool(HANZI_PATTERN.search(str(value))) if value is not None else False def model_check(text): url f{cfg[base_url]}/v1/chat/completions headers {Authorization: fBearer {cfg[api_key]}, Content-Type: application/json} payload { model: cfg[model_id], messages: [ {role: system, content: 只回答 yes 或 no。}, {role: user, content: f是否含中文{text}}, ], temperature: 0.1, } r requests.post(url, headersheaders, jsonpayload, timeoutcfg[timeout]) r.raise_for_status() return r.json()[choices][0][message][content].strip().lower() df pd.read_csv(data/input.csv, dtypestr, keep_default_naFalse) before len(df) mask df.apply(lambda row: any(has_hanzi(v) for v in row), axis1) dirty int(mask.sum()) clean before - dirty assert before - dirty clean df[~mask].to_csv(data/output.csv, indexFalse, encodingutf-8-sig) print(fbefore{before} dirty{dirty} clean{clean})这套流程跑下来从 CSV 到数据库、从几行到几十万行都能一次跑通、结果可核对。Key 用 TaoToken 统一管理换模型、换场景都不用改代码结构只改config.json里的 Model ID 就行。