fix(monitor): 失败签名三修——ERROR行预滤(重试WARNING的Expecting value不误中source_changed)+418针收紧'418 Client Error'(裸418子串误中rows=41823进度行=anti_crawl假阳性实证)+签名按[type]归因signatures_by_type(xueqiu的418不再抬share_capital的failed=跨类型连坐修); 10-05巡检案三连红假阳全案入spec §5.2/§9.B-21,22(research_report band跨车道混基线留痕) [vps]

2026-10-05 08:23:16 +08:00
parent 3cdf7e211e
commit 85830b6887
5 changed files with 93 additions and 20 deletions
@@ -239,7 +239,7 @@ def collect_log_layer(data_root, now, window_h=_LOG_WINDOW_H):
    keep = [p for p in logs
            if (now - dt.datetime.fromtimestamp(os.path.getmtime(p))
                ).total_seconds() / 3600 <= window_h]
-    stats, sigs = {}, []
+    stats, sigs_by_type = {}, {}
    for p in keep:
        try:
            with open(p, encoding="utf-8", errors="replace") as f:
@@ -250,18 +250,28 @@ def collect_log_layer(data_root, now, window_h=_LOG_WINDOW_H):
            prev = stats.get(t)
            stats[t] = s if prev is None or s.get("rows", 0) >= prev.get("rows", 0) \
                else prev
-        sigs.extend(chk.scan_signatures(text))
-    return stats, sigs
+        # 10-05: 签名按 [type] 归因(跨类型连坐修), 跨文件按 sig 去重合并
+        for t, ss in chk.signatures_by_type(text).items():
+            bucket = sigs_by_type.setdefault(t, [])
+            have = {x["sig"] for x in bucket}
+            for s in ss:
+                if s["sig"] not in have:
+                    bucket.append(s)
+                    have.add(s["sig"])
+    return stats, sigs_by_type


-def log_layer_results(stats, sigs, today):
-    """统计行 failed>0 → 按 sig 分级告警; 断路器独立键; 其余 ok → passed."""
+def log_layer_results(stats, sigs_by_type, today):
+    """统计行 failed>0 → 按「该类型自己的」ERROR 行签名分级(10-05 归因修:
+    他类型反爬签名不再抬本类型 failed); 断路器独立键(全局); 其余 ok → passed."""
    results, passed = [], []
-    worst = max((s["severity"] for s in sigs), default=None)
-    breaker = any(s["sig"] == "breaker" for s in sigs)
+    breaker = any(s["sig"] == "breaker"
+                  for ss in sigs_by_type.values() for s in ss)
    for t, s in sorted(stats.items()):
        if s.get("failed", 0) > 0:
-            ev = {"stats": s, "signatures": sigs}
+            tsigs = sigs_by_type.get(t, [])
+            worst = max((x["severity"] for x in tsigs), default=None)
+            ev = {"stats": s, "signatures": tsigs}
            sev = "red" if (worst == "red" or breaker) else "yellow"
            results.append(_event(f"data-{t}-failed", sev,
                                  f"{t} 类型级 failed={s['failed']}",
@@ -271,19 +281,22 @@ def log_layer_results(stats, sigs, today):
    if breaker:
        results.append(_event("data-akshare-breaker", "red",
                              "断路器触发(熔断连累: 后续类型整批跳过)",
-                              "exit code 2 会话", {"signatures": sigs}, today=today))
+                              "exit code 2 会话", {"signatures": sigs_by_type},
+                              today=today))
    else:
        passed.append("data-akshare-breaker")
    return results, passed


-def log_only_results(entries, stats, sigs, today):
-    """valuation 等 log_only 项: 统计行存在且 failed=0(周班项平日无行属正常)."""
+def log_only_results(entries, stats, sigs_by_type, today):
+    """valuation 等 log_only 项: 统计行存在且 failed=0(周班项平日无行属正常).
+    10-05: red 判定用该类型自己的签名(归因修)."""
    results, passed = [], []
-    red_sig = any(s["severity"] == "red" for s in sigs)
    for e in entries:
        key = f"data-{e['name']}-stat"
        s = stats.get(e["name"])
+        red_sig = any(x["severity"] == "red"
+                      for x in sigs_by_type.get(e["name"], []))
        if s is None:
            if e["log_cadence"] == "weekly":
                continue
@@ -464,11 +477,11 @@ def main(argv=None):
                source="infra" if r["key"].startswith("infra-") else "data",
                today=today))

-    stats, sigs = collect_log_layer(args.data_root, now)
-    log_events, log_passed = log_layer_results(stats, sigs, today)
+    stats, sigs_by_type = collect_log_layer(args.data_root, now)
+    log_events, log_passed = log_layer_results(stats, sigs_by_type, today)
    lo_events, lo_passed = log_only_results(
        [e for e in reg.REGISTRY if e["kind"] == "log_only"
-         and e.get("enabled", True)], stats, sigs, today)
+         and e.get("enabled", True)], stats, sigs_by_type, today)
    events += log_events + lo_events
    passed += log_passed + lo_passed

@@ -165,7 +165,7 @@ SIGNATURES = [
    ("断路器触发", "breaker", "red"),
    ("RemoteDisconnected", "disconnect", "red"),
    ("Expecting value", "source_changed", "red"),
-    ("418", "anti_crawl", "red"),
+    ("418 Client Error", "anti_crawl", "red"),
    ("'NoneType' object is not subscriptable", "source_empty", "yellow"),
    ("KeyError", "field_changed", "red"),
 ]
@@ -182,15 +182,44 @@ def parse_stat_lines(text):
    return out


+def _error_lines(text):
+    """只留 ERROR 行(重试 WARNING 良性) + [FATAL] 断路器行。"""
+    return [ln for ln in text.splitlines()
+            if " ERROR " in ln or "[FATAL]" in ln]
+
+
 def scan_signatures(text):
+    """10-05 三修: ①只扫 ERROR/[FATAL] 行——'Expecting value' 出现在重试
+    WARNING 行(financial_abstract/000963 实证), 良性重试不再误中
+    source_changed; ②'418' 针收紧为 '418 Client Error'——裸 418 子串误中
+    rows=41823/ok=19418 进度行(anti_crawl 假阳性实证)。"""
+    lines = _error_lines(text)
    seen, out = set(), []
    for needle, sig, severity in SIGNATURES:
-        if needle in text and sig not in seen:
+        if any(needle in ln for ln in lines) and sig not in seen:
            seen.add(sig)
            out.append({"sig": sig, "severity": severity})
    return out


+_TYPE_TAG = re.compile(r"\[(\w+)\]")
+
+
+def signatures_by_type(text):
+    """签名按日志行 [type] 归因(10-05): 'ERROR [share_capital] ...' 行的
+    签名只归 share_capital——防跨类型连坐(xueqiu 的 418 把 share_capital
+    的 failed 抬红实证)。断路器行归触发类型, data_monitor 侧全局取。"""
+    out = {}
+    for ln in _error_lines(text):
+        m = _TYPE_TAG.search(ln)
+        t = m.group(1) if m else "_global"
+        sigs = out.setdefault(t, [])
+        for needle, sig, severity in SIGNATURES:
+            if needle in ln and sig not in {s["sig"] for s in sigs}:
+                sigs.append({"sig": sig, "severity": severity})
+    return out
+
+
 # ---------- 调度层: schtasks /query /fo LIST /v 解析(双语字段名) ----------

 _FIELD_PATTERNS = {