# -*- coding: utf-8 -*- import os import time import oss2 import logging import json path1 = '/mnt/disk001/dcl_data_process/src/python2/pjisuv/' path2 = '/mnt/disk001/dcl_data_process/src/python3/pjisuv/' logging.basicConfig(filename=path1 + 'log/xosc-pjisuv.log', level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') key1 = 'pjisuv/' sleep_time = 60 # 每多少秒扫描一次 error_bag_json = "/mnt/disk001/dcl_data_process/src/python2/pjisuv/xosc-errorBag.json" def parse_json_to_string_array(file_path): try: # 打开并读取JSON文件(Python 2中不支持encoding参数,需要使用codecs模块或处理文件读取后的编码) with open(file_path, 'r') as file: # 读取文件内容 file_content = file.read() # 解析JSON内容(Python 2中json.loads用于解析字符串) data = json.loads(file_content.decode('utf-8')) # 假设文件是UTF-8编码,这里需要手动解码 # 检查数据是否是一个列表,并且列表中的元素是否是字符串 if isinstance(data, list): for item in data: if not isinstance(item, basestring): # Python 2中字符串类型包括str和unicode,用basestring检查 raise ValueError("JSON数组中的元素不是字符串") return data else: return [] except Exception as e: return [] def list_to_json_file(data, file_path): """ 将列表转换为JSON格式并写入指定的文件路径。 如果文件已存在,则覆盖它。 参数: data (list): 要转换为JSON的列表。 file_path (str): 要写入JSON数据的文件路径。 """ # 将列表转换为JSON格式的字符串,并确保输出为UTF-8编码的字符串 json_data = json.dumps(data, ensure_ascii=False, indent=4) json_data_utf8 = json_data.encode('utf-8') # 编码为UTF-8 # 以写入模式打开文件,如果文件已存在则覆盖 with open(file_path, 'w') as file: # 将UTF-8编码的JSON字符串写入文件 file.write(json_data_utf8) def generate_xosc(parse_prefix, local_parse_dir, local_delete_list): try: os.chdir(path2) # todo simulation_hmi 实际生成的 orig,因为多功能车没有 hmi command2 = 'python3 simulation_hmi.py ' + local_parse_dir[:-1] # 生成orig logging.info("进入目录 %s 调用命令2: %s", path2, str(command2)) os.system(command2) local_xosc_path2 = local_parse_dir + 'simulation_orig/xosc/openx0.xosc' bucket.put_object_from_file(parse_prefix + 'scenario_orig.xosc', local_xosc_path2) bucket.put_object_from_file(parse_prefix + 'scenario_hmi.xosc', local_xosc_path2) # todo 将orig也上传成hmi,因为多功能车暂时没有hmi logging.info("上传 scenario_orig.xosc 成功: %s" % str(parse_prefix + 'scenario_orig.xosc')) local_delete_list.append(local_xosc_path2) except Exception as e: error_bag_list = parse_json_to_string_array(error_bag_json) error_bag_list.append(parse_prefix) list_to_json_file(error_bag_list, error_bag_json) logging.exception(parse_prefix, "生成 xosc 报错,将其添加到错误bag列表。", str(e)) # ------- 获取合并之后的bag包,解析出csv ------- if __name__ == '__main__': # 1 创建阿里云对象 auth = oss2.Auth('n8glvFGS25MrLY7j', 'xZ2Fozoarpfw0z28FUhtg8cu0yDc5d') endpoint = 'oss-cn-beijing-gqzl-d01-a.ops.gqzl-cloud.com' bucket = oss2.Bucket(auth, endpoint, 'open-bucket') while True: try: logging.info("开始新一轮扫描") prefix_list = [] # 2 获取已经上传完成的所有目录并分组 all_prefix_set = set() for obj1 in oss2.ObjectIterator(bucket, prefix=key1): # 获取csv if 'data_parse' in str(obj1.key) and str(obj1.key).count('/') == 4: all_prefix_set.add('/'.join(str(obj1.key).split('/')[:-1]) + '/') for obj2_key in all_prefix_set: try: local_delete_list = [] oss_delete_list = [] xosc1_done = False xosc2_done = False csv1_done = False csv2_done = False for obj3 in oss2.ObjectIterator(bucket, prefix=obj2_key): if '/scenario_orig.xosc' in str(obj3.key): xosc1_done = True # if '/scenario_hmi.xosc' in str(obj3.key): # xosc2_done = True if '/pos_orig.csv' in str(obj3.key): csv1_done = True # if '/pos_hmi.csv' in str(obj3.key): # csv2_done = True if xosc1_done or not csv1_done: continue error_bag_list = parse_json_to_string_array(error_bag_json) if str(obj2_key) in error_bag_list: continue logging.info("需要生成xosc: %s" % obj2_key) local_dir_full = path1 + obj2_key if not os.path.exists(local_dir_full): os.makedirs(local_dir_full) # todo 下载单个csv生成两个视频? bucket.get_object_to_file(obj2_key + 'pos_orig.csv', local_dir_full + 'pos_orig.csv') # bucket.get_object_to_file(obj2_key+'pos_hmi.csv', local_dir_full+'pos_hmi.csv') # local_delete_list.append(local_dir_full + 'pos_orig.csv') # todo 不删除了 做测试用 # local_delete_list.append(local_dir_full+'pos_hmi.csv') generate_xosc(obj2_key, local_dir_full, local_delete_list) # 删除本地临时文件 if len(local_delete_list) > 0: for local_delete in local_delete_list: try: os.remove(local_delete) except Exception as e: logging.exception("删除本地临时文件报错: %s" % str(e)) except Exception as e: logging.exception("局部异常处理: %s", str(e)) time.sleep(sleep_time) except Exception as e: logging.exception("全局异常处理: %s", str(e))