Codex 生产实战：10 个真实场景案例

{
  "task": "运维巡检",
  "schedule": "0 9 * * *",
  "model": "deepseek-chat",
  "skills": ["ssh-tools", "report-generator"]
}

# Codex 执行的巡检脚本
import subprocess
import json
from datetime import datetime

servers = json.load(open("servers.json"))
report = []

for server in servers:
    # SSH 执行远程命令
    cpu = subprocess.check_output(
        f"ssh {server['user']}@{server['host']} 'top -bn1 | grep Cpu'",
        shell=True
    ).decode()
    
    disk = subprocess.check_output(
        f"ssh {server['user']}@{server['host']} 'df -h /'",
        shell=True
    ).decode()
    
    services = subprocess.check_output(
        f"ssh {server['user']}@{server['host']} 'systemctl is-active nginx postgresql redis'",
        shell=True
    ).decode()
    
    report.append({
        "server": server['name'],
        "cpu": parse_cpu(cpu),
        "disk": parse_disk(disk),
        "services": services.strip().split('\n'),
        "timestamp": datetime.now().isoformat()
    })

# 生成报告
generate_report(report, format="markdown", output="daily_check.md")
# 发送邮件
send_email("ops@company.com", "每日巡检报告", "daily_check.md")

用户: "生成本周市场数据周报"

Codex 执行:
1. [API] 调用 Google Analytics API，获取本周流量数据
2. [SQL] 查询内部数据库，获取转化数据
3. [API] 调用 CRM API，获取线索数据
4. [Python] 合并数据，计算关键指标:
   - 流量: 125,430 访问 (+12.3% WoW)
   - 转化率: 3.2% (+0.5pp)
   - 线索数: 456 (+8.7%)
   - ROI: 4.2x
5. [Excel] 生成图表
6. [Markdown] 生成分析报告，含趋势图和数据表
7. [邮件] 发送给 marketing@company.com

# Codex 生成的数据合并逻辑
import pandas as pd

# 1. GA 数据
ga_data = fetch_ga_metrics(
    view_id="123456789",
    metrics=["sessions", "users", "pageviews"],
    date_range="last_7_days"
)

# 2. 数据库数据
db_data = query_db("""
    SELECT DATE(created_at) as date,
           COUNT(*) as conversions,
           SUM(amount) as revenue
    FROM orders
    WHERE created_at >= NOW() - INTERVAL 7 DAY
    GROUP BY DATE(created_at)
""")

# 3. CRM 数据
crm_data = call_crm_api("/leads", params={
    "created_after": "2026-06-18",
    "status": "qualified"
})

# 4. 合并分析
merged = ga_data.merge(db_data, on="date").merge(crm_data, on="date")
merged["conversion_rate"] = merged["conversions"] / merged["sessions"]
merged["roi"] = merged["revenue"] / merged["leads_cost"]

# 5. 生成报告
report = f"""
# 市场周报 ({merged['date'].min()} ~ {merged['date'].max()})

## 关键指标
| 指标 | 本周 | 上周 | 变化 |
|------|------|------|------|
| 总访问量 | {merged['sessions'].sum():,} | ... | {change}% |
| 转化率 | {merged['conversion_rate'].mean():.1%} | ... | ... |
| 收入 | ¥{merged['revenue'].sum():,} | ... | ... |
| ROI | {merged['roi'].mean():.1f}x | ... | ... |

## 趋势分析
{generate_trend_chart(merged)}

## 异常检测
{detect_anomalies(merged)}
"""

# .github/workflows/codex-review.yml
name: Codex Code Review
on:
  pull_request:
    types: [opened, updated]

jobs:
  review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Codex
        run: npm install -g @openai/codex
      - name: Run Review
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          codex --non-interactive "
            审查 PR #${{ github.event.number }}:
            1. 检查代码规范
            2. 检查潜在 bug
            3. 检查安全问题
            4. 检查性能问题
            5. 给出改进建议
            输出 Markdown 格式评论
          " --files "$(git diff --name-only origin/main...HEAD)" \
            > review_comment.md
      - name: Post Review
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            const comment = fs.readFileSync('review_comment.md', 'utf8');
            github.rest.issues.createComment({
              ...context.repo,
              issue_number: context.issue.number,
              body: comment
            });

## Codex 代码审查报告

### ✅ 通过项
- 代码规范符合 ESLint 配置
- TypeScript 类型标注完整
- 单元测试覆盖率 > 80%

