1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
| import sqlite3
import os
from collections import defaultdict
class CDPErrorAggregator:
"""CDP 错误聚合系统——收集、去重、分析页面异常"""
def __init__(self, ws, session_id, db_path="error_tracking.db"):
self.ws = ws
self.session_id = session_id
self.db_path = db_path
self._init_db()
def _init_db(self):
"""初始化 SQLite 数据库"""
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS errors (
id INTEGER PRIMARY KEY AUTOINCREMENT,
error_type TEXT,
message TEXT,
url TEXT,
line_number INTEGER,
column_number INTEGER,
stack_hash TEXT,
full_stack TEXT,
timestamp TEXT,
page_url TEXT,
session_id TEXT
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_stack_hash
ON errors(stack_hash)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_timestamp
ON errors(timestamp)
""")
conn.commit()
conn.close()
@staticmethod
def compute_stack_hash(stack_trace):
"""计算堆栈的哈希值用于去重"""
if not stack_trace:
return "no_stack"
call_frames = stack_trace.get("callFrames", [])
key_parts = []
for frame in call_frames[:3]:
key_parts.append(f"{frame.get('functionName', '')}@{frame.get('lineNumber', '')}")
return hashlib.md5("|".join(key_parts).encode()).hexdigest() if key_parts else "no_frames"
async def collect_with_aggregation(self, duration=60):
"""收集并聚合错误"""
await cdp(self.ws, "Runtime.enable", {}, self.session_id)
error_buffer = []
start = asyncio.get_event_loop().time()
current_page_url = ""
while (asyncio.get_event_loop().time() - start) < duration:
try:
msg = await asyncio.wait_for(self.ws.__anext__(), timeout=1)
data = json.loads(msg)
method = data.get("method", "")
params = data.get("params", {})
if method == "Page.frameNavigated":
frame = params.get("frame", {})
if frame.get("id") == params.get("frame", {}).get("loaderId"):
current_page_url = frame.get("url", "")
if method == "Runtime.exceptionThrown":
details = params.get("exceptionDetails", {})
stack_trace = details.get("stackTrace", {})
stack_hash = self.compute_stack_hash(stack_trace)
entry = {
"error_type": details.get("text", "").split(":")[0] if ":" in details.get("text", "") else "Error",
"message": details.get("text", ""),
"url": details.get("url", ""),
"line_number": details.get("lineNumber", 0),
"column_number": details.get("columnNumber", 0),
"stack_hash": stack_hash,
"full_stack": json.dumps(stack_trace),
"timestamp": datetime.now().isoformat(),
"page_url": current_page_url,
"session_id": self.session_id
}
error_buffer.append(entry)
except asyncio.TimeoutError:
continue
self._batch_insert(error_buffer)
return self.generate_report(error_buffer)
def _batch_insert(self, entries):
"""批量写入错误到数据库"""
conn = sqlite3.connect(self.db_path)
conn.executemany("""
INSERT INTO errors
(error_type, message, url, line_number, column_number,
stack_hash, full_stack, timestamp, page_url, session_id)
VALUES
(:error_type, :message, :url, :line_number, :column_number,
:stack_hash, :full_stack, :timestamp, :page_url, :session_id)
""", entries)
conn.commit()
conn.close()
def generate_report(self, error_buffer):
"""生成错误聚合报告"""
if not error_buffer:
return {"total": 0, "message": "未捕获到异常"}
by_hash = defaultdict(lambda: {"count": 0, "first": "", "last": "", "entry": None})
for entry in error_buffer:
h = entry["stack_hash"]
by_hash[h]["count"] += 1
by_hash[h]["entry"] = entry
if not by_hash[h]["first"]:
by_hash[h]["first"] = entry["timestamp"]
by_hash[h]["last"] = entry["timestamp"]
by_type = defaultdict(int)
for entry in error_buffer:
by_type[entry["error_type"]] += 1
by_page = defaultdict(int)
for entry in error_buffer:
by_page[entry["page_url"]] += 1
return {
"total": len(error_buffer),
"unique_errors": len(by_hash),
"collection_period": f"{error_buffer[0]['timestamp']} ~ {error_buffer[-1]['timestamp']}",
"by_type": dict(by_type),
"by_page": dict(by_page),
"unique_stacks": [
{
"hash": h,
"count": info["count"],
"type": info["entry"]["error_type"],
"message": info["entry"]["message"][:100],
"url": info["entry"]["url"],
"line": info["entry"]["line_number"],
"first_seen": info["first"],
"last_seen": info["last"]
}
for h, info in sorted(by_hash.items(), key=lambda x: -x[1]["count"])
]
}
|
