app.py

from googleapiclient.discovery import build
from langchain.tools import BaseTool
from pydantic import Extra
from langchain.utilities import GoogleSearchAPIWrapper
from langchain.agents import initialize_agent
from langchain.agents import ZeroShotAgent, Tool, AgentExecutor
from langchain import OpenAI, LLMChain, LLMMathChain
import pandas as pd
import csv
import json
import re, os
import requests
from typing import Tuple, List
from langchain.llms import OpenAI
from langchain.agents.agent_toolkits import ZapierToolkit
from langchain.utilities.zapier import ZapierNLAWrapper
import base64
from google.oauth2 import credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
from email.mime.text import MIMEText
import gradio as gr
from email.mime.multipart import MIMEMultipart

class YoutubeChannelStatistics(BaseTool):

    class Config(BaseTool.Config):
        extra = Extra.allow

    def __init__(self, api_key: str):
        super().__init__(
            name="youtube channel assistance",
            description="useful when you need to get everything about a youtube channel, including it's email address, video titles, data, view counts, average views, comment rate, like rate, suggested commission, and you can know this channel's topics from the titles of the videos. the input should be the name of the channel.",
            return_direct=True,
            verbose=False
        )
        self.api_key = api_key
        self.youtube = build("youtube", "v3", developerKey=self.api_key)

    def scrape_video_data(self, video_id: str, domain_file_name: str, data1_file_name: str) -> Tuple[List[str], int]:
        response = requests.get(
            f"https://www.googleapis.com/youtube/v3/videos?id={video_id}&key={self.api_key}&part=snippet,statistics")

        if response.status_code == 200:
            video_data = json.loads(response.text)
            channel_name = video_data["items"][0]["snippet"]["channelTitle"]
            title_name = video_data["items"][0]["snippet"]["title"]
            description = video_data["items"][0]["snippet"]["description"]
            view_count = video_data["items"][0]["statistics"]["viewCount"]
            video_link = f"https://www.youtube.com/watch?v={video_id}"

        # Extract email addresses from the video description
            email_list = []
            excluded_count = 0
            email_regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
            matches = re.findall(email_regex, description)
            excluded_words = []
            with open(domain_file_name, mode="r", encoding="utf-8") as domain_file:
                reader = csv.reader(domain_file)
                for row in reader:
                    excluded_words.append(row[0])
            domain_exclude_regex = r'\b(?:{})\b'.format('|'.join(excluded_words))
            for match in matches:
                if re.search(domain_exclude_regex, match, re.IGNORECASE):
                    excluded_count += 1
                else:
                    email_list.append(match)

        # Check if an email address is present in data1.csv
            data1_df = pd.read_csv(data1_file_name, usecols=["Email Address"])
            already_contacted = set(data1_df["Email Address"].tolist())

        # Filter email addresses containing characters from domain.csv
            email_list = [email for email in email_list if not re.search(domain_exclude_regex, email, re.IGNORECASE)]

        # Append a notice to email addresses found in data1.csv
            email_list = [email + " (already contacted)" if email in already_contacted else email for email in email_list]

            return email_list, excluded_count         

    def _run(self, tool_input: str) -> str:
        channel_name = tool_input
        max_videos = 5 
        # The rest of the _run method implementation remains the same

        channel_response = self.youtube.search().list(
            q=channel_name, type="channel", part="id", maxResults=1
        ).execute()
        channel_id = channel_response["items"][0]["id"]["channelId"]

        # Get videos in the channel
        video_response = self.youtube.search().list(
            channelId=channel_id, type="video", part="id,snippet", maxResults=max_videos, order="date"
        ).execute()

        video_ids = [video["id"]["videoId"] for video in video_response["items"]]
        video_titles = {video["id"]["videoId"]: video["snippet"]["title"] for video in video_response["items"]}

        # Get video statistics
        video_stats_response = self.youtube.videos().list(
            id=",".join(video_ids), part="statistics"
        ).execute()

        total_views = 0
        total_comments = 0
        total_likes = 0
        video_count = 0

        for item in video_stats_response["items"]:
            video_count += 1
            stats = item["statistics"]

            total_views += int(stats["viewCount"])
            total_comments += int(stats.get("commentCount", 0))
            total_likes += int(stats["likeCount"])

        if video_count == 0:
            return f"{channel_name} avg_views: 0, comment_rate: 0, like_rate: 0"

        avg_views = total_views / video_count
        comment_rate = total_comments / total_views
        like_rate = total_likes / total_views

        # Calculate commission value
        if comment_rate > 0.02:
            commission = avg_views * 0.1
        else:
            commission = avg_views * 0.09

        # Convert comment_rate and like_rate to percentages
        comment_rate_percent = comment_rate * 100
        like_rate_percent = like_rate * 100

        # Round the custom value
        custom_rounded = round(commission)

        result = f"{channel_name}: average views:{avg_views:.2f}, comments vs views:{comment_rate_percent:.2f}%, likes vs views:{like_rate_percent:.2f}%, suggested commission:{custom_rounded:.2f}\n\nVideo titles and view counts:\n"

        for item in video_stats_response["items"]:
            video_id = item["id"]
            view_count = int(item["statistics"]["viewCount"])
            title = video_titles[video_id]
            result += f"- {title} - {view_count} views\n"

        # Scrape video data for each video ID
        email_list = []
        excluded_count = 0
        domain_file_name = "domain.csv"
        data1_file_name = "data1.csv"
        found_email = None  # Store the first found email address

        videos_to_check = 1  # Number of videos to check in each iteration

        for i in range(0, len(video_ids), videos_to_check):
            # Check only a certain number of videos at a time
            current_video_ids = video_ids[i:i + videos_to_check]

            for video_id in current_video_ids:
                video_emails, excluded = self.scrape_video_data(video_id, domain_file_name, data1_file_name)
                email_list.extend(video_emails)
                excluded_count += excluded

                # Store the first found email address
                if not found_email and video_emails:
                    found_email = video_emails[0]

            # If an email address was found, stop checking
            if found_email:
                break

        # Add email list to the result string
        result += f"\nEmail addresses in the descriptions:\n"
        for email in email_list:
            result += f"- {email}\n"

        if found_email:
            result += f"\nThis channel's email address is: {found_email}\n"
        else:
            result += "No email address was found in this channel.\n"

        return result

    async def _arun(self, tool_input: str) -> str:
        return self._run(tool_input) 

