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

feat: improve audit exports and admin analytics

Browse Source
This commit is contained in:
Nelson
2026-08-25 18:49:56 +08:00
parent 910e67bd89
commit 5740bd26e7
55 changed files with 1654 additions and 269 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
ai_knowledge_base_v2/apps/backend/tests/test_admin_pagination.py
View File

@@ -101,7 +101,14 @@ def test_chat_detail_messages_are_paginated():
])
db.commit()
response = chat_detail(1, messagePage=2, messagePageSize=10, db=db, current_admin=object())
response = chat_detail(
1,
messagePage=2,
messagePageSize=10,
focusMessageId=None,
db=db,
current_admin=object(),
)
data = response["data"]
assert data["session"]["messageCount"] == 25
@@ -111,6 +118,40 @@ def test_chat_detail_messages_are_paginated():
assert data["messages"][0]["content"] == "消息11"
def test_chat_detail_focus_message_opens_its_page_in_chronological_order():
with _database() as db:
user = User(id=1, phone="13800000000", name="学员", daily_chat_limit=10)
session = ChatSession(id=1, user_id=1, title="长会话", message_count=25)
created_at = datetime(2026, 8, 25, 10, 0, 0)
db.add_all([user, session])
db.add_all([
ChatMessage(
id=index + 1,
session_id=1,
user_id=1,
role="user",
content=f"消息{index + 1}",
created_at=created_at,
)
for index in range(25)
])
db.commit()
response = chat_detail(
1,
messagePage=1,
messagePageSize=10,
focusMessageId=23,
db=db,
current_admin=object(),
)
data = response["data"]
assert data["messagesPage"]["page"] == 3
assert data["focusedMessageId"] == 23
assert [item["id"] for item in data["messages"]] == [21, 22, 23, 24, 25]
def test_chat_messages_endpoint_returns_only_message_page():
with _database() as db:
user = User(id=1, phone="13800000000", name="学员", daily_chat_limit=10)
@@ -149,11 +190,75 @@ def test_retrieval_and_attention_lists_are_paginated():
sessionId=None, knowledgeId=None, knowledgeCalled=None, statusValue="", attentionCreated=None,
page=3, pageSize=10, db=db, current_admin=object(),
)["data"]
attention = attention_list(priority="", statusValue="", page=3, pageSize=10, db=db, current_admin=object())["data"]
attention = attention_list(
priority="",
statusValue="",
dateFrom=None,
dateTo=None,
page=3,
pageSize=10,
db=db,
current_admin=object(),
)["data"]
assert retrieval["total"] == attention["total"] == 21
assert len(retrieval["items"]) == len(attention["items"]) == 1
def test_attention_list_filters_by_priority_and_trigger_time():
with _database() as db:
db.add_all([
HumanAttentionRecord(
session_id=1,
message_id=1,
user_id=1,
trigger_message="紧急问题",
problem_summary="紧急问题",
trigger_reason="测试",
priority="urgent",
status="pending",
created_at=datetime(2026, 8, 20, 9, 0, 0),
),
HumanAttentionRecord(
session_id=2,
message_id=2,
user_id=2,
trigger_message="普通问题",
problem_summary="普通问题",
trigger_reason="测试",
priority="normal",
status="pending",
created_at=datetime(2026, 8, 21, 9, 0, 0),
),
HumanAttentionRecord(
session_id=3,
message_id=3,
user_id=3,
trigger_message="过期紧急问题",
problem_summary="过期紧急问题",
trigger_reason="测试",
priority="urgent",
status="pending",
created_at=datetime(2026, 8, 10, 9, 0, 0),
),
])
db.commit()
data = attention_list(
priority="urgent",
statusValue="",
dateFrom=datetime(2026, 8, 20, 0, 0, 0),
dateTo=datetime(2026, 8, 22, 23, 59, 59),
page=1,
pageSize=10,
db=db,
current_admin=object(),
)["data"]
assert data["total"] == 1
assert data["items"][0]["messageId"] == 1
assert data["items"][0]["targetAvailable"] is False
def test_question_insights_clean_and_cluster_similar_user_questions():
with _database() as db:
user_a = User(id=1, phone="13800000000", name="学员A", daily_chat_limit=10)

58
ai_knowledge_base_v2/apps/backend/tests/test_admin_permissions.py
View File

