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_CONFIRM = "CONFIRM"
ACTION_CANCEL = "CANCEL"

WRITE_ACTIONS = {ACTION_DAY_PAUSE, ACTION_STOP, ACTION_RESUME}

_COMMAND_ID_RE = re.compile(r"\bcmd_[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


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 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 "查看暂停状态" 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 = tuple(
        dict.fromkeys(int(value) for value in _ACCOUNT_ID_RE.findall(text))
    )
    if not account_ids:
        raise ValueError("命令中没有有效账户 ID，也没有指定“全部”")
    return ParsedCommand(
        action=action,
        scope_type="ACCOUNTS",
        account_ids=account_ids,
    )