100 个 Agent Skills 盘点（八）：文档与文件处理

1
# Agent 解析 PDF 合同
2
import pdfplumber
3

4
def extract_contract_info(pdf_path):
5
    with pdfplumber.open(pdf_path) as pdf:
6
        full_text = ""
7
        tables = []
8

9
        for page in pdf.pages:
10
            # 提取文字
11
            full_text += page.extract_text() or ""
12

13
            # 提取表格
14
            page_tables = page.extract_tables()
15
            tables.extend(page_tables)
16

17
        # 用 LLM 分析合同内容
18
        analysis = llm.analyze(f"""
19
        请从以下合同文本中提取关键信息：
20
        1. 合同双方
21
        2. 合同期限
22
        3. 金额
23
        4. 关键条款
24

25
        文本：{full_text[:5000]}
26
        """)
27

28
        return {
29
            "text": full_text,
30
            "tables": tables,
31
            "analysis": analysis
32
        }

1
// Agent 提取 PDF 中的表格数据
2
const pdfParse = require('pdf-parse');
3
const fs = require('fs');
4

5
async function extractTablesFromPDF(pdfPath) {
6
  const buffer = fs.readFileSync(pdfPath);
7
  const data = await pdfParse(buffer);
8

9
  // 使用正则匹配表格结构
10
  const lines = data.text.split('\n');
11
  const tableData = [];
12
  let inTable = false;
13

14
  for (const line of lines) {
15
    if (line.includes('|') || line.match(/\s{3,}/)) {
16
      inTable = true;
17
      const cells = line.split(/\s{2,}|\|/).filter(Boolean).map(s => s.trim());
18
      tableData.push(cells);
19
    } else if (inTable) {
20
      inTable = false;
21
    }
22
  }
23

24
  return tableData;
25
}

1
# Agent 生成 Word 报告
2
from docx import Document
3
from docx.shared import Inches, Pt
4
from docx.enum.text import WD_ALIGN_PARAGRAPH
5

6
def generate_report(data, output_path):
7
    doc = Document()
8

9
    # 标题
10
    title = doc.add_heading('月度运营报告', 0)
11
    title.alignment = WD_ALIGN_PARAGRAPH.CENTER
12

13
    # 摘要
14
    doc.add_heading('执行摘要', level=1)
15
    doc.add_paragraph(data['summary'])
16

17
    # 数据表格
18
    doc.add_heading('详细数据', level=1)
19
    table = doc.add_table(rows=1, cols=4)
20
    table.style = 'Light Grid Accent 1'
21

22
    # 表头
23
    headers = ['指标', '本月', '上月', '变化']
24
    for i, header in enumerate(headers):
25
        table.rows[0].cells[i].text = header
26

27
    # 数据行
28
    for row_data in data['metrics']:
29
        row = table.add_row()
30
        for i, value in enumerate(row_data):
31
            row.cells[i].text = str(value)
32

33
    # 图表说明
34
    doc.add_heading('趋势分析', level=1)
35
    doc.add_paragraph(data['analysis'])
36

37
    doc.save(output_path)

1
// Agent 用模板生成合同
2
const docxtemplater = require('docxtemplater');
3
const fs = require('fs');
4

5
async function generateContract(templatePath, data) {
6
  const content = fs.readFileSync(templatePath, 'binary');
7
  const doc = new docxtemplater(content);
8

9
  doc.setData({
10
    partyA: data.partyA,
11
    partyB: data.partyB,
12
    amount: data.amount,
13
    startDate: data.startDate,
14
    endDate: data.endDate,
15
    terms: data.terms
16
  });
17

18
  doc.render();
19
  return doc.getZip().generate({ type: 'nodebuffer' });
20
}

1
# Agent 分析销售数据并生成报告
2
import openpyxl
3
from openpyxl.chart import BarChart, Reference
4
from openpyxl.styles import Font, PatternFill
5

6
def analyze_sales(excel_path):
7
    wb = openpyxl.load_workbook(excel_path)
8
    ws = wb.active
9

10
    # 读取数据
11
    sales_data = []
12
    for row in ws.iter_rows(min_row=2, values_only=True):
13
        sales_data.append({
14
            'product': row[0],
15
            'region': row[1],
16
            'amount': row[2],
17
            'quantity': row[3]
18
        })
19

20
    # 按产品汇总
21
    product_summary = {}
22
    for item in sales_data:
23
        product = item['product']
24
        product_summary[product] = product_summary.get(product, 0) + item['amount']
25

26
    # 创建报告工作簿
27
    report_wb = openpyxl.Workbook()
28
    report_ws = report_wb.active
29
    report_ws.title = "销售汇总"
30

31
    # 写入表头
32
    headers = ['产品', '销售额', '占比']
33
    for col, header in enumerate(headers, 1):
34
        cell = report_ws.cell(row=1, column=col, value=header)
35
        cell.font = Font(bold=True, color="FFFFFF")
36
        cell.fill = PatternFill("solid", fgColor="2563EB")
37

38
    # 写入数据
39
    total = sum(product_summary.values())
40
    for row, (product, amount) in enumerate(sorted(product_summary.items(), key=lambda x: -x[1]), 2):
