# -*- coding: utf-8 -*- """中台：题库管理（增删改查 + Excel 导入）。""" import os import uuid from flask import render_template, request, redirect, url_for, flash, current_app from flask_login import login_required from app.models import Question from app.extensions import db from app.admin import bp from app.admin.views import _admin_only from app.services import log_action ALLOWED_IMG = {"png", "jpg", "jpeg", "gif", "webp"} @bp.route("/questions") @login_required def questions(): _admin_only() q = request.args.get("q", "").strip() cat = request.args.get("cat", "").strip() qtype = request.args.get("qtype", "").strip() query = Question.query if q: query = query.filter(Question.stem.contains(q)) if cat: query = query.filter_by(category=cat) if qtype: query = query.filter_by(qtype=qtype) questions = query.order_by(Question.id.desc()).limit(300).all() cats = [r[0] for r in db.session.query(Question.category).distinct().all() if r[0]] qtypes = [r[0] for r in db.session.query(Question.qtype).distinct().all() if r[0]] return render_template("admin/questions.html", questions=questions, cats=cats, qtypes=qtypes, q=q, cat=cat, qtype=qtype, active="questions") @bp.route("/questions/new", methods=["GET", "POST"]) @bp.route("/questions//edit", methods=["GET", "POST"]) @login_required def question_edit(id=None): _admin_only() q = Question.query.get_or_404(id) if id else Question( qtype="单选", points=10, difficulty=1, answer="A", category="通用") if request.method == "POST": q.category = request.form.get("category", "通用").strip() or "通用" q.indication = request.form.get("indication", "").strip() q.qtype = request.form.get("qtype", "单选") q.stem = request.form.get("stem", "").strip() q.opt_a = request.form.get("opt_a", "").strip() q.opt_b = request.form.get("opt_b", "").strip() q.opt_c = request.form.get("opt_c", "").strip() q.opt_d = request.form.get("opt_d", "").strip() q.answer = request.form.get("answer", "A").strip().upper() q.explanation = request.form.get("explanation", "").strip() q.points = request.form.get("points", 10, type=int) f = request.files.get("image") if f and f.filename and "." in f.filename and \ f.filename.rsplit(".", 1)[1].lower() in ALLOWED_IMG: ext = f.filename.rsplit(".", 1)[1].lower() fn = "question/%s.%s" % (uuid.uuid4().hex, ext) path = os.path.join(current_app.config["UPLOAD_DIR"], fn) os.makedirs(os.path.dirname(path), exist_ok=True) f.save(path) q.image_url = "/uploads/" + fn elif request.form.get("clear_image"): q.image_url = None if not q.stem: flash("题干必填", "danger") else: if id is None: db.session.add(q) db.session.commit() log_action("question_create" if id is None else "question_edit", target_type="question", target_id=q.id, summary="题目：%s" % (q.stem[:30] if q.stem else "")) flash("已保存", "success") return redirect(url_for("admin.questions")) return render_template("admin/question_form.html", q=q, active="questions", is_new=(id is None)) @bp.route("/questions//delete", methods=["POST"]) @login_required def question_delete(id): _admin_only() q = Question.query.get_or_404(id) stem = q.stem db.session.delete(q) log_action("delete_question", target_type="question", target_id=id, summary="删除题目：%s" % (stem[:30] if stem else ""), commit=False) db.session.commit() flash("已删除", "success") return redirect(url_for("admin.questions")) @bp.route("/questions/batch-delete", methods=["POST"]) @login_required def question_batch_delete(): _admin_only() raw = request.form.get("qids", "") ids = [int(x) for x in raw.split(",") if x.strip().isdigit()] if not ids: flash("未选择题目", "danger") return redirect(url_for("admin.questions")) qs = Question.query.filter(Question.id.in_(ids)).all() n = len(qs) for q in qs: db.session.delete(q) log_action("delete_question_batch", target_type="question", summary="批量删除题目 %d 题" % n, commit=False) db.session.commit() flash("已删除 %d 题" % n, "success") return redirect(url_for("admin.questions")) @bp.route("/questions/import", methods=["GET", "POST"]) @login_required def question_import(): _admin_only() if request.method == "POST": f = request.files.get("file") if not f or not (f.filename or "").lower().endswith((".xlsx", ".xls")): flash("请上传 .xlsx / .xls 文件", "danger") return redirect(url_for("admin.question_import")) from openpyxl import load_workbook wb = load_workbook(f, read_only=True, data_only=True) ws = wb.active rows = list(ws.iter_rows(values_only=True)) added, skipped = 0, 0 for row in rows[1:]: # 列：题干|题型|A|B|C|D|答案|分类|分值|解析 try: stem = str(row[0]).strip() if len(row) > 0 and row[0] else "" if not stem: continue qtype = str(row[1]).strip() if len(row) > 1 and row[1] else "单选" opt_a = str(row[2]).strip() if len(row) > 2 and row[2] else "" opt_b = str(row[3]).strip() if len(row) > 3 and row[3] else "" opt_c = str(row[4]).strip() if len(row) > 4 and row[4] else "" opt_d = str(row[5]).strip() if len(row) > 5 and row[5] else "" answer = str(row[6]).strip().upper() if len(row) > 6 and row[6] else "A" category = str(row[7]).strip() if len(row) > 7 and row[7] else "通用" points = int(row[8]) if len(row) > 8 and row[8] else 10 explanation = str(row[9]).strip() if len(row) > 9 and row[9] else "" db.session.add(Question( category=category, qtype=qtype, stem=stem, opt_a=opt_a, opt_b=opt_b, opt_c=opt_c, opt_d=opt_d, answer=answer, explanation=explanation, points=points, difficulty=1)) added += 1 except Exception: skipped += 1 db.session.commit() log_action("question_import", target_type="question", summary="导入题库：新增 %d，跳过 %d" % (added, skipped)) flash("导入完成：新增 %d 题，跳过 %d 题" % (added, skipped), "success") return redirect(url_for("admin.questions")) return render_template("admin/question_import.html", active="questions")