370 lines
16 KiB
Python
370 lines
16 KiB
Python
import os
|
||
import sqlite3
|
||
import random
|
||
import string
|
||
from datetime import datetime
|
||
from flask import Flask, render_template, request, redirect, url_for, g, flash, jsonify
|
||
|
||
app = Flask(__name__)
|
||
app.secret_key = 'admin-secret-key-2026'
|
||
|
||
# 连接主应用的数据库
|
||
DATABASE = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'instance', 'database.db')
|
||
|
||
def get_db():
|
||
db = getattr(g, '_database', None)
|
||
if db is None:
|
||
db = g._database = sqlite3.connect(DATABASE)
|
||
db.row_factory = sqlite3.Row
|
||
return db
|
||
|
||
@app.teardown_appcontext
|
||
def close_connection(exception):
|
||
db = getattr(g, '_database', None)
|
||
if db is not None:
|
||
db.close()
|
||
|
||
def init_db():
|
||
"""确保 system_notification 和 invite_code 表存在"""
|
||
with app.app_context():
|
||
db = get_db()
|
||
db.executescript('''
|
||
CREATE TABLE IF NOT EXISTS system_notification (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
title TEXT NOT NULL,
|
||
content TEXT NOT NULL,
|
||
author_id INTEGER NOT NULL,
|
||
pinned INTEGER DEFAULT 0,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||
);
|
||
CREATE TABLE IF NOT EXISTS invite_code (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
code TEXT UNIQUE NOT NULL,
|
||
user_id INTEGER NOT NULL,
|
||
application_id INTEGER NOT NULL,
|
||
used INTEGER DEFAULT 0,
|
||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
||
used_at TIMESTAMP
|
||
);
|
||
''')
|
||
db.commit()
|
||
|
||
# PLACEHOLDER_ROUTES
|
||
|
||
def generate_invite_code():
|
||
return 'TC-' + ''.join(random.choices(string.ascii_uppercase + string.digits, k=8))
|
||
|
||
def add_notification(db, user_id, ntype, content, from_user='', post_id=0):
|
||
db.execute('INSERT INTO notification (user_id, type, content, from_user, post_id, read) VALUES (?,?,?,?,?,0)',
|
||
(user_id, ntype, content, from_user, post_id))
|
||
|
||
# ========== 后台首页 ==========
|
||
@app.route('/admin')
|
||
def dashboard():
|
||
db = get_db()
|
||
stats = {
|
||
'users': db.execute('SELECT COUNT(*) as c FROM user').fetchone()['c'],
|
||
'contests': db.execute('SELECT COUNT(*) as c FROM contest').fetchone()['c'],
|
||
'exams': db.execute('SELECT COUNT(*) as c FROM exam').fetchone()['c'],
|
||
'posts': db.execute('SELECT COUNT(*) as c FROM post').fetchone()['c'],
|
||
'pending_teacher': db.execute("SELECT COUNT(*) as c FROM teacher_application WHERE status='pending'").fetchone()['c'],
|
||
'pending_contest': db.execute("SELECT COUNT(*) as c FROM contest_application WHERE status='pending'").fetchone()['c'],
|
||
}
|
||
return render_template('dashboard.html', stats=stats)
|
||
|
||
# ========== 杯赛管理 ==========
|
||
@app.route('/admin/competitions')
|
||
def competitions():
|
||
db = get_db()
|
||
contests = db.execute('SELECT * FROM contest ORDER BY created_at DESC').fetchall()
|
||
return render_template('competitions.html', competitions=contests)
|
||
|
||
@app.route('/admin/competitions/stop/<int:id>', methods=['POST'])
|
||
def stop_competition(id):
|
||
db = get_db()
|
||
db.execute("UPDATE contest SET status = 'ended' WHERE id = ?", (id,))
|
||
db.commit()
|
||
return redirect(url_for('competitions'))
|
||
|
||
@app.route('/admin/competitions/activate/<int:id>', methods=['POST'])
|
||
def activate_competition(id):
|
||
db = get_db()
|
||
db.execute("UPDATE contest SET status = 'upcoming' WHERE id = ?", (id,))
|
||
db.commit()
|
||
return redirect(url_for('competitions'))
|
||
|
||
# ========== 杯赛申请 ==========
|
||
@app.route('/admin/competition_apps')
|
||
def competition_apps():
|
||
db = get_db()
|
||
apps = db.execute('''
|
||
SELECT ca.*, u.name as applicant_name FROM contest_application ca
|
||
JOIN user u ON ca.user_id = u.id
|
||
ORDER BY ca.applied_at DESC
|
||
''').fetchall()
|
||
return render_template('competition_apps.html', apps=apps)
|
||
|
||
@app.route('/admin/competition_apps/approve/<int:id>', methods=['POST'])
|
||
def approve_competition_app(id):
|
||
db = get_db()
|
||
ca = db.execute('SELECT * FROM contest_application WHERE id = ?', (id,)).fetchone()
|
||
if not ca or ca['status'] != 'pending':
|
||
flash('该申请已处理', 'error')
|
||
return redirect(url_for('competition_apps'))
|
||
# 创建杯赛
|
||
db.execute('INSERT INTO contest (name, organizer, description, start_date, end_date, status, participants, created_by) VALUES (?,?,?,?,?,?,?,?)',
|
||
(ca['name'], ca['organizer'], ca['description'], '待定', '待定', 'upcoming', 0, ca['user_id']))
|
||
contest_id = db.execute('SELECT last_insert_rowid()').fetchone()[0]
|
||
# 创建成员关系（owner）
|
||
db.execute('INSERT INTO contest_membership (user_id, contest_id, role) VALUES (?,?,?)', (ca['user_id'], contest_id, 'owner'))
|
||
# 创建杯赛讨论群
|
||
db.execute('INSERT INTO chat_room (type, name, creator_id, contest_id) VALUES (?,?,?,?)',
|
||
('contest', ca['name'] + ' 讨论群', ca['user_id'], contest_id))
|
||
room_id = db.execute('SELECT last_insert_rowid()').fetchone()[0]
|
||
db.execute('INSERT INTO chat_room_member (room_id, user_id, role) VALUES (?,?,?)', (room_id, ca['user_id'], 'admin'))
|
||
# 更新申请状态
|
||
db.execute("UPDATE contest_application SET status = 'approved', reviewed_at = ? WHERE id = ?", (datetime.utcnow(), id))
|
||
add_notification(db, ca['user_id'], 'contest_result', f'您申请举办的杯赛「{ca["name"]}」已通过审核！', from_user='系统')
|
||
db.commit()
|
||
flash('申请已批准，杯赛已创建', 'success')
|
||
return redirect(url_for('competition_apps'))
|
||
|
||
@app.route('/admin/competition_apps/reject/<int:id>', methods=['POST'])
|
||
def reject_competition_app(id):
|
||
db = get_db()
|
||
ca = db.execute('SELECT * FROM contest_application WHERE id = ?', (id,)).fetchone()
|
||
if not ca or ca['status'] != 'pending':
|
||
flash('该申请已处理', 'error')
|
||
return redirect(url_for('competition_apps'))
|
||
db.execute("UPDATE contest_application SET status = 'rejected', reviewed_at = ? WHERE id = ?", (datetime.utcnow(), id))
|
||
add_notification(db, ca['user_id'], 'contest_result', f'您申请举办的杯赛「{ca["name"]}」未通过审核。', from_user='系统')
|
||
db.commit()
|
||
flash('申请已拒绝', 'success')
|
||
return redirect(url_for('competition_apps'))
|
||
|
||
# PLACEHOLDER_TEACHER_ROUTES
|
||
|
||
# ========== 教师申请（邀请码流程） ==========
|
||
@app.route('/admin/teacher_apps')
|
||
def teacher_apps():
|
||
db = get_db()
|
||
apps = db.execute('''
|
||
SELECT ta.*, u.name as username, c.name as contest_name
|
||
FROM teacher_application ta
|
||
JOIN user u ON ta.user_id = u.id
|
||
JOIN contest c ON ta.contest_id = c.id
|
||
ORDER BY ta.applied_at DESC
|
||
''').fetchall()
|
||
return render_template('teacher_apps.html', apps=apps)
|
||
|
||
@app.route('/admin/teacher_apps/approve/<int:id>', methods=['POST'])
|
||
def approve_teacher_app(id):
|
||
db = get_db()
|
||
ta = db.execute('SELECT * FROM teacher_application WHERE id = ?', (id,)).fetchone()
|
||
if not ta or ta['status'] != 'pending':
|
||
flash('该申请已处理', 'error')
|
||
return redirect(url_for('teacher_apps'))
|
||
# 检查是否已是杯赛成员
|
||
existing = db.execute('SELECT id FROM contest_membership WHERE user_id = ? AND contest_id = ?',
|
||
(ta['user_id'], ta['contest_id'])).fetchone()
|
||
if existing:
|
||
db.execute("UPDATE teacher_application SET status = 'rejected', reviewed_at = ? WHERE id = ?", (datetime.utcnow(), id))
|
||
db.commit()
|
||
flash('用户已是杯赛成员，申请已拒绝', 'error')
|
||
return redirect(url_for('teacher_apps'))
|
||
# 生成邀请码
|
||
code = generate_invite_code()
|
||
while db.execute('SELECT id FROM invite_code WHERE code = ?', (code,)).fetchone():
|
||
code = generate_invite_code()
|
||
db.execute('INSERT INTO invite_code (code, user_id, application_id) VALUES (?,?,?)', (code, ta['user_id'], ta['id']))
|
||
db.execute("UPDATE teacher_application SET status = 'approved', reviewed_at = ? WHERE id = ?", (datetime.utcnow(), id))
|
||
# 获取杯赛名称
|
||
contest = db.execute('SELECT name FROM contest WHERE id = ?', (ta['contest_id'],)).fetchone()
|
||
contest_name = contest['name'] if contest else ''
|
||
# 查找或创建私聊室，发送邀请码消息
|
||
admin_id = db.execute("SELECT id FROM user WHERE role = 'admin' LIMIT 1").fetchone()
|
||
sender_id = admin_id['id'] if admin_id else 1
|
||
# 查找已有私聊
|
||
room = db.execute('''
|
||
SELECT cr.id FROM chat_room cr
|
||
WHERE cr.type = 'private'
|
||
AND cr.id IN (SELECT room_id FROM chat_room_member WHERE user_id = ?)
|
||
AND cr.id IN (SELECT room_id FROM chat_room_member WHERE user_id = ?)
|
||
''', (sender_id, ta['user_id'])).fetchone()
|
||
if room:
|
||
room_id = room['id']
|
||
else:
|
||
db.execute('INSERT INTO chat_room (type, creator_id) VALUES (?,?)', ('private', sender_id))
|
||
room_id = db.execute('SELECT last_insert_rowid()').fetchone()[0]
|
||
db.execute('INSERT INTO chat_room_member (room_id, user_id, role) VALUES (?,?,?)', (room_id, sender_id, 'member'))
|
||
db.execute('INSERT INTO chat_room_member (room_id, user_id, role) VALUES (?,?,?)', (room_id, ta['user_id'], 'member'))
|
||
# 发送私聊消息
|
||
msg = f'恭喜！您申请成为杯赛「{contest_name}」老师已通过审核。\n请使用以下邀请码激活您的教师身份：\n\n🎫 邀请码：{code}\n\n请前往「申请成为老师」页面，在邀请码输入框中输入此码完成激活。\n注意：此邀请码仅限您本人使用，且仅限一次。'
|
||
db.execute('INSERT INTO message (room_id, sender_id, type, content) VALUES (?,?,?,?)', (room_id, sender_id, 'system', msg))
|
||
add_notification(db, ta['user_id'], 'teacher_result',
|
||
f'您申请成为杯赛「{contest_name}」老师已通过审核，请查看私聊消息获取邀请码。', from_user='系统')
|
||
db.commit()
|
||
flash(f'申请已批准，邀请码 {code} 已通过私聊发送给老师', 'success')
|
||
return redirect(url_for('teacher_apps'))
|
||
|
||
@app.route('/admin/teacher_apps/reject/<int:id>', methods=['POST'])
|
||
def reject_teacher_app(id):
|
||
db = get_db()
|
||
ta = db.execute('SELECT * FROM teacher_application WHERE id = ?', (id,)).fetchone()
|
||
if not ta or ta['status'] != 'pending':
|
||
flash('该申请已处理', 'error')
|
||
return redirect(url_for('teacher_apps'))
|
||
db.execute("UPDATE teacher_application SET status = 'rejected', reviewed_at = ? WHERE id = ?", (datetime.utcnow(), id))
|
||
contest = db.execute('SELECT name FROM contest WHERE id = ?', (ta['contest_id'],)).fetchone()
|
||
contest_name = contest['name'] if contest else ''
|
||
add_notification(db, ta['user_id'], 'teacher_result',
|
||
f'您申请成为杯赛「{contest_name}」老师未通过审核。', from_user='系统')
|
||
db.commit()
|
||
flash('申请已拒绝', 'success')
|
||
return redirect(url_for('teacher_apps'))
|
||
|
||
# PLACEHOLDER_REMAINING_ROUTES
|
||
|
||
# ========== 考试管理 ==========
|
||
@app.route('/admin/exams')
|
||
def exams():
|
||
db = get_db()
|
||
exams = db.execute('''
|
||
SELECT e.*, c.name as contest_name, u.name as creator_name
|
||
FROM exam e
|
||
LEFT JOIN contest c ON e.contest_id = c.id
|
||
LEFT JOIN user u ON e.creator_id = u.id
|
||
ORDER BY e.created_at DESC
|
||
''').fetchall()
|
||
return render_template('exams.html', exams=exams)
|
||
|
||
@app.route('/admin/exams/stop/<int:id>', methods=['POST'])
|
||
def stop_exam(id):
|
||
db = get_db()
|
||
db.execute("UPDATE exam SET status = 'closed' WHERE id = ?", (id,))
|
||
db.commit()
|
||
flash('考试已停止', 'success')
|
||
return redirect(url_for('exams'))
|
||
|
||
@app.route('/admin/exams/activate/<int:id>', methods=['POST'])
|
||
def activate_exam(id):
|
||
db = get_db()
|
||
db.execute("UPDATE exam SET status = 'available' WHERE id = ?", (id,))
|
||
db.commit()
|
||
flash('考试已恢复', 'success')
|
||
return redirect(url_for('exams'))
|
||
|
||
# ========== 用户管理 ==========
|
||
@app.route('/admin/users')
|
||
def users():
|
||
db = get_db()
|
||
all_users = db.execute('SELECT * FROM user ORDER BY created_at DESC').fetchall()
|
||
return render_template('users.html', users=all_users)
|
||
|
||
@app.route('/admin/users/toggle_ban/<int:id>', methods=['POST'])
|
||
def toggle_ban_user(id):
|
||
db = get_db()
|
||
cur = db.execute('SELECT is_banned FROM user WHERE id = ?', (id,)).fetchone()
|
||
new_val = 0 if cur['is_banned'] else 1
|
||
db.execute('UPDATE user SET is_banned = ? WHERE id = ?', (new_val, id))
|
||
db.commit()
|
||
flash('用户状态已更新', 'success')
|
||
return redirect(url_for('users'))
|
||
|
||
@app.route('/admin/users/set_role/<int:id>', methods=['POST'])
|
||
def set_user_role(id):
|
||
role = request.form.get('role', 'student')
|
||
db = get_db()
|
||
db.execute('UPDATE user SET role = ? WHERE id = ?', (role, id))
|
||
db.commit()
|
||
flash('角色已更新', 'success')
|
||
return redirect(url_for('users'))
|
||
|
||
# ========== 帖子管理 ==========
|
||
@app.route('/admin/posts')
|
||
def posts():
|
||
db = get_db()
|
||
all_posts = db.execute('''
|
||
SELECT p.*, u.name as author_name FROM post p
|
||
LEFT JOIN user u ON p.author_id = u.id
|
||
ORDER BY p.created_at DESC
|
||
''').fetchall()
|
||
return render_template('posts.html', posts=all_posts)
|
||
|
||
@app.route('/admin/posts/delete/<int:id>', methods=['POST'])
|
||
def delete_post(id):
|
||
db = get_db()
|
||
db.execute('DELETE FROM post WHERE id = ?', (id,))
|
||
db.commit()
|
||
flash('帖子已删除', 'success')
|
||
return redirect(url_for('posts'))
|
||
|
||
@app.route('/admin/posts/toggle_pin/<int:id>', methods=['POST'])
|
||
def toggle_pin_post(id):
|
||
db = get_db()
|
||
cur = db.execute('SELECT pinned FROM post WHERE id = ?', (id,)).fetchone()
|
||
new_val = 0 if cur['pinned'] else 1
|
||
db.execute('UPDATE post SET pinned = ? WHERE id = ?', (new_val, id))
|
||
db.commit()
|
||
flash('帖子置顶状态已更新', 'success')
|
||
return redirect(url_for('posts'))
|
||
|
||
# PLACEHOLDER_NOTIFICATION_ROUTES
|
||
|
||
# ========== 公告管理 ==========
|
||
@app.route('/admin/notifications')
|
||
def notifications():
|
||
db = get_db()
|
||
anns = db.execute('''
|
||
SELECT sn.*, u.name as author_name FROM system_notification sn
|
||
LEFT JOIN user u ON sn.author_id = u.id
|
||
ORDER BY sn.pinned DESC, sn.created_at DESC
|
||
''').fetchall()
|
||
return render_template('notifications.html', announcements=anns)
|
||
|
||
@app.route('/admin/notifications/create', methods=['POST'])
|
||
def create_notification():
|
||
title = request.form.get('title', '').strip()
|
||
content = request.form.get('content', '').strip()
|
||
pinned = 1 if request.form.get('pinned') else 0
|
||
if not title or not content:
|
||
flash('标题和内容不能为空', 'error')
|
||
return redirect(url_for('notifications'))
|
||
db = get_db()
|
||
admin = db.execute("SELECT id FROM user WHERE role = 'admin' LIMIT 1").fetchone()
|
||
author_id = admin['id'] if admin else 1
|
||
now = datetime.utcnow()
|
||
db.execute('INSERT INTO system_notification (title, content, author_id, pinned, created_at, updated_at) VALUES (?,?,?,?,?,?)',
|
||
(title, content, author_id, pinned, now, now))
|
||
db.commit()
|
||
flash('公告已发布', 'success')
|
||
return redirect(url_for('notifications'))
|
||
|
||
@app.route('/admin/notifications/edit/<int:id>', methods=['POST'])
|
||
def edit_notification(id):
|
||
title = request.form.get('title', '').strip()
|
||
content = request.form.get('content', '').strip()
|
||
pinned = 1 if request.form.get('pinned') else 0
|
||
if not title or not content:
|
||
flash('标题和内容不能为空', 'error')
|
||
return redirect(url_for('notifications'))
|
||
db = get_db()
|
||
db.execute('UPDATE system_notification SET title=?, content=?, pinned=?, updated_at=? WHERE id=?',
|
||
(title, content, pinned, datetime.utcnow(), id))
|
||
db.commit()
|
||
flash('公告已更新', 'success')
|
||
return redirect(url_for('notifications'))
|
||
|
||
@app.route('/admin/notifications/delete/<int:id>', methods=['POST'])
|
||
def delete_notification(id):
|
||
db = get_db()
|
||
db.execute('DELETE FROM system_notification WHERE id = ?', (id,))
|
||
db.commit()
|
||
flash('公告已删除', 'success')
|
||
return redirect(url_for('notifications'))
|
||
|
||
if __name__ == '__main__':
|
||
init_db()
|
||
app.run(debug=True, port=5001)
|