Implement admin Excel import settings and chat preview

ai_knowledge_base_v2/apps/backend/app/api/admin_users.py

@@ -1,7 +1,13 @@
 from __future__ import annotations
 
 import re
-from fastapi import APIRouter, Depends, HTTPException, Query, status
+from datetime import date, datetime
+from io import BytesIO
+
+from fastapi import APIRouter, Depends, File, HTTPException, Query, UploadFile, status
+from fastapi.responses import StreamingResponse
+from openpyxl import Workbook, load_workbook
+from pydantic import ValidationError
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
@@ -9,12 +15,15 @@ from app.core.database import get_db
 from app.core.dependencies import get_current_admin
 from app.core.responses import api_success
 from app.models.admin import Admin
+from app.models.ai_config import SystemConfig
 from app.models.user import User
-from app.schemas.admin import AdminUserCreateRequest, AdminUserImportRequest, AdminUserUpdateRequest
+from app.schemas.admin import AdminUserCreateRequest, AdminUserImportItem, AdminUserImportRequest, AdminUserUpdateRequest
 from app.services.admin_service import OperationLogService
 
 router = APIRouter()
 
+STUDENT_TEMPLATE_HEADERS = ["手机号", "姓名", "昵称", "每日聊天额度", "状态", "有效期"]
+
 
 @router.get("/user/list")
 def list_users(
@@ -49,7 +58,7 @@ def create_user(
     else:
         user = User(phone=phone, daily_chat_used=0)
 
-    _apply_user_payload(user, payload)
+    _apply_user_payload(user, payload, _default_daily_chat_limit(db))
     db.add(user)
     db.flush()
     OperationLogService.write(db, admin_id=current_admin.id, module="user", action="create", target_id=user.id)
@@ -67,12 +76,104 @@ def import_users(
     if not payload.students:
         raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请提供学员数据")
 
+    return api_success(_import_user_items(list(enumerate(payload.students, start=1)), db, current_admin))
+
+
+@router.get("/user/import/template")
+def download_user_import_template(
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> StreamingResponse:
+    workbook = Workbook()
+    sheet = workbook.active
+    sheet.title = "学员导入模板"
+    sheet.append(STUDENT_TEMPLATE_HEADERS)
+    sheet.append(["13800000001", "张三", "三三", _default_daily_chat_limit(db), "启用", "2026-12-31 23:59:59"])
+    sheet.append(["13800000002", "李四", "", "", "启用", ""])
+
+    widths = [18, 14, 14, 16, 12, 22]
+    for index, width in enumerate(widths, start=1):
+        sheet.column_dimensions[chr(64 + index)].width = width
+
+    stream = BytesIO()
+    workbook.save(stream)
+    stream.seek(0)
+    return StreamingResponse(
+        stream,
+        media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
+        headers={"Content-Disposition": 'attachment; filename="student_import_template.xlsx"'},
+    )
+
+
+@router.post("/user/import/excel")
+def import_users_excel(
+    file: UploadFile = File(...),
+    db: Session = Depends(get_db),
+    current_admin: Admin = Depends(get_current_admin),
+) -> dict:
+    filename = (file.filename or "").lower()
+    if not filename.endswith(".xlsx"):
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="请上传 .xlsx 格式的 Excel 文件")
+
+    content = file.file.read()
+    if not content:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Excel 文件为空")
+
+    try:
+        workbook = load_workbook(BytesIO(content), data_only=True)
+    except Exception as exc:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Excel 文件解析失败") from exc
+
+    sheet = workbook.active
+    header_map = _excel_header_map(next(sheet.iter_rows(min_row=1, max_row=1, values_only=True), ()))
+    for required in ("手机号", "姓名"):
+        if required not in header_map:
+            raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"模板缺少必填列：{required}")
+
+    students: list[tuple[int, AdminUserImportItem]] = []
+    parse_failures: list[dict] = []
+    for row_number, row in enumerate(sheet.iter_rows(min_row=2, values_only=True), start=2):
+        if not any(_cell_text(cell) for cell in row):
+            continue
+        try:
+            item = AdminUserImportItem(
+                phone=_cell_text(_excel_value(row, header_map, "手机号")),
+                name=_cell_text(_excel_value(row, header_map, "姓名")),
+                nickname=_cell_text(_excel_value(row, header_map, "昵称")) or None,
+                dailyChatLimit=_parse_optional_int(_excel_value(row, header_map, "每日聊天额度")),
+                status=_parse_status(_excel_value(row, header_map, "状态")),
+                expiredAt=_parse_optional_datetime(_excel_value(row, header_map, "有效期")),
+            )
+            students.append((row_number, item))
+        except (ValueError, ValidationError) as exc:
+            parse_failures.append(
+                {
+                    "row": row_number,
+                    "phone": _cell_text(_excel_value(row, header_map, "手机号")),
+                    "reason": _row_error_message(exc),
+                }
+            )
+
+    if not students and not parse_failures:
+        raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Excel 中没有可导入的学员数据")
+
+    result = _import_user_items(students, db, current_admin, parse_failures)
+    return api_success(result)
+
+
+def _import_user_items(
+    students: list[tuple[int, AdminUserImportItem]],
+    db: Session,
+    current_admin: Admin,
+    initial_failures: list[dict] | None = None,
+) -> dict:
     created = 0
     updated = 0
-    failed: list[dict] = []
+    failed: list[dict] = list(initial_failures or [])
     seen: set[str] = set()
+    default_daily_limit = _default_daily_chat_limit(db)
 
-    for index, item in enumerate(payload.students, start=1):
+    for row_number, item in students:
         phone = _normalize_phone(item.phone)
         name = item.name.strip()
         try:
@@ -90,10 +191,10 @@ def import_users(
             else:
                 user.is_deleted = 0
                 updated += 1
-            _apply_user_payload(user, item)
+            _apply_user_payload(user, item, default_daily_limit)
             db.add(user)
         except ValueError as exc:
-            failed.append({"row": index, "phone": item.phone, "reason": str(exc)})
+            failed.append({"row": row_number, "phone": item.phone, "reason": str(exc)})
 
     if created or updated:
         db.flush()
@@ -107,7 +208,7 @@ def import_users(
         )
         db.commit()
 
-    return api_success({"created": created, "updated": updated, "failed": len(failed), "failures": failed})
+    return {"created": created, "updated": updated, "failed": len(failed), "failures": failed}
 
 
 @router.get("/user/{user_id}")
@@ -179,21 +280,99 @@ def _user_dict(user: User) -> dict:
     }
 
 
-def _apply_user_payload(user: User, payload: AdminUserCreateRequest) -> None:
-    settings_limit = 100
-    try:
-        from app.core.config import get_settings
-
-        settings_limit = get_settings().default_daily_chat_limit
-    except Exception:
-        pass
+def _apply_user_payload(
+    user: User,
+    payload: AdminUserCreateRequest | AdminUserImportItem,
+    default_daily_limit: int,
+) -> None:
     user.name = payload.name.strip()
     user.nickname = payload.nickname.strip() if payload.nickname else None
     user.status = payload.status
-    user.daily_chat_limit = payload.dailyChatLimit if payload.dailyChatLimit is not None else settings_limit
+    user.daily_chat_limit = payload.dailyChatLimit if payload.dailyChatLimit is not None else default_daily_limit
     user.expired_at = payload.expiredAt.replace(tzinfo=None) if payload.expiredAt is not None else None
 
 
+def _default_daily_chat_limit(db: Session) -> int:
+    config = db.scalar(select(SystemConfig).where(SystemConfig.config_key == "daily_chat_limit"))
+    if config is not None and config.config_value.strip():
+        try:
+            return max(0, min(100000, int(float(config.config_value.strip()))))
+        except ValueError:
+            pass
+    try:
+        from app.core.config import get_settings
+
+        return get_settings().default_daily_chat_limit
+    except Exception:
+        return 100
+
+
+def _excel_header_map(headers: tuple) -> dict[str, int]:
+    return {_cell_text(header): index for index, header in enumerate(headers) if _cell_text(header)}
+
+
+def _excel_value(row: tuple, header_map: dict[str, int], header: str) -> object | None:
+    index = header_map.get(header)
+    if index is None or index >= len(row):
+        return None
+    return row[index]
+
+
+def _cell_text(value: object | None) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, float) and value.is_integer():
+        return str(int(value))
+    return str(value).strip()
+
+
+def _parse_optional_int(value: object | None) -> int | None:
+    text = _cell_text(value)
+    if not text:
+        return None
+    try:
+        parsed = int(float(text))
+    except ValueError as exc:
+        raise ValueError("每日聊天额度必须是数字") from exc
+    if parsed < 0 or parsed > 100000:
+        raise ValueError("每日聊天额度必须在 0 到 100000 之间")
+    return parsed
+
+
+def _parse_status(value: object | None) -> int:
+    text = _cell_text(value).lower()
+    if not text:
+        return 1
+    if text in {"1", "启用", "正常", "是", "true", "enabled"}:
+        return 1
+    if text in {"0", "禁用", "停用", "否", "false", "disabled"}:
+        return 0
+    raise ValueError("状态只能填写启用/禁用或 1/0")
+
+
+def _parse_optional_datetime(value: object | None) -> datetime | None:
+    if value is None or _cell_text(value) == "":
+        return None
+    if isinstance(value, datetime):
+        return value.replace(tzinfo=None)
+    if isinstance(value, date):
+        return datetime.combine(value, datetime.min.time())
+
+    text = _cell_text(value)
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d", "%Y/%m/%d %H:%M:%S", "%Y/%m/%d"):
+        try:
+            return datetime.strptime(text, fmt)
+        except ValueError:
+            continue
+    raise ValueError("有效期格式应为 YYYY-MM-DD 或 YYYY-MM-DD HH:mm:ss")
+
+
+def _row_error_message(exc: ValueError | ValidationError) -> str:
+    if isinstance(exc, ValidationError):
+        return "字段格式不正确"
+    return str(exc)
+
+
 def _normalize_phone(phone: str) -> str:
     return re.sub(r"\D", "", phone or "")
 

ai_knowledge_base_v2/apps/backend/requirements.txt

@@ -9,3 +9,5 @@ PyJWT>=2.10.0
 PyMySQL>=1.1.1
 cryptography>=45.0.0
 httpx>=0.28.0
+openpyxl>=3.1.5
+python-multipart>=0.0.20