# This files contains your custom actions which can be used to run # custom Python code. # # See this guide on how to implement these action: # https://rasa.com/docs/rasa/custom-actions #This is a simple example for a custom action which utters "Hello World!" from typing import Any, Text, Dict, List from rasa_sdk import Action from rasa_sdk.events import SlotSet,UserUttered,FollowupAction from rasa_sdk.executor import CollectingDispatcher from rasa_sdk.interfaces import Tracker import re import requests from bs4 import BeautifulSoup from urllib.parse import quote import time ########################## class Action_lists_emplois(Action): def name(self) -> Text: return "action_lists_emplois" def run(self, dispatcher: CollectingDispatcher, tracker: "Tracker", domain: Dict[Text, Any]) -> List[Dict[Text, Any]]: url = "https://www.mytek.tn/nos-offres-d-emploi" dispatcher.utter_message(text=f"Pour gérer la liste des emplois, visitez {url}") return [] ########################## class Action_get_order_status(Action): def name(self) -> Text: return "action_get_order_status" def run(self, dispatcher: CollectingDispatcher, tracker, domain: Dict[Text, Any]): order_id = tracker.get_slot('order_id') # Check if order_id is not None if order_id is not None: # Define the API endpoint url = f"https://192.168.100.172/magento2/pub/rest/all/V1/orders/{order_id}/statuses" # Headers for the request headers = { 'accept': 'application/json', 'Authorization': 'Bearer 3ehuu099rchpyqhk6eyookwcn0cixwwk' } # Send a request to the API response = requests.get(url, headers=headers, verify=False) # verify=False for self-signed certificates # Check if the request was successful if response.status_code == 200: # Assuming the response contains a JSON with the order status order_status = response.json() dispatcher.utter_message(text=f"Order status for {order_id}: {order_status}") else: # Handle unexpected status codes dispatcher.utter_message(text=f"Failed to retrieve order status for {order_id}. Please try again.") else: dispatcher.utter_message(text="No order ID provided.") return [] ########################## class Action_get_product_info(Action): def name(self) -> Text: return "action_get_product_info" def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]: product_url = tracker.get_slot("product_url") try: product_page = requests.get(product_url) soup = BeautifulSoup(product_page.content, "html.parser") # Find the specific table with 'FICHE TECHNIQUE' as caption fiche_technique_table = soup.find("table", id="product-attribute-specs-table") product_info = {} # Extract each row in the table if fiche_technique_table: rows = fiche_technique_table.find_all("tr") for row in rows: columns = row.find_all("th") values = row.find_all("td") if columns and values: label = columns[0].text.strip() value = values[0].text.strip() product_info[label] = value markdown_table = "" for label, value in product_info.items(): markdown_table += f" {label} : {value} \n" dispatcher.utter_message(text=f"Product Information: \n{markdown_table}") return [] except Exception as e: dispatcher.utter_message(text="Failed to fetch product information.") return [] ########################## class Action_get_product(Action): def name(self) -> Text: return "action_get_product" def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]: product = tracker.get_slot("product") if not product: dispatcher.utter_message(text="Please provide the name of the product.") return [] product = product.upper() # Convert product name to uppercase encoded_product = quote(product) # URL-safe encoding url = f"https://www.mytek.tn/search/ajax/suggest/?q={encoded_product}" try: response = requests.get(url) search_results = response.json() product_results = [result for result in search_results if result['type'] == 'product'] if product_results: buttons = [] for product_data in product_results[:3]: # Process up to the first three products title = product_data.get('title', 'No title available') price_html = product_data.get('price', 'Price not available') price_pattern = r'\d{1,3}(?:\u202f\d{3})*,\d{2}' price_match = re.search(price_pattern, price_html) price = price_match.group(0).replace('\u202f', '').replace(',', '.') + " DT" if price_match else "Price not available" image = product_data.get('image', 'No image available') url = product_data.get('url', 'No URL available') # Create message with markdown including an image if available message_text = f"**{title}**\nPrice: {price}\n" if image != 'No image available': message_text += f"![Product Image]({image})\n" message_text += f"[View Product]({url})" # Send each product's details dispatcher.utter_message(text=message_text) # Accumulate buttons for all products buttons.append({"title": "Get Info", "payload": f"/ask_product_info{{\"product_url\": \"{url}\"}}"}) # Send all buttons after all product details if buttons: dispatcher.utter_message(text="Get more information:", buttons=buttons) return [SlotSet("product_url", url)] else: dispatcher.utter_message(text="Aucun produit correspondant trouvé. Veuillez vérifier le nom et réessayer.") except Exception as e: dispatcher.utter_message(text=f"Error retrieving product information: {str(e)}") return [] ########################## class Action_reset_product(Action): def name(self) -> Text: return "action_reset_product" def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]: return [SlotSet("product", None)] ########################## class Action_default_fallback(Action): def name(self)-> Text: return "action_default_fallback" def run(self,dispatcher:CollectingDispatcher, tracker:Tracker, domain:Dict[Text,Any]) -> List[Dict[Text,Any]]: dispatcher.utter_message(text="Samahni mafhemteksh !") return [] ########################## class ActionListsPromos(Action): def name(self) -> Text: return "action_lists_promos" def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]: main_page_url = "https://www.mytek.tn" try: response = requests.get(main_page_url) soup = BeautifulSoup(response.content, "html.parser") promo_links = soup.find_all("div", class_=["img-right col-lg-6 col-sm-12","img-left col-lg-6 col-sm-12" ,"pad15"]) carousel_items = [] for link in promo_links: a_tag = link.find("a", class_="mkcache") if not a_tag: continue href = a_tag.get('href') # URL of the product page img_tag = a_tag.find("img") if not img_tag: continue img_src = img_tag.get('src') # Image source URL raw_name = img_tag.get('alt') # Product name from the alt attribute if raw_name: cleaned_name = raw_name.replace('-', ' ').upper() # Construct carousel item item = { "title": cleaned_name, "image_url": img_src, "default_action": { "type": "web_url", "url": href, "webview_height_ratio": "tall", }, "buttons": [ { "type": "web_url", "url": href, "title": "View Details" } ] } carousel_items.append(item) # Send carousel message to the user dispatcher.utter_message(attachment={ "type": "template", "payload": { "template_type": "generic", "elements": carousel_items } }) except Exception as e: dispatcher.utter_message(text="An error occurred while fetching promos.") print(e) # Print the exception for debugging purposes return [SlotSet("product_url", href)] ########################## class Action_check_order_id(Action): def name(self): return "action_check_order_id" def run(self, dispatcher, tracker, domain): order_id = tracker.get_slot('order_id') if order_id: # Order ID is already provided, so get the order status return [FollowupAction("action_get_order_status")] else: # Ask the user to provide the order ID dispatcher.utter_message(template="utter_get_order_id") return [] #####class ActionPreprocessUserMessage(Action): #def name(self): # return "action_preprocess_user_message" #def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: dict): # Get the last user message # original_message = tracker.latest_message.get('text') # Convert the message to uppercase # modified_message = original_message.upper() # Create a new UserUttered event with the modified message # new_event = UserUttered(text=modified_message, metadata={"is_converted": True}) # Return the new UserUttered event to update the tracker # return [new_event]