217 lines
9.5 KiB
Python
217 lines
9.5 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from app.plugins.base import PluginRegistry
|
|
|
|
|
|
class MockPluginExecutor:
|
|
def register(self, registry: PluginRegistry) -> PluginRegistry:
|
|
registry.register("plugin.order.query", self._query_order)
|
|
registry.register("plugin.logistics.query", self._query_logistics)
|
|
registry.register("plugin.order.cancel", self._cancel_order)
|
|
registry.register("plugin.service.transfer_human", self._transfer_human)
|
|
registry.register("plugin.cabin.navigation.cancel", self._navigate_cancel)
|
|
registry.register("plugin.cabin.navigation", self._navigate)
|
|
registry.register("plugin.cabin.ac.on", self._ac_on)
|
|
registry.register("plugin.cabin.ac.off", self._ac_off)
|
|
registry.register("plugin.cabin.ac_control", self._set_ac)
|
|
registry.register("plugin.cabin.fan.up", self._fan_up)
|
|
registry.register("plugin.cabin.fan.down", self._fan_down)
|
|
registry.register("plugin.cabin.defog.front_on", self._defog_front_on)
|
|
registry.register("plugin.cabin.defog.rear_on", self._defog_rear_on)
|
|
registry.register("plugin.cabin.window.open", self._window_open)
|
|
registry.register("plugin.cabin.window.close", self._window_close)
|
|
registry.register("plugin.cabin.sunroof.open", self._sunroof_open)
|
|
registry.register("plugin.cabin.sunroof.close", self._sunroof_close)
|
|
registry.register("plugin.cabin.doors.lock", self._lock_doors)
|
|
registry.register("plugin.cabin.doors.unlock", self._unlock_doors)
|
|
registry.register("plugin.cabin.music_play", self._play_music)
|
|
registry.register("plugin.cabin.music.pause", self._pause_music)
|
|
registry.register("plugin.cabin.music.next", self._next_track)
|
|
registry.register("plugin.cabin.music.previous", self._previous_track)
|
|
registry.register("plugin.cabin.volume.up", self._volume_up)
|
|
registry.register("plugin.cabin.volume.down", self._volume_down)
|
|
registry.register("plugin.cabin.volume.mute", self._volume_mute)
|
|
registry.register("plugin.cabin.lights.on", self._lights_on)
|
|
registry.register("plugin.cabin.lights.off", self._lights_off)
|
|
registry.register("plugin.cabin.seat_heat.on", self._seat_heat_on)
|
|
registry.register("plugin.cabin.seat_heat.off", self._seat_heat_off)
|
|
registry.register("plugin.cabin.mirror.fold", self._mirror_fold)
|
|
registry.register("plugin.cabin.mirror.unfold", self._mirror_unfold)
|
|
registry.register("plugin.cabin.wiper.on", self._wiper_on)
|
|
registry.register("plugin.cabin.wiper.off", self._wiper_off)
|
|
return registry
|
|
|
|
def _query_order(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"success": True,
|
|
"message": f"订单 {slots['order_id']} 当前待发货。",
|
|
"data": {"order_status": "pending_shipment"},
|
|
}
|
|
|
|
def _query_logistics(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"success": True,
|
|
"message": f"订单 {slots['order_id']} 最新物流状态为运输中。",
|
|
"data": {"logistics_status": "shipping"},
|
|
}
|
|
|
|
def _cancel_order(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"success": True,
|
|
"message": f"订单 {slots['order_id']} 已取消。",
|
|
"data": {"cancel_status": "cancelled"},
|
|
}
|
|
|
|
def _transfer_human(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"success": True,
|
|
"message": "已为你转接人工客服，请稍候。",
|
|
"data": {"queue_no": "A12", "reason": slots.get("reason", "用户请求")},
|
|
}
|
|
|
|
def _navigate(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"success": True,
|
|
"message": f"已开始导航到 {slots['destination']}。",
|
|
"data": {"route_id": "route_001", "destination": slots["destination"]},
|
|
}
|
|
|
|
def _navigate_cancel(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已结束当前导航。", {"navigation": "stopped"})
|
|
|
|
def _ac_on(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已打开空调。", {"power": "on"})
|
|
|
|
def _ac_off(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已关闭空调。", {"power": "off"})
|
|
|
|
def _set_ac(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"success": True,
|
|
"message": f"已将空调调到 {slots['temperature']} 度。",
|
|
"data": {"temperature": slots["temperature"]},
|
|
}
|
|
|
|
def _fan_up(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已调大风量。", {"fan": "up"})
|
|
|
|
def _fan_down(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已调小风量。", {"fan": "down"})
|
|
|
|
def _defog_front_on(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已打开前挡除雾。", {"defog": "front_on"})
|
|
|
|
def _defog_rear_on(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已打开后挡除雾。", {"defog": "rear_on"})
|
|
|
|
def _window_open(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已打开车窗。", {"window": "open"})
|
|
|
|
def _window_close(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已关闭车窗。", {"window": "close"})
|
|
|
|
def _sunroof_open(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已打开天窗。", {"sunroof": "open"})
|
|
|
|
def _sunroof_close(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已关闭天窗。", {"sunroof": "close"})
|
|
|
|
def _lock_doors(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已锁定车门。", {"doors": "locked"})
|
|
|
|
def _unlock_doors(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已解锁车门。", {"doors": "unlocked"})
|
|
|
|
def _play_music(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
target = slots.get("song") or slots.get("genre") or "默认歌单"
|
|
return {
|
|
"success": True,
|
|
"message": f"正在播放 {target}。",
|
|
"data": {"play_target": target},
|
|
}
|
|
|
|
def _pause_music(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已暂停播放。", {"music": "paused"})
|
|
|
|
def _next_track(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已切到下一首。", {"music": "next"})
|
|
|
|
def _previous_track(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已切到上一首。", {"music": "previous"})
|
|
|
|
def _volume_up(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已调大音量。", {"volume": "up"})
|
|
|
|
def _volume_down(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已调小音量。", {"volume": "down"})
|
|
|
|
def _volume_mute(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已静音。", {"volume": "mute"})
|
|
|
|
def _lights_on(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已打开车灯。", {"lights": "on"})
|
|
|
|
def _lights_off(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已关闭车灯。", {"lights": "off"})
|
|
|
|
def _seat_heat_on(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已打开座椅加热。", {"seat_heat": "on"})
|
|
|
|
def _seat_heat_off(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已关闭座椅加热。", {"seat_heat": "off"})
|
|
|
|
def _mirror_fold(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已折叠后视镜。", {"mirror": "folded"})
|
|
|
|
def _mirror_unfold(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已展开后视镜。", {"mirror": "unfolded"})
|
|
|
|
def _wiper_on(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已打开雨刷。", {"wiper": "on"})
|
|
|
|
def _wiper_off(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
_ = slots
|
|
return self._simple_action("好的，已关闭雨刷。", {"wiper": "off"})
|
|
|
|
def _fallback(self, slots: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"success": True,
|
|
"message": "已接收请求，当前使用 mock 插件返回成功结果。",
|
|
"data": slots,
|
|
}
|
|
|
|
def _simple_action(self, message: str, data: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"success": True,
|
|
"message": message,
|
|
"data": data,
|
|
}
|