将waae任务报告中页面上的连续帧截图合并为视频，并导出json文件

script_video.py

+import json
+import numpy as np
+import os
+import os.path as osp
+import shutil
+from datetime import date
+from collections import defaultdict as dd
+from moviepy import ImageSequenceClip
+
+import pandas as pd
+
+def create_excel_with_columns(data, file_name="app_report.xlsx"):
+    columns = [
+        "验证码", "非文本链接", "非文本控件", "多媒体", "功能性组件访问",
+        "装饰性内容访问", "非装饰性组件", "漂浮窗", "闪光", "焦点顺序",
+        "错误原因提示",
+    ]
+    
+    expanded_columns = []
+    for col in columns:
+        expanded_columns.append(f"{col}-预测")
+        expanded_columns.append(f"{col}-真实")
+    new_index = []
+    df = pd.DataFrame(index=range(len(data)), columns=expanded_columns)
+    for id, screen in enumerate(data):
+        for id1, k in enumerate(data[screen]['通过情况']):
+            df.iloc[id, 2*id1] = k
+        new_index.append(screen)
+
+    df.index = new_index
+    df.to_excel(file_name, index=True, engine='openpyxl')
+    
+    print('end')
+
+
+# 处理通过情况
+def process_data(data, entry):
+    result = [1] * len(entry)
+    for k in data:
+        result[k] = 0
+    
+    return result
+
+def process_page(data, page_id):
+    # 根据page_id，确定页面截图和主page的路径
+    page = data[int(page_id)]
+    
+    screenshot_path = [os.path.basename(box["local_image_path_full"]) for box in page["focus_boxes"]]
+
+    image_path = page['img_path']
+    
+    return screenshot_path, image_path
+
+def process(file_list, data, entry, video_save_path, save_path):
+    result_finial = {}
+    cnt = 0
+
+    for file in file_list:
+        result = dd(list) # 记录通过情况id
+        result2 = dd(list) # 记录错误条目名称
+        # print(osp.join(file, 'report.json'))
+        report = json.load(open(osp.join(file, 'report.json'), 'r', encoding='utf-8'))
+        try:
+            errors = report['check_result']['errors']
+            pages =  report['check_result']['pages']
+            app_name = report['appName']
+            task_id = report['hcTaskId']
+        except:
+            continue
+        for k in errors:
+            screen = str(k['screenshot_id'])
+            page_id = str(k['page_id'])
+            rule_id = k['rule_id']
+            if rule_id not in list(data.keys()):
+                continue
+            # 判断图片是否存在
+            if not osp.exists(osp.join(file, 'screenshots', screen + '.png')):
+                continue
+            error_reson = k['error_reason']
+            s = f'{data[rule_id][1]}.{entry[data[rule_id][1]]}'
+            result[page_id].append(s)
+            result2[page_id].append(data[rule_id][1])
+
+        for page_id in result:
+            id = f'{save_path}_{cnt}'
+            video_path = osp.join(video_save_path, f'{app_name}-{task_id}-{page_id}.mp4')
+            screenshot_path, image_path = process_page(pages, page_id)
+            image_path = osp.join(file, 'screenshots', image_path.split('/')[-1])
+            
+            img_path = osp.join(save_path, f'{id}.png')
+
+            # 当前页面的所有截图
+            screenshot_path = [osp.join(file, 'screenshots',item) for item in screenshot_path]
+            
+            # 合成视频
+            images_to_video(screenshot_path,video_path,fps=1)
+
+            result_finial[id] = {'页面名': f'{id}', '合成视频路径': video_path, 
+                                 '页面截图路径': image_path, 
+                                 '所有截图路径': screenshot_path}
+            shutil.copy(image_path, img_path)
+            result_finial[id]['判断条目'] = entry
+            result_finial[id]['错误条目'] = result[page_id]
+            result_finial[id]['通过情况'] = process_data(result2[page_id], entry)
+
+            cnt += 1
+        # print(result_finial, )
+        print(len(result_finial))
+    return result_finial
+
+def load_file_path(path):
+    file_list = []
+    for i in os.listdir(path):
+        if i.startswith('8') and len(i)==4:
+            file_list.append(osp.join(path, i))
+
+    return file_list
+
+def images_to_video(image_files, output_video_path, fps=1):
+    
+    clip = ImageSequenceClip(image_files, fps=fps)
+    clip.write_videofile(output_video_path)
+    print(f"视频已成功保存至：{output_video_path}")
+
+
+def run(tast_dir='task_dir',map_dict_path='app_video_map.txt'):
+    save_path = str(date.today()).replace('-', '_')
+    video_save_path = osp.join(save_path, 'videos')
+    report_save_path = osp.join(save_path, 'app_video_report.json')
+    report_save_path_excel = osp.join(save_path, 'app_video_report.xlsx')
+    os.makedirs(video_save_path, exist_ok=True)
+
+    file_list = load_file_path(tast_dir)
+
+    with open(map_dict_path, 'r') as f:
+        data1 = f.readlines()
+    data = dd(list)
+    keys = dd()
+    entry = []
+    for k in data1:
+        k = k[:-1].split(' ')
+        try:
+            data[int(k[0])].append(k[1])
+            if k[1] not in keys:
+                keys[k[1]] = len(keys)
+                entry.append(k[1])
+        except:
+            pass
+    
+    for k in data.keys():
+        data[k].append(keys[data[k][0]])
+        
+    result_finial = process(file_list, data, entry, video_save_path, save_path)
+
+    json.dump(result_finial, open(report_save_path, 'w', encoding='utf-8'), indent=4, ensure_ascii=False)
+
+    create_excel_with_columns(result_finial, report_save_path_excel)
+
+if __name__ == '__main__':
+    # TODO: 使用时，传入waae数据根目录(task_dir)和检测的条目(app_video_map.txt)
+    run()
\ No newline at end of file