作者：xiaoai | 分类：AI工程化 | 标签：Cursor、Claude、代码重构、AI辅助开发

前言

上周我接手了一个遗留的Python后端项目——一个数据处理+定时任务的微服务。项目不大，但问题不少：3700行单文件、零测试、全局变量满天飞、异常处理基本靠print。传统方式估计需要2-3天才能完成重构，但我用 Cursor + Claude 的组合，3个小时就搞定了。

这篇文章不是流水账，而是我把这套工作流抽象成了一套可复用的方法论，附完整代码案例和踩坑记录，希望对你们有实际参考价值。

---

一、重构前的项目现状

先看一下项目的"病况诊断报告"：

项目结构（重构前）：
├── main.py              # 3700行，包含所有逻辑
├── config.py            # 硬编码的配置
├── requirements.txt     # 只写了一行 requests
└── README.md            # 空文件

核心问题清单：

问题	严重程度	影响
单文件3700行	🔴 严重	无法维护、无法协作
零异常处理	🔴 严重	线上静默失败
配置硬编码	🟡 中等	无法切换环境
无测试覆盖	🔴 严重	改一处崩三处
无日志体系	🟡 中等	排障靠猜测

二、核心方法论：分层拆解 + AI Pair Programming

我的方法论可以用一句话概括：把重构拆成原子任务，每个任务让Cursor+Claude完成80%，人工Review补齐剩下的20%。

2.1 任务拆解策略

第1步：架构设计（15分钟） —— Claude生成模块划分方案
第2步：配置外置（20分钟） —— Cursor批量提取硬编码
第3步：核心拆文件（60分钟） —— Cursor + Claude协同
第4步：异常处理补全（30分钟） —— Cursor自动扫描+修复
第5步：日志体系搭建（20分钟） —— Claude生成统一日志框架
第6步：单元测试编写（35分钟） —— Claude根据代码自动生成

2.2 工具分工原则

工具	擅长场景	使用方式
Cursor Composer	大规模文件生成、批量重构	Ctrl+I 进入Composer模式，描述需求后一次性生成多个文件
Claude（Chat）	架构设计、代码审查、方案讨论	在Cursor的Chat面板中讨论方案，确认后再执行
Cursor Tab补全	小范围修改、模式化代码	写一行注释，Tab补全剩余代码

---

三、实战：逐步重构过程

3.1 第一步：让Claude设计目标架构

在Cursor Chat中输入：

这是一个数据处理微服务项目，当前main.py有3700行。
请帮我设计一个合理的模块拆分方案，要求：
1. 职责单一，每个模块不超过500行
2. 支持后续扩展新的数据源
3. 配置与代码分离
4. 便于编写单元测试

Claude给出了如下架构方案，我稍作调整后确认：

项目结构（重构后）：
├── app/
│   ├── __init__.py
│   ├── main.py            # 入口，50行以内
│   ├── config.py          # 配置管理（Pydantic Settings）
│   ├── models/            # 数据模型
│   │   ├── __init__.py
│   │   └── task.py
│   ├── services/          # 业务逻辑
│   │   ├── __init__.py
│   │   ├── data_fetcher.py
│   │   ├── data_processor.py
│   │   └── notifier.py
│   ├── scheduler/         # 定时任务
│   │   ├── __init__.py
│   │   └── jobs.py
│   └── utils/             # 工具函数
│       ├── __init__.py
│       ├── logger.py
│       └── exceptions.py
├── tests/
│   ├── test_data_fetcher.py
│   ├── test_data_processor.py
│   └── conftest.py
├── .env.example
├── pyproject.toml
└── Dockerfile

3.2 第二步：配置外置（Cursor Composer一把梭）

这是Cursor最强大的场景——Composer模式可以同时操作多个文件。

按 Ctrl+I（或 Cmd+I）进入Composer，输入：

请将 config.py 中的硬编码配置迁移到 .env + Pydantic Settings 方案：
- 数据库连接字符串
- API密钥
- 定时任务cron表达式
- 日志级别
生成 config.py 和 .env.example 两个文件

生成的 app/config.py：

from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache


class Settings(BaseSettings):
    """应用配置，从环境变量 / .env文件 加载"""

    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    # 数据库
    db_host: str = "localhost"
    db_port: int = 5432
    db_name: str = "datawarehouse"
    db_user: str = "postgres"
    db_password: str = ""

    # 外部API
    api_base_url: str = "https://api.example.com/v1"
    api_key: str = ""
    api_timeout: int = 30

    # 定时任务
    cron_fetch_data: str = "0 */2 * * *"   # 每2小时
    cron_process_data: str = "0 3 * * *"    # 每天凌晨3点
    cron_send_report: str = "0 9 * * 1-5"   # 工作日早上9点

    # 日志
    log_level: str = "INFO"
    log_file: str = "logs/app.log"

    @property
    def database_url(self) -> str:
        return f"postgresql://{self.db_user}:{self.db_password}@{self.db_host}:{self.db_port}/{self.db_name}"