### ⚠️ 警告
- `src/auth.ts:45`: `any` 类型应替换为具体类型
- `src/api/users.ts:78`: SQL 查询未使用参数化，存在注入风险

### 🔴 严重问题
- `src/utils/crypto.ts:12`: 使用了不安全的 MD5 哈希，应改为 bcrypt

### 💡 改进建议
- `src/services/userService.ts`: N+1 查询问题，建议使用 JOIN
- `src/middleware/auth.ts`: 缺少 rate limiting

用户: "扫描 src/routes/ 目录，从代码注释生成完整的 API 文档"

Codex 执行:
1. 扫描所有路由文件，识别 40 个端点
2. 解析 JSDoc 注释和 TypeScript 类型
3. 生成 OpenAPI 3.0 spec
4. 渲染为 HTML 文档 (Swagger UI)
5. 部署到 GitHub Pages

# Codex 生成的文档提取逻辑
import re
from pathlib import Path

def extract_api_docs(routes_dir: str) -> list:
    docs = []
    for file in Path(routes_dir).glob("**/*.ts"):
        content = file.read_text()
        
        # 提取路由定义
        routes = re.findall(
            r'@(?:Get|Post|Put|Delete|Patch)\([\'"](.+?)[\'"]\)',
            content
        )
        
        # 提取 JSDoc 注释
        jsdoc_blocks = re.findall(
            r'/\*\*(.*?)\*/',
            content,
            re.DOTALL
        )
        
        # 提取 TypeScript 接口
        interfaces = re.findall(
            r'interface (\w+) \{(.*?)\}',
            content,
            re.DOTALL
        )
        
        docs.append({
            "file": str(file),
            "routes": routes,
            "docs": jsdoc_blocks,
            "types": interfaces
        })
    
    return docs

# 生成 OpenAPI spec
def to_openapi(docs: list) -> dict:
    spec = {
        "openapi": "3.0.0",
        "info": {"title": "API Documentation", "version": "1.0.0"},
        "paths": {}
    }
    
    for doc in docs:
        for route in doc["routes"]:
            method = route["method"].lower()
            path = route["path"]
            spec["paths"][path] = {
                method: {
                    "summary": route.get("summary", ""),
                    "parameters": route.get("params", []),
                    "requestBody": route.get("body", {}),
                    "responses": route.get("responses", {})
                }
            }
    
    return spec

用户: "为 src/ 目录下所有未测试的文件生成单元测试，目标覆盖率 80%"

Codex 执行:
1. 扫描 src/，对比 tests/，找出 52 个未测试文件
2. 为每个文件生成测试:
   a. 分析导出的函数和类
   b. 生成正常用例
   c. 生成边界用例
   d. 生成异常用例
3. 运行测试，修复失败的用例
4. 生成覆盖率报告

// Codex 为 src/utils/format.ts 生成的测试
import { describe, test, expect } from 'vitest';
import { formatDate, formatCurrency, truncateText } from '../src/utils/format';

describe('formatDate', () => {
  test('正常日期', () => {
    expect(formatDate('2026-06-25')).toBe('2026年6月25日');
  });
  
  test('边界: 空字符串', () => {
    expect(formatDate('')).toBe('');
  });
  
  test('边界: null', () => {
    expect(formatDate(null as any)).toBe('');
  });
  
  test('异常: 无效日期', () => {
    expect(() => formatDate('invalid')).toThrow('Invalid date');
  });
  
  test('边界: 闰年', () => {
    expect(formatDate('2024-02-29')).toBe('2024年2月29日');
  });
});

describe('formatCurrency', () => {
  test('正常金额', () => {
    expect(formatCurrency(1234.56)).toBe('¥1,234.56');
  });
  
  test('边界: 零', () => {
    expect(formatCurrency(0)).toBe('¥0.00');
  });
  
  test('边界: 负数', () => {
    expect(formatCurrency(-100)).toBe('-¥100.00');
  });
  
  test('边界: 大数', () => {
    expect(formatCurrency(9999999999.99)).toBe('¥9,999,999,999.99');
  });
});

用户: "处理 contracts/ 目录下的所有 PDF，提取关键信息到 Excel"

Codex 执行:
1. 读取 500 个 PDF 文件
2. 用 NLP 提取:
   - 甲方名称
   - 乙方名称
   - 合同金额
   - 签署日期
   - 合同期限
