- name
- system-data-intelligence
- description
- >
- IMPORTANT
- 只要涉及上述任一场景，必须使用此 skill，不得因"任务简单"而跳过。
System Data Intelligence Skill
🔍 快速决策树
收到任务，先走此树定路径，再执行对应节点的代码：
用户任务
├─ 有文件需要读取？
│ ├─ Windows → [WIN-PATH]
│ ├─ macOS → [MAC-PATH]
│ └─ Linux → [LINUX-PATH]
│
├─ 需要查数据库？ → [DB-PATH]
├─ 需要调 API？ → [API-PATH]
├─ 含敏感数据？ → [SECURITY-PATH] ← 先执行，再做其他
├─ 需要分析数据？ → [ANALYSIS-PATH]
└─ 需要出图表？ → [VIZ-PATH]组合任务执行顺序（最常见）：
敏感检测 → 读文件/拉数据 → 分析 → 可视化 → 输出报告[WIN-PATH] Windows 文件读取
自动降级链： win32com → openpyxl（Excel） / python-docx（Word）
# 统一入口，自动选策略
from scripts.doc_parser import detect_and_load
data = detect_and_load("report.xlsx") # 自动 win32com → openpyxl 降级
# 或指定平台脚本
# python scripts/win_excel_reader.py report.xlsx [sheet_name]
# python scripts/wps_extractor.py data.etWPS 专属 COM 名称：
| 应用 | COM 名称 |
|---|---|
| WPS 表格（.et） | KET.Application |
| WPS 文字（.wps） | KWPS.Application |
注意： COM 操作结束后必须调用 Quit()，已封装到 safe_com_session 上下文管理器中。
完整手册 → references/windows-api.md
[MAC-PATH] macOS 文件读取
自动降级链： xlwings → openpyxl（Excel） / python-docx（Word）
# 统一入口，自动选策略
from scripts.doc_parser import detect_and_load
data = detect_and_load("report.xlsx") # 自动 xlwings → openpyxl 降级
# 或指定平台脚本
# python scripts/mac_excel_reader.py report.xlsx [sheet_name]权限问题（首次运行必看）：
系统设置 → 隐私与安全性 → 辅助功能 → 开启"终端"或"Python"权限
系统设置 → 隐私与安全性 → 自动化 → 允许控制"Microsoft Excel"完整手册 → references/macos-api.md
[LINUX-PATH] Linux 文件读取
策略： 无 COM/AppleScript，全部用纯 Python 库解析，老格式先用 LibreOffice 转换。
# 统一入口
from scripts.doc_parser import detect_and_load
data = detect_and_load("report.xlsx") # 直接 openpyxl
# 老格式先转换
# libreoffice --headless --convert-to xlsx input.xls --outdir /tmp/
# libreoffice --headless --convert-to docx input.doc --outdir /tmp/无头服务器可视化（必须设置）：
import matplotlib
matplotlib.use('Agg') # 放在所有 import matplotlib.pyplot 之前中文字体安装：
sudo apt install fonts-noto-cjk && fc-cache -fv完整手册 → references/linux-api.md
[DB-PATH] 数据库查询
from scripts.db_connector import DBConnector
db = DBConnector("mysql+pymysql://user:pass@host:3306/mydb")
# 普通查询 → DataFrame
df = db.query("SELECT * FROM orders WHERE status = :s", params={"s": "paid"})
# 大表分块（避免 OOM）
for chunk in db.query_chunked("SELECT * FROM logs", chunk_size=10000):
process(chunk)
db.close()连接 URL 速查：
| 数据库 | URL 格式 |
|---|---|
| SQLite | sqlite:///data.db |
| MySQL | mysql+pymysql://user:pass@host:3306/db |
| PostgreSQL | postgresql+psycopg2://user:pass@host:5432/db |
| SQL Server | mssql+pymssql://user:pass@host:1433/db |
CLI 用法：
python scripts/db_connector.py "sqlite:///data.db" "SELECT * FROM users" -o result.csv[API-PATH] REST API 数据接入
from scripts.api_loader import APILoader
api = APILoader(base_url="https://api.example.com", timeout=30, max_retries=3)
api.set_auth_token("your_token") # 或 api.set_api_key("key")
# 单次请求
df = api.fetch_json_to_df("/users", data_key="results")
# 分页自动合并
df = api.fetch_paginated("/items", page_param="page", per_page=100, data_key="data")
# CSV 接口
df = api.fetch_csv_to_df("https://data.example.com/export.csv")
api.close()CLI 用法：
python scripts/api_loader.py "https://api.example.com/data" \
--data-key results --output data.csv重试策略： 默认 3 次，指数退避（1s → 2s → 4s），网络超时、5xx 均触发重试。
[SECURITY-PATH] 敏感数据处理
原则：检测到敏感数据，先脱敏再分析，任务结束后清理临时文件。
from scripts.security_utils import secure_context
with secure_context() as ctx:
df = load_data(filepath)
# Step 1: 扫描（返回 {列名: [命中规则]}）
findings = ctx.masker.detect_sensitive(df)
# Step 2: 脱敏（仅处理文本列，不破坏结构）
df_safe = ctx.masker.mask_dataframe(df)
# Step 3: 安全临时目录（退出自动删除）
tmp = ctx.cleanup.create_temp_dir()
df_safe.to_csv(tmp / "safe_data.csv", index=False)
# with 块退出后：临时文件自动删除，内存自动 gc脱敏效果：
| 规则 | 原始 | 脱敏后 |
|---|---|---|
| 手机号 | 13812345678 | 138****5678 |
| 身份证 | 110101199001011234 | 110101********1234 |
| 邮箱 | test@example.com | t***@example.com |
| 银行卡 | 6222021234567890 | 6222 **** **** 7890 |
| IP | 192.168.1.100 | 192.168.*.* |
[ANALYSIS-PATH] 数据深度分析
分析层次：
Level 1: 描述性 → 数据现状（均值 / 中位数 / 缺失率 / 分布）
Level 2: 诊断性 → 根因探查（相关性 / 异常检测 / 重复行）
Level 3: 预测性 → 趋势建模（滚动均值 / 环比同比 / 季节性）
Level 4: 规范性 → 决策支持（优化建议 / 洞察摘要）执行：
from scripts.deep_analyzer import DeepAnalyzer
import pandas as pd
df = pd.DataFrame(data['sheets']['Sheet1']) # 来自 detect_and_load
report = DeepAnalyzer(df).run_full_analysis()
# report 结构：
# {
# 'overview': { shape, columns, dtypes, memory_mb }
# 'quality': { missing_rate, duplicate_rows, completeness }
# 'distributions': { col: { mean, median, std, min, max, q25, q75, skewness } }
# 'correlations': { full_matrix, strong_correlations }
# 'anomalies': { col: { outlier_count, outlier_pct } }
# 'insights': [ "...", "..." ] ← 文字洞察列表
# }时间序列分析（可选）：
from scripts.deep_analyzer import analyze_time_series
ts = analyze_time_series(df, date_col="date", value_col="revenue")
# 返回：period / trend_7d / trend_30d / mom_latest / yoy_latest / seasonality_strengthCLI 用法：
python scripts/deep_analyzer.py sales.xlsx date revenue
# 输出：outputs/analysis_result.json + outputs/summary.md分析参考 → references/viz-patterns.md
[VIZ-PATH] 数据可视化
图表选型：
数据特征
├─ 时间序列 → 折线图 / 面积图 → create_dashboard()
├─ 类别对比 → 柱状图 / 雷达图 → create_radar()
├─ 层级占比 → 树状图 / 旭日图 → create_treemap()
├─ 流量转化 → 桑基图 / 漏斗图 → create_sankey() / create_funnel()
├─ 相关分布 → 散点图 / 热力图 → create_dashboard()
├─ 节点关系 → 网络图 → create_network_graph()
└─ 地理分布 → 散点地图 / 填色地图 → create_geo_map() / create_choropleth()执行：
from scripts.viz_engine import (
create_dashboard, create_network_graph, create_geo_map,
create_choropleth, create_sankey, create_radar,
create_treemap, create_funnel, export_viz,
)
# 综合仪表盘（接受 DeepAnalyzer 的 report）
fig = create_dashboard(report, title="销售数据仪表盘")
export_viz(fig, "outputs/report", formats=["html", "png"])
# 网络关系图
fig = create_network_graph(
edges=[("总部", "上海"), ("总部", "北京"), ("上海", "客户A")],
title="业务关系图",
)
# 地理散点图
fig = create_geo_map(df, lat_col="lat", lon_col="lon",
value_col="sales", label_col="city")CLI 用法：
python scripts/viz_engine.py outputs/analysis_result.json outputs/
# 生成：outputs/report_YYYYMMDD_HHMMSS.html + .png图表样式规范 → references/viz-patterns.md
⚡ 统一文件入口
不知道用哪个脚本？永远从这里开始：
from scripts.doc_parser import detect_and_load
data = detect_and_load("/path/to/any/file", chunk_size=10000, timeout=120)
# 自动做：OS 检测 → 策略选择 → 失败降级 → 大文件分块 → 返回标准化 dict
# data 结构：
# {
# 'type': 'spreadsheet' | 'document' | 'text' | 'json' | 'project'
# 'sheets': { sheet_name: [[row], [row], ...] } # 表格类
# 'content': "..." # 文本类
# '_file': "filename.xlsx"
# '_os': 'Darwin' | 'Windows' | 'Linux'
# '_strategy': 'openpyxl' | 'win32com' | 'xlwings' # 实际使用策略
# }支持格式： .xlsx .xls .xlsm .et .docx .doc .wps .txt .md .rtz .csv .json
📤 输出规范
每次任务完成，必须输出以下四项：
| 输出 | 路径 | 说明 |
|---|---|---|
| 交互报告 | outputs/report_YYYYMMDD_HHMMSS.html | Plotly，可在浏览器操作 |
| 静态图 | outputs/report_YYYYMMDD_HHMMSS.png | 高清，适合嵌入文档 |
| 结构化数据 | outputs/analysis_result.json | 供后续脚本使用 |
| 洞察摘要 | outputs/summary.md | ≤5 行关键结论 |
💡 执行心法
直接行动，不问格式。 用户说"分析一下"，就给他完整的数据故事：解析→分析→可视化→报告，一气呵成。 遇到错误不停止。 首选策略失败 → 自动降级 → 记录日志 → 继续任务。告知用户"使用了备选方案"即可。 安全优先。 发现手机号、身份证等敏感字段 → 先脱敏 → 再分析 → 临时文件用 secure_context 自动清理。