简介这是一套面向Python开发者与闲鱼商家的自动化客服系统源码基于FastAPI与WebSocket实现闲鱼消息实时监听与智能响应解决多账号人工回复效率低、响应不及时、发货易出错等运营痛点。资源包共57个文件含20个核心Python模块如ws_utils.py、ai_reply_engine.py、image_uploader.py、13个前端样式与交互文件CSS/HTML/JS、3个Docker部署配置docker-compose.yml及双语Dockerfile、以及日志采集、安全解密、二维码登录等关键工具脚本整体压缩包仅1.53MB轻量易部署。已有809人学习下载涵盖从环境搭建、多用户权限管理、关键词分级匹配策略到AI上下文回复、图片自动CDN上传及防重复智能发货等完整链路代码结构清晰、模块职责分明附带README.md与详细配置说明可直接二次开发或快速落地私域客服提效场景。1. 闲鱼自动回复系统不是“群控脚本”而是带 JWT 隔离 WebSocket 实时通道 OpenAI 上下文理解的生产级客服中台你搜“闲鱼自动回复源码”90% 的结果是 Python Selenium 模拟点击、定时轮询消息、发完就断连的“半残废脚本”——它们卡在登录页、被风控踢号、消息延迟 3 分钟起步、多账号互相干扰甚至把买家问“能包邮吗”直接回复成“已发货”。而xianyu-auto-reply不是这种玩具。它用 FastAPI 构建了真正的服务化架构每个用户注册后拥有独立数据库 schemaPostgreSQL、独立 JWT token、独立 WebSocket 连接池每个闲鱼账号以独立进程运行在 Docker 容器里通过ws_utils.py封装的长连接心跳保活机制维持 24 小时不掉线消息流走message_utils.py的优先级路由引擎——先查商品 ID 是否命中专属回复规则再匹配该商品下的专用关键词最后 fallback 到全局通用词库或 OpenAI 上下文生成调用前自动注入商品标题、价格、买家历史对话片段。这不是“自动化”是把闲鱼客服流水线搬进了 Linux 容器。适合日均咨询量超 50 条的个体卖家、小团队运营者、二手数码/教材/手作类目店主——你不需要懂 WebSocket 协议细节但必须能看懂docker-compose.yml里restart: unless-stopped的含义以及为什么global_config.yml中ai_reply_enabled: true默认关闭。2. 从源码结构到核心链路FastAPI 服务层、WebSocket 接入层、规则引擎三层解耦设计2.1 目录即架构6 类模块分工明确拒绝“一坨 Python”整个项目目录不是扁平堆砌而是按职责分层组织。reply_server.py是 FastAPI 入口暴露/api/v1/login、/api/v1/rules/import等 REST 接口ws_utils.py和xianyu_utils.py构成 WebSocket 接入层封装了闲鱼 PC 端 WebSocket 协议逆向解析含xianyu_js_version_2.js提供的加密签名逻辑ai_reply_engine.py和message_utils.py组成规则引擎层前者调 OpenAI API 并做 prompt 工程含变量替换模板后者实现五级优先级策略order_detail_fetcher.py和secure_freeshipping_ultra.py是业务插件层专攻发货逻辑db_manager.py和cookie_manager.py是数据管理层前者用 SQLAlchemy ORM 建模用户/账号/规则三张核心表后者管理 Cookie 加密存储AES-256-CBC 盐值派生utils/下的image_uploader.py和file_log_collector.py是基础设施层前者对接七牛云/阿里云 OSS配置在config.py后者按小时切割日志并压缩归档。这种分层让调试变得可定位当买家消息没触发回复先查ws_utils.py的on_message()是否收到原始 payload再进message_utils.py的route_reply()看规则匹配路径最后验ai_reply_engine.py的generate_contextual_reply()是否返回空字符串。2.2 WebSocket 连接不是“连上就行”关键在心跳与重连状态机闲鱼 PC 端 WebSocket 连接有严格会话生命周期首次连接需携带X-Device-Id和X-Session-Id头每 30 秒必须发送{type:heartbeat}心跳包连续 2 次未响应即断连。ws_utils.py用asyncio实现了状态机驱动的重连逻辑# ws_utils.py class XianyuWebSocket: def __init__(self, account_id: str): self.account_id account_id self._state disconnected # disconnected / connecting / connected / reconnecting self._reconnect_delay 1.0 # 初始重连间隔秒 self._max_reconnect_delay 300.0 # 最大重连间隔5分钟 async def connect(self): while self._state ! connected: try: self.ws await websockets.connect( fwss://message.xianyu.com/ws?account_id{self.account_id}, extra_headers{X-Device-Id: self._get_device_id(), X-Session-Id: self._get_session_id()} ) self._state connected self._reconnect_delay 1.0 # 成功后重置延迟 asyncio.create_task(self._heartbeat_loop()) # 启动心跳协程 logger.info(f[{self.account_id}] WebSocket connected) except Exception as e: logger.warning(f[{self.account_id}] Connect failed: {e}, retry in {self._reconnect_delay}s) self._state reconnecting await asyncio.sleep(self._reconnect_delay) self._reconnect_delay min(self._reconnect_delay * 1.5, self._max_reconnect_delay) # 指数退避这段代码的关键不在连接本身而在self._reconnect_delay min(self._reconnect_delay * 1.5, self._max_reconnect_delay)—— 它实现了指数退避重连。实测中闲鱼服务器在高并发时段会主动断开部分连接若用固定 1 秒重试会导致大量账号同时发起连接请求触发 IP 限频而指数退避让各账号错峰重连成功率从 62% 提升至 98.7%。_heartbeat_loop()内部用asyncio.wait_for(ws.send(...), timeout5)包裹心跳发送并捕获asyncio.TimeoutError触发断连重建这是避免“假连接”的核心。2.3 规则引擎的五级优先级不是理论而是硬编码的 if-elif-else 链message_utils.py中route_reply()函数是规则调度中枢其逻辑直白但致命# message_utils.py def route_reply(self, msg: dict, user_id: str, item_id: str None) - str: # Level 1: 指定商品回复最高优先级 if item_id and (specific_reply : self._get_specific_item_reply(user_id, item_id)): return self._render_template(specific_reply, msg, item_id) # Level 2: 商品专用关键词需 item_id 存在且匹配 if item_id and (keyword_reply : self._match_item_keywords(user_id, item_id, msg.get(content, ))): return self._render_template(keyword_reply, msg, item_id) # Level 3: 通用关键词全局生效 if (general_reply : self._match_general_keywords(user_id, msg.get(content, ))): return self._render_template(general_reply, msg, item_id) # Level 4: 默认回复用户配置的兜底话术 if (default_reply : self._get_default_reply(user_id)): return self._render_template(default_reply, msg, item_id) # Level 5: AI 回复仅当显式启用且 API Key 有效 if self.ai_enabled and self._is_ai_eligible(msg): return self.ai_engine.generate_contextual_reply(msg, user_id, item_id) return # 无匹配不回复注意self._render_template()的调用位置——它在每一级都执行而非只在最终返回前。这意味着变量替换如{buyer_name}、{item_title}在匹配阶段就完成避免了“匹配到规则却因变量缺失导致回复为空”的玄学翻车。_is_ai_eligible()会检查消息长度50 字符、是否含敏感词黑名单过滤、是否为重复消息MD5 去重缓存防止 AI 被滥用刷 token。这个 if-elif-else 链没有用策略模式抽象因为闲鱼场景下规则匹配速度比扩展性更重要实测单条消息平均耗时 12.3msi5-10210U而策略模式引入的反射调用会增加 8~15ms 开销。2.4 自动发货不是“发个文字”而是规格-卡券-图片的三维映射secure_freeshipping_ultra.py实现了发货逻辑其核心是ShippingRule数据模型# models.py class ShippingRule(Base): __tablename__ shipping_rules id Column(Integer, primary_keyTrue) user_id Column(String(36), nullableFalse) item_id Column(String(64), nullableTrue) # 为空则为全局规则 spec_match Column(String(255), nullableTrue) # 如 内存:16GB|硬盘:512GB coupon_code Column(Text, nullableTrue) # 卡券文本 image_url Column(String(512), nullableTrue) # CDN 图片地址 delay_seconds Column(Integer, default0) # 发货延时 trigger_type Column(Enum(pay, offer, msg), defaultpay) # 触发条件 priority Column(Integer, default0) # 优先级数值越大越优先发货时order_detail_fetcher.py先解析买家下单详情含规格字符串再调用ShippingRule.match_rule(user_id, item_id, spec_string)进行三维匹配精确匹配spec_match字段完全等于订单规格如颜色:黑色|内存:16GB模糊匹配若精确失败则用正则提取规格关键词如r内存:(\dGB)再查是否存在spec_match LIKE %内存:16GB%的规则兜底匹配若前两者皆无则取item_id IS NULL AND spec_match IS NULL的全局规则匹配成功后根据coupon_code、image_url、delay_seconds组合生成发货动作——不是简单send_text()而是构造闲鱼协议要求的富文本消息含图片 base64 或 CDN URL、卡券可复制文本块。delay_seconds由asyncio.sleep()实现但实际部署时建议改用 Redis Delayed Queueredis-py的zaddzrangebyscore避免阻塞事件循环。提示secure_freeshipping_ultra.py中send_shipping_message()方法默认使用requests.post()调闲鱼内部 API但该接口未公开文档。项目通过抓包xianyu-group.png所在页面的 XHR 请求逆向得出Header 必须包含X-XSRF-TOKEN从cookie_manager.py获取和Content-Type: application/json。若发现发货失败先检查X-XSRF-TOKEN是否过期有效期 2 小时而非怀疑网络问题。3. Docker 部署不是“docker-compose up”而是环境隔离、配置注入、日志归集三步闭环3.1 Dockerfile-cn 与 docker-compose-cn.yml中文环境专用镜像构建链项目提供Dockerfile-cn而非标准Dockerfile原因在于闲鱼 JS 逆向依赖nodejs运行时用于执行xianyu_js_version_2.js中的加密函数而国内网络无法稳定拉取node:18-alpine镜像。Dockerfile-cn显式指定国内镜像源# Dockerfile-cn FROM registry.cn-hangzhou.aliyuncs.com/library/node:18-alpine # 设置国内 npm 源 RUN npm config set registry https://registry.npmmirror.com \ npm install -g pm2 # 复制 Python 依赖国内 pip 源已在 requirements.txt 中声明 COPY requirements.txt . RUN pip install --index-url https://pypi.tuna.tsinghua.edu.cn/simple/ -r requirements.txt # 复制源码 COPY . /app WORKDIR /app # 暴露端口 EXPOSE 8000 CMD [pm2, start, entrypoint.sh]docker-compose-cn.yml则针对中文环境优化了 volume 挂载和时区# docker-compose-cn.yml version: 3.8 services: api: build: context: . dockerfile: Dockerfile-cn image: xianyu-auto-reply:latest restart: unless-stopped environment: - TZAsia/Shanghai # 关键否则日志时间错乱 - DATABASE_URLpostgresql://user:passdb:5432/xianyu - OPENAI_API_KEY${OPENAI_API_KEY} volumes: - ./logs:/app/logs:rw # 日志挂载便于宿主机收集 - ./uploads:/app/static/uploads:rw # 图片上传目录 - ./config:/app/config:ro # 配置文件只读挂载 ports: - 8000:8000 depends_on: - db db: image: postgres:14-alpine restart: unless-stopped environment: - POSTGRES_DBxianyu - POSTGRES_USERuser - POSTGRES_PASSWORDpass volumes: - ./pgdata:/var/lib/postgresql/data:rw command: postgres -c log_statementall -c log_min_duration_statement1000注意TZAsia/Shanghai和log_min_duration_statement1000前者确保logging模块输出的时间戳与本地一致避免排查时序问题后者让 PostgreSQL 记录所有耗时超 1 秒的 SQL方便定位慢查询如规则匹配时的LIKE全表扫描。3.2 config.py 与 global_config.yml配置分离的双保险机制项目采用两层配置config.py是代码级常量如MAX_WS_CONNECTIONS 5global_config.yml是运行时可变参数如ai_reply_enabled: false。这种分离解决了“改代码重启 vs 改配置热加载”的矛盾。global_config.yml由reply_server.py在启动时读取并注入FastAPI的app.state# reply_server.py app.on_event(startup) async def load_global_config(): try: with open(config/global_config.yml, r, encodingutf-8) as f: app.state.config yaml.safe_load(f) logger.info(Global config loaded) except FileNotFoundError: logger.error(global_config.yml not found, using defaults) app.state.config {ai_reply_enabled: False, log_level: INFO}app.state.config可在任意路由中访问例如/api/v1/rules/test接口会检查app.state.config.get(ai_reply_enabled, False)决定是否调用 AI 引擎。global_config.yml的修改无需重启服务但需注意ai_reply_enabled切换为true后ai_reply_engine.py会立即生效而ws_utils.py的连接参数如心跳间隔仍需重启容器才能更新——这是设计取舍连接参数属底层协议变更频率极低AI 开关属业务策略需动态调整。3.3 日志归集file_log_collector.py 不是轮询而是 inotify 实时监听file_log_collector.py用inotify替代传统cron轮询解决日志切割延迟问题。它监听./logs/目录的IN_MOVED_TO事件文件移动完成一旦检测到新日志文件如app-2024-06-15.log立即执行压缩归档# file_log_collector.py import inotify.adapters import subprocess import os def watch_logs(log_dir: str): i inotify.adapters.Inotify() i.add_watch(log_dir, maskinotify.constants.IN_MOVED_TO) for event in i.event_gen(yield_nonesFalse): (_, type_names, path, filename) event if IN_MOVED_TO in type_names and filename.endswith(.log): full_path os.path.join(log_dir, filename) # 使用 pigz多线程 gzip加速压缩 subprocess.run([pigz, -f, full_path], checkTrue) logger.info(fCompressed {filename}) if __name__ __main__: watch_logs(./logs)pigz是关键——实测压缩 100MB 日志gzip耗时 42 秒pigz -p 4仅需 11 秒。inotify的IN_MOVED_TO事件比IN_CREATE更可靠因为日志框架如RotatingFileHandler先写临时文件再原子重命名IN_MOVED_TO确保文件写入完成才触发压缩避免压缩半截日志。3.4 避坑Docker 部署的四个血泪经验现象容器启动后psql连接拒绝docker logs api显示Connection refused原因docker-compose-cn.yml中depends_on仅控制启动顺序不等待db服务就绪。PostgreSQL 启动需 5~10 秒初始化而api容器启动脚本entrypoint.sh立即执行alembic upgrade head此时 DB 未 ready。解决在entrypoint.sh开头添加健康检查循环#!/bin/sh # entrypoint.sh until pg_isready -h db -U user -d xianyu; do echo Waiting for PostgreSQL... sleep 2 done exec $现象WebSocket 连接频繁断开ws_utils.py日志显示Connection closed原因Docker 网络默认 MTU 为 1500而某些云厂商如腾讯云 CVM物理网卡 MTU 为 9000导致 TCP 分片丢失WebSocket 心跳包被丢弃。解决在docker-compose-cn.yml的api服务中添加sysctlssysctls: - net.ipv4.tcp_keepalive_time60 - net.ipv4.tcp_keepalive_intvl10 - net.ipv4.tcp_keepalive_probes6并设置容器 MTU 为 1400network_mode: bridge extra_hosts: - host.docker.internal:host-gateway现象AI 回复返回空字符串OpenAI API Key 测试正常原因ai_reply_engine.py中generate_contextual_reply()对输入消息做长度截断msg_content[:200]但闲鱼消息可能含 emoji 或特殊符号len()计算字节数而非字符数导致截断点落在 UTF-8 多字节字符中间引发UnicodeDecodeError异常被捕获后静默返回空。解决改用msg_content.encode(utf-8)[:200].decode(utf-8, errorsignore)安全截断。现象批量导入 Excel 规则后部分关键词不生效原因Excel 文件编码非 UTF-8常见于 Windows 记事本另存为pandas.read_excel()默认用openpyxl引擎对 GBK 编码文件解析失败空单元格被填充为nan规则匹配时nan nan为False导致跳过。解决在rules/import接口添加编码探测from chardet import detect def import_rules(file): raw file.read() encoding detect(raw).get(encoding, utf-8) df pd.read_excel(io.BytesIO(raw), engineopenpyxl, encodingencoding)4. 多账号管理不是“开多个进程”而是基于 asyncio 的资源配额与状态同步4.1 账号进程隔离每个账号一个 asyncio.Task而非多线程XianyuAutoAsync.py是账号监控主入口它不使用threading或multiprocessing而是纯asyncio任务调度# XianyuAutoAsync.py class AccountManager: def __init__(self): self.tasks {} # {account_id: asyncio.Task} self.status {} # {account_id: {ws_connected: bool, last_msg_time: float}} async def start_account(self, account_id: str): if account_id in self.tasks and not self.tasks[account_id].done(): logger.warning(fAccount {account_id} already running) return task asyncio.create_task(self._run_account(account_id)) self.tasks[account_id] task self.status[account_id] {ws_connected: False, last_msg_time: time.time()} async def _run_account(self, account_id: str): ws XianyuWebSocket(account_id) while True: try: await ws.connect() await ws.listen() # 阻塞直到断连 except Exception as e: logger.error(f[{account_id}] WS error: {e}) await asyncio.sleep(5) # 断连后等待 5 秒再重试asyncio.create_task()创建的任务共享同一事件循环内存占用仅为线程的 1/10实测 100 账号仅占 120MB RAM且无 GIL 争抢。self.status字典用threading.Lock保护但仅在GET /api/v1/accounts/status接口读取时加锁写操作如ws_connected更新在ws_utils.py的on_open()和on_close()回调中直接赋值避免锁竞争。4.2 批量操作的原子性/api/v1/accounts/batch/start不是并发启动而是队列串行前端点击“批量启动”时reply_server.py的batch_start_accounts()接口接收账号 ID 列表但并非for id in ids: await manager.start_account(id)并发执行——这会导致瞬间创建 50 个 WebSocket 连接触发闲鱼风控。实际采用asyncio.Queue限流# reply_server.py app.post(/api/v1/accounts/batch/start) async def batch_start_accounts(ids: List[str]): queue asyncio.Queue() for acc_id in ids: await queue.put(acc_id) # 启动 3 个消费者每 2 秒处理一个账号 consumers [asyncio.create_task(_consume_queue(queue)) for _ in range(3)] await asyncio.gather(*consumers) return {status: queued} async def _consume_queue(queue: asyncio.Queue): while not queue.empty(): acc_id await queue.get() await account_manager.start_account(acc_id) await asyncio.sleep(2) # 间隔 2 秒 queue.task_done()queue的maxsize10限制待处理队列长度await asyncio.sleep(2)确保连接错峰。实测 50 账号批量启动耗时约 100 秒50*2但成功率从并发模式的 31% 提升至 99.2%。4.3 实时状态同步用 Server-Sent Events (SSE) 替代轮询前端账号状态页/static/index.html不使用setInterval(() fetch(/status), 5000)轮询而是 SSE 长连接// index.html const eventSource new EventSource(/api/v1/accounts/sse); eventSource.onmessage (event) { const status JSON.parse(event.data); updateAccountStatus(status.account_id, status); };后端reply_server.py的/api/v1/accounts/sse接口用StreamingResponseapp.get(/api/v1/accounts/sse) async def sse_status(request: Request): async def event_generator(): while True: if await request.is_disconnected(): break # 从 account_manager.status 获取最新状态 yield fdata: {json.dumps(account_manager.get_status())}\n\n await asyncio.sleep(3) # 3秒推送一次 return StreamingResponse(event_generator(), media_typetext/event-stream)SSE 降低服务器负载100 个在线用户轮询需 100*200 20,000 QPSSSE 仅需 100 连接维持QPS 为 0只有推送。await request.is_disconnected()检测客户端断开及时清理资源。4.4 避坑多账号状态的三个典型故障现象部分账号显示“离线”但docker ps查看容器仍在运行原因XianyuAutoAsync.py的_run_account()任务因未捕获的异常如KeyError访问不存在的msg[content]退出但asyncio.Task的done()为True后未被AccountManager清理status字典仍保留旧状态。解决在_run_account()结尾添加finally块finally: if account_id in self.tasks: del self.tasks[account_id] if account_id in self.status: self.status[account_id][ws_connected] False现象批量停止账号后仍有消息被回复原因stop_account()方法仅取消asyncio.Task但ws_utils.py的listen()协程可能正在await ws.recv()取消信号需等待 recv 返回才生效期间新消息仍会进入on_message()。解决在XianyuWebSocket中添加self._should_stop标志on_message()开头检查def on_message(self, msg): if getattr(self, _should_stop, False): return # 正常处理stop_account()先设标志再取消任务。现象账号状态页刷新后所有账号状态变为“未知”原因SSE 连接断开后前端未重连eventSource对象被 GC但页面未监听onerror事件。解决前端添加重连逻辑eventSource.onerror () { console.log(SSE disconnected, reconnecting...); eventSource.close(); setTimeout(() { eventSource new EventSource(/api/v1/accounts/sse); }, 5000); };5. 安全防护不是“加个密码”而是 JWT 隔离、验证码防爆破、安全日志三重加固5.1 JWT 用户隔离每个用户数据库 schema 独立非共享表db_manager.py的init_db()不是创建单一数据库而是为每个用户动态创建 schema# db_manager.py async def init_user_schema(user_id: str, engine: AsyncEngine): async with engine.begin() as conn: await conn.execute(text(fCREATE SCHEMA IF NOT EXISTS user_{user_id})) # 表创建语句中指定 schema await conn.run_sync(Base.metadata.create_all, bindengine, tables[ User.__table__.tometadata(), Account.__table__.tometadata(), Rule.__table__.tometadata() ])User、Account、Rule模型的__table_args__指定 schemaclass User(Base): __tablename__ users __table_args__ {schema: user_{user_id}}实际部署时user_id作为路径参数传入确保user_abc123和user_def456的数据物理隔离。即使 SQL 注入得逞攻击者也只能访问当前 JWT 解析出的user_id对应 schema无法跨库查询。5.2 图形验证码不是简单captcha库而是 Redis 令牌绑定register.html的验证码由/api/v1/captcha接口生成但关键在captcha_utils.py的令牌绑定# captcha_utils.py def generate_captcha(redis_client: Redis, user_ip: str) - Tuple[str, str]: text random_string(4) # 令牌与 IP 绑定防暴力 token secrets.token_urlsafe(16) redis_client.setex(fcaptcha:{token}, 300, f{text}:{user_ip}) # 5分钟有效期 return token, generate_image(text) app.post(/api/v1/register) async def register(captcha_token: str, captcha_input: str, request: Request): ip request.client.host stored redis_client.get(fcaptcha:{captcha_token}) if not stored or not stored.decode().startswith(f{captcha_input}:): raise HTTPException(400, Invalid captcha) # 验证通过后删除令牌 redis_client.delete(fcaptcha:{captcha_token})stored.decode().startswith(f{captcha_input}:)是关键——它验证输入与存储的验证码匹配且:后的 IP 地址与当前请求 IP 一致防止令牌被截获后异地使用。Redis 的SETEX确保令牌 5 分钟自动过期。5.3 安全日志file_log_collector.py 的日志分级与审计字段file_log_collector.py不仅压缩日志还注入审计字段。logging配置中Formatter添加user_id和ip_address# logging_config.py LOGGING_CONFIG { formatters: { audit: { format: %(asctime)s | %(levelname)-8s | %(name)s | %(user_id)s | %(ip_address)s | %(message)s } }, handlers: { file: { class: logging.handlers.RotatingFileHandler, formatter: audit, filename: ./logs/audit.log, maxBytes: 10485760, backupCount: 5 } } }reply_server.py的中间件注入这些字段app.middleware(http) async def add_audit_info(request: Request, call_next): response await call_next(request) # 获取 JWT 中的 user_id token request.headers.get(Authorization, ).replace(Bearer , ) user_id decode_jwt(token).get(sub, anonymous) if token else anonymous # 记录审计日志 logger_audit.info(fRequest: {request.method} {request.url.path}, extra{user_id: user_id, ip_address: request.client.host}) return responseaudit.log包含所有敏感操作POST /api/v1/rules/import、DELETE /api/v1/accounts/123、PUT /api/v1/config/ai_enabled每条日志含user_id和ip_address满足等保 2.0 审计要求。5.4 避坑安全防护的三个致命疏漏现象JWT token 泄露后攻击者可永久访问原因config.py中JWT_SECRET_KEY硬编码为dev-secret-key且未提供.env文件覆盖机制。解决在entrypoint.sh中检查环境变量if [ -z $JWT_SECRET_KEY ]; then echo Error: JWT_SECRET_KEY not set exit 1 fidocker-compose-cn.yml通过environment注入environment: - JWT_SECRET_KEY${JWT_SECRET_KEY}现象图形验证码可被 OCR 自动识别原因generate_image()使用PIL绘制简单字体无扭曲、无噪点。解决集成captcha库的ImageCaptcha启用干扰线from captcha.image import ImageCaptcha image ImageCaptcha(width160, height60, font_sizes[20, 25, 30]) image.draw_noise_curve(draw, color0x000000) image.draw_noise_dots(draw, color0x000000, width1)现象/api/v1/login接口遭暴力破解无限尝试原因登录接口未做速率限制redis中未记录失败次数。解决在login路由中添加app.post(/api/v1/login) async def login(credentials: OAuth2PasswordRequestForm Depends()): ip request.client.host key flogin_fail:{ip} fail_count int(redis_client.get(key) or 0) if fail_count 5: raise HTTPException(429, Too many attempts, try again later) if not verify_password(credentials.password, hashed): redis_client.incr(key) redis_client.expire(key, 3600) # 1小时 raise HTTPException(401, Incorrect credentials) # 登录成功重置计数 redis_client.delete(key)6. 生产验证用真实闲鱼消息流压测确认五级优先级、AI 上下文、发货防重三件事真能跑通6.1 压测方案用locust模拟 50 买家并发发消息观测规则匹配耗时不用ab或wrk因为闲鱼消息是 WebSocket 流需模拟真实协议。本文还有配套的精品资源点击获取