90 lines
3.2 KiB
Python
90 lines
3.2 KiB
Python
import sys
|
|
import json
|
|
import asyncio
|
|
from twikit import Client
|
|
|
|
async def get_user_tweets(username, count=10):
|
|
client = Client('en-US')
|
|
|
|
try:
|
|
client.load_cookies('cookies.json')
|
|
except:
|
|
print(json.dumps({"error": "Not logged in. Run login script first."}))
|
|
return
|
|
|
|
try:
|
|
user = await client.get_user_by_screen_name(username)
|
|
user_id = user.id
|
|
|
|
tweets_data = []
|
|
tweets = await client.get_user_tweets(user_id, 'Tweets', count=count)
|
|
|
|
for tweet in tweets:
|
|
if tweet.text.startswith('RT @'):
|
|
continue
|
|
|
|
# Get full text - twikit truncates long tweets at 280 chars
|
|
# Full text is available on the tweet object under different attributes
|
|
full_text = None
|
|
|
|
# Try note_tweet first (X Premium long posts are stored as "note tweets")
|
|
if hasattr(tweet, 'note_tweet') and tweet.note_tweet:
|
|
note = tweet.note_tweet
|
|
if hasattr(note, 'text'):
|
|
full_text = note.text
|
|
elif isinstance(note, dict):
|
|
full_text = note.get('text') or note.get('note_tweet_results', {}).get('result', {}).get('text')
|
|
|
|
# Fall back to full_text attribute if available
|
|
if not full_text and hasattr(tweet, 'full_text') and tweet.full_text:
|
|
full_text = tweet.full_text
|
|
|
|
# Fall back to regular text
|
|
if not full_text:
|
|
full_text = tweet.text
|
|
|
|
# Strip trailing ellipsis if still truncated (shouldn't happen but just in case)
|
|
if full_text.endswith('…'):
|
|
full_text = full_text[:-1]
|
|
|
|
media = []
|
|
if tweet.media:
|
|
for m in tweet.media:
|
|
media_url = None
|
|
media_type = m.type
|
|
|
|
if hasattr(m, 'media_url'):
|
|
media_url = m.media_url
|
|
|
|
if media_type == 'video' and hasattr(m, 'streams') and m.streams:
|
|
best_stream = max(m.streams, key=lambda s: s.bitrate if hasattr(s, 'bitrate') and s.bitrate else 0)
|
|
if hasattr(best_stream, 'url'):
|
|
media_url = best_stream.url
|
|
|
|
if media_url:
|
|
media.append({
|
|
'type': media_type,
|
|
'url': media_url
|
|
})
|
|
|
|
tweets_data.append({
|
|
'id': tweet.id,
|
|
'text': full_text,
|
|
'media': media,
|
|
'created_at': tweet.created_at
|
|
})
|
|
|
|
print(json.dumps(tweets_data))
|
|
except Exception as e:
|
|
print(json.dumps({"error": str(e)}))
|
|
|
|
if __name__ == '__main__':
|
|
username = sys.argv[1] if len(sys.argv) > 1 else None
|
|
count = int(sys.argv[2]) if len(sys.argv) > 2 else 10
|
|
|
|
if not username:
|
|
print(json.dumps({"error": "Username required"}))
|
|
sys.exit(1)
|
|
|
|
asyncio.run(get_user_tweets(username, count))
|