历史数据导入导出

# -*- coding: UTF-8 -*-
"""
@Description: 处理从 sa 的 query/sql 导出的数据
"""
 
import json
import argparse
 
BATCH_SIZE = 10000  # 可根据内存调整
 
def process_file(input_file, output_file, error_file):
    buffer = []
    error_buffer = []
    try:
        with open(input_file, 'r', encoding='utf-8') as infile, \
                open(output_file, 'w', encoding='utf-8') as outfile, \
                open(error_file, 'w', encoding='utf-8') as errfile:
 
            for line_number, line in enumerate(infile, start=1):
                try:
                    data = json.loads(line)
                    if 'properties' in data and need_process(data):
                        properties = data['properties']
                        identities = {}
 
                        # 移除 $idmap_reason
                        properties.pop('$idmap_reason', None)
 
                        # 提取 identity_ 字段
                        for key in list(properties.keys()):
                            if key.startswith('identity_') or key.startswith('$identity_'):
                                value = properties.pop(key)
                                if isinstance(value, list) and len(value) == 1:
                                    value = value[0]
                                identities[key] = value
 
                        if identities:
                            if len(identities) == 1 and '$identity_distinct_id' in identities:
                                distinct_id_value = identities['$identity_distinct_id']
                                if not isinstance(distinct_id_value, str):
                                    data['identities'] = identities
                            else:
                                data['identities'] = identities
 
                    buffer.append(json.dumps(data, ensure_ascii=False) + '\n')
 
                except Exception as e:
                    error_buffer.append(f"Line {line_number}: {str(e)}\n")
 
                # 批量写入
                if len(buffer) >= BATCH_SIZE:
                    outfile.writelines(buffer)
                    buffer.clear()
                if len(error_buffer) >= BATCH_SIZE:
                    errfile.writelines(error_buffer)
                    error_buffer.clear()
 
            # 写入剩余
            if buffer:
                outfile.writelines(buffer)
            if error_buffer:
                errfile.writelines(error_buffer)
    except FileNotFoundError as fnfe:
        print(f"Error: {str(fnfe)}")
 
def need_process(data):
    if 'identities' in data:
        return False
    if 'version' in data and data['version'] == 2:
        return False
    return True
 
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description='处理从 sa 的 query/sql 导出的数据')
    parser.add_argument('--input_file', type=str, help='输入文件名')
    parser.add_argument('--output_file', type=str, default='processed_event_output.json', help='输出文件名,默认值 processed_output.json')
    parser.add_argument('--error_file', type=str, default='default_event_error_output.txt', help='错误文件名,默认值为 default_error_output.txt')
    args = parser.parse_args()
    process_file(args.input_file, args.output_file, args.error_file)