from datetime import UTC, datetime import pytest from app.models.entities import Conversation, ConversationMessage from app.schemas.risk_prediction import SupplierFrequency from app.services.conversations import ConversationService def test_conversation_message_data_reads_persisted_risk_result() -> None: message = ConversationMessage( id=10, conversation_id=3, role="assistant", content="已完成风险预估。", structured_data={ "riskResult": { "subjectId": "软管新品", "score": 82, "level": "low", "modelVersion": "baseline-safety-v2", "factors": [], } }, created_at=datetime.now(UTC), ) result = ConversationService._message(message) assert result.role == "assistant" assert result.risk_result is not None assert result.risk_result.score == 82 assert result.risk_result.level.value == "low" def test_conversation_message_data_reads_persisted_semantic_and_response_card() -> None: message = ConversationMessage( id=11, conversation_id=3, role="assistant", content="已完成分析。", structured_data={ "semantic": {"intent": "solution", "category": "洗发水"}, "responseCard": { "type": "solution", "title": "AI 案例推理生成", "similarCases": [{"case_id": 101, "similarity": 0.82}], "suggestions": ["完成包材相容性测试"], }, }, created_at=datetime.now(UTC), ) result = ConversationService._message(message) assert result.semantic == {"intent": "solution", "category": "洗发水"} assert result.response_card is not None assert result.response_card["similarCases"][0]["case_id"] == 101 assert result.response_card["suggestions"] == ["完成包材相容性测试"] def test_conversation_data_includes_all_messages() -> None: conversation = Conversation( id=3, user_id=1, title="软管新品评估", created_at=datetime.now(UTC), updated_at=datetime.now(UTC), messages=[ ConversationMessage( id=1, conversation_id=3, role="user", content="评估软管新品", created_at=datetime.now(UTC), ), ConversationMessage( id=2, conversation_id=3, role="assistant", content="已完成分析", created_at=datetime.now(UTC), ), ], ) result = ConversationService._data(conversation) assert result.message_count == 2 assert [message.role for message in result.messages] == ["user", "assistant"] def test_only_explicit_risk_requests_generate_risk_cards() -> None: assert ConversationService._is_risk_request( "你好", {"intent": "greet", "risk_requested": False}, ) is False assert ConversationService._is_risk_request( "请只回复：AI 连通成功。", {"intent": "general", "risk_requested": False}, ) is False assert ConversationService._is_risk_request( "请评估这个产品的风险", {"intent": "assessment", "risk_requested": True}, ) is True @pytest.mark.parametrize( ("text", "expected"), [ ("评估新品：洗发水，包装材质：泵头", "risk_assessment"), ("查物料风险：软管", "material"), ("搜索相似案例：软管印刷遇水脱落", "risk_assessment"), ("生成 AI 案例推理方案：铝瓶+喷头密封验证", "solution"), ], ) def test_tool_requests_keep_their_intent_for_card_routing(text: str, expected: str) -> None: assert ConversationService._response_type(text, {"intent": "general"}) == expected def test_legacy_general_semantics_hide_old_risk_card() -> None: messages = [ ConversationMessage( id=1, conversation_id=1, role="user", content="你好", created_at=datetime.now(UTC), ), ConversationMessage( id=2, conversation_id=1, role="assistant", content="你好，我可以帮你做什么？", structured_data={ "semantic": {"intent": "general"}, "riskResult": { "subjectId": "误判", "score": 61, "level": "moderately_low", "modelVersion": "baseline-safety-v2", "factors": [], }, }, created_at=datetime.now(UTC), ), ] assert ConversationService._risk_result_for_message(messages, 1) is None assert ConversationService._is_risk_request( "我要做一款脱毛膏新品，100g，PE软管包装，加工厂准备使用瑞沣", {"intent": "general", "risk_requested": False}, ) is True def test_product_assessment_semantics_override_nlu_miss() -> None: normalized = ConversationService._normalize_semantics( "我要做一款脱毛膏新品，100g，PE软管包装，加工厂准备使用瑞沣", {"intent": "general", "risk_requested": False, "category": "脱毛膏"}, ) assert normalized["intent"] == "assessment" assert normalized["risk_requested"] is True def test_short_known_supplier_follow_up_is_routed_to_supplier_card() -> None: assert ConversationService._infer_supplier_follow_up( "佳亿", ["佳亿", "祥印", "瑞沣"], ) == "佳亿" assert ConversationService._infer_supplier_follow_up( "这个供应商", ["佳亿", "祥印"], ) is None def test_unknown_standalone_name_is_treated_as_new_supplier() -> None: assert ConversationService._infer_unknown_supplier_follow_up("德盛") == "德盛" assert ConversationService._infer_unknown_supplier_follow_up("新供应商A") is None assert ConversationService._infer_unknown_supplier_follow_up("测试供应商") is None assert ConversationService._infer_unknown_supplier_follow_up("新供应商B") is None assert ConversationService._infer_unknown_supplier_follow_up("洗发水") is None assert ConversationService._infer_unknown_supplier_follow_up("帮我查德盛") is None assert ConversationService._response_type("测试供应商", {"intent": "supplier"}) == "general" assert ConversationService._response_type("新供应商A", {"intent": "supplier"}) == "general" @pytest.mark.parametrize( ("count", "a_count", "b_count", "expected"), [ (0, 0, 0, "新供应商"), (12, 0, 1, "S级 · 战略"), (40, 0, 4, "A级 · 优质"), (70, 0, 10, "B级 · 关注"), (40, 0, 16, "B级 · 关注"), (10, 1, 0, "C级 · 高风险"), (121, 0, 0, "C级 · 高风险"), ], ) def test_supplier_tier_uses_prd_frequency_and_defect_rules( count: int, a_count: int, b_count: int, expected: str, ) -> None: assert ConversationService._supplier_tier( count, a_count=a_count, b_count=b_count, ) == expected def test_supplier_management_suggestions_follow_prd_actions() -> None: assert ConversationService._supplier_management_suggestions("B级 · 关注") == [ "加强巡检", "新品打样加严", ] assert ConversationService._supplier_management_suggestions("新供应商") == [ "无历史数据", "首单全检", "安排驻厂巡检", ] @pytest.mark.asyncio async def test_supplier_card_uses_prd_proxy_frequency_and_tier_advice( monkeypatch: pytest.MonkeyPatch, ) -> None: service = object.__new__(ConversationService) async def history_stats(*_args, **_kwargs): return 70, 0, 10 async def top_anomaly_types(*_args, **_kwargs): return [{"name": "泵头异常", "count": 12}] monkeypatch.setattr(service, "_supplier_history_stats", history_stats) monkeypatch.setattr(service, "_top_anomaly_types", top_anomaly_types) result = await service._supplier_card( "祥印供应商风险怎么样", {"supplier": "祥印"}, ) assert result["anomalyCount"] == 70 assert result["cooperationCount"] is None assert result["frequencyCoefficient"] == 1.0 assert result["tier"] == "B级 · 关注" assert result["suggestions"] == ["加强巡检", "新品打样加严"] assert result["frequencyBasis"] == "暂无匹配采购订单" assert result["dataAvailable"] is False def test_new_supplier_frequency_serializes_zero_proxy_count() -> None: result = SupplierFrequency( supplier_name="新供应商", cooperation_count=0, anomaly_count=0, tier="新供应商", suggestion="无历史数据；首单全检；安排驻厂巡检", note="合作次数为近3年异常记录数代理值，不代表真实采购订单合作次数。", ).model_dump(by_alias=True) assert result["cooperationCount"] == 0 assert result["anomalyCount"] == 0 assert result["frequencyBasis"] == "暂无匹配采购订单" def test_unspecified_supplier_returns_displayable_no_data_frequency_card() -> None: result = SupplierFrequency( note="未指定供应商，暂无供应商频次分析。", frequency_basis="未提供供应商", ).model_dump(by_alias=True) assert result["dataAvailable"] is False assert result["alpha"] is None assert result["note"] == "未指定供应商，暂无供应商频次分析。" def test_similar_cases_are_typed_before_risk_result_serialization() -> None: from app.schemas.risk_prediction import RiskAssessmentResult, SimilarCase result = RiskAssessmentResult( subject_id="test", score=52, level="medium", model_version="test", factors=[], similar_cases=[SimilarCase(case_id=1, similarity=0.75)], ) payload = result.model_dump(mode="json") assert payload["similar_cases"][0]["case_id"] == 1 @pytest.mark.asyncio async def test_any_product_without_history_uses_solution_card(monkeypatch: pytest.MonkeyPatch) -> None: service = object.__new__(ConversationService) async def no_history(_: str) -> int: return 0 monkeypatch.setattr(service, "_count_category_anomalies", no_history) response_type = ConversationService._response_type( "评估防晒霜新品，50g，软管包装", {"intent": "assessment", "risk_requested": True, "category": "防晒霜"}, ) result = await service._use_solution_card_without_history( response_type, "评估防晒霜新品，50g，软管包装", {"intent": "assessment", "risk_requested": True, "category": "防晒霜"}, ) assert response_type == "risk_assessment" assert result == "solution" @pytest.mark.asyncio async def test_unknown_product_without_category_uses_solution_card() -> None: service = object.__new__(ConversationService) result = await service._use_solution_card_without_history( "risk_assessment", "评估这个新产品的风险", {"intent": "assessment", "risk_requested": True}, ) assert result == "solution"