后端代码优化

一次关于拆分、解耦和异常治理的代码进化记录

一、单一职责落地

痛点回顾

ai_chat_service.py在一个文件里同时承担了：

• LLM 流式调用编排
• 会话记忆管理
• 用户问题意图识别
• SQL 安全校验

每次改动任何一个维度，都得在这700行里小心翻找，而且任何修改都有波及其他功能的风险。

重构方案

按单一职责原则拆分为独立模块：

services/ai_chat/ ├── __init__.py # 统一导出，对外透明 ├── service.py # 主服务：stream_chat 入口 ├── memory.py # MemoryManager：记忆管理 ├── planner.py # LLMQuestionPlanner：意图识别 └── sql_safety.py # SQL 安全作用域

关键的细节在__init__.py：

fromopenapi_server.services.ai_chat.serviceimportAIChatService,get_ai_chat_servicefromopenapi_server.services.ai_chat.memoryimportMemoryManager# ...__all__=["AIChatService","get_ai_chat_service","MemoryManager",...]

调用方只需要改 import 路径，符号名称完全不变：

# 优化前fromopenapi_server.services.ai_chat_serviceimportAIChatService# 优化后fromopenapi_server.services.ai_chatimportAIChatService

收益

二、代码去重

痛点回顾

uploadWechatBillData和uploadAlipayBillData两个接口，除了解析 Excel/CSV 的细节不同，后续的 AI 分类、分批入库、SSE 进度推送完全一致，维护成本翻倍。

重构方案

提取公共方法_process_uploaded_bills()：

asyncdef_process_uploaded_bills(parsed_data:list[Bill],user:UserDep,platform_name:str,batch_size:int=50,)->StreamingResponse:"""处理上传账单的公共逻辑：AI分类 + 分批入库 + SSE推送"""asyncdefevent_stream():foriinrange(0,len(parsed_data),batch_size):batch=parsed_data[i:i+batch_size]# AI 分类results=awaitclassify_bill_batch(batch)forj,rinenumerate(results):batch[j].transaction_type=r.transaction_type# 入库withSession(engine)asdb_session:db_session.add_all(batch)db_session.commit()yield_sse_event({"processed":processed,"total":total})yield_sse_event({"status":"completed"})returnStreamingResponse(event_stream(),media_type="text/event-stream")

两个上传接口瞬间精简：

@BillAPIRouter.post('/uploadWechatBillData')asyncdefupload_wechat_bill_data(excelData:UploadFile,user:UserDep):parsed_data=parse_wechat_bill(excelData)returnawait_process_uploaded_bills(parsed_data,user,"微信")@BillAPIRouter.post('/uploadAlipayBillData')asyncdefupload_alipay_bill_data(csvData:UploadFile,user:UserDep):parsed_data=parse_alipay_bill(csvData)returnawait_process_uploaded_bills(parsed_data,user,"支付宝")

启发

去重的关键不在于机械地提取函数，而在于识别业务流程的本质——两个接口的差异只在于第一步的解析，后续流程完全相同。这种“模板方法”模式在业务开发中非常实用。

三、模型文件拆分：从混乱到有序

30多个 Pydantic 模型堆在apis/models.py一个文件里，每次改一个接口的模型都要翻半天。按业务域拆分：

apis/models/ ├── auth.py # 登录/注册/刷新Token ├── bill.py # 账单CRUD ├── analysis.py # 月度报告/分类统计/趋势 ├── budget.py # 预算 └── ai_chat.py # AI聊天

__init__.py统一重导出，调用方无需任何修改：

fromopenapi_server.apis.modelsimportBillItem# 完全没变

四、异常处理统一：告别散落的 HTTPException

重构前

每个路由都直接raise HTTPException(status_code=xxx, detail="...")：

ifbillisNone:raiseHTTPException(status_code=404,detail="账单不存在")ifnotuser:raiseHTTPException(status_code=401,detail="用户不存在")

问题：

• 错误消息硬编码，分散在各处
• 同类错误的 HTTP 状态码在不同路由可能不一致
• 无法统一做日志告警、重试等策略

重构后

定义异常层次结构：

classAppException(Exception):def__init__(self,message:str,status_code:int=500):self.message=message self.status_code=status_codeclassResourceNotFoundError(AppException):def__init__(self,resource_type:str,resource_id:int|str=None):message=f"{resource_type}不存在"+(f" (ID:{resource_id})"ifresource_idelse"")super().__init__(message,status_code=404)classBillNotFoundError(ResourceNotFoundError):def__init__(self,bill_id:int=None):super().__init__("账单",bill_id)classUnauthorizedError(AppException):def__init__(self,message:str="权限不足"):super().__init__(message,status_code=401)

配合中间件统一捕获处理：

classExceptionHandlerMiddleware(BaseHTTPMiddleware):asyncdefdispatch(self,request,call_next):try:returnawaitcall_next(request)exceptAppExceptionase:logger.warning("应用异常: %s, 路径: %s",e.message,request.url.path)returnJSONResponse(status_code=e.status_code,content={"detail":e.message})exceptSQLAlchemyErrorase:logger.error("数据库异常: %s",str(e))returnJSONResponse(status_code=500,content={"detail":"数据库操作失败"})exceptExceptionase:logger.exception("未处理异常: %s",type(e).__name__)returnJSONResponse(status_code=500,content={"detail":"服务器内部错误"})

路由层语义清晰：

# 重构前raiseHTTPException(status_code=404,detail="账单不存在")# 重构后raiseBillNotFoundError(bill_id)

替换统计

五、两个细节

5.1 语法错误修复

services/ai_chat/service.py隐藏的语法错误——Prompt 字符串里的 ASCII 双引号与 Python 字符串定界符冲突：

# 有问题的代码"禁止使用同比或环比，统一写"与指定对比"周期相比"。"# ↑ 这个双引号结束了字符串# 修复后——使用中文引号"禁止使用同比或环比，统一写“与指定对比”周期相比。"

这类语法错误在动态语言中很容易被忽视，直到特定路径被触发才暴露。

5.2 全局变量的隐患

重构前模块级全局变量在 import 时就创建实例：

_ai_chat_service=AIChatService()# 模块加载时立即创建defget_ai_chat_service():return_ai_chat_service

问题：

• 不必要的初始化开销
• 测试时必须用dependency_overridesmock，属于“后门”注入
• 不利于延迟初始化和多实例管理

改为@lru_cache实现懒加载单例：

fromfunctoolsimportlru_cache@lru_cache(maxsize=1)defget_ai_chat_service():returnAIChatService()

小结

这次重构的核心原则是：在保持对外接口完全不变的前提下，彻底重构内部架构。