Python+Flask与SQLite实现OnlineJudge过题数统计与排行榜
简介基于 Python 与 Flask 框架实现的一套轻量级在线判题过题数统计网站项目主要面向高校计算机类毕业设计或课程作业场景适合需要掌握网站开发、数据采集与可视化展示的开发者参考学习。整个压缩包共包含六十一个文件体积约八百六十二千字节其中有十余个 Python 程序文件负责网站后台核心逻辑与爬虫抓取任务还包含多个前端页面模板、样式表与交互脚本用于结果查询、数据统计和排行展示另附容器化部署所需的配置文件、依赖清单及说明文档便于快速搭建运行环境。在功能上项目内置可扩展的爬虫模块能对接 Codeforces、洛谷、POJ 等知名在线判题平台自动获取用户的过题数量并完成统计与排行同时实现了用户登录、题目维护、提交结果反馈、数据库存储等配套模块整体代码结构清晰符合课程设计与毕业设计的要求。目前已有八十二人学习使用对于想深入理解 Flask 开发流程、爬虫编写以及前后端联调的读者来说是一份不错的实战参考资料。1. 轻量级 OnlineJudge 过题数统计网站本质是一个去重计数问题先把口径定死用 Python 和 Flask 做 OnlineJudge 过题数统计听起来就是写几个页面然后连数据库但真正动手后几乎所有返工都出在“口径”上同一个用户在同一道题上提交十次、只有一次 Accepted过题数算 1 还是算 10同一道题在不同 OJ 平台里题号不一样要不要合并OJ 返回的时间戳是 UTC统计“近 30 天过题趋势”要不要先转时区。这些问题在写第一个路由之前不定下来后面每加一个查询就要回头改表结构。本文按一条课程设计和毕设里最常见的实现路线展开Python 抓取 OJ 提交记录清洗成统一字段写入 SQLite再用 Flask 把聚合结果渲染成排行榜和个人统计页。适合正在做课程作业的学生也适合想给团队内部做一个轻量刷题看板的工程师。2. 先用 SQLite 立住三个表过题数统计的数据模型怎么设计2.1 users、problems、submissions 三张表各管什么轻量方案的存储首选 SQLite零配置文件、数据库就是一个 .db 文件、Python 标准库自带 sqlite3 驱动完全匹配 Flask 项目的交付方式。在按页面写代码前先把表结构确定下来。常见的设计是维度表加事实表users 管人problems 管题submissions 管每一次提交。CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, nickname TEXT DEFAULT , class_name TEXT DEFAULT ); CREATE TABLE IF NOT EXISTS problems ( id INTEGER PRIMARY KEY AUTOINCREMENT, oj_source TEXT NOT NULL, problem_key TEXT NOT NULL, title TEXT DEFAULT , url TEXT DEFAULT , UNIQUE (oj_source, problem_key) ); CREATE TABLE IF NOT EXISTS submissions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, problem_id INTEGER NOT NULL, status TEXT NOT NULL, submitted_at TEXT NOT NULL, remote_id TEXT NOT NULL, UNIQUE (remote_id, user_id), FOREIGN KEY (user_id) REFERENCES users(id), FOREIGN KEY (problem_id) REFERENCES problems(id) );逻辑说明submissions 是核心事实表users 和 problems 是维度表。status 字段存 OJ 原始判定字符串而不是只存 0/1 布尔值这样以后想统计“错误提交次数”“通过率”或“编译错误分布”时可以直接从已有数据算不用重新抓一遍。参数说明problems 表用UNIQUE (oj_source, problem_key)做联合唯一约束。oj_source 表示数据来源是哪个平台problem_key 表示该平台里的题号。不同 OJ 里题目没有人会强行统一编号最多在展示层加一个映射这样设计对外部数据源的兼容性最好。submissions 表用UNIQUE (remote_id, user_id)去重remote_id 是 OJ 返回的提交 ID。因为部分老式 OJ 的分页接口会把同一提交 ID 在不同查询条件下重复返回加上 user_id 组成联合唯一键后入库时就能安全忽略重复记录。2.2 过题数去重查询的关键写法过题数的本质是“按用户去重后的题目数”。同一个问题被同一用户多次 Accepted只能贡献一次计数。SQL 里的标准写法是 COUNT(DISTINCT problem_id)。SELECT u.username, COUNT(DISTINCT s.problem_id) AS solved_count FROM submissions s JOIN users u ON u.id s.user_id WHERE s.status IN (accepted, Accepted, AC) GROUP BY u.id ORDER BY solved_count DESC;逻辑说明COUNT(DISTINCT s.problem_id) 保证同一个人同一道题只出现一次WHERE 里的状态条件把通过判定圈定住。要注意的是 status 字段如果入库前没有做归一化查询时就得把各种“通过”的写法都列出来。更稳妥的做法是在抓取清洗阶段就把状态统一成小写英文查询条件只需要写s.status accepted这一行。这样设计之后可以顺手加一个索引排行榜页和个人详情页的查询都会用到它CREATE INDEX idx_submissions_user_status ON submissions(user_id, status);这个索引把 user_id 和 status 组合起来SQLite 在过滤“某个用户的 accepted 提交”时能直接走索引数据量到十几万条仍然能保持毫秒级响应。2.3 平台维度与班级维度怎么扩展课程作业里经常出现两个额外需求按平台看用户过题数按班级看排名。这两个都不需要改表结构。按平台看对 problems 表做一次 JOIN再用 GROUP BY 把 oj_source 带出来SELECT p.oj_source, COUNT(DISTINCT s.problem_id) AS solved_count FROM submissions s JOIN problems p ON p.id s.problem_id WHERE s.user_id 1 AND s.status accepted GROUP BY p.oj_source;按班级看users 表里已经有 class_name 字段WHERE 条件加u.class_name ?即可。需要记住的原则是统计维度能通过现有字段组合出来就不要再单独建表除非业务真的需要把多个 OJ 里同一道题合并成一个逻辑题目——那种情况才需要加一张 problem_alias 映射表轻量项目一般不建议做。3. Python 抓 OJ 提交记录公开 API、模拟登录和状态归一化3.1 三种数据源接入方式怎么选不同 OJ 对外暴露数据的方式大致分三类。第一类是带公开 API 的比如 Codeforces 一类的平台直接请求接口拿 JSON第二类需要模拟登录后抓页面常见于校内 OJ 和不少国产老牌 OJ第三类是给数据库直查权限或提供导出文件一般在比赛管理后台才有。毕设和课程作业里最常遇到的是前两类。接入方式的选型标准只有一条优先选能拿到“结构化字段”的方案。公开 API 的 JSON 数据天然带提交 ID、判定结果、时间戳解析成本最低模拟登录抓页面虽然要写选择器但好在绝大多数 OJ 页面都有一张提交记录表格BeautifulSoup 解析成本也不高。真正不建议做的是用 Selenium 驱动浏览器去点页面慢而且容易被 OJ 的登录验证机制拦住。3.2 公开 API 拉取提交列表的最小实现requests 是这类任务的标配。以下代码按分页循环拉取某个用户的提交记录并做了一个粗略的翻页保护。import requests def fetch_user_submissions(api_url: str, handle: str, per_page: int 200) - list[dict]: params {handle: handle, count: per_page} all_records [] for offset in range(0, 1000, per_page): params[from] offset 1 params[count] per_page resp requests.get(api_url, paramsparams, timeout10) resp.raise_for_status() payload resp.json() if payload.get(status) ! OK: break batch payload.get(result, []) all_records.extend(batch) if len(batch) per_page: break return all_records逻辑说明用 from/count 参数模拟“从第几条开始取”每次取 200 条直到某次返回不足 200 条就停止。有些 OJ 接口用 page 参数而不是 from改成params[page] offset // per_page 1即可整体循环结构不用变。参数说明timeout10 是 requests 的连接加读取超时时间。设成 3 秒在 OJ 响应慢时容易误判失败设成 30 秒又会让一次同步卡住太久。10 秒是一个平衡值。注意这里没有做重试真实场景里网络抖动很常见建议对 requests 的异常做一次指数退避重试重试间隔从 1 秒翻倍到 4 秒超过 3 次就放弃该页并把失败的 handle 记到日志里。3.3 模拟登录抓页面的最小方案没有公开 API 时requests.Session 配合 BeautifulSoup 是一套最稳定的组合。Session 会保持登录后的 cookie后续请求不需要重复登录。from bs4 import BeautifulSoup import requests session requests.Session() login_resp session.post( https://oj.example.com/login, data{username: your_account, password: your_password}, timeout10, ) login_resp.raise_for_status() resp session.get( https://oj.example.com/submissions?uid1001, timeout10, ) resp.encoding resp.apparent_encoding soup BeautifulSoup(resp.text, html.parser) records [] for row in soup.select(table#submission tbody tr): cols [td.get_text(stripTrue) for td in row.find_all(td)] if len(cols) 6: continue records.append({ remote_id: cols[0], problem_key: cols[2], status: cols[4], submitted_at: cols[5], })逻辑说明不同 OJ 的页面结构只差在 select 选择器和列顺序上主体逻辑一致。建议先保存一份真实页面的 HTML 到本地用解析脚本离线调试调通后再连线上环境。注意resp.encoding那两行很多 OJ 页面没有明确声明编码不手动设置就会出现中文全部乱码、后面正则匹配不到关键词的问题。这里有一个容易踩的坑页面里可能出现“Wrong Answer”这种带空格的判定td.get_text(stripTrue) 会把空格压缩成单个普通空格而 MySQL 查询时又要做归一化。所以解析阶段不要急着把原始字符串直接入库做完下一小节的归一化再写进 SQLite。3.4 状态归一化和脏数据容忍策略各 OJ 对“通过”的写法五花八门Accepted、AC、Accepted (AC)、正确。抓取阶段统一转小写后映射成标准状态后续所有查询只认这一套标准值。STATUS_ALIAS { accepted: accepted, ac: accepted, accept: accepted, correct: accepted, 正确: accepted, wronganswer: wrong_answer, wrong answer: wrong_answer, wa: wrong_answer, compileerror: compile_error, compile error: compile_error, ce: compile_error, } def normalize_status(raw: str) - str: key raw.strip().lower().replace( , ) key key.replace( , ) return STATUS_ALIAS.get(key, other)逻辑说明先把所有连续空格去掉避免 “Wrong Answer” 和 “WrongAnswer” 匹配不到同一个键再查映射表。查不到的状态统一落成 other而不是抛异常。这样设计是因为抓取任务通常是批量跑几十个用户某一条状态解析不出来就中断整个任务会让所有用户的数据都停在旧版本上。宁可让这一条记录进不了过题统计也要保证同步任务继续往下走。4. Flask 路由、SQL 聚合和模板渲染把过题数变成可读的统计页4.1 路由划分与 SQLite 连接管理Flask 项目的路由划分按“页面”来最直观。排行榜一个页面个人详情一个页面再加一个手动触发同步的入口三个路由即可。数据库连接用 Flask 的 g 对象做请求级管理如下所示from flask import Flask, g, render_template import sqlite3 app Flask(__name__) DATABASE oj_stats.db def get_db(): if db not in g: g.db sqlite3.connect(DATABASE) g.db.row_factory sqlite3.Row g.db.execute(PRAGMA journal_modeWAL) return g.db app.teardown_appcontext def close_db(exc): db g.pop(db, None) if db is not None: db.close()逻辑说明SQLite 默认是读共享、写独占打开 WAL 模式后读写可以并行Flask 在开发模式下同时有多个请求进来时不会频繁报 database is locked。row_factory 设为 sqlite3.Row查询结果就可以用row[username]这种方式取值返回给模板时也更安全。4.2 排行榜聚合查询过题数和总提交一次算清排行榜页面的信息量一般只要求两列用户名、过题数。但总提交次数是很容易被顺带问到的数据与其以后加不如在第一条 SQL 里直接把两个指标都查出来。app.route(/) def index(): db get_db() rows db.execute( SELECT u.username, COUNT(DISTINCT s.problem_id) AS solved, COUNT(s.id) AS total_submissions FROM submissions s JOIN users u ON u.id s.user_id GROUP BY u.id ORDER BY solved DESC, total_submissions ASC ).fetchall() return render_template(index.html, rowsrows)注意这里没有在 WHERE 里过滤 status而是把所有提交都统计进 total_submissions。逻辑说明total_submissions 想要的是“提交次数”这个事实不管是否通过都要算进去solved 通过 COUNT(DISTINCT problem_id) 单独控制只有当某条提交是 accepted 时 DISTINCT 才会把这个题目算进去。等等这样写有一个逻辑漏洞solved 直接统计所有题目的去重数没有限定 statusaccepted那么这个值会把 WA 也带上。修正方法是用条件聚合COUNT(DISTINCT CASE WHEN s.status accepted THEN s.problem_id END) AS solved, COUNT(s.id) AS total_submissions逻辑说明CASE WHEN 只对 accepted 的提交保留 problem_id非 accepted 行返回 NULLCOUNT(DISTINCT) 会自动跳过 NULL。这样 solved 的语义就是“被 Accepted 过的不同题目数”与 total_submissions 互不干扰。ORDER BY solved DESC, total_submissions ASC 的含义是过题数相同的人提交次数少的排前面这符合排行榜的常规习惯。4.3 个人详情页与近 30 天过题趋势个人详情页在排行榜点击用户名进入展示该用户每题通过情况外加一个趋势统计。趋势部分用 substr 把 submitted_at 的 ISO 字符串截成“年-月-日”再按天做分组app.route(/user/username) def user_detail(username: str): db get_db() trend db.execute( SELECT substr(s.submitted_at, 1, 10) AS day, COUNT(DISTINCT s.problem_id) AS solved FROM submissions s JOIN users u ON u.id s.user_id WHERE u.username ? AND s.status accepted AND s.submitted_at date(now, -30 day) GROUP BY day ORDER BY day , (username,)).fetchall() return render_template(user_detail.html, usernameusername, trendtrend)逻辑说明date(now, -30 day) 是 SQLite 内置的日期函数直接把系统日期往前推 30 天。submitted_at 在入库时统一存成 ISO 8601 字符串比如 2025-03-16T10:22:31substr 取前 10 位就得到日期部分。如果没有统一格式这里会混杂各种分隔符导致分组错乱——这就是 2.3 节坚持规范时间字段的回报。4.4 Jinja2 模板中的数据渲染约定模板里的原则只有一条只做展示不做运算。排行榜页面渲染时过题数和总提交次数都已经是 SQL 算好的数字模板只负责循环输出table thead tr th排名/th th用户名/th th过题数/th th总提交/th /tr /thead tbody {% for row in rows %} tr td{{ loop.index }}/td td a href{{ url_for(user_detail, usernamerow[username]) }} {{ row[username] }} /a /td td{{ row[solved] }}/td td{{ row[total_submissions] }}/td /tr {% endfor %} /tbody /table逻辑说明url_for 根据路由函数名生成链接会自动处理用户名里的特殊字符。如果这里图省事直接拼href/user/{{ row[username] }}用户名里出现斜杠或问号时会出现路由错乱甚至被当作参数传递。Jinja2 模板里也不建议调用自定义的统计方法模板加载慢是小事更麻烦的是视图函数的返回结构一变模板里依赖的字段名容易静默失效。5. 增量更新与统计自检轻量方案最后要补的两个动作5.1 用 ON CONFLICT 做增量写入避免全表重建第一次同步成功后后续刷新应该只做增量更新。直接删表重建会让排行榜在刷新期间出现空白而且每次全量抓取的请求量大容易触发 OJ 限流。SQLite 的 upsert 写法可以优雅解决这个问题INSERT INTO submissions (user_id, problem_id, status, submitted_at, remote_id) VALUES (?, ?, ?, ?, ?) ON CONFLICT (remote_id, user_id) DO UPDATE SET status excluded.status, submitted_at excluded.submitted_at;逻辑说明remote_id 和 user_id 组成唯一键重复的提交记录会走 DO UPDATE 分支把 status 和 submitted_at 刷新成最新值。这样同一道题从 WA 变 AC 时过题数会自动增长而不会插入一条新记录造成重复计数。配合 Flask CLI 命令把同步动作做成可手动调用的入口app.cli.command(sync-oj) click.option(--handle, default) def sync_oj(handle: str): 拉取指定用户的 OJ 提交记录并写入 SQLite。 records fetch_and_normalize(handle) db get_db() db.executemany( INSERT INTO submissions (user_id, problem_id, status, submitted_at, remote_id) VALUES (?, ?, ?, ?, ?) ON CONFLICT (remote_id, user_id) DO UPDATE SET status excluded.status , records) db.commit()逻辑说明executemany 批量写入比一条条 execute 快一个数量级。同步命令跑完后排行榜再刷新一次就能看到新数据不需要重启 Flask。5.2 统计自检的验证技巧增量更新做完最后一步是用一个小脚本验证统计口径是否符合预期。构造三个手工记录跑一遍排行榜查询然后断言结果sqlite3 oj_stats.db INSERT INTO submissions (user_id, problem_id, status, submitted_at, remote_id) VALUES (1, 1, accepted, 2025-03-01T10:00:00, r1), (1, 1, wrong_answer, 2025-03-01T11:00:00, r2), (1, 2, accepted, 2025-03-02T10:00:00, r3); SELECT COUNT(DISTINCT problem_id) FROM submissions WHERE user_id 1 AND status accepted; 期望输出是 2代表两道不同题目被 Accepted另外一条 WA 不影响结果。再把 r1 的 remote_id 改成 r3 的 remote_id 重新同步确认 upsert 后过题数仍然为 2。执行完这组验证后把它写成 pytest 测试以后抓取逻辑改动时就不会在不知不觉中破坏统计口径。整条本地验证链路跑通后再接入 linux 服务器上的 cron 定时任务或者 Windows 计划任务这个轻量的 OnlineJudge 统计网站就具备交给老师或同学试用的完整度了。本文还有配套的精品资源点击获取