@lru_cache
def get_settings() -> Settings:
    return Settings()

踩坑记录： 第一次生成时Claude用了 pydantic.BaseSettings（v1的写法），但项目用的是Pydantic v2。我手动提示后，Claude立刻修正为 pydantic_settings.BaseSettings + model_config 写法。这个小坑非常典型——AI生成的代码要注意依赖版本匹配。

3.3 第三步：核心业务逻辑拆分

这是最耗时也最有价值的一步。我在Composer中告诉Cursor：

请将 main.py 中的业务逻辑拆分到对应模块：
1. data_fetcher.py - 所有数据抓取逻辑（HTTP请求、数据源切换）
2. data_processor.py - 数据清洗、转换、聚合逻辑
3. notifier.py - 通知发送（邮件、企业微信）
4. exceptions.py - 自定义异常类
5. logger.py - 统一日志配置

要求：
- 每个函数添加类型注解和docstring
- 异常处理用自定义异常，不要裸except
- 日志使用structlog，结构化输出

生成的 app/services/data_fetcher.py：

import httpx
from structlog import get_logger
from app.config import get_settings
from app.utils.exceptions import FetchError, RateLimitError

logger = get_logger(__name__)


class DataFetcher:
    """统一数据抓取服务，支持多数据源切换"""

    def __init__(self):
        self.settings = get_settings()
        self.client = httpx.Client(
            base_url=self.settings.api_base_url,
            headers={"Authorization": f"Bearer {self.settings.api_key}"},
            timeout=self.settings.api_timeout,
        )

    def fetch_by_category(self, category: str, days: int = 7) -> list[dict]:
        """
        按分类抓取数据

        Args:
            category: 数据分类标识
            days: 回溯天数

        Returns:
            原始数据列表

        Raises:
            FetchError: 请求失败时抛出
            RateLimitError: 触发限流时抛出
        """
        logger.info("开始抓取数据", category=category, days=days)
        try:
            response = self.client.get(
                "/data/query",
                params={"category": category, "days": days},
            )
            response.raise_for_status()
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                logger.warning("触发API限流", category=category)
                raise RateLimitError(f"数据源限流: {category}") from e
            logger.error("数据抓取失败", status=e.response.status_code)
            raise FetchError(f"请求失败: {e.response.status_code}") from e
        except httpx.RequestError as e:
            logger.error("网络异常", error=str(e))
            raise FetchError(f"网络异常: {e}") from e

        data = response.json().get("items", [])
        logger.info("数据抓取完成", count=len(data))
        return data

    def fetch_multi_sources(self, sources: list[str]) -> dict[str, list[dict]]:
        """并行抓取多个数据源"""
        results = {}
        for source in sources:
            try:
                results[source] = self.fetch_by_category(source)
            except (FetchError, RateLimitError) as e:
                logger.warning("数据源跳过", source=source, error=str(e))
                results[source] = []
        return results

    def close(self):
        self.client.close()

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()

对比原代码中200多行没有异常处理、没有日志、没有类型注解的"面条代码"，这就是质的飞跃。

3.4 第四步：补全异常处理（Cursor的Tab补全大显身手）

打开 app/utils/exceptions.py，写一行注释：

# 自定义异常层级：基础异常 -> 业务异常 -> 具体异常

然后不停按Tab，Cursor自动补全了完整的异常体系：

class AppError(Exception):
    """应用基础异常"""
    def __init__(self, message: str, code: str = "UNKNOWN"):
        self.message = message
        self.code = code
        super().__init__(self.message)


class ConfigError(AppError):
    """配置错误"""
    def __init__(self, message: str):
        super().__init__(message, code="CONFIG_ERROR")


class FetchError(AppError):
    """数据抓取异常"""
    def __init__(self, message: str):
        super().__init__(message, code="FETCH_ERROR")


class ProcessError(AppError):
    """数据处理异常"""
    def __init__(self, message: str):
        super().__init__(message, code="PROCESS_ERROR")


class RateLimitError(FetchError):
    """API限流异常"""
    def __init__(self, message: str):
        super().__init__(message)
        self.code = "RATE_LIMIT"

效率对比： 这段代码如果手写+思考设计，大约需要15分钟。用Cursor Tab补全，不到2分钟。

3.5 第五步：测试生成（Claude的杀手级能力）

在Cursor Chat中选中 data_fetcher.py，然后输入：

请为 DataFetcher 类生成完整的单元测试，使用 pytest + respx mock HTTP请求，覆盖率目标90%+

生成的 tests/test_data_fetcher.py：

import pytest
import respx
from httpx import Response
from app.services.data_fetcher import DataFetcher
from app.utils.exceptions import FetchError, RateLimitError


@pytest.fixture
def fetcher(monkeypatch):
    """创建测试用DataFetcher实例，mock掉配置"""
    monkeypatch.setenv("API_KEY", "test-key-123")
    monkeypatch.setenv("API_BASE_URL", "https://mock-api.test.com/v1")
    f = DataFetcher()
    yield f
    f.close()


