#!/usr/bin/env python3 """ Gmail to SQLite Pipeline Monitors Gmail for crypto leads and saves to database """ import os import sqlite3 from datetime import datetime 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 base64 import re from dotenv import load_dotenv load_dotenv() SCOPES = ['https://www.googleapis.com/auth/gmail.readonly'] class GmailPipeline: """Gmail to SQLite pipeline""" def __init__(self): self.service = self._authenticate() self.db_conn = self._init_database() def _authenticate(self): """Authenticate with Gmail API""" creds = None if os.path.exists('token.json'): creds = Credentials.from_authorized_user_file('token.json', SCOPES) if not creds or not creds.valid: if creds and creds.expired and creds.refresh_token: creds.refresh(Request()) else: flow = InstalledAppFlow.from_client_secrets_file( 'credentials.json', SCOPES) creds = flow.run_local_server(port=0) with open('token.json', 'w') as token: token.write(creds.to_json()) return build('gmail', 'v1', credentials=creds) def _init_database(self): """Initialize SQLite database""" conn = sqlite3.connect('leads.db') cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS leads ( id INTEGER PRIMARY KEY AUTOINCREMENT, email_id TEXT UNIQUE, subject TEXT, sender TEXT, sender_email TEXT, body TEXT, received_date TEXT, labels TEXT, processed INTEGER DEFAULT 0, notes TEXT ) ''') conn.commit() return conn def fetch_crypto_leads(self, label='Crypto Lead'): """Fetch emails with specific label""" try: # Get label ID labels = self.service.users().labels().list(userId='me').execute() label_id = None for l in labels.get('labels', []): if l['name'] == label: label_id = l['id'] break if not label_id: print(f"Label '{label}' not found. Creating it...") label_body = {'name': label} created_label = self.service.users().labels().create( userId='me', body=label_body).execute() label_id = created_label['id'] # Fetch messages results = self.service.users().messages().list( userId='me', labelIds=[label_id]).execute() messages = results.get('messages', []) print(f"Found {len(messages)} messages with label '{label}'") for msg in messages: self._process_message(msg['id']) return len(messages) except Exception as e: print(f"Error fetching emails: {e}") return 0 def _process_message(self, msg_id): """Process individual message""" try: message = self.service.users().messages().get( userId='me', id=msg_id, format='full').execute() headers = message['payload']['headers'] # Extract metadata subject = next((h['value'] for h in headers if h['name'] == 'Subject'), '') from_header = next((h['value'] for h in headers if h['name'] == 'From'), '') date = next((h['value'] for h in headers if h['name'] == 'Date'), '') # Parse sender sender_match = re.match(r'(.*?)\s*<(.+?)>', from_header) if sender_match: sender_name = sender_match.group(1).strip() sender_email = sender_match.group(2).strip() else: sender_name = from_header sender_email = from_header # Extract body body = self._get_message_body(message) # Get labels label_ids = message.get('labelIds', []) labels = ','.join(label_ids) # Save to database cursor = self.db_conn.cursor() cursor.execute(''' INSERT OR IGNORE INTO leads (email_id, subject, sender, sender_email, body, received_date, labels) VALUES (?, ?, ?, ?, ?, ?, ?) ''', (msg_id, subject, sender_name, sender_email, body, date, labels)) self.db_conn.commit() if cursor.rowcount > 0: print(f"✅ Saved: {subject[:50]}...") except Exception as e: print(f"Error processing message {msg_id}: {e}") def _get_message_body(self, message): """Extract message body""" try: if 'parts' in message['payload']: parts = message['payload']['parts'] for part in parts: if part['mimeType'] == 'text/plain': data = part['body'].get('data', '') if data: return base64.urlsafe_b64decode(data).decode('utf-8') else: data = message['payload']['body'].get('data', '') if data: return base64.urlsafe_b64decode(data).decode('utf-8') except Exception as e: print(f"Error extracting body: {e}") return '' def export_to_notion(self): """Export leads to Notion (requires notion-client)""" try: from notion_client import Client notion = Client(auth=os.getenv('NOTION_API_KEY')) database_id = os.getenv('NOTION_DATABASE_ID') cursor = self.db_conn.cursor() cursor.execute('SELECT * FROM leads WHERE processed = 0') leads = cursor.fetchall() for lead in leads: notion.pages.create( parent={'database_id': database_id}, properties={ 'Name': {'title': [{'text': {'content': lead[2]}}]}, # subject 'Email': {'email': lead[4]}, # sender_email 'Status': {'select': {'name': 'New'}}, 'Source': {'select': {'name': 'Gmail'}}, } ) # Mark as processed cursor.execute('UPDATE leads SET processed = 1 WHERE id = ?', (lead[0],)) self.db_conn.commit() print(f"✅ Exported {len(leads)} leads to Notion") except ImportError: print("❌ notion-client not installed. Run: pip install notion-client") except Exception as e: print(f"Error exporting to Notion: {e}") def print_statistics(self): """Print statistics""" cursor = self.db_conn.cursor() cursor.execute('SELECT COUNT(*) FROM leads') total = cursor.fetchone()[0] cursor.execute('SELECT COUNT(*) FROM leads WHERE processed = 1') processed = cursor.fetchone()[0] print("\n" + "="*50) print("📧 EMAIL PIPELINE STATISTICS") print("="*50) print(f"Total Leads: {total}") print(f"Processed: {processed}") print(f"Pending: {total - processed}") print("="*50 + "\n") def close(self): """Close database connection""" self.db_conn.close() def main(): """Main entry point""" import argparse parser = argparse.ArgumentParser(description='Gmail to SQLite Pipeline') parser.add_argument('--label', default='Crypto Lead', help='Gmail label to monitor') parser.add_argument('--export-notion', action='store_true', help='Export to Notion') parser.add_argument('--stats', action='store_true', help='Show statistics only') args = parser.parse_args() pipeline = GmailPipeline() try: if args.stats: pipeline.print_statistics() else: pipeline.fetch_crypto_leads(label=args.label) if args.export_notion: pipeline.export_to_notion() pipeline.print_statistics() finally: pipeline.close() if __name__ == '__main__': main()