79 lines
2.9 KiB
Python
79 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Callable, Dict, Awaitable
|
|
|
|
from db.customer_risk_db import risk_db
|
|
from utils.content_filter import should_block_customer_smart
|
|
|
|
|
|
@dataclass
|
|
class RiskDecision:
|
|
blocked: bool
|
|
category: str = "none"
|
|
reason: str = ""
|
|
source: str = "none" # manual/ai_filter/fallback
|
|
profile: Dict[str, Any] | None = None
|
|
|
|
|
|
class RiskService:
|
|
def evaluate_customer(self, customer_id: str) -> Dict[str, Any]:
|
|
return risk_db.evaluate_customer(customer_id)
|
|
|
|
def check_manual_block(self, customer_id: str) -> RiskDecision:
|
|
profile = self.evaluate_customer(customer_id)
|
|
if bool(profile.get("do_not_serve")):
|
|
return RiskDecision(
|
|
blocked=True,
|
|
category="manual_block",
|
|
reason="do_not_serve",
|
|
source="manual",
|
|
profile=profile,
|
|
)
|
|
return RiskDecision(blocked=False, source="manual", profile=profile)
|
|
|
|
async def check_text_block(
|
|
self,
|
|
text: str,
|
|
*,
|
|
political_detector: Callable[[str], bool],
|
|
map_detector: Callable[[str], bool],
|
|
) -> RiskDecision:
|
|
try:
|
|
hit, category, reason = await should_block_customer_smart(text)
|
|
map_hit = map_detector(text) or category == "map"
|
|
political_hit = political_detector(text) or category == "political"
|
|
if hit or map_hit or political_hit:
|
|
c = category
|
|
if map_hit:
|
|
c = "map"
|
|
elif political_hit:
|
|
c = "political"
|
|
return RiskDecision(
|
|
blocked=True,
|
|
category=c or "other",
|
|
reason=reason or "sensitive_text",
|
|
source="ai_filter",
|
|
)
|
|
return RiskDecision(blocked=False, category="none", source="ai_filter")
|
|
except Exception:
|
|
map_hit = map_detector(text)
|
|
political_hit = political_detector(text)
|
|
if map_hit:
|
|
return RiskDecision(blocked=True, category="map", reason="fallback_match", source="fallback")
|
|
if political_hit:
|
|
return RiskDecision(blocked=True, category="political", reason="fallback_match", source="fallback")
|
|
return RiskDecision(blocked=False, category="none", source="fallback")
|
|
|
|
@staticmethod
|
|
def build_reject_text(category: str) -> str:
|
|
if category == "map":
|
|
return "地图这类不做哈，这边不接地图相关需求。"
|
|
if category == "sexual":
|
|
return "这类不做哈，涉黄擦边内容都不接。"
|
|
if category == "violent":
|
|
return "这类不做哈，暴力血腥相关都不接。"
|
|
if category == "political":
|
|
return "这类不做哈，政治相关图片和人物都不接。"
|
|
return "这类不做哈，这边不接这类需求。"
|