| import os |
| import requests |
| import json |
| from urllib.parse import urlparse |
| import asyncio |
| import edge_tts |
| from dotenv import load_dotenv |
|
|
| |
| load_dotenv() |
|
|
| def extract_asin_from_amazon_url(url): |
| """Extracts the ASIN from an Amazon product URL.""" |
| parsed_url = urlparse(url) |
| path_parts = parsed_url.path.split('/') |
| for i, part in enumerate(path_parts): |
| if part == 'dp' and i < len(path_parts) - 1: |
| return path_parts[i + 1] |
| return None |
|
|
| def get_product_details(asin, country="US"): |
| """Fetches product details using the RapidAPI.""" |
| url = os.getenv('RAPIDAPI_URL') |
| querystring = {"asin": asin, "country": country} |
| headers = { |
| "x-rapidapi-key": os.getenv('RAPIDAPI_KEY'), |
| "x-rapidapi-host": os.getenv('RAPIDAPI_HOST') |
| } |
| response = requests.get(url, headers=headers, params=querystring) |
| return response.json() |
|
|
| def save_product_data(product_data, file_path='data.json'): |
| """Saves extracted product data to a JSON file.""" |
| extracted_data = { |
| "title": product_data["data"]["product_title"], |
| "description": product_data["data"]["product_description"], |
| "price": product_data["data"]["product_price"], |
| "images": product_data["data"]["product_photos"] |
| } |
| with open(file_path, 'w') as json_file: |
| json.dump(extracted_data, json_file, indent=4) |
| print(f"Extracted product data saved to {file_path}") |
|
|
| def load_data_from_json(file_path='data.json'): |
| """Loads product data from the JSON file.""" |
| try: |
| with open(file_path, "r") as file: |
| data = json.load(file) |
| title = data.get("title", "") |
| description = data.get("description", title) |
| return title, description |
| except FileNotFoundError: |
| print(f"{file_path} file not found. Please make sure it exists in the same directory.") |
| return "", "" |
| except json.JSONDecodeError: |
| print(f"Error decoding {file_path}. Please make sure it's a valid JSON file.") |
| return "", "" |
|
|
| def generate_ad_script(title, description, language): |
| """Generates an ad script using the new API.""" |
| url = "https://picfy.xyz/api/amz.php" |
| prompt = f"""Create a short advertisement script for: |
| Title: {title} |
| Description: {description} |
| Language: {language} |
| Provide a 2 to 3 line ad script in the requested language.""" |
|
|
| try: |
| response = requests.post( |
| url, |
| headers={'Content-Type': 'application/json'}, |
| json={'input': prompt}, |
| timeout=30 |
| ) |
| response.raise_for_status() |
| |
| result = response.json() |
| ad_script = result.get('message', '').strip() |
| |
| if not ad_script: |
| return {"ad_script": "Error: No ad script generated"} |
| |
| return {"ad_script": ad_script} |
| |
| except requests.RequestException as e: |
| print(f"API request error: {e}") |
| return {"ad_script": "Error generating ad script"} |
|
|
| def save_ad_script(ad_script, file_path='ad_script.txt'): |
| """Saves the generated ad script to a text file.""" |
| with open(file_path, "w", encoding="utf-8") as file: |
| file.write(ad_script) |
| print(f"Ad script saved to {file_path}") |
|
|
| async def list_voices(): |
| """Lists available voices for text-to-speech.""" |
| voices = await edge_tts.list_voices() |
| return voices |
|
|
| async def text_to_speech(voice, text, output_file): |
| """Converts text to speech using the selected voice.""" |
| communicate = edge_tts.Communicate(text, voice) |
| await communicate.save(output_file) |
