Add Gmail and email bot scripts to repo
Some checks failed
Build and Deploy to Production / build-and-deploy (push) Has been cancelled
Some checks failed
Build and Deploy to Production / build-and-deploy (push) Has been cancelled
This commit is contained in:
176
scripts/gmail/gmail.py
Executable file
176
scripts/gmail/gmail.py
Executable file
@@ -0,0 +1,176 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Gmail Assistant for OpenClaw
|
||||
Read and send emails via Gmail IMAP/SMTP
|
||||
"""
|
||||
|
||||
import json
|
||||
import imaplib
|
||||
import smtplib
|
||||
import email
|
||||
from email.mime.text import MIMEText
|
||||
from email.mime.multipart import MIMEMultipart
|
||||
from datetime import datetime
|
||||
import os
|
||||
import sys
|
||||
|
||||
CREDENTIALS_FILE = "/root/.openclaw/credentials/gmail.json"
|
||||
|
||||
def load_credentials():
|
||||
"""Load Gmail credentials from secure file"""
|
||||
if not os.path.exists(CREDENTIALS_FILE):
|
||||
print(f"❌ Credentials file not found: {CREDENTIALS_FILE}")
|
||||
print("Please create it from the template:")
|
||||
print(f" cp {CREDENTIALS_FILE}.template {CREDENTIALS_FILE}")
|
||||
print(" # Then edit with your email and app password")
|
||||
sys.exit(1)
|
||||
|
||||
with open(CREDENTIALS_FILE, 'r') as f:
|
||||
return json.load(f)
|
||||
|
||||
def connect_imap(creds):
|
||||
"""Connect to Gmail IMAP server"""
|
||||
mail = imaplib.IMAP4_SSL(creds['imap_server'], creds['imap_port'])
|
||||
mail.login(creds['email'], creds['app_password'])
|
||||
return mail
|
||||
|
||||
def connect_smtp(creds):
|
||||
"""Connect to Gmail SMTP server"""
|
||||
server = smtplib.SMTP(creds['smtp_server'], creds['smtp_port'])
|
||||
server.starttls()
|
||||
server.login(creds['email'], creds['app_password'])
|
||||
return server
|
||||
|
||||
def fetch_unread_emails(limit=10):
|
||||
"""Fetch unread emails from inbox"""
|
||||
creds = load_credentials()
|
||||
mail = connect_imap(creds)
|
||||
|
||||
mail.select('inbox')
|
||||
status, messages = mail.search(None, 'UNSEEN')
|
||||
|
||||
if status != 'OK' or not messages[0]:
|
||||
print("📭 No unread emails")
|
||||
mail.logout()
|
||||
return []
|
||||
|
||||
email_ids = messages[0].split()[-limit:] # Get last 'limit' unread
|
||||
emails = []
|
||||
|
||||
for e_id in email_ids:
|
||||
status, msg_data = mail.fetch(e_id, '(RFC822)')
|
||||
if status == 'OK':
|
||||
raw_email = msg_data[0][1]
|
||||
msg = email.message_from_bytes(raw_email)
|
||||
|
||||
emails.append({
|
||||
'id': e_id.decode(),
|
||||
'from': msg['From'],
|
||||
'subject': msg['Subject'],
|
||||
'date': msg['Date'],
|
||||
'body': get_email_body(msg)
|
||||
})
|
||||
|
||||
mail.logout()
|
||||
return emails
|
||||
|
||||
def get_email_body(msg):
|
||||
"""Extract text body from email message"""
|
||||
if msg.is_multipart():
|
||||
for part in msg.walk():
|
||||
if part.get_content_type() == "text/plain":
|
||||
return part.get_payload(decode=True).decode('utf-8', errors='ignore')
|
||||
else:
|
||||
return msg.get_payload(decode=True).decode('utf-8', errors='ignore')
|
||||
return ""
|
||||
|
||||
def send_email(to_address, subject, body, preview_only=True):
|
||||
"""Send an email (or preview it)"""
|
||||
creds = load_credentials()
|
||||
|
||||
msg = MIMEMultipart()
|
||||
msg['From'] = creds['email']
|
||||
msg['To'] = to_address
|
||||
msg['Subject'] = subject
|
||||
msg['Date'] = datetime.now().strftime('%a, %d %b %Y %H:%M:%S +0000')
|
||||
|
||||
msg.attach(MIMEText(body, 'plain'))
|
||||
|
||||
if preview_only:
|
||||
print("\n" + "="*60)
|
||||
print("📧 EMAIL PREVIEW (Not Sent)")
|
||||
print("="*60)
|
||||
print(f"From: {msg['From']}")
|
||||
print(f"To: {msg['To']}")
|
||||
print(f"Subject: {msg['Subject']}")
|
||||
print(f"Date: {msg['Date']}")
|
||||
print("-"*60)
|
||||
print(body)
|
||||
print("="*60)
|
||||
print("\n✉️ To send this email, run with --send flag\n")
|
||||
return
|
||||
|
||||
# Actually send
|
||||
server = connect_smtp(creds)
|
||||
server.send_message(msg)
|
||||
server.quit()
|
||||
print(f"✅ Email sent to {to_address}")
|
||||
|
||||
def list_emails(limit=5):
|
||||
"""List recent unread emails"""
|
||||
emails = fetch_unread_emails(limit)
|
||||
|
||||
if not emails:
|
||||
return
|
||||
|
||||
print(f"\n📬 {len(emails)} Unread Email(s):\n")
|
||||
for i, email_data in enumerate(emails, 1):
|
||||
print(f"{i}. From: {email_data['from']}")
|
||||
print(f" Subject: {email_data['subject']}")
|
||||
print(f" Date: {email_data['date']}")
|
||||
print(f" Preview: {email_data['body'][:100]}...")
|
||||
print()
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
print("Gmail Assistant for OpenClaw")
|
||||
print("\nUsage:")
|
||||
print(" gmail.py list [limit] - List unread emails")
|
||||
print(" gmail.py read <id> - Read full email")
|
||||
print(" gmail.py send <to> <subject> <body> - Preview email")
|
||||
print(" gmail.py send --send <to> <subject> <body> - Actually send")
|
||||
sys.exit(1)
|
||||
|
||||
command = sys.argv[1]
|
||||
|
||||
if command == "list":
|
||||
limit = int(sys.argv[2]) if len(sys.argv) > 2 else 5
|
||||
list_emails(limit)
|
||||
|
||||
elif command == "send":
|
||||
if len(sys.argv) < 5:
|
||||
print("Usage: gmail.py send [--send] <to> <subject> <body>")
|
||||
sys.exit(1)
|
||||
|
||||
preview_only = True
|
||||
args = sys.argv[2:]
|
||||
|
||||
if args[0] == "--send":
|
||||
preview_only = False
|
||||
args = args[1:]
|
||||
|
||||
if len(args) < 3:
|
||||
print("Usage: gmail.py send [--send] <to> <subject> <body>")
|
||||
sys.exit(1)
|
||||
|
||||
to_address = args[0]
|
||||
subject = args[1]
|
||||
body = args[2]
|
||||
|
||||
send_email(to_address, subject, body, preview_only)
|
||||
|
||||
else:
|
||||
print(f"Unknown command: {command}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user