Anri/feurBot
Archived
1
0
Fork 0
Code Issues 2 Pull requests Projects Releases Wiki Activity

Don't answer on private accounts

Browse source
This commit is contained in:
Mylloon 2022-08-11 12:20:44 +02:00
parent b83944959b
commit 805ab1bb57
Signed by: Anri
GPG key ID: A82D63DFF8D1317F
1 changed files with 14 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

24
main.py
View file

@ -4,7 +4,7 @@ from random import choice
from re import findall, sub from re import findall, sub
from dotenv import load_dotenv from dotenv import load_dotenv
from tweepy import Client, StreamingClient, StreamRule, Tweet, errors from tweepy import Client, StreamingClient, StreamRule, Tweet, User, errors
def load(variables) -> dict: def load(variables) -> dict:
@ -88,21 +88,25 @@ class Listener(StreamingClient):
def on_connect(self): def on_connect(self):
print(f"Début du scroll sur Twitter...") print(f"Début du scroll sur Twitter...")
def _get_user(self, uid: int) -> str: def _get_user(self, uid: int) -> User:
"""Return username by ID, with cache support""" """Return username by ID, with cache support"""
# If not cached # If not cached
if not uid in self.cache: if not uid in self.cache:
# Fetch from Twitter # Fetch from Twitter
self.cache[uid] = self.client.get_user( self.cache[uid] = self.client.get_user(
id=uid, user_auth=True).data.username id=uid, user_fields="protected", user_auth=True).data
# Return the username # Return the user
return self.cache[uid] return self.cache[uid]
def on_tweet(self, tweet: Tweet): def on_tweet(self, tweet: Tweet):
# Check if the tweet is not a retweet # Check if the tweet is not a retweet
if not tweet.text.startswith("RT @"): if not tweet.text.startswith("RT @"):
username = self._get_user(tweet.author_id) # Cancel if author of the first tweet in the conversation is in private
if tweet.conversation_id:
if self._get_user(self.client.get_tweet(id=tweet.conversation_id, tweet_fields="author_id", user_auth=True).data.author_id).protected:
return
author = self._get_user(tweet.author_id)
# Clean the tweet # Clean the tweet
lastWord = cleanTweet(tweet.text) lastWord = cleanTweet(tweet.text)
@ -119,7 +123,7 @@ class Listener(StreamingClient):
infoLastWord += f"dernier mot : {lastWord.split()[-1:][0]}" infoLastWord += f"dernier mot : {lastWord.split()[-1:][0]}"
newline = "" newline = ""
print( print(
f"Tweet trouvé de {username} ({infoLastWord})...{newline}", end=" ") f"Tweet trouvé de {author.username} ({infoLastWord})...{newline}", end=" ")
# Ignore a tweet # Ignore a tweet
if lastWord == None or len(lastWord) == 0: if lastWord == None or len(lastWord) == 0:
@ -134,7 +138,7 @@ class Listener(StreamingClient):
# Check repetition # Check repetition
repetition = findall(r"di(\S+)", lastWord) repetition = findall(r"di(\S+)", lastWord)
if(len(repetition) > 0): if len(repetition) > 0:
# We need to repeat something... # We need to repeat something...
answer = repeater(repetition[0]) answer = repeater(repetition[0])
@ -165,11 +169,11 @@ class Listener(StreamingClient):
# Send the tweet with the answer # Send the tweet with the answer
self.client.create_tweet( self.client.create_tweet(
in_reply_to_tweet_id=tweet.id, text=choice(answer)) in_reply_to_tweet_id=tweet.id, text=choice(answer))
print(f"{username} s'est fait {answer[0]} !") print(f"{author.username} s'est fait {answer[0]} !")
except errors.Forbidden: except errors.Forbidden:
if keys["VERBOSE"]: if keys["VERBOSE"]:
print( print(
f"{errorMessage[:-2]} ! Tweet supprimé ou auteur ({username}) en privé/bloqué.") f"{errorMessage[:-2]} ! Tweet supprimé ou auteur ({author.username}) en privé/bloqué.")
else: else:
if keys["VERBOSE"]: if keys["VERBOSE"]:
print("Annulation car le dernier mot n'est pas intéressant.") print("Annulation car le dernier mot n'est pas intéressant.")
@ -366,7 +370,7 @@ def start():
stream.add_rules([StreamRule(rule) for rule in rules]) stream.add_rules([StreamRule(rule) for rule in rules])
# Apply the filter # Apply the filter
stream.filter(threaded=True, tweet_fields="author_id") stream.filter(threaded=True, tweet_fields=["author_id", "conversation_id"])
if __name__ == "__main__": if __name__ == "__main__":