import os import csv from collections import defaultdict def _decode_mixed_csv_line(raw_line): for encoding in ('utf-8-sig', 'utf-8', 'gb18030'): try: return raw_line.decode(encoding) except UnicodeDecodeError: continue return raw_line.decode('gb18030', errors='replace') def _read_csv_lines(csv_path): with open(csv_path, 'rb') as f: return [_decode_mixed_csv_line(line) for line in f] def _csv_value(row, index): if index >= len(row): return '' return str(row[index] or '').strip() def _latest_task_key(task_time, seq): task_time = str(task_time or '').strip() return (1 if task_time else 0, task_time, seq) def _load_latest_task_rows(task_paths, package_idx, time_idx, allowed_packages=None): rows_by_package = {} best_keys = {} seq = 0 for task_path in task_paths: if not os.path.exists(task_path): continue try: reader = csv.reader(_read_csv_lines(task_path)) next(reader, None) for row in reader: package_name = _csv_value(row, package_idx) if allowed_packages is not None and package_name not in allowed_packages: continue if not package_name: continue seq += 1 key = _latest_task_key(_csv_value(row, time_idx), seq) if package_name in best_keys and key <= best_keys[package_name]: continue best_keys[package_name] = key rows_by_package[package_name] = row except Exception as e: print(f"读取 {task_path} 时出错: {e}") return rows_by_package def _normalize_success_detail(detail_info): detail_info = str(detail_info or '').strip() if '|' not in detail_info: return detail_info parts = detail_info.split('|', 1) if len(parts) > 1: return parts[1].strip() return detail_info def extract_second_level_domain(domain): if not domain: return "" domain = str(domain).lower().replace('http://', '').replace('https://', '').strip().strip('"\'') parts = domain.split('.') if len(parts) <= 2: return domain return "*." + ".".join(parts[-2:]) def load_package_mapping(current_dir): package_to_app = {} app_to_package = {} package_list_path = os.path.join(current_dir, 'package_list.csv') if not os.path.exists(package_list_path): print("未找到 package_list.csv，将使用包名作为应用名。") return None, None try: with open(package_list_path, 'r', encoding='utf-8-sig', newline='') as f: reader = csv.DictReader(f) for row in reader: app_name = (row.get('app_name') or '').strip() package_name = (row.get('package_name') or '').strip() if package_name and app_name: package_to_app[package_name] = app_name app_to_package[app_name] = package_name except Exception as e: print(f"读取 package_list.csv 时出错: {e}") return None, None return package_to_app, app_to_package def load_traffic_profile(current_dir, allowed_packages=None): self_ratio_map = {} traffic_profile_path = os.path.join(current_dir, 'app_traffic_profile_20260324.csv') if not os.path.exists(traffic_profile_path): print("未找到 app_traffic_profile_20260324.csv，将不使用 self_ratio。") return self_ratio_map try: with open(traffic_profile_path, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row in reader: package_name = row.get('app_id', '').strip() if allowed_packages is not None and package_name not in allowed_packages: continue if package_name: try: self_ratio = float(row.get('self_ratio', 0) or 0) self_ratio_map[package_name] = self_ratio except ValueError: pass except Exception as e: print(f"读取 app_traffic_profile_20260224.csv 时出错: {e}") return self_ratio_map def load_success_tasks(current_dir, allowed_packages=None): import glob success_packages = set() success_info = {} success_time = {} success_worker = {} success_task_files = sorted(glob.glob(os.path.join(current_dir, 'success_tasks*.csv'))) if not success_task_files: return success_packages, success_info, success_time, success_worker for package_name, row in _load_latest_task_rows( success_task_files, package_idx=3, time_idx=0, allowed_packages=allowed_packages, ).items(): success_packages.add(package_name) success_info[package_name] = _normalize_success_detail(_csv_value(row, 6)) success_time[package_name] = _csv_value(row, 0) success_worker[package_name] = _csv_value(row, 4) return success_packages, success_info, success_time, success_worker def load_non_success_tasks(current_dir, allowed_packages=None): failure_info = {} failure_type = {} failure_time = {} failure_worker = {} non_success_rows = {} failed_rows = _load_latest_task_rows( [os.path.join(current_dir, 'failed_tasks.csv')], package_idx=3, time_idx=0, allowed_packages=allowed_packages, ) retry_rows = _load_latest_task_rows( [os.path.join(current_dir, 'retry_tasks.csv')], package_idx=3, time_idx=0, allowed_packages=allowed_packages, ) for package_name, row in failed_rows.items(): non_success_rows[package_name] = { 'detail': _csv_value(row, 7) if len(row) >= 8 else _csv_value(row, 6), 'failure_type': _csv_value(row, 5), 'time': _csv_value(row, 0), 'worker': _csv_value(row, 4), } for package_name, row in retry_rows.items(): if package_name in non_success_rows: continue non_success_rows[package_name] = { 'detail': _csv_value(row, 6), 'failure_type': _csv_value(row, 5), 'time': _csv_value(row, 0), 'worker': _csv_value(row, 4), } for package_name, row in non_success_rows.items(): failure_info[package_name] = row['detail'] failure_type[package_name] = row['failure_type'] failure_time[package_name] = row['time'] failure_worker[package_name] = row['worker'] return failure_info, failure_type, failure_time, failure_worker def parse_utg_js(js_path, target_package=None): import re import json result = { 'num_nodes': 0, 'num_reached_activities': 0, 'app_num_total_activities': 0 } try: with open(js_path, 'r', encoding='utf-8-sig') as f: content = f.read() json_match = re.search(r'var\s+utg\s*=\s*(\{.*\})\s*;?\s*$', content, re.DOTALL) if json_match: try: utg_data = json.loads(json_match.group(1)) app_package = utg_data.get('app_package', '') filter_package = target_package if target_package else app_package nodes = utg_data.get('nodes', []) if filter_package: filtered_count = sum(1 for node in nodes if node.get('package') == filter_package) result['num_nodes'] = filtered_count else: result['num_nodes'] = len(nodes) num_reached_match = utg_data.get('num_reached_activities', 0) result['num_reached_activities'] = num_reached_match if isinstance(num_reached_match, int) else 0 app_total_match = utg_data.get('app_num_total_activities', 0) result['app_num_total_activities'] = app_total_match if isinstance(app_total_match, int) else 0 except json.JSONDecodeError as e: print(f"解析 UTG JSON 时出错: {e}") num_nodes_match = re.search(r'"num_nodes"\s*:\s*(\d+)', content) if num_nodes_match: result['num_nodes'] = int(num_nodes_match.group(1)) num_reached_match = re.search(r'"num_reached_activities"\s*:\s*(\d+)', content) if num_reached_match: result['num_reached_activities'] = int(num_reached_match.group(1)) app_total_match = re.search(r'"app_num_total_activities"\s*:\s*(\d+)', content) if app_total_match: result['app_num_total_activities'] = int(app_total_match.group(1)) else: num_nodes_match = re.search(r'"num_nodes"\s*:\s*(\d+)', content) if num_nodes_match: result['num_nodes'] = int(num_nodes_match.group(1)) num_reached_match = re.search(r'"num_reached_activities"\s*:\s*(\d+)', content) if num_reached_match: result['num_reached_activities'] = int(num_reached_match.group(1)) app_total_match = re.search(r'"app_num_total_activities"\s*:\s*(\d+)', content) if app_total_match: result['app_num_total_activities'] = int(app_total_match.group(1)) except Exception as e: print(f"解析 UTG 文件 {js_path} 时出错: {e}") return result def process_all_data(): app_data = defaultdict(lambda: { 'domains': set(), 'second_level_domains': set(), 'droidbot_steps': 0, 'gui_agent_steps': 0, 'duration_seconds': 0, 'num_nodes': 0, 'num_reached_activities': 0, 'app_num_total_activities': 0 }) current_dir = os.getcwd() package_to_app, app_to_package = load_package_mapping(current_dir) allowed_packages = set(package_to_app.keys()) if package_to_app is not None else None batch_result_files = [] print("开始遍历文件收集数据...") for root, dirs, files in os.walk(current_dir): for file in files: if file.endswith('.txt'): file_path = os.path.join(root, file) try: with open(file_path, 'r', encoding='utf-8-sig') as f: for line in f: parts = line.strip().split(',') if len(parts) >= 3: package_name = parts[0].strip() app_name = parts[1].strip() domain_name = parts[2].strip() if allowed_packages is not None and package_name not in allowed_packages: continue if domain_name != "model_data:" and package_name: app_data[package_name]['domains'].add(domain_name) second_level_domain = extract_second_level_domain(domain_name) if second_level_domain: app_data[package_name]['second_level_domains'].add(second_level_domain) except Exception as e: print(f"处理文件 {file_path} 时出错: {e}") if file.endswith('.csv') and file.startswith('batch_result'): file_path = os.path.join(root, file) batch_result_files.append(file_path) if file.endswith('_utg.js'): file_path = os.path.join(root, file) package_name = file.replace('_utg.js', '') if allowed_packages is not None and package_name not in allowed_packages: continue utg_result = parse_utg_js(file_path, package_name) app_data[package_name]['num_nodes'] = max( app_data[package_name]['num_nodes'], utg_result['num_nodes'] ) app_data[package_name]['num_reached_activities'] = max( app_data[package_name]['num_reached_activities'], utg_result['num_reached_activities'] ) app_data[package_name]['app_num_total_activities'] = max( app_data[package_name]['app_num_total_activities'], utg_result['app_num_total_activities'] ) batch_result_files.sort(key=lambda x: os.path.getmtime(x)) print(f"找到 {len(batch_result_files)} 个 batch_result 文件，按修改时间排序处理...") for file_path in batch_result_files: try: with open(file_path, 'r', encoding='utf-8-sig') as f: reader = csv.DictReader(f) for row in reader: package_name = row.get('package_name', '').strip() if allowed_packages is not None and package_name not in allowed_packages: continue if not package_name: continue try: droidbot_steps = int(row.get('droidbot_steps', 0) or 0) app_data[package_name]['droidbot_steps'] = droidbot_steps except ValueError: pass try: guiagent_steps = int(row.get('guiagent_steps', 0) or 0) app_data[package_name]['gui_agent_steps'] = guiagent_steps except ValueError: pass try: duration_seconds = float(row.get('duration_seconds', 0) or 0) app_data[package_name]['duration_seconds'] = duration_seconds except ValueError: pass except Exception as e: print(f"处理文件 {file_path} 时出错: {e}") found_packages = set(app_data.keys()) print(f"从文件中共找到 {len(found_packages)} 个包的数据。") success_packages, success_info, success_time, success_worker = load_success_tasks( current_dir, allowed_packages=allowed_packages, ) failure_info, failure_type, failure_time, failure_worker = load_non_success_tasks( current_dir, allowed_packages=allowed_packages, ) self_ratio_map = load_traffic_profile(current_dir, allowed_packages=allowed_packages) if success_packages: print(f"已加载 {len(success_packages)} 个成功任务。") if failure_info: print(f"已加载 {len(failure_info)} 个非成功任务。") if self_ratio_map: print(f"已加载 {len(self_ratio_map)} 个应用的流量配置。") final_packages = set() if package_to_app is not None: final_packages = found_packages.intersection(allowed_packages) print(f"根据 package_list.csv 筛选后，剩余 {len(final_packages)} 个包。") print(f"过滤掉了 {len(found_packages) - len(final_packages)} 个不在列表中的包。") else: final_packages = found_packages print("未进行筛选，使用所有找到的包。") output_file = 'app_domain_summary.csv' headers = ['app_name', 'package_name', 'test_success', 'test_time', 'test_host', 'self_ratio', 'unique_domain_count', 'unique_domain_names', 'unique_second_level_domain_count', 'unique_second_level_domains', 'droidbot_steps', 'gui_agent_steps', 'duration_seconds', 'num_nodes', 'num_reached_activities', 'app_num_total_activities', 'task_detail', 'failure_type'] try: with open(output_file, 'w', newline='', encoding='utf-8-sig') as csvfile: writer = csv.writer(csvfile) writer.writerow(headers) count = 0 for package_name in sorted(final_packages): info = app_data[package_name] if package_to_app: app_name = package_to_app.get(package_name, package_name) else: app_name = package_name domains = sorted(list(info['domains'])) second_level_domains = sorted(list(info['second_level_domains'])) is_success = package_name in success_packages if is_success: test_time = success_time.get(package_name, '') test_host = success_worker.get(package_name, '') task_detail = success_info.get(package_name, '') package_failure_type = '' else: test_time = failure_time.get(package_name, '') test_host = failure_worker.get(package_name, '') task_detail = failure_info.get(package_name, '') package_failure_type = failure_type.get(package_name, '') self_ratio = self_ratio_map.get(package_name, '') writer.writerow([ app_name, package_name, is_success, test_time, test_host, self_ratio, len(domains), ", ".join(domains), len(second_level_domains), ", ".join(second_level_domains), info['droidbot_steps'], info['gui_agent_steps'], info['duration_seconds'], info['num_nodes'], info['num_reached_activities'], info['app_num_total_activities'], task_detail, package_failure_type ]) count += 1 print(f"处理完成！最终统计了 {count} 个 App。") print(f"结果已保存至: {output_file}") except Exception as e: print(f"写入 CSV 时出错: {e}") if __name__ == "__main__": process_all_data()