IGNORE/app-final-no-placeholders.py

import os
import re
import psycopg2
from psycopg2 import pool
import requests
import pandas as pd
from datetime import datetime
from bs4 import BeautifulSoup
from googlesearch import search
import gradio as gr
import boto3
from botocore.exceptions import NoCredentialsError, PartialCredentialsError
import openai
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging

# Configuration
aws_access_key_id = os.getenv("AWS_ACCESS_KEY_ID", "your-aws-access-key")
aws_secret_access_key = os.getenv("AWS_SECRET_ACCESS_KEY", "your-aws-secret-key")
region_name = "us-east-1"

openai.api_key = os.getenv("OPENAI_API_KEY", "your-openai-api-key")
openai.api_base = os.getenv("OPENAI_API_BASE", "https://api.openai.com/v1/")
openai_model = "text-davinci-003"

db_params = {
    "user": "your-postgres-user",
    "password": "your-postgres-password",
    "host": "your-postgres-host",
    "port": "your-postgres-port",
    "dbname": "your-postgres-dbname",
    "sslmode": "require"
}

# Initialize AWS SES client
ses_client = boto3.client('ses',
                          aws_access_key_id=aws_access_key_id,
                          aws_secret_access_key=aws_secret_access_key,
                          region_name=region_name)

# Connection pool for PostgreSQL
db_pool = pool.SimpleConnectionPool(1, 10, **db_params)

# HTTP session with retry strategy
session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[502, 503, 504])
session.mount('https://', HTTPAdapter(max_retries=retries))

# Setup logging
logging.basicConfig(level=logging.INFO, filename='app.log', filemode='a',
                    format='%(asctime)s - %(levelname)s - %(message)s')

# Initialize database
def init_db():
    conn = None
    try:
        conn = db_pool.getconn()
        conn.close()
        logging.info("Successfully connected to the database!")
    except Exception as e:
        logging.error(f"Failed to connect to the database: {e}")
    finally:
        if conn:
            db_pool.putconn(conn)

init_db()

# Fetch the most recent or frequently used template ID
def fetch_recent_template_id():
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            cursor.execute('SELECT id FROM email_templates ORDER BY last_used DESC LIMIT 1')
            recent_template_id = cursor.fetchone()[0]
        return recent_template_id
    except Exception as e:
        logging.error(f"Failed to fetch the most recent template: {e}")
        return None
    finally:
        if conn:
            db_pool.putconn(conn)

# Auto-save drafts every few seconds
def auto_save_drafts(draft_content):
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            cursor.execute('INSERT INTO drafts (content, saved_at) VALUES (%s, %s) RETURNING id',
                           (draft_content, datetime.now()))
        conn.commit()
        logging.info("Draft saved successfully.")
    except Exception as e:
        logging.error(f"Failed to save draft: {e}")
    finally:
        if conn:
            db_pool.putconn(conn)

# Auto-restore drafts when user returns
def auto_restore_drafts():
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            cursor.execute('SELECT content FROM drafts ORDER BY saved_at DESC LIMIT 1')
            draft_content = cursor.fetchone()[0]
        return draft_content
    except Exception as e:
        logging.error(f"Failed to restore draft: {e}")
        return ""
    finally:
        if conn:
            db_pool.putconn(conn)

# Save each search query automatically
def save_search_query(query):
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            cursor.execute('INSERT INTO search_terms (status, fetched_emails, last_processed_at) VALUES (%s, %s, %s) RETURNING id',
                           ('pending', 0, None))
            search_term_id = cursor.fetchone()[0]
        conn.commit()
        return search_term_id
    except Exception as e:
        logging.error(f"Failed to save search query: {e}")
        return None
    finally:
        if conn:
            db_pool.putconn(conn)

# Fetch unsent emails and auto-sort them by priority or date
def fetch_unsent_emails():
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            cursor.execute('SELECT * FROM generated_emails WHERE email_sent=0 ORDER BY email_id')
            unsent_emails = cursor.fetchall()
        return unsent_emails
    except Exception as e:
        logging.error(f"Failed to fetch unsent emails: {e}")
        return []
    finally:
        if conn:
            db_pool.putconn(conn)

# Enhanced function for tracking progress and sending emails
def track_progress_and_send(from_address, reply_to):
    progress = 0
    total_emails = len(fetch_unsent_emails())
    for email in fetch_unsent_emails():
        send_email_via_aws(email[2], email[3], email[4], from_address, reply_to)
        progress += 1
        update_progress_bar(progress / total_emails)
        notify_user(f"Sent {progress}/{total_emails} emails.")
    return "All emails sent successfully."

