我用AI工具重构了3个老项目_效率提升300%

我用 AI 工具重构了 3 个老项目,效率提升 300%

适用读者:全栈开发、技术 Leader、对 AI 编程感兴趣的工程师

技术栈:Python 3.11+ / FastAPI / Cursor / Claude / Copilot

阅读时长:约 15 分钟


一、为什么开始用 AI 重构?

去年接手了 3 个"祖传代码"项目:

项目状态痛点
A. 内部 API 服务Flask 单文件 2000 行无类型、难维护、文档缺失
B. 电商爬虫Python 2 + 同步 requests慢、漏数据、经常崩溃
C. 数据处理平台单体 5000 行没人敢动、改一处崩三处

3 个项目加起来,如果用传统方式重构,需要 3~4 周

实际用 AI 工具辅助后:7 天全部完成

下面分享具体怎么做的。


二、AI 工具选型

工具优势适用场景
Cursor编辑器内 AI,支持整个项目上下文大型重构、跨文件改动
GitHub Copilot行内补全、轻量写样板代码、补全
Claude / GPT-4长上下文、深度推理架构设计、复杂逻辑
Cody(Sourcegraph)跨仓库代码搜索理解老项目全貌

我的工作流:

  • Cursor做主力编辑器
  • Claude做架构讨论和复杂设计
  • Copilot补全样板代码

三、项目 A:Flask → FastAPI 重构

背景

老代码:一个app.py2000 行,所有路由、数据库操作、工具函数堆在一起。

# 重构前:app.py 节选(典型问题)fromflaskimportFlask,request,jsonifyimportpymysql app=Flask(__name__)# 全局 DB 连接,没有连接池DB_CONFIG={"host":"localhost","user":"root","password":"xxx","db":"shop",}defget_user(user_id):# 数据库连接每次都新建conn=pymysql.connect(**DB_CONFIG)# SQL 拼字符串,有注入风险sql=f"SELECT * FROM users WHERE id ={user_id}"# 没有异常处理cur=conn.cursor()cur.execute(sql)returncur.fetchone()@app.route("/user/<int:user_id>",methods=["GET"])defuser_api(user_id):# 参数没有校验# 返回字典直接 JSON 化,字段名不统一user=get_user(user_id)returnjsonify({"id":user[0],"name":user[1],"email":user[2],"create_time":str(user[3]),# datetime 转字符串})@app.route("/user",methods=["POST"])defcreate_user():data=request.get_json()# 没有校验,直接入库# ...

AI 提示词

请把以下 Flask 代码重构为 FastAPI。 要求: 1. 用 Pydantic 定义请求/响应模型 2. 用 Depends 做数据库连接管理 3. 添加类型注解 4. 用 SQLAlchemy 替换裸 SQL 5. 保留原接口路径和语义

重构后

# 重构后:app.py 拆分后的样子# models.py - Pydantic 模型(AI 自动生成,只需微调)frompydanticimportBaseModel,EmailStr,FieldfromdatetimeimportdatetimefromtypingimportOptionalclassUserResponse(BaseModel):# API 响应模型,自动生成 OpenAPI 文档id:intname:stremail:EmailStr create_time:datetime# 内部字段不暴露给前端internal_note:Optional[str]=Field(default=None,exclude=True)classUserCreate(BaseModel):# 创建用户请求模型,自带校验name:str=Field(min_length=1,max_length=50)email:EmailStr age:int=Field(ge=0,le=150)
# database.py - 数据库连接池(AI 给出的标准模板)fromsqlalchemyimportcreate_enginefromsqlalchemy.ormimportsessionmaker,SessionfromtypingimportGenerator DATABASE_URL="mysql+pymysql://root:xxx@localhost/shop"# 创建连接池(替代每次新建连接)engine=create_engine(DATABASE_URL,pool_size=10,max_overflow=20)SessionLocal=sessionmaker(bind=engine)defget_db()->Generator[Session,None,None]:# FastAPI 依赖注入,自动管理连接生命周期db=SessionLocal()try:yielddbfinally:db.close()
# main.py - FastAPI 入口fromfastapiimportFastAPI,Depends,HTTPExceptionfromsqlalchemy.ormimportSessionfrommodelsimportUserResponse,UserCreatefromdatabaseimportget_db app=FastAPI(title="User API",version="2.0")@app.get("/user/{user_id}",response_model=UserResponse)defget_user(user_id:int,db:Session=Depends(get_db)):# 类型注解 + 依赖注入,IDE 友好user=db.query(UserORM).filter(UserORM.id==user_id).first()ifnotuser:raiseHTTPException(status_code=404,detail="用户不存在")returnuser# SQLAlchemy 对象自动按 response_model 转换@app.post("/user",response_model=UserResponse,status_code=201)defcreate_user(data:UserCreate,db:Session=Depends(get_db)):# Pydantic 自动校验,非法请求直接 422user=UserORM(**data.model_dump())db.add(user)db.commit()db.refresh(user)returnuser

