dcl_data_process/src/python2/pjisuv/callback-pjisuv.py

# -*- coding: utf-8 -*-
import json
import time
import urllib2
import oss2
from datetime import datetime, timedelta
import logging

path1 = '/mnt/disk001/dcl_data_process/src/python2/pjisuv/'

logging.basicConfig(filename=path1 + 'log/callback-pjisuv.log', level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s')

key1 = 'pjisuv/'
sleep_time = 60  # 每多少秒扫描一次

url1_private = "http://10.14.86.127:9081/device/auth"
url2_private = "http://10.14.86.127:9081/device/data/callback"

error_bag_json = "/mnt/disk001/dcl_data_process/src/python2/pjisuv/callback-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 add_hour(date_string, hour_number):
    original_date = datetime.strptime(date_string, "%Y-%m-%d-%H-%M-%S")
    new_date = original_date + timedelta(hours=hour_number)
    return new_date.strftime("%Y-%m-%d-%H-%M-%S")


if __name__ == '__main__':
    # 1 登录验证 。
    auth = oss2.Auth('n8glvFGS25MrLY7j', 'xZ2Fozoarpfw0z28FUhtg8cu0yDc5d')

    # 2 填写自定义域名，例如example.com。获取桶。
    # 3 填写Bucket名称，并设置is_cname=True来开启CNAME。CNAME是指将自定义域名绑定到存储空间。
    endpoint = 'oss-cn-beijing-gqzl-d01-a.ops.gqzl-cloud.com'
    bucket_name = 'open-bucket'
    bucket = oss2.Bucket(auth, endpoint, bucket_name)

    while True:
        logging.info("开始新一轮扫描")
        local_delete_list = []
        oss_delete_list = []
        upload_completed_prefix_list = []
        # 4 获取即将被合并的bag目录
        logging.info("开始扫描目录： %s" % str(key1))
        for obj1 in oss2.ObjectIterator(bucket, prefix=key1):
            if 'callback.json' in str(obj1.key):
                time.sleep(1)
                prefix = '/'.join(str(obj1.key).split('/')[:-1])

                pos_orig_csv = False
                pos_hmi_csv = False
                drive_csv = False
                camera_mp4 = False
                pcd_forwardlook_mp4 = False
                pcd_overlook_mp4 = False
                scenario_orig_xosc = False
                scenario_hmi_xosc = False
                scenario_orig_mp4 = False
                scenario_hmi_mp4 = False
                camera_bag = False
                fusion_bag = False
                plan_bag = False
                control_bag = False
                callback_json = False
                for obj2 in oss2.ObjectIterator(bucket, prefix=prefix + '/'):
                    if 'callback.json' in str(obj2.key):
                        callback_json = True
                    if 'camera.mp4' in str(obj2.key):
                        camera_mp4 = True
                    if 'drive.csv' in str(obj2.key):
                        drive_csv = True
                    if 'pcd_forwardlook.mp4' in str(obj2.key):
                        pcd_forwardlook_mp4 = True
                    if 'pcd_overlook.mp4' in str(obj2.key):
                        pcd_overlook_mp4 = True
                    if 'pos_orig.csv' in str(obj2.key):
                        pos_orig_csv = True
                    if 'scenario_orig.mp4' in str(obj2.key):
                        scenario_orig_mp4 = True
                    if 'scenario_orig.xosc' in str(obj2.key):
                        scenario_orig_xosc = True

                if not callback_json or not camera_mp4 or not drive_csv or not pcd_forwardlook_mp4 or not pcd_overlook_mp4 or not pos_orig_csv or not scenario_orig_mp4 or not scenario_orig_xosc:
                    continue
                time.sleep(2)
                logging.info("发送： %s" % str(obj1.key))
                # 1 获取json内容
                json_content = bucket.get_object(str(obj1.key)).read()
                # 2 获取token
                json_object = json.loads(json_content)
                data1 = {
                    "equipmentNo": json_object['equipmentNo'],
                    "secretKey": json_object['secretKey']
                }
                json_data1 = json.dumps(data1)
                request1 = urllib2.Request(url1_private, json_data1,
                                           headers={'Content-Type': 'application/json'})
                response1 = urllib2.urlopen(request1)
                result_json1 = response1.read()
                result_object1 = json.loads(result_json1)
                try:
                    access_token = result_object1.get('data').get('accessToken')
                    logging.info("bag文件为：%s" % str(json_object['rosBagPath']))
                    old_date = json_object['dataName']
                    equipment_no = json_object['equipmentNo']
                    old_file_path = json_object['filePath']
                    old_ros_bag_path = json_object['rosBagPath']
                    task_id = json_object['taskId']
                    trigger_id = json_object['triggerId']
                except Exception as e:
                    logging.exception("callback报错： %s" % str(e))
                    continue

                upload = False
                if 'userId' in json_object:
                    logging.info("手动上传的数据")
                    upload = True
                    old_date = ''
                else:
                    logging.info("自动采集的数据")
                    upload = False

                old_delete_list = []
                new_date = ''
                # 复制 data_parse
                for obj_old in oss2.ObjectIterator(bucket, prefix=old_file_path):
                    old_delete_list.append(str(obj_old.key))
                    if 'callback.json' in str(obj_old.key):
                        if not upload:
                            new_date = add_hour(old_date, 8)
                            # 将时区统一
                            bucket.copy_object(bucket_name, str(obj_old.key),
                                               str(obj_old.key).replace(old_date, new_date).replace('callback.json',
                                                                                                    'callback_done.json'))
                        else:
                            bucket.copy_object(bucket_name, str(obj_old.key),
                                               str(obj_old.key).replace('callback.json',
                                                                        'callback_done.json'))

                    else:
                        if not upload:
                            new_date = add_hour(old_date, 8)
                            bucket.copy_object(bucket_name, str(obj_old.key),
                                               str(obj_old.key).replace(old_date, new_date))

                # 处理是否上传
                if not upload:
                    try:
                        bucket.copy_object(bucket_name, old_ros_bag_path, old_ros_bag_path.replace(old_date, new_date))
                        bucket.delete_object(old_ros_bag_path)
                        bucket.batch_delete_objects(old_delete_list)
                    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)
                        continue

                if upload:
                    data2 = {
                        'userId': json_object['userId'],
                        "dataName": new_date,
                        "dataSize": json_object['dataSize'],
                        "equipmentNo": equipment_no,
                        "filePath": old_file_path.replace(old_date, new_date),
                        "rosBagPath": old_ros_bag_path.replace(old_date, new_date),
                        "taskId": task_id,
                        "triggerId": trigger_id
                    }
                else:
                    data2 = {
                        "dataName": new_date,
                        "dataSize": bucket.get_object_meta(json_object['rosBagPath']).content_length,
                        "equipmentNo": equipment_no,
                        "filePath": old_file_path.replace(old_date, new_date),
                        "rosBagPath": old_ros_bag_path.replace(old_date, new_date),
                        "taskId": task_id,
                        "triggerId": trigger_id
                    }
                json_data2 = json.dumps(data2)
                logging.info("回调接口请求中：%s" % url2_private)
                request2 = urllib2.Request(url2_private, json_data2,
                                           headers={'Content-Type': 'application/json',
                                                    'authorization': access_token})

                response2 = urllib2.urlopen(request2)
                result_json2 = response2.read()
                result_object2 = json.loads(result_json2)
                logging.info("回调接口请求结果为： %s", result_object2)
        time.sleep(sleep_time)