import os
import requests
import json
from dotenv import load_dotenv
import tweepy
import time
# --- Load environment variables ---
load_dotenv()
# --- X API Credentials (from .env file) ---
# Ensure these variables are set in your .env file
try:
X_API_KEY = os.getenv("X_API_KEY")
X_API_SECRET = os.getenv("X_API_SECRET")
X_ACCESS_TOKEN = os.getenv("X_ACCESS_TOKEN")
X_ACCESS_TOKEN_SECRET = os.getenv("X_ACCESS_TOKEN_SECRET")
if not all([X_API_KEY, X_API_SECRET, X_ACCESS_TOKEN, X_ACCESS_TOKEN_SECRET]):
raise ValueError("One or more X API credentials are missing from the .env file.")
except ValueError as e:
print(f"Configuration Error: {e}")
print("Please ensure your .env file is correctly set up with all X API keys and tokens.")
exit()
# --- Your X Profile Link (for tweet composition) ---
MY_X_PROFILE_LINK = "https://x.com/quaily_official" # This is your profile link, not a direct product link
# --- Function to download image ---
def download_image(image_url, filename="temp_image.jpg"):
"""Downloads an image from a URL and saves it to a file."""
try:
response = requests.get(image_url, stream=True)
response.raise_for_status()
with open(filename, 'wb') as file:
for chunk in response.iter_content(chunk_size=8192):
file.write(chunk)
return filename
except requests.exceptions.RequestException as e:
print(f"Error downloading image from {image_url}: {e}")
return None
# --- Main function to post to X ---
def post_to_x(products_data):
"""
Authenticates with X and posts products with images and links.
"""
# Authenticate with X
try:
auth = tweepy.OAuthHandler(X_API_KEY, X_API_SECRET)
auth.set_access_token(X_ACCESS_TOKEN, X_ACCESS_TOKEN_SECRET)
api = tweepy.API(auth)
api.verify_credentials()
print("X API Authentication successful!")
except tweepy.TweepyException as e:
print(f"Error during X API authentication: {e}")
print("Please double-check your API keys and tokens in the .env file.")
return
print(f"Attempting to post {len(products_data)} products to X...")
for i, product in enumerate(products_data):
print(f"\nProcessing product {i+1}/{len(products_data)}: {product.get('title', 'Untitled')}")
image_url = product.get('image_url')
amazon_link = product.get('amazon_link')
title = product.get('title', 'Check out this amazing product!')
if not image_url or not amazon_link:
print(f"Skipping product due to missing image URL or Amazon link: {title}")
continue
# 1. Download the image
image_filepath = download_image(image_url)
if not image_filepath:
print(f"Skipping product due to image download failure: {title}")
continue
media_id = None
try:
# 2. Upload image to X (media_upload)
print(f"Uploading image for {title}...")
# Ensure the client is v1 for media_upload, which Tweepy's API object uses by default.
# For Tweepy v4+, the API object is typically Client v1.
# If you were using Client v2, it would be client_v2.media_upload(...)
media = api.media_upload(image_filepath)
media_id = media.media_id_string
print(f"Image uploaded, media_id: {media_id}")
except tweepy.TweepyException as e:
print(f"Error uploading image to X for {title}: {e}")
# Clean up temporary image file even if upload fails
os.remove(image_filepath)
continue # Skip to next product
# 3. Compose the tweet text
# X has a 280 character limit for tweets. Links are shortened.
# Shorten title if necessary to make space for link and hashtags.
max_title_length = 200 # Adjust based on how much space you want for the link and hashtags
if len(title) > max_title_length:
title = title[:max_title_length].rsplit(' ', 1)[0] + '...' # Truncate at last space
tweet_text = f"{title}\n\nShop now: {amazon_link}\n\n#AmazonAffiliate #quaily"
# Optional: Add your X profile link if desired, though the product link is more direct.
# tweet_text = f"{title}\n\nShop now: {amazon_link}\n\nFind more on {MY_X_PROFILE_LINK}\n#AmazonAffiliate #quaily"
try:
# 4. Publish the tweet
print(f"Publishing tweet for {title}...")
api.update_status(status=tweet_text, media_ids=[media_id])
print(f"Successfully tweeted for: {title}")
except tweepy.TweepyException as e:
print(f"Error tweeting for {title}: {e}")
finally:
# Clean up the downloaded temporary image file
if os.path.exists(image_filepath):
os.remove(image_filepath)
print(f"Cleaned up temporary image file: {image_filepath}")
# Add a delay between posts to avoid hitting X's rate limits
# For 10 products, 5-10 seconds is usually safe.
# For more products, consider longer delays or smarter rate limit handling.
time_to_wait = 10 # seconds
print(f"Waiting for {time_to_wait} seconds before next post...")
time.sleep(time_to_wait)
if __name__ == "__main__":
# --- Load product data from JSON file ---
try:
with open('extracted_products.json', 'r', encoding='utf-8') as f:
products = json.load(f)
print(f"Loaded {len(products)} products from extracted_products.json.")
except FileNotFoundError:
print("Error: extracted_products.json not found. Please run the scraping script first.")
products = []
except json.JSONDecodeError:
print("Error: Could not decode extracted_products.json. Check file format.")
products = []
# Limit to 10 products for a test run, or adjust as needed
products_to_post = products[:10] # Posts the first 10 products
if products_to_post:
post_to_x(products_to_post)
print("\nProcess finished.")
else:
print("No products to post. Exiting.")