🚀 Auto-deploy: BotVPS atualizado em 29/03/2026 18:02:27
This commit is contained in:
107
tools.py
107
tools.py
@@ -147,44 +147,44 @@ def cronos_query(arg: str) -> str:
|
||||
target_dir = os.path.join(MEMORY_ROOT, folder)
|
||||
return run_bash_command(f"grep -rniI '{query}' {target_dir} | head -n 20")
|
||||
|
||||
def list_gmail_emails(account: str) -> str:
|
||||
"""Lista os últimos 5 e-mails com Título e Remetente. Aceita apelidos: ma, mr, adm, 4r."""
|
||||
mapping = {
|
||||
# ============================================================
|
||||
# GOOGLE WORKSPACE TOOLS
|
||||
# ============================================================
|
||||
|
||||
ACCOUNT_MAPPING = {
|
||||
"ma": "gws-mr", "mr": "gws-mr", "marcos": "gws-mr",
|
||||
"adm": "gws-adm", "empresa": "gws-adm",
|
||||
"4r": "gws-4r", "familia": "gws-4r", "fam": "gws-4r"
|
||||
}
|
||||
|
||||
clean_account = account.strip().lower().replace("gws-", "")
|
||||
account = mapping.get(clean_account, f"gws-{clean_account}" if not clean_account.startswith("gws") else clean_account)
|
||||
def resolve_account(account_alias: str) -> str:
|
||||
clean = account_alias.strip().lower()
|
||||
return ACCOUNT_MAPPING.get(clean, f"gws-{clean}")
|
||||
|
||||
# 1. Obtém a lista de IDs
|
||||
def list_gmail_emails(account: str) -> str:
|
||||
"""Lista os últimos 5 e-mails com Título e Remetente. Aceita apelidos: ma, mr, adm, 4r."""
|
||||
account = resolve_account(account)
|
||||
list_cmd = f"{account} gmail users messages list --params '{{\"userId\": \"me\", \"maxResults\": 5}}'"
|
||||
res = run_bash_command(list_cmd)
|
||||
|
||||
try:
|
||||
data = json.loads(res)
|
||||
messages = data.get("messages", [])
|
||||
if not messages:
|
||||
return "Nenhum e-mail encontrado."
|
||||
if not messages: return "Nenhum e-mail encontrado."
|
||||
|
||||
result_text = "📧 **Últimos E-mails:**\n"
|
||||
for i, msg in enumerate(messages, 1):
|
||||
msg_id = msg["id"]
|
||||
# 2. Busca detalhes de cada e-mail (Metadata apenas para ser rápido)
|
||||
details_cmd = f"{account} gmail users messages get --params '{{\"userId\": \"me\", \"id\": \"{msg_id}\", \"format\": \"metadata\", \"metadataHeaders\": [\"Subject\", \"From\"]}}'"
|
||||
details_res = run_bash_command(details_cmd)
|
||||
|
||||
try:
|
||||
details = json.loads(details_res)
|
||||
headers = details.get("payload", {}).get("headers", [])
|
||||
subject = next((h["value"] for h in headers if h["name"] == "Subject"), "Sem Assunto")
|
||||
sender = next((h["value"] for h in headers if h["name"] == "From"), "Desconhecido")
|
||||
|
||||
result_text += f"{i}. **De:** {sender}\n **Assunto:** {subject}\n **ID:** `{msg_id}`\n\n"
|
||||
except:
|
||||
result_text += f"{i}. [Erro ao carregar detalhes do ID: {msg_id}]\n\n"
|
||||
|
||||
return result_text
|
||||
except Exception as e:
|
||||
return f"Erro ao listar e-mails: {str(e)}\nResposta bruta: {res[:200]}"
|
||||
@@ -193,48 +193,66 @@ def gmail_manage_label(arg: str) -> str:
|
||||
"""Cria ou busca marcadores (labels). Arg: account name (ex: adm alibaba)"""
|
||||
try:
|
||||
parts = arg.split(maxsplit=1)
|
||||
account = parts[0]
|
||||
label_name = parts[1] if len(parts) > 1 else ""
|
||||
account_alias = parts[0]
|
||||
label_name = parts[1].strip() if len(parts) > 1 else ""
|
||||
if not label_name: return "Erro: Nome do marcador não fornecido."
|
||||
account = resolve_account(account_alias)
|
||||
|
||||
mapping = {"ma": "gws-mr", "mr": "gws-mr", "adm": "gws-adm", "4r": "gws-4r"}
|
||||
account = mapping.get(account.lower(), account)
|
||||
# 1. Verifica se já existe
|
||||
list_res = run_bash_command(f"{account} gmail users labels list --params '{{\"userId\": \"me\"}}'")
|
||||
try:
|
||||
# Extrai apenas o JSON se houver lixo no stdout (ex: Using keyring...)
|
||||
json_match = re.search(r"(\{.*\})", list_res, re.S)
|
||||
list_res_clean = json_match.group(1) if json_match else list_res
|
||||
labels_data = json.loads(list_res_clean)
|
||||
for l in labels_data.get("labels", []):
|
||||
if l["name"].lower() == label_name.lower():
|
||||
return f"Marcador '{l['name']}' já existe (ID: {l['id']})."
|
||||
except: pass
|
||||
|
||||
# Tenta criar o marcador
|
||||
cmd = f"{account} gmail users labels create --json '{{\"name\": \"{label_name}\", \"labelListVisibility\": \"labelShow\", \"messageListVisibility\": \"show\"}}'"
|
||||
return run_bash_command(cmd)
|
||||
except Exception as e: return f"Erro ao gerenciar marcador: {str(e)}"
|
||||
# 2. Tenta criar
|
||||
cmd = f"{account} gmail users labels create --params '{{\"userId\": \"me\"}}' --json '{{\"name\": \"{label_name}\", \"labelListVisibility\": \"labelShow\", \"messageListVisibility\": \"show\"}}'"
|
||||
res = run_bash_command(cmd)
|
||||
return f"Criação de '{label_name}': {res}"
|
||||
except Exception as e: return f"Erro marcador: {str(e)}"
|
||||
|
||||
def gmail_manage_filter(arg: str) -> str:
|
||||
"""Cria um filtro para e-mails. Arg: account subject_or_from label_name (ex: adm alibaba alibaba)"""
|
||||
"""Cria um filtro para e-mails. Arg: account criteria label_name (ex: adm alibaba alibaba)"""
|
||||
try:
|
||||
parts = arg.split(maxsplit=2)
|
||||
if len(parts) < 3: return "Erro: Use 'conta termo marcador'. Ex: adm alibaba alibaba"
|
||||
account, criteria, label_id = parts[0], parts[1], parts[2]
|
||||
if len(parts) < 3: return "Erro: Use 'conta criteria marcador'. Ex: adm alibaba alibaba"
|
||||
account_alias, criteria, label_name = parts[0], parts[1], parts[2]
|
||||
account = resolve_account(account_alias)
|
||||
|
||||
mapping = {"ma": "gws-mr", "mr": "gws-mr", "adm": "gws-adm", "4r": "gws-4r"}
|
||||
account = mapping.get(account.lower(), account)
|
||||
# Busca o ID do marcador pelo nome
|
||||
label_id = label_name
|
||||
list_res = run_bash_command(f"{account} gmail users labels list --params '{{\"userId\": \"me\"}}'")
|
||||
try:
|
||||
json_match = re.search(r"(\{.*\})", list_res, re.S)
|
||||
list_res_clean = json_match.group(1) if json_match else list_res
|
||||
labels_data = json.loads(list_res_clean)
|
||||
for l in labels_data.get("labels", []):
|
||||
if l["name"].lower() == label_name.lower():
|
||||
label_id = l["id"]
|
||||
break
|
||||
except: pass
|
||||
|
||||
# Tenta criar o filtro (exemplo: assunto contém o critério ou de quem vem)
|
||||
# Primeiro, buscamos se o marcador existe para pegar o ID?
|
||||
# Na verdade no Gmail API você pode usar o nome do marcador se for por --json direto.
|
||||
# Mas gws gmail users settings filters create requer um Filter object.
|
||||
criteria_obj = {"from": criteria} if "@" in criteria else {"query": criteria}
|
||||
filter_obj = {
|
||||
"criteria": {"from": criteria}, # Simplificado: busca pelo remetente
|
||||
"criteria": criteria_obj,
|
||||
"action": {"addLabelIds": [label_id]}
|
||||
}
|
||||
cmd = f"{account} gmail users settings filters create --json '{json.dumps(filter_obj)}'"
|
||||
cmd = f"{account} gmail users settings filters create --params '{{\"userId\": \"me\"}}' --json '{json.dumps(filter_obj)}'"
|
||||
return run_bash_command(cmd)
|
||||
except Exception as e: return f"Erro ao gerenciar filtro: {str(e)}"
|
||||
except Exception as e: return f"Erro filtro: {str(e)}"
|
||||
|
||||
def drive_find(arg: str) -> str:
|
||||
"""Busca arquivos no Drive por nome. Arg: account query (ex: ma financeiro)"""
|
||||
try:
|
||||
parts = arg.split(maxsplit=1)
|
||||
account = parts[0]
|
||||
account_alias = parts[0]
|
||||
query = parts[1] if len(parts) > 1 else ""
|
||||
mapping = {"ma": "gws-mr", "mr": "gws-mr", "marcos": "gws-mr", "adm": "gws-adm", "4r": "gws-4r", "fam": "gws-4r"}
|
||||
account = mapping.get(account.lower(), account)
|
||||
account = resolve_account(account_alias)
|
||||
q = f"name contains '{query}'" if query else ""
|
||||
cmd = f"{account} drive files list"
|
||||
if q: cmd += f" --params '{{\"q\": \"{q}\"}}'"
|
||||
@@ -252,9 +270,8 @@ def drive_upload(arg: str) -> str:
|
||||
"""Upload de arquivo para o Drive. Arg: account filepath (ex: ma /tmp/relat.pdf)"""
|
||||
try:
|
||||
parts = arg.split(maxsplit=1)
|
||||
account, filepath = parts[0], parts[1]
|
||||
mapping = {"ma": "gws-mr", "mr": "gws-mr", "adm": "gws-adm", "4r": "gws-4r"}
|
||||
account = mapping.get(account.lower(), account)
|
||||
account_alias, filepath = parts[0], parts[1]
|
||||
account = resolve_account(account_alias)
|
||||
filename = os.path.basename(filepath)
|
||||
cmd = f"{account} drive files create --json '{{\"name\": \"{filename}\"}}' --output {filepath}"
|
||||
return run_bash_command(cmd)
|
||||
@@ -264,21 +281,15 @@ def calendar_agenda(arg: str) -> str:
|
||||
"""Busca os próximos eventos no calendário. Arg: account timeframe (timeframe: today, tomorrow, week, days=N)"""
|
||||
try:
|
||||
parts = arg.split(maxsplit=1)
|
||||
account = parts[0]
|
||||
account_alias = parts[0]
|
||||
timeframe = parts[1] if len(parts) > 1 else "--today"
|
||||
|
||||
mapping = {"ma": "gws-mr", "mr": "gws-mr", "adm": "gws-adm", "4r": "gws-4r"}
|
||||
account = mapping.get(account.lower(), account)
|
||||
|
||||
# Converte timeframe para flag correta
|
||||
if not timeframe.startswith("--"):
|
||||
timeframe = f"--{timeframe}"
|
||||
|
||||
account = resolve_account(account_alias)
|
||||
if not timeframe.startswith("--"): timeframe = f"--{timeframe}"
|
||||
cmd = f"{account} calendar +agenda {timeframe}"
|
||||
return run_bash_command(cmd)
|
||||
except Exception as e: return f"Erro no Calendário: {str(e)}"
|
||||
|
||||
# Mapeamento para o Agente entender quais tools ele possui (será usado no loop ReAct)
|
||||
# Mapeamento para o Agente entender quais tools ele possui
|
||||
AVAILABLE_TOOLS = {
|
||||
"run_bash_command": {
|
||||
"description": "Executa comandos Linux na VPS. Use para docker, git, mkdir, touch, etc.",
|
||||
|
||||
54
validate_tools.py
Normal file
54
validate_tools.py
Normal file
@@ -0,0 +1,54 @@
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
from tools import gmail_manage_label, gmail_manage_filter, list_gmail_emails, run_bash_command, resolve_account
|
||||
|
||||
def test_account(account_alias: str):
|
||||
print(f"\n--- INICIANDO TESTES PARA CONTA: {account_alias} ---")
|
||||
|
||||
# 1. Teste de Listagem de E-mails
|
||||
print(f"[TEST] Listando e-mails...")
|
||||
emails = list_gmail_emails(account_alias)
|
||||
if "Sucesso" in emails or "📧" in emails:
|
||||
print(f"✅ SUCESSO na listagem (Amostra): {emails[:50]}...")
|
||||
else:
|
||||
print(f"❌ FALHA na listagem: {emails}")
|
||||
|
||||
# 2. Teste de Criação de Marcador (Label)
|
||||
test_label = "ANTIGRAVITY_PROACTIVE_TEST"
|
||||
print(f"[TEST] Criando marcador '{test_label}'...")
|
||||
res_label = gmail_manage_label(f"{account_alias} {test_label}")
|
||||
if "Sucesso" in res_label or "já existe" in res_label or "Criação" in res_label:
|
||||
print(f"✅ SUCESSO no marcador: {res_label}")
|
||||
else:
|
||||
print(f"❌ FALHA no marcador: {res_label}")
|
||||
|
||||
# 3. Teste de Criação de Filtro
|
||||
print(f"[TEST] Criando filtro de teste...")
|
||||
res_filter = gmail_manage_filter(f"{account_alias} test@example.com {test_label}")
|
||||
if "Sucesso" in res_filter or "Sucesso" in str(res_filter) or "{" in str(res_filter):
|
||||
print(f"✅ SUCESSO no filtro: {res_filter}")
|
||||
else:
|
||||
print(f"❌ FALHA no filtro: {res_filter}")
|
||||
|
||||
# 4. Limpeza
|
||||
print(f"[TEST] Limpando...")
|
||||
account = resolve_account(account_alias)
|
||||
# Busca ID para deletar
|
||||
list_res = run_bash_command(f"{account} gmail users labels list")
|
||||
try:
|
||||
data = json.loads(list_res)
|
||||
for l in data.get("labels", []):
|
||||
if l["name"] == test_label:
|
||||
run_bash_command(f"{account} gmail users labels delete --params '{{\"userId\": \"me\", \"id\": \"{l['id']}\"}}'")
|
||||
print(f"✅ Limpeza concluída.")
|
||||
break
|
||||
except: pass
|
||||
|
||||
if __name__ == "__main__":
|
||||
accounts = ["adm", "mr", "4r"]
|
||||
for acc in accounts:
|
||||
try:
|
||||
test_account(acc)
|
||||
except Exception as e:
|
||||
print(f"❌ Erro crítico ao testar {acc}: {e}")
|
||||
Reference in New Issue
Block a user