Flask+SQLite 实现模特大面试分组调度系统

在中国国际时装周这类大型公开活动的执行现场模特大面试通常不是一个小规模邀约而是一个需要把大量候选人按时间、场次和组别快速编排的分组调度过程。像“新思维第二组”这样的标识背后其实对应着一套完整的业务规则谁报名了、谁通过初筛、谁被编进哪个测评组、组内顺序怎么排、每个评委打了多少分、最终成绩从哪条链路汇总。这个环节过去主要靠 Excel 和人工叫号候选人一多问题就会集中爆发名单版本不一致、分组合并漏人、签到和评分脱节、最终结果需要反复对表。本文不讨论某场活动的具体安排而是把“模特大面试”抽象成一个通用线下面试管理需求用 Flask SQLite 实现一个可运行的最小闭环系统完整覆盖报名、分组、签到、评分、汇总五个环节。这套设计不只适用于时装周选角也可以套用到演员试镜、校园招聘面试、艺术类考试等任何按批次分组测评的业务场景。后面所有代码和示例数据都以工程演练为目的落地到真实活动时需要根据你的活动规模、字段和流程重新调整。1. 先把“候选人、组、评分”拆成一套状态机1.1 现场执行真正需要的数据不是花名册而是状态很多活动的第一步是收简历所以最容易做出来的功能是“候选人花名册”。但真实执行中运营人员更关心的是每个人的当前状态是刚报名还是已经被编入某一组还是已经签到还是面试结束。组的状态也一样是待开始、面试中还是已经结束并进入汇总。如果只用一张 Excel 表管理状态通常体现在三处名单里加一列“是否到场”、主持人手里有一张纸板、评委在评分表上手动填分数。三处数据各自维护就会出现经典的“名单不一致”问题。技术方案的价值不是把名单从 Excel 复制成网页而是让候选人状态、组状态、评分状态形成一条可查询、可回溯的链路。1.2 业务主流程与最小角色定义一个线下面试活动可以压缩成六个动作候选人报名进入待分组池。运营人员创建组别例如“新思维第一组”“新思维第二组”。系统按容量自动分配候选人并生成组内序号。现场主持人把组状态从“待开始”切到“面试中”。候选人在所属组签到评委按维度打分。组结束后汇总平均分生成组内排名。这里至少存在三类角色候选人不需要登录系统运营负责报名和分组现场执行人员负责签到评委只负责打分。第一版系统可以不引入完整登录体系但接口语义上要区分出操作场景否则后面加权限时会非常痛苦。1.3 为什么用“组”而不是直接排一个大队列大面试通常有多个评委、多个房间和多个时间段。如果所有候选人排成一条大队列任何一个迟到或临时取消都会导致整条流水线阻塞。分组的本质是把一个大流程切成多个可以并行的小流程。从“新思维第二组”这种命名也能看出组别既承担了时间编排功能也承担了地点编排功能。一组的候选人去 A 房间二组的候选人去 B 房间一组先面试二组后面试。因此在数据模型里组不能只是一个字符串它要有自己的容量、房间、时间和状态。2. 数据表设计先定状态再写接口2.1 四张表和它们解决的问题系统拆成四张核心表即可覆盖第一个可运行版本表名对应业务实体解决的核心问题candidate候选人存储报名信息interview_group面试组存储组别、容量、房间、时间group_member候选人与组的关联解决“一个人分配到哪个组”的问题score_record评委评分解决“每个评委给候选人打了多少分”的问题很多人会省略 group_member直接在 candidate 表里加一个 group_id 字段。这在候选人只参加一场面试时能跑通但一旦活动有多轮或者候选人需要先进入候选池再确认组别单字段方案就很别扭。中间表是更稳妥的选择。2.2 用 DDL 把规则固化下来建表 SQL 保存为 schema.sqlPRAGMA foreign_keys ON; CREATE TABLE IF NOT EXISTS candidate ( id INTEGER PRIMARY KEY AUTOINCREMENT, candidate_no TEXT NOT NULL UNIQUE, name TEXT NOT NULL, gender TEXT NOT NULL DEFAULT 女, height_cm INTEGER, city TEXT, agency TEXT, phone TEXT, source_channel TEXT, status TEXT NOT NULL DEFAULT PENDING, created_at TEXT NOT NULL DEFAULT (datetime(now, localtime)) ); CREATE TABLE IF NOT EXISTS interview_group ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_no TEXT NOT NULL UNIQUE, group_name TEXT NOT NULL, round_name TEXT NOT NULL DEFAULT 第一轮大面试, capacity INTEGER NOT NULL DEFAULT 20, room TEXT, interview_time TEXT, status TEXT NOT NULL DEFAULT WAIT ); CREATE TABLE IF NOT EXISTS group_member ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_id INTEGER NOT NULL REFERENCES interview_group(id), candidate_id INTEGER NOT NULL REFERENCES candidate(id), seq INTEGER NOT NULL DEFAULT 0, checkin_time TEXT, score_status TEXT NOT NULL DEFAULT UNSCORED, UNIQUE(group_id, candidate_id) ); CREATE TABLE IF NOT EXISTS score_record ( id INTEGER PRIMARY KEY AUTOINCREMENT, group_member_id INTEGER NOT NULL REFERENCES group_member(id), judge_no TEXT NOT NULL, dimension_look INTEGER NOT NULL DEFAULT 0, dimension_show INTEGER NOT NULL DEFAULT 0, dimension_express INTEGER NOT NULL DEFAULT 0, remark TEXT, created_at TEXT NOT NULL DEFAULT (datetime(now, localtime)), UNIQUE(group_member_id, judge_no) );这里有几个关键约束candidate_no 唯一作用是防止同一个人被重复导入。group_member 中 group_id 与 candidate_id 联合唯一作用是一个候选人在一个组里只出现一次。score_record 中 group_member_id 与 judge_no 联合唯一作用是同一个评委对同一个候选人只保留一条评分重复提交时执行更新而不是插入。2.3 字段取值与业务含义候选人的 status 字段建议用固定枚举不要自由输入取值含义PENDING已报名等待分配组ASSIGNED已分配进某个组CANCELLED已取消或未通过初筛组的 status 对应现场节奏取值含义WAIT待开始仍允许分配候选人ONGOING面试中停止分配允许签到END面试结束只读这样设计的好处是接口可以通过状态约束业务流程。比如组状态不是 ONGOING 时签到接口应该拒绝组状态已经是 END 时评分接口应该拒绝或至少弹警告。2.4 为什么要保留 group_member 中间表中间表把“候选人基本信息”和“候选人本次面试表现”分开。身高、机构、城市这些信息属于一次报名长期有效签到时间、组内序号、评分状态则属于某一次特定面试。两者混在一张表里一旦同一候选人参加多轮面试就要反复复制基础信息数据不一致的概率会显著上升。3. 用 Flask SQLite 搭一个最小可运行后端3.1 环境准备最低环境如下依赖版本建议说明Python3.10 及以上本文示例使用 3.10 语法Flask3.xWeb 框架SQLitePython 内置无需单独安装安装依赖mkdir casting-demo cd casting-demo python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate pip install flask3.0.33.2 项目结构第一版保持简单三个文件足够casting-demo/ ├── schema.sql # 建表语句 ├── app.py # Flask 接口 └── requirements.txt # 依赖数据库文件 casting.db 会在首次启动时自动生成不需要提交到代码仓库。3.3 数据库初始化逻辑在 app.py 中先封装数据库连接和建库逻辑import os import sqlite3 from datetime import datetime from flask import Flask, g, jsonify, request BASE_DIR os.path.dirname(os.path.abspath(__file__)) DATABASE os.path.join(BASE_DIR, casting.db) app Flask(__name__) def get_db(): if db not in g: conn sqlite3.connect(DATABASE) conn.row_factory sqlite3.Row conn.execute(PRAGMA foreign_keys ON) g.db conn return g.db app.teardown_appcontext def close_db(exc): db g.pop(db, None) if db is not None: db.close() def init_db(): conn sqlite3.connect(DATABASE) schema_path os.path.join(BASE_DIR, schema.sql) with open(schema_path, r, encodingutf-8) as f: conn.executescript(f.read()) conn.commit() conn.close()注意两个细节。第一连接后要设置row_factory sqlite3.Row这样查询结果可以像字典一样按列名访问。第二SQLite 默认不强制外键约束所以要执行一次PRAGMA foreign_keys ON否则 group_member 里的外键约束形同虚设。3.4 启动并确认健康检查在 app.py 末尾加上入口和健康检查接口app.get(/api/health) def health(): return jsonify({ok: True, app: casting-demo}) app.get(/) def index(): return jsonify({tips: visit /api/health}) if __name__ __main__: if not os.path.exists(DATABASE): init_db() app.run(host127.0.0.1, port5000, debugTrue)启动python app.py验证curl http://127.0.0.1:5000/api/health正常返回{app:casting-demo,ok:true}此时 casting.db 已经自动生成schema.sql 中的表也都已创建。可以用 sqlite3 客户端确认sqlite3 casting.db .tables4. 核心接口实现从报名一路走到成绩汇总4.1 批量导入候选人报名阶段通常是运营人员手里已经有一份 Excel 或报名小程序导出的数据。最实用的接口是批量导入而不是让运营在系统里一条条手工新增。在 app.py 中加入导入接口app.post(/api/candidates/import) def import_candidates(): db get_db() payload request.get_json(forceTrue) if not isinstance(payload, list): return jsonify({error: body must be list}), 400 row db.execute(SELECT MAX(candidate_no) AS max_no FROM candidate).fetchone() max_no row[max_no] if row and row[max_no] else C0000 next_seq int(max_no[1:]) 1 inserted [] skipped_count 0 for item in payload: name (item.get(name) or ).strip() if not name: skipped_count 1 continue candidate_no item.get(candidate_no) or C{:04d}.format(next_seq) try: db.execute( INSERT INTO candidate (candidate_no, name, gender, height_cm, city, agency, phone, source_channel, status) VALUES (?, ?, ?, ?, ?, ?, ?, ?, PENDING) , ( candidate_no, name, item.get(gender) or 女, item.get(height_cm), item.get(city), item.get(agency), item.get(phone), item.get(source_channel), ), ) inserted.append(candidate_no) except sqlite3.IntegrityError: # candidate_no 重复说明已经导入过 skipped_count 1 next_seq 1 db.commit() return jsonify({inserted: inserted, skipped_count: skipped_count})这里的编号生成逻辑适用于固定宽度编号比如 C0001、C0002。如果原始材料里已经有候选编号就直接使用如果为空系统再生成。candidate_no 命中唯一约束时采用跳过策略避免一次重复导入把整个批次全部回滚。4.2 创建场次分组在导入候选人之后需要先创建组再执行分配。创建组的接口如下app.post(/api/groups) def create_group(): db get_db() p request.get_json(forceTrue) group_no (p.get(group_no) or ).strip() group_name (p.get(group_name) or ).strip() if not group_no or not group_name: return jsonify({error: group_no and group_name required}), 400 try: cur db.execute( INSERT INTO interview_group (group_no, group_name, round_name, capacity, room, interview_time, status) VALUES (?, ?, ?, ?, ?, ?, WAIT) , ( group_no, group_name, p.get(round_name) or 第一轮大面试, int(p.get(capacity) or 20), p.get(room), p.get(interview_time), ), ) except sqlite3.IntegrityError: return jsonify({error: group_no duplicate}), 409 db.commit() return jsonify({id: cur.lastrowid, group_no: group_no})请求示例curl -X POST http://127.0.0.1:5000/api/groups \ -H Content-Type: application/json \ -d { group_no: G01, group_name: 新思维第一组, round_name: 第二轮大面试, capacity: 3, room: A204, interview_time: 2025-07-20 09:00 }返回结果为新建组的 id 和 group_no。后续所有操作都应使用数据库 id而不是组名字符串因为组名可能重复也可能被人为修改。4.3 自动分配候选人并生成组内顺序把所有 PENDING 状态的候选人均匀分配到所有 WAIT 状态的组是线下面试最常用的分配逻辑。这里使用轮询分配避免候选人集中进入第一组app.post(/api/distribute) def distribute(): db get_db() pending db.execute( SELECT id FROM candidate WHERE status PENDING ORDER BY id ).fetchall() groups db.execute( SELECT * FROM interview_group WHERE status WAIT ORDER BY id ).fetchall() if not pending: return jsonify({error: no pending candidate}), 400 if not groups: return jsonify({error: create WAIT group first}), 400 total_capacity sum(g[capacity] for g in groups) if len(pending) total_capacity: return jsonify({error: capacity not enough}), 400 group_index 0 assigned_count {g[id]: 0 for g in groups} assigned [] for cand in pending: # 轮询尝试每个组找到一个还有容量的组 for _ in range(len(groups)): group groups[group_index % len(groups)] group_index 1 if assigned_count[group[id]] group[capacity]: assigned_count[group[id]] 1 seq assigned_count[group[id]] db.execute( INSERT INTO group_member (group_id, candidate_id, seq) VALUES (?, ?, ?), (group[id], cand[id], seq), ) db.execute( UPDATE candidate SET status ASSIGNED WHERE id ?, (cand[id],), ) assigned.append( { candidate_id: cand[id], group_id: group[id], seq: seq, } ) break db.commit() return jsonify({assigned: assigned})分配完成后candidate 的 status 变为 ASSIGNEDgroup_member 中写入组内顺序 seq。组内顺序建议用 1、2、3 的连续序号这对应现场叫号顺序不需要和人名单行号强绑定。4.4 现场签到与组状态流转面试开始前现场执行人员要把组状态从 WAIT 切到 ONGOINGapp.post(/api/groups/int:group_id/status) def change_group_status(group_id): db get_db() p request.get_json(forceTrue) status p.get(status) if status not in (WAIT, ONGOING, END): return jsonify({error: invalid status}), 400 cur db.execute( UPDATE interview_group SET status ? WHERE id ?, (status, group_id), ) if cur.rowcount 0: return jsonify({error: group not found}), 404 db.commit() return jsonify({id: group_id, status: status})组进入 ONGOING 后候选人才能签到app.post(/api/checkin) def checkin(): db get_db() p request.get_json(forceTrue) candidate_id p.get(candidate_id) group_id p.get(group_id) if not candidate_id or not group_id: return jsonify({error: candidate_id and group_id are required}), 400 member db.execute( SELECT gm.id, gm.checkin_time, g.status AS group_status FROM group_member gm JOIN interview_group g ON g.id gm.group_id WHERE gm.candidate_id ? AND gm.group_id ? , (candidate_id, group_id), ).fetchone() if not member: return jsonify({error: candidate not in this group}), 404 if member[group_status] ! ONGOING: return jsonify({error: group must be ONGOING}), 400 if member[checkin_time]: return jsonify( { warning: already checked in, checkin_time: member[checkin_time], } ), 200 now datetime.now().strftime(%Y-%m-%d %H:%M:%S) db.execute( UPDATE group_member SET checkin_time ? WHERE id ?, (now, member[id]), ) db.commit() return jsonify({id: member[id], checkin_time: now})签到必须绑定 candidate_id 和 group_id不能只传候选人编号。原因是候选人可能存在于多个轮次不指定组就不知道是在哪个房间、哪个场次签到。重复签到不报错而是返回已有的签到时间这样现场口头询问“他是不是已经签过了”时系统能给出准确答案。4.5 评委评分录入现场评委通常有多个每个人对同一个候选人打一组分。接口按group_member_id judge_no作为唯一评分记录app.post(/api/scores) def add_score(): db get_db() p request.get_json(forceTrue) required [candidate_id, group_id, judge_no] for key in required: if key not in p: return jsonify({error: f{key} required}), 400 member db.execute( SELECT gm.id FROM group_member gm JOIN interview_group g ON g.id gm.group_id WHERE gm.candidate_id ? AND gm.group_id ? , (p[candidate_id], p[group_id]), ).fetchone() if not member: return jsonify({error: candidate not in this group}), 404 # 每个维度 0-10 分超出范围时截断到边界 dimension_look min(max(int(p.get(dimension_look, 0)), 0), 10) dimension_show min(max(int(p.get(dimension_show, 0)), 0), 10) dimension_express min(max(int(p.get(dimension_express, 0)), 0), 10) try: db.execute( INSERT INTO score_record (group_member_id, judge_no, dimension_look, dimension_show, dimension_express, remark) VALUES (?, ?, ?, ?, ?, ?) , ( member[id], p[judge_no], dimension_look, dimension_show, dimension_express, p.get(remark), ), ) except sqlite3.IntegrityError: # 同一评委重复提交按一次更新处理 db.execute( UPDATE score_record SET dimension_look ?, dimension_show ?, dimension_express ?, remark ? WHERE group_member_id ? AND judge_no ? , ( dimension_look, dimension_show, dimension_express, p.get(remark), member[id], p[judge_no], ), ) db.commit() return jsonify({ok: True})这里的维度只做示例实际活动可以换成“台步表现”“镜头表现”“表达沟通”。维度名称变了但表的纵向结构不需要变新增维度只是新增一个字段或一个子表的问题。0 到 10 的边界校验看起来简单却能防止评委误输入 100 分导致平均分完全失真。4.6 分组结果汇总最后一环是按组汇总候选人平均分并按照分数倒序排列。没有评分的候选人排在最后方便运营区分“已评分”和“待评分”app.get(/api/results) def get_results(): db get_db() group_id request.args.get(group_id, typeint) if not group_id: return jsonify({error: group_id query param required}), 400 rows db.execute( SELECT gm.seq, c.candidate_no, c.name, c.gender