做数据迁移或者系统对接的时候，最烦的就是两边格式对不上。最近我在做一个数据对接的小项目，需要把WorkBuddy导出的任务记录转换成分析平台要求的DSH格式。WorkBuddy是团队一直在用的任务与工时管理软件，它导出的JSON结构记录着任务、子任务、评论、负责人这些信息；而下游的数据分析平台只接受DSH，一种按行为单位、以key=value形式组织的结构化文本格式。两个系统之间没有官方转换工具，数据要进入分析管道，只能自己写转换程序。

这篇文章就把我完整实现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:00+08:00", "due_date": "2024-05-10", "assignee": { "name": "张三", "email": "zhangsan@example.com", "team": "BI组" }, "subtasks": [ {"title": "清洗数据", "done": true, "hours": 3}, {"title": "制作图表", "done": true, "hours": 2.5} ], "comments": [ {"author": "李四", "content": "注意统计口径", "time": "2024-05-03T10:00:00+08:00"} ] } ], "exported_at": "2024-06-01T00:00:00+08:00" }

这种结构对人类很友好，但对下游系统并不友好。分析平台要的是扁平的、列名清晰的输入，它不可能自己去解析JSON里多层嵌套的assignee、subtasks和comments。如果直接把这份JSON喂给平台，平台解析器要么直接拒绝，要么只取到最外层字段，任务详情、工时、负责人这些关键信息全部丢失。所以转换的第一步，就是把这棵“树”拍平成一条“线”。

1.2 DSH格式到底有什么要求

DSH这种格式，简单说就是每一行一个键值对，字段用等号连接，不同记录之间用分隔行隔开。文件开头可以写一个类型声明行，告诉解析器这批数据的版本和类型。我这边对接的平台要求的格式是这样：

task:record:v1 id=T-202405-001 title=完成季度数据报表 project=数据组 status=done priority=2 estimate_hours=8.5 spent_hours=8.5 created_at=1714613400 due_date=1715270400 assignee=张三|BI组|zhangsan@example.com tags=报表,季度 subtask_count=2 subtask_done=2 comment_count=1 ---

特点很明显：第一，所有字段都是key=value的文本行，解析起来不需要考虑嵌套层级，按行读、按等号切分就能还原；第二，时间统一成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脚本，建议在代码里显式指定encoding="utf-8"，不要依赖系统默认编码。

3. 核心转换逻辑：从JSON到DSH的完整路径

3.1 字段映射表先想清楚

转换的第一步不是写代码，而是把字段映射表列出来。哪些字段要原样保留，哪些字段要改名，哪些字段要用嵌套数据计算出来，都先想好。我做了一张这样的对照表：

这张表是整个脚本的地基。写代码之前先做映射，看起来是个笨功夫，实际上能避免绝大多数低级错误。比如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_done，comments变成comment_count，典型应用场景是下游只需要“有多少”“完成多少”这类汇总信息。第二种是拼接，把数组或对象用分隔符串成单行字符串，比如tags用逗号拼，assignee用竖线拼，典型应用场景是下游需要看到内容但不需要单独拆列。第三种是跳过，如果嵌套内容对下游完全没用，直接不输出，省得DSH文件里堆一堆垃圾行。

这三种手法没有绝对的好坏，完全看下游系统需要什么。我最初想把评论内容也拼进DSH，后来发现平台根本不消费这个字段，就把它砍掉了，只保留comment_count。少一个字段，解析和写入都少一分出错的可能。如果你在做的也是类似的转换项目，建议先问清楚下游到底消费哪些字段，再决定用哪种手法，不要想当然地“全都带上”。

3.3 时间字段的规范化是个大坑

时间处理是整个项目里最容易被忽视、也最值得展开说的一块。WorkBuddy导出的是带时区的ISO8601格式，比如2024-05-02T09:30:00+08: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(hours=8)) def date_to_epoch(value): dt = datetime.strptime(value, "%Y-%m-%d").replace(tzinfo=CST) return str(int(dt.timestamp()))

这种“统一时区”的思路，在做跨系统数据交换时尤其重要。只要DSH文件里的时间戳全部基于同一个时区换算，下游不管在哪个国家、哪台服务器上解析，拿到的都是同一个绝对时间点。我在项目里还专门写了一个测试用例，把同一份数据放到东八区和UTC时区的两台机器上跑，验证输出时间戳完全一致，才敢放心交接给下游。

3.4 序列化输出与自校验

拍平后的记录要转成DSH文本，最后一个关键环节是序列化。我用的方法是逐行拼key=value，先拼标题行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(parents=True, exist_ok=True) 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, encoding="utf-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平台解析，两边编码对不上，中文就全部变成了问号或者乱码。

解决方式很简单，所有文件读写都显式指定encoding="utf-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 常见异常速查表

把这段时间遇到的异常和排查方法整理成一个速查表，应该能帮你少走不少弯路：

这张表看起来每一条都很简单，但每一条都是真实踩过坑才总结出来的。格式转换的项目表面上在写代码，实际上大部分时间都花在和各种异常格式打交道。提前把这张表贴在项目文档里，后续无论是自己维护还是交给别人接手，都能省很多排查时间。

6. 沉淀下来的心得与后续扩展

6.1 这个小工具带来的意外复用

workbuddy-to-dsh这个小工具跑到现在，稳定性和效率都达到预期。最让我意外的是，它不仅在原定的分析平台对接场景里用上了，后来有同事要把历史任务数据导进一个新的看板系统，也是直接拿这个脚本改了改字段映射就搞定了。可见格式转换这种需求一旦沉淀成工具，复用价值远大于写代码时付出的那半天功夫。

这个复用过程也让我意识到，转换脚本的核心不是某一个字段的映射，而是那个“拍平嵌套结构、统一字段类型、规范输出格式”的框架。换一个目标格式，改的只是FIELD_MAP和序列化函数，整个处理流程完全不用动。

6.2 对同类转换工具的三个坚持

经历过这个项目之后，我对写格式转换类工具形成了三个坚持。第一，字段映射表必须单独留档，不光是写在代码注释里，最好单独建一页文档。因为几个月之后你回来看这个项目，能快速让你想起当时为什么这么映射的，只有那份映射表。第二，输出文件一定要带元信息头。我给DSH输出文件的头部加了一行格式为# source=xxx.json exported=2024-06-01T00:00:00+08:00的注释行，这个改动在排障的时候帮了大忙：下游一旦发现某条数据对不上，直接看DSH文件头就能定位到是哪一份源文件生成的，不用再猜来猜去。第三，不要一开始就追求完美。先把主流程跑通，用真实数据验证，再根据报错一点点补边界情况，比一开始就把所有异常可能性都考虑全要高效得多，因为很多边界情况你不实际碰到，是根本想象不到的。

数据转换工程的可靠性，往往就是从这些不起眼的元信息和映射文档里来的。工具本身可能只有几十行代码，但把边界情况处理干净、把追溯信息留全，它才能真正从“能用”变成“好用”。