598 lines
20 KiB
Python
598 lines
20 KiB
Python
from datetime import datetime
|
|
from os import environ
|
|
from random import choice
|
|
from re import findall, sub
|
|
|
|
from dotenv import load_dotenv
|
|
from tweepy import Client, StreamingClient, StreamRule, Tweet, User, errors
|
|
|
|
|
|
def load(variables) -> dict:
|
|
"""Load environment variables"""
|
|
keys = {}
|
|
load_dotenv() # load .env file
|
|
for var in variables:
|
|
try:
|
|
if var == "VERBOSE": # check is VERBOSE is set
|
|
try:
|
|
if environ[var].lower() == "true":
|
|
res = True
|
|
except:
|
|
res = False # if not its False
|
|
elif var == "WHITELIST": # check if WHITELIST is set
|
|
try:
|
|
res = list(set(environ[var].split(",")) - {""})
|
|
except:
|
|
res = [] # if not its an empty list
|
|
elif var == "FORCELIST": # check if FORCELIST is set
|
|
try:
|
|
res = list(set(environ[var].split(",")) - {""})
|
|
except:
|
|
res = [] # if not its an empty list
|
|
else:
|
|
res = environ[var]
|
|
if var == "PSEUDOS":
|
|
# create a list for the channels and remove blank channels and doubles
|
|
res = list(set(res.split(",")) - {""})
|
|
keys[var] = res
|
|
except KeyError:
|
|
print(
|
|
f"Veuillez définir la variable d'environnement {var} (fichier .env supporté)")
|
|
exit(1)
|
|
return keys
|
|
|
|
|
|
def cleanTweet(tweet: str) -> str | None:
|
|
"""Remove all unwanted elements from the tweet"""
|
|
# Convert to lower case
|
|
tweet = tweet.lower()
|
|
# Remove URLs
|
|
tweet = sub(r"(https?:\/\/\S+|www.\S+)", " ", tweet)
|
|
# Check all hashtags
|
|
hashtagMatch = findall(r"#\S+", tweet)
|
|
# If less than 3
|
|
if len(hashtagMatch) < 3:
|
|
# Remove them
|
|
tweet = sub(r"#\S+", " ", tweet)
|
|
else:
|
|
# Too much hashtags in the tweet -> so ignore it
|
|
return None
|
|
# Remove usernames
|
|
tweet = sub(r"@\S+", " ", tweet)
|
|
# Remove everything who isn't a letter/number/space
|
|
tweet = sub(r" *?[^\w\s]+", " ", tweet)
|
|
# Remove element of the word only if the last syllable can be matched
|
|
# (so more words will be answered without adding them manually)
|
|
tweet = sub(r"\S+(?=si|ci)", " ", tweet)
|
|
|
|
# Remove key smashing in certains words
|
|
# 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)
|
|
|
|
return tweet.strip()
|
|
|
|
|
|
class Listener(StreamingClient):
|
|
"""Watch for tweets that match criteria in real-time"""
|
|
|
|
def __init__(
|
|
self,
|
|
bearer_token,
|
|
client: Client,
|
|
):
|
|
super(Listener, self).__init__(bearer_token, wait_on_rate_limit=True)
|
|
self.client = client
|
|
self.cache = {}
|
|
|
|
def on_connect(self):
|
|
print(f"Début du scroll sur Twitter...")
|
|
|
|
def _get_user(self, uid: int) -> User:
|
|
"""Return username by ID, with cache support"""
|
|
# If not cached
|
|
if not uid in self.cache:
|
|
# Fetch from Twitter
|
|
self.cache[uid] = self.client.get_user(
|
|
id=uid, user_fields="protected", user_auth=True).data
|
|
|
|
# Return the user
|
|
return self.cache[uid]
|
|
|
|
def on_tweet(self, tweet: Tweet):
|
|
# Check if the tweet is not a retweet
|
|
if not tweet.text.startswith("RT @"):
|
|
# Cancel if author of the first tweet in the conversation is in private
|
|
if tweet.conversation_id:
|
|
if tweet.id != tweet.conversation_id:
|
|
if keys["VERBOSE"]:
|
|
print("Thread...", end=" ")
|
|
base_tweet = self.client.get_tweet(
|
|
id=tweet.conversation_id, tweet_fields="author_id", user_auth=True).data
|
|
# Sometimes Twitter don't give what we want
|
|
if hasattr(base_tweet, 'author_id'):
|
|
base_author = self._get_user(base_tweet.author_id)
|
|
# Check if account is in private mode
|
|
if base_author.protected:
|
|
if keys["VERBOSE"]:
|
|
print(
|
|
"Auteur du premier tweet en privé, pas de réponses.")
|
|
return
|
|
else:
|
|
if keys["VERBOSE"]:
|
|
print(
|
|
"Auteur du premier tweet en publique...", end=" ")
|
|
else:
|
|
# Can't check the status of the first tweet in the thread, ignoring for safety
|
|
if keys["VERBOSE"]:
|
|
print(
|
|
"Impossible de vérifier le status de l'auteur du fil.")
|
|
return
|
|
else:
|
|
# Can't check if we're in a thread or not
|
|
if keys["VERBOSE"]:
|
|
print("Impossible de vérifier si le tweet est dans un thread.")
|
|
return
|
|
author = self._get_user(tweet.author_id)
|
|
# Clean the tweet
|
|
lastWord = cleanTweet(tweet.text)
|
|
|
|
# Log
|
|
if keys["VERBOSE"]:
|
|
infoLastWord = "dernier mot : "
|
|
newline = "\n"
|
|
match lastWord:
|
|
case None:
|
|
infoLastWord += "tweet ignoré car trop de hashtags"
|
|
case w if len(w) == 0:
|
|
infoLastWord += "tweet pas intéressant"
|
|
case _:
|
|
infoLastWord += f"dernier mot : {lastWord.split()[-1:][0]}"
|
|
newline = ""
|
|
print(
|
|
f"Tweet trouvé de {author.username} ({infoLastWord})...{newline}", end=" ")
|
|
|
|
# Ignore a tweet
|
|
if lastWord == None or len(lastWord) == 0:
|
|
return
|
|
|
|
# Fetch the last word of the tweet
|
|
lastWord = lastWord.split()[-1:][0]
|
|
|
|
# Check if the last word found is a supported word
|
|
if lastWord in universalBase:
|
|
answer = None
|
|
|
|
# Check repetition
|
|
repetition = findall(r"di(\S+)", lastWord)
|
|
if len(repetition) > 0:
|
|
# We need to repeat something...
|
|
answer = repeater(repetition[0])
|
|
|
|
# Fetch an other adequate (better) response
|
|
for mot in base.items():
|
|
if lastWord in mot[1]:
|
|
# Handle specific case
|
|
if mot[0] == "bon":
|
|
# Between 7am and 5pm
|
|
if datetime.now().hour in range(7, 17):
|
|
answer = answers[mot[0]][0] # jour
|
|
else:
|
|
answer = answers[mot[0]][1] # soir
|
|
else:
|
|
# Normal answer
|
|
answer = answers[mot[0]]
|
|
|
|
# If no answer has been found
|
|
if answer == None:
|
|
if keys["VERBOSE"]:
|
|
print(f"{errorMessage} Aucune réponse trouvée.")
|
|
|
|
# If an answer has been found
|
|
else:
|
|
if keys["VERBOSE"]:
|
|
print(f"Envoie d'un {answer[0]}...", end=" ")
|
|
try:
|
|
# Send the tweet with the answer
|
|
self.client.create_tweet(
|
|
in_reply_to_tweet_id=tweet.id, text=choice(answer))
|
|
print(f"{author.username} s'est fait -{answer[0]} !")
|
|
except errors.Forbidden:
|
|
if keys["VERBOSE"]:
|
|
print(
|
|
f"{errorMessage[:-2]} ! Tweet supprimé ou auteur ({author.username}) en privé/bloqué.")
|
|
else:
|
|
if keys["VERBOSE"]:
|
|
print("Annulation car le dernier mot n'est pas intéressant.")
|
|
|
|
def on_request_error(self, status_code):
|
|
print(f"{errorMessage[:-2]} ({status_code}) !", end=" ")
|
|
match status_code:
|
|
case 420:
|
|
if keys["VERBOSE"]:
|
|
print("Déconnecter du flux.")
|
|
case 429:
|
|
if keys["VERBOSE"]:
|
|
print("En attente de reconnexion...")
|
|
case _:
|
|
# newline
|
|
print("")
|
|
return False
|
|
|
|
|
|
def repeater(word: str) -> str:
|
|
"""Formating a word who need to be repeated"""
|
|
# Remove first letter if the first letter is a "S" or a "T"
|
|
# Explanation: Trigger word for the repeater is "di" and sometimes it is
|
|
# "dis", sometimes its "dit", that's why we need to remove this 2 letters
|
|
# from the final answer
|
|
if word[0] == "s" or word[0] == "t":
|
|
word = word[1:]
|
|
|
|
# Random format from the base answer
|
|
return createBaseAnswers(word)
|
|
|
|
|
|
def getFriends(client: Client, users: list[str]) -> list:
|
|
"""Get all friends of choosen users"""
|
|
friends_list = []
|
|
# Get IDs of the user's friends
|
|
for user in users:
|
|
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))
|
|
return friends_list[0]
|
|
|
|
|
|
def createBaseTrigger(lists: list[list]) -> list:
|
|
"""Merges all given lists into one"""
|
|
listing = []
|
|
for liste in lists:
|
|
listing.extend(liste)
|
|
return list(set(listing))
|
|
|
|
|
|
def createBaseAnswers(word: str) -> list:
|
|
"""Generates default answers for a given word"""
|
|
irritating_word = [
|
|
"lol",
|
|
"👀",
|
|
"XD",
|
|
]
|
|
|
|
return [
|
|
# Assuming the first element of this list is always word, don't change it
|
|
word,
|
|
f"({word})",
|
|
word.upper(),
|
|
f"{word} {choice(irritating_word)}",
|
|
f"{word}...",
|
|
]
|
|
|
|
|
|
def createClient(consumer_key, consumer_secret, access_token, access_token_secret) -> Client:
|
|
"""Create a client for the Twitter API v2"""
|
|
client = Client(
|
|
consumer_key=consumer_key,
|
|
consumer_secret=consumer_secret,
|
|
access_token=access_token,
|
|
access_token_secret=access_token_secret
|
|
)
|
|
|
|
if keys["VERBOSE"]:
|
|
try:
|
|
print(
|
|
f"Authentification réussie en tant que @{client.get_me().data.username}.\n")
|
|
|
|
# Compte ignorés
|
|
if keys["WHITELIST"] == []:
|
|
whitelist = "Aucun"
|
|
else:
|
|
whitelist = f"@{', @'.join(keys['WHITELIST'])}"
|
|
print(f"Liste des comptes ignorés : {whitelist}.")
|
|
|
|
# Compte forcés
|
|
if keys["FORCELIST"] == []:
|
|
forcelist = "Aucun"
|
|
else:
|
|
forcelist = f"@{', @'.join(keys['FORCELIST'])}"
|
|
print(f"Liste des comptes forcés : {forcelist}.")
|
|
|
|
# Compte aux following suivis
|
|
if keys["PSEUDOS"] == []:
|
|
pseudos = "Aucun"
|
|
else:
|
|
pseudos = f"@{', @'.join(keys['PSEUDOS'])}"
|
|
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")
|
|
except:
|
|
print("Erreur d'authentification.")
|
|
exit(1)
|
|
|
|
return client
|
|
|
|
|
|
def create_rules(tracked_users: list[str]) -> list[str]:
|
|
"""Create rules for tracking users, by respecting the twitter API policies"""
|
|
rules = []
|
|
|
|
# Repeating rules
|
|
repeat = "-is:retweet lang:fr ("
|
|
|
|
# Buffer
|
|
buffer = repeat
|
|
|
|
tracked_users.sort()
|
|
|
|
# Track users
|
|
for user in tracked_users:
|
|
# Check if the rule don't exceeds the maximum length of a rule (512)
|
|
# 5 is len of "from:"
|
|
# 1 is len for closing parenthesis
|
|
if len(buffer) + len(user) + 5 + 1 > 512:
|
|
rules.append(buffer[:-4] + ")")
|
|
buffer = repeat
|
|
buffer += f"from:{user} OR "
|
|
|
|
if len(buffer) > 0:
|
|
rules.append(buffer[:-4] + ")")
|
|
|
|
if len(rules) > 25:
|
|
raise BufferError("Too much rules.")
|
|
|
|
return rules
|
|
|
|
|
|
def start():
|
|
"""Start the bot"""
|
|
client = createClient(
|
|
keys["CONSUMER_KEY"],
|
|
keys["CONSUMER_SECRET"],
|
|
keys["TOKEN"],
|
|
keys["TOKEN_SECRET"],
|
|
)
|
|
|
|
# Only track specifics users
|
|
# Including users in forcelist and removing users in whitelist
|
|
tracked_users = [
|
|
i for i in [
|
|
user.data["username"] for user in getFriends(client, keys["PSEUDOS"])
|
|
] + keys["FORCELIST"] if i not in keys["WHITELIST"]
|
|
]
|
|
|
|
stream = Listener(keys["BEARER_TOKEN"], client)
|
|
|
|
# Gathering rules
|
|
rules = [rule for rule in create_rules(tracked_users)]
|
|
|
|
# Check if rules already exists
|
|
old_rules = stream.get_rules().data
|
|
old_rules_values = []
|
|
if old_rules:
|
|
old_rules_values = [rule.value for rule in old_rules]
|
|
need_changes = False
|
|
# Same amount of rules
|
|
if len(old_rules_values) == len(rules):
|
|
for rule in rules:
|
|
# Check if Twitter doesn't know the rule and change rules if needed
|
|
if rule not in old_rules_values:
|
|
need_changes = True
|
|
break
|
|
else:
|
|
need_changes = True
|
|
|
|
if need_changes:
|
|
if keys["VERBOSE"]:
|
|
print("Changes needed... ", end=" ")
|
|
# Clean old rules
|
|
if old_rules:
|
|
if keys["VERBOSE"]:
|
|
print("deleting old rules... ", end=" ")
|
|
stream.delete_rules([rule.id for rule in old_rules])
|
|
|
|
# Add new rules
|
|
if keys["VERBOSE"]:
|
|
print("sending new filter to Twitter.")
|
|
stream.add_rules([StreamRule(rule) for rule in rules])
|
|
|
|
# Apply the filter
|
|
stream.filter(tweet_fields=["author_id", "conversation_id"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
"""
|
|
TOKEN is the Access Token available in the Authentication Tokens section under the Access Token and Secret sub-heading
|
|
TOKEN_SECRET is the Access Token Secret available in the Authentication Tokens section under the Access Token and Secret sub-heading
|
|
CONSUMER_KEY is the API Key available in the Consumer Keys section under the API Key and Secret sub-heading
|
|
CONSUMER_SECRET is the API Secret Key available in the Consumer Keys section under the API Key and Secret sub-heading
|
|
BEARER_TOKEN is the Bearer Token available in the Authentication Tokens section under the Bearer Token sub-heading
|
|
--
|
|
PSEUDOS is a list of account you want to listen, all of his·er following (guys followed by PSEUDO) will be sniped
|
|
WHITELSIT is a list of account who are protected from the bot
|
|
FORCELIST is a list of account who are targeted by the bot, if user is in the whitelist, he·r will be ignored
|
|
---
|
|
VERBOSE enable some debugs log
|
|
"""
|
|
# Error message
|
|
errorMessage = "Une erreur survient !"
|
|
|
|
# Words who trigger the bot (keys in lowercase)
|
|
base = {
|
|
"quoi": ["quoi", "koi", "quoient"],
|
|
"oui": ["oui", "ui", "wi"],
|
|
"non": ["non", "nn"],
|
|
"nan": ["nan"],
|
|
"hein": ["hein", "1", "un"],
|
|
"ci": ["ci", "si"],
|
|
"con": ["con"],
|
|
"ok": ["ok", "okay", "oké", "k"],
|
|
"ouais": ["ouais", "oué"],
|
|
"comment": ["comment"],
|
|
"mais": ["mais", "mé"],
|
|
"fort": ["fort"],
|
|
"coup": ["coup", "cou"],
|
|
"ça": ["ça", "sa"],
|
|
"bon": ["bon"],
|
|
"qui": ["qui", "ki"],
|
|
"sur": ["sur", "sûr"],
|
|
"pas": ["pas", "pa"],
|
|
"ka": ["ka", "kha"],
|
|
"fais": ["fais", "fait"],
|
|
"tant": ["tant", "temps", "tend", "tends"],
|
|
"et": ["et"],
|
|
"la": ["la", "là"],
|
|
"tki": ["tki"],
|
|
"moi": ["moi", "mwa"],
|
|
"toi": ["toi", "toit"],
|
|
"top": ["top"],
|
|
"jour": ["jour", "bonjour"],
|
|
"ya": ["ya", "y'a"],
|
|
"yo": ["yo"],
|
|
"ni": ["ni"],
|
|
"re": ["re", "reu", "reuh"],
|
|
"quand": ["quand", "kan", "qand", "quan"],
|
|
"sol": ["sol"],
|
|
"vois": ["vois", "voit", "voie", "voi"],
|
|
"akhy": ["akhy", "aquis", "aquit"],
|
|
}
|
|
|
|
# Answers for all the triggers (keys in lowercase)
|
|
answers = {
|
|
"quoi": createBaseAnswers("feur")
|
|
+ createBaseAnswers("feuse")
|
|
+ [
|
|
"https://twitter.com/Myshawii/status/1423219640025722880/video/1",
|
|
"https://twitter.com/Myshawii/status/1423219684552417281/video/1",
|
|
"feur (-isson -ictalope -diatre -uil)",
|
|
"https://twitter.com/Myshawii/status/1455469162202075138/video/1",
|
|
"https://twitter.com/Myshawii/status/1552026689101860865/video/1",
|
|
"https://twitter.com/Myshawii/status/1553112547678720001/photo/1"
|
|
],
|
|
|
|
"oui": createBaseAnswers("stiti")
|
|
+ createBaseAnswers("fi"),
|
|
|
|
"non": createBaseAnswers("bril")
|
|
+ [
|
|
"bril (-liant)"
|
|
],
|
|
|
|
"nan": createBaseAnswers("cy"),
|
|
|
|
"hein": createBaseAnswers("deux")
|
|
+ createBaseAnswers("bécile")
|
|
+ [
|
|
"2"
|
|
],
|
|
|
|
"ci": createBaseAnswers("tron")
|
|
+ createBaseAnswers("prine"),
|
|
|
|
"con": createBaseAnswers("combre")
|
|
+ createBaseAnswers("gelé")
|
|
+ createBaseAnswers("pas"),
|
|
|
|
"ok": createBaseAnswers("sur glace"),
|
|
|
|
"ouais": createBaseAnswers("stern"),
|
|
|
|
"comment": createBaseAnswers("tateur")
|
|
+ createBaseAnswers("tatrice")
|
|
+ createBaseAnswers("dant Cousteau"),
|
|
|
|
"mais": createBaseAnswers("on")
|
|
+ [
|
|
"on (-dulation)"
|
|
],
|
|
|
|
"fort": createBaseAnswers("boyard")
|
|
+ [
|
|
"boyard (-ennes)"
|
|
],
|
|
|
|
"coup": createBaseAnswers("teau"),
|
|
|
|
"ça": createBaseAnswers("perlipopette")
|
|
+ createBaseAnswers("von")
|
|
+ createBaseAnswers("pristi")
|
|
+ [
|
|
"pristi (-gnasse)"
|
|
],
|
|
|
|
"bon": [
|
|
createBaseAnswers("jour"),
|
|
createBaseAnswers("soir")
|
|
],
|
|
|
|
"qui": createBaseAnswers("wi")
|
|
+ createBaseAnswers("mono"),
|
|
|
|
"sur": createBaseAnswers("prise"),
|
|
|
|
"pas": createBaseAnswers("nini")
|
|
+ createBaseAnswers("steur")
|
|
+ createBaseAnswers("trimoine")
|
|
+ createBaseAnswers("té")
|
|
+ createBaseAnswers("stis"),
|
|
|
|
"ka": createBaseAnswers("pitaine")
|
|
+ createBaseAnswers("pitulation"),
|
|
|
|
"fais": createBaseAnswers("rtile"),
|
|
|
|
"tant": createBaseAnswers("gente")
|
|
+ createBaseAnswers("tation"),
|
|
|
|
"et": createBaseAnswers("eint")
|
|
+ createBaseAnswers("ain"),
|
|
|
|
"la": createBaseAnswers("vabo")
|
|
+ createBaseAnswers("vande"),
|
|
|
|
"tki": createBaseAnswers("la"),
|
|
|
|
"moi": createBaseAnswers("tié")
|
|
+ createBaseAnswers("sson")
|
|
+ createBaseAnswers("sissure"),
|
|
|
|
"toi": createBaseAnswers("lette")
|
|
+ createBaseAnswers("ture"),
|
|
|
|
"top": createBaseAnswers("inambour"),
|
|
|
|
"jour": createBaseAnswers("nal"),
|
|
|
|
"ya": createBaseAnswers("hourt"),
|
|
|
|
"yo": createBaseAnswers("ghourt")
|
|
+ createBaseAnswers("yo"),
|
|
|
|
"ni": createBaseAnswers("cotine"),
|
|
|
|
"re": createBaseAnswers("pas")
|
|
+ createBaseAnswers("veil")
|
|
+ createBaseAnswers("tourne"),
|
|
|
|
"quand": createBaseAnswers("dide")
|
|
+ createBaseAnswers("tal")
|
|
+ createBaseAnswers("didat"),
|
|
|
|
"sol": createBaseAnswers("itaire"),
|
|
|
|
"vois": createBaseAnswers("ture"),
|
|
|
|
"akhy": createBaseAnswers("nator"),
|
|
}
|
|
|
|
# List of all the trigger words
|
|
universalBase = createBaseTrigger(list(base.values()))
|
|
|
|
# Loading environment variables
|
|
keys = load(["TOKEN", "TOKEN_SECRET", "CONSUMER_KEY", "CONSUMER_SECRET",
|
|
"BEARER_TOKEN", "PSEUDOS", "VERBOSE", "WHITELIST", "FORCELIST"])
|
|
|
|
# Start the bot
|
|
start()
|