#!/usr/bin/env python3 """ 埋点测试覆盖率分析工具 - 主脚本 从飞书文档自动生成测试用例和覆盖率报告 """ import os import sys import json import re import argparse from datetime import datetime from typing import List, Dict, Any, Optional # 导入 datatest_feishu 的 FeishuClient # 动态查找 datatest_feishu 目录 current_dir = os.path.dirname(os.path.abspath(__file__)) parent_dir = os.path.dirname(current_dir) datatest_feishu_dir = os.path.join(os.path.dirname(parent_dir), 'datatest_feishu') sys.path.insert(0, datatest_feishu_dir) # `convert_to_table.py` 在导入时会 `from feishu_client import FeishuClient`， # 它期望 `feishu_client.py` 能直接被 Python 在当前搜路径找到。 sys.path.insert(0, os.path.join(datatest_feishu_dir, 'scripts')) from scripts.feishu_client import FeishuClient try: # 复用 datatest_feishu 的解析器，把飞书 blocks 结构化成事件/属性 from scripts.convert_to_table import parse_events as datatest_parse_events except Exception: datatest_parse_events = None class TrackingRequirement: """埋点需求定义""" def __init__(self, event_name: str, block_index: int = 0): self.event_name = event_name self.fields: List[Dict[str, Any]] = [] self.description = "" self.trigger_scene = "" self.block_index = block_index # 记录在文档中的位置，用于排序 def add_field(self, name: str, field_type: str, description: str, required: bool = True, enum_values: List[str] = None): self.fields.append({ 'name': name, 'type': field_type, 'description': description, 'required': required, 'enum_values': enum_values or [] }) class TrackingImplementation: """埋点实现分析""" def __init__(self, event_name: str): self.event_name = event_name self.fields: Dict[str, Any] = {} self.code_location = "" self.code_snippet = "" def add_field(self, name: str, field_type: str, value_expr: str): self.fields[name] = { 'type': field_type, 'value': value_expr } class FeishuDocParser: """飞书文档解析器""" def __init__(self, client: FeishuClient): self.client = client def parse_tracking_doc(self, doc_url: str) -> List[TrackingRequirement]: """解析飞书埋点文档""" # 提取 token doc_token = self._extract_token(doc_url) # 获取文档内容 blocks = self.client.get_doc_blocks(doc_token) # 优先复用 datatest_feishu 的解析器（更贴近飞书文档实际格式） if datatest_parse_events: events = datatest_parse_events(blocks) # type: ignore[misc] return self._map_events_to_requirements(events) # 兜底：使用 tracktest 自带的轻量解析 return self._parse_blocks(blocks) def _map_events_to_requirements(self, events: List[Dict[str, Any]]) -> List[TrackingRequirement]: """把 datatest_feishu parse_events 的结果映射成 tracktest 的 TrackingRequirement""" requirements: List[TrackingRequirement] = [] for idx, event in enumerate(events): event_name = (event.get('event_name') or '').strip() if not event_name: continue req = TrackingRequirement(event_name=event_name, block_index=idx) req.description = event.get('title', '') or '' req.trigger_scene = event.get('scenario', '') or '' for prop in event.get('properties', []) or []: field_name = (prop.get('name') or '').strip() if not field_name: continue field_type = (prop.get('type') or 'STRING').strip() field_desc = (prop.get('description') or '').strip() # 当前 CoverageAnalyzer 不使用 enum_values；这里做一个轻量兼容解析 enum_values = self._extract_enum_values(field_desc) req.add_field( name=field_name, field_type=field_type, description=field_desc, required=True, enum_values=enum_values, ) if req.fields: requirements.append(req) return requirements def _extract_enum_values(self, description: str) -> List[str]: """从字段描述里尽量提取枚举值（启发式）""" if not description: return [] # 兼容 tracktest 旧逻辑：xxx：枚举1、枚举2（...） enum_match = re.search(r'[:：](.+?)（', description) if not enum_match: # 备选：xxx：枚举1、枚举2, ... enum_match = re.search(r'[:：](.+?)(?:$|，|,|；|;)', description) if not enum_match: return [] enum_str = enum_match.group(1) # 按常见分隔拆分，并清理引号/空格 values = [v.strip().strip('"\'') for v in re.split(r'[、，,]', enum_str)] return [v for v in values if v] def _extract_token(self, url: str) -> str: """从 URL 提取 token""" patterns = [ r'/wiki/([a-zA-Z0-9]+)', r'/docx/([a-zA-Z0-9]+)', ] for pattern in patterns: match = re.search(pattern, url) if match: return match.group(1) raise ValueError(f"无法从 URL 提取 token: {url}") def _parse_blocks(self, blocks: List[Dict]) -> List[TrackingRequirement]: """解析文档块，提取埋点需求""" requirements = [] current_event = None for i, block in enumerate(blocks): block_type = block.get('block_type') text = self._extract_text(block) # 跳过空文本块 if not text.strip(): continue # 检测是否是埋点事件标题 if self._is_event_title(text): # 保存前一个事件 if current_event and current_event.fields: # 只保存有字段的事件 requirements.append(current_event) # 创建新事件，记录块索引 event_name = self._extract_event_name(text) current_event = TrackingRequirement(event_name, block_index=i) current_event.description = text continue # 解析字段定义（当前有事件在处理中） if current_event: # 解析字段 field = self._parse_field_definition(text) if field: current_event.add_field(**field) # 解析埋点场景 if '埋点场景' in text or '触发时机' in text: current_event.trigger_scene = text # 添加最后一个事件 if current_event and current_event.fields: requirements.append(current_event) # 按文档顺序排序（使用 block_index） requirements.sort(key=lambda x: x.block_index) return requirements def _extract_text(self, block: Dict) -> str: """提取块的文本内容""" # 飞书API返回的块结构可能有不同的字段 possible_keys = ['text', 'paragraph', 'heading2', 'heading3', 'heading4', 'heading5', 'heading6', 'heading7'] for key in possible_keys: if key in block: elements = block[key].get('elements', []) texts = [] for elem in elements: if 'text_run' in elem: texts.append(elem['text_run'].get('content', '')) return ''.join(texts).strip() return "" def _is_event_title(self, text: str) -> bool: """判断是否是埋点事件标题""" # 格式1: "事件名称： EventName" 或 "事件名称 ： EventName" if '事件名称' in text and ('：' in text or ':' in text): match = re.search(r'事件名称\s*[：:]\s*(\w+)', text) if match and match.group(1): return True # 格式2: "XXX——服务端" 或 "XXX—— cocos传，原生客户端上报" (heading7格式) # 这类标题包含"——"并且前面有内容 if '——' in text: parts = text.split('——') if len(parts) >= 2 and parts[0].strip(): # 确保"——"前面有内容 return True return False def _extract_event_name(self, text: str) -> str: """从标题提取事件名""" # 格式1: "事件名称： EventName" 或 "事件名称 ： EventName" if '事件名称' in text: match = re.search(r'事件名称\s*[：:]\s*(\w+)', text) if match: return match.group(1).strip() # 格式2: "XXX——服务端" 或 "加好友—— cocos传，原生客户端上报" # 提取"——"前面的部分作为事件名 if '——' in text: event_name = text.split('——')[0].strip() # 移除特殊符号 event_name = re.sub(r'【.*?】', '', event_name) return event_name # 其他格式：移除特殊标记 text = re.sub(r'【.*?】', '', text) text = text.strip() return text def _parse_field_definition(self, text: str) -> Optional[Dict]: """解析字段定义""" # 格式: field_name = 字段说明, TYPE（不区分大小写） # 示例: game_type = 游戏ID，XX（Parchisi飞行棋），NUMBER pattern = r'(\w+)\s*=\s*(.+?)[,，]\s*(STRING|NUMBER|BOOL|list)' match = re.search(pattern, text, re.IGNORECASE) if match: field_name = match.group(1) description = match.group(2).strip() field_type = match.group(3).upper() # 统一转为大写 # 提取枚举值 - 在描述中查找冒号或中文冒号后的内容 enum_values = [] enum_match = re.search(r'[:：](.+?)（', description) if enum_match: enum_str = enum_match.group(1) enum_values = [v.strip().strip('"\'') for v in re.split(r'[、，,]', enum_str)] return { 'name': field_name, 'field_type': field_type if field_type != 'LIST' else 'list', # 保持list小写 'description': description, 'enum_values': enum_values } return None class CodeAnalyzer: """代码分析器""" def __init__(self, code_path: str): self.code_path = code_path self.language = self._detect_language() def _detect_language(self) -> str: """检测代码语言""" if self.code_path.endswith('.go'): return 'go' elif self.code_path.endswith(('.ts', '.js')): return 'typescript' else: raise ValueError(f"不支持的代码文件类型: {self.code_path}") def analyze_tracking_implementation(self, event_name: str) -> Optional[TrackingImplementation]: """分析埋点实现""" with open(self.code_path, 'r', encoding='utf-8') as f: code = f.read() if self.language == 'go': return self._analyze_go_tracking(code, event_name) elif self.language == 'typescript': return self._analyze_ts_tracking(code, event_name) return None def _analyze_go_tracking(self, code: str, event_name: str) -> Optional[TrackingImplementation]: """分析 Go 埋点实现""" # 查找埋点函数 pattern = rf'func\s+\(.*?\)\s+track{event_name}\s*\([^)]*\)\s*{{\s*(.+?)\n\s*huiwan\.SensorCustomizedProperties' match = re.search(pattern, code, re.DOTALL) if not match: return None impl = TrackingImplementation(event_name) # 查找 properties 定义 props_pattern = r'properties\s*:=\s*map\[string\]interface{}\s*{(.+?)}' props_match = re.search(props_pattern, code[match.start():], re.DOTALL) if props_match: props_content = props_match.group(1) # 解析每个字段 field_pattern = r'"(\w+)":\s*(.+?)(?:,|\n)' for field_match in re.finditer(field_pattern, props_content): field_name = field_match.group(1) value_expr = field_match.group(2).strip() # 推断类型 field_type = self._infer_go_type(value_expr) impl.add_field(field_name, field_type, value_expr) return impl def _analyze_ts_tracking(self, code: str, event_name: str) -> Optional[TrackingImplementation]: """分析 TypeScript 埋点实现""" # 查找 track 方法调用 pattern = rf'track\(["\'](\w+)["\'],\s*{{\s*(.+?)\s*}}\)' for match in re.finditer(pattern, code, re.DOTALL): if match.group(1) == event_name: impl = TrackingImplementation(event_name) props_content = match.group(2) # 解析字段 field_pattern = r'(\w+):\s*(.+?)(?:,|\n)' for field_match in re.finditer(field_pattern, props_content): field_name = field_match.group(1) value_expr = field_match.group(2).strip() field_type = self._infer_ts_type(value_expr) impl.add_field(field_name, field_type, value_expr) return impl return None def _infer_go_type(self, value_expr: str) -> str: """推断 Go 值的类型""" if 'int(' in value_expr or 'int32(' in value_expr: return 'int' elif 'strconv.Itoa' in value_expr or '+ "_" +' in value_expr or 'strconv.FormatBool' in value_expr: return 'string' elif value_expr in ['true', 'false']: return 'bool' elif value_expr.startswith('"') or value_expr.startswith("'"): return 'string' else: return 'unknown' def _infer_ts_type(self, value_expr: str) -> str: """推断 TypeScript 值的类型""" if value_expr.startswith('"') or value_expr.startswith("'") or value_expr.startswith('`'): return 'string' elif value_expr in ['true', 'false']: return 'bool' elif value_expr.isdigit(): return 'number' else: return 'unknown' class CoverageAnalyzer: """覆盖率分析器""" def analyze(self, requirement: TrackingRequirement, implementation: Optional[TrackingImplementation]) -> Dict: """分析覆盖率""" if not implementation: return { 'field_coverage': 0.0, 'type_matching': 0.0, 'overall_score': 'F', 'overall_percentage': 0.0, 'missing_fields': [f['name'] for f in requirement.fields], 'implemented_fields': [], 'type_mismatches': [] } # 字段覆盖率 required_fields = [f['name'] for f in requirement.fields if f.get('required', True)] implemented_fields = list(implementation.fields.keys()) common_fields = set(required_fields) & set(implemented_fields) field_coverage = len(common_fields) / len(required_fields) * 100 if required_fields else 0 # 缺失字段 missing_fields = set(required_fields) - set(implemented_fields) # 类型匹配 type_mismatches = [] for field in requirement.fields: field_name = field['name'] if field_name in implementation.fields: expected_type = field['type'].lower() actual_type = implementation.fields[field_name]['type'] if not self._types_compatible(expected_type, actual_type): type_mismatches.append({ 'field': field_name, 'expected': expected_type, 'actual': actual_type }) type_matching = (len(common_fields) - len(type_mismatches)) / len(common_fields) * 100 if common_fields else 0 # 整体评分 overall_score = (field_coverage * 0.7 + type_matching * 0.3) grade = 'F' if overall_score >= 90: grade = 'A' elif overall_score >= 80: grade = 'B' elif overall_score >= 70: grade = 'C' elif overall_score >= 60: grade = 'D' return { 'field_coverage': field_coverage, 'type_matching': type_matching, 'overall_score': grade, 'overall_percentage': overall_score, 'missing_fields': list(missing_fields), 'implemented_fields': implemented_fields, 'type_mismatches': type_mismatches } def _types_compatible(self, expected: str, actual: str) -> bool: """检查类型是否兼容""" type_mapping = { 'string': ['string', 'str'], 'number': ['int', 'number', 'int32', 'int64', 'float'], 'bool': ['bool', 'boolean'], } for standard_type, compatible_types in type_mapping.items(): if expected in compatible_types and actual in compatible_types: return True return False class ReportGenerator: """报告生成器""" def generate_markdown_report(self, game_name: str, requirement: TrackingRequirement, implementation: Optional[TrackingImplementation], coverage: Dict, output_path: str): """生成 Markdown 覆盖率报告""" report_lines = [] # 标题 report_lines.append(f"# {game_name} {requirement.event_name} 埋点覆盖率分析报告\n") report_lines.append(f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") report_lines.append(f"**分析方法**: 静态代码分析 + 需求文档对比\n") report_lines.append("---\n") # 总体覆盖率 report_lines.append("## 📊 总体覆盖率\n") report_lines.append("| 维度 | 覆盖率 | 评级 |") report_lines.append("|------|--------|------|") report_lines.append(f"| 字段覆盖率 | {coverage['field_coverage']:.1f}% | {coverage['overall_score']} |") report_lines.append(f"| 类型匹配度 | {coverage['type_matching']:.1f}% | - |") report_lines.append(f"| **整体评分** | **{coverage['overall_percentage']:.1f}%** | **{coverage['overall_score']}** |") report_lines.append("") # 字段明细表 report_lines.append("## 📋 字段覆盖率明细\n") report_lines.append("| # | 字段名 | 需求类型 | 实现类型 | 状态 | 问题 |") report_lines.append("|---|--------|----------|----------|------|------|") for idx, field in enumerate(requirement.fields, 1): field_name = field['name'] expected_type = field['type'] if field_name in coverage['implemented_fields']: if implementation and field_name in implementation.fields: actual_type = implementation.fields[field_name]['type'] else: actual_type = "unknown" # 检查类型匹配 is_type_mismatch = any(m['field'] == field_name for m in coverage['type_mismatches']) if is_type_mismatch: status = "⚠️" issue = f"类型不匹配：需求{expected_type}，实际{actual_type}" else: status = "✅" issue = "-" else: actual_type = "-" status = "❌" issue = f"**缺失：{field['description']}**" report_lines.append(f"| {idx} | {field_name} | {expected_type} | {actual_type} | {status} | {issue} |") report_lines.append("") # 问题汇总 if coverage['missing_fields'] or coverage['type_mismatches']: report_lines.append("## 🚨 问题汇总\n") if coverage['missing_fields']: report_lines.append("### ❌ 缺失字段\n") for field_name in coverage['missing_fields']: field_def = next((f for f in requirement.fields if f['name'] == field_name), None) if field_def: report_lines.append(f"- **{field_name}**: {field_def['description']} ({field_def['type']})") report_lines.append("") if coverage['type_mismatches']: report_lines.append("### ⚠️ 类型不匹配\n") for mismatch in coverage['type_mismatches']: report_lines.append(f"- **{mismatch['field']}**: 需求要求 {mismatch['expected']}，实际实现 {mismatch['actual']}") report_lines.append("") # 修复建议 report_lines.append("## 🔧 修复建议\n") for field_name in coverage['missing_fields']: field_def = next((f for f in requirement.fields if f['name'] == field_name), None) if field_def: report_lines.append(f"### 添加 {field_name} 字段\n") report_lines.append(f"**说明**: {field_def['description']}\n") report_lines.append("**修复代码**:") report_lines.append("```go") report_lines.append(f'properties["{field_name}"] = // TODO: 实现获取逻辑') report_lines.append("```\n") # 写入文件 with open(output_path, 'w', encoding='utf-8') as f: f.write('\n'.join(report_lines)) print(f"✅ 覆盖率报告已生成: {output_path}") def main(): parser = argparse.ArgumentParser(description='埋点测试覆盖率分析工具') parser.add_argument('--feishu-url', required=True, help='飞书文档URL') parser.add_argument('--code-path', required=True, help='代码文件路径') parser.add_argument('--game-name', default='Game', help='游戏名称') parser.add_argument('--event-name', help='指定分析的事件名（可选）') parser.add_argument('--output-dir', default='.', help='输出目录') args = parser.parse_args() # 初始化客户端 client = FeishuClient() # 解析文档 print(f"📄 正在解析飞书文档...") parser = FeishuDocParser(client) requirements = parser.parse_tracking_doc(args.feishu_url) print(f"✅ 解析完成，找到 {len(requirements)} 个埋点事件") # 分析代码 print(f"🔍 正在分析代码实现...") analyzer = CodeAnalyzer(args.code_path) # 分析覆盖率 coverage_analyzer = CoverageAnalyzer() report_gen = ReportGenerator() for requirement in requirements: if args.event_name and requirement.event_name != args.event_name: continue print(f"\n分析埋点: {requirement.event_name}") implementation = analyzer.analyze_tracking_implementation(requirement.event_name) coverage = coverage_analyzer.analyze(requirement, implementation) # 生成报告 output_path = os.path.join( args.output_dir, f"{args.game_name}_{requirement.event_name}_覆盖率报告_{datetime.now().strftime('%Y%m%d')}.md" ) report_gen.generate_markdown_report( args.game_name, requirement, implementation, coverage, output_path ) # 打印摘要 print(f" 字段覆盖率: {coverage['field_coverage']:.1f}%") print(f" 整体评分: {coverage['overall_score']} ({coverage['overall_percentage']:.1f}%)") if __name__ == '__main__': main()