my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	"""Twitter/X Client - Wrapper for Twitter API v2 using Tweepy"""
#2
#3	import tweepy
#4	from typing import Optional, List
#5	from dataclasses import dataclass
#6	import logging
#7
#8	logger = logging.getLogger(__name__)
#9
#10
#11	@dataclass
#12	class TweetResult:
#13	"""Result from posting a tweet"""
#14	tweet_id: str
#15	text: str
#16	url: str
#17	success: bool
#18	error: Optional[str] = None
#19
#20
#21	class TwitterClient:
#22	"""Python client for Twitter API v2"""
#23
#24	def __init__(
#25	self,
#26	consumer_key: str,
#27	consumer_secret: str,
#28	access_token: str,
#29	access_token_secret: str,
#30	bearer_token: Optional[str] = None,
#31	):
#32	"""
#33	Initialize Twitter client with API credentials.
#34
#35	Args:
#36	consumer_key: Twitter API consumer key
#37	consumer_secret: Twitter API consumer secret
#38	access_token: Twitter access token
#39	access_token_secret: Twitter access token secret
#40	bearer_token: Optional bearer token for read-only operations
#41	"""
#42	self.consumer_key = consumer_key
#43	self.consumer_secret = consumer_secret
#44	self.access_token = access_token
#45	self.access_token_secret = access_token_secret
#46	self.bearer_token = bearer_token
#47
#48	# Initialize Tweepy client for API v2
#49	self.client = tweepy.Client(
#50	bearer_token=bearer_token,
#51	consumer_key=consumer_key,
#52	consumer_secret=consumer_secret,
#53	access_token=access_token,
#54	access_token_secret=access_token_secret,
#55	)
#56
#57	# Store authenticated user info
#58	self._user_info = None
#59
#60	logger.info("Twitter client initialized")
#61
#62	async def get_authenticated_user(self) -> dict:
#63	"""Get information about the authenticated user"""
#64	try:
#65	if not self._user_info:
#66	me = self.client.get_me()
#67	self._user_info = {
#68	"id": me.data.id,
#69	"username": me.data.username,
#70	"name": me.data.name,
#71	}
#72	return self._user_info
#73	except Exception as e:
#74	logger.error(f"Failed to get authenticated user: {e}")
#75	raise
#76
#77	async def post_tweet(
#78	self,
#79	text: str,
#80	reply_to_tweet_id: Optional[str] = None,
#81	quote_tweet_id: Optional[str] = None,
#82	) -> TweetResult:
#83	"""
#84	Post a tweet to Twitter/X.
#85
#86	Args:
#87	text: Tweet content (max 280 characters for standard accounts)
#88	reply_to_tweet_id: Optional tweet ID to reply to
#89	quote_tweet_id: Optional tweet ID to quote tweet
#90
#91	Returns:
#92	TweetResult with tweet details and URL
#93	"""
#94	try:
#95	# Validate text length
#96	if len(text) > 280:
#97	return TweetResult(
#98	tweet_id="",
#99	text=text,
#100	url="",
#101	success=False,
#102	error=f"Tweet too long: {len(text)} characters (max 280)"
#103	)
#104
#105	# Post the tweet
#106	kwargs = {"text": text}
#107	if reply_to_tweet_id:
#108	kwargs["in_reply_to_tweet_id"] = reply_to_tweet_id
#109	if quote_tweet_id:
#110	kwargs["quote_tweet_id"] = quote_tweet_id
#111
#112	response = self.client.create_tweet(**kwargs)
#113
#114	tweet_id = response.data["id"]
#115	user_info = await self.get_authenticated_user()
#116	username = user_info["username"]
#117	tweet_url = f"https://twitter.com/{username}/status/{tweet_id}"
#118
#119	logger.info(f"Successfully posted tweet: {tweet_url}")
#120
#121	return TweetResult(
#122	tweet_id=tweet_id,
#123	text=text,
#124	url=tweet_url,
#125	success=True,
#126	)
#127
#128	except tweepy.TweepyException as e:
#129	error_msg = str(e)
#130	logger.error(f"Failed to post tweet: {error_msg}")
#131	return TweetResult(
#132	tweet_id="",
#133	text=text,
#134	url="",
#135	success=False,
#136	error=error_msg,
#137	)
#138	except Exception as e:
#139	error_msg = f"Unexpected error: {str(e)}"
#140	logger.error(f"Failed to post tweet: {error_msg}")
#141	return TweetResult(
#142	tweet_id="",
#143	text=text,
#144	url="",
#145	success=False,
#146	error=error_msg,
#147	)
#148
#149	async def get_recent_tweets(
#150	self,
#151	username: Optional[str] = None,
#152	max_results: int = 10
#153	) -> List[dict]:
#154	"""
#155	Get recent tweets from a user.
#156
#157	Args:
#158	username: Username to fetch tweets from (defaults to authenticated user)
#159	max_results: Maximum number of tweets to return (default 10, max 100)
#160
#161	Returns:
#162	List of tweet dictionaries
#163	"""
#164	try:
#165	if username:
#166	# Get user by username
#167	user = self.client.get_user(username=username)
#168	user_id = user.data.id
#169	else:
#170	# Get authenticated user
#171	user_info = await self.get_authenticated_user()
#172	user_id = user_info["id"]
#173
#174	# Get user's tweets
#175	tweets = self.client.get_users_tweets(
#176	id=user_id,
#177	max_results=min(max_results, 100),
#178	tweet_fields=["created_at", "public_metrics", "entities"]
#179	)
#180
#181	if not tweets.data:
#182	return []
#183
#184	return [
#185	{
#186	"id": tweet.id,
#187	"text": tweet.text,
#188	"created_at": str(tweet.created_at),
#189	"metrics": tweet.public_metrics if hasattr(tweet, "public_metrics") else None,
#190	}
#191	for tweet in tweets.data
#192	]
#193
#194	except Exception as e:
#195	logger.error(f"Failed to get recent tweets: {e}")
#196	return []
#197
#198	async def search_tweets(
#199	self,
#200	query: str,
#201	max_results: int = 10
#202	) -> List[dict]:
#203	"""
#204	Search for tweets matching a query.
#205
#206	Args:
#207	query: Search query string
#208	max_results: Maximum number of results (default 10, max 100)
#209
#210	Returns:
#211	List of matching tweets
#212	"""
#213	try:
#214	tweets = self.client.search_recent_tweets(
#215	query=query,
#216	max_results=min(max_results, 100),
#217	tweet_fields=["created_at", "public_metrics", "author_id"]
#218	)
#219
#220	if not tweets.data:
#221	return []
#222
#223	return [
#224	{
#225	"id": tweet.id,
#226	"text": tweet.text,
#227	"created_at": str(tweet.created_at),
#228	"author_id": tweet.author_id,
#229	"metrics": tweet.public_metrics if hasattr(tweet, "public_metrics") else None,
#230	}
#231	for tweet in tweets.data
#232	]
#233
#234	except Exception as e:
#235	logger.error(f"Failed to search tweets: {e}")
#236	return []
#237
#238	async def close(self):
#239	"""Close the client connection"""
#240	# Tweepy client doesn't require explicit closing
#241	logger.info("Twitter client closed")
#242

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public