# -*- encoding=utf8 -*- import re import json import logging logger = logging.getLogger(__name__) def parse_utg_js(js_path, target_package=None): """ 解析 utg.js 文件,提取 UTG 统计信息 Args: js_path: utg.js 文件路径 target_package: 目标包名(用于过滤节点) Returns: dict: { 'num_nodes': int, 'num_reached_activities': int, 'app_num_total_activities': int } """ result = { 'num_nodes': 0, 'num_reached_activities': 0, 'app_num_total_activities': 0 } try: with open(js_path, 'r', encoding='utf-8') 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 = utg_data.get('num_reached_activities', 0) result['num_reached_activities'] = num_reached if isinstance(num_reached, int) else 0 app_total = utg_data.get('app_num_total_activities', 0) result['app_num_total_activities'] = app_total if isinstance(app_total, int) else 0 except json.JSONDecodeError as e: logger.warning(f"解析 UTG JSON 时出错: {e}") _fallback_parse(content, result) else: _fallback_parse(content, result) except Exception as e: logger.error(f"解析 UTG 文件 {js_path} 时出错: {e}") return result def _fallback_parse(content, result): """正则表达式后备解析""" 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))