簡(jiǎn)介面向需要把純文本數(shù)據(jù)整理為結(jié)構(gòu)化文檔的編程學(xué)習(xí)者與數(shù)據(jù)處理人員這份小工具提供從 txt 到 xml 的轉(zhuǎn)換實(shí)現(xiàn)重點(diǎn)解決 txt 缺乏元素層級(jí)、屬性描述和跨系統(tǒng)交換能力的問(wèn)題。壓縮包共 37 個(gè)文件、約 65KB以 C# 源碼、WPF 界面文件、項(xiàng)目解決方案與工程配置以及可執(zhí)行程序?yàn)橹鞑⒏缴倭抠Y源、清單和設(shè)置文件構(gòu)成一個(gè)可查看、編譯或運(yùn)行的完整小程序項(xiàng)目。已有 3445 人學(xué)習(xí)下載說(shuō)明其在文本轉(zhuǎn) XML 這類(lèi)基礎(chǔ)數(shù)據(jù)加工場(chǎng)景中具有一定參考面。讀者可從中了解 txt 行列數(shù)據(jù)的讀取與解析、XML 元素和屬性的映射、文件 I/O、異常處理與調(diào)試測(cè)試思路并依據(jù)自身數(shù)據(jù)布局調(diào)整轉(zhuǎn)換邏輯對(duì)不熟悉 XML 的讀者也能借這個(gè)小項(xiàng)目熟悉結(jié)構(gòu)化數(shù)據(jù)組織方式與不同系統(tǒng)間的數(shù)據(jù)交換流程。1. txt 轉(zhuǎn) xml 到底在轉(zhuǎn)什么先認(rèn)清三種輸入形態(tài)某公司的巡檢數(shù)據(jù)一線用記事本按「一行一條、制表符分隔」記下來(lái)十年攢了幾萬(wàn)份下游報(bào)表系統(tǒng)是早年采購(gòu)的接口只接受 xml。中間這道轉(zhuǎn)換每天早上要人手跑一遍直到有人寫(xiě)了這個(gè)小工具。txt 數(shù)據(jù)轉(zhuǎn)換為 xml 數(shù)據(jù)難點(diǎn)從來(lái)不在「轉(zhuǎn)」這個(gè)動(dòng)作而在于 txt 根本沒(méi)有 schema 可言——同一批文件里字段個(gè)數(shù)、分隔符、空值寫(xiě)法可能每份都不一樣。工具的職責(zé)其實(shí)是替你把「格式約定」這件事釘死解析行與列映射成節(jié)點(diǎn)路徑再寫(xiě)出帶聲明、帶縮進(jìn)、編碼明確的 xml。它適合手上已經(jīng)有成堆結(jié)構(gòu)化文本、又必須對(duì)接只認(rèn) xml 的老系統(tǒng)或第三方接口的人反過(guò)來(lái)如果你的數(shù)據(jù)連穩(wěn)定列結(jié)構(gòu)都沒(méi)有先補(bǔ)規(guī)范比找工具劃算。2. 拆開(kāi)這個(gè)小工具解析層、映射層、寫(xiě)出層把腳本從頭到尾讀一遍會(huì)發(fā)現(xiàn)它是很干凈的三段式解析層負(fù)責(zé)把一行文本切成列表映射層負(fù)責(zé)把列表下標(biāo)翻譯成 xml 節(jié)點(diǎn)路徑寫(xiě)出層負(fù)責(zé)序列化、縮進(jìn)和編碼聲明。三層之間只靠一個(gè)字典配置和一組列表傳遞數(shù)據(jù)沒(méi)有任何一層知道另外兩層的細(xì)節(jié)。這種切法的好處是換數(shù)據(jù)源時(shí)只動(dòng)解析層換下游結(jié)構(gòu)時(shí)只動(dòng)映射配置出問(wèn)題也能一眼定位是哪一層的鍋不用在幾百行里猜。我見(jiàn)過(guò)有人偷懶直接寫(xiě)字符串模板recordid{}/id/record.format(row[0])樣例數(shù)據(jù)跑得飛快正式數(shù)據(jù)一上來(lái)就崩。原因是模板法把三層的邊界全抹平了字段里的、沒(méi)轉(zhuǎn)義空值沒(méi)兜底節(jié)點(diǎn)層級(jí)寫(xiě)死在字符串里改一次要全文搜索。所以哪怕只是個(gè)小工具也值得按層次切開(kāi)后面加字段、加校驗(yàn)、加批量處理都是順手的事。2.1 三種 txt 形態(tài)與對(duì)應(yīng)的解析策略在動(dòng)手寫(xiě)解析器之前先給手上的 txt 分個(gè)類(lèi)這決定了后面用什么讀法。日常遇到的絕大多數(shù)文本逃不出下面三種形態(tài)混著出現(xiàn)的也有通常頭部是鍵值信息、正文是行式數(shù)據(jù)。形態(tài)典型特征推薦解析方式高頻易錯(cuò)點(diǎn)分隔符行式每行列數(shù)一致csv / tsv / 豎線 / 分號(hào)line.split(delim, maxsplit)分隔符被表格軟件換成空格定寬列式主frame程序按字節(jié)位置對(duì)齊肉眼看著像空格分開(kāi)按固定下標(biāo)切片line[a:b]中文字符占位寬度與字節(jié)不一致鍵值塊式keyvalue成段空行分段逐行匹配kv空行切記錄值里本身含有號(hào)混合式前若干行是元信息之后才是數(shù)據(jù)表先定位表頭行再按行式處理表頭行位置不固定分隔符行式最省心前提是分隔符真的是你以為的那個(gè)。從 Excel 復(fù)制粘貼出來(lái)的「制表符分隔」經(jīng)常已經(jīng)變成四個(gè)空格肉眼看不出差別split(\t)只會(huì)返回一個(gè)元素。所以工具里加了一個(gè)分隔符嗅探拿前 N 行做投票只有每行列數(shù)完全一致且大于 1 才算候選候選里按「制表符優(yōu)先、空格最后」的順序取。CANDIDATE_DELIMS [\t, |, ,, ;, ] def sniff_delimiter(sample_lines, min_cols2): 用樣本行給候選分隔符投票返回列數(shù)最多的那個(gè)無(wú)法判斷返回 None scores {} for d in CANDIDATE_DELIMS: cols [len(line.split(d)) for line in sample_lines if line.strip()] if not cols: continue # 列數(shù)必須穩(wěn)定且不少于 min_cols否則視為誤判 if len(set(cols)) 1 and cols[0] min_cols: scores[d] cols[0] if not scores: return None return max(scores, keyscores.get)CANDIDATE_DELIMS的順序是有意排的空格最容易被誤判成合法分隔符放最后制表符最可靠放第一。min_cols默認(rèn) 2如果你的數(shù)據(jù)只有一列整行當(dāng)一個(gè)字段嗅探必然失敗這時(shí)候要允許人工強(qiáng)制指定--delimiter。函數(shù)返回None不是異常而是給上層一個(gè)明確信號(hào)這文件得人工看一眼別硬猜。定寬列式是另一個(gè)坑。有些老設(shè)備的輸出按固定字節(jié)寬度對(duì)齊日志里看著整整齊齊實(shí)際是中文字符占了兩到三個(gè)字節(jié)寬度按字符下標(biāo)切片全錯(cuò)位。遇到這種先把文件按utf-8讀進(jìn)來(lái)用len(line.encode(utf-8))核對(duì)寬度實(shí)在不行就退回「按連續(xù)空格壓縮成單分隔符」的寬松模式寧可丟一點(diǎn)精度也別靜默錯(cuò)位。鍵值塊式通常用于配置類(lèi)數(shù)據(jù)keyvalue一行一個(gè)空行表示一條記錄結(jié)束。解析時(shí)要注意值里可能含所以split(, 1)而不是split()?？招星杏涗浿筮€要丟掉純空白的塊否則最后一條記錄后面跟著的換行會(huì)多生成一個(gè)空節(jié)點(diǎn)。2.2 字段映射把列號(hào)寫(xiě)進(jìn)配置而不是寫(xiě)進(jìn)代碼映射關(guān)系一旦硬編碼進(jìn)代碼換一個(gè)數(shù)據(jù)源就得改腳本改完還要重新測(cè)一遍成本比想象中高。所以這個(gè)工具把字段映射抽成一份 JSON 配置腳本只認(rèn)配置、不認(rèn)業(yè)務(wù)。{ source: {delimiter: \t, encoding: utf-8-sig, has_header: true}, root: records, record: record, fields: [ {index: 0, path: meta/id, type: str, required: true}, {index: 1, path: meta/time, type: datetime, format: %Y-%m-%d %H:%M:%S}, {index: 2, path: payload/name, type: str, default: }, {index: 3, path: payload/value,type: float, default: 0} ] }這份配置表達(dá)的信息量其實(shí)不小source.delimiter為null時(shí)啟用自動(dòng)嗅探has_header決定第一條有效行是表頭還是數(shù)據(jù)path用斜杠表示節(jié)點(diǎn)層級(jí)父節(jié)點(diǎn)不存在就自動(dòng)創(chuàng)建。required的字段取值為空直接拋錯(cuò)default的字段則靜默補(bǔ)默認(rèn)值——兩者的區(qū)別是「數(shù)據(jù)壞了」和「數(shù)據(jù)缺了」處理策略不該一樣。def build_record(record_el, row, fields): for spec in fields: idx spec[index] raw row[idx].strip() if idx len(row) else if raw and spec.get(required): raise ValueError(必填字段為空: spec[path]) text convert(raw, spec) # 類(lèi)型轉(zhuǎn)換見(jiàn) 4.3 node record_el for part in spec[path].split(/)[:-1]: child node.find(part) # 顯式判空不要寫(xiě) node.find(p) or SubElement(...) if child is None: child ET.SubElement(node, part) node child ET.SubElement(node, spec[path].split(/)[-1]).text text這里有一行注釋值得單獨(dú)說(shuō)不要用node.find(part) or ET.SubElement(node, part)這種簡(jiǎn)寫(xiě)。Element 對(duì)象的布爾值取決于它有沒(méi)有子元素一個(gè)已經(jīng)存在但沒(méi)有子節(jié)點(diǎn)的meta會(huì)被判成False于是重復(fù)創(chuàng)建同名節(jié)點(diǎn)最后 xml 里出現(xiàn)兩個(gè)meta下游用find()只能拿到第一個(gè)值全丟。這是標(biāo)準(zhǔn)庫(kù)留給所有人的一個(gè)坑寫(xiě)成顯式判空雖然多兩行但省掉的排查時(shí)間遠(yuǎn)不止兩行。index用的是列下標(biāo)而不是表頭名看起來(lái)不夠優(yōu)雅但對(duì)老數(shù)據(jù)更穩(wěn)——很多 txt 的表頭是中文、帶空格、甚至同名重復(fù)按下標(biāo)定位反而不會(huì)錯(cuò)。如果數(shù)據(jù)確實(shí)規(guī)范可以在配置里加一個(gè)可選name字段加載時(shí)用表頭行反查下標(biāo)兩種方式并存。2.3 XML 寫(xiě)出聲明、縮進(jìn)、編碼三件事寫(xiě)出層看著最簡(jiǎn)單實(shí)際上是最容易「本地跑得好、交給下游就報(bào)錯(cuò)」的一環(huán)。要點(diǎn)就三個(gè)xml 聲明里的 encoding 必須和實(shí)際寫(xiě)盤(pán)的編碼一致縮進(jìn)只影響可讀性但寫(xiě)出來(lái)好看與否直接影響人工核對(duì)效率空節(jié)點(diǎn)的寫(xiě)法x /和x/x在部分老解析器里行為不同要顯式定死。def write_xml(tree, out_path, encodingutf-8): 序列化并落盤(pán)encoding 同時(shí)決定文件編碼和聲明內(nèi)容 try: ET.indent(tree, space ) # Python 3.9 才有低版本走下面的兜底 except AttributeError: _manual_indent(tree.getroot()) tree.write( out_path, encodingencoding, # 與聲明里的 encoding 屬性保持一致 xml_declarationTrue, # 輸出 ?xml version1.0 encodingutf-8? short_empty_elementsTrue # 空節(jié)點(diǎn)寫(xiě)成 x /而不是 x/x )xml_declarationTrue不加的話某些只吃嚴(yán)格 xml 的系統(tǒng)會(huì)直接拒收encoding傳utf-8時(shí) ElementTree 不會(huì)額外寫(xiě) BOM這一點(diǎn)對(duì)下游比對(duì)友好省得文件頭多出三個(gè)不可見(jiàn)字節(jié)導(dǎo)致哈希不一致。ET.indent是 3.9 才進(jìn)標(biāo)準(zhǔn)庫(kù)的如果運(yùn)行環(huán)境版本不確定就保留那個(gè)try/except兜底否則在舊機(jī)器上會(huì)直接AttributeError而這行報(bào)錯(cuò)出現(xiàn)在寫(xiě)盤(pán)前前面幾分鐘的解析全白做。還有一個(gè)容易忽略的細(xì)節(jié)tree.write()會(huì)以二進(jìn)制模式打開(kāi)文件傳進(jìn)去的encoding只影響序列化方式。如果你的目標(biāo)是「文件本身是 utf-8 無(wú) BOM」這樣寫(xiě)沒(méi)問(wèn)題如果下游明確要求帶 BOM就得自己用codecs.BOM_UTF8拼一下別指望encodingutf-8-sig能直接生效。3. 跑通第一條轉(zhuǎn)換鏈路命令行參數(shù)與批量目錄單文件轉(zhuǎn)換只是驗(yàn)證階段真正上線一定是整個(gè)目錄批量跑。這一章按「先跑通一個(gè)文件 → 再批量 → 最后校驗(yàn)」的順序推進(jìn)每一步都能獨(dú)立驗(yàn)證出了問(wèn)題不用一次懷疑全鏈路。參數(shù)設(shè)計(jì)上我傾向于把「輸入、輸出、配置」三個(gè)必填項(xiàng)顯式化其余全部給默認(rèn)值這樣日常調(diào)用最短特殊場(chǎng)景又能覆蓋。3.1 最小可運(yùn)行腳本與六個(gè)必設(shè)參數(shù)命令行接口用argparse就夠了不需要引入第三方框架。參數(shù)不多但每一個(gè)都對(duì)應(yīng)一類(lèi)現(xiàn)實(shí)需求分隔符要能強(qiáng)制指定嗅探失敗時(shí)救命編碼要能指定老文件不是 utf-8出錯(cuò)策略要能選批量跑的時(shí)候不能一條壞數(shù)據(jù)就全停。import argparse def build_parser(): p argparse.ArgumentParser(descriptiontxt - xml 批量轉(zhuǎn)換) p.add_argument(-i, --input, requiredTrue, help輸入文件或目錄) p.add_argument(-o, --output, requiredTrue, help輸出文件或目錄) p.add_argument(-c, --config, requiredTrue, help字段映射 JSON 路徑) p.add_argument(-d, --delimiter, defaultNone, help強(qiáng)制指定分隔符默認(rèn)自動(dòng)嗅探) p.add_argument(-e, --encoding, defaultauto, help源文件編碼auto 表示探測(cè)) p.add_argument(--recursive, actionstore_true, help遞歸遍歷子目錄) p.add_argument(--on-error, choices[skip, abort, report], defaultreport, help單文件失敗時(shí)的處理策略) p.add_argument(--chunk, typeint, default5000, help每多少條記錄寫(xiě)一次盤(pán)) return p參數(shù)說(shuō)明里最需要解釋的是--on-error。skip是壞文件直接跳過(guò)、不留痕跡適合每天定時(shí)跑、只要結(jié)果能出就行的場(chǎng)景abort是遇到第一個(gè)錯(cuò)誤立刻退出適合調(diào)試期能最快暴露問(wèn)題report是默認(rèn)值繼續(xù)跑完但把所有失敗文件的路徑和異常收集起來(lái)最后統(tǒng)一打印。生產(chǎn)上我一般用report跑完看一眼失敗清單比中途停掉再重跑省事得多。--chunk是為大文件準(zhǔn)備的。默認(rèn) 5000 條記錄落一次盤(pán)意味著一個(gè)幾十萬(wàn)行的文件不會(huì)在內(nèi)存里堆出一個(gè)巨大的 ElementTree也不會(huì)因?yàn)樽詈笠恍斜罎?dǎo)致前面幾十分鐘白跑。代價(jià)是分片寫(xiě)入需要自己管理根節(jié)點(diǎn)閉合不能用一次性tree.write()得用增量寫(xiě)法。--delimiter默認(rèn)None而不是\t這個(gè)設(shè)計(jì)很關(guān)鍵。默認(rèn)給個(gè)具體值會(huì)讓用戶忘了嗅探的存在而嗅探失敗時(shí)必須顯式指定反而更清楚。同理--encoding默認(rèn)auto走探測(cè)邏輯探測(cè)不出來(lái)再報(bào)錯(cuò)讓人指定不要默默按 utf-8 讀那樣只會(huì)讀到一堆亂碼而不報(bào)錯(cuò)是更壞的結(jié)果。3.2 批量轉(zhuǎn)換目錄鏡像、命名規(guī)則與失敗清單批量處理的核心約定是「輸出目錄結(jié)構(gòu)和輸入目錄結(jié)構(gòu)鏡像」輸入data/2024/03/a.txt就輸出out/2024/03/a.xml只換擴(kuò)展名、只換根目錄。這樣下游按日期或按設(shè)備號(hào)索引的邏輯完全不用改也不用擔(dān)心同名文件互相覆蓋。from pathlib import Path def convert_tree(src_dir, dst_dir, pattern*.txt, recursiveFalse, on_errorreport, convert_oneNone): src_dir, dst_dir Path(src_dir), Path(dst_dir) it src_dir.rglob(pattern) if recursive else src_dir.glob(pattern) failed, done [], 0 for src in sorted(it): rel src.relative_to(src_dir).with_suffix(.xml) dst dst_dir / rel dst.parent.mkdir(parentsTrue, exist_okTrue) # 鏡像目錄缺哪層補(bǔ)哪層 try: convert_one(src, dst) done 1 except Exception as exc: failed.append((str(src), repr(exc))) if on_error abort: raise print(完成 {} 個(gè)失敗 {} 個(gè).format(done, len(failed))) return failedsorted(it)不是為了美觀是為了讓失敗清單的順序穩(wěn)定方便和上一次的清單做 diffwith_suffix(.xml)只換最后一個(gè)后綴a.txt.bak這種命名會(huì)變成a.txt.xml如果數(shù)據(jù)里真有這種備份文件遍歷時(shí)用pattern加if src.suffix .txt雙重過(guò)濾。dst.parent.mkdir(parentsTrue, exist_okTrue)是鏡像目錄的關(guān)鍵少了它遞歸模式下一進(jìn)子目錄就FileNotFoundError。失敗清單一定要落盤(pán)不能只print。跑批任務(wù)經(jīng)常是后臺(tái)執(zhí)行終端輸出丟了等下游反饋數(shù)據(jù)缺失時(shí)你已經(jīng)無(wú)從查起。最簡(jiǎn)單的做法是寫(xiě)一個(gè)failed.tsv兩列分別是源路徑和異常摘要下次重跑直接讀這個(gè)清單只重跑失敗的比全量重跑快一個(gè)數(shù)量級(jí)。還有一個(gè)隱性約定輸出文件先寫(xiě)臨時(shí)名再原子改名。批量轉(zhuǎn)換中途被 kill 是常態(tài)直接往目標(biāo)路徑寫(xiě)會(huì)留下半截 xml下游讀到解析失敗的殘文件還以為數(shù)據(jù)壞了。寫(xiě)法是dst.with_suffix(.xml.tmp)寫(xiě)完再os.replace一行代碼換來(lái)的確定性很值。3.3 轉(zhuǎn)換后的校驗(yàn)回讀、計(jì)數(shù)與結(jié)構(gòu)比對(duì)轉(zhuǎn)換腳本自己說(shuō)成功不算成功得讓 xml 解析器說(shuō)成功。校驗(yàn)分三層成本從低到高語(yǔ)法層用解析器把文件讀一遍結(jié)構(gòu)層比對(duì)記錄數(shù)和關(guān)鍵字段是否存在內(nèi)容層抽幾條和源 txt 逐字段對(duì)照。import xml.etree.ElementTree as ET def verify(xml_path, expect_recordsNone, required_paths(meta/id,)): tree ET.parse(xml_path) # 語(yǔ)法錯(cuò)誤在這一步就會(huì)拋 root tree.getroot() records root.findall(record) if expect_records is not None and len(records) ! expect_records: raise AssertionError(記錄數(shù)不符: {} ! {}.format(len(records), expect_records)) for r in records[:50]: # 抽查前 50 條別全量遍歷拖慢流程 for path in required_paths: if r.find(path) is None: raise AssertionError(缺少節(jié)點(diǎn): path) return len(records)ET.parse是最硬的一道關(guān)任何標(biāo)簽未閉合、非法字符、編碼聲明與實(shí)際不符都會(huì)在這里炸出來(lái)成本極低建議每個(gè)文件都跑。記錄數(shù)比對(duì)的期望值從哪來(lái)最省事的辦法是數(shù)源文件的有效行數(shù)去掉表頭和空行和 xml 里的record節(jié)點(diǎn)數(shù)對(duì)齊兩邊差一條就說(shuō)明某行解析異常被吞了。結(jié)構(gòu)層校驗(yàn)只抽查前 50 條是刻意的取舍。全量遍歷幾十萬(wàn)節(jié)點(diǎn)在批量場(chǎng)景下會(huì)把總耗時(shí)翻倍而結(jié)構(gòu)錯(cuò)誤幾乎必然出現(xiàn)在頭部抽查足夠發(fā)現(xiàn)系統(tǒng)性問(wèn)題。真正需要逐條比對(duì)的是內(nèi)容層那屬于抽樣人工核對(duì)或者單獨(dú)寫(xiě)回歸腳本的范疇不該塞進(jìn)每次跑批的默認(rèn)路徑。如果機(jī)器上有xmllint加一道外部校驗(yàn)也很劃算。它和 Python 解析器的實(shí)現(xiàn)不同能覆蓋一些標(biāo)準(zhǔn)庫(kù)容忍但嚴(yán)格解析器不容忍的寫(xiě)法# 只做語(yǔ)法校驗(yàn)不輸出內(nèi)容返回非 0 表示文件有問(wèn)題 find out -name *.xml -print0 | xargs -0 -n1 xmllint --noout-print0配xargs -0是為了應(yīng)付帶空格的文件名批量腳本里凡是遍歷文件的地方都建議這么寫(xiě)別等某天出現(xiàn)一個(gè)report 2024 03.txt才發(fā)現(xiàn)整條流水線斷在空格上。4. 特殊字符與編碼轉(zhuǎn)義、BOM、空值三處最容易翻車(chē)的地方前三章跑的是「干凈數(shù)據(jù)」現(xiàn)實(shí)里的 txt 從來(lái)不干凈。業(yè)務(wù)字段里帶、備注里帶尖括號(hào)、從舊系統(tǒng)導(dǎo)出的文件是 GBK、空值有時(shí)候是空字符串有時(shí)候是NULL有時(shí)候干脆缺列。這些問(wèn)題不會(huì)讓腳本立刻崩但會(huì)讓下游拿到「能解析、值不對(duì)」的 xml排查成本比直接報(bào)錯(cuò)高得多。這一章把三類(lèi)問(wèn)題拆開(kāi)處理。4.1 五個(gè)必須處理的字符與 CDATA 的取舍xml 里有五個(gè)字符在文本節(jié)點(diǎn)中具有特殊含義、、、、。前兩個(gè)不轉(zhuǎn)義會(huì)直接破壞文檔結(jié)構(gòu)后三個(gè)在屬性值里必須轉(zhuǎn)義、在文本節(jié)點(diǎn)里可轉(zhuǎn)可不轉(zhuǎn)。好消息是只要用序列化器寫(xiě)節(jié)點(diǎn)這五個(gè)字符都會(huì)被自動(dòng)處理你什么都不用做壞消息是很多人會(huì)在某一步「優(yōu)化」成字符串拼接把保護(hù)全拆了。# 錯(cuò)誤示范手工拼接字段里一個(gè) 就讓整個(gè)文件解析失敗 bad name{}/name.format(name) # 正確做法交給序列化器文本賦值即可 el ET.SubElement(record_el, name) el.text name值得強(qiáng)調(diào)的是el.text賦值和el.set(attr, value)都會(huì)自動(dòng)轉(zhuǎn)義但如果你往el.text里塞的是已經(jīng)轉(zhuǎn)義過(guò)的amp;序列化器會(huì)再轉(zhuǎn)一次變成amp;amp;下游解析出來(lái)就是字面量amp;。這類(lèi)雙重轉(zhuǎn)義問(wèn)題在數(shù)據(jù)經(jīng)過(guò)兩套系統(tǒng)時(shí)特別常見(jiàn)判斷方法很簡(jiǎn)單源數(shù)據(jù)里看到amp;就說(shuō)明上游已經(jīng)轉(zhuǎn)過(guò)一次你要做的是還原而不是再轉(zhuǎn)。CDATA 是另一個(gè)常見(jiàn)訴求。有人覺(jué)得字段里有 html 片段必須用![CDATA[...]]包起來(lái)實(shí)際上標(biāo)準(zhǔn)庫(kù)的 ElementTree 不支持直接輸出 CDATA需要手工拼或者換 lxml。多數(shù)情況下沒(méi)必要CDATA 的唯一作用是讓和不用轉(zhuǎn)義而序列化器已經(jīng)幫你轉(zhuǎn)義了下游解析出來(lái)的文本內(nèi)容完全一致。只有當(dāng)字段內(nèi)容本身就是一段需要人眼閱讀的 xml/html 片段、且下游明確要求原樣保留時(shí)才值得折騰否則純屬給自己加一道工序。真正需要留意的是控制字符。從老設(shè)備或數(shù)據(jù)庫(kù)導(dǎo)出的文本里偶爾混入\x00、\x0b這類(lèi)不可見(jiàn)字符它們?cè)?xml 1.0 里是非法的序列化時(shí)不報(bào)錯(cuò)寫(xiě)出來(lái)解析時(shí)炸。穩(wěn)妥做法是在賦值前統(tǒng)一過(guò)濾import re ILLEGAL re.compile(r[\x00-\x08\x0b\x0c\x0e-\x1f]) def clean(text): return ILLEGAL.sub(, text) if text else text這個(gè)正則放過(guò)的\x09制表符、\x0a換行、\x0d回車(chē)是 xml 允許的別一起干掉否則多行文本會(huì)被壓成一行。4.2 編碼探測(cè)與 BOMutf-8-sig 與 gb18030編碼問(wèn)題最氣人的地方在于它經(jīng)常不報(bào)錯(cuò)。用 utf-8 去讀一個(gè) GBK 文件Python 可能直接拋UnicodeDecodeError這是好事但如果文件里恰好全是 ASCII 字符讀進(jìn)來(lái)一切正常等某天出現(xiàn)一個(gè)中文人名整批數(shù)據(jù)才開(kāi)始崩。所以編碼處理要分兩步先探測(cè)探測(cè)不出來(lái)就報(bào)錯(cuò)讓人指定絕不靜默降級(jí)。def open_text(path, encodingauto): 按指定編碼打開(kāi)auto 時(shí)先看 BOM再按 utf-8 試讀 if encoding and encoding ! auto: return open(path, r, encodingencoding, newline) with open(path, rb) as f: head f.read(4) if head.startswith(b\xef\xbb\xbf): return open(path, r, encodingutf-8-sig, newline) # 吃掉 BOM return open(path, r, encodingutf-8, newline)newline這個(gè)參數(shù)容易被忽略。默認(rèn)的通用換行模式會(huì)把\r\n統(tǒng)一成\n對(duì)多數(shù)場(chǎng)景沒(méi)問(wèn)題但如果某個(gè)字段本身要保留原始換行符統(tǒng)一轉(zhuǎn)換就丟了信息。批量轉(zhuǎn)換工具我傾向于關(guān)掉自動(dòng)轉(zhuǎn)換讓數(shù)據(jù)原樣進(jìn)來(lái)需要處理再顯式處理。BOM 是另一個(gè)高頻坑。帶 BOM 的 utf-8 文件用encodingutf-8打開(kāi)第一個(gè)字段的開(kāi)頭會(huì)多出一個(gè)不可見(jiàn)的\ufeff肉眼完全看不出來(lái)但用它做字典鍵、做字符串比較、寫(xiě)進(jìn) xml 都會(huì)出問(wèn)題。判斷方法repr(row[0])看開(kāi)頭是不是\ufeff。解決就是utf-8-sig它在讀取時(shí)自動(dòng)剝掉 BOM寫(xiě)入時(shí)不加 BOM。GBK 系文件沒(méi)有 BOM 可依靠utf-8試讀失敗來(lái)兜底是最簡(jiǎn)單的策略def read_with_fallback(path): for enc in (utf-8-sig, gb18030): try: with open(path, r, encodingenc, newline) as f: return f.read(), enc except UnicodeDecodeError: continue raise UnicodeDecodeError(無(wú)法確定編碼請(qǐng)用 --encoding 指定: str(path))gb18030而不是gbk是因?yàn)榍罢呤浅芨采w更多生僻字代價(jià)是偶爾會(huì)把非 GBK 文件「成功」解成亂碼。所以這個(gè)兜底只該在明確知道數(shù)據(jù)源是中文舊系統(tǒng)時(shí)啟用不要無(wú)腦加進(jìn)通用流程。探測(cè)順序也別搞反先 utf-8 再 gb18030反過(guò)來(lái)會(huì)讓很多 utf-8 文件被誤判。4.3 空值、缺字段與類(lèi)型推斷空值處理是最容易被低估的一節(jié)。同一條記錄字段為空可能有三種含義值確實(shí)為空、這個(gè)字段在源數(shù)據(jù)里不存在、這個(gè)字段本來(lái)該有值但采集失敗了。三種情況在 xml 里應(yīng)該有不同的表達(dá)全塞成空字符串等于把信息丟了。工具里的約定是配置里標(biāo)了required的字段為空直接報(bào)錯(cuò)標(biāo)了default的字段為空補(bǔ)默認(rèn)值既沒(méi)標(biāo) required 也沒(méi)標(biāo) default 的字段為空就寫(xiě)空節(jié)點(diǎn)。這個(gè)規(guī)則簡(jiǎn)單到不用記但能覆蓋九成場(chǎng)景。類(lèi)型轉(zhuǎn)換則要防兩個(gè)坑。第一個(gè)是浮點(diǎn)數(shù)str(0.1 0.2)會(huì)得到0.30000000000000004直接寫(xiě)進(jìn) xml 下游拿去算金額就出事了第二個(gè)是科學(xué)計(jì)數(shù)法很小的數(shù)會(huì)被str()成1e-05很多只認(rèn)定點(diǎn)小數(shù)的老系統(tǒng)直接報(bào)格式錯(cuò)誤。from datetime import datetime def convert(raw, spec): 按配置把原始字符串轉(zhuǎn)成 xml 文本空值走 default if raw or raw is None: return spec.get(default, ) t spec.get(type, str) if t int: return str(int(raw)) if t float: # 定點(diǎn)格式化保留 6 位后裁掉多余的 0避免科學(xué)計(jì)數(shù)法和浮點(diǎn)尾差 return format(float(raw), .6f).rstrip(0).rstrip(.) if t datetime: return datetime.strptime(raw, spec[format]).isoformat() return raw.6f這個(gè)精度不是隨便定的。金融和計(jì)量場(chǎng)景一般到小數(shù)點(diǎn)后 4 到 6 位再多就是噪聲r(shí)strip(0).rstrip(.)是為了讓1.500000變成1.5、2.000000變成2.再變成2注意順序不能反先裁 0 再裁點(diǎn)否則2.000000裁完點(diǎn)是2.會(huì)留下一個(gè)非法的小數(shù)點(diǎn)。datetime的format必須寫(xiě)在配置里而不是代碼里因?yàn)樵磾?shù)據(jù)的日期格式五花八門(mén)%Y/%m/%d、%Y-%m-%d %H:%M:%S、甚至20240301都有。統(tǒng)一轉(zhuǎn)成 ISO 8601 再寫(xiě)進(jìn) xml下游排序和比較都不用再解析一次。5. 避坑記錄五條從現(xiàn)場(chǎng)撿回來(lái)的報(bào)錯(cuò)下面這五條都不是理論推演是批量跑真實(shí)數(shù)據(jù)時(shí)踩出來(lái)的。每條按「現(xiàn)象 → 原因 → 解決」寫(xiě)如果你正準(zhǔn)備跑第一批全量數(shù)據(jù)建議先對(duì)照著掃一眼自己的輸入。5.1 所有行只解析出一列分隔符憑空消失現(xiàn)象很典型本地用記事本打開(kāi)文件肉眼能看到字段之間有明顯間隔腳本跑完卻每條記錄只有一個(gè)字段值是一整行拼接后的字符串sniff_delimiter返回None。原因是分隔符在流轉(zhuǎn)過(guò)程中被換了。從 Excel 或網(wǎng)頁(yè)表格復(fù)制粘貼出來(lái)的「制表符」經(jīng)過(guò)郵件、聊天工具、某些編輯器保存之后制表符\t會(huì)被展開(kāi)成 4 個(gè)或 8 個(gè)空格也有的是反過(guò)來(lái)源文件本來(lái)是多個(gè)空格對(duì)齊被某個(gè)工具壓成了一個(gè)空格。兩種情況下split(\t)都拿不到多列。解決辦法分兩層。第一層是嗅探函數(shù)必須把空格列進(jìn)候選并且允許「連續(xù)空格視為一個(gè)分隔符」的寬松模式用re.split(r , line.strip())處理。第二層是加一道列數(shù)斷言解析出的第一行如果只有 1 列而配置里的fields有 4 個(gè)直接報(bào)錯(cuò)退出并打印前 200 個(gè)字符的repr讓人一眼看到實(shí)際的分隔符是什么。我在批量腳本里把這條斷言做成了默認(rèn)開(kāi)啟寧可跑失敗也不要跑出一堆看似正常的錯(cuò)數(shù)據(jù)。5.2 Windows 上跑批中文全變成問(wèn)號(hào)或直接拋編碼錯(cuò)現(xiàn)象是在開(kāi)發(fā)機(jī)上跑得好好的腳本換到另一臺(tái)機(jī)器尤其是中文 Windows 服務(wù)器上日志里的中文變成???嚴(yán)重時(shí)直接UnicodeEncodeError: gbk codec cant encode character。原因和轉(zhuǎn)換邏輯無(wú)關(guān)出在標(biāo)準(zhǔn)輸出。Python 在 Windows 上的sys.stdout默認(rèn)編碼跟隨系統(tǒng)區(qū)域設(shè)置中文環(huán)境就是 GBK。腳本里任何一句print(處理完成: 中文路徑)都會(huì)走這個(gè)編碼遇到 GBK 表示不了的字符就炸。更隱蔽的是如果腳本把日志重定向到文件文件也會(huì)按 GBK 寫(xiě)用 utf-8 打開(kāi)就是亂碼。解決辦法是別依賴默認(rèn)編碼。寫(xiě)文件時(shí)顯式encodingutf-8所有open()調(diào)用都帶上需要print調(diào)試信息時(shí)在入口處加一句sys.stdout.reconfigure(encodingutf-8, errorsreplace)3.7 起可用或者干脆把日志寫(xiě)進(jìn)文件而不是打在終端。errorsreplace是保命的它讓日志不會(huì)因?yàn)橐粋€(gè)字符打不出來(lái)而中斷整個(gè)跑批。5.3 數(shù)值被寫(xiě)成科學(xué)計(jì)數(shù)法下游報(bào)類(lèi)型不匹配現(xiàn)象是轉(zhuǎn)換后的 xml 里出現(xiàn)value1e-05/value或value3.0000000000000004/value本地看沒(méi)人發(fā)現(xiàn)下游做數(shù)值解析時(shí)報(bào)格式錯(cuò)誤或者算出來(lái)的金額差了幾分錢(qián)。原因是中間經(jīng)過(guò)了float再str()。二進(jìn)制浮點(diǎn)表示不了 0.1 這種十進(jìn)制小數(shù)累加之后尾差就出來(lái)了小于 0.0001 的數(shù)str()時(shí)自動(dòng)切科學(xué)計(jì)數(shù)法這是 Python 的顯示規(guī)則不是 bug。解決辦法是在配置里給每個(gè)數(shù)值字段顯式標(biāo)type: float轉(zhuǎn)換時(shí)走定點(diǎn)格式化并且在格式化之前用Decimal(str(raw))收一道把源字符串按十進(jìn)制語(yǔ)義解析避免float()引入誤差。如果字段是金額更好的做法是配置里加個(gè)scale: 2轉(zhuǎn)換時(shí)乘 100 存整數(shù)、寫(xiě) xml 時(shí)再除以 100全程不碰浮點(diǎn)。5.4 xml 里出現(xiàn)重復(fù)的同名節(jié)點(diǎn)下游只取到第一個(gè)現(xiàn)象是下游反饋「某些記錄的 name 全空了」你把文件打開(kāi)一看結(jié)構(gòu)是這樣的recordpayloadnameA/name/payloadpayloadnameB/name/payload/record兩個(gè)payload并列下游用find(payload/name)只拿到 AB 被忽略。原因是映射代碼里用了node.find(part) or ET.SubElement(node, part)這種簡(jiǎn)寫(xiě)。Element 的布爾值由子元素?cái)?shù)量決定一個(gè)已存在但暫時(shí)沒(méi)有子節(jié)點(diǎn)的父節(jié)點(diǎn)會(huì)被判成假于是又創(chuàng)建了一個(gè)同名節(jié)點(diǎn)。層級(jí)越深越容易觸發(fā)meta下面先建了id之后才建time的那一瞬間就是假的。解決辦法就是 2.2 里那段顯式判空。另外在寫(xiě)出之前加一道自檢遍歷每條記錄的直接子節(jié)點(diǎn)名如果出現(xiàn)重復(fù)就報(bào)錯(cuò)。這道檢查成本很低但能擋住一整類(lèi)結(jié)構(gòu)性問(wèn)題。5.5 大文件跑一半內(nèi)存爆掉機(jī)器卡死現(xiàn)象是處理幾萬(wàn)行的小文件一切正常換成一個(gè)幾十萬(wàn)行的日志文件腳本跑到中途內(nèi)存曲線直線上升最后MemoryError或者機(jī)器開(kāi)始換頁(yè)卡死。原因是兩層疊加f.readlines()一次性把整個(gè)文件讀進(jìn)內(nèi)存ElementTree 的整棵樹(shù)也全在內(nèi)存里幾十萬(wàn)個(gè)節(jié)點(diǎn)每個(gè)都是對(duì)象開(kāi)銷(xiāo)遠(yuǎn)大于原始文本。一個(gè) 200MB 的文本能撐出 2GB 以上的內(nèi)存占用。解決辦法是兩頭都改成流式。讀用for line in f:逐行迭代不要readlines()寫(xiě)用增量模式每--chunk條記錄就把當(dāng)前這批子節(jié)點(diǎn)序列化追加到文件然后清空。增量寫(xiě)的實(shí)現(xiàn)稍微繞一點(diǎn)需要手工寫(xiě)?xml ...?聲明和根節(jié)點(diǎn)開(kāi)標(biāo)簽中間追加序列化后的片段最后補(bǔ)上根節(jié)點(diǎn)閉標(biāo)簽。如果下游不強(qiáng)制要求單文件更省事的方案是按固定行數(shù)切片輸出多個(gè) xmlpart-0001.xml、part-0002.xml下游按序讀取即可改造成本幾乎為零。6. 進(jìn)階把映射關(guān)系外置成模板順手加一道回歸校驗(yàn)字段映射寫(xiě)進(jìn) JSON 之后下一個(gè)自然的需求是「多個(gè)數(shù)據(jù)源共用大部分字段只差幾列」。這時(shí)候復(fù)制多份配置會(huì)讓維護(hù)變成災(zāi)難改一個(gè)公共字段要改十幾份文件。我的做法是給配置加一層extends子配置只聲明差異字段加載時(shí)先繼承父配置再覆蓋實(shí)現(xiàn)也就二十行。模板字段含義常見(jiàn)取值extends父模板路徑相對(duì)當(dāng)前配置解析base/common.jsonsource.encoding源文件編碼auto走探測(cè)utf-8-sig/gb18030fields[].index列下標(biāo)子模板可用-1表示追加到末尾0~n-1fields[].required為空時(shí)是報(bào)錯(cuò)還是補(bǔ)默認(rèn)值true/falseoutput.pretty是否縮進(jìn)交付給人看開(kāi)給機(jī)器讀可關(guān)true/falseimport json from pathlib import Path def load_config(path, _seenNone): 支持 extends 繼承子配置的 fields 按 path 覆蓋父配置 _seen _seen or set() path Path(path).resolve() if path in _seen: raise ValueError(配置循環(huán)繼承: str(path)) _seen.add(path) cfg json.loads(path.read_text(encodingutf-8)) parent cfg.pop(extends, None) if not parent: return cfg base load_config(path.parent / parent, _seen) merged dict(base) merged.update({k: v for k, v in cfg.items() if k ! fields}) by_path {f[path]: f for f in base.get(fields, [])} for f in cfg.get(fields, []): by_path[f[path]] f # 同路徑覆蓋新路徑追加 merged[fields] list(by_path.values()) return merged_seen這個(gè)集合是防循環(huán)繼承的A 繼承 B、B 又繼承 A 的情況在配置文件里比想象中容易出現(xiàn)尤其是多人維護(hù)的時(shí)候。fields用path做鍵而不是用index是因?yàn)橥粋€(gè)字段在不同數(shù)據(jù)源里列位置經(jīng)常變但節(jié)點(diǎn)路徑是穩(wěn)定的語(yǔ)義標(biāo)識(shí)。配套的回歸校驗(yàn)也很關(guān)鍵。改完配置之后拿一份已知正確輸出的樣本重新跑一遍把結(jié)果和基線文件做結(jié)構(gòu)化比對(duì)——不比字節(jié)比「記錄數(shù)、每條記錄的子節(jié)點(diǎn)路徑集合、關(guān)鍵字段的值」這樣縮進(jìn)變化、屬性順序變化不會(huì)誤報(bào)真正的字段丟失和值錯(cuò)誤一個(gè)都跑不掉。def diff_structure(a_root, b_root): 比對(duì)兩棵樹(shù)的節(jié)點(diǎn)路徑與文本返回差異列表 def walk(el, prefix): cur prefix / el.tag yield cur, (el.text or ).strip() for c in el: yield from walk(c, cur) left, right dict(walk(a_root)), dict(walk(b_root)) return [(k, left.get(k), right.get(k)) for k in set(left) | set(right) if left.get(k) ! right.get(k)]這套東西我用了很久最大的價(jià)值不是發(fā)現(xiàn) bug而是讓「改配置」這件事從「提心吊膽跑一遍看看」變成「跑完看差異列表是不是空的」。有一回一個(gè)同事把default從一個(gè)空字符串改成0結(jié)構(gòu)比對(duì)立刻列出了兩千多條差異五分鐘就回滾了要是靠人工抽查這個(gè)改動(dòng)會(huì)安安靜靜躺在數(shù)據(jù)里等下游算出一個(gè)全零的報(bào)表才被發(fā)現(xiàn)。從那以后我每次動(dòng)映射配置都強(qiáng)制先跑一遍基線比對(duì)再上線哪怕只是改了一個(gè)縮進(jìn)。希望這套思路能幫到你源碼包和示例配置在整理好的下載里拿到后先跑自帶的樣例再換自己的字段。本文還有配套的精品資源點(diǎn)擊獲取