41
        report_ws.cell(row=row, column=1, value=product)
42
        report_ws.cell(row=row, column=2, value=amount)
43
        report_ws.cell(row=row, column=3, value=f"{amount/total*100:.1f}%")
44

45
    # 添加图表
46
    chart = BarChart()
47
    chart.title = "产品销售额对比"
48
    data = Reference(report_ws, min_col=2, min_row=1, max_row=len(product_summary)+1)
49
    categories = Reference(report_ws, min_col=1, min_row=2, max_row=len(product_summary)+1)
50
    chart.add_data(data, titles_from_data=True)
51
    chart.set_categories(categories)
52
    report_ws.add_chart(chart, "E2")
53

54
    report_wb.save("sales_report.xlsx")

1
# Agent 处理大型 CSV 文件
2
import pandas as pd
3

4
def process_large_csv(input_path, output_path):
5
    # 分块读取大文件
6
    chunks = pd.read_csv(input_path, chunksize=10000, encoding='utf-8')
7

8
    results = []
9
    for chunk in chunks:
10
        # 数据清洗
11
        chunk = chunk.dropna(subset=['email'])  # 删除无邮箱的行
12
        chunk['email'] = chunk['email'].str.lower().str.strip()
13
        chunk['amount'] = pd.to_numeric(chunk['amount'], errors='coerce')
14

15
        # 数据转换
16
        chunk['date'] = pd.to_datetime(chunk['date'])
17
        chunk['month'] = chunk['date'].dt.to_period('M')
18

19
        results.append(chunk)
20

21
    # 合并所有数据
22
    df = pd.concat(results)
23

24
    # 分析
25
    summary = df.groupby('month').agg(
26
        total_amount=('amount', 'sum'),
27
        count=('email', 'count'),
28
        unique_users=('email', 'nunique')
29
    ).reset_index()
30

31
    summary.to_csv(output_path, index=False)
32
    return summary

1
// Agent 自动生成文档目录
2
const fs = require('fs');
3