class GmailDraftCreator:
    def __init__(self):
        self.creds = self.get_credentials()
        self.service = build('gmail', 'v1', credentials=self.creds)

    def get_credentials(self):
        creds = None
        if os.path.exists('token.json'):
            creds = credentials.Credentials.from_authorized_user_file('token.json')

        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', ['https://www.googleapis.com/auth/gmail.compose'])
                creds = flow.run_local_server(port=0)
            with open('token.json', 'w') as token:
                token.write(creds.to_json())

        return creds

    def create_draft(self, to, subject, body):
        message = self.create_message(to, subject, body)
        draft = {
            'message': {
                'raw': base64.urlsafe_b64encode(message.as_bytes()).decode('utf-8')
            }
        }
        draft = self.service.users().drafts().create(userId='me', body=draft).execute()
        return draft

    @staticmethod
    def create_message(to, subject, body):
        message = MIMEMultipart()
        text = MIMEText(body)
        message.attach(text)
        message['to'] = to
        message['subject'] = subject
        return message

    def run(self, input_string):
        input_string = input_string.replace("Action Input: ", "").strip()
        input_list = input_string.strip().split("', '")
        if len(input_list) != 3:
            print("Invalid input format.")
            return None

        email_address, subject, body = [s.strip("'") for s in input_list]
        try:
            self.create_draft(email_address, subject, body)
            print("Gmail draft created successfully!")
        except HttpError as error:
            print(f"An error occurred: {error}")
            return None


api_key = 'AIzaSyCl5OiDNHfHLY0cEIL0QlSeVzPDkxEukKE'  # Replace with your actual API key  AIzaSyD4MtRHpxBL2SAk2eA9RIdpbYGlUlUZYd4
llm=OpenAI(temperature=0)    

yt_search_tool = YoutubeChannelStatistics(api_key)
search = GoogleSearchAPIWrapper(google_api_key=api_key, google_cse_id = '159edbaf5e68e4d9e')

gmail_draft_creator = GmailDraftCreator()

tools = [
    Tool(
        name="youtube channel assistance",
        func=yt_search_tool.run,
        description="useful when you need to get everything about a youtube channel, including it's email address, video titles, data, view counts, average views, comment rate, like rate, suggested commission, and you can know this channel's topics from the titles of the videos. the input should be the name of the channel.",
        return_direct=False
    ),
    Tool(
        name="send emails",
        func= gmail_draft_creator.run,
        description="useful when you need to send emails to a youtube channel, the input format should be: 'email address', 'subject', 'body'. for example: 'xx@mail.com','nice to meet you','hello, how are you?'",
        return_direct=False
    ),
]

tool_names = [tool.name for tool in tools]

#改动promt会影响input string for email generate
prefix = """You're a social channel analyzer, answer the following questions based on the following info. You have access to the following tools:"""
suffix = """Begin! remember to use the default tool "youtube channel" first to get average views, comment rate, like rate of this youtube channel. and then answer the following question:

Question: {input} 
{agent_scratchpad}"""

prompt = ZeroShotAgent.create_prompt(
    tools, 
    prefix=prefix, 
    suffix=suffix, 
    input_variables=["input", "agent_scratchpad"]
)


llm_chain = LLMChain(llm=llm, prompt=prompt)

agent = ZeroShotAgent(llm_chain = llm_chain, allowed_tools = tool_names, max_iterations=2 )

agent_executor = AgentExecutor.from_agent_and_tools(agent=agent, tools=tools, verbose=True)

# With this function:
def agent_executor_function(input_text: str) -> str:
    result = agent_executor.run(input_text)
    return result

# Create a Gradio interface
iface = gr.Interface(
    fn=agent_executor_function,
    inputs=gr.inputs.Textbox(lines=3, placeholder="Enter your command here..."),
    outputs=gr.outputs.Textbox(),
    title="Social Channel Analyzer",
    description="Analyze YouTube channels and send email invitations.",
)

# Launch the app on Hugging Face Spaces
iface.launch()