AgenticControl/read_inbox.py

import json
import os
import sqlite3
import re
from datetime import datetime
from google.oauth2.credentials import Credentials
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
import sys
import base64
from dotenv import load_dotenv

load_dotenv(dotenv_path=os.path.join(os.environ.get('WORKSPACE_ROOT', '.'), 'backend/.env'))

CLIENT_ID = os.getenv('GOOGLE_CLIENT_ID')
CLIENT_SECRET = os.getenv('GOOGLE_CLIENT_SECRET')
DATABASE_FILE = os.path.join(os.environ.get('WORKSPACE_ROOT', '.'), 'backend/database.json')
SENT_DB_PATH = os.path.join(os.environ.get('WORKSPACE_ROOT', '.'), 'Database/FollowUps/sent_emails.db')
QUEUE_DB_PATH = os.path.join(os.environ.get('WORKSPACE_ROOT', '.'), 'Database/EmailsSent/email_to_be_sent.db')
INBOX_DB_DIR = os.path.join(os.environ.get('WORKSPACE_ROOT', '.'), 'Database/Inbox')
INBOX_DB_PATH = os.path.join(INBOX_DB_DIR, 'inbox.db')

def init_db():
    os.makedirs(INBOX_DB_DIR, exist_ok=True)
    conn = sqlite3.connect(INBOX_DB_PATH)
    cursor = conn.cursor()
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS inbox_threads (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            thread_id TEXT UNIQUE,
            company_email TEXT,
            company_name TEXT,
            subject TEXT,
            last_messages_json TEXT,
            date_received DATETIME,
            status TEXT DEFAULT 'pending_reply'
        )
    ''')
    conn.commit()
    return conn

def get_contacted_emails():
    emails = {}
    
    # Check sent_emails.db
    if os.path.exists(SENT_DB_PATH):
        try:
            conn = sqlite3.connect(SENT_DB_PATH)
            cursor = conn.cursor()
            cursor.execute("SELECT company_email, company_name FROM sent_applications")
            for row in cursor.fetchall():
                email, name = row
                if email:
                    emails[email.strip().lower()] = name
            conn.close()
        except Exception as e:
            pass

    # Check email_to_be_sent.db as requested by user
    if os.path.exists(QUEUE_DB_PATH):
        try:
            conn = sqlite3.connect(QUEUE_DB_PATH)
            cursor = conn.cursor()
            cursor.execute("SELECT company_email, company_name FROM ready_emails")
            for row in cursor.fetchall():
                email, name = row
                if email:
                    emails[email.strip().lower()] = name
            conn.close()
        except Exception as e:
            pass

    return emails

def parse_email_address(header_val):
    if not header_val: return ""
    match = re.search(r'[\w\.-]+@[\w\.-]+', header_val)
    if match:
        return match.group(0).lower()
    return ""

def get_body(payload):
    body = ""
    if 'parts' in payload:
        for part in payload['parts']:
            if part['mimeType'] == 'text/plain':
                data = part['body'].get('data', '')
                if data:
                    try:
                        body += base64.urlsafe_b64decode(data).decode('utf-8')
                    except Exception:
                        pass
            elif 'parts' in part:
                body += get_body(part)
    elif payload.get('mimeType') == 'text/plain':
        data = payload['body'].get('data', '')
        if data:
            try:
                body += base64.urlsafe_b64decode(data).decode('utf-8')
            except Exception:
                pass
    return body

def fetch_inbox():
    if not os.path.exists(DATABASE_FILE):
        print(json.dumps({"error": f"Credentials {DATABASE_FILE} not found."}))
        return

    with open(DATABASE_FILE, 'r') as f:
        user_db_data = json.load(f)

    access_token = user_db_data.get('access_token')
    refresh_token = user_db_data.get('refresh_token')

    creds = Credentials(
        token=access_token,
        refresh_token=refresh_token,
        token_uri="https://oauth2.googleapis.com/token",
        client_id=CLIENT_ID,
        client_secret=CLIENT_SECRET,
        scopes=["https://www.googleapis.com/auth/gmail.modify"]
    )

    if creds and creds.expired and creds.refresh_token:
        creds.refresh(Request())

    service = build('gmail', 'v1', credentials=creds)
    conn = init_db()
    cursor = conn.cursor()

    contacted_emails = get_contacted_emails()

    try:
        # Fetch recent messages from Inbox
        results = service.users().messages().list(userId='me', labelIds=['INBOX'], maxResults=50).execute()
        messages = results.get('messages', [])
        
        fetched_count = 0
        new_threads = []

        for msg in messages:
            msg_id = msg['id']
            thread_id = msg['threadId']
            
            # Check if we already processed this thread
            cursor.execute("SELECT id FROM inbox_threads WHERE thread_id = ?", (thread_id,))
            if cursor.fetchone():
                continue

            # Fetch full message
            full_msg = service.users().messages().get(userId='me', id=msg_id, format='metadata', metadataHeaders=['From', 'Subject', 'Date']).execute()
            headers = full_msg.get('payload', {}).get('headers', [])
            
            from_email_header = next((h['value'] for h in headers if h['name'].lower() == 'from'), "")
            from_email = parse_email_address(from_email_header)
            subject = next((h['value'] for h in headers if h['name'].lower() == 'subject'), "No Subject")
            
            if from_email in contacted_emails:
                company_name = contacted_emails[from_email]
                
                # Fetch thread to get last 5 messages (format=full to get payload body)
                thread = service.users().threads().get(userId='me', id=thread_id, format='full').execute()
                thread_messages = thread.get('messages', [])
                
                # Extract text
                last_5 = thread_messages[-5:]
                history = []
                for t_msg in last_5:
                    t_payload = t_msg.get('payload', {})
                    t_headers = t_payload.get('headers', [])
                    t_from = next((h['value'] for h in t_headers if h['name'].lower() == 'from'), "")
                    t_date = next((h['value'] for h in t_headers if h['name'].lower() == 'date'), "")
                    
                    body_content = get_body(t_payload)
                    snippet = body_content.strip() if body_content else t_msg.get('snippet', '')
                    
                    history.append({
                        "from": t_from,
                        "date": t_date,
                        "snippet": snippet
                    })

                cursor.execute('''
                    INSERT INTO inbox_threads (thread_id, company_email, company_name, subject, last_messages_json, date_received)
                    VALUES (?, ?, ?, ?, ?, ?)
                ''', (thread_id, from_email, company_name, subject, json.dumps(history), datetime.now().strftime("%Y-%m-%d %H:%M:%S")))
                conn.commit()
                fetched_count += 1
                new_threads.append({
                    "thread_id": thread_id,
                    "company_email": from_email,
                    "subject": subject
                })

        print(json.dumps({"success": True, "fetched_count": fetched_count, "new_threads": new_threads}))

    except Exception as e:
        print(json.dumps({"error": str(e)}))

if __name__ == "__main__":
    fetch_inbox()