Agent/examples/tencent_realtime_control/operator_commands.py

"""Deterministic parsing for Feishu advertising control commands."""

from __future__ import annotations

import re
from dataclasses import dataclass


ACTION_DAY_PAUSE = "DAY_PAUSE"
ACTION_STOP = "STOP"
ACTION_RESUME = "RESUME"
ACTION_STATUS = "STATUS"
ACTION_TODAY_SPEND = "TODAY_SPEND"
ACTION_CONFIRM = "CONFIRM"
ACTION_CANCEL = "CANCEL"
ACTION_REJECT = "REJECT"

WRITE_ACTIONS = {ACTION_DAY_PAUSE, ACTION_STOP, ACTION_RESUME}
READ_ACTIONS = {ACTION_STATUS, ACTION_TODAY_SPEND}
INTENT_ACTIONS = WRITE_ACTIONS | READ_ACTIONS

SCOPE_ALL = "ALL"
SCOPE_ACCOUNTS = "ACCOUNTS"
SCOPE_AUTOMATION = "AUTOMATION"
SCOPE_MISSING = "MISSING"

_COMMAND_ID_RE = re.compile(r"\b(?:cmd|roi)_[0-9A-Za-z_-]+\b", re.IGNORECASE)
_ACCOUNT_ID_RE = re.compile(r"(?<!\d)(\d{7,12})(?!\d)")


@dataclass(frozen=True)
class ParsedCommand:
    action: str
    scope_type: str = "ACCOUNTS"
    account_ids: tuple[int, ...] = ()
    command_id: str | None = None


@dataclass(frozen=True)
class CommandIntent:
    action: str | None
    scope_type: str = SCOPE_MISSING
    account_ids: tuple[int, ...] = ()
    missing_fields: tuple[str, ...] = ()
    confidence: float = 1.0
    parse_source: str = "deterministic"

    @property
    def complete(self) -> bool:
        if self.action == ACTION_STATUS:
            return True
        if self.action == ACTION_TODAY_SPEND:
            if self.scope_type in {SCOPE_ALL, SCOPE_AUTOMATION}:
                return True
            return self.scope_type == SCOPE_ACCOUNTS and bool(self.account_ids)
        if self.action not in WRITE_ACTIONS:
            return False
        if self.scope_type == SCOPE_ALL:
            return True
        return self.scope_type == SCOPE_ACCOUNTS and bool(self.account_ids)

    def to_parsed_command(self) -> ParsedCommand:
        if not self.complete or self.action is None:
            raise ValueError("命令意图尚未补充完整")
        return ParsedCommand(
            action=self.action,
            scope_type=self.scope_type,
            account_ids=self.account_ids,
        )


def normalize_text(raw: str) -> str:
    text = str(raw or "").strip()
    translations = str.maketrans(
        {
            "，": ",",
            "、": ",",
            "；": ",",
            ";": ",",
            "：": " ",
            ":": " ",
            "\t": " ",
            "\r": " ",
            "\n": " ",
        }
    )
    return re.sub(r"\s+", " ", text.translate(translations)).strip()


def extract_account_ids(raw: str) -> tuple[int, ...]:
    return tuple(
        dict.fromkeys(int(value) for value in _ACCOUNT_ID_RE.findall(normalize_text(raw)))
    )


def parse_deterministic_intent(raw: str) -> CommandIntent | None:
    """Parse safe command forms, including incomplete commands for clarification."""
    text = normalize_text(raw)
    if not text:
        return None

    if "查看暂停状态" in text or "查询暂停状态" in text:
        return CommandIntent(action=ACTION_STATUS, scope_type=SCOPE_ALL)

    if (
        text.startswith(("今天", "暂停", "停止"))
        and any(value in text for value in ("今天", "到明天", "本日"))
    ):
        action = ACTION_DAY_PAUSE
    elif text.startswith(("停止", "持续停止", "永久停止")):
        action = ACTION_STOP
    elif text.startswith(("暂停", "今天暂停", "暂停到明天", "今天不投")):
        action = ACTION_DAY_PAUSE
    elif text.startswith(("恢复", "开启", "继续投放")):
        action = ACTION_RESUME
    else:
        return None

    if any(value in text for value in ("全部", "所有", "自动化账户", "纳管账户")):
        return CommandIntent(action=action, scope_type=SCOPE_ALL)
    account_ids = extract_account_ids(text)
    if account_ids:
        return CommandIntent(
            action=action,
            scope_type=SCOPE_ACCOUNTS,
            account_ids=account_ids,
        )
    return CommandIntent(
        action=action,
        missing_fields=("scope",),
    )


def parse_scope_reply(raw: str) -> tuple[str, tuple[int, ...]] | None:
    """Parse a scope-only answer used to complete an active conversation draft."""
    text = normalize_text(raw)
    if not text:
        return None
    if any(value in text for value in ("全部", "所有", "自动化账户", "纳管账户")):
        return SCOPE_ALL, ()
    account_ids = extract_account_ids(text)
    if account_ids:
        return SCOPE_ACCOUNTS, account_ids
    return None


def parse_command(raw: str) -> ParsedCommand | None:
    text = normalize_text(raw)
    if not text:
        return None

    lowered = text.lower()
    command_match = _COMMAND_ID_RE.search(text)
    if lowered.startswith("确认"):
        if not command_match:
            raise ValueError("确认命令缺少 command_id")
        return ParsedCommand(
            action=ACTION_CONFIRM,
            command_id=command_match.group(0).lower(),
        )
    if lowered.startswith("取消"):
        if not command_match:
            raise ValueError("取消命令缺少 command_id")
        return ParsedCommand(
            action=ACTION_CANCEL,
            command_id=command_match.group(0).lower(),
        )
    if lowered.startswith("拒绝"):
        if not command_match or not command_match.group(0).lower().startswith("roi_"):
            raise ValueError("拒绝命令缺少 roi_run_id")
        return ParsedCommand(
            action=ACTION_REJECT,
            command_id=command_match.group(0).lower(),
        )

    if "查看暂停状态" in text or "查询暂停状态" in text:
        return ParsedCommand(action=ACTION_STATUS, scope_type="ALL")

    if text.startswith(("停止", "持续停止", "永久停止")):
        action = ACTION_STOP
    elif text.startswith(("暂停", "今天暂停", "暂停到明天", "今天不投")):
        action = ACTION_DAY_PAUSE
    elif text.startswith(("恢复", "开启", "继续投放")):
        action = ACTION_RESUME
    else:
        return None

    if "全部" in text or "所有" in text:
        return ParsedCommand(action=action, scope_type="ALL")

    account_ids = extract_account_ids(text)
    if not account_ids:
        raise ValueError("命令中没有有效账户 ID，也没有指定“全部”")
    return ParsedCommand(
        action=action,
        scope_type="ACCOUNTS",
        account_ids=account_ids,
    )