import sqlite3 from contextlib import contextmanager from datetime import UTC, datetime, timedelta from html import escape from io import BytesIO from typing import Annotated, cast import jwt import pandas as pd from fastapi import Depends, FastAPI, File, HTTPException, Query, Request, UploadFile from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, StreamingResponse from pydantic import BaseModel, BeforeValidator, InstanceOf SECRET_KEY = 'QET5hv3A@148fefo#5#W5x#aF%9Y#8!2' ALGORITHM = 'HS256' ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=['*'], allow_credentials=True, allow_methods=['*'], allow_headers=['*'], ) # ==================== 数据库配置 ==================== DB_PATH = 'score_system.db' @contextmanager def get_db(write=False): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute('PRAGMA journal_mode=WAL') conn.execute('PRAGMA busy_timeout=5000') conn.execute('PRAGMA foreign_keys=ON') if write: conn.execute('BEGIN IMMEDIATE') try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def init_db(): with get_db() as conn: cursor = conn.cursor() cursor.execute( """CREATE TABLE IF NOT EXISTS batches (id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, created_at TIMESTAMP)""" ) cursor.execute( """CREATE TABLE IF NOT EXISTS programs (sn TEXT PRIMARY KEY, name TEXT NOT NULL, unit TEXT, performer TEXT, category TEXT, batch_id INTEGER NOT NULL, order_num INTEGER, created_at TIMESTAMP, UNIQUE(batch_id, order_num), FOREIGN KEY (batch_id) REFERENCES batches(id) ON DELETE CASCADE ON UPDATE CASCADE)""" ) cursor.execute( """CREATE TABLE IF NOT EXISTS judges (id TEXT PRIMARY KEY, name TEXT NOT NULL, order_num INTEGER, created_at TIMESTAMP)""" ) cursor.execute( """CREATE TABLE IF NOT EXISTS scores (id INTEGER PRIMARY KEY AUTOINCREMENT, judge_id TEXT NOT NULL, program_sn TEXT NOT NULL, content_score REAL, stage_score REAL, art_score REAL, overall_score REAL, admin_total_score REAL, comment TEXT, updated_at TIMESTAMP, UNIQUE(judge_id, program_sn), FOREIGN KEY (program_sn) REFERENCES programs(sn) ON DELETE CASCADE ON UPDATE CASCADE, FOREIGN KEY (judge_id) REFERENCES judges(id) ON DELETE CASCADE ON UPDATE CASCADE)""" ) cursor.execute( """CREATE TABLE IF NOT EXISTS admins (id TEXT PRIMARY KEY, username TEXT UNIQUE NOT NULL, password TEXT NOT NULL, created_at TIMESTAMP)""" ) cursor.execute( """CREATE TABLE IF NOT EXISTS settings (key TEXT PRIMARY KEY, value TEXT)""" ) cursor.execute('SELECT COUNT(*) FROM admins') if cursor.fetchone()[0] == 0: cursor.execute( 'INSERT INTO admins (id, username, password, created_at) VALUES (?, ?, ?, ?)', ('admin_001', 'admin', 'admin123', datetime.now()), ) cursor.execute("SELECT COUNT(*) FROM settings WHERE key = 'main_title'") if cursor.fetchone()[0] == 0: cursor.execute( "INSERT INTO settings (key, value) VALUES ('main_title', '文化艺术展评'), ('sub_title', '文艺汇演')" ) cursor.execute('SELECT COUNT(*) FROM batches') if cursor.fetchone()[0] == 0: batches_data = [] cursor.executemany( 'INSERT INTO batches (name, created_at) VALUES (?, ?)', batches_data, ) cursor.execute('SELECT COUNT(*) FROM judges') if cursor.fetchone()[0] == 0: judges_data = [] cursor.executemany( 'INSERT INTO judges (id, name, order_num, created_at) VALUES (?, ?, ?, ?)', judges_data, ) cursor.execute("SELECT COUNT(*) FROM settings WHERE key = 'award_first'") if cursor.fetchone()[0] == 0: cursor.execute( "INSERT INTO settings (key, value) VALUES ('award_first', '3'), ('award_second', '6'), ('award_third', '10')" ) cursor.execute("SELECT COUNT(*) FROM settings WHERE key = 'award_first_name'") if cursor.fetchone()[0] == 0: cursor.execute( "INSERT INTO settings (key, value) VALUES ('award_first_name', '一等奖'), ('award_second_name', '二等奖'), ('award_third_name', '三等奖')" ) init_db() def create_access_token(data: dict): to_encode = data.copy() expire = datetime.now(UTC) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({'exp': expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) def verify_token(token: str): try: return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) except jwt.PyJWTError as exc: raise HTTPException(status_code=401, detail='Token 无效') from exc async def require_admin(request: Request): token = request.headers.get('Authorization', '').replace('Bearer ', '') payload = verify_token(token) if payload.get('role') != 'admin': raise HTTPException(status_code=403, detail='需要管理员权限') return payload # ==================== Pydantic模型 ==================== def escape_html(value: str) -> str: if value is None: return '' return escape(str(value).strip()) EscapedStr = Annotated[str, BeforeValidator(escape_html)] class AdminLogin(BaseModel): username: EscapedStr password: EscapedStr class JudgeLogin(BaseModel): judge_id: EscapedStr judge_name: EscapedStr class JudgeCreate(BaseModel): id: EscapedStr name: EscapedStr class JudgeUpdate(BaseModel): name: EscapedStr class ProgramCreate(BaseModel): sn: EscapedStr name: EscapedStr unit: EscapedStr performer: EscapedStr category: EscapedStr batch_id: int class ProgramUpdate(BaseModel): name: EscapedStr unit: EscapedStr performer: EscapedStr category: EscapedStr batch_id: int class BatchCreate(BaseModel): name: EscapedStr class ScoreDetailUpdate(BaseModel): program_sn: EscapedStr field: EscapedStr value: float | None = None class ScoreBatchUpdate(BaseModel): program_sn: EscapedStr judge_scores: dict[EscapedStr, float | None] class CommentUpdate(BaseModel): program_sn: EscapedStr comment: EscapedStr class AwardSettings(BaseModel): first: int second: int third: int class AwardNamesSettings(BaseModel): first_name: EscapedStr second_name: EscapedStr third_name: EscapedStr class TitleSettings(BaseModel): main_title: EscapedStr sub_title: EscapedStr # ==================== 根路径 ==================== @app.get('/') async def root(): return HTMLResponse(content=HTML_TEMPLATE) # ==================== 管理员登录API ==================== @app.post('/api/admin/login') async def admin_login(login_data: AdminLogin): with get_db() as conn: cursor = conn.cursor() cursor.execute( 'SELECT id, username FROM admins WHERE username = ? AND password = ?', (login_data.username, login_data.password), ) admin = cursor.fetchone() if not admin: raise HTTPException(status_code=401, detail='管理员用户名或密码错误') token = create_access_token( {'id': admin['id'], 'username': admin['username'], 'role': 'admin'} ) return { 'success': True, 'admin_id': admin['id'], 'username': admin['username'], 'token': token, } @app.get('/api/admin/check') async def check_admin(admin_id: str): with get_db() as conn: cursor = conn.cursor() cursor.execute('SELECT id FROM admins WHERE id = ?', (admin_id,)) exists = cursor.fetchone() is not None return {'exists': exists} @app.post('/api/judge/login') async def judge_login(login_data: JudgeLogin): with get_db() as conn: cursor = conn.cursor() cursor.execute( 'SELECT id, name FROM judges WHERE id = ? AND name = ?', (login_data.judge_id, login_data.judge_name), ) judge = cursor.fetchone() if not judge: raise HTTPException(status_code=401, detail='评委编号或姓名错误') return {'success': True, 'judge_id': judge['id'], 'judge_name': judge['name']} @app.get('/api/judge/check') async def check_judge(judge_id: str): with get_db() as conn: cursor = conn.cursor() cursor.execute('SELECT id FROM judges WHERE id = ?', (judge_id,)) exists = cursor.fetchone() is not None return {'exists': exists} # ==================== 设置API ==================== @app.get('/api/settings/titles') async def get_title_settings(): with get_db() as conn: cursor = conn.cursor() cursor.execute( "SELECT key, value FROM settings WHERE key IN ('main_title', 'sub_title')" ) titles = {row['key']: row['value'] for row in cursor.fetchall()} return { 'main_title': titles.get('main_title', ''), 'sub_title': titles.get('sub_title', ''), } @app.post('/api/settings/titles', dependencies=[Depends(require_admin)]) async def update_title_settings(titles: TitleSettings): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute( "UPDATE settings SET value = ? WHERE key = 'main_title'", (titles.main_title,), ) cursor.execute( "UPDATE settings SET value = ? WHERE key = 'sub_title'", (titles.sub_title,) ) return {'success': True} @app.get('/api/settings/awards') async def get_award_settings(): with get_db() as conn: cursor = conn.cursor() cursor.execute( "SELECT key, value FROM settings WHERE key IN ('award_first', 'award_second', 'award_third')" ) counts = {row['key']: int(row['value']) for row in cursor.fetchall()} cursor.execute( "SELECT key, value FROM settings WHERE key IN ('award_first_name', 'award_second_name', 'award_third_name')" ) names = {row['key']: row['value'] for row in cursor.fetchall()} return { 'first': counts.get('award_first', 3), 'second': counts.get('award_second', 6), 'third': counts.get('award_third', 10), 'first_name': names.get('award_first_name', '一等奖'), 'second_name': names.get('award_second_name', '二等奖'), 'third_name': names.get('award_third_name', '三等奖'), } @app.post('/api/settings/awards', dependencies=[Depends(require_admin)]) async def update_award_settings(settings: AwardSettings): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute( "UPDATE settings SET value = ? WHERE key = 'award_first'", (str(settings.first),), ) cursor.execute( "UPDATE settings SET value = ? WHERE key = 'award_second'", (str(settings.second),), ) cursor.execute( "UPDATE settings SET value = ? WHERE key = 'award_third'", (str(settings.third),), ) return {'success': True} @app.post('/api/settings/award_names', dependencies=[Depends(require_admin)]) async def update_award_names(settings: AwardNamesSettings): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute( "UPDATE settings SET value = ? WHERE key = 'award_first_name'", (settings.first_name,), ) cursor.execute( "UPDATE settings SET value = ? WHERE key = 'award_second_name'", (settings.second_name,), ) cursor.execute( "UPDATE settings SET value = ? WHERE key = 'award_third_name'", (settings.third_name,), ) return {'success': True} @app.get('/api/criteria') async def get_criteria(): return { 'content': { 'label': '内容取材', 'max': 20, 'description': '主题鲜明，健康向上，编排新颖，原创性加分', }, 'stage': { 'label': '舞台风度', 'max': 10, 'description': '服装统一整洁，精神饱满，出场退场从容得体', }, 'art': { 'label': '表演艺术水平', 'max': 60, 'description': '动作流畅协调，节奏感强，技巧娴熟，富有感染力', }, 'overall': { 'label': '综合评价', 'max': 10, 'description': '整体效果印象深刻，现场驾驭能力强', }, } # ==================== 批次管理API ==================== @app.get('/api/batches') async def get_batches(): with get_db() as conn: cursor = conn.cursor() cursor.execute('SELECT id, name, created_at FROM batches ORDER BY id') batches = [{'id': row['id'], 'name': row['name']} for row in cursor.fetchall()] return batches @app.post('/api/batches', dependencies=[Depends(require_admin)]) async def create_batch(batch: BatchCreate): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute( 'INSERT INTO batches (name, created_at) VALUES (?, ?)', (batch.name, datetime.now()), ) new_id = cursor.lastrowid return {'success': True, 'id': new_id, 'name': batch.name} @app.delete('/api/batches/{batch_id}', dependencies=[Depends(require_admin)]) async def delete_batch(batch_id: int): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute('DELETE FROM batches WHERE id = ?', (batch_id,)) if cursor.rowcount == 0: raise HTTPException(status_code=404, detail='批次不存在') return {'success': True} @app.delete('/api/batches/{batch_id}/programs', dependencies=[Depends(require_admin)]) async def clear_batch_programs(batch_id: int): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute('DELETE FROM programs WHERE batch_id = ?', (batch_id,)) if cursor.rowcount == 0: raise HTTPException(status_code=404, detail='批次不存在') return {'success': True} # ==================== 评委管理API ==================== @app.get('/api/judges') async def get_judges(): with get_db() as conn: cursor = conn.cursor() cursor.execute('SELECT id, name, order_num FROM judges ORDER BY order_num') judges = [ {'id': row['id'], 'name': row['name'], 'order_num': row['order_num']} for row in cursor.fetchall() ] return judges @app.post('/api/judges', dependencies=[Depends(require_admin)]) async def create_judge(judge: JudgeCreate): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute('SELECT id FROM judges WHERE id = ?', (judge.id,)) if cursor.fetchone(): raise HTTPException(status_code=400, detail='评委编号已存在') cursor.execute('SELECT MAX(order_num) FROM judges') max_order = cursor.fetchone()[0] or 0 cursor.execute( 'INSERT INTO judges (id, name, order_num, created_at) VALUES (?, ?, ?, ?)', (judge.id, judge.name, max_order + 1, datetime.now()), ) cursor.execute('SELECT sn FROM programs') for prog in cursor.fetchall(): cursor.execute( 'INSERT OR IGNORE INTO scores (judge_id, program_sn, updated_at) VALUES (?, ?, ?)', (judge.id, prog['sn'], datetime.now()), ) return {'success': True} @app.put('/api/judges/{judge_id}', dependencies=[Depends(require_admin)]) async def update_judge(judge_id: str, judge: JudgeUpdate): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute('SELECT id FROM judges WHERE id = ?', (judge_id,)) if not cursor.fetchone(): raise HTTPException(status_code=404, detail='评委不存在') cursor.execute( 'UPDATE judges SET name = ? WHERE id = ?', (judge.name, judge_id) ) return {'success': True} @app.delete('/api/judges/{judge_id}', dependencies=[Depends(require_admin)]) async def delete_judge(judge_id: str): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute('DELETE FROM judges WHERE id = ?', (judge_id,)) if cursor.rowcount == 0: raise HTTPException(status_code=404, detail='评委不存在') return {'success': True} # ==================== 节目管理API ==================== @app.get('/api/programs') async def get_programs(batch_id: int | None = Query(None)): with get_db() as conn: cursor = conn.cursor() if batch_id: cursor.execute( 'SELECT sn, name, unit, performer, category, batch_id, order_num FROM programs WHERE batch_id = ? ORDER BY order_num', (batch_id,), ) else: cursor.execute( 'SELECT sn, name, unit, performer, category, batch_id, order_num FROM programs ORDER BY batch_id, order_num' ) programs = [ { 'sn': row['sn'], 'name': row['name'], 'unit': row['unit'], 'performer': row['performer'], 'category': row['category'], 'batch_id': row['batch_id'], 'order_num': row['order_num'], } for row in cursor.fetchall() ] return programs @app.post('/api/programs', dependencies=[Depends(require_admin)]) async def create_program(program: ProgramCreate): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO programs (sn, name, unit, performer, category, batch_id, order_num, created_at) VALUES (?, ?, ?, ?, ?, ?, (SELECT COALESCE(MAX(order_num), 0) + 1 FROM programs WHERE batch_id = ?), ?) """, ( program.sn, program.name, program.unit, program.performer, program.category, program.batch_id, datetime.now(), program.batch_id, ), ) if cursor.rowcount == 0: raise HTTPException(status_code=400, detail='节目序号已存在') cursor.execute('SELECT id FROM judges') for judge in cursor.fetchall(): cursor.execute( 'INSERT OR IGNORE INTO scores (judge_id, program_sn, updated_at) VALUES (?, ?, ?)', (judge['id'], program.sn, datetime.now()), ) return {'success': True} @app.post('/api/programs/reorder', dependencies=[Depends(require_admin)]) async def reorder_programs(data: dict): """批量更新节目顺序，data: {'batch_id': 1, 'sn_list': ['P001', 'P002', ...]}""" batch_id = data.get('batch_id') sn_list = data.get('sn_list', []) if not batch_id or not sn_list: raise HTTPException(status_code=400, detail='缺少参数') with get_db(write=True) as conn: cursor = conn.cursor() # 1. 先将该批次所有节目的 order_num 清空，避免唯一约束冲突 cursor.execute( 'UPDATE programs SET order_num = NULL WHERE batch_id = ?', (batch_id,) ) # 2. 按新顺序重新赋值 for idx, sn in enumerate(sn_list, start=1): cursor.execute( 'UPDATE programs SET order_num = ? WHERE sn = ? AND batch_id = ?', (idx, sn, batch_id), ) return {'success': True} @app.post('/api/programs/import', dependencies=[Depends(require_admin)]) async def import_programs( request: Request, batch_id: int = Query(...), file: UploadFile = File(...) ): MAX_MB = 10 content_length = request.headers.get('content-length') if content_length and int(content_length) > MAX_MB * 1024 * 1024: raise HTTPException(status_code=413, detail=f'文件大小超过 {MAX_MB}MB 限制') try: contents = await file.read() df = pd.read_excel(BytesIO(contents)) required_columns = ['节目编号', '节目名称', '演出单位', '表演者', '节目类别'] for col in required_columns: if col not in df.columns: raise HTTPException(status_code=400, detail=f'Excel缺少必要列：{col}') with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute( 'SELECT COALESCE(MAX(order_num), 0) FROM programs WHERE batch_id = ?', (batch_id,), ) max_order = cursor.fetchone()[0] next_order = max_order + 1 inserted_count = 0 updated_count = 0 for _, row in df.iterrows(): sn = escape_html(row['节目编号']) name = escape_html(row['节目名称']) unit = escape_html(row['演出单位']) if pd.notna(row['演出单位']) else '' performer = ( escape_html(row['表演者']) if pd.notna(row['表演者']) else '' ) category = ( escape_html(row['节目类别']) if pd.notna(row['节目类别']) else '' ) if not sn or not name: continue # 检查节目是否已存在 cursor.execute('SELECT sn, batch_id FROM programs WHERE sn = ?', (sn,)) exists = cursor.fetchone() if exists: # 如果节目已存在且不属于当前导入的批次，则报错，禁止跨批次移动 if exists['batch_id'] != batch_id: raise HTTPException( status_code=400, detail=f'节目编号 {sn} 已存在于批次ID {exists["batch_id"]} 中，不能通过导入移动到当前批次。' f'如需移动，请先删除原节目或使用编辑功能修改批次。', ) # 同批次更新（保留原有 order_num） cursor.execute( """UPDATE programs SET name = ?, unit = ?, performer = ?, category = ?, batch_id = ? WHERE sn = ?""", (name, unit, performer, category, batch_id, sn), ) updated_count += 1 else: # 新增节目，分配新的 order_num cursor.execute( """INSERT INTO programs (sn, name, unit, performer, category, batch_id, order_num, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", ( sn, name, unit, performer, category, batch_id, next_order, datetime.now(), ), ) next_order += 1 inserted_count += 1 # 为新节目初始化所有评委的评分记录（与原逻辑一致） cursor.execute('SELECT id FROM judges') for judge in cursor.fetchall(): cursor.execute( 'INSERT OR IGNORE INTO scores (judge_id, program_sn, updated_at) VALUES (?, ?, ?)', (judge['id'], sn, datetime.now()), ) return { 'success': True, 'message': f'导入完成：新增 {inserted_count} 个节目，更新 {updated_count} 个节目', } except Exception as exc: raise HTTPException( status_code=400, detail=exc.detail if isinstance(exc, HTTPException) else str(exc), ) from exc @app.put('/api/programs/{program_sn}', dependencies=[Depends(require_admin)]) async def update_program(program_sn: str, program: ProgramUpdate): with get_db(write=True) as conn: cursor = conn.cursor() # 获取当前节目的信息（特别是原 batch_id 和 order_num） cursor.execute( 'SELECT batch_id, order_num FROM programs WHERE sn = ?', (program_sn,) ) old = cursor.fetchone() if not old: raise HTTPException(status_code=404, detail='节目不存在') new_batch_id = program.batch_id # 如果批次发生了变更 if old['batch_id'] != new_batch_id: # 获取目标批次的最大 order_num cursor.execute( 'SELECT COALESCE(MAX(order_num), 0) FROM programs WHERE batch_id = ?', (new_batch_id,), ) max_order = cursor.fetchone()[0] new_order_num = max_order + 1 # 新节目放到末尾 else: # 批次未变，保留原 order_num new_order_num = old['order_num'] # 更新节目信息，包括 batch_id 和 order_num cursor.execute( """ UPDATE programs SET name = ?, unit = ?, performer = ?, category = ?, batch_id = ?, order_num = ? WHERE sn = ? """, ( program.name, program.unit, program.performer, program.category, new_batch_id, new_order_num, program_sn, ), ) return {'success': True} @app.delete('/api/programs/{program_sn}', dependencies=[Depends(require_admin)]) async def delete_program(program_sn: str): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute('DELETE FROM programs WHERE sn = ?', (program_sn,)) if cursor.rowcount == 0: raise HTTPException(status_code=404, detail='节目不存在') return {'success': True} # ==================== 评分API ==================== @app.get('/api/scores/{judge_id}') async def get_judge_scores(judge_id: str): with get_db() as conn: cursor = conn.cursor() cursor.execute( 'SELECT program_sn, content_score, stage_score, art_score, overall_score, comment FROM scores WHERE judge_id = ?', (judge_id,), ) rows = cursor.fetchall() scores = {} for row in rows: scores[row['program_sn']] = { 'content': row['content_score'], 'stage': row['stage_score'], 'art': row['art_score'], 'overall': row['overall_score'], 'comment': row['comment'] or '', } cursor.execute('SELECT sn FROM programs') for row in cursor.fetchall(): if row['sn'] not in scores: scores[row['sn']] = { 'content': None, 'stage': None, 'art': None, 'overall': None, 'comment': '', } return scores @app.post('/api/scores/{judge_id}') async def update_score(judge_id: str, score_data: ScoreDetailUpdate): max_scores = {'content': 20, 'stage': 10, 'art': 60, 'overall': 10} if score_data.field not in max_scores: raise HTTPException(status_code=400, detail='无效评分项') val = score_data.value if val is not None and (val < 0 or val > max_scores[score_data.field]): raise HTTPException( status_code=400, detail=f'分值必须在0-{max_scores[score_data.field]}之间' ) field_map = { 'content': 'content_score', 'stage': 'stage_score', 'art': 'art_score', 'overall': 'overall_score', } with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute( f""" INSERT INTO scores (judge_id, program_sn, {field_map[score_data.field]}, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(judge_id, program_sn) DO UPDATE SET {field_map[score_data.field]} = excluded.{field_map[score_data.field]}, updated_at = excluded.updated_at """, (judge_id, score_data.program_sn, val, datetime.now()), ) cursor.execute( 'SELECT content_score, stage_score, art_score, overall_score FROM scores WHERE judge_id = ? AND program_sn = ?', (judge_id, score_data.program_sn), ) row = cursor.fetchone() total = 0 if row: for v in [ row['content_score'], row['stage_score'], row['art_score'], row['overall_score'], ]: if v is not None: total += v return {'success': True, 'total': total} @app.post('/api/comment/{judge_id}') async def update_comment(judge_id: str, comment_data: CommentUpdate): with get_db(write=True) as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO scores (judge_id, program_sn, comment, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(judge_id, program_sn) DO UPDATE SET comment = excluded.comment, updated_at = excluded.updated_at """, (judge_id, comment_data.program_sn, comment_data.comment, datetime.now()), ) return {'success': True} @app.post('/api/scores/admin/batch', dependencies=[Depends(require_admin)]) async def batch_update_scores(batch_data: ScoreBatchUpdate): """管理员批量更新总分""" with get_db() as conn: cursor = conn.cursor() for judge_id, total_score in batch_data.judge_scores.items(): if total_score is not None and (total_score < 0 or total_score > 100): raise HTTPException(status_code=400, detail='总分必须在0-100之间') cursor.execute('SELECT id FROM judges WHERE id = ?', (judge_id,)) if not cursor.fetchone(): raise HTTPException(status_code=404, detail=f'评委 {judge_id} 不存在') with get_db(write=True) as conn: cursor = conn.cursor() for judge_id, total_score in batch_data.judge_scores.items(): if total_score is None: cursor.execute( """ UPDATE scores SET admin_total_score = NULL, updated_at = ? WHERE judge_id = ? AND program_sn = ? """, (datetime.now(), judge_id, batch_data.program_sn), ) else: cursor.execute( """ INSERT INTO scores (judge_id, program_sn, admin_total_score, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(judge_id, program_sn) DO UPDATE SET admin_total_score = excluded.admin_total_score, updated_at = excluded.updated_at """, (judge_id, batch_data.program_sn, total_score, datetime.now()), ) return {'success': True} @app.post('/api/scores/program/batch') async def get_programs_scores(data: dict): sn_list = data.get('sn_list', []) if isinstance(sn_list, str): sn_list = [s.strip() for s in sn_list.split(',') if s.strip()] if not sn_list: return {} with get_db() as conn: cursor = conn.cursor() placeholders = ','.join(['?' for _ in sn_list]) cursor.execute( f""" SELECT program_sn, judge_id, CASE WHEN (admin_total_score IS NOT NULL OR content_score IS NOT NULL OR stage_score IS NOT NULL OR art_score IS NOT NULL OR overall_score IS NOT NULL) THEN COALESCE(admin_total_score, COALESCE(content_score,0) + COALESCE(stage_score,0) + COALESCE(art_score,0) + COALESCE(overall_score,0)) ELSE NULL END AS final_total FROM scores WHERE program_sn IN ({placeholders}) """, sn_list, ) rows = cursor.fetchall() result = {} for row in rows: prog_sn = row['program_sn'] if prog_sn not in result: result[prog_sn] = {} result[prog_sn][row['judge_id']] = row['final_total'] return result # ==================== 汇总统计API ==================== @app.get('/api/aggregate') async def get_aggregate(batch_id: int | None = Query(None)): with get_db() as conn: cursor = conn.cursor() if batch_id: cursor.execute( 'SELECT sn, name, unit, performer, category FROM programs WHERE batch_id = ? ORDER BY order_num', (batch_id,), ) else: cursor.execute( 'SELECT sn, name, unit, performer, category FROM programs ORDER BY batch_id, order_num' ) programs = cursor.fetchall() if not programs: return [] sn_list = [p['sn'] for p in programs] placeholders = ','.join(['?' for _ in sn_list]) cursor.execute( f""" SELECT t.program_sn, AVG(CASE WHEN t.has_score THEN t.final_total ELSE NULL END) AS avg_score, MAX(CASE WHEN t.has_score THEN t.final_total ELSE NULL END) AS max_score, MIN(CASE WHEN t.has_score THEN t.final_total ELSE NULL END) AS min_score, COUNT(DISTINCT CASE WHEN t.has_score THEN t.judge_id ELSE NULL END) AS scored_count FROM ( SELECT program_sn, judge_id, CASE WHEN admin_total_score IS NOT NULL THEN admin_total_score WHEN (content_score IS NOT NULL OR stage_score IS NOT NULL OR art_score IS NOT NULL OR overall_score IS NOT NULL) THEN COALESCE(content_score,0) + COALESCE(stage_score,0) + COALESCE(art_score,0) + COALESCE(overall_score,0) ELSE NULL END AS final_total, (admin_total_score IS NOT NULL OR content_score IS NOT NULL OR stage_score IS NOT NULL OR art_score IS NOT NULL OR overall_score IS NOT NULL) AS has_score FROM scores WHERE program_sn IN ({placeholders}) ) t GROUP BY t.program_sn """, sn_list, ) stats = {} for row in cursor.fetchall(): stats[row['program_sn']] = { 'avg_score': row['avg_score'] if row['avg_score'] is not None else None, 'max_score': row['max_score'], 'min_score': row['min_score'], 'scored_count': row['scored_count'] or 0, } results = [] for prog in programs: stat = stats.get( prog['sn'], {'avg_score': None, 'max_score': 0, 'min_score': 0, 'scored_count': 0}, ) results.append( { 'sn': prog['sn'], 'name': prog['name'], 'unit': prog['unit'], 'performer': prog['performer'], 'category': prog['category'], 'avg_score': round(stat['avg_score'], 3) if stat['avg_score'] is not None else None, 'max_score': stat['max_score'], 'min_score': stat['min_score'], 'scored_count': stat['scored_count'], } ) results.sort( key=lambda x: x['avg_score'] if x['avg_score'] else 0, reverse=True ) for idx, r in enumerate(results): r['rank'] = idx + 1 return results @app.get('/api/progress') async def get_progress(): with get_db() as conn: cursor = conn.cursor() total_programs = cursor.execute('SELECT COUNT(*) FROM programs').fetchone()[0] total_judges = cursor.execute('SELECT COUNT(*) FROM judges').fetchone()[0] total_cells = total_programs * total_judges scored_cells = cursor.execute( 'SELECT COUNT(*) FROM scores WHERE content_score IS NOT NULL OR stage_score IS NOT NULL OR art_score IS NOT NULL OR overall_score IS NOT NULL OR admin_total_score IS NOT NULL' ).fetchone()[0] progress = round(scored_cells / total_cells * 100, 1) if total_cells > 0 else 0 return { 'total_programs': total_programs, 'total_judges': total_judges, 'scored_programs': scored_cells, 'total_cells': total_cells, 'progress': progress, } @app.get('/api/export_excel') async def export_excel(batch_id: int | None = Query(None)): aggregate = await get_aggregate(batch_id) data = [ { '名次': item['rank'], '节目编号': item['sn'], '节目名称': item['name'], '演出单位': item['unit'], '表演者': item['performer'], '节目类别': item['category'], '平均分': item['avg_score'], '最高分': item['max_score'], '最低分': item['min_score'], '评分评委数': item['scored_count'], } for item in aggregate ] df = pd.DataFrame(data) output = BytesIO() with pd.ExcelWriter(output, engine='openpyxl') as writer: df.to_excel(writer, sheet_name='评分汇总表', index=False) output.seek(0) filename = f'{"batch_" + str(batch_id) if batch_id else "all"}_scores.xlsx' return StreamingResponse( output, media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', headers={'Content-Disposition': f'attachment; filename={filename}'}, ) # ==================== 打印页面API ==================== @app.get('/print/program/{program_sn}') async def print_program_score(program_sn: str): with get_db() as conn: prog = conn.execute( 'SELECT sn, name, unit, performer, category FROM programs WHERE sn = ?', (program_sn,), ).fetchone() if not prog: raise HTTPException(status_code=404, detail='节目不存在') judges = conn.execute( 'SELECT id, name FROM judges ORDER BY order_num' ).fetchall() scores_data, all_scores = {}, [] for judge in judges: score = conn.execute( """SELECT content_score, stage_score, art_score, overall_score, admin_total_score, comment FROM scores WHERE judge_id = ? AND program_sn = ?""", (judge['id'], program_sn), ).fetchone() total = None has_score = False comment = '' if score: # 优先使用管理员总分 if score['admin_total_score'] is not None: total = score['admin_total_score'] has_score = True else: # 计算明细分之和 sum_score = 0 if score['content_score'] is not None: sum_score += score['content_score'] has_score = True if score['stage_score'] is not None: sum_score += score['stage_score'] has_score = True if score['art_score'] is not None: sum_score += score['art_score'] has_score = True if score['overall_score'] is not None: sum_score += score['overall_score'] has_score = True if has_score: total = sum_score comment = score['comment'] or '' if has_score: all_scores.append(total) scores_data[judge['name']] = { 'total': total, 'comment': comment, 'has_score': has_score, } max_score = max(all_scores) if all_scores else 0 min_score = min(all_scores) if all_scores else 0 avg_score = round(sum(all_scores) / len(all_scores), 3) if all_scores else 0 titles = await get_title_settings() return HTMLResponse( content=render_print_program_template( prog, judges, scores_data, max_score, min_score, avg_score, titles ) ) @app.get('/print/blank/{program_sn}') async def print_blank_score(program_sn: str): with get_db() as conn: prog = conn.execute( 'SELECT sn, name, unit, performer, category FROM programs WHERE sn = ?', (program_sn,), ).fetchone() if not prog: raise HTTPException(status_code=404, detail='节目不存在') judges = conn.execute( 'SELECT id, name FROM judges ORDER BY order_num' ).fetchall() criteria = await get_criteria() titles = await get_title_settings() return HTMLResponse( content=render_print_blank_template(prog, judges, criteria, titles) ) async def get_award_winners(batch_id: int | None = Query(None)): aggregate = await get_aggregate(batch_id) settings = await get_award_settings() first_count, second_count, third_count = ( settings['first'], settings['second'], settings['third'], ) winners = {'first': [], 'second': [], 'third': []} for idx, prog in enumerate(aggregate): if idx < first_count: winners['first'].append(prog) elif idx < first_count + second_count: winners['second'].append(prog) elif idx < first_count + second_count + third_count: winners['third'].append(prog) return winners @app.get('/print/award/{award_type}') async def print_award_list(award_type: str, batch_id: int | None = Query(None)): winners = await get_award_winners(batch_id) if award_type not in winners: raise HTTPException(status_code=404, detail='奖项类型不存在') award_settings = await get_award_settings() award_names = { 'first': award_settings['first_name'], 'second': award_settings['second_name'], 'third': award_settings['third_name'], } titles = await get_title_settings() batch_name = '' if batch_id: with get_db() as conn: batch = conn.execute( 'SELECT name FROM batches WHERE id = ?', (batch_id,) ).fetchone() if batch: batch_name = f'（{batch["name"]}）' return HTMLResponse( content=render_print_award_template( award_type, award_names[award_type], winners[award_type], titles, batch_name ) ) def render_print_program_template( program, judges, scores_data, max_score, min_score, avg_score, titles ): html = f"""{program['name']} - 评分详情

