python - Posting to Twitter Api a post with media init - Stack Overflow

admin管理员组

文章数量:1124026

I am trying to copy a tweet with its media and tweet it on my account everything is working till we reach the adding the media part I researched and found that I have to first all load the media say image to Twitter to get the media_id and it has to be loaded through my account meaning I can't get the media_id from a tweet tweeted by another user I have to load it again myself, and then add this with the tweet_text when tweeting,

so u can find here I tried both twitter API v1 an

res = client.create_tweet(
            text=tweet_text
           # media_ids=media_ids if media_ids else None
        )
        print(res)
        """ res = client.create_tweet(
            text=tweet_text + "new",
            media_ids= media_ids if media_ids else None
        ) """

and both through the sam error

error': '403 Forbidden\nYou are not permitted to perform this action.'

note again everything works when u comment the line media_ids=media_ids if media_ids else None

from requests import Response
import tweepy
import re
import json
import time
from types import SimpleNamespace
import requests

def extract_tweet_id(tweet_link):
    match = re.search(r"status/(\d+)", tweet_link)
    if match:
        return match.group(1)
    return None
def clean_tweet_text(tweet_text):
    # Regular expression to match URLs
    url_pattern = r'https?://\S+'
    cleaned_text = re.sub(url_pattern, '', tweet_text).strip()
    return cleaned_text
def extract_media_id(media_key):
    match = re.search(r"_(\d+)", media_key)
    if match:
        return match.group(1)
    return None

def download_media(url, filename):
    print(f"download_media({url}, {filename})")
    response = requests.get(url)
    with open(filename, 'wb') as f:
        f.write(response.content)
    print(f"result enwfilename {filename}")
    return filename


def reupload_media(api, media_url):
    print(f"reupload_media({api}, {media_url}):")
    # Download media
    filename = download_media(media_url, "temp_media_file.png")
    # Upload to Twitter
    uploaded_media = api.media_upload(filename)
    print(f"result is new_media_id: {uploaded_media.media_id}")
    return uploaded_media.media_id

def fetch_tweet(client, api,  tweet_link):
    tweet_id = extract_tweet_id(tweet_link=tweet_link)
    if tweet_id:
        try:
            tweet = client.get_tweet(
            tweet_id,
            expansions=["author_id", "attachments.media_keys"],
            tweet_fields=["created_at", "text", "attachments"],
            media_fields=["url"]    
            )
            media_ids = []
            if "media" in tweet.includes:
                print('there exists some media')
                for media in tweet.includes["media"]:
                    media_url = media.url
                    print(f"Media URL: {media.url}")
                    media_id = reupload_media(api, media_url)
                    media_ids.append(media_id)
            return SimpleNamespace(tweet_text=  clean_tweet_text(tweet.data.text), media_ids= media_ids)
        except tweepy.TweepyException as e:
                if "Too Many Requests" in str(e):
                    print("Rate limit exceeded. sleep 15 min Waiting for reset...")
                    time.sleep(15 * 60)  # Wait 5 minutes
                else:
                    print(f"Error fetching tweet details: {e}")
    else:
        print("Invalid tweet link") 


def post_tweet(client, api, tweet_text, media_ids=None, hashtags=None):
    try:
        # Process hashtags
        if hashtags:
            tweet_text += " " + " ".join(f"{tag}" for tag in hashtags.split())
        
        res = client.create_tweet(
            text=tweet_text
           # media_ids=media_ids if media_ids else None
        )
        print(res)
        """ res = client.create_tweet(
            text=tweet_text + "new",
            media_ids= media_ids if media_ids else None
        ) """
        res = api.update_status(status=tweet_text, media_ids = media_ids if media_ids else None)
        print(res)

        return {"status": True, "response": res}
    except Exception as e:
        return {"status": False, "error": str(e)}




def lambda_handler(event, context):
    body = event.get("body", "{}")
    if isinstance(body, str):
        body = json.loads(body)  # Convert to dictionary if it's a JSON string
    
    # Dynamically handle payload structure
    if "body" in body and isinstance(body["body"], dict):
        # Handle the Python `requests.post` format
        payload = body["body"]
    else:
        # Handle the JavaScript `fetch` format
        payload = body
        
    tweet_link = payload.get("tweet_link", "")
    hashtags = payload.get("hashtags", "")
    api_key = payload.get("api_key", "")
    api_secret = payload.get("api_secret", "")
    access_token = payload.get("access_token", "")
    access_secret = payload.get("access_secret", "")
    bearer_key = payload.get("bearer_key", "")

    client = tweepy.Client(bearer_token = bearer_key,consumer_key =  api_key,consumer_secret =  api_secret,access_token =  access_token,access_token_secret =  access_secret,return_type=type[Response],wait_on_rate_limit=True)

    auth = tweepy.OAuth1UserHandler(
        consumer_key=api_key,
        consumer_secret=api_secret,
        access_token=access_token,
        access_token_secret=access_secret
    )      
    api = tweepy.API(auth)

    res = fetch_tweet(client, api, tweet_link)
    
    media_ids = res.media_ids
    tweet_text =  res.tweet_text
    print(media_ids)
    print(tweet_text)
    res = post_tweet(client=client, api=api, tweet_text=tweet_text,media_ids=media_ids,hashtags=hashtags)
    if (res['status']):
        print(f"successfully posted the tweet {res}")
    else:
        print(f"An error occurred posting a tweet: {res}")   
    

本文标签：