131 lines
4.3 KiB
Python
131 lines
4.3 KiB
Python
import time
|
|
from typing import Dict, Any
|
|
from app.models.request import AuditRequest
|
|
from app.models.response import AuditResponse, DecisionPath
|
|
|
|
class AuditService:
|
|
"""AI审核服务 - 最小实现版本"""
|
|
|
|
def __init__(self):
|
|
pass
|
|
|
|
async def audit_order(self, request: AuditRequest) -> AuditResponse:
|
|
"""
|
|
执行AI审核 - 简化版本，模拟原有逻辑
|
|
"""
|
|
start_time = time.time()
|
|
|
|
# 模拟决策路径
|
|
decision_path = []
|
|
|
|
# 1. 前置条件检查
|
|
if not self._check_preconditions(request, decision_path):
|
|
return self._build_response("REFUSE", 0.9, "前置条件不满足", decision_path, start_time)
|
|
|
|
# 2. 业务规则检查
|
|
business_result = self._check_business_rules(request, decision_path)
|
|
if business_result != "PASS":
|
|
return self._build_response(business_result, 0.85, "业务规则验证失败", decision_path, start_time)
|
|
|
|
# 3. 默认通过
|
|
decision_path.append(DecisionPath(
|
|
node="FINAL_DECISION",
|
|
condition="all_checks_passed",
|
|
result=True
|
|
))
|
|
|
|
return self._build_response("PASS", 0.95, "AI审核通过", decision_path, start_time)
|
|
|
|
def _check_preconditions(self, request: AuditRequest, decision_path: list) -> bool:
|
|
"""前置条件检查"""
|
|
# 订单状态检查
|
|
if request.order_status != "WAITEXAMINE":
|
|
decision_path.append(DecisionPath(
|
|
node="PRE_CHECK",
|
|
condition="order_status == WAITEXAMINE",
|
|
result=False
|
|
))
|
|
return False
|
|
|
|
# OEM检查
|
|
if request.oem_id != 2:
|
|
decision_path.append(DecisionPath(
|
|
node="PRE_CHECK",
|
|
condition="oem_id == 2",
|
|
result=False
|
|
))
|
|
return False
|
|
|
|
# 车辆类型检查
|
|
if request.car_type != "0":
|
|
decision_path.append(DecisionPath(
|
|
node="PRE_CHECK",
|
|
condition="car_type == 0",
|
|
result=False
|
|
))
|
|
return False
|
|
|
|
decision_path.append(DecisionPath(
|
|
node="PRE_CHECK",
|
|
condition="all_preconditions_met",
|
|
result=True
|
|
))
|
|
|
|
return True
|
|
|
|
def _check_business_rules(self, request: AuditRequest, decision_path: list) -> str:
|
|
"""业务规则检查"""
|
|
|
|
# 公司名称检查
|
|
if request.buyer_name and self._is_company_name(request.buyer_name):
|
|
decision_path.append(DecisionPath(
|
|
node="COMPANY_CHECK",
|
|
condition="is_not_company",
|
|
result=False
|
|
))
|
|
return "REFUSE"
|
|
|
|
# 姓名一致性检查
|
|
if request.buyer_name and request.card_name and request.buyer_name != request.card_name:
|
|
decision_path.append(DecisionPath(
|
|
node="NAME_CONSISTENCY",
|
|
condition="buyer_name == card_name",
|
|
result=False
|
|
))
|
|
return "REFUSE"
|
|
|
|
# 计分检查 (简化版)
|
|
total_score = (request.id_card_score or 0) + (request.bill_score or 0)
|
|
if total_score < 50:
|
|
decision_path.append(DecisionPath(
|
|
node="SCORING_CHECK",
|
|
condition="total_score >= 50",
|
|
result=False
|
|
))
|
|
return "MANUAL"
|
|
|
|
decision_path.append(DecisionPath(
|
|
node="BUSINESS_RULES",
|
|
condition="all_business_rules_passed",
|
|
result=True
|
|
))
|
|
|
|
return "PASS"
|
|
|
|
def _is_company_name(self, name: str) -> bool:
|
|
"""检查是否为公司名称"""
|
|
company_keywords = ["有限公司", "股份公司", "集团", "企业", "责任公司"]
|
|
return any(keyword in name for keyword in company_keywords)
|
|
|
|
def _build_response(self, result: str, confidence: float, reason: str,
|
|
decision_path: list, start_time: float) -> AuditResponse:
|
|
"""构建响应对象"""
|
|
processing_time = int((time.time() - start_time) * 1000)
|
|
|
|
return AuditResponse(
|
|
result=result,
|
|
confidence=confidence,
|
|
reason=reason,
|
|
decision_path=decision_path,
|
|
processing_time_ms=processing_time
|
|
) |