{titles['main_title']}

{titles['sub_title']} · 评分详情表

节目编号： {program['sn']}

节目名称： {program['name']}

演出单位： {program['unit']}

表演者： {program['performer']}

节目类别： {program['category']}

📊 平均分
{avg_score}
🎯 最高分
{max_score}
⭐ 最低分
{min_score}
""" for judge in judges: s = scores_data.get(judge['name'], {}) total_display = ( '-' if not s.get('has_score') else (s.get('total') if s.get('total') is not None else '-') ) html += f"""""" html += f"""
评委姓名总分(100分)评委点评
{judge['name']}{total_display}{s.get('comment', '')}

评委组签名：____________________

日期：{datetime.now().strftime('%Y年%m月%d日')}

""" return html def render_print_blank_template(program, judges, criteria, titles): html = f"""{program['name']} - 空白评分表 """ for _judge in judges: html += f"""

{titles['main_title']}

{titles['sub_title']} · 节目评分表

节目编号： {program['sn']} | 节目名称： {program['name']}

演出单位： {program['unit']} | 表演者： {program['performer']} | 节目类别： {program['category']}

""" for _key, cri in criteria.items(): html += f"""""" html += """
评分项目满分得分评分要点
{cri['label']}{cri['max']}分______{cri['description']}
合计100分______
评委签名：____________________日期：____年____月____日
""" html += ( '
' '' ) return html def render_print_award_template(award_type, award_name, winners, titles, batch_name=''): html = f"""{award_name}{batch_name}获奖名单

{titles['main_title']}

{titles['sub_title']}{batch_name}
{award_name}
""" for idx, prog in enumerate(winners, 1): html += f"""""" html += f"""
名次节目编号节目名称演出单位表演者节目类别平均分
{idx}{prog['sn']}{prog['name']}{prog['unit']}{prog['performer']}{prog['category']}{prog['avg_score']}
评委组签名：____________________日期：{datetime.now().strftime('%Y年%m月%d日')}
""" return html # ==================== HTML模板（前端部分） ==================== HTML_TEMPLATE = """ 文艺汇演评分系统

加载中...

🎯 评分打分
🖨️ 打印中心
请先登录选择节目评分

📊 评分汇总表

名次节目编号节目名称演出单位表演者类别平均分最高分最低分评委数操作
"""