做数据迁移或者系统对接的时候最烦的就是两边格式对不上。最近我在做一个数据对接的小项目需要把WorkBuddy导出的任务记录转换成分析平台要求的DSH格式。WorkBuddy是团队一直在用的任务与工时管理软件它导出的JSON结构记录着任务、子任务、评论、负责人这些信息而下游的数据分析平台只接受DSH一种按行为单位、以keyvalue形式组织的结构化文本格式。两个系统之间没有官方转换工具数据要进入分析管道只能自己写转换程序。这篇文章就把我完整实现workbuddy-to-dsh的整个过程记录下来包括字段映射怎么设计、嵌套数据怎么拍平、时区和时间格式怎么处理、批量转换脚本怎么组织以及实际调试中踩过的坑。无论你最终面对的目标格式是不是DSH这套思路在JSON和自定义文本格式互转的场景下都能直接参考。1. 为什么需要这个转换工具1.1 WorkBuddy导出的数据长什么样先看输入侧。WorkBuddy导出的JSON并不是一个平坦结构每一项任务是一个对象里面既有标题、状态、优先级这类标量字段也有负责人、标签、子任务、评论这些嵌套数据。一个典型的任务记录大概长这样{ tasklist: [ { id: T-202405-001, title: 完成季度数据报表, project: 数据组, tags: [报表, 季度], status: done, priority: 2, estimate_hours: 8.5, spent_minutes: 510, created_at: 2024-05-02T09:30:0008:00, due_date: 2024-05-10, assignee: { name: 张三, email: zhangsanexample.com, team: BI组 }, subtasks: [ {title: 清洗数据, done: true, hours: 3}, {title: 制作图表, done: true, hours: 2.5} ], comments: [ {author: 李四, content: 注意统计口径, time: 2024-05-03T10:00:0008:00} ] } ], exported_at: 2024-06-01T00:00:0008:00 }这种结构对人类很友好但对下游系统并不友好。分析平台要的是扁平的、列名清晰的输入它不可能自己去解析JSON里多层嵌套的assignee、subtasks和comments。如果直接把这份JSON喂给平台平台解析器要么直接拒绝要么只取到最外层字段任务详情、工时、负责人这些关键信息全部丢失。所以转换的第一步就是把这棵“树”拍平成一条“线”。1.2 DSH格式到底有什么要求DSH这种格式简单说就是每一行一个键值对字段用等号连接不同记录之间用分隔行隔开。文件开头可以写一个类型声明行告诉解析器这批数据的版本和类型。我这边对接的平台要求的格式是这样task:record:v1 idT-202405-001 title完成季度数据报表 project数据组 statusdone priority2 estimate_hours8.5 spent_hours8.5 created_at1714613400 due_date1715270400 assignee张三|BI组|zhangsanexample.com tags报表,季度 subtask_count2 subtask_done2 comment_count1 ---特点很明显第一所有字段都是keyvalue的文本行解析起来不需要考虑嵌套层级按行读、按等号切分就能还原第二时间统一成Unix时间戳跨系统不会有时区歧义无论下游服务器在哪个地域解析出来的都是同一个绝对时间点第三列表和嵌套对象要聚合成单行字符串用竖线、逗号这样的分隔符把子信息串起来。DSH的“结构化”并不是指层级结构而是指字段结构和类型约定都提前定义好解析端不需要做任何猜测。1.3 为什么不能靠手动整理或者改后缀糊弄有人可能会想数据量不大手动整理不就行了或者直接把JSON文件改名成.dsh。这两种做法都行不通。手动整理在几十条记录的时候或许可行但任务记录是每周都在增长的这周可能几十条下个月就是上千条手工维护的边际成本会快速上升。而且手抄数据极易出错时间戳算错一位、工时单位搞混都是常态。至于改后缀那就更不靠谱了。下游解析器看到JSON的花括号和冒号根本不会认报错只是时间问题。还有一层原因容易被忽略DSH不只是“换个后缀的文本”它有自己约定好的字段名和值类型规则。比如spent_hours要求数字created_at要求时间戳这些约束只有通过转换程序才能系统性地保证。转换程序的核心价值不是省一次两次的力气而是把“从WorkBuddy到DSH”这条路径固化下来每次导出新数据跑一遍脚本就能得到干净可用的结果。2. 转换方案选型脚本、映射表还是现成工具2.1 三条路线对比动手之前我先评估了三条路线。第一条是用现成的数据转换工具比如各种ETL工具或者在线转换网站。这类工具适合字段一一对应的场景但WorkBuddy的数据里有嵌套对象和数组绝大多数在线工具不支持自定义这种聚合逻辑字段映射配置起来反而比写代码还麻烦。第二条是用可视化编排平台走一遍配置流程但为了一个内部使用的小转换器引入一套重平台运维成本明显不划算。第三条就是直接用Python写一个几十行的转换脚本输入JSON输出DSH中间的逻辑完全可控。三条路线对比下来结论很清楚数据格式转换这种需求越具体、格式越私有越应该用脚本解决。ETL工具擅长的是“已知格式到已知格式”的转换而这里的目标格式是下游平台自己定义的只有脚本能精确控制每一行的输出。更何况这是一个长期要用的工具脚本写完之后后续改映射、加字段都是分钟级的事。2.2 为什么用Python而不是其他语言我选Python的原因有三个。第一标准库自带json模块和datetime模块解析WorkBuddy导出、处理时间戳转换都不需要装第三方依赖在任何一台机器上都能直接跑第二字符串处理能力对路DSH这种按行拼文本的格式在Python里用列表收集再join写起来非常顺手第三脚本后期维护成本低我自己需要偶尔调整映射规则Python改起来比编译型语言快得多。如果说项目要嵌入到已有的服务里那当然应该用对应的语言重写但作为一个独立使用的转换小工具Python是性价比最高的选择。2.3 环境准备清单环境方面要求很低Python 3.8以上就够用不需要虚拟环境不需要pip安装任何包。我本机用的是Python 3.11直接在项目目录下建了两个文件一个workbuddy_to_dsh.py放核心转换逻辑一个batch_convert.py放批量处理入口。如果你的Python版本比较老注意fromisoformat这个方法在3.7之后才支持带时区的ISO字符串3.6及以下需要自己手动解析这个细节在后面的时间处理部分会再提到。提示脚本要统一用UTF-8编码读写文件尤其是任务描述里如果有中文、日文或者特殊符号编码不一致会导致输出乱码。Windows环境下面运行Python脚本建议在代码里显式指定encodingutf-8不要依赖系统默认编码。3. 核心转换逻辑从JSON到DSH的完整路径3.1 字段映射表先想清楚转换的第一步不是写代码而是把字段映射表列出来。哪些字段要原样保留哪些字段要改名哪些字段要用嵌套数据计算出来都先想好。我做了一张这样的对照表WorkBuddy字段DSH字段处理方式idid原样保留titletitle原样保留projectproject原样保留statusstatus原样保留prioritypriority原样保留estimate_hoursestimate_hours原样保留spent_minutesspent_hours分钟转小时created_atcreated_atISO时间转Unix时间戳due_datedue_date日期转Unix时间戳assigneeassignee展开为姓名|团队|邮箱tagstags数组转逗号分隔字符串subtaskssubtask_count / subtask_done计算子任务总数与完成数commentscomment_count计算评论数这张表是整个脚本的地基。写代码之前先做映射看起来是个笨功夫实际上能避免绝大多数低级错误。比如spent_minutes和spent_hours如果不先列出来写代码的时候很容易忘了单位换算导致下游看到的工时数据差了60倍。我见过不止一次就是因为省了这一步最后对账的时候发现所有工时不匹配只能重新导数据。字段映射规则定好之后写转换函数就顺了。我把核心的映射逻辑放在flatten_task函数里它的职责是把一个嵌套的任务对象拍平成一个扁平的字典键是DSH字段值是待输出的字符串或数字def flatten_task(task): record {} for src_key, dst_key in FIELD_MAP.items(): if src_key in task: record[dst_key] task[src_key] if assignee in task: a task[assignee] record[assignee] {}|{}|{}.format( a.get(name, ), a.get(team, ), a.get(email, ) ) if tags in task: record[tags] ,.join(task[tags]) if subtasks in task: subs task[subtasks] record[subtask_count] len(subs) record[subtask_done] sum(1 for s in subs if s.get(done)) if comments in task: record[comment_count] len(task[comments]) return record这里有个设计取舍值得说一下我没有把assignee里的name、team、email分别映射成三个DSH字段而是聚合成一个字符串。原因是下游平台的DSH字段列表是固定的新增字段要改平台的解析配置代价比较高而把三个信息用竖线串进一个字段既保住了信息又不破坏DSH的字段结构。下游需要单独取某个值的话按竖线split一下就行成本很低。3.2 嵌套结构拍平的三种通用手法JSON转扁平格式时嵌套数据的处理是最容易卡壳的地方。我总结下来就三种手法。第一种是聚合把数组里的元素转换成统计值比如subtasks变成subtask_count和subtask_donecomments变成comment_count典型应用场景是下游只需要“有多少”“完成多少”这类汇总信息。第二种是拼接把数组或对象用分隔符串成单行字符串比如tags用逗号拼assignee用竖线拼典型应用场景是下游需要看到内容但不需要单独拆列。第三种是跳过如果嵌套内容对下游完全没用直接不输出省得DSH文件里堆一堆垃圾行。这三种手法没有绝对的好坏完全看下游系统需要什么。我最初想把评论内容也拼进DSH后来发现平台根本不消费这个字段就把它砍掉了只保留comment_count。少一个字段解析和写入都少一分出错的可能。如果你在做的也是类似的转换项目建议先问清楚下游到底消费哪些字段再决定用哪种手法不要想当然地“全都带上”。3.3 时间字段的规范化是个大坑时间处理是整个项目里最容易被忽视、也最值得展开说的一块。WorkBuddy导出的是带时区的ISO8601格式比如2024-05-02T09:30:0008:00而DSH要求Unix时间戳。这里面有两层逻辑一是把ISO字符串解析成datetime对象二是换算成UTC之后取timestamp。def time_to_epoch(value): if not value: return try: dt datetime.fromisoformat(value) return str(int(dt.timestamp())) except Exception: return value看起来简单但实际要小心两个点。第一点fromisoformat在Python 3.7之前对带时区字符串的支持不完整如果你的运行环境是旧版本需要先手动解析时区偏移。第二点due_date这种纯日期字段WorkBuddy导出的是2024-05-10这种格式没有时区信息如果直接交给fromisoformat解析timestamp会按当前机器时区计算不同机器上跑出来的结果可能差8个小时。正确做法是给纯日期补一个默认时区比如统一按东八区处理from datetime import timezone, timedelta CST timezone(timedelta(hours8)) def date_to_epoch(value): dt datetime.strptime(value, %Y-%m-%d).replace(tzinfoCST) return str(int(dt.timestamp()))这种“统一时区”的思路在做跨系统数据交换时尤其重要。只要DSH文件里的时间戳全部基于同一个时区换算下游不管在哪个国家、哪台服务器上解析拿到的都是同一个绝对时间点。我在项目里还专门写了一个测试用例把同一份数据放到东八区和UTC时区的两台机器上跑验证输出时间戳完全一致才敢放心交接给下游。3.4 序列化输出与自校验拍平后的记录要转成DSH文本最后一个关键环节是序列化。我用的方法是逐行拼keyvalue先拼标题行task:record:v1然后按固定字段顺序输出最后每条记录后面加三个连字符作为分隔行。def to_dsh_line(key, value): text str(value) text text.replace(\\, \\\\).replace(\n, \\n) return {}{}.format(key, text) def convert(data): lines [task:record:v1] for task in data.get(tasklist, []): flat flatten_task(task) for k, v in flat.items(): if k in (created_at, due_date): v time_to_epoch(v) lines.append(to_dsh_line(k, v)) lines.append(---) return \n.join(lines)这里做了两层防护。第一层是转义任务标题或评论内容里如果恰好有换行符直接输出会把DSH的“一行一个键值对”结构打破所以我把换行符替换成\n字面量反斜杠本身也要转义否则解析端还原的时候会混淆。第二层是空值过滤flatten_task里只收集存在的字段任务里没写due_date就不会输出due_date行DSH解析器对缺字段的记录也要能容忍这一条需要提前和下游平台约定好。输出完成后我还会做一次快速校验检查生成的文件里每条记录是否都有id字段检查行数是否等于预期值检查时间戳是否能被解析成合法时间。把这些校验写在脚本末尾能在数据出错的第一时间就把问题挡在管道外面而不是等下游解析器报错再来排查。4. 进阶玩法批量转换和增量更新4.1 命令行入口设计单文件的转换脚本很容易写但用起来舒服还需要一个像样的命令行入口。我把核心逻辑和入口拆成了两个文件workbuddy_to_dsh.py只负责转换逻辑暴露parse_workbuddy和convert两个函数batch_convert.py负责接收命令行参数、遍历文件、调用转换函数、处理异常。命令行入口我设计成最简单的两个参数输入JSON路径和输出DSH路径。不想引入argparse直接用sys.argv判断保持脚本零依赖。使用方式是workbuddy_to_dsh.py input.json output.dsh一次转换一个文件。这个入口对单次转换场景已经够用实际使用中更常见的是目录级批量转换。4.2 目录级批量处理WorkBuddy会按周导出多份JSON放在同一个导出目录里我需要把这个目录下所有.json文件都转换一遍。批量脚本的核心逻辑如下def main(): src_dir Path(sys.argv[1]) dst_dir Path(sys.argv[2]) dst_dir.mkdir(parentsTrue, exist_okTrue) files sorted(src_dir.glob(*.json)) for f in files: try: data w2d.parse_workbuddy(f) text w2d.convert(data) out dst_dir / (f.stem .dsh) out.write_text(text, encodingutf-8) print({} - {} ({} tasks).format(f.name, out.name, len(data[tasklist]))) except Exception as exc: print(FAIL {}: {}.format(f.name, exc)) print(batch done, {} files.format(len(files)))这个脚本里有几个细节是实际跑过之后才加上的。一是把输出文件名定为输入文件的主文件名加.dsh后缀这样源文件和目标文件能一一对应出问题的时候方便回溯。二是每个文件单独try-except单个文件转换失败不会中断整个批次日志里会明确打印哪个文件失败了。三是用sorted排序保证批量转换的顺序是确定的不会因为文件系统的枚举顺序不同导致输出顺序不稳定。4.3 增量更新怎么判断还有一个高频需求是增量更新上周已经转换过的任务这周不想再全量处理一遍。我的做法是靠文件的生成时间做判断。批量脚本里增加一个参数--since只处理修改时间晚于指定时间戳的JSON文件since 0.0 if --since in sys.argv: idx sys.argv.index(--since) since float(sys.argv[idx 1]) for f in files: if f.stat().st_mtime since: continue # 转换逻辑这个方案简单可靠前提是WorkBuddy导出文件的修改时间能反映新增内容。如果导出工具生成新文件时总会刷新mtime那么增量判断就是准的。如果平台会修改旧文件或者mtime不可靠就需要在JSON内容里找version或者exported_at字段做对比复杂度会高一些但我实际使用下来mtime方案已经足够了。5. 实战中的坑与排查记录5.1 中文乱码问题第一次批量转换之后我打开DSH文件检查发现任务标题里的中文全部变成了乱码。排查下来不是转换逻辑的问题而是文件读写编码不一致。我最初写输出文件时用了系统默认编码在Windows上默认可能是GBK而DSH文件在Linux平台解析两边编码对不上中文就全部变成了问号或者乱码。解决方式很简单所有文件读写都显式指定encodingutf-8。同时我在读取JSON时也显式声明编码utf-8-sig防止某些JSON文件自带BOM头导致json.load直接报错。这个坑不算深但特别容易在跨平台协作时出现因为每个开发者本机的默认编码可能都不一样。5.2 字段缺失导致DSH结构不完整有一批比较老的任务记录没有assignee字段也没有subtasks字段。最初版本的flatten_task用task[assignee]直接取值遇到缺失字段就抛KeyError整个批次转换中断。后来统一改成task.get(assignee)并用if判断是否存在缺失字段就跳过输出。这里要特别注意一个约定缺失字段到底是不输出还是输出空值我最终选择不输出原因有两个。一是DSH文件更干净下游解析器不需要处理大量空行二是下游平台约定对缺失字段采用默认值不输出和输出默认值效果等价。但这个约定一定要提前和下游系统的负责人确认两边口径不一致是最隐蔽的错误来源。5.3 大文件转换超时的优化单次导出的JSON如果包含数千个任务字符串拼接的性能就会成为瓶颈。一开始我用text line \n的方式逐行拼接转换一个3000条任务的导出文件耗时接近10秒。后来改成把所有行放进列表最后用\n.join(lines)一次性拼接耗时降到1秒以内。这个优化原理很简单Python字符串是不可变对象反复用拼接会不断创建新字符串是O(n²)的复杂度而join是一次性分配内存做拼接复杂度是O(n)。对于这种行数在万级以内的转换任务join方案已经完全够用不需要引入更复杂的流式写入方案。如果你的数据量到了几十万行再考虑分批写入或者用生成器逐行输出。5.4 常见异常速查表把这段时间遇到的异常和排查方法整理成一个速查表应该能帮你少走不少弯路现象可能原因排查方法中文变成乱码文件读写编码不一致统一使用utf-8或utf-8-sig读写KeyError: assignee旧数据缺少嵌套字段改用get加存在性判断JSONDecodeError文件带BOM头或空文件先检查文件大小再用utf-8-sig读取时间戳差8小时fromisoformat解析纯日期时按本地时区纯日期字段显式补充时区再转时间戳输出文件有奇怪换行原始字段含换行符输出前对值做转义处理批量转换中途失败单文件异常未捕获每个文件单独try-except转换很慢字符串用反复拼接改用列表加join一次性拼接这张表看起来每一条都很简单但每一条都是真实踩过坑才总结出来的。格式转换的项目表面上在写代码实际上大部分时间都花在和各种异常格式打交道。提前把这张表贴在项目文档里后续无论是自己维护还是交给别人接手都能省很多排查时间。6. 沉淀下来的心得与后续扩展6.1 这个小工具带来的意外复用workbuddy-to-dsh这个小工具跑到现在稳定性和效率都达到预期。最让我意外的是它不仅在原定的分析平台对接场景里用上了后来有同事要把历史任务数据导进一个新的看板系统也是直接拿这个脚本改了改字段映射就搞定了。可见格式转换这种需求一旦沉淀成工具复用价值远大于写代码时付出的那半天功夫。这个复用过程也让我意识到转换脚本的核心不是某一个字段的映射而是那个“拍平嵌套结构、统一字段类型、规范输出格式”的框架。换一个目标格式改的只是FIELD_MAP和序列化函数整个处理流程完全不用动。6.2 对同类转换工具的三个坚持经历过这个项目之后我对写格式转换类工具形成了三个坚持。第一字段映射表必须单独留档不光是写在代码注释里最好单独建一页文档。因为几个月之后你回来看这个项目能快速让你想起当时为什么这么映射的只有那份映射表。第二输出文件一定要带元信息头。我给DSH输出文件的头部加了一行格式为# sourcexxx.json exported2024-06-01T00:00:0008:00的注释行这个改动在排障的时候帮了大忙下游一旦发现某条数据对不上直接看DSH文件头就能定位到是哪一份源文件生成的不用再猜来猜去。第三不要一开始就追求完美。先把主流程跑通用真实数据验证再根据报错一点点补边界情况比一开始就把所有异常可能性都考虑全要高效得多因为很多边界情况你不实际碰到是根本想象不到的。数据转换工程的可靠性往往就是从这些不起眼的元信息和映射文档里来的。工具本身可能只有几十行代码但把边界情况处理干净、把追溯信息留全它才能真正从“能用”变成“好用”。