297 lines
8.3 KiB
Python
297 lines
8.3 KiB
Python
import json
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
try:
|
|
import pyautogui
|
|
pyautogui.FAILSAFE = True
|
|
pyautogui.PAUSE = 0.06
|
|
_PYAUTOGUI = True
|
|
except ImportError:
|
|
_PYAUTOGUI = False
|
|
|
|
try:
|
|
import pyperclip
|
|
_PYPERCLIP = True
|
|
except ImportError:
|
|
_PYPERCLIP = False
|
|
|
|
def _base_dir() -> Path:
|
|
if getattr(sys, "frozen", False):
|
|
return Path(sys.executable).parent
|
|
return Path(__file__).resolve().parent.parent
|
|
|
|
def _get_os() -> str:
|
|
try:
|
|
cfg = json.loads(
|
|
(_base_dir() / "config" / "api_keys.json").read_text(encoding="utf-8")
|
|
)
|
|
return cfg.get("os_system", "windows").lower()
|
|
except Exception:
|
|
return "windows"
|
|
|
|
|
|
def _require_pyautogui():
|
|
if not _PYAUTOGUI:
|
|
raise RuntimeError("PyAutoGUI not installed. Run: pip install pyautogui")
|
|
|
|
|
|
def _paste_text(text: str) -> None:
|
|
_require_pyautogui()
|
|
|
|
os_name = _get_os()
|
|
paste_hotkey = ("command", "v") if os_name == "mac" else ("ctrl", "v")
|
|
|
|
if _PYPERCLIP:
|
|
pyperclip.copy(text)
|
|
time.sleep(0.15)
|
|
pyautogui.hotkey(*paste_hotkey)
|
|
time.sleep(0.1)
|
|
else:
|
|
pyautogui.write(text, interval=0.03)
|
|
|
|
|
|
def _clear_and_paste(text: str) -> None:
|
|
_require_pyautogui()
|
|
os_name = _get_os()
|
|
select_all = ("command", "a") if os_name == "mac" else ("ctrl", "a")
|
|
pyautogui.hotkey(*select_all)
|
|
time.sleep(0.1)
|
|
pyautogui.press("delete")
|
|
time.sleep(0.1)
|
|
_paste_text(text)
|
|
|
|
def _open_app(app_name: str) -> bool:
|
|
_require_pyautogui()
|
|
os_name = _get_os()
|
|
|
|
try:
|
|
if os_name == "windows":
|
|
pyautogui.press("win")
|
|
time.sleep(0.5)
|
|
_paste_text(app_name)
|
|
time.sleep(0.6)
|
|
pyautogui.press("enter")
|
|
time.sleep(2.5)
|
|
return True
|
|
|
|
elif os_name == "mac":
|
|
result = subprocess.run(
|
|
["open", "-a", app_name],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
if result.returncode != 0:
|
|
result = subprocess.run(
|
|
["open", "-a", f"{app_name}.app"],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
time.sleep(2.5)
|
|
return result.returncode == 0
|
|
|
|
else:
|
|
launched = False
|
|
for launcher in [
|
|
["gtk-launch", app_name.lower()],
|
|
[app_name.lower()],
|
|
]:
|
|
try:
|
|
subprocess.Popen(
|
|
launcher,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
launched = True
|
|
break
|
|
except FileNotFoundError:
|
|
continue
|
|
time.sleep(2.5)
|
|
return launched
|
|
|
|
except Exception as e:
|
|
print(f"[SendMessage] ⚠️ Could not open {app_name}: {e}")
|
|
return False
|
|
|
|
|
|
def _open_browser_url(url: str) -> bool:
|
|
import webbrowser
|
|
try:
|
|
webbrowser.open(url)
|
|
time.sleep(4.0)
|
|
return True
|
|
except Exception as e:
|
|
print(f"[SendMessage] ⚠️ Could not open browser: {e}")
|
|
return False
|
|
|
|
def _search_in_app(query: str) -> None:
|
|
_require_pyautogui()
|
|
os_name = _get_os()
|
|
search_hotkey = ("command", "f") if os_name == "mac" else ("ctrl", "f")
|
|
|
|
pyautogui.hotkey(*search_hotkey)
|
|
time.sleep(0.5)
|
|
_clear_and_paste(query)
|
|
time.sleep(1.0)
|
|
|
|
def _desktop_send(app_name: str, receiver: str, message: str) -> str:
|
|
if not _open_app(app_name):
|
|
return f"Could not open {app_name}."
|
|
|
|
time.sleep(1.0)
|
|
_search_in_app(receiver)
|
|
pyautogui.press("enter")
|
|
time.sleep(0.8)
|
|
|
|
_paste_text(message)
|
|
time.sleep(0.2)
|
|
pyautogui.press("enter")
|
|
time.sleep(0.3)
|
|
return f"Message sent to {receiver} via {app_name}."
|
|
|
|
def _send_whatsapp(receiver: str, message: str) -> str:
|
|
return _desktop_send("WhatsApp", receiver, message)
|
|
|
|
def _send_telegram(receiver: str, message: str) -> str:
|
|
return _desktop_send("Telegram", receiver, message)
|
|
|
|
def _send_signal(receiver: str, message: str) -> str:
|
|
return _desktop_send("Signal", receiver, message)
|
|
|
|
|
|
def _send_discord(receiver: str, message: str) -> str:
|
|
return _desktop_send("Discord", receiver, message)
|
|
|
|
|
|
def _send_instagram(receiver: str, message: str) -> str:
|
|
_require_pyautogui()
|
|
|
|
if not _open_browser_url("https://www.instagram.com/direct/new/"):
|
|
return "Could not open Instagram in browser."
|
|
|
|
_paste_text(receiver)
|
|
time.sleep(1.5)
|
|
|
|
pyautogui.press("down")
|
|
time.sleep(0.3)
|
|
pyautogui.press("enter")
|
|
time.sleep(0.4)
|
|
|
|
for _ in range(4):
|
|
pyautogui.press("tab")
|
|
time.sleep(0.15)
|
|
pyautogui.press("enter")
|
|
time.sleep(2.0)
|
|
|
|
_paste_text(message)
|
|
time.sleep(0.2)
|
|
pyautogui.press("enter")
|
|
time.sleep(0.3)
|
|
|
|
return f"Message sent to {receiver} via Instagram."
|
|
|
|
|
|
def _send_messenger(receiver: str, message: str) -> str:
|
|
_require_pyautogui()
|
|
|
|
if not _open_browser_url("https://www.messenger.com/"):
|
|
return "Could not open Messenger in browser."
|
|
|
|
|
|
_search_in_app(receiver)
|
|
time.sleep(0.5)
|
|
pyautogui.press("down")
|
|
time.sleep(0.3)
|
|
pyautogui.press("enter")
|
|
time.sleep(1.0)
|
|
|
|
_paste_text(message)
|
|
time.sleep(0.2)
|
|
pyautogui.press("enter")
|
|
time.sleep(0.3)
|
|
|
|
return f"Message sent to {receiver} via Messenger."
|
|
|
|
_PLATFORM_MAP = [
|
|
({"whatsapp", "wp", "wapp"}, _send_whatsapp),
|
|
({"telegram", "tg"}, _send_telegram),
|
|
({"instagram", "ig", "insta"}, _send_instagram),
|
|
({"signal"}, _send_signal),
|
|
({"discord"}, _send_discord),
|
|
({"messenger", "facebook", "fb"}, _send_messenger),
|
|
]
|
|
|
|
|
|
def _resolve_platform(platform_str: str):
|
|
key = platform_str.lower().strip()
|
|
for keywords, handler in _PLATFORM_MAP:
|
|
if any(k in key for k in keywords):
|
|
return handler
|
|
return lambda r, m: _desktop_send(platform_str.strip().title(), r, m)
|
|
|
|
|
|
def send_message(
|
|
parameters: dict,
|
|
response=None,
|
|
player=None,
|
|
session_memory=None,
|
|
) -> str:
|
|
params = parameters or {}
|
|
receiver = params.get("receiver", "").strip()
|
|
message_text = params.get("message_text", "").strip()
|
|
platform = params.get("platform", "whatsapp").strip()
|
|
|
|
if not receiver:
|
|
return "Please specify a recipient."
|
|
if not message_text:
|
|
return "Please specify the message content."
|
|
if not _PYAUTOGUI:
|
|
return "PyAutoGUI is not installed — cannot control the desktop."
|
|
|
|
preview = message_text[:50] + ("…" if len(message_text) > 50 else "")
|
|
print(f"[SendMessage] 📨 {platform} → {receiver}: {preview}")
|
|
if player:
|
|
player.write_log(f"[msg] {platform} → {receiver}")
|
|
|
|
try:
|
|
handler = _resolve_platform(platform)
|
|
result = handler(receiver, message_text)
|
|
except Exception as e:
|
|
result = f"Could not send message: {e}"
|
|
|
|
print(f"[SendMessage] {'✅' if 'sent' in result.lower() else '❌'} {result}")
|
|
if player:
|
|
player.write_log(f"[msg] {result}")
|
|
|
|
return result
|
|
|
|
|
|
# ── Tool declaration (auto-discovered by core/action_loader.py) ──────────────
|
|
TOOL = {
|
|
"name": "send_message",
|
|
"description": "Sends a text message via WhatsApp, Telegram, or other messaging platform.",
|
|
"parameters": {
|
|
"type": "OBJECT",
|
|
"properties": {
|
|
"receiver": {
|
|
"type": "STRING",
|
|
"description": "Recipient contact name"
|
|
},
|
|
"message_text": {
|
|
"type": "STRING",
|
|
"description": "The message to send"
|
|
},
|
|
"platform": {
|
|
"type": "STRING",
|
|
"description": "Platform: WhatsApp, Telegram, etc."
|
|
}
|
|
},
|
|
"required": [
|
|
"receiver",
|
|
"message_text",
|
|
"platform"
|
|
]
|
|
},
|
|
"handler": send_message,
|
|
}
|