#!/usr/bin/env python3 """ Atlas Brain — Enis WhatsApp monitor + auto-reply Polls Evolution API for new Enis messages, replies via local LLM with guardrails. State tracked in /opt/atlas/brain/enis_state.json """ import json, os, time, requests, logging, re from pathlib import Path from datetime import datetime STATE_FILE = "/opt/atlas/brain/enis_state.json" LOG_FILE = "/opt/atlas/logs/enis_monitor.log" EVO_URL = "http://localhost:8120" EVO_KEY = "atlas-evo-key-2026" ENIS_JID = "124674495234167@lid" LLM_URL = "http://localhost:8888/v1/chat/completions" POLL_SEC = 30 Path("/opt/atlas/brain").mkdir(parents=True, exist_ok=True) Path("/opt/atlas/logs").mkdir(parents=True, exist_ok=True) logging.basicConfig( filename=LOG_FILE, level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s" ) log = logging.getLogger("enis-monitor") SYSTEM_PROMPT = """You are Atlas Brain — the AI core of Atlas Corporation, built by Chaib Aarab in Tilburg, NL. You are talking to Enis, a friend and fellow AI/tech builder who is building his own second brain. Your personality: knowledgeable, direct, friendly, Muslim-aware (use mashAllah/SubhanAllah naturally when fitting). You speak as a peer builder, not as an assistant. HARD GUARDRAILS — never break these: 1. NEVER share passwords, API keys, access tokens, internal IPs, or any credentials. 2. NEVER discuss Chaib's personal finances, bijstand, gemeente/municipality matters, or welfare benefits. 3. NEVER make financial promises or commitments on behalf of Atlas Corporation. 4. NEVER claim to be human or deny being an AI when sincerely asked. 5. NEVER discuss client data, Odoo leads, or confidential business details. 6. NEVER share the internal architecture (specific ports, container names, secret paths). 7. If asked to do something Chaib hasn't approved, say you'll flag it for him. 8. If Enis asks for the git repo: https://git.atlascorporation.nl/chaib/atlas-os (this is public, safe to share). 9. Keep replies concise — WhatsApp, not essays. 2-4 sentences max unless a technical question needs more. 10. You can discuss: AI/LLM tech, second brain architecture, Atlas OS, how you work (at a high level), Enis's projects. If you are uncertain whether something is safe to share, say "I'll check with Chaib on that." """ def load_state(): if Path(STATE_FILE).exists(): return json.loads(Path(STATE_FILE).read_text()) return {"last_ts": 0, "last_id": ""} def save_state(s): Path(STATE_FILE).write_text(json.dumps(s)) def get_new_enis_messages(since_ts): try: r = requests.post( f"{EVO_URL}/chat/findMessages/atlas", headers={"apikey": EVO_KEY, "Content-Type": "application/json"}, json={"where": {"key": {"remoteJid": ENIS_JID, "fromMe": False}}, "limit": 20}, timeout=10 ) data = r.json() records = data.get("messages", {}).get("records", []) if isinstance(data, dict) else data new = [m for m in records if m.get("messageTimestamp", 0) > since_ts] new.sort(key=lambda m: m.get("messageTimestamp", 0)) return new except Exception as e: log.error(f"fetch error: {e}") return [] def extract_text(m): msg = m.get("message", {}) if "conversation" in msg: return msg["conversation"] if "extendedTextMessage" in msg: return msg["extendedTextMessage"].get("text", "") if "imageMessage" in msg: return f"[image: {msg['imageMessage'].get('caption', '')}]" if "audioMessage" in msg: return "[voice message]" if "videoMessage" in msg: return "[video]" if "documentMessage" in msg: return f"[file: {msg['documentMessage'].get('fileName', '')}]" return "" def _groq_key(): try: txt = open('/opt/atlas/router-secrets.env').read() m = re.search(r'GROQ_API_KEY=(.+)', txt) return m.group(1).strip() if m else '' except: return '' def generate_reply(user_text): messages = [ {'role': 'system', 'content': SYSTEM_PROMPT}, {'role': 'user', 'content': user_text} ] try: r = requests.post(LLM_URL, json={ 'model': 'atlas-free', 'messages': messages, 'max_tokens': 300, 'stream': False, 'temperature': 0.7 }, timeout=30) result = r.json()['choices'][0]['message']['content'].strip() if result: log.info('replied via local LLM') return result except Exception as e: log.warning(f'local LLM fail: {e}') try: key = _groq_key() r = requests.post( 'https://api.groq.com/openai/v1/chat/completions', json={'model': 'llama-3.3-70b-versatile', 'messages': messages, 'max_tokens': 300}, headers={'Authorization': f'Bearer {key}'}, timeout=30 ) result = r.json()['choices'][0]['message']['content'].strip() log.info('replied via Groq') return result except Exception as e: log.error(f'Groq fail: {e}') return None def send_reply(text): try: payload = {"number": ENIS_JID, "textMessage": {"text": text}} r = requests.post( f"{EVO_URL}/message/sendText/atlas", headers={"apikey": EVO_KEY, "Content-Type": "application/json"}, json=payload, timeout=15 ) resp = r.json() if resp.get("key"): log.info(f"sent OK: id={resp['key'].get('id','?')}") return True log.warning(f"send failed: {resp}") return False except Exception as e: log.error(f"send error: {e}") return False def main(): log.info("Enis monitor started") state = load_state() while True: try: msgs = get_new_enis_messages(state["last_ts"]) for m in msgs: ts = m.get("messageTimestamp", 0) mid = m.get("key", {}).get("id", "") txt = extract_text(m) if not txt or mid == state.get("last_id"): state["last_ts"] = max(state["last_ts"], ts) save_state(state) continue log.info(f"new msg ts={ts} id={mid}: {txt[:80]}") reply = generate_reply(txt) if reply: time.sleep(2) # brief typing pause ok = send_reply(reply) log.info(f"reply={'sent' if ok else 'FAILED'}: {reply[:60]}") state["last_ts"] = max(state["last_ts"], ts) state["last_id"] = mid save_state(state) time.sleep(3) # gap between multi-message bursts except Exception as e: log.error(f"loop error: {e}") time.sleep(POLL_SEC) if __name__ == "__main__": main()