Публикация в Twitter Api сообщения с медиа-инициализацией ⇐ Python

[Anonymous]

Я пытаюсь скопировать твит с его медиафайлами и разместить его в своем аккаунте
все работает, пока мы не дошли до добавления медиачасти
Я исследовал и обнаружил, что мне нужно сначала загрузить все СМИ отправляют изображение в Твиттер, чтобы получить media_id, и его нужно загрузить через мою учетную запись, что означает, что я не могу получить media_id из твита, опубликованного другим пользователем. Мне нужно загрузить его снова самому, а затем добавить его сtwitter_text при твите ,
чтобы вы могли найти здесь, я попробовал оба API Twitter v1 и
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
) """

и то и другое из-за той же ошибки
error': '403 Forbidden\nYou are not permitted to perform this action.'

еще раз обратите внимание, что все работает, когда вы комментируете строку media_ids=media_ids, если 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}")



Подробнее здесь: https://stackoverflow.com/questions/793 ... media-init