import json
|
from datetime import datetime
|
|
from fastapi import APIRouter, Depends, File, Form, UploadFile
|
from sqlalchemy import func, select
|
from sqlalchemy.orm import Session
|
|
from app.auth import require_admin
|
from app.database import get_db
|
from app.import_service import parse_batch_exam_ids, purge_batch, run_import
|
from app.models import ExaminationExamExaminee, ExaminationPaperSource, ImportBatch, User
|
|
router = APIRouter(prefix="/api/admin/import", tags=["admin-import"])
|
|
|
@router.get("/batches")
|
def list_batches(
|
db: Session = Depends(get_db),
|
admin: User = Depends(require_admin),
|
):
|
batches = db.scalars(select(ImportBatch).order_by(ImportBatch.imported_at.desc())).all()
|
out = []
|
for b in batches:
|
cnt = db.scalar(
|
select(func.count())
|
.select_from(ExaminationExamExaminee)
|
.where(ExaminationExamExaminee.import_batch_id == b.id)
|
)
|
paper_name = None
|
if b.paper_source_id:
|
ps = db.get(ExaminationPaperSource, b.paper_source_id)
|
paper_name = ps.paper_name if ps else None
|
out.append(
|
{
|
"id": b.id,
|
"subject_name": b.subject_name,
|
"exam_id": b.exam_id,
|
"exam_ids": parse_batch_exam_ids(b),
|
"zip_filename": b.zip_filename,
|
"imported_at": b.imported_at.isoformat() if b.imported_at else "",
|
"status": b.status,
|
"attempt_count": cnt or 0,
|
"paper_name": paper_name,
|
"report": json.loads(b.report_json or "{}"),
|
}
|
)
|
return {"items": out}
|
|
|
@router.post("")
|
async def import_zip(
|
subject_name: str = Form(...),
|
file: UploadFile = File(...),
|
db: Session = Depends(get_db),
|
admin: User = Depends(require_admin),
|
):
|
data = await file.read()
|
return run_import(
|
db,
|
subject_name=subject_name,
|
zip_filename=file.filename or "upload.zip",
|
file_bytes=data,
|
user_id=admin.id,
|
)
|
|
|
@router.delete("/batches/{batch_id}")
|
def delete_batch(
|
batch_id: int,
|
db: Session = Depends(get_db),
|
admin: User = Depends(require_admin),
|
):
|
counts = purge_batch(db, batch_id)
|
return {"deleted_counts": counts}
|