3. 写入 contracts_summary.xlsx
4. 标记无法提取的文件

import pdfplumber
import re
import openpyxl

def extract_contract_info(pdf_path: str) -> dict:
    with pdfplumber.open(pdf_path) as pdf:
        text = ""
        for page in pdf.pages[:3]:  # 通常前3页包含关键信息
            text += page.extract_text() or ""
    
    info = {
        "文件名": Path(pdf_path).name,
        "甲方": extract_party(text, "甲方"),
        "乙方": extract_party(text, "乙方"),
        "金额": extract_amount(text),
        "签署日期": extract_date(text),
        "期限": extract_term(text),
        "提取状态": "成功"
    }
    
    if not info["甲方"] or not info["乙方"]:
        info["提取状态"] = "需人工检查"
    
    return info

def extract_amount(text: str) -> str:
    patterns = [
        r'合同金额[：:]\s*人民币\s*([\d,]+\.?\d*)\s*元',
        r'总金额[：:]\s*([\d,]+\.?\d*)\s*[元￥]',
        r'金额[：:]\s*¥\s*([\d,]+\.?\d*)'
    ]
    for pattern in patterns:
        match = re.search(pattern, text)
        if match:
            return f"¥{match.group(1)}"
    return "未提取到"

用户: "爬取以下 5 个竞品网站的产品定价，与上周数据对比，生成变化报告"

Codex 执行:
1. [浏览器] 逐个打开竞品网站定价页面
2. [数据提取] 提取产品名称、价格、功能对比
3. [数据对比] 读取上周数据，计算变化
4. [报告生成] 生成 Markdown 报告 + Excel 明细
5. [通知] 发送到产品团队 Slack 频道

from playwright.sync_api import sync_playwright
import pandas as pd
import json

competitors = json.load(open("competitors.json"))

def scrape_pricing():
    all_data = []
    
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        
        for comp in competitors:
            page = browser.new_page()
            page.goto(comp["pricing_url"])
            page.wait_for_selector(comp["price_selector"])
            
            products = page.query_selector_all(comp["product_selector"])
            
            for product in products:
                name = product.query_selector(comp["name_selector"]).text_content()
                price = product.query_selector(comp["price_selector"]).text_content()
                
                all_data.append({
                    "competitor": comp["name"],
                    "product": name,
                    "price": parse_price(price),
                    "url": comp["pricing_url"],
                    "date": datetime.now().strftime("%Y-%m-%d")
                })
            
            page.close()
        
        browser.close()
    
    return pd.DataFrame(all_data)

# 对比上周数据
this_week = scrape_pricing()
last_week = pd.read_csv("last_week_pricing.csv")

merged = this_week.merge(last_week, on=["competitor", "product"], suffixes=("_new", "_old"))
merged["price_change"] = merged["price_new"] - merged["price_old"]
merged["change_pct"] = (merged["price_change"] / merged["price_old"] * 100).round(2)

# 生成报告
changes = merged[merged["price_change"] != 0]
print(f"发现 {len(changes)} 个价格变动")

用户: "对接支付宝支付 API、顺丰物流 API、阿里云短信 API，包含完整测试"

Codex 执行:
1. 读取 3 个 API 文档（通过 URL）
2. 生成 API 客户端代码
3. 实现错误处理和重试逻辑
4. 编写集成测试
5. 生成使用文档

# Codex 生成的支付 API 客户端
import httpx
import hashlib
import time
from typing import Optional
from dataclasses import dataclass

@dataclass
class PaymentResult:
    success: bool
    order_id: str
    payment_url: Optional[str]
    error: Optional[str]