@@ -16,8 +16,9 @@ def test_super_admin_has_all_permissions() -> None:
assert permissions_for(admin) == ALL_PERMISSION_CODES
assert {"feedback.view", "feedback.detail", "feedback.export", "feedback.delete"} <= ALL_PERMISSION_CODES
assert "prompt.batch" in ALL_PERMISSION_CODES
assert "behavior.view" in ALL_PERMISSION_CODES
assert {"behavior.view", "behavior.export"} <= ALL_PERMISSION_CODES
assert {"attention.view", "attention.edit", "attention.config", "attention.preview"} <= ALL_PERMISSION_CODES
assert {"records.view", "records.export"} <= ALL_PERMISSION_CODES
def test_role_permissions_are_restricted_to_catalog() -> None:
@@ -81,6 +82,61 @@ def test_user_behavior_routes_require_behavior_permission() -> None:
enforce_admin_access(request, denied)
assert exc.value.status_code == 403
export_request = Request({"type": "http", "method": "GET", "path": "/api/admin/user-behavior/export", "headers": []})
with pytest.raises(HTTPException) as exc:
enforce_admin_access(export_request, allowed)
assert exc.value.status_code == 403
export_role = Role(code="behavior-exporter", name="行为报表导出", permissions=json.dumps(["behavior.export"]))
exporter = Admin(
id=6,
username="behavior-exporter",
password="hash",
name="行为报表导出",
status=1,
must_change_password=0,
is_super_admin=0,
role=export_role,
)
assert enforce_admin_access(export_request, exporter) is exporter
@pytest.mark.parametrize(
"path",
[
"/api/admin/chat/export",
"/api/admin/chat/14/export",
"/api/admin/question-insights/export",
],
)
def test_record_export_routes_require_export_permission(path: str) -> None:
viewer = Admin(
id=7,
username="records-viewer",
password="hash",
name="记录查看员",
status=1,
must_change_password=0,
is_super_admin=0,
role=Role(code="records-view", name="记录查看", permissions=json.dumps(["records.view"])),
)
request = Request({"type": "http", "method": "GET", "path": path, "headers": []})
with pytest.raises(HTTPException) as exc:
enforce_admin_access(request, viewer)
assert exc.value.status_code == 403
exporter = Admin(
id=8,
username="records-exporter",
password="hash",
name="记录导出员",
status=1,
must_change_password=0,
is_super_admin=0,
role=Role(code="records-export", name="记录导出", permissions=json.dumps(["records.export"])),
)
assert enforce_admin_access(request, exporter) is exporter
@pytest.mark.parametrize(
("path", "method", "permission"),

63
ai_knowledge_base_v2/apps/backend/tests/test_chat_export.py Normal file
View File

@@ -0,0 +1,63 @@
from __future__ import annotations
from datetime import datetime
from openpyxl import load_workbook
from app.core.ai_content_label import AI_GENERATED_NOTICE
from app.models.chat import ChatMessage, ChatSession
from app.models.user import User
from app.services.chat_export_service import ChatExportService
def test_chat_export_contains_all_messages_in_order_and_ai_notice() -> None:
session = ChatSession(
id=9,
user_id=3,
title="完整对话测试",
message_count=99,
last_message_at=datetime(2026, 8, 25, 8, 2),
updated_at=datetime(2026, 8, 25, 8, 3),
)
user = User(id=3, phone="13800000000", name="测试学员", daily_chat_limit=10)
messages = [
ChatMessage(
id=21,
session_id=9,
user_id=3,
role="user",
content="第一个问题",
message_status="FINISHED",
created_at=datetime(2026, 8, 25, 8),
),
ChatMessage(
id=22,
session_id=9,
user_id=3,
role="assistant",
content="第二个回答",
message_status="FINISHED",
token_input=20,
token_output=10,
response_time_ms=800,
created_at=datetime(2026, 8, 25, 8, 1),
),
]
stream = ChatExportService.build(
session=session,
user=user,
source_name="千问千答直接访问",
messages=messages,
)
workbook = load_workbook(stream)
assert workbook.sheetnames == ["会话信息", "完整对话"]
assert workbook["会话信息"]["B8"].value == 2
sheet = workbook["完整对话"]
assert sheet.max_row == 3
assert sheet["B2"].value == 21
assert sheet["E2"].value == "第一个问题"
assert sheet["B3"].value == 22
assert AI_GENERATED_NOTICE in sheet["E3"].value
assert sheet["D2"].value == datetime(2026, 8, 25, 16)

4
ai_knowledge_base_v2/apps/backend/tests/test_message_feedback.py
View File

@@ -1,6 +1,6 @@
from __future__ import annotations
from datetime import datetime
from datetime import datetime, timedelta
from types import SimpleNamespace
from sqlalchemy import create_engine
@@ -53,5 +53,5 @@ def test_feedback_export_workbook_is_formatted_and_formula_safe() -> None:
assert sheet["E2"].value.startswith("'=")
assert sheet["H1"].value == "对应AI回答（AI生成）"
assert "AI生成内容，请结合实际情况核对后使用。" in sheet["H2"].value
assert sheet["I2"].value == created_at
assert sheet["I2"].value == created_at + timedelta(hours=8)
assert sheet.tables["FeedbackRecords"].ref == "A1:J2"

27
ai_knowledge_base_v2/apps/backend/tests/test_time_contract.py Normal file
View File

@@ -0,0 +1,27 @@
from __future__ import annotations
from datetime import date, datetime
from zoneinfo import ZoneInfo
from app.core.responses import api_success
from app.core.time_utils import business_day_boundary, to_business_naive
def test_api_success_serializes_naive_database_time_as_explicit_utc() -> None:
response = api_success({"createdAt": datetime(2026, 8, 25, 8, 30, 12, 123456)})
assert response["data"]["createdAt"] == "2026-08-25T08:30:12.123Z"
def test_api_success_normalizes_aware_business_time_to_utc() -> None:
response = api_success(
{"createdAt": datetime(2026, 8, 25, 16, 30, tzinfo=ZoneInfo("Asia/Shanghai"))}
)
assert response["data"]["createdAt"] == "2026-08-25T08:30:00.000Z"
def test_beijing_day_boundary_and_excel_display_conversion() -> None:
assert business_day_boundary(date(2026, 8, 25)) == datetime(2026, 8, 24, 16)
assert business_day_boundary(date(2026, 8, 25), end_exclusive=True) == datetime(2026, 8, 25, 16)
assert to_business_naive(datetime(2026, 8, 25, 8, 30)) == datetime(2026, 8, 25, 16, 30)

58
ai_knowledge_base_v2/apps/backend/tests/test_user_behavior_analytics.py
View File

@@ -1,7 +1,9 @@
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from io import BytesIO
from openpyxl import load_workbook
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
@@ -11,19 +13,27 @@ from app.models.behavior import UserBehaviorEvent
from app.models.user import User
from app.schemas.behavior import UserBehaviorEventCreate
from app.services.user_behavior_service import UserBehaviorService
from app.services.user_behavior_export_service import UserBehaviorExportService
def _engine():
return create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool)
def _event(event_id: str, code: str, *, target_type: str | None = None, target_id: int | None = None):
def _event(
event_id: str,
code: str,
*,
target_type: str | None = None,
target_id: int | None = None,
occurred_at: datetime | None = None,
):
return UserBehaviorEventCreate(
clientEventId=event_id,
eventCode=code,
targetType=target_type,
targetId=target_id,
occurredAt=datetime.now(UTC),
occurredAt=occurred_at or datetime.now(UTC),
)
@@ -67,18 +77,19 @@ def test_behavior_overview_user_list_and_timeline() -> None:
]
db.add_all(users)
db.commit()
base_time = datetime.now(UTC) - timedelta(minutes=5)
UserBehaviorService.record_batch(
db,
user=users[0],
items=[
_event("11111111-1111-1111-1111-111111111111", "app_open"),
_event("22222222-2222-2222-2222-222222222222", "send_question_click", target_type="session", target_id=3),
_event("11111111-1111-1111-1111-111111111111", "app_open", occurred_at=base_time),
_event("22222222-2222-2222-2222-222222222222", "send_question_click", target_type="session", target_id=3, occurred_at=base_time + timedelta(minutes=1)),
],
)
UserBehaviorService.record_batch(
db,
user=users[1],
items=[_event("33333333-3333-3333-3333-333333333333", "app_open")],
items=[_event("33333333-3333-3333-3333-333333333333", "app_open", occurred_at=base_time + timedelta(minutes=2))],
)
overview = UserBehaviorService.overview(db, start=None, end=None)
assert overview["totalEvents"] == 3
@@ -90,9 +101,46 @@ def test_behavior_overview_user_list_and_timeline() -> None:
listed = UserBehaviorService.users(db, start=None, end=None, keyword="甲", page=1, page_size=20)
assert listed["total"] == 1
assert listed["items"][0]["eventCount"] == 2
by_count = UserBehaviorService.users(
db,
start=None,
end=None,
keyword="",
page=1,
page_size=20,
sort_by="eventCount",
sort_order="asc",
)
assert [item["userName"] for item in by_count["items"]] == ["乙用户", "甲用户"]
by_last_time = UserBehaviorService.users(
db,
start=None,
end=None,
keyword="",
page=1,
page_size=20,
sort_by="lastEventAt",
sort_order="desc",
)
assert [item["userName"] for item in by_last_time["items"]] == ["乙用户", "甲用户"]
timeline = UserBehaviorService.timeline(db, user_id=1, start=None, end=None, page=1, page_size=50)
assert [item["eventName"] for item in timeline["items"]] == ["进入答疑页面", "发送问题"]
workbook_stream = UserBehaviorExportService.build_workbook(
db,
start=None,
end=None,
keyword="",
sort_by="eventCount",
sort_order="asc",
)
workbook = load_workbook(BytesIO(workbook_stream.getvalue()))
assert workbook.sheetnames == ["报表说明", "每日趋势", "功能使用频率", "用户汇总"]
assert workbook["报表说明"]["B11"].value == 3
assert workbook["用户汇总"]["B2"].value == "乙用户"
assert workbook["用户汇总"]["D2"].value == 1
assert workbook["用户汇总"]["C2"].number_format == "@"
def test_behavior_retention_deletes_only_expired_rows() -> None:
engine = _engine()