#!/usr/bin/env python3 # AP + Modem 一体化工作流：获取/解压 → 归档 → 过滤 → 绘图 → 分析 import argparse import re import subprocess from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import List, Optional, Tuple from archive_unit import ArchiveProcessor from aplog_unit import AplogLocator from aplog_filter_unit import AplogFilterUnit from analysis_report_unit import AnalysisReportUnit from fs_utils import ensure_dir, find_prefixed_ancestor, move_preserve, prune_empty_dirs from log_patterns import has_aplog_fragments, is_ap_text_name, looks_like_aplog_dir from mdlog_filter_unit import MdlogFilterUnit from netlog_unit import NetlogCollector from plot_unit import PlotUnit from time_window_utils import NamedTimeWindow, parse_named_time_windows from workflow_layout import LAYOUT SCRIPT_ROOT = Path(__file__).resolve().parent SKILLS_ROOT = SCRIPT_ROOT.parents[1] # .../skills WORKFLOW_VERSION = "1.0.0" WORKFLOW_SKILL_ROOT = SCRIPT_ROOT.parent LOG_WORKFLOW_ROOT = WORKFLOW_SKILL_ROOT.parent APLOG_MERGE_ROOT = LOG_WORKFLOW_ROOT / "skill_log_merge_aplog" / "scripts" / "aplog_merge_tool" APLOG_FILTER_ROOT = LOG_WORKFLOW_ROOT / "skill_log_filter_aplog" / "scripts" / "aplog_filter_tool" / "aplog" APLOG_PLOT_ROOT = LOG_WORKFLOW_ROOT / "skill_log_plot_aplog" / "scripts" MDLOG_FILTER_ROOT = ( SKILLS_ROOT / "skill_code_ai_md_power" / "scripts" / "2_data_process_tool" / "mdlog_filter" / "skill_log_filter_mdlog" / "scripts" / "mdlog_filter_tool" ) MERGER = APLOG_MERGE_ROOT / "aplog_processor.py" AP_ANALYZER = APLOG_FILTER_ROOT / "mainAPLogHandler.py" MD_ANALYZER = MDLOG_FILTER_ROOT / "mainApp.py" MD_LOGPLOT = MDLOG_FILTER_ROOT / "LogPlot.py" AP_REG_SIGNAL_PLOT = APLOG_FILTER_ROOT / "additional_tool" / "plot_logcatAplog_reg_signal.py" AP_TPUT_PLOT = APLOG_FILTER_ROOT / "additional_tool" / "aplog_plot_tput.py" AP_MODEM_ACTIVITY_PLOT = APLOG_PLOT_ROOT / "aplog_plot_modemactivityinfo.py" AP_UID_TPUT_PLOT = APLOG_PLOT_ROOT / "aplog_plot_uid_tput.py" AP_WINDOW_OVERVIEW_PLOT = APLOG_PLOT_ROOT / "aplog_plot_window_overview.py" AP_SERVING_CELL_PLOT = APLOG_PLOT_ROOT / "aplog_plot_serving_cell.py" COMPACT_CSV_ROOT = LOG_WORKFLOW_ROOT / "skill_log_compact_csv" COMPACT_CSV_GENERATOR = COMPACT_CSV_ROOT / "scripts" / "generate_elevator_compact_csv.py" COMPACT_CSV_ELEVATOR_RULES = ( COMPACT_CSV_ROOT / "resources" / "compact_csv_elevator" / "modem_last_1s_avg_elevator_lines1_4_transposed.csv" ) FEISHU_REPORT_ASSETS = ( LOG_WORKFLOW_ROOT / "skill_log_feishu_analysis_report" / "scripts" / "prepare_feishu_report_assets.py" ) def _is_session_dir_name(name: str) -> bool: return re.match(r"^log\d+_\d+", name) is not None def _guess_year_from_path(path: Path) -> Optional[str]: for part in path.parts: match = re.search(r"(20\d{2})", part) if match: return match.group(1) return None def _guess_full_date_from_path(path: Path) -> Optional[str]: match = re.search(r"(20\d{2})(\d{2})(\d{2})", str(path)) if not match: return None return f"{match.group(1)}-{match.group(2)}-{match.group(3)}" def _normalize_compact_date(date_text: Optional[str], input_path: Path) -> str: if date_text: normalized = date_text.strip() if re.fullmatch(r"\d{4}-\d{2}-\d{2}", normalized): return normalized if re.fullmatch(r"\d{2}-\d{2}", normalized): year = _guess_year_from_path(input_path) or str(datetime.now().year) return f"{year}-{normalized}" guessed = _guess_full_date_from_path(input_path) if guessed: return guessed raise ValueError(f"无法为 compact csv 推导日期，请显式传入 --date: input={input_path}") def _build_compact_window_label(time_range: Optional[str]) -> Optional[str]: if not time_range: return None try: start_text, end_text = [item.strip() for item in time_range.split("-", 1)] except ValueError: return None start_match = re.fullmatch(r"(\d{2}):(\d{2}):(\d{2})", start_text) end_match = re.fullmatch(r"(\d{2}):(\d{2}):(\d{2})", end_text) if not start_match or not end_match: return None return ( f"{start_match.group(1)}{start_match.group(2)}{start_match.group(3)}_" f"{end_match.group(1)}{end_match.group(2)}{end_match.group(3)}" ) def _find_performance_summary_csv(input_path: Path) -> Optional[Path]: if input_path.is_file(): search_root = input_path.parent else: search_root = input_path matches = sorted(search_root.rglob("*SmoothedTestSummary.csv")) if not matches: return None return matches[0] def _find_input_root(path: Path) -> Optional[Path]: return find_prefixed_ancestor(path, "input") def _find_child_input_root(path: Path) -> Optional[Path]: if not path.is_dir(): return None candidates = sorted( child for child in path.iterdir() if child.is_dir() and child.name.startswith("input") ) return candidates[0] if candidates else None def _discover_child_device_inputs(path: Path) -> List[Path]: if not path.is_dir(): return [] device_inputs: List[Path] = [] for child in sorted(p for p in path.iterdir() if p.is_dir()): input_root = _find_child_input_root(child) if input_root is not None: device_inputs.append(input_root) return device_inputs def _build_run_session_root(base: Path, run_id: str, keep_leaf: bool) -> Optional[Path]: input_root = _find_input_root(base) if input_root is None: return None run_root = input_root.parent / "runs" / f"{run_id}_v{WORKFLOW_VERSION}" if keep_leaf and base != input_root: return run_root / base.name return run_root def detect_session_root( input_path: Path, explicit_root: Optional[Path], source_root: Optional[Path] = None, run_id: Optional[str] = None, ) -> Path: """Session root detection. Keep logic deterministic. No extra "compat" fallbacks. """ effective_run_id = run_id or datetime.now().strftime("%Y%m%d_%H%M%S") if explicit_root: explicit = explicit_root.resolve() runs_root = _build_run_session_root(explicit, effective_run_id, keep_leaf=explicit != _find_input_root(explicit)) return runs_root if runs_root is not None else explicit p = input_path.resolve() anchor = (source_root or p).resolve() runs_root = _build_run_session_root(anchor, effective_run_id, keep_leaf=False) if runs_root is None: runs_root = _build_run_session_root(p, effective_run_id, keep_leaf=False) if runs_root is not None: return runs_root if p.is_file() and p.suffix.lower() in {".zip", ".7z"} and source_root is not None: return source_root.resolve() if p.is_dir(): # Directory input means "organize inside this directory". return p q = p for _ in range(4): if _is_session_dir_name(q.name): return q if _is_session_dir_name(q.parent.name): return q.parent q = q.parent return input_path.parent @dataclass class Workflow: archive: ArchiveProcessor aplog_locator: AplogLocator netlog_collector: NetlogCollector aplog_filter_unit: AplogFilterUnit mdlog_filter_unit: MdlogFilterUnit plot_unit: PlotUnit analysis_report_unit: AnalysisReportUnit def resolve_scan_root_and_aplog_dir(self, input_path: Path, out_root: Path) -> Tuple[Path, Path]: base = self.archive.ensure_extracted_root(input_path, out_root) cur_root = base if cur_root.is_dir(): try: child_names = [p.name for p in cur_root.iterdir()] except Exception: child_names = [] if any(is_ap_text_name(name) for name in child_names): return cur_root, cur_root try: if looks_like_aplog_dir(cur_root) or has_aplog_fragments(child_names): return cur_root, cur_root except Exception: pass for _ in range(self.archive.max_depth): aplog_dir = self.aplog_locator.detect_aplog_dir(cur_root) if aplog_dir: return cur_root, aplog_dir bug_parent = self.aplog_locator.find_bugreport_txt_parent(cur_root) if bug_parent: return cur_root, bug_parent nxt = self.archive.extract_next_nested(cur_root, extract_root=out_root) if not nxt: break cur_root = nxt return cur_root, cur_root # Stage implementations live in dedicated units: # - filtering: `aplog_filter_unit.py`, `mdlog_filter_unit.py` # - plotting: `plot_unit.py` # - local analysis summary: `analysis_report_unit.py` def prepare_feishu_report_assets( self, session_root: Path, template_url: str = "", language: str = "zh", ) -> None: if not FEISHU_REPORT_ASSETS.exists(): print(f"[stage:report] missing Feishu asset script: {FEISHU_REPORT_ASSETS}") return cmd = [ "python3", str(FEISHU_REPORT_ASSETS), "--session-root", str(session_root), "--language", language, ] if template_url: cmd.extend(["--template-url", template_url]) print("[stage:report] skill_log_feishu_analysis_report asset manifest") result = subprocess.run(cmd, check=False) if result.returncode != 0: print(f"[stage:report] Feishu asset manifest failed: exit={result.returncode}") def reorganize_session(self, session_root: Path, source_root: Path, aplog_dir: Path) -> None: ensure_dir(session_root / LAYOUT.aplog_raw) ensure_dir(session_root / LAYOUT.aplog_filter) ensure_dir(session_root / LAYOUT.mdlog_filter) ensure_dir(session_root / LAYOUT.net_log_raw) ensure_dir(session_root / LAYOUT.others) protected = [ session_root / LAYOUT.aplog_raw, session_root / LAYOUT.aplog_filter, session_root / LAYOUT.mdlog_filter, session_root / LAYOUT.net_log_raw, session_root / LAYOUT.others, ] self.netlog_collector.collect(session_root, protected_dirs=protected) others_dir = session_root / LAYOUT.others for child in list(session_root.iterdir()): if child.name in LAYOUT.reserved_dirnames(): continue move_preserve(child, session_root, others_dir) def _find_aplog_compact_input_dir(self, session_root: Path) -> Optional[Path]: ap_root = session_root / LAYOUT.aplog_filter if not ap_root.exists(): return None candidates = [ap_root] + sorted(path for path in ap_root.rglob("*") if path.is_dir()) for candidate in candidates: if (candidate / "opt_elevator_event.txt").exists() or (candidate / "opt_airport_key_event.txt").exists(): return candidate return None def _find_md_compact_input_dir(self, session_root: Path) -> Optional[Path]: md_root = session_root / LAYOUT.mdlog_filter if not md_root.exists(): return None candidates = [] direct = md_root / "auto_analysis" if direct.exists(): candidates.append(direct) candidates.extend(sorted(path for path in md_root.rglob("auto_analysis") if path.is_dir())) seen = set() for candidate in candidates: resolved = candidate.resolve() if resolved in seen: continue seen.add(resolved) if list(candidate.glob("sim*.csv")): return candidate return None def run_compact_csv( self, session_root: Path, input_path: Path, time_range: Optional[str], date: Optional[str], scenario: str, ) -> None: if scenario != "elevator": raise RuntimeError(f"Unsupported compact csv scenario: {scenario}") if not COMPACT_CSV_GENERATOR.exists(): raise RuntimeError(f"Missing compact csv generator script: {COMPACT_CSV_GENERATOR}") if not COMPACT_CSV_ELEVATOR_RULES.exists(): raise RuntimeError(f"Missing compact csv rules: {COMPACT_CSV_ELEVATOR_RULES}") window_label = _build_compact_window_label(time_range) if not window_label: print("[stage:compact_csv] 缺少有效 time_range，跳过 compact csv") return ap_dir = self._find_aplog_compact_input_dir(session_root) md_dir = self._find_md_compact_input_dir(session_root) if ap_dir is None or md_dir is None: raise RuntimeError( "[stage:compact_csv] 缺少 compact csv 输入目录: " f"ap_dir={ap_dir}, md_dir={md_dir}, session_root={session_root}" ) compact_date = _normalize_compact_date(date, input_path) output_dir = ensure_dir(session_root / LAYOUT.analysis / "compact_csv" / scenario) performance_csv = _find_performance_summary_csv(input_path) cmd = [ "python3", str(COMPACT_CSV_GENERATOR), "--transposed-rules", str(COMPACT_CSV_ELEVATOR_RULES), "--ap-dir", str(ap_dir), "--md-dir", str(md_dir), "--date", compact_date, "--windows", window_label, "--output-dir", str(output_dir), ] if performance_csv is not None: cmd.extend(["--performance-csv", str(performance_csv)]) print(f"[stage:compact_csv] performance_csv={performance_csv}") print(f"[stage:compact_csv] {scenario} -> {output_dir}") result = subprocess.run(cmd, check=False) if result.returncode != 0: raise RuntimeError( f"[stage:compact_csv] 生成失败: scenario={scenario}, exit={result.returncode}" ) # Filtering stages are handled by units: # - `AplogFilterUnit` (AP filter outputs) # - `MdlogFilterUnit` (MD filter + auto_analysis outputs) def process_single_input( self, input_path: Path, out_root: Path, prefix: str, time_range: Optional[str], date: Optional[str], reorg: bool, mdlog_only_masks: Optional[str] = None, mdlog_dump_all_visible: bool = False, mdlog_full_visible_only: bool = False, session_root_override: Optional[Path] = None, run_id: Optional[str] = None, feishu_assets: bool = True, feishu_template_url: str = "", report_language: str = "zh", compact_csv_scenario: Optional[str] = None, ) -> None: ensure_dir(out_root) print(f"输入: {input_path}") print(f"工作目录: {out_root}") session_root = detect_session_root( input_path, session_root_override, source_root=input_path, run_id=run_id, ) scan_root, aplog_dir = self.resolve_scan_root_and_aplog_dir(input_path, session_root) print(f"日志目录: {aplog_dir}") session_root = detect_session_root( input_path, session_root_override, source_root=scan_root, run_id=run_id, ) if not reorg: return # 准备会话目录 -> 过滤 -> 绘图 -> 分析 try: print(f"[stage:prepare] session_root={session_root}") self.reorganize_session(session_root, scan_root, aplog_dir) print("[stage:filter] aplog_filter_tool") self.aplog_filter_unit.run(session_root, aplog_dir=aplog_dir, time_range=time_range, date=date, prefix=prefix) print("[stage:filter] mdlog_filter_tool") self.mdlog_filter_unit.run( session_root, source_root=scan_root, time_range=time_range, date=date, only_masks=mdlog_only_masks, dump_all_visible=mdlog_dump_all_visible, full_visible_only=mdlog_full_visible_only, ) if compact_csv_scenario: self.run_compact_csv( session_root=session_root, input_path=input_path, time_range=time_range, date=date, scenario=compact_csv_scenario, ) print("[stage:plot] aplog/mdlog") self.plot_unit.run(session_root) print("[stage:analyze] write local report input") report_path = self.analysis_report_unit.write_local_report(session_root) print(f"[stage:analyze] report: {report_path}") if feishu_assets: self.prepare_feishu_report_assets( session_root, template_url=feishu_template_url, language=report_language, ) finally: removed_dirs = prune_empty_dirs(session_root) if removed_dirs: print(f"[stage:cleanup] removed empty dirs: {len(removed_dirs)}") def main() -> None: parser = argparse.ArgumentParser(description="AP+MD 一体化工作流：归档→过滤→绘图→分析") parser.add_argument("--input", "-i", required=True, help="输入路径：zip/7z/目录/单文件") parser.add_argument( "--out-dir", "-o", default=str(Path.home() / "disk4T" / "jieli" / "workflow"), help="兼容参数：仅用于少量临时工作文件，不决定最终产物位置", ) parser.add_argument("--prefix", default="logcat_pacific_", help="合并文件重命名前缀") parser.add_argument("--time-range", help="合并时间范围 HH:MM:SS-HH:MM:SS，可选") parser.add_argument("--date", help="合并日期或范围 MM-DD 或 MM-DD,MM-DD，可选") parser.add_argument("--time-points-file", help="阶段时间点文件，按成对进/出时间批量生成多个窗口 run") parser.add_argument("--reorg", action="store_true", help="完成分析后，将会话目录按 aplog/mdlog/others 结构归档") parser.add_argument("--session-root", help="显式指定会话根目录") parser.add_argument("--mdlog-only-masks", help="仅运行指定 mdlog masks，逗号分隔") parser.add_argument( "--mdlog-dump-all-visible", action="store_true", help="额外导出 QCAT 全可见 mdlog 文本与 strings 兜底索引，不受 mask 白名单限制", ) parser.add_argument( "--mdlog-full-visible-only", action="store_true", help="只导出 MD full visible 证据，跳过 mask 结构化解析", ) parser.add_argument( "--compact-csv-scenario", choices=["elevator"], help="可选：在 AP/MD filter 后生成 compact csv", ) parser.add_argument("--no-feishu-assets", action="store_true", help="不生成飞书报告 asset manifest") parser.add_argument("--feishu-template-url", default="", help="可选飞书报告模板文档 URL") parser.add_argument("--report-language", default="zh", choices=["zh", "en"], help="报告语言") args = parser.parse_args() input_path = Path(args.input).resolve() out_root = Path(args.out_dir).resolve() ensure_dir(out_root) explicit_session_root = Path(args.session_root).resolve() if args.session_root else None run_id = datetime.now().strftime("%Y%m%d_%H%M%S") archive = ArchiveProcessor(max_depth=4) workflow = Workflow( archive=archive, aplog_locator=AplogLocator(), netlog_collector=NetlogCollector(layout=LAYOUT), aplog_filter_unit=AplogFilterUnit( merger_script=MERGER, ap_analyzer_script=AP_ANALYZER, aplog_raw_dirname=LAYOUT.aplog_raw, aplog_filter_dirname=LAYOUT.aplog_filter, ), mdlog_filter_unit=MdlogFilterUnit( md_analyzer_script=MD_ANALYZER, mdlog_filter_dirname=LAYOUT.mdlog_filter, ), plot_unit=PlotUnit( layout=LAYOUT, md_logplot_script=MD_LOGPLOT, md_subids="1", ap_reg_signal_plot_script=AP_REG_SIGNAL_PLOT, ap_tput_plot_script=AP_TPUT_PLOT, ap_uid_tput_plot_script=AP_UID_TPUT_PLOT, ap_modem_activity_plot_script=AP_MODEM_ACTIVITY_PLOT, ap_window_overview_plot_script=AP_WINDOW_OVERVIEW_PLOT, ap_serving_cell_plot_script=AP_SERVING_CELL_PLOT, ), analysis_report_unit=AnalysisReportUnit(layout=LAYOUT), ) time_windows = _resolve_time_windows(args.time_points_file, args.date) device_dirs = archive.prepare_device_dirs(input_path) if input_path.is_dir() and explicit_session_root is None else [] if device_dirs: print(f"检测到 {len(device_dirs)} 个顶层压缩包，按各设备 input* 同级 runs 独立批量处理") for device_dir in device_dirs: device_input = _find_child_input_root(device_dir) or device_dir _run_workflow_windows( workflow=workflow, input_path=device_input, out_root=out_root / device_dir.name, prefix=args.prefix, base_time_range=args.time_range, base_date=args.date, reorg=True, mdlog_only_masks=args.mdlog_only_masks, mdlog_dump_all_visible=args.mdlog_dump_all_visible, mdlog_full_visible_only=args.mdlog_full_visible_only, session_root_override=None, run_id=run_id, windows=time_windows, feishu_assets=not args.no_feishu_assets, feishu_template_url=args.feishu_template_url, report_language=args.report_language, compact_csv_scenario=args.compact_csv_scenario, ) return child_device_inputs = _discover_child_device_inputs(input_path) if input_path.is_dir() and explicit_session_root is None else [] if len(child_device_inputs) >= 2: print(f"检测到 {len(child_device_inputs)} 个设备目录（child input*），按设备独立批量处理") for device_input in child_device_inputs: _run_workflow_windows( workflow=workflow, input_path=device_input, out_root=out_root / device_input.parent.name, prefix=args.prefix, base_time_range=args.time_range, base_date=args.date, reorg=True, mdlog_only_masks=args.mdlog_only_masks, mdlog_dump_all_visible=args.mdlog_dump_all_visible, mdlog_full_visible_only=args.mdlog_full_visible_only, session_root_override=None, run_id=run_id, windows=time_windows, feishu_assets=not args.no_feishu_assets, feishu_template_url=args.feishu_template_url, report_language=args.report_language, compact_csv_scenario=args.compact_csv_scenario, ) return _run_workflow_windows( workflow=workflow, input_path=input_path, out_root=out_root, prefix=args.prefix, base_time_range=args.time_range, base_date=args.date, reorg=args.reorg, mdlog_only_masks=args.mdlog_only_masks, mdlog_dump_all_visible=args.mdlog_dump_all_visible, mdlog_full_visible_only=args.mdlog_full_visible_only, session_root_override=explicit_session_root, run_id=run_id, windows=time_windows, feishu_assets=not args.no_feishu_assets, feishu_template_url=args.feishu_template_url, report_language=args.report_language, compact_csv_scenario=args.compact_csv_scenario, ) def _resolve_time_windows(time_points_file: Optional[str], date: Optional[str]) -> List[NamedTimeWindow]: if not time_points_file: return [] points_path = Path(time_points_file).resolve() windows = parse_named_time_windows(points_path, date=date) print(f"从 {points_path} 解析到 {len(windows)} 个阶段窗口") for window in windows: print( f" - {window.label}: " f"{window.start_text}:00 -> {window.end_text}:59" ) return windows def _run_workflow_windows( workflow: Workflow, input_path: Path, out_root: Path, prefix: str, base_time_range: Optional[str], base_date: Optional[str], reorg: bool, mdlog_only_masks: Optional[str], mdlog_dump_all_visible: bool, mdlog_full_visible_only: bool, session_root_override: Optional[Path], run_id: str, windows: List[NamedTimeWindow], feishu_assets: bool, feishu_template_url: str, report_language: str, compact_csv_scenario: Optional[str], ) -> None: if not windows: workflow.process_single_input( input_path, out_root=out_root, prefix=prefix, time_range=base_time_range, date=base_date, reorg=reorg, mdlog_only_masks=mdlog_only_masks, mdlog_dump_all_visible=mdlog_dump_all_visible, mdlog_full_visible_only=mdlog_full_visible_only, session_root_override=session_root_override, run_id=run_id, feishu_assets=feishu_assets, feishu_template_url=feishu_template_url, report_language=report_language, compact_csv_scenario=compact_csv_scenario, ) return for window in windows: print(f"[window] {window.label} {window.time_range}") workflow.process_single_input( input_path, out_root=out_root, prefix=prefix, time_range=window.time_range, date=window.date or base_date, reorg=reorg, mdlog_only_masks=mdlog_only_masks, mdlog_dump_all_visible=mdlog_dump_all_visible, mdlog_full_visible_only=mdlog_full_visible_only, session_root_override=session_root_override, run_id=f"{run_id}_{window.label}", feishu_assets=feishu_assets, feishu_template_url=feishu_template_url, report_language=report_language, compact_csv_scenario=compact_csv_scenario, ) if __name__ == "__main__": main()