效果:

  • 行数:2000 → 600(主文件)+ 200(模型)+ 100(数据库)
  • 自动 OpenAPI 文档:/docs直接看
  • 类型安全:IDE 提示,运行时校验
  • 重构时间:3 天 → 1 天

四、项目 B:Python 2 爬虫现代化

背景

# 重构前:spider.py(典型老代码)# -*- coding: utf-8 -*-# Python 2.7importurllib2# Python 2 模块importreimporttimeimportsysreload(sys)sys.setdefaultencoding("utf-8")deffetch(url):# 没有超时,经常卡死response=urllib2.urlopen(url)returnresponse.read()defparse(html):# 正则硬解析,稍微改版就崩items=re.findall(r'<div class="item">(.+?)</div>',html,re.DOTALL)returnitemsdefmain():foriinrange(100):url="https://example.com/list?page="+str(i)html=fetch(url)items=parse(html)# 打印看,没存库,没日志foriteminitems:printitem# Python 2 语法time.sleep(1)# 反爬靠 sleepif__name__=="__main__":main()

AI 提示词

请把这个 Python 2 爬虫升级到 Python 3.11,要求: 1. 用 httpx 异步,支持并发 2. 用 BeautifulSoup 替代正则 3. 用 loguru 结构化日志 4. 加重试和异常处理 5. 数据存到 SQLite 6. 加上请求频率控制

重构后

# 重构后:spider.py# Python 3.11,异步 + 结构化importasyncioimporthttpxfrombs4importBeautifulSoupfromloguruimportloggerfromtypingimportListimportsqlite3fromdataclassesimportdataclass@dataclassclassProduct:# 用 dataclass 定义数据结构,比 dict 清晰title:strprice:floaturl:strclassAsyncSpider:# 异步爬虫,单进程并发 10 个请求# 对比老代码:100 个页面要 100 秒 -> 现在 10 秒def__init__(self,max_concurrent:int=10):self.semaphore=asyncio.Semaphore(max_concurrent)# httpx.AsyncClient 复用连接,比每次新建快self.client=httpx.AsyncClient(timeout=10.0,headers={"User-Agent":"Mozilla/5.0"},http2=True,# 启用 HTTP/2)asyncdeffetch(self,url:str,retries:int=3)->str:# 加重试机制,网络抖动不怕asyncwithself.semaphore:forattemptinrange(retries):try:resp=awaitself.client.get(url)resp.raise_for_status()returnresp.textexcepthttpx.HTTPErrorase:logger.warning(f"第{attempt+1}次重试{url}:{e}")awaitasyncio.sleep(2**attempt)# 指数退避raiseRuntimeError(f"请求失败{url}")defparse(self,html:str)->List[Product]:# BeautifulSoup 替代正则,改版不崩soup=BeautifulSoup(html,"html.parser")products=[]foriteminsoup.select("div.item"):try:products.append(Product(title=item.select_one(".title").text.strip(),price=float(item.select_one(".price").text.replace("¥","")),url=item.select_one("a")["href"],))except(AttributeError,ValueError)ase:# 单条解析失败不影响整体logger.error(f"解析失败:{e}")returnproductsasyncdefcrawl_page(self,page:int)->List[Product]:url=f"https://example.com/list?page={page}"html=awaitself.fetch(url)returnself.parse(html)asyncdefrun(self,total_pages:int=100):# 并发抓取所有页tasks=[self.crawl_page(i)foriinrange(total_pages)]all_results=awaitasyncio.gather(*tasks)# 扁平化products=[pforbatchinall_resultsforpinbatch]self.save(products)logger.info(f"抓取完成,共{len(products)}条")defsave(self,products:List[Product]):# 存储到 SQLite(也可以换成 MySQL/PostgreSQL)conn=sqlite3.connect("products.db")forpinproducts:conn.execute("INSERT OR REPLACE INTO products VALUES (?, ?, ?)",(p.title,p.price,p.url),)conn.commit()conn.close()asyncdefclose(self):awaitself.client.aclose()# 入口if__name__=="__main__":asyncdefmain():spider=AsyncSpider(max_concurrent=10)try:awaitspider.run(total_pages=100)finally:awaitspider.close()asyncio.run(main())

