在中国国际时装周这类大型公开活动的执行现场,模特大面试通常不是一个小规模邀约,而是一个需要把大量候选人按时间、场次和组别快速编排的分组调度过程。像“新思维第二组”这样的标识,背后其实对应着一套完整的业务规则:谁报名了、谁通过初筛、谁被编进哪个测评组、组内顺序怎么排、每个评委打了多少分、最终成绩从哪条链路汇总。这个环节过去主要靠 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.sql:
PRAGMA 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 环境准备
最低环境如下:
| 依赖 | 版本建议 | 说明 |
|---|---|---|
| Python | 3.10 及以上 | 本文示例使用 3.10 语法 |
| Flask | 3.x | Web 框架 |
| SQLite | Python 内置 | 无需单独安装 |
安装依赖:
mkdir casting-demo cd casting-demo python -m venv .venv # Windows .venv\Scripts\activate # macOS / Linux source .venv/bin/activate pip install flask==3.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", encoding="utf-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(host="127.0.0.1", port=5000, debug=True)启动:
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 ".tables"4. 核心接口实现:从报名一路走到成绩汇总
4.1 批量导入候选人
报名阶段通常是运营人员手里已经有一份 Excel 或报名小程序导出的数据。最实用的接口是批量导入,而不是让运营在系统里一条条手工新增。
在 app.py 中加入导入接口:
@app.post("/api/candidates/import") def import_candidates(): db = get_db() payload = request.get_json(force=True) 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(force=True) 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 变为 ASSIGNED,group_member 中写入组内顺序 seq。组内顺序建议用 1、2、3 的连续序号,这对应现场叫号顺序,不需要和人名单行号强绑定。
4.4 现场签到与组状态流转
面试开始前,现场执行人员要把组状态从 WAIT 切到 ONGOING:
@app.post("/api/groups/<int:group_id>/status") def change_group_status(group_id): db = get_db() p = request.get_json(force=True) 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(force=True) 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(force=True) 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", type=int) 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