#!/usr/bin/env python3 """ Approval Processor for Swing Shift Bot. Scans the approval queue, sends pending items to Larry on Telegram, and processes approved/denied actions. Works with OpenClaw's Telegram channel. Queue directory: ops/approvals/pending/ Processed directory: ops/approvals/processed/ Each approval file is a JSON with: - type: "sms", "email", "campaign", "booking" - to: recipient - message/subject/body: content to send - status: "pending_approval" → "approved" / "denied" """ import json import os import sys import time import urllib.request from datetime import datetime from pathlib import Path PROJECT_DIR = Path(__file__).parent.parent PENDING_DIR = PROJECT_DIR / "ops" / "approvals" / "pending" PROCESSED_DIR = PROJECT_DIR / "ops" / "approvals" / "processed" TELEGRAM_CHAT_ID = "1440579195" PROCESSED_DIR.mkdir(parents=True, exist_ok=True) PENDING_DIR.mkdir(parents=True, exist_ok=True) def send_telegram(text): """Send message to Larry via OpenClaw gateway.""" try: data = json.dumps({ "channel": "telegram", "to": TELEGRAM_CHAT_ID, "text": text, }).encode() req = urllib.request.Request( "http://127.0.0.1:18789/api/send", data=data, headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(req, timeout=10) as resp: return json.loads(resp.read().decode()) except Exception as e: print(f"Telegram send failed: {e}", file=sys.stderr) return None def format_approval_message(item, filename): """Format a pending approval for Telegram display.""" item_type = item.get("type", "unknown") lines = [f"📋 APPROVAL NEEDED — {item_type.upper()}"] if item_type == "sms": lines.append(f"📱 To: {item.get('to', '?')}") lines.append(f"📝 Template: {item.get('template', 'custom')}") msg = item.get("message", "") if len(msg) > 200: msg = msg[:200] + "..." lines.append(f"💬 Message: {msg}") elif item_type == "email": lines.append(f"📧 To: {item.get('to', '?')}") lines.append(f"📝 Subject: {item.get('subject', '?')}") body = item.get("body", "") if len(body) > 200: body = body[:200] + "..." lines.append(f"💬 Preview: {body}") elif item_type == "campaign": lines.append(f"📣 Campaign: {item.get('name', '?')}") lines.append(f"👥 Recipients: {item.get('recipient_count', '?')}") lines.append(f"📝 Subject: {item.get('subject', '?')}") else: lines.append(f"Details: {json.dumps(item, indent=2)[:300]}") lines.append(f"\nCreated: {item.get('created', '?')}") lines.append(f"File: {filename}") lines.append("\nReply 'approve [filename]' or 'deny [filename]'") lines.append("Or reply 'approve all' to approve everything pending.") return "\n".join(lines) def scan_and_notify(): """Scan pending approvals and notify Larry.""" pending = list(PENDING_DIR.glob("*.json")) if not pending: print("No pending approvals.") return 0 print(f"Found {len(pending)} pending approvals") for path in pending: try: item = json.loads(path.read_text()) except json.JSONDecodeError: print(f" Skipping corrupt file: {path.name}") continue if item.get("notified"): continue # Already sent to Telegram msg = format_approval_message(item, path.name) result = send_telegram(msg) if result: item["notified"] = True item["notified_at"] = datetime.now().isoformat() path.write_text(json.dumps(item, indent=2)) print(f" Notified: {path.name}") else: print(f" Failed to notify: {path.name}") return len(pending) def process_approval(filename, action="approve"): """Process an approval/denial.""" path = PENDING_DIR / filename if not path.exists(): # Try partial match matches = [p for p in PENDING_DIR.glob("*.json") if filename in p.name] if len(matches) == 1: path = matches[0] elif len(matches) > 1: print(f"Ambiguous: {[m.name for m in matches]}") return False else: print(f"Not found: {filename}") return False item = json.loads(path.read_text()) item["status"] = "approved" if action == "approve" else "denied" item["processed_at"] = datetime.now().isoformat() # Move to processed dest = PROCESSED_DIR / path.name dest.write_text(json.dumps(item, indent=2)) path.unlink() print(f" {action.upper()}: {path.name}") if action == "approve": execute_approved(item) return True def process_all(action="approve"): """Approve or deny all pending items.""" pending = list(PENDING_DIR.glob("*.json")) for path in pending: process_approval(path.name, action) return len(pending) def execute_approved(item): """Execute an approved action.""" item_type = item.get("type", "") if item_type == "sms": # Import and send via TextMagic print(f" Executing SMS to {item.get('to')}") try: from textmagic_sender import send_sms result = send_sms(item["to"], item["message"], item.get("media_url")) print(f" SMS result: {result}") except Exception as e: print(f" SMS execution failed: {e}") elif item_type == "email": print(f" Email sending not yet automated — flagged for manual send") send_telegram(f"✅ Approved email to {item.get('to')} — send manually from MailerLite") elif item_type == "campaign": print(f" Campaign launch not yet automated — flagged for manual activation") send_telegram(f"✅ Approved campaign '{item.get('name')}' — activate in MailerLite") else: print(f" Unknown type '{item_type}' — no auto-execution") if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Swing Shift Approval Processor") sub = parser.add_subparsers(dest="command") sub.add_parser("scan", help="Scan pending and notify on Telegram") sub.add_parser("list", help="List pending approvals") ap = sub.add_parser("approve", help="Approve an item") ap.add_argument("filename", nargs="?", default="all", help="Filename or 'all'") dp = sub.add_parser("deny", help="Deny an item") dp.add_argument("filename", help="Filename to deny") args = parser.parse_args() if args.command == "scan": scan_and_notify() elif args.command == "list": pending = list(PENDING_DIR.glob("*.json")) if not pending: print("No pending approvals.") for p in pending: item = json.loads(p.read_text()) print(f" [{item.get('type','?')}] {p.name} — {item.get('status','?')}") elif args.command == "approve": if args.filename == "all": n = process_all("approve") print(f"Approved {n} items") else: process_approval(args.filename, "approve") elif args.command == "deny": process_approval(args.filename, "deny") else: parser.print_help()