#!/usr/bin/env python3
"""Email monitoring bot that alerts on specific subject lines."""
import os
import sys
import json
from datetime import datetime
# Add project root to path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
# Import config
import config
def get_credentials():
"""Get valid credentials for Gmail API."""
creds = None
# Check if token file exists
if os.path.exists(config.TOKEN_FILE):
creds = Credentials.from_authorized_user_file(config.TOKEN_FILE, config.SCOPES)
# If credentials are not valid, refresh or re-authenticate
if not creds or not creds.valid:
if creds and creds.expired and creds.refresh_token:
creds.refresh(Request())
# Save refreshed token
with open(config.TOKEN_FILE, 'w') as token:
token.write(creds.to_json())
else:
# Re-authenticate
if not os.path.exists(config.CREDENTIALS_FILE):
print(f"Error: Credentials file not found at {config.CREDENTIALS_FILE}")
print("Download from: https://console.cloud.google.com/apis/credentials")
return None
flow = InstalledAppFlow.from_client_secrets_file(
config.CREDENTIALS_FILE, config.SCOPES
)
creds = flow.run_local_server(port=0)
# Save token
with open(config.TOKEN_FILE, 'w') as token:
token.write(creds.to_json())
return creds
def get_gmail_service():
"""Build Gmail API service."""
creds = get_credentials()
if not creds:
return None
try:
service = build('gmail', 'v1', credentials=creds)
return service
except Exception as e:
print(f"Error building Gmail service: {e}")
return None
def get_unread_emails(service, max_results=50):
"""Get unread emails from inbox."""
try:
# Search for unread emails
results = service.users().messages().list(
userId='me',
q='is:unread',
maxResults=max_results
).execute()
messages = results.get('messages', [])
return messages
except Exception as e:
print(f"Error fetching emails: {e}")
return []
def get_email_details(service, message_id):
"""Get full email details."""
try:
message = service.users().messages().get(
userId='me',
id=message_id,
format='full'
).execute()
return message
except Exception as e:
print(f"Error fetching message {message_id}: {e}")
return None
def extract_subject(message):
"""Extract subject from email message."""
headers = message.get('payload', {}).get('headers', [])
for header in headers:
if header['name'].lower() == 'subject':
return header['value']
return ''
def extract_sender(message):
"""Extract sender from email message."""
headers = message.get('payload', {}).get('headers', [])
for header in headers:
if header['name'].lower() == 'from':
return header['value']
return ''
def extract_snippet(message):
"""Extract snippet (preview) from email."""
return message.get('snippet', 'No preview available')
def send_telegram_alert(subject, sender, snippet, timestamp):
"""Send alert to Telegram."""
if not config.TELEGRAM_BOT_TOKEN or not config.TELEGRAM_CHAT_ID:
print("Warning: Telegram credentials not configured")
print("Set TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID environment variables")
return False
try:
import requests
# Format timestamp
dt = datetime.fromtimestamp(timestamp / 1000)
time_str = dt.strftime('%Y-%m-%d %H:%M')
# Create message
message = f"""📧 *New Email Alert*
*Subject:* {subject}
*From:* {sender}
*Time:* {time_str}
*Preview:* {snippet[:100]}..."""
# Send to Telegram
url = f"https://api.telegram.org/bot{config.TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
'chat_id': config.TELEGRAM_CHAT_ID,
'text': message,
'parse_mode': 'Markdown'
}
response = requests.post(url, json=payload, timeout=10)
response.raise_for_status()
print(f"✅ Alert sent to Telegram")
return True
except Exception as e:
print(f"❌ Error sending Telegram alert: {e}")
return False
def mark_as_read(service, message_id):
"""Mark email as read to avoid duplicate alerts."""
try:
service.users().messages().modify(
userId='me',
id=message_id,
body={'removeLabelIds': ['UNREAD']}
).execute()
return True
except Exception as e:
print(f"Error marking message {message_id} as read: {e}")
return False
def check_emails():
"""Main function to check for matching emails."""
print(f"🔍 Checking for emails with subject: '{config.SUBJECT_FILTER}'")
# Get Gmail service
service = get_gmail_service()
if not service:
print("❌ Failed to connect to Gmail")
return
# Get unread emails
messages = get_unread_emails(service, config.MAX_RESULTS)
print(f"📭 Found {len(messages)} unread emails")
if not messages:
print("✅ No unread emails")
return
# Check each email
alert_count = 0
for message in messages:
message_id = message['id']
# Get full message details
message_data = get_email_details(service, message_id)
if not message_data:
continue
# Extract subject
subject = extract_subject(message_data)
# Check if subject matches
if config.SUBJECT_FILTER.lower() in subject.lower():
print(f"🎯 Match found: '{subject}'")
# Extract details
sender = extract_sender(message_data)
snippet = extract_snippet(message_data)
timestamp = message_data.get('internalDate', 0)
# Send alert
if send_telegram_alert(subject, sender, snippet, timestamp):
alert_count += 1
# Mark as read to avoid duplicate alerts
mark_as_read(service, message_id)
if alert_count > 0:
print(f"✅ Sent {alert_count} alert(s)")
else:
print("✅ No matching emails found")
def test_telegram():
"""Test Telegram connection."""
if not config.TELEGRAM_BOT_TOKEN or not config.TELEGRAM_CHAT_ID:
print("❌ Telegram not configured")
print("Set TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID environment variables")
return
try:
import requests
message = "🤖 Email monitoring bot is ready!"
url = f"https://api.telegram.org/bot{config.TELEGRAM_BOT_TOKEN}/sendMessage"
payload = {
'chat_id': config.TELEGRAM_CHAT_ID,
'text': message
}
response = requests.post(url, json=payload, timeout=10)
response.raise_for_status()
print("✅ Telegram test message sent successfully")
except Exception as e:
print(f"❌ Telegram test failed: {e}")
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description='Email monitoring bot')
parser.add_argument('--test-telegram', action='store_true', help='Test Telegram connection')
parser.add_argument('--subject', type=str, help='Override subject filter')
args = parser.parse_args()
# Override subject if provided
if args.subject:
config.SUBJECT_FILTER = args.subject
if args.test_telegram:
test_telegram()
else:
check_emails()