# Function to send emails via AWS SES
def send_email_via_aws(to_address, subject, body_html, from_address, reply_to):
    try:
        response = ses_client.send_email(
            Source=from_address,
            Destination={'ToAddresses': [to_address]},
            Message={
                'Subject': {'Data': subject},
                'Body': {
                    'Html': {'Data': body_html}
                }
            },
            ReplyToAddresses=[reply_to]
        )
        return response['MessageId']
    except (NoCredentialsError, PartialCredentialsError) as e:
        logging.error(f"AWS credentials error: {e}")
        return None
    except Exception as e:
        logging.error(f"Failed to send email via AWS SES: {e}")
        return None

# Function to update the progress bar
def update_progress_bar(progress):
    # Implement your code to update the progress bar here
    pass

# Function to notify the user with a message
def notify_user(message):
    # Implement your code to notify the user here
    pass

# Function to scrape emails from Google search results
def scrape_emails(search_query, num_results):
    results = []
    search_urls = list(search(search_query, num_results=num_results))

    for url in search_urls:
        try:
            response = session.get(url, timeout=10)
            response.encoding = 'utf-8'
            soup = BeautifulSoup(response.text, 'html.parser')
            emails = find_emails(response.text)
            for email in emails:
                results.append((search_query, email, url))
                save_lead(search_query, email, url)
        except Exception as e:
            logging.error(f"Failed to scrape {url}: {e}")

    return pd.DataFrame(results, columns=["Search Query", "Email", "URL"])

# Function to find emails in HTML text
def find_emails(html_text):
    email_pattern = re.compile(r'[\w\.-]+@[\w\.-]+')
    emails = email_pattern.findall(html_text)
    return emails

# Function to save a lead (search query, email, URL)
def save_lead(search_query, email, url):
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            cursor.execute('INSERT INTO leads (search_query, email, url) VALUES (%s, %s, %s)',
                           (search_query, email, url))
        conn.commit()
        logging.info(f"Lead saved successfully: {search_query}, {email}, {url}")
    except Exception as e:
        logging.error(f"Failed to save lead: {e}")
    finally:
        if conn:
            db_pool.putconn(conn)

# Function to generate email content using OpenAI
def generate_ai_content(search_query, email):
    prompt = f"Write an email to {email} about {search_query}:\n\n"
    response = openai.Completion.create(engine=openai_model, prompt=prompt)
    return response.choices[0].text.strip()

# Function to fetch an email template by ID
def fetch_template(template_id):
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            cursor.execute('SELECT content FROM email_templates WHERE id=%s', (template_id,))
            template_content = cursor.fetchone()[0]
        return template_content
    except Exception as e:
        logging.error(f"Failed to fetch template {template_id}: {e}")
        return ""
    finally:
        if conn:
            db_pool.putconn(conn)

# Function to update the last_used timestamp of a template
def update_template_last_used(template_id):
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            current_time = datetime.now()
            cursor.execute('UPDATE email_templates SET last_used=%s WHERE id=%s', (current_time, template_id))
        conn.commit()
        logging.info(f"Template {template_id} last_used timestamp updated.")
    except Exception as e:
        logging.error(f"Failed to update template {template_id} last_used timestamp: {e}")
    finally:
        if conn:
            db_pool.putconn(conn)

# Function to generate and save emails
def generate_and_save_emails(search_query, template_id, num_emails):
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            for _ in range(num_emails):
                email = fetch_recent_lead(search_query)
                ai_content = generate_ai_content(search_query, email)
                full_content = fetch_template(template_id)
                full_content = full_content.replace("{{content}}", ai_content)
                cursor.execute('INSERT INTO generated_emails (search_query, email, content, status, email_sent) VALUES (%s, %s, %s, %s, %s)',
                               (search_query, email, full_content, 'draft', 0))
        conn.commit()
        logging.info(f"Generated and saved {num_emails} emails successfully.")
    except Exception as e:
        logging.error(f"Failed to generate and save emails: {e}")
    finally:
        if conn:
            db_pool.putconn(conn)