4
function generateTOC(markdownContent) {
5
  const headings = [];
6
  const lines = markdownContent.split('\n');
7

8
  for (const line of lines) {
9
    const match = line.match(/^(#{1,6})\s+(.+)/);
10
    if (match) {
11
      const level = match[1].length;
12
      const title = match[2].trim();
13
      const slug = title
14
        .toLowerCase()
15
        .replace(/[^\w一-龥]+/g, '-')
16
        .replace(/^-|-$/g, '');
17

18
      headings.push({ level, title, slug });
19
    }
20
  }
21

22
  // 生成目录
23
  let toc = '## 目录\n\n';
24
  for (const heading of headings) {
25
    const indent = '  '.repeat(heading.level - 1);
26
    toc += `${indent}- [${heading.title}](#${heading.slug})\n`;
27
  }
28

29
  return toc;
30
}
31

32
// Agent 批量处理 Markdown 文件
33
const files = fs.readdirSync('./docs').filter(f => f.endsWith('.md'));
34
for (const file of files) {
35
  const content = fs.readFileSync(`./docs/${file}`, 'utf-8');
36
  const toc = generateTOC(content);
37
  // 在第一个标题后插入目录
38
  const updated = content.replace(/(^# .+$)/m, `$1\n\n${toc}`);
39
  fs.writeFileSync(`./docs/${file}`, updated);
40
}

1
// Agent 批量处理产品图片
2
const sharp = require('sharp');
3
const fs = require('fs');
4
const path = require('path');
5

6
async function processProductImages(inputDir, outputDir) {
7
  const files = fs.readdirSync(inputDir).filter(f =>
8
    /\.(jpg|jpeg|png|webp)$/i.test(f)
9
  );
10

11
  for (const file of files) {
12
    const inputPath = path.join(inputDir, file);
13
    const name = path.parse(file).name;
14

15
    // 生成多种尺寸
16
    const sizes = [
17
      { width: 1200, suffix: 'large' },   // 产品详情页
18
      { width: 600, suffix: 'medium' },    // 列表页
19
      { width: 300, suffix: 'thumbnail' }, // 缩略图
20
      { width: 100, suffix: 'icon' }       // 图标
21
    ];
22

23
    for (const size of sizes) {
24
      await sharp(inputPath)
25
        .resize(size.width, null, { withoutEnlargement: true })
26
        .webp({ quality: 80 })
27
        .toFile(path.join(outputDir, `${name}-${size.suffix}.webp`));
28
    }
29

30
    console.log(`✅ Processed: ${file}`);
31
  }
32
}

1
# Agent 用 Whisper 做会议录音转文字
2
import whisper
3

4
def transcribe_meeting(audio_path):
5
    model = whisper.load_model("medium")
6
    result = model.transcribe(audio_path, language="zh")
7

8
    # 生成带时间戳的文字稿
9
    transcript = ""
10
    for segment in result["segments"]:
11
        start = format_time(segment["start"])
12
        end = format_time(segment["end"])
13
        text = segment["text"]
14
        transcript += f"[{start} → {end}] {text}\n"
15

16
    # 用 LLM 生成会议纪要
17
    summary = llm.generate(f"""
18
    根据以下会议录音文字稿，生成会议纪要：
19
    1. 会议主题
20
    2. 讨论要点（3-5 个）
21
    3. 决议事项
22
    4. 待办事项及负责人
23

24
    文字稿：{transcript}
25
    """)
26

27
    return {"transcript": transcript, "summary": summary}

1
// Agent 批量格式转换
2
const fs = require('fs');
3
const yaml = require('js-yaml');
4
const csvtojson = require('csvtojson');
5

6
async function convertFormat(inputPath, fromFormat, toFormat) {
7
  let data;
8

9
  // 读取
10
  switch (fromFormat) {
11
    case 'json':
12
      data = JSON.parse(fs.readFileSync(inputPath, 'utf-8'));
13
      break;
14
    case 'yaml':
15
      data = yaml.load(fs.readFileSync(inputPath, 'utf-8'));
16
      break;
17
    case 'csv':
18
      data = await csvtojson().fromFile(inputPath);
19
      break;
20
  }
21

22
  // 写入
23
  const outputPath = inputPath.replace(`.${fromFormat}`, `.${toFormat}`);
24
  switch (toFormat) {
25
    case 'json':
26
      fs.writeFileSync(outputPath, JSON.stringify(data, null, 2));
27
      break;
28
    case 'yaml':
29
      fs.writeFileSync(outputPath, yaml.dump(data));
30
      break;
31
    case 'csv':
32
      const headers = Object.keys(data[0]).join(',');
33
      const rows = data.map(row => Object.values(row).join(','));
34
      fs.writeFileSync(outputPath, [headers, ...rows].join('\n'));
35
      break;
36
  }
37

38
  console.log(`✅ Converted: ${inputPath} → ${outputPath}`);
39
}

1
// Agent 通过 Sentry MCP Server 自动分析错误并创建 Issue
2
// 1. 查询最近 1 小时的错误事件
3
const recentErrors = await mcp.sentry.queryErrors({
4
  project: 'my-web-app',
5
  timeRange: '1h',
6
  sort: 'frequency',
7
  limit: 10,
8
});
9

10
// 2. 按错误类型聚合，找出最常见的问题
11
const topIssues = aggregateByType(recentErrors);
12

13
for (const issue of topIssues.slice(0, 3)) {
14
  // 3. 分析错误堆栈，定位根因
15
  const analysis = await llm.analyze({
16
    prompt: `分析以下错误堆栈，找出根本原因并给出修复建议：
17
      错误类型：${issue.title}
18
      堆栈信息：${issue.stacktrace}
19
      发生次数：${issue.count}
20
      首次出现：${issue.firstSeen}`,
21
  });
22

23
  // 4. 自动创建 GitHub Issue 并关联 Sentry
24
  await mcp.github.createIssue({
25
    repo: 'my-org/my-web-app',
26
    title: `[Sentry] ${issue.title}`,
27
    body: `## 错误分析\n\n${analysis.summary}\n\n` +
28
          `## 堆栈信息\n\n\`\`\`\n${issue.stacktrace}\n\`\`\`\n\n` +
29
          `## 修复建议\n\n${analysis.suggestion}\n\n` +
30
          `> Sentry Issue: ${issue.permalink}`,
31
    labels: ['bug', 'sentry', 'auto-created'],
32
    assignees: [analysis.suggestedAssignee],
33
  });
34
}
35

36
console.log(`✅ 已分析 ${recentErrors.length} 个错误，创建了 ${topIssues.length} 个 Issue`);

1
# Agent 自动生成 API 文档
2
import ast
3
import json
4

5
def generate_api_docs(source_code):
6
    """从 Python 代码自动生成 API 文档"""
7
    tree = ast.parse(source_code)
8
    endpoints = []
9

10
    for node in ast.walk(tree):
11
        if isinstance(node, ast.FunctionDef):
12
            # 检查是否有路由装饰器
13
            for decorator in node.decorator_list:
14
                if isinstance(decorator, ast.Call):
15
                    if hasattr(decorator.func, 'attr'):
16
                        method = decorator.func.attr  # get, post, put, delete
17
                        if method in ('get', 'post', 'put', 'delete'):
18
                            # 提取路径
19
                            path = decorator.args[0].value if decorator.args else '/'
20

21
                            # 提取文档字符串
22
                            docstring = ast.get_docstring(node) or ''
23

24
                            # 提取参数
25
                            params = []
26
                            for arg in node.args.args:
27
                                params.append(arg.arg)
28

29
                            endpoints.append({
30
                                'method': method.upper(),
31
                                'path': path,
32
                                'function': node.name,
33
                                'params': params,
34
                                'description': docstring
35
                            })
36

37
    # 生成 Markdown 文档
38
    doc = "# API Documentation\n\n"
39
    for ep in endpoints:
40
        doc += f"## {ep['method']} {ep['path']}\n\n"
41
        doc += f"**Function:** `{ep['function']}`\n\n"
42
        doc += f"**Parameters:** {', '.join(ep['params'])}\n\n"
43
        doc += f"{ep['description']}\n\n---\n\n"
44

45
    return doc