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-24 17:49:56 +08:00
parent 6ffdd2c259
commit 910e67bd89
53 changed files with 1808 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -223,6 +223,7 @@ def test_question_insights_clean_and_cluster_similar_user_questions():
assert workbook["洞察结果"]["J2"].value == "是"
assert workbook["相似问法与样例"].max_row > 2
assert workbook["统计说明"]["B10"].value == 1
assert workbook["统计说明"]["B12"].value == "AI生成内容，请结合实际情况核对后使用。"
def test_question_insight_export_rejects_reversed_date_range():

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

@@ -17,6 +17,7 @@ def test_super_admin_has_all_permissions() -> None:
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 {"attention.view", "attention.edit", "attention.config", "attention.preview"} <= ALL_PERMISSION_CODES
def test_role_permissions_are_restricted_to_catalog() -> None:
@@ -79,3 +80,30 @@ def test_user_behavior_routes_require_behavior_permission() -> None:
with pytest.raises(HTTPException) as exc:
enforce_admin_access(request, denied)
assert exc.value.status_code == 403
@pytest.mark.parametrize(
("path", "method", "permission"),
[
("/api/admin/attention/config", "GET", "attention.config"),
("/api/admin/attention/config", "PUT", "attention.config"),
("/api/admin/attention/preview/config", "GET", "attention.preview"),
("/api/admin/attention/preview/messages", "GET", "attention.preview"),
("/api/admin/attention/preview", "POST", "attention.preview"),
],
)
def test_attention_config_and_preview_routes_use_granular_permissions(path: str, method: str, permission: str) -> None:
role = Role(code="attention-specialist", name="人工关注专员", permissions=json.dumps([permission]))
admin = Admin(
id=8,
username="attention-specialist",
password="hash",
name="人工关注专员",
status=1,
must_change_password=0,
is_super_admin=0,
role=role,
)
request = Request({"type": "http", "method": method, "path": path, "headers": []})
assert enforce_admin_access(request, admin) is admin

5
ai_knowledge_base_v2/apps/backend/tests/test_agent_batch_tests.py
View File

@@ -106,11 +106,14 @@ def test_create_job_snapshots_config_and_export_keeps_failed_rows() -> None:
exported = AgentBatchTestService.export_workbook(job, items)
workbook = load_workbook(exported, data_only=True)
sheet = workbook["批量测试结果"]
assert sheet["C2"].value == "第一个答案"
assert sheet["C1"].value == "答案（AI生成）"
assert sheet["C2"].value.startswith("第一个答案")
assert "AI生成内容，请结合实际情况核对后使用。" in sheet["C2"].value
assert sheet["D2"].value == "成功"
assert sheet["D3"].value == "失败"
assert sheet["E3"].value == "供应商超时"
assert sheet.freeze_panes == "A2"
assert workbook["任务信息"]["B10"].value == "AI生成内容，请结合实际情况核对后使用。"
def test_import_rejects_excel_formula_question() -> None:

3
ai_knowledge_base_v2/apps/backend/tests/test_content_generation_configs.py
View File

