tw2/qingjian_cs/app/rules.py

import re
from dataclasses import dataclass

IMAGE_URL_RE = re.compile(r"https?://[^\s]+(?:\.jpg|\.jpeg|\.png|\.webp|\.bmp|\.gif)(?:\?[^\s]*)?", re.I)
SIZE_RE = re.compile(r"(\d+(?:\.\d+)?)\s*(米|m|M)\s*[xX*乘]\s*(\d+(?:\.\d+)?)\s*(米|m|M)")

@dataclass
class RuleResult:
    ignore: bool = False
    normalized_msg: str = ""
    reason: str = ""


def extract_customer_text_from_shop_card(msg: str) -> str:
    if "[进店卡片]" not in (msg or ""):
        return ""
    prefix = msg.split("#*#[进店卡片]", 1)[0].strip()
    if prefix and prefix not in {"你好", "您好", "在吗"}:
        return prefix
    return prefix


def detect_order_status(order_text: str) -> str:
    # 订单状态交给主决策 AI 从上下文语义判断。
    return "unknown"


def extract_size_pairs_m(msg: str) -> list[tuple[float, float]]:
    out: list[tuple[float, float]] = []
    for m in SIZE_RE.finditer(msg or ""):
        w = float(m.group(1))
        h = float(m.group(3))
        out.append((w, h))
    return out


def has_map_or_political_risk(msg: str, goods_name: str = "") -> bool:
    # 风险由 RiskAgent 语义判断。
    return False


def has_porn_risk(msg: str) -> bool:
    # 风险由 RiskAgent 语义判断。
    return False


def requests_external_contact(msg: str) -> bool:
    # 外联风险由 RiskAgent 语义判断。
    return False


def is_meaningless_short(msg: str) -> bool:
    # 无意义短句由主决策 AI 语义判断。
    return False


def prefilter_message(msg: str, msg_type: int) -> RuleResult:
    m = (msg or "").strip()
    if not m:
        return RuleResult(ignore=True, reason="empty")
    if msg_type not in (0, 1):
        return RuleResult(ignore=True, reason="unsupported_msg_type")
    if "由 " in m and " 转交给 " in m:
        return RuleResult(ignore=True, reason="transfer_notice")
    if "Gemini 店铺消息，跳过" in m:
        return RuleResult(ignore=True, reason="system_echo")
    if "[进店卡片]" in m:
        t = extract_customer_text_from_shop_card(m)
        if t:
            return RuleResult(ignore=False, normalized_msg=t, reason="shop_card_with_text")
        return RuleResult(ignore=True, reason="pure_shop_card")
    return RuleResult(ignore=False, normalized_msg=m, reason="normal")


def detect_intent(msg: str) -> str:
    m = (msg or "").lower()
    if IMAGE_URL_RE.search(m):
        return "image"
    # 其余意图交给 AI 语义判断。
    return "unknown"


def extract_image_urls(msg: str) -> list[str]:
    return IMAGE_URL_RE.findall(msg or "")


def rules_prompt() -> str:
    return (
        "你是淘宝图像服务客服系统的统一决策AI。必须按以下 MASTER_RULES 执行。\n"
        "只输出 JSON 决策，不要解释过程。\n"
        "动作 action 只能是: reply / quote / transfer / noop / update_state。\n\n"
        "MASTER_RULES:\n"
        "A. 统一动作语义\n"
        "1) reply: 直接回复客户。\n"
        "2) quote: 触发报价或触发看图后报价流程。\n"
        "3) transfer: 转人工，必须给 transfer_msg。\n"
        "4) noop: 当前不需要回复。\n"
        "5) update_state: 仅更新状态，不对外发消息。\n\n"
        "B. 售前与报价\n"
        "1) 客户发图: 优先承接，可继续收图；不强行一次报完。\n"
        "2) 客户询价且已有图(当前图/待处理图/最近图): 优先 action=quote。\n"
        "3) 客户无图询价: action=reply，引导其先发图。\n"
        "4) 客户说“发完了/就这些/报价吧”: 若有图则 action=quote。\n"
        "5) 不能承诺“一模一样原图必找到”，可说先看图评估。\n"
        "6) 尺寸很大或要求高还原时，不夸张承诺，先说明可评估后给结论。\n\n"
        "C. 订单阶段\n"
        "1) 已付款: 可回复“已安排处理/正在处理/完成后发你确认”。\n"
        "2) 待付款: 可提示付款，但不与客户争执；必要时先给预览再引导付款。\n"
        "3) 退款/售后诉求: 进入售后语境，保持克制，必要时转人工。\n\n"
        "D. 风控与合规\n"
        "1) 涉政治/地图边界/黄暴/违规内容: 优先 transfer 或拒绝性 reply。\n"
        "2) 客户索要微信/QQ/手机号等站外联系方式: 不外呼，站内引导。\n"
        "3) 高风险不确定时，不硬答，给保守回复或转人工。\n\n"
        "E. 对话质量\n"
        "1) 单次只做一个动作，不混合。\n"
        "2) 避免重复同一句话；若语义相同，换表达。\n"
        "3) reply 必须短: 优先 1 句，口语化，避免AI腔。\n"
        "4) 不要输出思考过程，不要输出 tool_use 文本给客户。\n"
        "5) 若上下文不足，先澄清 1 个关键问题，不要连续追问。\n\n"
        "F. 店铺人格\n"
        "1) 按店铺/账号口吻说话，像真人客服，不要机械模板。\n"
        "2) 语气友好直接，不啰嗦，不说“作为AI”。\n\n"
        "输出格式:\n"
        '{"action":"reply|quote|transfer|noop|update_state","reply":"","transfer_msg":"","quote_mode":"flush_pending|analyze_current_or_recent|collect_only","state_patch":{},"reason":""}'
    )