GuangYu/QuestionProject
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 支持按权益名称导入用户

Browse Source
This commit is contained in:
Nelson
2026-08-19 15:11:49 +08:00
parent efe835be81
commit 6ffdd2c259
5 changed files with 213 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
ai_knowledge_base_v2/apps/backend/app/api/admin_users.py
View File

@@ -35,7 +35,7 @@ from app.api.pagination import page_result
router = APIRouter()
STUDENT_TEMPLATE_HEADERS = ["手机号", "姓名", "昵称", "每日聊天额度", "状态", "有效期"]
STUDENT_TEMPLATE_HEADERS = ["手机号", "姓名", "昵称", "每日聊天额度", "状态", "有效期", "权益名称"]
class AdminGenerateReportRequest(BaseModel):
@@ -155,17 +155,7 @@ 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
workbook = _user_import_template_workbook(db)
stream = BytesIO()
workbook.save(stream)
stream.seek(0)
@@ -176,6 +166,32 @@ def download_user_import_template(
)
def _user_import_template_workbook(db: Session) -> Workbook:
workbook = Workbook()
sheet = workbook.active
sheet.title = "学员导入模板"
sheet.append(STUDENT_TEMPLATE_HEADERS)
available_plan = EntitlementService.list_plans(db)
example_plan_name = available_plan[0].name if available_plan else ""
sheet.append(
[
"13800000001",
"张三",
"三三",
_default_daily_chat_limit(db),
"启用",
"2026-12-31 23:59:59",
example_plan_name,
]
)
sheet.append(["13800000002", "李四", "", "", "启用", "", ""])
widths = [18, 14, 14, 16, 12, 22, 22]
for index, width in enumerate(widths, start=1):
sheet.column_dimensions[chr(64 + index)].width = width
return workbook
@router.post("/user/import/excel")
def import_users_excel(
file: UploadFile = File(...),
@@ -214,6 +230,7 @@ def import_users_excel(
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, "有效期")),
entitlementName=_cell_text(_excel_value(row, header_map, "权益名称")) or None,
)
students.append((row_number, item))
except (ValueError, ValidationError) as exc:
@@ -243,6 +260,11 @@ def _import_user_items(
failed: list[dict] = list(initial_failures or [])
seen: set[str] = set()
default_daily_limit = _default_daily_chat_limit(db)
entitlement_plans = {
plan.name.strip(): plan
for plan in EntitlementService.list_plans(db)
if plan.name.strip()
}
for row_number, item in students:
phone = _normalize_phone(item.phone)
@@ -264,6 +286,28 @@ def _import_user_items(
updated += 1
_apply_user_payload(user, item, default_daily_limit)
db.add(user)
db.flush()
entitlement_name = (item.entitlementName or "").strip()
entitlement_plan = entitlement_plans.get(entitlement_name)
if entitlement_plan is not None:
EntitlementService.assign_user_plan(
db,
user=user,
plan_id=entitlement_plan.id,
operated_by=current_admin.id,
remark="用户批量导入按权益名称匹配",
)
else:
EntitlementService.clear_user_plan(
db,
user=user,
operated_by=current_admin.id,
reason=(
f"导入的权益名称不存在：{entitlement_name}"
if entitlement_name
else "导入时未填写权益名称"
),
)
except ValueError as exc:
failed.append({"row": row_number, "phone": item.phone, "reason": str(exc)})

1
ai_knowledge_base_v2/apps/backend/app/schemas/admin.py
View File

@@ -99,6 +99,7 @@ class AdminUserImportItem(BaseModel):
status: int = Field(default=1, ge=0, le=1)
dailyChatLimit: int | None = Field(default=None, ge=0, le=100000)
expiredAt: datetime | None = None
entitlementName: str | None = Field(default=None, max_length=80)
class AdminUserImportRequest(BaseModel):

35
ai_knowledge_base_v2/apps/backend/app/services/entitlement_service.py
View File

@@ -312,6 +312,41 @@ class EntitlementService:
)
return entitlement
@staticmethod
def clear_user_plan(
db: Session,
*,
user: User,
operated_by: int | None,
reason: str | None = None,
) -> None:
"""Remove explicit entitlements so the user returns to the unassigned/default state."""
current_rows = list(
db.scalars(
select(UserEntitlement)
.where(UserEntitlement.user_id == user.id, UserEntitlement.status == "active")
.order_by(UserEntitlement.created_at.desc(), UserEntitlement.id.desc())
)
)
if not current_rows:
return
for current in current_rows:
current.status = "replaced"
db.add(current)
latest = current_rows[0]
db.add(
UserEntitlementLog(
user_id=user.id,
entitlement_id=latest.id,
from_plan_id=latest.plan_id,
to_plan_id=None,
action="clear",
detail_json=json.dumps({"reason": reason}, ensure_ascii=False),
operated_by=operated_by,
created_at=_now(),
)
)
def plan_dict(plan: EntitlementPlan) -> dict:
return {

120
ai_knowledge_base_v2/apps/backend/tests/test_admin_user_import.py Normal file
View File

@@ -0,0 +1,120 @@
from __future__ import annotations
from types import SimpleNamespace
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
from app.api.admin_users import (
STUDENT_TEMPLATE_HEADERS,
_import_user_items,
_user_import_template_workbook,
)
from app.models import Base
from app.models.entitlement import EntitlementPlan, UserEntitlement
from app.models.user import User
from app.schemas.admin import AdminUserImportItem
def _database() -> Session:
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine)
return Session(engine)
def test_user_import_template_contains_entitlement_name_and_uses_available_plan_example():
with _database() as db:
db.add_all(
[
EntitlementPlan(id=1, name="深度陪伴版", plan_type="deep", status=1, sort_order=1),
EntitlementPlan(id=2, name="已停用版", plan_type="basic", status=0, sort_order=0),
]
)
db.commit()
sheet = _user_import_template_workbook(db).active
assert [cell.value for cell in sheet[1]] == STUDENT_TEMPLATE_HEADERS
assert sheet.cell(row=2, column=7).value == "深度陪伴版"
assert sheet.cell(row=3, column=7).value == ""
def test_user_import_matches_entitlement_by_name_and_clears_missing_or_unknown_names():
with _database() as db:
available = EntitlementPlan(id=1, name="深度陪伴版", plan_type="deep", status=1)
old_plan = EntitlementPlan(id=2, name="旧权益", plan_type="basic", status=1)
disabled = EntitlementPlan(id=3, name="停用权益", plan_type="basic", status=0)
existing_user = User(id=10, phone="13800000010", name="旧姓名", daily_chat_limit=10)
db.add_all([available, old_plan, disabled, existing_user])
db.flush()
db.add(UserEntitlement(user_id=existing_user.id, plan_id=old_plan.id, status="active"))
db.commit()
result = _import_user_items(
[
(
2,
AdminUserImportItem(
phone="13800000011",
name="匹配用户",
entitlementName="深度陪伴版",
),
),
(
3,
AdminUserImportItem(
phone="13800000010",
name="更新用户",
entitlementName="系统不存在的权益",
),
),
(
4,
AdminUserImportItem(
phone="13800000012",
name="留空用户",
entitlementName=None,
),
),
(
5,
AdminUserImportItem(
phone="13800000013",
name="停用权益用户",
entitlementName="停用权益",
),
),
],
db,
SimpleNamespace(id=99),
)
assert result == {"created": 3, "updated": 1, "failed": 0, "failures": []}
matched_user = db.scalar(select(User).where(User.phone == "13800000011"))
matched_entitlement = db.scalar(
select(UserEntitlement).where(
UserEntitlement.user_id == matched_user.id,
UserEntitlement.status == "active",
)
)
assert matched_entitlement is not None
assert matched_entitlement.plan_id == available.id
for phone in ("13800000010", "13800000012", "13800000013"):
user = db.scalar(select(User).where(User.phone == phone))
assert db.scalar(
select(UserEntitlement).where(
UserEntitlement.user_id == user.id,
UserEntitlement.status == "active",
)
) is None
old_entitlement = db.scalar(
select(UserEntitlement).where(UserEntitlement.user_id == existing_user.id)
)
assert old_entitlement.status == "replaced"