class AlipayClient:
    def __init__(self, app_id: str, private_key: str, sandbox: bool = False):
        self.app_id = app_id
        self.private_key = private_key
        self.base_url = (
            "https://openapi-sandbox.dl.alipaydev.com/gateway.do"
            if sandbox else "https://openapi.alipay.com/gateway.do"
        )
        self.client = httpx.AsyncClient(timeout=30)
    
    async def create_payment(
        self,
        order_id: str,
        amount: float,
        subject: str,
        notify_url: str,
        return_url: str
    ) -> PaymentResult:
        params = {
            "app_id": self.app_id,
            "method": "alipay.trade.page.pay",
            "charset": "utf-8",
            "sign_type": "RSA2",
            "timestamp": time.strftime("%Y-%m-%d %H:%M:%S"),
            "version": "1.0",
            "biz_content": json.dumps({
                "out_trade_no": order_id,
                "total_amount": f"{amount:.2f}",
                "subject": subject,
                "product_code": "FAST_INSTANT_TRADE_PAY"
            }),
            "notify_url": notify_url,
            "return_url": return_url
        }
        
        params["sign"] = self._sign(params)
        
        for attempt in range(3):
            try:
                resp = await self.client.post(self.base_url, data=params)
                result = resp.json()
                
                if result.get("alipay_trade_page_pay_response", {}).get("code") == "10000":
                    return PaymentResult(
                        success=True,
                        order_id=order_id,
                        payment_url=result["alipay_trade_page_pay_response"]["qr_code"]
                    )
                else:
                    return PaymentResult(
                        success=False,
                        order_id=order_id,
                        payment_url=None,
                        error=result.get("msg", "Unknown error")
                    )
            except httpx.TimeoutException:
                if attempt == 2:
                    return PaymentResult(False, order_id, None, "API timeout")
                await asyncio.sleep(2 ** attempt)
    
    def _sign(self, params: dict) -> str:
        sorted_params = sorted(params.items())
        sign_string = "&".join(f"{k}={v}" for k, v in sorted_params if v)
        signature = hashlib.sha256(sign_string.encode()).hexdigest()
        return signature

# .github/workflows/codex-pipeline.yml
name: Codex CI/CD Pipeline
on:
  push:
    branches: [develop]
  pull_request:
    branches: [main]

jobs:
  codex-review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Install Codex
        run: npm install -g @openai/codex
      
      - name: Code Review
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          codex --non-interactive "
            审查本次提交的代码变更:
            1. 安全漏洞检查
            2. 性能问题检查
            3. 代码规范检查
            如发现问题，直接修复并提交修复 commit
          " --files "$(git diff --name-only HEAD~1)"
      
      - name: Auto-fix
        if: failure()
        run: |
          codex --non-interactive "
            上一步发现的问题，请自动修复
          " --auto-commit
      
      - name: Run Tests
        run: |
          codex --non-interactive "
            运行测试，如果失败请分析原因并修复
          " --command "npm test"
      
      - name: Generate Changelog
        run: |
          codex --non-interactive "
            根据 git log 生成本次变更的 changelog
          " --output CHANGELOG_ENTRY.md

用户工单 → Codex 分析 → 分类:
  → [FAQ类] 自动回复 (60%)
  → [Bug类] 创建 GitHub Issue + 回复 (20%)
  → [复杂问题] 转人工 + 提供上下文 (20%)

# Codex 工单处理流程
def process_ticket(ticket: dict) -> dict:
    # 1. 分析工单意图
    analysis = codex.analyze(f"""
    分析以下客服工单:
    标题: {ticket['subject']}
    内容: {ticket['body']}
    
    输出 JSON:
    - category: faq | bug | feature_request | billing | other
    - urgency: low | medium | high | critical
    - confidence: 0-1
    - suggested_response: 如果是 FAQ 类，给出回复内容
    - related_docs: 相关文档链接列表
    """)
    
    # 2. 按类别处理
    if analysis["category"] == "faq" and analysis["confidence"] > 0.85:
        # 自动回复
        send_ticket_reply(ticket["id"], analysis["suggested_response"])
        close_ticket(ticket["id"], "auto_resolved")
        return {"status": "auto_resolved"}
    
    elif analysis["category"] == "bug":
        # 创建 GitHub Issue
        issue = create_github_issue(
            title=f"[Customer] {ticket['subject']}",
            body=f"""
## 客户工单
{ticket['body']}

## 客户信息
- 邮箱: {ticket['email']}
- 紧急度: {analysis['urgency']}

## 环境信息
{ticket.get('metadata', {})}
""",
            labels=["bug", f"priority:{analysis['urgency']}"]
        )
        reply = f"感谢反馈，已创建 Issue #{issue['number']}，我们会尽快处理。"
        send_ticket_reply(ticket["id"], reply)
        return {"status": "issue_created", "issue_id": issue['number']}
    
    else:
        # 转人工，但提供上下文
        assign_to_human(ticket["id"], context={
            "analysis": analysis,
            "suggested_response": analysis.get("suggested_response", ""),
            "related_docs": analysis.get("related_docs", [])
        })
        return {"status": "escalated"}