#!/usr/bin/env python3 """Analysis report unit for local workflow output.""" from __future__ import annotations from dataclasses import dataclass import json from pathlib import Path import re from typing import Dict, Iterable, List, Optional, Sequence, Set, Tuple from fs_utils import ensure_dir from workflow_layout import SessionLayout PRIMARY_CSV_ARCHIVE_DIR_NAME = "_primary_csv" TEXT_SUFFIXES = (".txt", ".log", ".csv") PLOT_SUFFIXES = (".html", ".png") INTERMEDIATE_DIR_NAMES = {"work", ".work"} MASK_STATUS_FILE_NAME = "mask_status.json" MAX_HIGHLIGHTS_PER_FILE = 12 MAX_TIMELINE_GROUPS = 40 MAX_TIMELINE_ITEMS_PER_GROUP = 6 MAX_LINE_CHARS = 360 CATEGORY_LABELS = { "data": "数据通路", "device": "设备信息", "gps": "GPS/定位", "network": "网络/小区", "other": "其它", "power": "功耗/睡眠", "ril": "RIL/Radio", "sensor": "传感器", } TIMESTAMP_RE = re.compile(r"(?P\b\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?)") KEYWORD_RE = re.compile( r"(" r"sleep|sleepRadio|modemSleepRatio|mpss|apss|wakeup|wake\s*lock|qrtr|" r"IPA|rmnet|WAN|LAN|netif|uid|bytes|traffic|" r"ServingCell|mServingCellInfo|ratType|EUTRAN|NGRAN|NR|LTE|isOos|oos|REG_|REG_HOME|PLMN|" r"RSRP|SNR|RSRQ|band=|freq=|cellId|" r"GPS|Gnss|location|Geofence|died|rebind|Unable|fail|error|timeout|" r"sensor|elevator|stuck|stall|reset|crash|exception" r")", re.IGNORECASE, ) @dataclass(frozen=True) class FileScan: path: Path category: str line_count: int first_timestamp: str last_timestamp: str highlights: List[Tuple[int, str]] @dataclass(frozen=True) class AnalysisReportUnit: layout: SessionLayout def _list_plot_files(self, root: Path, limit: Optional[int] = 30) -> List[Path]: if not root.exists(): return [] files = [p for p in root.rglob("*") if p.is_file() and p.name.endswith(PLOT_SUFFIXES)] files.sort(key=lambda p: str(p.relative_to(root))) return files[:limit] if limit is not None else files def _list_files(self, root: Path, suffix: str, limit: Optional[int] = None) -> List[Path]: if not root.exists(): return [] files = [ p for p in root.rglob("*") if p.is_file() and p.name.endswith(suffix) and self._is_report_source(p, root) ] files.sort(key=lambda p: str(p.relative_to(root))) return files[:limit] if limit is not None else files def _is_report_source(self, path: Path, root: Path) -> bool: try: rel_parts = path.relative_to(root).parts except ValueError: rel_parts = path.parts if PRIMARY_CSV_ARCHIVE_DIR_NAME in rel_parts: return False if any(part in INTERMEDIATE_DIR_NAMES for part in rel_parts): return False if any(part.startswith("temp_") for part in rel_parts): return False return True def _list_text_outputs(self, root: Path) -> List[Path]: if not root.exists(): return [] files = [ p for p in root.rglob("*") if p.is_file() and p.name.endswith(TEXT_SUFFIXES) and self._is_report_source(p, root) ] return sorted(files, key=lambda p: str(p.relative_to(root))) def _list_mask_status_files(self, root: Path) -> List[Path]: if not root.exists(): return [] return sorted(root.glob(f"*/work/{MASK_STATUS_FILE_NAME}"), key=lambda p: str(p.relative_to(root))) def _category_for(self, path: Path) -> str: name = path.name.lower() if name.startswith("power_") or "sleep" in name or "wakeup" in name: return "power" if name.startswith("reg_") or name.startswith("rf_") or "networkbrain" in name: return "network" if name.startswith("data_") or "netif" in name or "uid" in name or "ipa" in name: return "data" if name.startswith("gps_") or "gnss" in name or "location" in name: return "gps" if name.startswith("sensor_") or "elevator" in name: return "sensor" if name.startswith("ril_"): return "ril" if name.startswith("device_"): return "device" return "other" def _trim_line(self, line: str) -> str: text = " ".join(line.strip().split()) if len(text) <= MAX_LINE_CHARS: return text return text[: MAX_LINE_CHARS - 3] + "..." def _scan_text_file(self, path: Path) -> FileScan: line_count = 0 first_timestamp = "" last_timestamp = "" matched: List[Tuple[int, str]] = [] fallback: List[Tuple[int, str]] = [] try: with path.open("r", encoding="utf-8", errors="replace") as fp: for line_no, raw_line in enumerate(fp, 1): line_count = line_no line = raw_line.rstrip("\n") timestamp = TIMESTAMP_RE.search(line) if timestamp: ts = timestamp.group("ts") if not first_timestamp: first_timestamp = ts last_timestamp = ts if len(fallback) < 2 and line.strip(): fallback.append((line_no, line)) if KEYWORD_RE.search(path.name) or KEYWORD_RE.search(line): matched.append((line_no, self._trim_line(line))) except OSError as exc: matched.append((0, f"Read failed: {exc}")) if matched: if len(matched) <= MAX_HIGHLIGHTS_PER_FILE: highlights = matched else: head_count = MAX_HIGHLIGHTS_PER_FILE // 2 tail_count = MAX_HIGHLIGHTS_PER_FILE - head_count highlights = matched[:head_count] + matched[-tail_count:] else: highlights = [(line_no, self._trim_line(line)) for line_no, line in fallback] return FileScan( path=path, category=self._category_for(path), line_count=line_count, first_timestamp=first_timestamp, last_timestamp=last_timestamp, highlights=highlights, ) def _scan_files(self, paths: Iterable[Path]) -> List[FileScan]: return [self._scan_text_file(path) for path in paths] def _relative(self, path: Path, session_root: Path) -> str: try: return str(path.relative_to(session_root)) except ValueError: return str(path) def _append_inventory( self, lines: List[str], title: str, scans: Sequence[FileScan], ) -> None: lines.append(f"## {title}") lines.append("") lines.append(f"- 已检查文件数：{len(scans)}") by_category: Dict[str, List[FileScan]] = {} for scan in scans: by_category.setdefault(scan.category, []).append(scan) for category in sorted(by_category): category_scans = by_category[category] total_lines = sum(scan.line_count for scan in category_scans) category_label = CATEGORY_LABELS.get(category, category) lines.append(f"- {category_label}：{len(category_scans)} 个文件，{total_lines} 行") lines.append("") def _format_timeline_event(self, timestamp: str, text: str) -> str: event = text.strip() if event.startswith(timestamp): event = event[len(timestamp):].strip() return event def _read_scan_lines(self, scan: FileScan) -> List[str]: try: return scan.path.read_text(encoding="utf-8", errors="replace").splitlines() except OSError: return [] def _find_scan(self, scans: Sequence[FileScan], name: str) -> Optional[FileScan]: for scan in scans: if scan.path.name == name: return scan return None def _find_scans_by_pattern(self, scans: Sequence[FileScan], pattern: str) -> List[FileScan]: regex = re.compile(pattern) return [scan for scan in scans if regex.search(scan.path.name)] def _format_counter(self, counter: Dict[str, int]) -> str: if not counter: return "无" return ", ".join(f"{key}={counter[key]}" for key in sorted(counter)) def _percent(self, count: int, total: int) -> str: if total <= 0: return "0.0%" return f"{count * 100.0 / total:.1f}%" def _extract_key_values(self, text: str, keys: Sequence[str]) -> List[str]: values: List[str] = [] for key in keys: match = re.search(rf"\b{re.escape(key)}\s*[:=]\s*([^,\s}}]+)", text) if match: values.append(f"{key}={match.group(1)}") return values def _extract_key_value_map(self, text: str, keys: Sequence[str]) -> Dict[str, str]: values: Dict[str, str] = {} for key in keys: match = re.search(rf"\b{re.escape(key)}\s*[:=]\s*([^,\s}}]+)", text) if match: values[key] = match.group(1) return values def _extract_log_tag(self, text: str) -> str: match = re.search(r"\b[VDIWEF]\s+([A-Za-z0-9_.-]{2,80})\s*:", text) if match: return match.group(1) return "" def _summarize_timeline_event(self, category: str, text: str) -> str: event = " ".join(text.strip().split()) lower = event.lower() if re.search(r"(^|\s)at\s+[A-Za-z0-9_.$]+\(.*\)", event): return "异常堆栈展开（已归并，具体栈帧回看 filter 产物）" if "Process:" in event and "AndroidRuntime" in event: return "应用崩溃进程信息" if "ExceptionInInitializerError" in event: return "应用异常类型：ExceptionInInitializerError" if "mServingCellInfo update" in event or "CellInfo{" in event: fields = self._extract_key_values( event, ("inService", "isOos", "ratType", "pci", "freq", "band", "rsrp", "snr", "rsrq", "cellId", "mcc", "mnc"), ) if fields: return "服务小区状态：" + ", ".join(fields) if "DATA_REGISTRATION_STATE" in event or "VOICE_REGISTRATION_STATE" in event: state_match = re.search(r"regState:\s*([^,\s]+)", event) rat_match = re.search(r"rat:\s*([^,\s]+)", event) operator_match = re.search(r"operatorNumeric:\s*([^,\s}]+)", event) parts = [] if "DATA_REGISTRATION_STATE" in event: parts.append("数据注册") if "VOICE_REGISTRATION_STATE" in event: parts.append("语音注册") if state_match: parts.append(f"state={state_match.group(1)}") if rat_match: parts.append(f"rat={rat_match.group(1)}") if operator_match: parts.append(f"PLMN={operator_match.group(1)}") if parts: return "注册状态：" + ", ".join(parts) if "setPreferredDataModem" in event: match = re.search(r"phoneId[:=]\s*(-?\d+)", event) return f"DDS 请求切换到 phoneId={match.group(1)}" if match else "DDS 请求切换" if "DDS switch success" in event: match = re.search(r"phoneId\s*=\s*(-?\d+)", event) return f"DDS 切换成功 phoneId={match.group(1)}" if match else "DDS 切换成功" if "DATA_SUBSCRIPTION_CHANGED" in event or "ACTION_DEFAULT_DATA_SUBSCRIPTION_CHANGED" in event: return "默认数据卡订阅变化" if "UNSOL_DATA_CALL_LIST_CHANGED" in event or "SetupDataCallResult" in event: fields = self._extract_key_values(event, ("active", "type", "ifname")) return "DataCall 状态变化：" + ", ".join(fields) if fields else "DataCall 状态变化" if "DataStallParameters" in event: fields = self._extract_key_values(event, ("bsr", "grant", "pdcpDiscRatio", "rsrp", "isDataStall", "reason")) return "DataStall 参数：" + ", ".join(fields) if fields else "DataStall 参数上报" if "potential DataStall detected" in event: return "DataStall 潜在异常检测" if "DataStall:processDataStall" in event: fields = self._extract_key_values(event, ("currentCount",)) return "DataStall 处理计数：" + ", ".join(fields) if fields else "DataStall 处理计数更新" if "brain info cell changed" in event: old_match = re.search(r"oldCellId=([^,\s]+)", event) new_match = re.search(r"newCellId=([^,\s]+)", event) if old_match and new_match: return f"NetworkBrain 小区变化：oldCellId={old_match.group(1)}, newCellId={new_match.group(1)}" return "NetworkBrain 小区变化" if "setRecoveryAction" in event: match = re.search(r"setRecoveryAction:\s*([A-Za-z0-9_]+)", event) return f"数据恢复动作：{match.group(1)}" if match else "数据恢复动作更新" if "GET_CELL_INFO_LIST" in event: if "CellInfoNr" in event: return "小区列表刷新：包含 NR 小区" if "CellInfoLte" in event: return "小区列表刷新：包含 LTE 小区" if "[]" in event: return "小区列表刷新：返回空列表" return "小区列表刷新" if "NetworkRegistrationInfo" in event: fields = self._extract_key_values(event, ("domain", "transportType", "registrationState", "accessNetworkTechnology")) return "网络注册信息回调：" + ", ".join(fields) if fields else "网络注册信息回调" if "StatusForAccessTech" in event: fields = self._extract_key_values(event, ("networkMode", "registered", "status", "restrictCause")) return "IMS 接入技术状态：" + ", ".join(fields) if fields else "IMS 接入技术状态变化" if "notifyFenceChanged" in event: fields = self._extract_key_values(event, ("fenceType", "oucid", "fenceName")) return "围栏/小区关联通知：" + ", ".join(fields[:2]) if fields else "围栏/小区关联通知" if "onNrIconChange" in event or "onNrIconType" in event: fields = self._extract_key_values(event, ("slotId", "icon", "NrIconType")) return "RadioService NR 图标通知：" + ", ".join(fields) if fields else "RadioService NR 图标通知" if "getNrIconGroup" in event: fields = self._extract_key_values(event, ("nrIconType", "is6Rx", "show6RxConfig")) return "5G 图标状态：" + ", ".join(fields) if fields else "5G 图标状态刷新" if "battery_level" in event: match = re.search(r"battery_level:\s*\[([^\]]+)\]", event) return f"电池电量状态：[{match.group(1)}]" if match else "电池电量状态变化" if "charge counter" in lower: fields = self._extract_key_values(event, ("level", "voltage", "power")) return "充电/电量计状态：" + ", ".join(fields) if fields else "充电/电量计状态变化" if "FATAL EXCEPTION" in event: return "应用/系统异常：FATAL EXCEPTION" if "null object reference" in lower: return "异常风险：空对象引用" if "RefactorFlagAssert" in event: return "系统断言：RefactorFlagAssert" if "SingleThroughputCalculation" in event or "L4NetEvaluate" in event: fields = self._extract_key_values(event, ("ifName", "sendRate(Bps)", "recvRate(Bps)", "mRssi", "mBandwidth")) return "吞吐/Wi-Fi 状态：" + ", ".join(fields) if fields else "吞吐/Wi-Fi 状态刷新" tag = self._extract_log_tag(event) if category == "network": return f"网络事件：{tag}" if tag else "网络事件：未分类" if category == "data": return f"数据通路事件：{tag}" if tag else "数据通路事件：未分类" if category == "power": return f"功耗/电量事件：{tag}" if tag else "功耗/电量事件：未分类" if category == "device": return f"设备/系统事件：{tag}" if tag else "设备/系统事件：未分类" return f"其它事件：{tag}" if tag else "其它事件：未分类" def _append_timeline(self, lines: List[str], scans: Sequence[FileScan]) -> None: grouped: Dict[str, List[Tuple[str, str, str]]] = {} total_events = 0 for scan in scans: for _, text in scan.highlights: match = TIMESTAMP_RE.search(text) if match: ts = match.group("ts") event = self._format_timeline_event(ts, text) summary = self._summarize_timeline_event(scan.category, event) bucket = ts[:11] grouped.setdefault(bucket, []).append((ts, scan.category, summary)) total_events += 1 lines.append("## 跨产物时间线") lines.append("") lines.append("- 本节按时间窗口梳理问题场景、参数变化和关键事件，不重复列文件来源；原始行追溯请回到对应 filter 产物。") if not grouped: lines.append("- 未发现带时间戳的关键事件。") lines.append("") return if len(grouped) > MAX_TIMELINE_GROUPS: lines.append(f"- 共发现 {total_events} 条带时间戳事件，聚合为 {len(grouped)} 个时间窗口，当前展示前 {MAX_TIMELINE_GROUPS} 个窗口。") else: lines.append(f"- 共发现 {total_events} 条带时间戳事件，聚合为 {len(grouped)} 个时间窗口。") for bucket in sorted(grouped)[:MAX_TIMELINE_GROUPS]: events = grouped[bucket] categories = sorted({CATEGORY_LABELS.get(category, category) for _, category, _ in events}) first_ts = min(ts for ts, _, _ in events) last_ts = max(ts for ts, _, _ in events) lines.append(f"### {bucket}") lines.append("") lines.append(f"- 时间范围：{first_ts} -> {last_ts}") lines.append(f"- 涉及主题：{', '.join(categories)}") lines.append(f"- 事件规模：{len(events)} 条") summaries: List[str] = [] seen: Set[str] = set() for _, _, summary in events: if summary in seen: continue seen.add(summary) summaries.append(summary) if len(summaries) >= MAX_TIMELINE_ITEMS_PER_GROUP: break lines.append("- 关键变化：") for summary in summaries: lines.append(f" - {summary}") if len(seen) < len(events): lines.append(f" - 另有 {len(events) - len(seen)} 条同窗口事件已归并。") lines.append("") lines.append("") def _append_focus_summary(self, lines: List[str], scans: Sequence[FileScan]) -> None: focus_rules = [ ("设备信息", ("device",)), ("功耗与睡眠", ("power",)), ("网络与小区", ("network", "ril")), ("数据通路", ("data",)), ("GPS/定位", ("gps",)), ("传感器", ("sensor",)), ] lines.append("## 分析焦点检查") lines.append("") lines.append("- 本节只做主题级摘要，文件覆盖见后文产物覆盖摘要。") lines.append("") for title, categories in focus_rules: selected = [scan for scan in scans if scan.category in categories] lines.append(f"### {title}") lines.append("") if not selected: lines.append("- 未发现对应过滤产物。这只能说明当前证据缺失，不能直接证明行为正常。") lines.append("") continue highlight_count = sum(len(scan.highlights) for scan in selected) timestamps = [ ts for scan in selected for ts in (scan.first_timestamp, scan.last_timestamp) if ts ] time_span = "无时间戳" if timestamps: time_span = f"{min(timestamps)} -> {max(timestamps)}" representative = "未提取到代表性摘录。" for scan in selected: if scan.highlights: representative = self._summarize_timeline_event(scan.category, scan.highlights[0][1]) break lines.append(f"- 已检查产物数：{len(selected)}") lines.append(f"- 关键摘录数：{highlight_count}") lines.append(f"- 时间覆盖：{time_span}") lines.append(f"- 代表性现象：{representative}") lines.append("") def _append_analysis_summary( self, lines: List[str], scans: Sequence[FileScan], ap_scans: Sequence[FileScan], md_scans: Sequence[FileScan], ) -> None: texts = [text for scan in scans for _, text in scan.highlights] joined = "\n".join(texts) lower = joined.lower() timestamps = [ ts for scan in scans for ts in (scan.first_timestamp, scan.last_timestamp) if ts ] time_span = "无时间戳" if timestamps: time_span = f"{min(timestamps)} -> {max(timestamps)}" lines.append("## 分析结论摘要") lines.append("") lines.append(f"- 覆盖范围：AP 产物 {len(ap_scans)} 个，MD 产物 {len(md_scans)} 个，时间范围 {time_span}。") lines.append("- 结论性质：这是基于过滤产物的自动归纳，正文只保留状态链路、量化指标和异常信号；原始行按需回到 filter 产物追溯。") conclusions: List[str] = [] if "isoos=true" in lower and ("isoos=false" in lower or "inservice=true" in lower): conclusions.append("网络侧存在从 OOS/非服务态到 inService/非 OOS 的状态变化，重点看恢复窗口内小区、RAT、DDS 和 DataCall 的先后关系。") elif "isoos=true" in lower: conclusions.append("网络侧出现 OOS 证据，但当前自动摘要未同时捕获明确恢复点，需要回到对应 filter 产物确认恢复路径。") if "reg_home" in lower or "regstate: reg_home" in lower: conclusions.append("注册层有 REG_HOME 证据，说明至少部分窗口内注册状态回到 home；若用户体感仍异常，应重点看恢复耗时和数据业务可用性。") if "datastall" in lower or "data stall" in lower: conclusions.append("数据通路出现 DataStall 相关信号，需把 OOS/小区变化、DDS 切换、DataCall 连接状态与 DataStall 上报放在同一时间窗口分析。") if "setpreferreddatamodem" in lower or "dds switch success" in lower or "data_subscription_changed" in lower: conclusions.append("默认数据卡/DDS 在问题窗口内发生切换或刷新，可能影响数据注册、DataCall 重建和上层网络可用性判断。") if "recovery_action_radio_restart" in lower or "radio_restart" in lower: conclusions.append("出现 radio recovery / restart 动作信号，这通常不是普通驻网刷新，需确认触发前是否已有 DataStall、长时间无服务或小区异常。") if "fatal exception" in lower or "null object reference" in lower: conclusions.append("AP 侧伴随应用或系统异常日志；这些异常未必是 modem 根因，但会干扰用户体感和上层状态判断，不能和通信链路证据混为同一层结论。") if not any(scan.category == "gps" for scan in scans): conclusions.append("当前过滤结果缺少 GPS/定位专项产物，不能从本报告证明 GPS 行为正常或异常。") if not any(scan.category == "sensor" for scan in scans): conclusions.append("当前过滤结果缺少传感器专项产物，不能从本报告证明 sensor 行为正常或异常。") if not conclusions: conclusions.append("当前自动归纳未捕获明确异常链路，只能说明现有规则未命中强信号；仍需结合具体问题时间点补充人工判断。") for item in conclusions: lines.append(f"- {item}") lines.append("") lines.append("### 建议优先核查链路") lines.append("") lines.append("1. 先定位用户问题发生分钟，查看同窗口是否同时出现 OOS、RAT/小区变化、DDS 切换、DataCall 重建或 DataStall。") lines.append("2. 若存在 OOS -> REG_HOME / inService 恢复，重点看恢复耗时和恢复后数据业务是否立即可用。") lines.append("3. 若存在 radio recovery / restart，优先回看触发前 1-2 分钟的数据通路和注册状态变化。") lines.append("4. AP crash/null reference 只作为上层干扰信号，除非时间上严格领先通信异常，否则不要直接作为 modem 根因。") lines.append("") def _append_quantitative_analysis(self, lines: List[str], ap_scans: Sequence[FileScan]) -> None: lines.append("## 关键状态量化") lines.append("") lines.append("- 量化口径来自 `skill_log_keyword_aplog`：注册态优先看 `reg_sim*_data_reg_rat/voice_reg_rat`，`reg_data_sub_serv_cell` 只作为 NetworkBrain 业务侧快照。") lines.append("") self._append_serving_cell_metrics(lines, ap_scans) self._append_registration_metrics(lines, ap_scans) self._append_data_path_metrics(lines, ap_scans) lines.append("") def _append_serving_cell_metrics(self, lines: List[str], ap_scans: Sequence[FileScan]) -> None: scan = self._find_scan(ap_scans, "reg_data_sub_serv_cell.txt") or self._find_scan(ap_scans, "opt_networkbrain.txt") lines.append("### NetworkBrain 服务小区快照") lines.append("") if not scan: lines.append("- 缺少 `reg_data_sub_serv_cell.txt` / `opt_networkbrain.txt`，无法量化业务侧 OOS/inService 快照。") lines.append("") return samples: List[Dict[str, str]] = [] for raw_line in self._read_scan_lines(scan): if "mServingCellInfo update" not in raw_line and "CellInfo{" not in raw_line: continue fields = self._extract_key_value_map( raw_line, ("inService", "isOos", "dataRegState", "ratType", "pci", "freq", "band", "rsrp", "snr", "rsrq", "cellId", "mcc", "mnc"), ) if not fields: continue match = TIMESTAMP_RE.search(raw_line) fields["ts"] = match.group("ts") if match else "" samples.append(fields) total = len(samples) if total == 0: lines.append(f"- `{scan.path.name}` 存在，但未解析到可量化的 serving cell 样本。") lines.append("") return is_oos_counts: Dict[str, int] = {} in_service_counts: Dict[str, int] = {} data_reg_counts: Dict[str, int] = {} rat_counts: Dict[str, int] = {} cell_ids: Set[str] = set() rsrp_values: List[int] = [] oos_transitions = 0 recovery_transitions = 0 rat_changes = 0 last_oos = "" last_rat = "" for item in samples: is_oos = item.get("isOos", "") in_service = item.get("inService", "") data_reg = item.get("dataRegState", "") rat = item.get("ratType", "") if is_oos: is_oos_counts[is_oos] = is_oos_counts.get(is_oos, 0) + 1 if in_service: in_service_counts[in_service] = in_service_counts.get(in_service, 0) + 1 if data_reg: data_reg_counts[data_reg] = data_reg_counts.get(data_reg, 0) + 1 if rat: rat_counts[rat] = rat_counts.get(rat, 0) + 1 if item.get("cellId"): cell_ids.add(item["cellId"]) if item.get("rsrp"): try: rsrp_values.append(int(item["rsrp"])) except ValueError: pass if last_oos and is_oos and last_oos != is_oos: if last_oos == "false" and is_oos == "true": oos_transitions += 1 if last_oos == "true" and is_oos == "false": recovery_transitions += 1 if last_rat and rat and last_rat != rat: rat_changes += 1 if is_oos: last_oos = is_oos if rat: last_rat = rat first_ts = next((item.get("ts", "") for item in samples if item.get("ts")), "") last_ts = next((item.get("ts", "") for item in reversed(samples) if item.get("ts")), "") oos_true = is_oos_counts.get("true", 0) in_service_true = in_service_counts.get("true", 0) data_oos = data_reg_counts.get("1", 0) data_in_service = data_reg_counts.get("0", 0) lines.append(f"- 来源：`{scan.path.name}`，样本数 {total}，时间范围 {first_ts or '未知'} -> {last_ts or '未知'}。") lines.append(f"- isOos=true：{oos_true}/{total}（{self._percent(oos_true, total)}）；isOos=false：{is_oos_counts.get('false', 0)}/{total}（{self._percent(is_oos_counts.get('false', 0), total)}）。") lines.append(f"- inService=true：{in_service_true}/{total}（{self._percent(in_service_true, total)}）；dataRegState=0(IN_SERVICE)：{data_in_service}/{total}（{self._percent(data_in_service, total)}），dataRegState=1(OUT_OF_SERVICE)：{data_oos}/{total}（{self._percent(data_oos, total)}）。") lines.append(f"- OOS 转入次数：{oos_transitions}；OOS 恢复次数：{recovery_transitions}；RAT 切换次数：{rat_changes}；不同 cellId 数：{len(cell_ids)}。") lines.append(f"- RAT 分布：{self._format_counter(rat_counts)}。") if rsrp_values: avg_rsrp = sum(rsrp_values) / len(rsrp_values) lines.append(f"- RSRP 样本：{len(rsrp_values)}，范围 {min(rsrp_values)}~{max(rsrp_values)} dBm，均值 {avg_rsrp:.1f} dBm。") lines.append("- 解读：该章节是业务侧快照，只能证明 NetworkBrain 观察到的 OOS/inService 抖动；最终注册结论仍以后面的 RIL 注册态 SSOT 为准。") lines.append("") def _append_registration_metrics(self, lines: List[str], ap_scans: Sequence[FileScan]) -> None: lines.append("### RIL 注册态 SSOT") lines.append("") reg_scans = self._find_scans_by_pattern(ap_scans, r"reg_sim\d+_(data|voice)_reg_rat\.txt$") if not reg_scans: lines.append("- 缺少 `reg_sim*_data_reg_rat.txt` / `reg_sim*_voice_reg_rat.txt`，无法按 SSOT 量化注册态。") lines.append("") return for scan in reg_scans: state_counts: Dict[str, int] = {} rat_counts: Dict[str, int] = {} plmn_counts: Dict[str, int] = {} total = 0 non_registered = 0 for raw_line in self._read_scan_lines(scan): state_match = re.search(r"regState:\s*([A-Z0-9_]+)", raw_line) rat_match = re.search(r"rat:\s*([A-Z0-9_]+)", raw_line) plmn_match = re.search(r"operatorNumeric:\s*([^,\s}]+)", raw_line) if not state_match: continue total += 1 state = state_match.group(1) state_counts[state] = state_counts.get(state, 0) + 1 if state not in ("REG_HOME", "REG_ROAMING"): non_registered += 1 if rat_match: rat = rat_match.group(1) rat_counts[rat] = rat_counts.get(rat, 0) + 1 if plmn_match: plmn = plmn_match.group(1) plmn_counts[plmn] = plmn_counts.get(plmn, 0) + 1 role = "数据注册" if "_data_" in scan.path.name else "语音注册" lines.append(f"- `{scan.path.name}`（{role}）：样本数 {total}，非注册态 {non_registered}/{total}（{self._percent(non_registered, total)}）。") lines.append(f" - regState 分布：{self._format_counter(state_counts)}。") lines.append(f" - RAT 分布：{self._format_counter(rat_counts)}。") lines.append(f" - PLMN 分布：{self._format_counter(plmn_counts)}。") lines.append("") def _append_data_path_metrics(self, lines: List[str], ap_scans: Sequence[FileScan]) -> None: lines.append("### 数据通路 / DDS / DataStall") lines.append("") dds_scan = self._find_scan(ap_scans, "data_defaultDataSub.txt") if dds_scan: requests = 0 success = 0 phone_counts: Dict[str, int] = {} for raw_line in self._read_scan_lines(dds_scan): if "setPreferredDataModem" in raw_line: requests += 1 match = re.search(r"phoneId[:=]\s*(-?\d+)", raw_line) if match: phone_counts[match.group(1)] = phone_counts.get(match.group(1), 0) + 1 if "DDS switch success" in raw_line: success += 1 lines.append(f"- DDS：setPreferredDataModem 请求 {requests} 次，DDS switch success {success} 次，请求 phoneId 分布：{self._format_counter(phone_counts)}。") else: lines.append("- DDS：缺少 `data_defaultDataSub.txt`。") data_stall_scan = self._find_scan(ap_scans, "data_stall.txt") modemdiag_scan = self._find_scan(ap_scans, "opt_modemdiag.txt") data_lines = [] if data_stall_scan: data_lines.extend(self._read_scan_lines(data_stall_scan)) if modemdiag_scan: data_lines.extend(self._read_scan_lines(modemdiag_scan)) if not data_lines: lines.append("- DataStall：缺少 `data_stall.txt` / `opt_modemdiag.txt`。") lines.append("") return potential = sum(1 for line in data_lines if "potential DataStall detected" in line) param_lines = [line for line in data_lines if "DataStallParameters" in line] actual_true = sum(1 for line in param_lines if "isDataStall=true" in line) actual_false = sum(1 for line in param_lines if "isDataStall=false" in line) valid = sum(1 for line in data_lines if "ValidationStatusChanged" in line and "VALID" in line) disconnected = sum(1 for line in data_lines if "All InternetDataNetwork Disconnected" in line) connected = sum(1 for line in data_lines if "At Least One InternetDataNetwork Connected" in line) current_counts = [] rsrp_values = [] for line in data_lines: count_match = re.search(r"currentCount[:=]\s*(\d+)", line) if count_match: current_counts.append(int(count_match.group(1))) rsrp_match = re.search(r"rsrp=([-]?\d+)", line) if rsrp_match: rsrp_values.append(int(rsrp_match.group(1))) lines.append(f"- DataStall：potential detected {potential} 次，参数上报 {len(param_lines)} 次，其中 isDataStall=true {actual_true} 次、false {actual_false} 次。") if current_counts: lines.append(f"- DataStall 处理计数最大值：{max(current_counts)}。") if rsrp_values: lines.append(f"- DataStall 参数中的 RSRP：{min(rsrp_values)}~{max(rsrp_values)} dBm，均值 {sum(rsrp_values) / len(rsrp_values):.1f} dBm。") lines.append(f"- InternetDataNetwork：connected 事件 {connected} 次，VALID 事件 {valid} 次，disconnected 事件 {disconnected} 次。") lines.append("") def _append_md_status_summary(self, lines: List[str], mdlog_filter: Path, session_root: Path) -> None: status_files = self._list_mask_status_files(mdlog_filter) lines.append("## MD 解析状态") lines.append("") if not status_files: lines.append("- 未发现 `mask_status.json`，无法从状态清单证明 MD filter 完成态。") lines.append("") return for status_file in status_files: rel = self._relative(status_file, session_root) try: data = json.loads(status_file.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: lines.append(f"- `{rel}`：读取失败，{exc}") continue global_masks = data.get("global_masks", {}) if not isinstance(global_masks, dict): lines.append(f"- `{rel}`：格式异常，缺少 `global_masks`。") continue by_status: Dict[str, List[str]] = {} blocked_masks: List[str] = [] for mask_name, item in global_masks.items(): if not isinstance(item, dict): by_status.setdefault("UNKNOWN", []).append(str(mask_name)) continue by_status.setdefault(str(item.get("status", "UNKNOWN")), []).append(str(mask_name)) diagnostics = item.get("diagnostics") if isinstance(diagnostics, dict): blocked = diagnostics.get("packet_decode_blocked") if isinstance(blocked, dict): count = blocked.get("count", 0) txt_file = blocked.get("txt_file", "") markers = ", ".join(str(m) for m in blocked.get("markers", [])) blocked_masks.append( f"{mask_name}(count={count}, txt={txt_file}, markers={markers})" ) total = sum(len(names) for names in by_status.values()) success = len(by_status.get("SUCCESS_NON_EMPTY", [])) empty = len(by_status.get("EMPTY_SUCCESS", [])) failed = len(by_status.get("FAILED", [])) unknown = total - success - empty - failed lines.append( f"- `{rel}`：total={total}, SUCCESS_NON_EMPTY={success}, " f"EMPTY_SUCCESS={empty}, FAILED={failed}, UNKNOWN={unknown}" ) failed_masks = by_status.get("FAILED", []) if failed_masks: sample = ", ".join(failed_masks[:20]) suffix = f"，另有 {len(failed_masks) - 20} 个" if len(failed_masks) > 20 else "" lines.append(f" - 失败 mask 示例：{sample}{suffix}") if blocked_masks: sample = "；".join(blocked_masks[:20]) suffix = f"；另有 {len(blocked_masks) - 20} 个" if len(blocked_masks) > 20 else "" lines.append( " - QCAT packet payload 加密或当前 QCAT 不支持解析：" f"{sample}{suffix}。这些 mask 的 CSV 可能缺失或不完整，不能据此判断 NR 侧无事件。" ) lines.append("") def _append_md_full_visible_summary(self, lines: List[str], mdlog_filter: Path, session_root: Path) -> None: summary_files = sorted(mdlog_filter.rglob("full_visible/full_visible_summary.json")) lines.append("## MD 全量可见证据") lines.append("") if not summary_files: lines.append("- 未发现 `full_visible_summary.json`；本次报告没有纳入 QCAT 全可见导出或 strings 兜底索引。") lines.append("") return for summary_file in summary_files: rel = self._relative(summary_file, session_root) try: data = json.loads(summary_file.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: lines.append(f"- `{rel}`：读取失败，{exc}") continue qcat = data.get("qcat_all_visible", {}) strings = data.get("strings_visible", {}) qcat_status = qcat.get("status", "UNKNOWN") if isinstance(qcat, dict) else "UNKNOWN" qcat_count = qcat.get("record_count", 0) if isinstance(qcat, dict) else 0 strings_status = strings.get("status", "UNKNOWN") if isinstance(strings, dict) else "UNKNOWN" strings_count = strings.get("row_count", 0) if isinstance(strings, dict) else 0 input_folder = data.get("input_folder", "") lines.append( f"- `{rel}`：input=`{input_folder}`，" f"QCAT={qcat_status}/{qcat_count} records，strings={strings_status}/{strings_count} lines。" ) if isinstance(qcat, dict): coverage = qcat.get("coverage_summary", {}) if isinstance(coverage, dict): record_types = coverage.get("record_type_counts", {}) categories = coverage.get("category_counts", {}) defined_count = coverage.get("defined_filter_record_count", 0) unmapped_count = coverage.get("unmapped_filter_record_count", 0) if record_types: type_text = ", ".join(f"{key}={value}" for key, value in sorted(record_types.items())) lines.append(f" - QCAT 可见类型分布：{type_text}") if categories: category_text = ", ".join(f"{key}={value}" for key, value in sorted(categories.items())) lines.append(f" - QCAT 语义粗分类：{category_text}") lines.append( " - QCAT filter 覆盖：" f"现有规则内 records={defined_count}，现有规则外可见 records={unmapped_count}。" ) unmapped = coverage.get("top_unmapped_log_ids", []) if isinstance(unmapped, list) and unmapped: sample = ", ".join( f"{item.get('value', '')}({item.get('count', 0)})" for item in unmapped[:20] if isinstance(item, dict) ) lines.append( " - 现有 mask 未定义但 QCAT 可见的 log id 示例：" f"{sample}。这些记录必须作为 raw evidence 纳入人工检索，不能因无 CSV 而判定无事件。" ) index_files = qcat.get("index_files", []) if index_files: index_rel = [self._relative(Path(p), session_root) for p in index_files[:5]] lines.append(f" - QCAT index：{', '.join(f'`{p}`' for p in index_rel)}") if isinstance(strings, dict): coverage = strings.get("coverage_summary", {}) if isinstance(coverage, dict): categories = coverage.get("category_counts", {}) if categories: category_text = ", ".join(f"{key}={value}" for key, value in sorted(categories.items())) lines.append(f" - strings 语义粗分类：{category_text}") index_file = strings.get("index_file") if index_file: lines.append(f" - strings index：`{self._relative(Path(index_file), session_root)}`") lines.append("") def write_local_report(self, session_root: Path) -> Path: out_dir = ensure_dir(session_root / self.layout.analysis) aplog_filter = session_root / self.layout.aplog_filter mdlog_filter = session_root / self.layout.mdlog_filter plots_dir = session_root / self.layout.analysis / "plots" ap_scans = self._scan_files(self._list_text_outputs(aplog_filter)) md_scans = self._scan_files(self._list_text_outputs(mdlog_filter)) all_scans = ap_scans + md_scans md_xlsx = self._list_files(mdlog_filter, ".xlsx") report = out_dir / "analysis_report.md" lines: List[str] = [] lines.append("# 日志工作流分析报告") lines.append("") lines.append(f"- 会话目录：`{session_root}`") lines.append(f"- AP 过滤目录：`{aplog_filter}`") lines.append(f"- MD 过滤目录：`{mdlog_filter}`") lines.append("") lines.append("## 报告约束") lines.append("") lines.append("- 本报告已枚举并检查当前会话下的所有 AP/MD 过滤产物。") lines.append("- 空产物或缺失主题只能说明证据缺失，不能直接证明行为正常。") lines.append("- 正文只保留分析结论、状态链路和关键参数变化，不展开原始日志摘录。") lines.append("- 产物覆盖只保留分类统计；原始行按需回到 filter 产物查看，避免报告退化成文件索引。") lines.append("- AP log 时间基准为北京时间，MD csv 时间基准为 UTC；联合分析必须按 AP = MD + 8h 对齐。") lines.append("") self._append_analysis_summary(lines, all_scans, ap_scans, md_scans) self._append_quantitative_analysis(lines, ap_scans) self._append_timeline(lines, all_scans) self._append_focus_summary(lines, all_scans) lines.append("## 产物覆盖摘要") lines.append("") lines.append(f"- 已检查 AP 文本产物数：{len(ap_scans)}") lines.append(f"- 已检查 MD 文本/CSV 产物数：{len(md_scans)}") lines.append(f"- 发现 MD xlsx 产物数：{len(md_xlsx)}") lines.append("") self._append_inventory(lines, "AP 产物覆盖", ap_scans) self._append_inventory(lines, "MD 产物覆盖", md_scans) self._append_md_status_summary(lines, mdlog_filter, session_root) self._append_md_full_visible_summary(lines, mdlog_filter, session_root) lines.append("## 图表产物") plot_files = self._list_plot_files(plots_dir, limit=None) lines.append(f"- 图表产物数：{len(plot_files)}") for p in plot_files: lines.append(f" - `{self._relative(p, session_root)}`") lines.append("") lines.append("## 下一步：飞书报告") lines.append("- 使用 `skill_log_feishu_analysis_report`，并把本文档与 `feishu_report_assets.md/json` 一起作为报告输入。") lines.append("- AP+MD 联合分析必须显式写出时间基准：`AP = Beijing time`，`MD csv = UTC`，结论前按 `AP = MD + 8h` 对齐。") lines.append("- 每个关键指标都应同时给出摘要表和对应图。") lines.append("- 飞书正文优先使用单指标 PNG，不要把一张超长 combined 图作为唯一主图。") lines.append("- 如果存在多图 HTML 报告，飞书文档中必须同时提供直接跳转链接和原始 HTML 附件。") lines.append("- 使用 `lark-cli docs +media-insert` 插图时，先切到图片目录，再使用类似 `./figure.png` 的相对路径。") lines.append("- 最终飞书链接必须是干净永久链接，格式为 `https://bytedance.larkoffice.com/docx/`。") lines.append("- 如果需要引用 NR 分析指南，使用 `skill_log_analysis_workflow` 中的 NR/5G 章节。") lines.append("") report.write_text("\n".join(lines) + "\n", encoding="utf-8") return report