Tencent Cloud CAT - Network Quality Analysis
将 CAT 网络质量分析与 PDF 报告生成串联为一条顺序管道。调用 API 获取 AI 分析结果,解析为结构化 JSON,再通过 Playwright/Chromium 渲染为专业 PDF 报告——一条命令完成。
Architecture
┌─────────────────┐ ┌──────────────────┐ ┌──────────────────────┐
│ CAT AI Console │────▶│ Markdown Parser │────▶│ PDF Generator │
│ (SSE → Text) │ │ (Rule-based) │ │ (HTML → Playwright) │
└────────┬────────┘ └──────────────────┘ └──────────────────────┘
│ ▲
│ pcap_check.done │
│ (tool_data) │
└──▶ _build_pcap_section ┘
(Data.items → 表格)
- SSE 分析调用: 调用
ProcessAIEventsStreamAPI,消费 SSE 事件流,获取完整的 Markdown 分析文本。支持多种 done 事件类型:agent.start— 包含会话 ID(SessionID),记录后继续消费后续事件agent.done— 包含最终分析文本(FullText),仅此事件终止流消费pcap_check.done— 包含抓包候选列表(Data.items),记录 tool_data 后继续消费后续事件- 其他
*.done— 记录日志后继续消费
- Markdown 解析: 使用纯规则解析器(无 LLM 依赖)将 Markdown 分析文本转为 PDF 生成所需的结构化 JSON。支持多级标题映射:
#/##→ 主章节(section)###→ 子章节(独立 section)####/#####→ 子块标题(关联到紧随其后的表格)- 自动过滤 Markdown 水平分割线(
---、***、___)
- PDF 生成: 将结构化 JSON 渲染为 HTML,通过 Playwright 驱动 Chromium 无头浏览器将 HTML 转为 PDF,原生支持彩色 emoji 和高质量中文排版。
如果 SSE 流中包含 pcap_check.done 事件,其 Data.items(抓包候选列表)会被自动转为额外的表格章节追加到报告末尾,包含城市、运营商、错误码、任务ID、拨测时间、拨测点编码等字段,同时以格式化表格输出到控制台(stderr)。
三个步骤通过纯函数顺序调用串联,无额外框架依赖。
Prerequisites
1. Install Python packages
pip3 install tencentcloud-sdk-python playwright certifi
2. Install Chromium browser (Playwright dependency)
python3 -m playwright install chromium
⚠️ Linux: Chromium requires system-level shared libraries. If installation fails, install system dependencies first:
Option A (recommended): Use Playwright's built-in command (requires sudo):
python3 -m playwright install --with-deps chromiumOption B: Install dependencies manually:
Debian / Ubuntu:
sudo apt-get update sudo apt-get install -y libnss3 libatk1.0-0 libatk-bridge2.0-0 libcups2 \ libdrm2 libgbm1 libxkbcommon0 libxcomposite1 libxdamage1 libxrandr2 \ libpango-1.0-0 libcairo2 libasound2CentOS / RHEL:
sudo yum install -y nss atk at-spi2-atk cups-libs libdrm mesa-libgbm \ libxkbcommon libXcomposite libXdamage libXrandr pango cairo alsa-libThen re-run
python3 -m playwright install chromium.
3. Configure Tencent Cloud credentials
Via either method:
环境变量优先级: 优先使用
CAT_xxx系列变量,未设置时自动降级使用TENCENTCLOUD_xxx系列变量。
Method A: .env file (recommended)
Create a .env file in the script directory or current working directory:
CAT_SECRET_ID=your-secret-id
CAT_SECRET_KEY=your-secret-key
CAT_TOKEN=your-token
CAT_ENDPOINT=cat.ai.tencentcloudapi.com
Method B: Environment variables
export CAT_SECRET_ID='your-secret-id'
export CAT_SECRET_KEY='your-secret-key'
export CAT_ENDPOINT='cat.ai.tencentcloudapi.com' # 可选,自定义 API 端点
💡 也兼容旧版变量名
TENCENTCLOUD_SECRET_ID/TENCENTCLOUD_SECRET_KEY/TENCENTCLOUD_TOKEN/TENCENTCLOUD_ENDPOINT,但建议优先使用CAT_xxx。
Usage
Run the bundled script:
python3 ${SKILL_DIR}/scripts/cat_network_quality_analysis.py \
--query-text "<QueryText>" \
[--task-id "<TaskID>"] \
[--start-time <StartTime>] \
[--end-time <EndTime>] \
[--session-id "<SessionID>"] \
[--output <output.pdf>]
${SKILL_DIR}refers to the base directory of this skill.--query-textis required: the analysis query to send to CAT AI Console.--output: path for the output PDF. Must be provided when TaskID is present and the query is a summarization/analysis request (see Workflow rules). Defaults to{task_id}_error_report.pdf.
Example
python3 ${SKILL_DIR}/scripts/cat_network_quality_analysis.py \
--query-text "请总结分析最近3小时的错误情况" \
--task-id "task-xxxxxxxx" \
--start-time 1773658839444 \
--end-time 1773745239444 \
--output report.pdf
Handling the Response
The script outputs progress to stderr and the final PDF path to stdout:
[Stream] 正在调用 CAT AI Console…
[Stream] pcap_check.done — 完整文本 0 chars
[Stream] 附带工具数据: pcap_check
[Stream] agent.done — 完整文本 4096 chars
[Stream] 分析完成,共 15 个事件
[Stream] SessionID: 17f57xxxxxxxxxxxxxxxx5b224055354
[Pipeline] 分析文本就绪 (4096 chars)
[Pipeline] 正在将分析结果转为结构化报告…
[Pipeline] 已追加抓包分析候选列表章节
============================================================
🔍 抓包分析候选列表
============================================================
检测到 3 个错误记录配置了"错误时抓包"模式,以下为抓包分析候选项。
城市 | 运营商 | 错误码 | 任务ID | 拨测时间 | 拨测点编码
---------+----------------+--------+------------------+---------------------+-----------
慕尼黑 | Germany_COLT | 600 | task-xxxxxxxx | 2026-03-17 07:06:35 | xxxxxx
...
============================================================
[Pipeline] 解析完成: 12 个章节, 7 个表格
[PDF] 正在生成 HTML…
[PDF] 正在启动 Chromium 渲染 PDF…
[PDF] PDF 报告已生成: report.pdf
The script also outputs SessionID=<session_id> to stdout (after the PDF path). You MUST capture and remember this SessionID for use in subsequent pcap analysis invocations (Step 5b).
Workflow
⏱ Execution mode — Background + Poll output file (IMPORTANT)
This script may take a long time (SSE streaming + Chromium PDF rendering typically takes 90–150 seconds). OpenClaw's
execute_commandhas a built-in timeout on the entire shell invocation, so evennohup ... & wait $PIDwill be killed when the timeout fires onwait.You MUST use a two-phase approach: launch in background, then poll for the output file.
Phase 1 — Launch (one command, returns immediately)
nohup python3 ${SKILL_DIR}/scripts/cat_network_quality_analysis.py [args...] > /tmp/cat_report_stdout.log 2> /tmp/cat_report_stderr.log & echo "launched"This command will return instantly. Do NOT add
waithere.Phase 2 — Three-stage adaptive polling (separate commands, call repeatedly)
The script typically takes 90–150 seconds (SSE streaming ~60–120s + Chromium PDF rendering ~10–30s). To minimize wasted poll calls while still catching early errors quickly, use a three-stage polling strategy:
| Stage | Poll # | Sleep interval | Purpose | |-------|--------|---------------|---------| | Quick probe | 1–3 |
sleep 10| Catch early failures (bad args, missing deps, auth errors) | | Long wait | 4–6 |sleep 30| Skip the SSE streaming phase where the script is always busy | | Regular poll | 7+ |sleep 10| Script should be near completion; check frequently |Poll command template (replace
<OUTPUT_PDF_PATH>with the actual--outputpath, and<SLEEP>with the appropriate interval):sleep <SLEEP> && test -f "<OUTPUT_PDF_PATH>" && echo "DONE" || (grep -i "error\|traceback\|exception" /tmp/cat_report_stderr.log && echo "HAS_ERROR" || echo "RUNNING")Polling rules:
RUNNING→ do nothing, do NOT reply to the user. Just silently issue the next poll command (with the appropriate sleep interval for the current stage). Repeat untilDONE.HAS_ERROR→ report the error to the user and stop.DONE→ the PDF file exists, the script has finished. Read the results:cat /tmp/cat_report_stderr.log # check for errors cat /tmp/cat_report_stdout.log # get final output (PDF path)IMPORTANT: While the output file has not appeared and no errors are detected, do NOT send any message to the user. Stay silent and keep polling. Only speak to the user when the file is generated (
DONE) or when an error is detected.Why NOT
nohup ... & wait?OpenClaw's timeout applies to the entire shell invocation.
waitblocks the shell, so the timeout kills the shell (and thewait), then OpenClaw reports "process terminated" — even though thenohupchild is still running in the background. The two-phase approach avoids this because each command returns quickly.Do NOT use
wait,timeout, or any other blocking pattern in the same command as the script launch.
PDF output rule
⚠️ When BOTH of the following conditions are met, you MUST generate and deliver a PDF report — even if the user did not explicitly ask for a PDF:
- The user provided a TaskID (e.g.
task-xxxxxxxx)- The user's intent is summarization or analysis (keywords: 总结, 分析, 汇总, 报告, summary, analyze, report, diagnose, 诊断, 排查)
If either condition is not met (no TaskID, or the user is just asking a simple question), run the analysis normally without PDF generation.
Pre-check — Missing TaskID handling (IMPORTANT)
🚫 If the user's intent is summarization/analysis but NO TaskID is provided, you MUST NOT execute this skill yet.
Instead, you MUST:
- Ask the user to provide the TaskID — reply with a message like: "请提供需要分析的拨测任务 ID(格式如
task-xxxxxxxx),我才能为您生成分析报告。"- Remind about WeChat Work (企微) group bots — append this reminder: "💡 如果您是在企微群中使用机器人,请记得 @机器人 来回复任务 ID,否则机器人无法收到您的消息。"
- Wait for the user's reply with the TaskID before proceeding to Step 1.
Do NOT attempt to run the script without a TaskID when the intent is summarization/analysis.
Step 1 — Collect parameters
- Confirm the user has provided
QueryText(required). Prompt for it if missing. - Collect optional parameters (
TaskID,StartTime,EndTime,SessionID). - Time range default: if the user did not specify
StartTime/EndTime, automatically default to the last 3 hours. Compute as:EndTime= current timestamp in millisecondsStartTime=EndTime− 10800000 (3 hours in milliseconds)- Do NOT ask the user for the time range — just use the default silently.
- Check PDF rule: if
TaskIDis present AND the query is a summarization/analysis intent → proceed to Step 2. Otherwise skip to Step 3 without--output.
Step 2 — Determine output path (when PDF rule is triggered)
Compute the output PDF path and pass it via --output.
2a. Choose report type by user intent
| User intent | Keywords | Report type |
|-------------|----------|-------------|
| Error-focused analysis/summary | 错误分析, 错误总结, 错误报告,报错分析,报错总结,报错报告, error analysis, error report, 排查, 诊断 | error_report |
| Overall analysis/summary | 整体分析, 整体总结, 总体分析, 全面分析, 汇总, overall, summary, 概况, 全局 | overall_report |
| Pcap analysis (triggered by Step 5b) | 用户选择抓包候选项后自动触发 | pcap_report |
Default: If the intent is ambiguous or does not clearly fall into either category, default to
error_report.
2b. Naming convention
Format (MUST follow):
{task_id}/{task_id}_{report_type}_{timestamp}.pdf
{task_id}— the task ID provided by the user (e.g.task-xxxxxxxx){report_type}— eithererror_reportoroverall_report(determined in 2a){timestamp}— current time formatted asYYYYMMDD_HHmmss(e.g.20260318_143025)
Examples:
- Error analysis for task
task-xxxxxxxxat 2026-03-18 14:30:25:task-xxxxxxxx/task-xxxxxxxx_error_report_20260318_143025.pdf - Overall summary for task
task-xxxxxxxxat 2026-03-18 14:30:25:task-xxxxxxxx/task-xxxxxxxx_overall_report_20260318_143025.pdf
Before executing the script, create the task directory if it does not exist:
mkdir -p {task_id}
Place the directory in the current working directory.
Step 3 — Execute
# With PDF output (when PDF rule is triggered):
python3 ${SKILL_DIR}/scripts/cat_network_quality_analysis.py \
--query-text "<QueryText>" \
--task-id "<TaskID>" \
--start-time <StartTime> \
--end-time <EndTime> \
--output <output.pdf>
# Without PDF output (simple query, no TaskID):
python3 ${SKILL_DIR}/scripts/cat_network_quality_analysis.py \
--query-text "<QueryText>" \
--start-time <StartTime> \
--end-time <EndTime>
Step 4 — Deliver results
If PDF was generated (PDF rule triggered):
- Send the PDF file — you MUST use the
messagetool to send the PDF file to the user. Do NOT use themediatool —mediais for images only and will fail or produce incorrect results for PDF files. Call themessagetool with the generated PDF path so the user receives the actual file in the conversation. Do NOT just print the path — you MUST deliver the file. - Provide a brief analysis summary — extract key findings (error count, error types, affected cities/ISPs, top issues) and present a concise summary.
- List actionable next steps — suggest prioritized actions based on the analysis.
- Capture the SessionID — parse the script's stdout for a line matching
SessionID=<value>. You MUST remember this SessionID for passing to the pcap analysis in Step 5b via--session-id.
If no PDF was generated:
- Reply with the analysis text directly in the conversation.
Step 5 — Pcap analysis follow-up (when pcap candidates exist)
After delivering results, check the script's stderr output (or the PDF report) for a "抓包分析候选列表" section. If it exists, you MUST perform the following interactive flow:
5a. Present the numbered candidate list
Extract each item from the pcap candidate table and present them as a numbered list to the user. Format:
📋 检测到以下抓包分析候选项,您可以选择其中一个进行深入抓包分析:
1. 城市: 慕尼黑 | 运营商: Germany_COLT | 错误码: 600 | 任务ID: task-xxxxxxxx | 拨测时间: 2026-03-17 07:06:35 | 拨测点: xxxxx
2. 城市: 柏林 | 运营商: Germany_DTAG | 错误码: 600 | 任务ID: task-xxxxxxxx | 拨测时间: 2026-03-17 07:05:22 | 拨测点: xxxxx
3. ...
请回复序号(如 `1`)选择要分析的项。
💡 如果您是在企微群中使用机器人,请记得 **@机器人** 来回复序号,否则机器人无法收到您的消息。
5b. Handle user selection
When the user replies with a number (e.g. 1, 2), map it back to the corresponding candidate item and re-invoke the script with:
--query-textset to a<structured_json>tagged payload containing the selected item's fields:--query-text "<structured_json>{\"task_id\":\"task-xxxxxxxx\",\"probe_time\":1773905195000,\"code\":\"xxxxx\"}</structured_json>"task_id— the candidate'stask_idprobe_time— the candidate'sprobe_time(original millisecond timestamp, NOT the formatted string)code— the candidate'scode(拨测点编码)
⚠️ CRITICAL: The
<structured_json>...</structured_json>tags are mandatory and must NOT be removed or altered. Only replace the values inside the JSON with the user's selected candidate data.
- Pass
--session-id— use the SessionID captured from the original invocation's stdout (Step 4). This maintains the conversation context for the AI Console. - Don't keep the same
--task-id,--start-time,--end-timeas the original invocation. - Generate a new
--outputpath following the naming convention (Step 2), using report typepcap_report.
Example full command:
python3 ${SKILL_DIR}/scripts/cat_network_quality_analysis.py \
--query-text "<structured_json>{\"task_id\":\"task-xxxxxxxx\",\"probe_time\":1773905195000,\"code\":\"xxxxx\"}</structured_json>" \
--session-id "17f57xxxxxxxxxxxxxxxx5b224055354" \
--output task-xxxxxxxx/task-xxxxxxxx_pcap_report_20260319_120000.pdf
Then follow the same execution mode (Phase 1 launch + Phase 2 poll) and deliver results (Step 4) workflow as before.
扫码联系在线客服