如何按JSON内timedate_saved字段拆分会话记录并按年/月目录重存

实现思路 #

• 遍历根目录下所有公司的comport文件夹，扫描所有enregistrement_开头的JSON文件
• 读取文件内容并解析为JSON对象，提取根节点的session哈希值、enregistrements子数组
• 遍历enregistrements数组内的每个元素，解析timedate_saved字段（格式为日 月 年 时:分:秒），提取年、月数值，月份统一保留2位避免排序混乱
• 自动生成目标存储路径{公司名}/comport/{年}/{月}/，路径不存在时自动递归创建
• 每个enregistrements元素单独序列化为合法JSON格式，写入目标路径下，文件名保留原有enregistrement_{session哈希}规则
• 存量处理完成后可配置定时任务定期处理新生成的文件，避免重复扫描历史文件

可运行Python代码 #

无需额外安装第三方依赖（标准库即可实现），代码内置异常处理兼容损坏文件：

import os
import json
from datetime import datetime

# 配置项，按需修改
ROOT_DIR = "d:/"  # 所有公司目录的根路径
BACKUP_OLD_FILE = True  # 处理完的原始文件是否移到备份目录，避免重复处理
BACKUP_DIR = "d:/json_backup/"

def process_single_file(file_path: str, company_name: str):
    # 跳过空文件
    if os.path.getsize(file_path) == 0:
        return
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except json.JSONDecodeError:
        print(f"文件损坏跳过：{file_path}")
        return
    
    session_hash = data.get("session")
    enregistrements = data.get("enregistrements", [])
    if not session_hash or not enregistrements:
        return
    
    for item in enregistrements:
        timedate_str = item.get("timedate_saved")
        if not timedate_str:
            continue
        try:
            # 解析日期，格式为"27 04 2020 12:39:21"
            save_time = datetime.strptime(timedate_str, "%d %m %Y %H:%M:%S")
            year = str(save_time.year)
            month = f"{save_time.month:02d}"
        except ValueError:
            print(f"日期格式错误跳过：{timedate_str}，文件：{file_path}")
            continue
        
        # 生成目标路径
        target_dir = os.path.join(ROOT_DIR, company_name, "comport", year, month)
        os.makedirs(target_dir, exist_ok=True)
        target_file = os.path.join(target_dir, f"enregistrement_{session_hash}")
        
        # 写入合法JSON
        with open(target_file, "w", encoding="utf-8") as f:
            json.dump(item, f, ensure_ascii=False, indent=2)
    
    # 原始文件备份
    if BACKUP_OLD_FILE:
        os.makedirs(BACKUP_DIR, exist_ok=True)
        os.rename(file_path, os.path.join(BACKUP_DIR, os.path.basename(file_path)))

if __name__ == "__main__":
    # 遍历所有公司目录
    for company_name in os.listdir(ROOT_DIR):
        company_comport_dir = os.path.join(ROOT_DIR, company_name, "comport")
        if not os.path.isdir(company_comport_dir):
            continue
        # 遍历comport下所有文件
        for file_name in os.listdir(company_comport_dir):
            if not file_name.startswith("enregistrement_"):
                continue
            file_path = os.path.join(company_comport_dir, file_name)
            if not os.path.isfile(file_path):
                continue
            process_single_file(file_path, company_name)
    print("全部处理完成")

注意事项 #

• 全量处理前务必先备份所有原始存量文件，避免操作失误丢失数据
• 可先注释掉备份逻辑，拿少量测试文件验证路径生成、JSON格式、日期解析均符合要求后再全量运行
• 全量处理数据量较大时，可安装tqdm库添加进度条，方便查看处理进度
• 新文件的增量处理可配置定时任务：Linux用crontab、Windows用计划任务，每1~2小时执行一次脚本即可

内容的提问来源于stack exchange，提问作者Théo Derville