mirror of
https://gitverse.ru/kpa39l/telepost.git
synced 2026-09-29 18:05:08 +00:00
126 lines
5.0 KiB
Python
126 lines
5.0 KiB
Python
import tweepy
|
|
from typing import Dict, Optional
|
|
from fastapi import HTTPException
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TwitterAPI:
|
|
def __init__(self, consumer_key: str, consumer_secret: str, access_token: str, access_token_secret: str):
|
|
"""
|
|
Инициализация Twitter API клиента
|
|
|
|
Args:
|
|
consumer_key: Ключ приложения Twitter
|
|
consumer_secret: Секрет приложения Twitter
|
|
access_token: Токен доступа
|
|
access_token_secret: Секрет токена доступа
|
|
"""
|
|
try:
|
|
# Создаем объект аутентификации
|
|
auth = tweepy.OAuthHandler(consumer_key, consumer_secret)
|
|
auth.set_access_token(access_token, access_token_secret)
|
|
|
|
# Создаем API клиент
|
|
self.api = tweepy.API(auth, wait_on_rate_limit=True)
|
|
|
|
# Проверяем валидность аутентификации
|
|
self.api.verify_credentials()
|
|
except Exception as e:
|
|
logger.error(f"Error initializing Twitter API: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to initialize Twitter API: {str(e)}")
|
|
|
|
def post_tweet(self, text: str) -> Dict:
|
|
"""
|
|
Опубликовать твит
|
|
|
|
Args:
|
|
text: Текст твита (до 280 символов)
|
|
|
|
Returns:
|
|
Информация о созданном твите
|
|
"""
|
|
try:
|
|
# Публикуем твит
|
|
tweet = self.api.update_status(text)
|
|
|
|
return {
|
|
"success": True,
|
|
"tweet_id": tweet.id,
|
|
"tweet_text": tweet.text,
|
|
"created_at": tweet.created_at.isoformat() if tweet.created_at else None
|
|
}
|
|
except tweepy.TweepyException as e:
|
|
logger.error(f"Error posting tweet: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to post tweet: {str(e)}")
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error posting tweet: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to post tweet: {str(e)}")
|
|
|
|
def post_thread(self, tweets: list) -> Dict:
|
|
"""
|
|
Опубликовать тред (цепочку твитов)
|
|
|
|
Args:
|
|
tweets: Список текстов твитов в порядке публикации
|
|
|
|
Returns:
|
|
Информация о созданном треде
|
|
"""
|
|
try:
|
|
previous_tweet_id = None
|
|
tweet_ids = []
|
|
|
|
for i, text in enumerate(tweets):
|
|
if previous_tweet_id:
|
|
# Отвечаем на предыдущий твит, чтобы создать тред
|
|
tweet = self.api.update_status(
|
|
status=text,
|
|
in_reply_to_status_id=previous_tweet_id,
|
|
auto_populate_reply_metadata=True
|
|
)
|
|
else:
|
|
# Первый твит в треде
|
|
tweet = self.api.update_status(text)
|
|
|
|
tweet_ids.append(tweet.id)
|
|
previous_tweet_id = tweet.id
|
|
|
|
return {
|
|
"success": True,
|
|
"tweet_ids": tweet_ids,
|
|
"thread_length": len(tweets)
|
|
}
|
|
except tweepy.TweepyException as e:
|
|
logger.error(f"Error posting thread: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to post thread: {str(e)}")
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error posting thread: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to post thread: {str(e)}")
|
|
|
|
def get_user_info(self) -> Dict:
|
|
"""
|
|
Получить информацию о текущем пользователе
|
|
|
|
Returns:
|
|
Информация о пользователе
|
|
"""
|
|
try:
|
|
user = self.api.verify_credentials()
|
|
|
|
return {
|
|
"id": user.id,
|
|
"screen_name": user.screen_name,
|
|
"name": user.name,
|
|
"followers_count": user.followers_count,
|
|
"friends_count": user.friends_count,
|
|
"statuses_count": user.statuses_count,
|
|
"description": user.description
|
|
}
|
|
except tweepy.TweepyException as e:
|
|
logger.error(f"Error getting user info: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to get user info: {str(e)}")
|
|
except Exception as e:
|
|
logger.error(f"Unexpected error getting user info: {str(e)}")
|
|
raise HTTPException(status_code=500, detail=f"Failed to get user info: {str(e)}") |