效果对比:

指标重构前重构后
抓取 100 页耗时100s10s
网络异常处理3 次重试 + 指数退避
改版后崩溃率80%< 5%
日志可读性print结构化 JSON 日志
重构时间估 3 天半天

五、项目 C:单体拆分 + 加测试

背景

5000 行的monolith.py,改一处崩三处,没人敢动。

AI 提示词(分步骤)

第 1 步:理解项目

请阅读 monolith.py,告诉我: 1. 整体架构(分层、模块依赖) 2. 核心业务逻辑有哪些 3. 哪些是"高频修改",哪些是"低频稳定" 4. 建议的拆分顺序

第 2 步:拆分方案

请把这个单体应用拆分为以下模块: - api/ (路由层) - services/(业务逻辑) - models/ (数据模型) - utils/ (工具函数) - config/ (配置) 要求保持接口兼容,逐步迁移。

第 3 步:补单元测试

请为 services/order_service.py 补充单元测试。 覆盖: 1. 正常下单流程 2. 库存不足的异常 3. 支付失败的回滚 4. 重复提交的幂等性 使用 pytest + pytest-mock

AI 给出的拆分结果

重构前: monolith.py (5000 行,所有东西都在里面) 重构后: api/ __init__.py user_api.py order_api.py services/ __init__.py user_service.py order_service.py models/ user.py order.py product.py utils/ db.py cache.py logger.py config/ settings.py constants.py tests/ test_user_service.py test_order_service.py test_payment.py

AI 生成的测试样例

# tests/test_order_service.py# AI 生成的单元测试importpytestfromunittest.mockimportMock,patchfromservices.order_serviceimportOrderServicefrommodels.orderimportOrderclassTestOrderService:# 测试订单服务的核心场景# 用 mock 隔离数据库,确保单元测试快速且独立@pytest.fixturedefservice(self):# 每个测试前都建一个全新的 servicereturnOrderService(db=Mock(),cache=Mock())deftest_create_order_success(self,service):# 场景 1:正常下单order_data={"user_id":1,"product_id":100,"quantity":2,}order=service.create_order(**order_data)assertorder.status=="pending"service.db.commit.assert_called_once()deftest_create_order_insufficient_stock(self,service):# 场景 2:库存不足service.db.query.return_value.filter.return_value.first.return_value.stock=0withpytest.raises(InsufficientStockError):service.create_order(user_id=1,product_id=100,quantity=1)deftest_create_order_payment_failure_rollback(self,service):# 场景 3:支付失败时,订单要回滚service.payment_client.charge.side_effect=PaymentError("银行拒绝")withpytest.raises(PaymentError):service.create_order(user_id=1,product_id=100,quantity=1)# 验证:订单被标记为 failed,而不是 pendingservice.db.rollback.assert_called_once()deftest_create_order_idempotent(self,service):# 场景 4:同一请求重复提交,只生成一个订单order_data={"user_id":1,"product_id":100,"quantity":2,"request_id":"abc-123",}service.create_order(**order_data)service.create_order(**order_data)# 重复请求# 验证:db.commit 只调用一次assertservice.db.commit.call_count==1

