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

follow pep8

Browse source
This commit is contained in:
Mylloon 2022-08-07 21:55:48 +02:00
parent 36792c1949
commit 52525cab51
Signed by: Anri
GPG key ID: A82D63DFF8D1317F
1 changed files with 95 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

51
main.py
View file

@ -33,13 +33,16 @@ def load(variables) -> dict:
else: else:
res = environ[var] res = environ[var]
if var == "PSEUDOS": if var == "PSEUDOS":
res = list(set(res.split(",")) - {""}) # create a list for the channels and remove blank channels and doubles # create a list for the channels and remove blank channels and doubles
res = list(set(res.split(",")) - {""})
keys[var] = res keys[var] = res
except KeyError: except KeyError:
print(f"Veuillez définir la variable d'environnement {var} (fichier .env supporté)") print(
f"Veuillez définir la variable d'environnement {var} (fichier .env supporté)")
exit(1) exit(1)
return keys return keys
def cleanTweet(tweet: str) -> str: def cleanTweet(tweet: str) -> str:
"""Remove all unwanted elements from the tweet""" """Remove all unwanted elements from the tweet"""
# Convert to lower case # Convert to lower case
@ -65,12 +68,15 @@ def cleanTweet(tweet: str) -> str:
# Remove key smashing in certains words # Remove key smashing in certains words
# uiii naaaan quoiiii noooon heiiin siiii # uiii naaaan quoiiii noooon heiiin siiii
tweet = sub(r"(?<=ui)i+|(?<=na)a+(?<!n)|(?<=quoi)i+|(?<=no)o+(?<!n)|(?<=hei)i+(?<!n)|(?<=si)i+", "", tweet) tweet = sub(
r"(?<=ui)i+|(?<=na)a+(?<!n)|(?<=quoi)i+|(?<=no)o+(?<!n)|(?<=hei)i+(?<!n)|(?<=si)i+", "", tweet)
return tweet.strip() return tweet.strip()
class Listener(StreamingClient): class Listener(StreamingClient):
"""Watch for tweets that match criteria in real-time""" """Watch for tweets that match criteria in real-time"""
def __init__( def __init__(
self, self,
bearer_token, bearer_token,
@ -92,11 +98,14 @@ class Listener(StreamingClient):
# Verify the age of the tweet # Verify the age of the tweet
if seniority(tweet.created_at): if seniority(tweet.created_at):
# Fetch the last word of the tweet # Fetch the last word of the tweet
tweet.username: str = self.client.get_user(id=tweet.author_id, user_auth=True).data.username tweet.username: str = self.client.get_user(
id=tweet.author_id, user_auth=True).data.username
lastWord = cleanTweet(tweet.text).split()[-1:][0] lastWord = cleanTweet(tweet.text).split()[-1:][0]
if keys["VERBOSE"]: if keys["VERBOSE"]:
infoLastWord = f"dernier mot : \"{lastWord}\"" if len(lastWord) > 0 else "tweet ignoré car trop de hashtags" infoLastWord = f"dernier mot : \"{lastWord}\"" if len(
print(f"Tweet trouvé de {tweet.username} ({infoLastWord})...", end=" ") lastWord) > 0 else "tweet ignoré car trop de hashtags"
print(
f"Tweet trouvé de {tweet.username} ({infoLastWord})...", end=" ")
# Check if the last word found is a supported word # Check if the last word found is a supported word
if lastWord in universalBase: if lastWord in universalBase:
answer = None answer = None
@ -132,7 +141,8 @@ class Listener(StreamingClient):
print(f"Envoie d'un {answer[0]}...", end=" ") print(f"Envoie d'un {answer[0]}...", end=" ")
try: try:
# Send the tweet with the answer # Send the tweet with the answer
self.client.create_tweet(in_reply_to_tweet_id=tweet.id, text=choice(answer)) self.client.create_tweet(
in_reply_to_tweet_id=tweet.id, text=choice(answer))
print(f"{tweet.username} s'est fait {answer[0]} !") print(f"{tweet.username} s'est fait {answer[0]} !")
except Exception as error: except Exception as error:
error = loads(error.response.text)["errors"][0] error = loads(error.response.text)["errors"][0]
@ -144,7 +154,8 @@ class Listener(StreamingClient):
# Show error only if relevant, always in verbose # Show error only if relevant, always in verbose
if show_error or keys["VERBOSE"]: if show_error or keys["VERBOSE"]:
print(f"{errorMessage[:-2]} ({error['code']}) ! {error['message']}") print(
f"{errorMessage[:-2]} ({error['code']}) ! {error['message']}")
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.")
@ -158,6 +169,7 @@ class Listener(StreamingClient):
print("\n") print("\n")
return False return False
def repeater(word: str) -> str: def repeater(word: str) -> str:
"""Formating a word who need to be repeated""" """Formating a word who need to be repeated"""
# Remove first letter if the first letter is a "S" or a "T" # Remove first letter if the first letter is a "S" or a "T"
@ -170,15 +182,18 @@ def repeater(word: str) -> str:
# Random format from the base answer # Random format from the base answer
return createBaseAnswers(word) return createBaseAnswers(word)
def getFriends(client: Client, users: list[str]) -> list: def getFriends(client: Client, users: list[str]) -> list:
"""Get all friends of choosen users""" """Get all friends of choosen users"""
friends_list = [] friends_list = []
# Get IDs of the user's friends # Get IDs of the user's friends
for user in users: for user in users:
user_id = client.get_user(username=user, user_auth=True).data.id user_id = client.get_user(username=user, user_auth=True).data.id
friends_list.extend(client.get_users_following(id=user_id, user_auth=True)) friends_list.extend(client.get_users_following(
id=user_id, user_auth=True))
return friends_list[0] return friends_list[0]
def seniority(date: datetime) -> bool: def seniority(date: datetime) -> bool:
"""Return True only if the given string date is less than one day old""" """Return True only if the given string date is less than one day old"""
# Convert string format to datetime format # Convert string format to datetime format
@ -190,6 +205,7 @@ def seniority(date: datetime) -> bool:
# False if older than a day, else True # False if older than a day, else True
return False if age.days >= 1 else True return False if age.days >= 1 else True
def createBaseTrigger(lists: list[list]) -> list: def createBaseTrigger(lists: list[list]) -> list:
"""Merges all given lists into one""" """Merges all given lists into one"""
listing = [] listing = []
@ -197,6 +213,7 @@ def createBaseTrigger(lists: list[list]) -> list:
listing.extend(liste) listing.extend(liste)
return list(set(listing)) return list(set(listing))
def createBaseAnswers(word: str) -> list: def createBaseAnswers(word: str) -> list:
"""Generates default answers for a given word""" """Generates default answers for a given word"""
irritating_word = [ irritating_word = [
@ -213,6 +230,7 @@ def createBaseAnswers(word: str) -> list:
f"{word}...", f"{word}...",
] ]
def createClient(consumer_key, consumer_secret, access_token, access_token_secret) -> Client: def createClient(consumer_key, consumer_secret, access_token, access_token_secret) -> Client:
"""Create a client for the Twitter API v2""" """Create a client for the Twitter API v2"""
client = Client( client = Client(
@ -225,7 +243,8 @@ def createClient(consumer_key, consumer_secret, access_token, access_token_secre
if keys["VERBOSE"]: if keys["VERBOSE"]:
try: try:
client.get_me().data.username client.get_me().data.username
print(f"Authentification réussie en tant que @{client.get_me().data.username}.\n") print(
f"Authentification réussie en tant que @{client.get_me().data.username}.\n")
# Compte ignorés # Compte ignorés
if keys['WHITELIST'] == []: if keys['WHITELIST'] == []:
@ -246,15 +265,18 @@ def createClient(consumer_key, consumer_secret, access_token, access_token_secre
pseudos = "Aucun" pseudos = "Aucun"
else: else:
pseudos = f"@{', @'.join(keys['PSEUDOS'])}" pseudos = f"@{', @'.join(keys['PSEUDOS'])}"
print(f"Les comptes suivis par ces comptes sont traqués : {pseudos}.\n") print(
f"Les comptes suivis par ces comptes sont traqués : {pseudos}.\n")
print("Notez que si un compte est dans la whitelist, il sera dans tout les cas ignoré.\n") print(
"Notez que si un compte est dans la whitelist, il sera dans tout les cas ignoré.\n")
except: except:
print("Erreur d'authentification.") print("Erreur d'authentification.")
exit(1) exit(1)
return client return client
def create_rules(tracked_users: list[str]) -> list[str]: def create_rules(tracked_users: list[str]) -> list[str]:
"""Create rules for tracking users, by respecting the twitter API policies""" """Create rules for tracking users, by respecting the twitter API policies"""
rules = [] rules = []
@ -280,6 +302,7 @@ def create_rules(tracked_users: list[str]) -> list[str]:
return rules return rules
def start(): def start():
"""Start the bot""" """Start the bot"""
client = createClient( client = createClient(
@ -303,6 +326,7 @@ def start():
stream.add_rules(StreamRule(rule)) stream.add_rules(StreamRule(rule))
stream.filter(threaded=True) stream.filter(threaded=True)
if __name__ == "__main__": if __name__ == "__main__":
""" """
TOKEN is the Access Token available in the Authentication Tokens section under the Access Token and Secret sub-heading TOKEN is the Access Token available in the Authentication Tokens section under the Access Token and Secret sub-heading
@ -489,7 +513,8 @@ if __name__ == "__main__":
universalBase = createBaseTrigger(list(base.values())) universalBase = createBaseTrigger(list(base.values()))
# Loading environment variables # Loading environment variables
keys = load(["TOKEN", "TOKEN_SECRET", "CONSUMER_KEY", "CONSUMER_SECRET", "BEARER_TOKEN", "PSEUDOS", "VERBOSE", "WHITELIST", "FORCELIST"]) keys = load(["TOKEN", "TOKEN_SECRET", "CONSUMER_KEY", "CONSUMER_SECRET",
"BEARER_TOKEN", "PSEUDOS", "VERBOSE", "WHITELIST", "FORCELIST"])
# Start the bot # Start the bot
start() start()