@@ -35,6 +35,7 @@ def test_template_preview_keeps_locked_notice_and_rejects_unknown_variables():
assert "问题：我第一次参加带练" in content
assert "不会自动发送给老师" in content
assert "不代表已经转人工处理" in content
assert content.count("AI生成内容，请结合实际情况核对后使用。") == 1
with pytest.raises(HTTPException) as error:
ContentGenerationConfigService.preview(
@@ -64,8 +65,10 @@ def test_weekly_and_monthly_report_defaults_have_independent_variables_and_safet
assert "本周纳入 36 条聊天消息" in weekly
assert "本周报告根据报告周期内的聊天记录自动整理" in weekly
assert "AI生成内容" in weekly
assert "本月纳入 5 份周报告" in monthly
assert "本月报告根据本月覆盖的周报告自动整理" in monthly
assert "AI生成内容" in monthly
assert {item["name"] for item in default_variables("weekly_report")} != {
item["name"] for item in default_variables("monthly_report")
}

6
ai_knowledge_base_v2/apps/backend/tests/test_help_cards.py
View File

@@ -15,7 +15,7 @@ from app.models.entitlement import EntitlementPlan
from app.models.growth import TeacherHelpCard
from app.models.user import User
from app.services.help_card_service import HelpCardService
from app.services.help_card_service import _format_time
from app.services.help_card_service import _format_time, help_card_dict
def _db() -> Session:
@@ -66,6 +66,7 @@ def test_generate_help_card_from_topic_summary_and_mark_copied():
assert "给老师的求助卡" in card.content
assert "自定义模板" in card.content
assert "不会自动发送给老师" in card.content
assert "AI生成内容" in card.content
assert "阴影人格练习步骤是否正确" in card.content
assert "情绪 / 身体感受" not in card.content
assert "已经尝试过或被建议的功课" not in card.content
@@ -76,6 +77,9 @@ def test_generate_help_card_from_topic_summary_and_mark_copied():
assert copied.copied == 1
assert copied.copied_at is not None
payload = help_card_dict(copied)
assert payload["aiGenerated"] is True
assert payload["content"].count("AI生成内容，请结合实际情况核对后使用。") == 1
other_user = User(id=2, phone="13800000002", name="其他学员", daily_chat_limit=100, daily_chat_used=0)
db.add(other_user)

214
ai_knowledge_base_v2/apps/backend/tests/test_human_attention_config.py Normal file
View File

@@ -0,0 +1,214 @@
from __future__ import annotations
from types import SimpleNamespace
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
from app.models import Base
from app.models.knowledge import HumanAttentionHistory, HumanAttentionJob, HumanAttentionRecord
from app.services.human_attention_service import (
DEFAULT_ATTENTION_PROMPT,
HumanAttentionService,
)
from app.services.human_attention_worker import HumanAttentionWorker
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 _config_payload(**overrides) -> dict:
payload = {
"enabled": True,
"keywordEnabled": True,
"aiEnabled": False,
"knowledgeMissingEnabled": True,
"urgentTerms": ["危险词"],
"importantTerms": ["重要词"],
"normalTerms": ["找人工"],
"promptTemplate": "自定义筛选提示词",
"recognitionItems": [
{"name": "复杂卡住", "description": "用户反复沟通后仍明显卡住", "priority": "important", "enabled": True}
],
}
payload.update(overrides)
return payload
def test_default_config_preserves_previous_hardcoded_behavior() -> None:
with _database() as db:
config = HumanAttentionService.get_config(db)
assert config.enabled is True
assert config.keyword_enabled is True
assert config.ai_enabled is False
assert "自杀" in config.urgent_terms
assert config.prompt_template == DEFAULT_ATTENTION_PROMPT
def test_config_can_be_saved_and_loaded() -> None:
with _database() as db:
saved = HumanAttentionService.save_config(db, _config_payload(aiEnabled=True), admin_id=9)
db.commit()
loaded = HumanAttentionService.get_config(db)
assert saved == loaded
assert loaded.ai_enabled is True
assert loaded.urgent_terms == ("危险词",)
assert loaded.prompt_template == "自定义筛选提示词"
def test_configured_keyword_and_knowledge_missing_rules_are_applied() -> None:
with _database() as db:
config = HumanAttentionService.save_config(db, _config_payload(), admin_id=9)
urgent = HumanAttentionService.evaluate(
db,
question="这里出现危险词",
answer="回答",
knowledge_missing=False,
config=config,
)
missing = HumanAttentionService.evaluate(
db,
question="普通课程问题",
answer="回答",
knowledge_missing=True,
config=config,
)
assert urgent.needs_attention is True
assert urgent.priority == "urgent"
assert urgent.source == "keyword"
assert missing.needs_attention is True
assert missing.source == "knowledge_missing"
def test_ai_preview_uses_unsaved_prompt_and_returns_structured_result(monkeypatch) -> None:
def fake_generate(*_args, **kwargs):
assert "尚未保存的提示词" in kwargs["prompt"]
assert "历史问题" in kwargs["prompt"]
return SimpleNamespace(
answer='{"needsAttention":true,"matchedItem":"复杂卡住","priority":"important","reason":"需要人工判断","summary":"历史问题摘要"}'
)
monkeypatch.setattr("app.services.human_attention_service.TrackedGenerationService.generate", fake_generate)
with _database() as db:
result = HumanAttentionService.preview(
db,
question="历史问题",
answer="历史回答",
knowledge_missing=False,
config_payload=_config_payload(
keywordEnabled=False,
knowledgeMissingEnabled=False,
aiEnabled=True,
promptTemplate="尚未保存的提示词",
),
user_id=7,
)
assert result.needs_attention is True
assert result.priority == "important"
assert result.source == "ai"
assert result.matched_item == "复杂卡住"
assert "尚未保存的提示词" in result.rendered_prompt
def test_create_if_needed_is_idempotent_for_same_message() -> None:
with _database() as db:
HumanAttentionService.save_config(db, _config_payload(), admin_id=9)
first = HumanAttentionService.create_if_needed(
db,
session_id=1,
message_id=11,
user_id=7,
question="这里出现重要词",
answer="回答",
knowledge_missing=False,
)
second = HumanAttentionService.create_if_needed(
db,
session_id=1,
message_id=11,
user_id=7,
question="这里出现重要词",
answer="回答",
knowledge_missing=False,
)
assert first is second
assert db.query(HumanAttentionRecord).count() == 1
assert db.query(HumanAttentionHistory).count() == 1
def test_runtime_ai_screening_is_queued_without_blocking_chat(monkeypatch) -> None:
def unexpected_generate(*_args, **_kwargs):
raise AssertionError("AI screening must not run in the chat transaction")
monkeypatch.setattr("app.services.human_attention_service.TrackedGenerationService.generate", unexpected_generate)
with _database() as db:
HumanAttentionService.save_config(
db,
_config_payload(aiEnabled=True, keywordEnabled=False, knowledgeMissingEnabled=False),
admin_id=9,
)
record = HumanAttentionService.create_if_needed(
db,
session_id=1,
message_id=21,
user_id=7,
question="需要语义判断的历史问题",
answer="历史回答",
knowledge_missing=False,
)
assert record is None
job = db.scalar(db.query(HumanAttentionJob).where(HumanAttentionJob.message_id == 21).statement)
assert job is not None
assert job.status == "pending"
assert "复杂卡住" in job.config_snapshot
def test_worker_processes_ai_screening_job(monkeypatch) -> None:
def fake_generate(*_args, **_kwargs):
return SimpleNamespace(
answer='{"needsAttention":true,"matchedItem":"复杂卡住","priority":"normal",'
'"reason":"多轮沟通后仍未解决","summary":"需要老师跟进"}'
)
monkeypatch.setattr("app.services.human_attention_service.TrackedGenerationService.generate", fake_generate)
with _database() as db:
HumanAttentionService.save_config(
db,
_config_payload(aiEnabled=True, keywordEnabled=False, knowledgeMissingEnabled=False),
admin_id=9,
)
HumanAttentionService.create_if_needed(
db,
session_id=1,
message_id=22,
user_id=7,
question="我尝试了很多次还是不知道怎么办",
answer="可以再试一次",
knowledge_missing=False,
)
job_id = HumanAttentionWorker.claim_next(db, worker_id="test-worker")
job = HumanAttentionWorker.execute_claimed(db, job_id=job_id, worker_id="test-worker")
assert job is not None
assert job.status == "completed"
record = db.scalar(db.query(HumanAttentionRecord).where(HumanAttentionRecord.message_id == 22).statement)
assert record is not None
assert record.priority == "important"
assert record.trigger_reason.startswith("复杂卡住：")

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

@@ -51,5 +51,7 @@ def test_feedback_export_workbook_is_formatted_and_formula_safe() -> None:
assert sheet.freeze_panes == "A2"
assert sheet["E2"].data_type == "s"
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.tables["FeedbackRecords"].ref == "A1:J2"

3
ai_knowledge_base_v2/apps/backend/tests/test_periodic_reports.py
View File

@@ -651,6 +651,7 @@ def test_user_report_payload_exposes_async_status_without_internal_error():
assert payload["status"] == "failed"
assert payload["content"] == ""
assert payload["aiGenerated"] is True
assert "errorMessage" not in payload
@@ -674,3 +675,5 @@ def test_report_payload_tracks_chat_and_weekly_report_sources():
assert payload["sourceMessageIds"] == [1, 2]
assert payload["sourceReportIds"] == [11, 12]
assert payload["aiGenerated"] is True
assert "AI生成内容，请结合实际情况核对后使用。" in payload["content"]

52
ai_knowledge_base_v2/apps/backend/tests/test_public_site_config.py Normal file
View File

@@ -0,0 +1,52 @@
from __future__ import annotations
import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.pool import StaticPool
from app.models.ai_config import SystemConfig
from app.models.base import Base
from app.services.public_site_config_service import PublicSiteConfigService
def _database() -> Session:
engine = create_engine(
"sqlite:///:memory:",
connect_args={"check_same_thread": False},
poolclass=StaticPool,
)
Base.metadata.create_all(engine, tables=[SystemConfig.__table__])
return Session(engine)
def test_public_site_config_is_empty_when_filing_text_is_not_configured() -> None:
with _database() as db:
db.add(SystemConfig(config_key="site_filing_url", config_value="https://beian.miit.gov.cn/"))
db.commit()
assert PublicSiteConfigService.public_config(db) == {"filingText": "", "filingUrl": ""}
def test_public_site_config_returns_trimmed_text_and_safe_url() -> None:
with _database() as db:
db.add_all(
[
SystemConfig(config_key="site_filing_text", config_value=" 京ICP备12345678号-1 "),
SystemConfig(config_key="site_filing_url", config_value=" https://beian.miit.gov.cn/ "),
]
)
db.commit()
assert PublicSiteConfigService.public_config(db) == {
"filingText": "京ICP备12345678号-1",
"filingUrl": "https://beian.miit.gov.cn/",
}
def test_filing_url_rejects_unsafe_protocol() -> None:
with pytest.raises(HTTPException) as exc:
PublicSiteConfigService.normalize_admin_value("site_filing_url", "javascript:alert(1)")
assert exc.value.status_code == 400

6
ai_knowledge_base_v2/apps/backend/tests/test_share_drafts.py
View File

@@ -14,7 +14,7 @@ from app.models.chat import ChatMessage, ChatSession, TopicSession
from app.models.entitlement import EntitlementPlan
from app.models.growth import ShareDraft
from app.models.user import User
from app.services.share_draft_service import ShareDraftService
from app.services.share_draft_service import ShareDraftService, share_draft_dict
def _db() -> Session:
@@ -65,6 +65,7 @@ def test_generate_share_draft_from_topic_summary_and_mark_copied():
assert "实修分享稿草稿" in draft.content
assert "自定义模板" in draft.content
assert "系统不会自动发送到任何群" in draft.content
assert "AI生成内容" in draft.content
assert "不代表结论" in draft.content
assert "情绪和身体反应" not in draft.content
assert "做了什么功课" not in draft.content
@@ -77,6 +78,9 @@ def test_generate_share_draft_from_topic_summary_and_mark_copied():
assert copied.copied == 1
assert copied.copied_at is not None
payload = share_draft_dict(copied)
assert payload["aiGenerated"] is True
assert payload["content"].count("AI生成内容，请结合实际情况核对后使用。") == 1
other_user = User(id=2, phone="13800000002", name="其他学员", daily_chat_limit=100, daily_chat_used=0)
db.add(other_user)