效果:

  • 5000 行 → 6 个模块,平均 800 行/模块
  • 测试覆盖率:0% → 78%
  • 修复 bug 平均时间:2 小时 → 15 分钟(因为有测试兜底)
  • 重构时间:1 周 → 3 天

六、3 个项目的总数据

项目传统估时AI 辅助实际提升
A. Flask → FastAPI3 天1 天300%
B. 爬虫现代化3 天0.5 天600%
C. 单体拆分 + 测试1 周3 天230%
合计13 天4.5 天~290%

注:数字含"AI 产出后人工 review + 调整"的时间。如果纯让 AI 写,可能再快 50%,但生产代码必须 review。


七、用好 AI 重构的 5 条经验

1. 先让 AI 理解代码,再让 AI 改

❌ 一上来就:把 X 改成 Y ✅ 先问:这个文件做什么?哪些是核心逻辑?哪些是次要的? 再问:如果要重构,你会怎么拆?给出方案。 最后:按方案改

2. 提供具体约束,不要开放式

❌ "请优化这段代码" ✅ "请把 O(n²) 改成 O(n),保留返回结果,加注释说明算法"

3. 一次只改一处,不要一次性大改

❌ 把整个 5000 行文件丢给 AI 一次性重构 ✅ 先拆函数 -> 再加类型 -> 再加测试,逐步推进

4. 一定要写测试,作为 AI 的"质检员"

# 重构前先写好"行为快照测试"deftest_legacy_behavior():# 把老代码的输出存下来,确保重构后输出一致assertnew_func(input)==snapshot

5. 保留 git 历史,频繁 commit

gitcommit-m"refactor: 用 AI 拆分 utils 模块"gitcommit-m"test: 给 order_service 加单元测试"

AI 改坏了可以一行命令回滚。


八、AI 还不能做什么(诚实清单)

能做的还做不好的
样板代码、CRUD业务核心逻辑(需要领域知识)
框架迁移、API 翻译跨团队架构决策
写单元测试写复杂集成测试
解释老代码理解"为什么这么写"的历史原因
重构 + 性能优化真正的性能 profiling
生成文档判断文档是否"对人友好"

最重要的判断:AI 重构后,人必须 review。AI 不知道业务上下文,可能改对了语法、改错了语义。


九、推荐工作流(可直接套用)

1. 阅读老代码 -> 列出"核心问题清单"(手写 10 分钟) 2. 让 AI 给出重构方案 + 拆分建议(AI 5 分钟) 3. 人工 review 方案,标注哪些可以 AI 自动改、哪些要自己改 4. 拆任务:小步快跑,每改一处就 commit 5. 让 AI 写测试,人工 review 6. 跑全套测试 + 手动冒烟 7. Code Review(让同事帮忙看一眼) 8. 上线灰度

十、总结

AI 工具不是"替代程序员",而是把程序员从重复劳动中解放出来

过去:

  • 写样板代码 60% 时间
  • 改 bug 30% 时间
  • 思考业务 10% 时间

用 AI 之后:

  • 写样板代码 10% 时间
  • 改 bug 15% 时间
  • 思考业务50%时间
  • Review AI 输出 25% 时间

本质变化:从"搬砖"变成"设计"

如果你还在犹豫要不要用 AI 重构老项目,我的建议是:选一个 1000 行左右的项目,先试试 1 天。你会立刻感受到区别。


参考工具

  • Cursor:https://www.cursor.com/
  • GitHub Copilot:https://github.com/features/copilot
  • Claude:https://claude.ai/
  • Cody(Sourcegraph):https://about.sourcegraph.com/cody