Codeforces Bot开发实战：从零构建自动化竞赛助手
适用人群：已会用 Python 写脚本、想省掉“复制-粘贴-等待网页刷新”这些机械动作的中级玩家
目标：把一次提交从平均 5 分钟压到 20 秒以内，训练效率翻 3 倍不是梦。
1. 手动刷题到底慢在哪？
先放一组我本地统计的“裸人工”数据（样本 100 场不包括 Div.4）：
| 步骤 | 平均耗时 |
|---|---|
| 读完题意→本地写完 | 9.3 min |
| 登录网页→找到题目页 | 1.1 min |
| 复制代码→选语言→粘贴→点击 Submit | 0.9 min |
| 等待评测队列+回退看结果 | 3.1 min |
| 如果 WA→Debug→再次提交 | 循环 2-4 次 |
一次 AC 平均5.1 min都耗在了“网页交互”上；一天 15 题就是 75 分钟纯机械劳动。
把这部分自动化，等于每天白捡1 小时写题时间，长期复利非常夸张。
2. 技术选型：Requests vs. aiohttp
| 指标 | Requests（同步） | aiohttp（异步） |
|---|---|---|
| 单线程 100 次空体 GET | 18.7 s | 2.1 s |
| 并发 100 次 Submit 仿真 | 49 s | 5.3 s |
| CPU 占用 | 5 % | 7 % |
| 代码可读性 |
结论：
- 网络 I/O 占大头，异步直接 *10 倍 QPS；
- Codeforces 官方限速 1 req/s，但“并发+队列”能把等待重叠，整体体感提升；
- aiohttp 的
timeout=ClientTimeout(total=20)与connector=TCPConnector(limit=30)足够应付。
下面整套 Bot 就基于asyncio + aiohttp搭建。
3. 核心模块拆解
3.1 API 鉴权（OAuth2 简化版）
Codeforces 没走标准 OAuth2，而是“Cookie + CSRF Token”双校验，步骤如下：
- GET 登录页→解析
csrf_token - POST 账号密码+Token→拿
SESSIONCookie - 后续所有请求带
Cookie: SESSION=xxx+X-CSRF-Token头
关键代码（已加异常重试）：
async def fetch_csrf(client: aiohttp.ClientSession) -> str: async with client.get("/enter") as resp: text = await resp.text() if resp.status != 200: raise RuntimeError("Login page unreachable") match = re.search(r'name="csrf_token".value="(.+?)"', text) if not match: raise RuntimeError("CSRF not found") return match.group(1) async def login(client, user: str, pwd: str) -> None: token = await fetch_csrf(client) payload = {"csrf_token": token, "handleOrEmail": user, "password": pwd} async with client.post("/enter", data=payload, allow_redirects=False) as resp: if resp.status != 302: raise RuntimeError("Login failed") # SESSION 已自动落入 CookieJar会话保持机制：
- 全局复用同一个
aiohttp.ClientSession(connector=conn, headers=base_header)- 设置
cookie_jar=aiohttp.CookieJar()让框架自动维护 Cookie- 退出时
await client.close()保证 TCP 连接池干净释放
3.2 提交监控器（轮询优化）
评测状态接口/api/contest.status返回 JSON，官方建议间隔 ≥ 2 s。
但实测 1.2 s 也不会 429，为了稳，用“指数退避”：
async def wait_for_verdict(client, contest_id, sub_id): delay, cap = 1, 16 while True: await asyncio.sleep(delay) stat = await get_submission_status(client, contest_id, sub_id) if stat in ("TESTING", "PRETTEST"): delay = min(delay * 2, cap) continue return stat- 平均等待 3.4 轮、总耗时 9 s，比固定 2 s 无脑轮询快 30 %。
- 速率限制规避：全局
asyncio.Semaphore(4)控制并发，防止瞬间 20+ 请求把 IP 封掉。
3.3 代码自动生成器（Jinja2 模板）
把“读入模板→填充→写文件→提交”做成一条链：
from jinja2 import Environment, FileSystemLoader env = Environment(loader=FileSystemLoader("templates")) tpl = env.get_template("main.py") def gen_code(desc: dict) -> str: return tpl.render( mod=int(1e9+7), T=desc["tc"], data_struct=desc.get("struct", "") )模板示例（节选）：
import sys, math sys.setrecursionlimit(2000000) def solve(): data = sys.stdin.readline().strip() # {{ data_struct }} print(ans) if __name__ == "__main__": solve()4. 完整可运行骨架（含异常处理）
import aiohttp, asyncio, re, json, time, hmac, hashlib from aiohttp import ClientTimeout, TCPConnector class CfBot: def __init__(self, user: str, pwd: str): self.user = user self.pwd = pwd self.sem = asyncio.Semaphore(4) self.jar = aiohttp.CookieJar() timeout = ClientTimeout(total=20) connector = TCPConnector(limit=30, ttl_dns_cache=300) self.session = aiohttp.ClientSession( connector=connector, timeout=timeout, cookie_jar=self.jar, headers={"User-Agent": "cfbot/1.0"}, ) async def __aenter__(self): await login(self.session, self.user, self.pwd) return self async def __aexit__(self, exc_type, exc, tb): await self.session.close() async def submit(self, contest_id: int, prob: str, lang: str, code: str) -> int: async with self.sem: # 速率限制 url = f"/contest/{contest_id}/submit" token = await fetch_csrf(self.session) data = { "csrf_token": token, "ftaa": "", "bfaa": "", "action": "submitSolution", "contestId": str(contest_id), "submittedProblemIndex": prob, "programTypeId": lang, # 例如 54 对应 PyPy 3 "source": code, } async with self.session.post(url, data=data, allow_redirects=False) as resp: loc = resp.headers.get("Location", "") if "my" not in loc: raise RuntimeError("Submit rejected, probably rate limited") # 解析新提交 ID match = re.search(r"submission/(\d+)", loc) if not match: raise RuntimeError("Cannot parse submission id") return int(match.group(1)) async def verdict(self, contest_id: int, sub_id: int) -> str: api = f"https://codeforces.com/api/contest/{contest_id}/submission/{sub_id}" async with self.session.get(api) as resp: if resp.status != 200: raise RuntimeError("API unavailable") data = await resp.json() return data["result"]["verdict"] async def safe_submit(self, *args, **kw): """带重试的封装""" for attempt in range(1, 4): try: return await self.submit(*args, **kw) except RuntimeError as e: if attempt == 3: raise await asyncio.sleep(2 ** attempt) # 指数退避响应数据校验：
- 所有 JSON 先
json.loads()再对字段做pydantic.BaseModel校验，缺字段直接抛ValidationError，防止后面空指针。- 对 HTML 用
BeautifulSoup抽数据时，加if not node: raise ParseError快速失败。
5. 性能压测 & 重试有效性
用 Locust 写 30 虚拟用户，脚本如下：
from locust import HttpUser, task, between class CfUser(HttpUser): wait_time = between(1, 2) def on_start(self): self.client.post("/enter", data={...}) # 简写 @task def status(self): with self.client.get("/api/contest.status?contestId=1775", catch_response=True) as r: if r.status_code != 200: r.failure("Got 429")结果：
- RPS 峰值 22 → 触发 429，Bot 侧立刻退避，成功率 99.2 %；
- 关闭重试的对照组成功率仅 84 %，且被临时 Ban 10 min。
结论：指数退避 + Semaphore 把“误伤”降到可接受范围。
6. 避坑指南
反爬虫
- 必须带
User-Agent，最好再带Referer； - 同 IP 同账号 1 min 内别超 30 次提交；
- 若被 302 到
/blocked，立刻停 30 min 并弹警告。
- 必须带
竞赛规则合规
- Bot 只能在自己账号、自己写的代码上跑，替别人提交=封号；
- 正式赛期间（Running）很多接口会 403，提前判断
phase != FINISHED直接拒绝运行。
本地缓存雪崩
- 把题目列表、语言 ID 映射放
aiocache.Redis，TTL 6 h； - 加随机 jitter，防止重启瞬间几千并发把 Redis 打挂；
- 回退到本地 JSON 文件，保证 Redis 挂掉也能启动。
- 把题目列表、语言 ID 映射放
7. 效果实测
跑两周（Daily 20 题）：
| 指标 | 手工 | Bot |
|---|---|---|
| 平均提交间隔 | 5 min 8 s | 38 s |
| 等待评测焦虑 | 高 | 低（自动轮询） |
| 代码模板错误 | 3 次/周 | 0（统一模板） |
| 训练时长节省 | — | 每天 65 min |
提升≈ 300 %没吹牛，主要省的是“人肉网页操作”。
8. 还能怎么玩？——开放话题
如何集成大模型实现智能代码修复？
- 把 WA 的测试差例喂给 Codex / ChatGLM，让它输出修正补丁；
- 用抽象语法树对比“原函数 vs 补丁”，确认只改逻辑不改框架；
- 自动再次提交→循环直到 AC 或达到重试上限。
思路有了，但风险是可能违反“一人一份代码”原则。
你会怎么设计提示词、过滤策略，既提速又不踩线？欢迎留言一起脑洞。
