social_media_crew.py

from crewai import Agent, Task, Crew, Process, LLM
from image_generator_tool import GenerateImageTool
from utils_tools import CalculateDiscountedPriceTool, CalculateDiscountValueTool, MerchantSelectorTool
import requests
from advanced_scrape_tool import AdvancedScrapingTool

class SocialMediaCrew:
    def __init__(self, openai_api_key: str, natura_api_token: str, openai_base_url: str, openai_model_name: str):
        self.openai_api_key = openai_api_key
        self.natura_api_token = natura_api_token
        self.openai_base_url = openai_base_url
        self.openai_model_name = openai_model_name
        self.scrape_tool = AdvancedScrapingTool()
        self.calculate_discounted_price_tool = CalculateDiscountedPriceTool()
        self.calculate_discount_value_tool = CalculateDiscountValueTool()
        self.image_generator_tool = GenerateImageTool()
        self.merchant_selector_tool = MerchantSelectorTool(natura_api_token=self.natura_api_token)

        print("Initializing SocialMediaCrew with BASE URL:", self.openai_base_url)
        print("Using OpenAI Model:", self.openai_model_name)
        print("Using OpenAI Key:", self.openai_api_key[:10])

        llm = LLM(
            api_key=self.openai_api_key, 
            model=self.openai_model_name, 
            base_url=self.openai_base_url
        )

        self.product_analyst = Agent(
            role='Product Analyst',
            goal='Analyze the provided URL and extract key product information',
            backstory=("You are an expert in analyzing product pages and extracting the most important information. You can identify the product name, its main features, and the target audience."),
            verbose=True,
            tools=[self.scrape_tool, 
                   self.calculate_discounted_price_tool, 
                   self.calculate_discount_value_tool],
            allow_delegation=False,
            llm=llm,
            max_retries=3
        )

        self.social_media_copywriter = Agent(
            role='Social Media Copywriter',
            goal='Create a compelling social media post in Portuguese to sell the product',
            backstory=("You are a creative copywriter specialized in the beauty and fragrance market. You know how to craft posts that are engaging, persuasive, and tailored for a Portuguese-speaking audience. You are an expert in using emojis and hashtags to increase engagement."),
            verbose=True,
            tools=[self.image_generator_tool],
            allow_delegation=False,
            llm=llm,
            max_retries=3
        )

    def _validate_url(self, product_url: str) -> bool:
        headers = {
            "accept": "*/*",
            "accept-language": "pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7",
            "sec-ch-ua": '"Not)A;Brand";v="8", "Chromium";v="138", "Google Chrome";v="138"',
            "sec-ch-ua-mobile": "?0",
            "sec-ch-ua-platform": '"Windows"',
            "sec-fetch-dest": "empty",
            "sec-fetch-mode": "cors",
            "sec-fetch-site": "cross-site",
            "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36",
        }
        try:
            response = requests.get(product_url, headers=headers)
            response.raise_for_status()
            if '<template data-dgst="NEXT_NOT_FOUND">' in response.text:
                return False
            return True
        except requests.exceptions.RequestException as e:
            print(f"Error checking URL: {e}")
            return False

    def _prepare_merchant(self, product_url: str):
        merchant = self.merchant_selector_tool.run(product_url)
        css_selector = merchant.keep_css_selectors()
        short_url = merchant.shorten_url(product_url)
        return merchant, css_selector, short_url

    def _create_analyze_product_task(self, product_url: str, css_selector: list[str], main_cupom_discount_percentage: float, short_url: str, original_price: float, discounted_price: float) -> Task:
        task_description = (f"1. Scrape the content of the URL: {product_url} using the 'scrape_tool' with param `keep_selectors` as `{css_selector}` and `base_selector` as `body`.\n"
                            "2. Extract the product name, key characteristics, and any other relevant DISCOUNT available.\n")

        if original_price is not None and original_price > 0 and discounted_price is not None and discounted_price > 0:
            task_description += (f"3. The user has provided the prices. Use ORIGINAL PRICE = {original_price} and DISCOUNTED PRICE = {discounted_price}.\n")
            final_best_price_source = str(discounted_price)
        else:
            task_description += ("3. Identify and extract the original product price and the final discounted price if existing from the scraped content. "
                                 "IGNORE any price breakdowns like 'produto' or 'consultoria'.\n")
            final_best_price_source = "the extracted final best price"

        task_description += (f"4. Use the 'Calculate Discounted Price Tool' with {final_best_price_source} and the provided DISCOUNT PERCENTAGE ({main_cupom_discount_percentage}) to get the CUPOM DISCOUNTED PRICE.\n"
                             "4.1 Use the 'Calculate Discount Value Tool' with ORIGINAL PRICE and CUPOM DISCOUNTED PRICE to get the TOTAL DISCOUNT PERCENTAGE.\n" 
                             f"5. Provide all this information, including the product name, ORIGINAL PRICE, DISCOUNTED PRICE (the one from step 3), CUPOM DISCOUNTED PRICE, and the generated short URL ({short_url}). If any of this information cannot be extracted, you MUST return 'MISSING_PRODUCT_INFO'.")

        return Task(
            description=task_description,
            agent=self.product_analyst,
            expected_output="A concise summary of the product including its name, key features, unique selling points, ORIGINAL PRICE, DISCOUNTED PRICE (the one used as the input in the tool 'Calculate Discounted Price Tool'), CUPOM DISCOUNTED PRICE, TOTAL DISCOUNT PERCENTAGE, and the SHORT SHAREABLE URL ({short_url}), OR 'MISSING_PRODUCT_INFO' if essential product details are not found."
        )

    def _create_post_task(self, analyze_product_task: Task, merchant, main_cupom: str, cupom_1: str, store_name: str) -> Task:
        template = merchant.get_template(main_cupom, cupom_1, store=store_name)
        return Task(
            description=(f"Based on the product analysis, create a CONCISE and DIRECT social media post in Portuguese, using Brazilian currency format, add one space after R$, like R$ 99,99, suitable for a WhatsApp group. \n If the input you receive is 'INVALID_URL' or 'MISSING_PRODUCT_INFO', you MUST stop and output only that same message.\n The post should strictly follow this template:\n {template}\n\n. Do not add backticks to response. Ensure a URL is always present in the output. Include a clear and short call to action and a MAXIMUM of 1 relevant emoji. DO NOT include hashtags. Keep it short and impactful and does not forget to include the backticks around the last paragraph.\n\n If the input you receive is 'INVALID_URL', you MUST stop and output only 'INVALID_URL'."),
            agent=self.social_media_copywriter,
            expected_output="A short, direct, and impactful social media post in Portuguese for WhatsApp, strictly following the provided template, dont include triple backticks, including the FINAL PRICE, any DISCOUNT with no decimal cases, the SHORT SHAREABLE URL, a call to action, and one emoji in the Title or in the Description. Description should be one sentence only. Description should stay before the title, as the template. No hashtags should be present. A URL must always be present in the final output, OR the message 'INVALID_URL' or 'MISSING_PRODUCT_INFO' if the page was not found or product info is missing.",
            context=[analyze_product_task]
        )

    def run_crew(self, product_url: str, store_name: str, main_cupom: str, main_cupom_discount_percentage: float, cupom_1: str, original_price: float, discounted_price: float) -> str:
        if not self._validate_url(product_url):
            return "INVALID_URL"

        merchant, css_selector, short_url = self._prepare_merchant(product_url)
        
        analyze_product_task = self._create_analyze_product_task(product_url, css_selector, main_cupom_discount_percentage, short_url, original_price, discounted_price)
        create_post_task = self._create_post_task(analyze_product_task, merchant, main_cupom, cupom_1, store_name)

        crew = Crew(
            agents=[self.product_analyst, self.social_media_copywriter],
            tasks=[analyze_product_task, create_post_task],
            process=Process.sequential
        )

        print(f"Crew is kicking off for URL: {product_url}")
        result = crew.kickoff()
        return result