diff --git a/tools.py b/tools.py index 1731d5d..f808d4a 100644 --- a/tools.py +++ b/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") +# ============================================================ +# 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" +} + +def resolve_account(account_alias: str) -> str: + clean = account_alias.strip().lower() + return ACCOUNT_MAPPING.get(clean, f"gws-{clean}") + 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 = { - "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) - - # 1. Obtém a lista de IDs + 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.", diff --git a/validate_tools.py b/validate_tools.py new file mode 100644 index 0000000..8fad0c0 --- /dev/null +++ b/validate_tools.py @@ -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}")