# Function to fetch the most recent lead for a search query
def fetch_recent_lead(search_query):
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            cursor.execute('SELECT email FROM leads WHERE search_query=%s ORDER BY saved_at DESC LIMIT 1', (search_query,))
            email = cursor.fetchone()[0]
        return email
    except Exception as e:
        logging.error(f"Failed to fetch recent lead for {search_query}: {e}")
        return ""
    finally:
        if conn:
            db_pool.putconn(conn)

# Function to update the status of generated emails
def update_email_status(email_id, status):
    conn = None
    try:
        conn = db_pool.getconn()
        with conn.cursor() as cursor:
            cursor.execute('UPDATE generated_emails SET status=%s WHERE email_id=%s', (status, email_id))
        conn.commit()
        logging.info(f"Email {email_id} status updated to {status}.")
    except Exception as e:
        logging.error(f"Failed to update email {email_id} status: {e}")
    finally:
        if conn:
            db_pool.putconn(conn)

# Function to bulk send emails via AWS SES
def bulk_send_emails_aws(from_address, reply_to):
    unsent_emails = fetch_unsent_emails()
    for email in unsent_emails:
        send_email_via_aws(email[2], email[3], email[4], from_address, reply_to)
        update_email_status(email[0], 'sent')
    return "All emails sent successfully."

# Function to bulk send emails via custom API
def bulk_send_emails_custom_api(from_address, reply_to):
    unsent_emails = fetch_unsent_emails()
    for email in unsent_emails:
        send_email_via_custom_api(email[2], email[3], email[4], from_address, reply_to)
        update_email_status_custom_db(email[0], 'sent')
    return "All emails sent successfully."

# Function to send emails via custom API
def send_email_via_custom_api(to_address, subject, body_html, from_address, reply_to):
    # Implement your custom email sending logic here
    # ...
    return "Email sent successfully via custom API."

# Function to update email status in a custom database
def update_email_status_custom_db(email_id, status):
    # Implement your custom database update logic here
    # ...
    return "Email status updated successfully in custom database."

# Gradio interface
with gr.Blocks() as gradio_app:
    gr.Markdown("# Email Campaign Management System")

    with gr.Tab("Search Emails"):
        search_query = gr.Textbox(label="Search Query", placeholder="e.g., 'Potential Customers in Madrid'")
        num_results = gr.Slider(1, 100, value=10, step=1, label="Number of Results")
        search_button = gr.Button("Search")
        results = gr.Dataframe(headers=["Search Query", "Email", "URL"])

        search_button.click(lambda query, num_results: scrape_emails(query, num_results),
                            inputs=[search_query, num_results], outputs=[results])

    with gr.Tab("Create Email Template"):
        template_content = gr.Textarea(placeholder="Email template content", default="Hi {{name}},\n\nThis is an email template with a placeholder for the name.\n\nRegards,\nYour Company", lines=10)
        save_template_button = gr.Button("Save Template")
        template_id = gr.Label("")

        save_template_button.click(lambda content: save_template(content), inputs=[template_content], outputs=[template_id])

    with gr.Tab("Generate and Send Emails"):
        search_query = gr.Textbox(label="Search Query", placeholder="e.g., 'Potential Customers in Madrid'")
        template_id = gr.Dropdown(label="Select Template", choices=["Select a template"])
        num_emails = gr.Slider(1, 1000, value=10, step=1, label="Number of Emails")
        generate_button = gr.Button("Generate and Save")
        send_button = gr.Button("Send Emails via AWS SES")
        custom_send_button = gr.Button("Send Emails via Custom API")

        generate_button.click(lambda query, template, num: generate_and_save_emails(query, template, num),
                               inputs=[search_query, template_id, num_emails])

        send_button.click(lambda from_address, reply_to: track_progress_and_send(from_address, reply_to),
                           inputs=[gr.InputBox(label="From Address", type="email"), gr.InputBox(label="Reply-To Address", type="email")])

        custom_send_button.click(lambda from_address, reply_to: bulk_send_emails_custom_api(from_address, reply_to),
                                inputs=[gr.InputBox(label="From Address", type="email"), gr.InputBox(label="Reply-To Address", type="email")])

    with gr.Tab("Manage Search Queries"):
        query_status = gr.Dropdown(label="Select Query Status", choices=["Select a status"])
        process_button = gr.Button("Process Queries")
        processed_queries = gr.Label("")

        process_button.click(lambda status: process_queries(status=status), inputs=[query_status], outputs=[processed_queries])

gradio_app.launch()