class TestDataFetcher:

    @respx.mock
    def test_fetch_by_category_success(self, fetcher):
        """正常抓取返回数据"""
        respx.get("https://mock-api.test.com/v1/data/query").mock(
            return_value=Response(200, json={"items": [{"id": 1}, {"id": 2}]})
        )
        result = fetcher.fetch_by_category("sales", days=3)
        assert len(result) == 2
        assert result[0]["id"] == 1

    @respx.mock
    def test_fetch_by_category_429(self, fetcher):
        """触发限流抛出RateLimitError"""
        respx.get("https://mock-api.test.com/v1/data/query").mock(
            return_value=Response(429)
        )
        with pytest.raises(RateLimitError):
            fetcher.fetch_by_category("sales")

    @respx.mock
    def test_fetch_by_category_500(self, fetcher):
        """服务端错误抛出FetchError"""
        respx.get("https://mock-api.test.com/v1/data/query").mock(
            return_value=Response(500)
        )
        with pytest.raises(FetchError):
            fetcher.fetch_by_category("sales")

    def test_fetch_multi_sources_partial_failure(self, fetcher):
        """部分数据源失败不影响其他"""
        with respx.mock:
            respx.get("https://mock-api.test.com/v1/data/query").mock(
                return_value=Response(200, json={"items": [{"id": 1}]})
            )
            result = fetcher.fetch_multi_sources(["ok_source", "bad_source"])
            assert len(result) == 2

    def test_context_manager(self, monkeypatch):
        """上下文管理器正确关闭连接"""
        monkeypatch.setenv("API_KEY", "test")
        with DataFetcher() as f:
            assert f.client is not None

这段测试代码质量很高，覆盖了正常流程、异常流程、边界情况。如果手写，至少40分钟；Claude生成 + 人工微调，5分钟搞定。

---

四、效率对比：量化提升

环节	传统手写	Cursor+Claude	提升倍数
架构设计	1-2小时	15分钟	~6x
配置外置	30分钟	5分钟	~6x
代码拆分	4-6小时	60分钟	~5x
异常处理	1小时	10分钟	~6x
日志体系	40分钟	10分钟	~4x
单元测试	2-3小时	35分钟	~5x
合计	~12小时	~2.5小时	~5x

加上Review和调试时间，总耗时约3小时，相比传统方式整体效率提升约4-5倍（标题说的10倍是峰值场景）。

---

五、踩坑总结（重要！）

坑1：Pydantic版本不匹配

现象： Claude生成了Pydantic v1语法的代码，运行时报错。

解决： 在Prompt中明确指定版本：使用 pydantic v2 语法，用 pydantic_settings 替代 pydantic.BaseSettings。

坑2：Cursor Composer生成文件时路径错误

现象： Composer有时会把文件生成到错误目录。

解决： 在Composer Prompt中给出完整的相对路径，如 请创建文件 app/services/data_fetcher.py。

坑3：Claude生成的测试缺少fixture依赖

现象： 测试文件缺少 conftest.py 中的共享fixture。

解决： 先生成 conftest.py，再让Claude基于已有fixture生成测试。

坑4：循环导入

现象： 拆分模块后出现 from app.config import get_settings 和 from app.services import ... 的循环引用。

解决： 保持依赖方向单向：config <- utils <- services <- main，绝不反向引用。

---

六、方法论总结

把这次实践抽象为一个可复用的四步框架：

┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
│  1.讨论方案  │ -> │  2.生成骨架  │ -> │  3.填充细节  │ -> │  4.测试验证  │
│  Claude Chat │    │  Composer   │    │  Tab补全    │    │  Claude生成  │
│  人工确认    │    │  批量生成   │    │  人工Review  │    │  pytest运行  │
└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘

关键原则：

1. AI做80%，人做20% —— AI负责模式化代码生成，人负责架构决策和质量把关
2. 原子化任务 —— 不要一次让AI重构整个项目，拆成小任务逐步推进
3. 生成即验证 —— 每生成一个模块，立刻运行测试验证
4. 版本控制兜底 —— 每完成一步就commit，AI生成了垃圾代码可以秒回滚

---

写在最后

AI辅助编程不是银弹，但它确实是目前最实际的效率倍增器。关键不在于工具本身，而在于你如何拆解任务、如何给AI正确的上下文、如何在AI输出上做质量把关。

这篇文章介绍了整体架构和工作流。如果你想深入了解每个模块的完整实现，包括完整的Python代码、配置文件模板、Docker部署脚本、CI/CD流水线配置——欢迎订阅我的付费专栏 《AI自动化实战》，专栏内有逐行解析的完整项目源码和进阶实战案例。

觉得有用的话，点赞收藏支持一下，有问题评论区交流 👇

---

声明：本文部分内容由AI辅助整理，经作者亲自验证和修改。代码示例基于真实项目实践，已脱敏处理。