Compare commits
47 commits
f9a64cfeb5
...
61a1d52bf8
Author | SHA1 | Date | |
---|---|---|---|
61a1d52bf8 | |||
83956d2d47 | |||
30be0904d0 | |||
c0c00a0759 | |||
dc28c29c6f | |||
945844712f | |||
bda4d8ccd4 | |||
43c962c3e0 | |||
805ab1bb57 | |||
b83944959b | |||
eda6804d38 | |||
8c57391b54 | |||
f351e477e3 | |||
6eab4a41b3 | |||
b1c5abe1f9 | |||
68144b56ca | |||
a1effc03c7 | |||
c8db7f6436 | |||
d23b9b57c3 | |||
30615db063 | |||
1eb5f84cc0 | |||
f595dcd83d | |||
dcda3d107e | |||
3b5c78f539 | |||
a0032708af | |||
95f2fe551d | |||
2d8ef4e0ce | |||
7fad7d09c6 | |||
11f5180346 | |||
2110a4d961 | |||
e347f7fcb7 | |||
d1ee920d7a | |||
9300961afa | |||
2513236f6c | |||
a40866e736 | |||
52525cab51 | |||
36792c1949 | |||
46037106e6 | |||
ed6e72270c | |||
f99ab71369 | |||
4e437ad675 | |||
c0b5c168a7 | |||
a2e118c38f | |||
86853437a9 | |||
0874d5211d | |||
756016ae8d | |||
78bfe13fa3 |
3 changed files with 407 additions and 282 deletions
|
@ -1,5 +1,6 @@
|
|||
TOKEN=
|
||||
TOKEN_SECRET=
|
||||
CONSUMER_KEY=
|
||||
CONSUMER_SECRET=
|
||||
BEARER_TOKEN=
|
||||
TOKEN=
|
||||
TOKEN_SECRET=
|
||||
PSEUDO=
|
||||
|
|
89
README.md
89
README.md
|
@ -2,33 +2,26 @@
|
|||
|
||||
Bot qui envoie automatiquement des réponses ennuyantes quand les personnes que tu suis finissent leur tweet par un mot spécial.
|
||||
|
||||
Certains mots peuvent servir de "trigger" sans être dans la liste, example : `aussi` n'est pas dans la liste, mais en retirant `aus`, on obtient `si`, qui est dans la liste.
|
||||
| Mot | Réponse | ¦ | Mot | Réponse | ¦ | Mot | Réponse | ¦ | Mot | Réponse | ¦ | Mot | Réponse | ¦ | Mot | Réponse
|
||||
------|--------- |:-:|---------|-----------|:-:|------|-------------------------------|:-:|-------------------------|------------|:-:|-------|------------------ |:-:|-------|-
|
||||
quoi | feur | ¦ | con | combre | ¦ | coup | teau | ¦ | ka | pitaine | ¦ | moi | tié/sson/sissure | ¦ | ni | cotine
|
||||
oui | stiti/fi | ¦ | ok | sur glace | ¦ | ça | pristi/perlipopette/von | ¦ | fais | rtile | ¦ | toi | lette/ture | ¦ | quand | dide/tal/didat
|
||||
non | bril | ¦ | ouais | stern | ¦ | bon | jour/soir (dépend de l'heure) | ¦ | tant (ou autre syntaxe) | gente | ¦ | top | inambour | ¦ | sol | itaire
|
||||
nan | cy | ¦ | comment | tateur | ¦ | qui | wi/mono | ¦ | et | eint/ain | ¦ | jour | nal | ¦ | vois | ture
|
||||
hein | deux | ¦ | mais | on | ¦ | sur | prise | ¦ | la | vabo/vande | ¦ | ya/yo | hourt/yo | ¦ | akhy | nator
|
||||
ci | tron | ¦ | fort | boyard | ¦ | pas | nini/steur | ¦ | tki | la | ¦ | re | pas/veil/tourne | ¦ |
|
||||
|
||||
N'hésitez pas à ouvrir un ticket ou faire une merge-request pour contribuer au projet.
|
||||
|
||||
## Lancer le Bot
|
||||
|
||||
Donner la permission `Read and Write` (ou `Read + Write + Direct Messages` mais aucun DM n'est envoyé) au bot dans `Settings` puis `App permissions`.
|
||||
Donner la permission au minimum `Read and Write` au bot dans `Settings` puis `App permissions`.
|
||||
|
||||
Les codes fourni par l'API de Twitter sont généralements disponible dans la page `Keys and tokens`.
|
||||
|
||||
Détails des variables d'environnement :
|
||||
| Variable | Explication et où elle se trouve
|
||||
----------------|-
|
||||
TOKEN | Token d'accès disponible dans la section `Authentication Tokens` sous la sous-rubrique `Access Token and Secret`
|
||||
TOKEN_SECRET | Token d'accès secret disponible dans la section `Authentication Tokens` sous la sous-rubrique `Access Token and Secret`
|
||||
CONSUMER_KEY | Clé API disponible dans la section `Consumer Keys`
|
||||
CONSUMER_SECRET | Clé secrète API disponible dans la section `Consumer Keys`
|
||||
PSEUDOS | Pseudos du ou des compte.s que vous voulez écouter pour le snipe (a séparer avec une virgule **sans** espaces)
|
||||
WHITELIST | Pseudos des comptes qui ne seront pas touché par le Bot (facultatif, a séparer avec une virgule **sans** espaces, par défaut la liste est vide)
|
||||
VERBOSE | Affiche plus de messages dans la console [False\|True] (facultatif, par défaut sur False)
|
||||
FORCELIST | Force le bot à écouter certains comptes (séparer les comptes avec une virgule **sans** espaces, par défaut la liste est vide)
|
||||
| Variable | Explication et où elle se trouve
|
||||
| --------------- |-
|
||||
| TOKEN | Token d'accès disponible dans la section `Authentication Tokens` sous la sous-rubrique `Access Token and Secret`
|
||||
| TOKEN_SECRET | Token d'accès secret disponible dans la section `Authentication Tokens` sous la sous-rubrique `Access Token and Secret`
|
||||
| CONSUMER_KEY | Clé API disponible dans la section `Consumer Keys` sous la sous-rubrique `API Key and Secret`
|
||||
| CONSUMER_SECRET | Clé secrète API disponible dans la section `Consumer Keys` sous la sous-rubrique `API Key and Secret`
|
||||
| BEARER_TOKEN | Token disponible dans la section `Authentication Tokens` sous la sous-rubrique `Bearer Token`
|
||||
| PSEUDOS | Pseudos du ou des compte.s que vous voulez écouter pour le snipe (a séparer avec une virgule **sans** espaces)
|
||||
| WHITELIST | Pseudos des comptes qui ne seront pas touché par le Bot (facultatif, a séparer avec une virgule **sans** espaces, par défaut la liste est vide)
|
||||
| VERBOSE | Affiche plus de messages dans la console [False\|True] (facultatif, par défaut sur False)
|
||||
| FORCELIST | Force le bot à écouter certains comptes (séparer les comptes avec une virgule **sans** espaces, par défaut la liste est vide)
|
||||
|
||||
### En local
|
||||
|
||||
|
@ -46,10 +39,11 @@ docker build https://git.kennel.ml/Anri/feurBot.git#main --tag feurbot:main && \
|
|||
docker run -d \
|
||||
--name="feurBot" \
|
||||
feurbot:main \
|
||||
--TOKEN="" \
|
||||
--TOKEN_SECRET="" \
|
||||
--CONSUMER_KEY="" \
|
||||
--CONSUMER_SECRET="" \
|
||||
--BEARER_TOKEN="" \
|
||||
--TOKEN="" \
|
||||
--TOKEN_SECRET="" \
|
||||
--PSEUDOS=""
|
||||
```
|
||||
Ou avec un `docker-compose.yml` :
|
||||
|
@ -60,10 +54,55 @@ services:
|
|||
build: https://git.kennel.ml/Anri/feurBot.git#main
|
||||
container_name: feurBot
|
||||
environment:
|
||||
- TOKEN=
|
||||
- TOKEN_SECRET=
|
||||
- CONSUMER_KEY=
|
||||
- CONSUMER_SECRET=
|
||||
- BEARER_TOKEN=
|
||||
- TOKEN=
|
||||
- TOKEN_SECRET=
|
||||
- PSEUDOS=
|
||||
restart: unless-stopped
|
||||
```
|
||||
|
||||
## Liste des mots
|
||||
Certains mots peuvent servir de déclencheur sans être dans la liste,
|
||||
example : `aussi` n'est pas dans la liste, mais en retirant `aus`,
|
||||
on obtient `si`, qui est dans la liste.
|
||||
|
||||
| Mot | Réponse |
|
||||
| ----------------------- | ----------------------------- |
|
||||
| quoi | feur/feuse |
|
||||
| oui | stiti/fi |
|
||||
| non | bril |
|
||||
| nan | cy |
|
||||
| hein | deux/bécile |
|
||||
| ci | tron/prine |
|
||||
| con | combre/gelé/pas |
|
||||
| ok | sur glace |
|
||||
| ouais | stern |
|
||||
| comment | tateur/tatrice/dant Cousteau |
|
||||
| mais | on |
|
||||
| fort | boyard |
|
||||
| coup | teau |
|
||||
| ça | pristi/von/perlipopette |
|
||||
| bon | jour/soir (dépend de l'heure) |
|
||||
| qui | wi/mono |
|
||||
| sur | prise |
|
||||
| pas | nini/steur/trimoine/té/stis |
|
||||
| ka | pitaine/pitulation |
|
||||
| fais | rtile |
|
||||
| tant (ou autre syntaxe) | gente/tation |
|
||||
| et | eint/ain |
|
||||
| la | vabo/vande |
|
||||
| tki | la |
|
||||
| moi | tié/sson/sissure |
|
||||
| toi | lette/ture |
|
||||
| top | inambour |
|
||||
| jour | nal |
|
||||
| ya | hourt |
|
||||
| yo | yo/ghourt |
|
||||
| ni | cotine |
|
||||
| re | pas/veil/tourne |
|
||||
| quand | dide/tal/didat |
|
||||
| sol | itaire |
|
||||
| vois | ture |
|
||||
| akhy | nator |
|
||||
|
|
595
main.py
595
main.py
|
@ -1,48 +1,51 @@
|
|||
from dotenv import load_dotenv
|
||||
from os import environ
|
||||
from tweepy import OAuth1UserHandler, API, Stream
|
||||
from re import sub, findall
|
||||
from random import choice
|
||||
from datetime import datetime
|
||||
from pytz import timezone
|
||||
from queue import Queue
|
||||
from json import loads
|
||||
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
|
||||
load_dotenv() # load .env file
|
||||
for var in variables:
|
||||
try:
|
||||
if var == "VERBOSE": # check is VERBOSE is set
|
||||
if var == "VERBOSE": # check is VERBOSE is set
|
||||
try:
|
||||
res = bool(environ[var])
|
||||
if environ[var].lower() == "true":
|
||||
res = True
|
||||
except:
|
||||
res = False # if not its False
|
||||
elif var == "WHITELIST": # check if WHITELIST is set
|
||||
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
|
||||
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
|
||||
res = [] # if not its an empty list
|
||||
else:
|
||||
res = environ[var]
|
||||
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
|
||||
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)
|
||||
return keys
|
||||
|
||||
def cleanTweet(tweet: str) -> str:
|
||||
|
||||
def cleanTweet(tweet: str) -> str | None:
|
||||
"""Remove all unwanted elements from the tweet"""
|
||||
# Convert to lower case
|
||||
tweet = tweet.lower()
|
||||
tweet = tweet.lower()
|
||||
# Remove URLs
|
||||
tweet = sub(r"(https?:\/\/\S+|www.\S+)", " ", tweet)
|
||||
# Check all hashtags
|
||||
|
@ -53,7 +56,7 @@ def cleanTweet(tweet: str) -> str:
|
|||
tweet = sub(r"#\S+", " ", tweet)
|
||||
else:
|
||||
# Too much hashtags in the tweet -> so ignore it
|
||||
return ""
|
||||
return None
|
||||
# Remove usernames
|
||||
tweet = sub(r"@\S+", " ", tweet)
|
||||
# Remove everything who isn't a letter/number/space
|
||||
|
@ -63,196 +66,173 @@ def cleanTweet(tweet: str) -> str:
|
|||
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)
|
||||
# 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(Stream):
|
||||
|
||||
class Listener(StreamingClient):
|
||||
"""Watch for tweets that match criteria in real-time"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
consumer_key,
|
||||
consumer_secret,
|
||||
access_token,
|
||||
access_token_secret,
|
||||
api: API = None,
|
||||
users: list = None,
|
||||
forcelist: list = None,
|
||||
q = Queue()
|
||||
bearer_token,
|
||||
client: Client,
|
||||
):
|
||||
super(Listener, self).__init__(consumer_key, consumer_secret, access_token, access_token_secret)
|
||||
self.q = q
|
||||
self.api = api
|
||||
self.accounts = [users, forcelist]
|
||||
self.listOfFriendsID = getFriendsID(api, users) + getIDs(api, forcelist)
|
||||
super(Listener, self).__init__(bearer_token, wait_on_rate_limit=True)
|
||||
self.client = client
|
||||
self.cache = {}
|
||||
|
||||
def on_connect(self):
|
||||
if self.accounts[1] == []:
|
||||
forcelist = "Aucun"
|
||||
else:
|
||||
forcelist = f"@{', @'.join(self.accounts[1])}"
|
||||
print(f"Début du scroll sur Twitter avec les abonnements de @{', @'.join(self.accounts[0])} et ces comptes en plus : {forcelist} comme timeline...")
|
||||
print(f"Début du scroll sur Twitter...")
|
||||
|
||||
def on_disconnect_message(notice):
|
||||
notice = notice["disconnect"]
|
||||
print(f"Déconnexion (code {notice['code']}).", end = " ")
|
||||
if len(notice["reason"]) > 0:
|
||||
print(f"Raison : {notice['reason']}")
|
||||
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
|
||||
|
||||
def on_status(self, status):
|
||||
json = status._json
|
||||
# Verify the author of the tweet
|
||||
if json["user"]["id"] in self.listOfFriendsID and json["user"]["screen_name"] not in keys["WHITELIST"]:
|
||||
# Verify the age of the tweet
|
||||
if seniority(json["created_at"]):
|
||||
# Verify if the tweet isn't a retweet
|
||||
if not hasattr(status, "retweeted_status"):
|
||||
# Fetch the tweet
|
||||
if "extended_tweet" in json:
|
||||
tweet = cleanTweet(status.extended_tweet["full_text"])
|
||||
else:
|
||||
tweet = cleanTweet(status.text)
|
||||
# Fetch the last word of the tweet
|
||||
lastWord = tweet.split()[-1:][0]
|
||||
if keys["VERBOSE"]:
|
||||
infoLastWord = f"dernier mot : \"{lastWord}\"" if len(lastWord) > 0 else "tweet ignoré car trop de hashtags"
|
||||
print(f"Tweet trouvé de {json['user']['screen_name']} ({infoLastWord})...", end = " ")
|
||||
# Check if the last word found is a supported word
|
||||
if lastWord in universalBase:
|
||||
answer = None
|
||||
# Return the user
|
||||
return self.cache[uid]
|
||||
|
||||
# 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 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.api.update_status(status = choice(answer), in_reply_to_status_id = json["id"], auto_populate_reply_metadata = True)
|
||||
print(f"{json['user']['screen_name']} s'est fait {answer[0]} !")
|
||||
except Exception as error:
|
||||
error = loads(error.response.text)["errors"][0]
|
||||
# https://developer.twitter.com/en/support/twitter-api/error-troubleshooting
|
||||
show_error = True
|
||||
if error["code"] == 385:
|
||||
error["message"] = f"Tweet supprimé ou auteur ({json['user']['screen_name']}) en privé/bloqué."
|
||||
show_error = False
|
||||
|
||||
# Show error only if relevant, always in verbose
|
||||
if show_error or keys["VERBOSE"]:
|
||||
print(f"{errorMessage[:-2]} ({error['code']}) ! {error['message']}")
|
||||
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 and 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("Annulation car le dernier mot n'est pas intéressant.")
|
||||
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
|
||||
author = self._get_user(tweet.author_id)
|
||||
# Clean the tweet
|
||||
lastWord = cleanTweet(tweet.text)
|
||||
|
||||
def do_stuff(self):
|
||||
"""Loop for the Listener"""
|
||||
while True:
|
||||
self.q.get()
|
||||
self.q.task_done()
|
||||
# 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 = " ")
|
||||
if status_code == 413:
|
||||
if keys["VERBOSE"]:
|
||||
print("La liste des mots est trop longue (triggerWords).")
|
||||
elif status_code == 420:
|
||||
if keys["VERBOSE"]:
|
||||
print("Déconnecter du flux.")
|
||||
else:
|
||||
print("\n")
|
||||
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':
|
||||
if word[0] == "s" or word[0] == "t":
|
||||
word = word[1:]
|
||||
|
||||
# Random format from the base answer
|
||||
return createBaseAnswers(word)
|
||||
|
||||
def getFriendsID(api: API, users: list[str]) -> list:
|
||||
|
||||
def getFriends(client: Client, users: list[str]) -> list:
|
||||
"""Get all friends of choosen users"""
|
||||
liste = []
|
||||
friends_list = []
|
||||
# Get IDs of the user's friends
|
||||
for user in users:
|
||||
liste.extend(api.get_friend_ids(screen_name=user))
|
||||
return list(set(liste))
|
||||
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 getIDs(api: API, users: list[str]) -> list:
|
||||
"""Get all the ID of users"""
|
||||
liste = []
|
||||
# Get IDs of the users
|
||||
for user in users:
|
||||
liste.append(api.get_user(screen_name=user)._json["id"])
|
||||
return list(set(liste))
|
||||
|
||||
def seniority(date: str) -> bool:
|
||||
"""Return True only if the given string date is less than one day old"""
|
||||
# Convert string format to datetime format
|
||||
datetimeObject = datetime.strptime(date, "%a %b %d %H:%M:%S +0000 %Y")
|
||||
# Twitter give us an UTC time
|
||||
datetimeObject = datetimeObject.replace(tzinfo = timezone("UTC"))
|
||||
# time now in UTC minus the time we got to get the age of the date
|
||||
age = datetime.now(timezone("UTC")) - datetimeObject
|
||||
# False if older than a day, else True
|
||||
return False if age.days >= 1 else True
|
||||
|
||||
def generateWords(array: list[str]) -> list:
|
||||
"""
|
||||
Retrieves all possible combinations for the given list and returns the result as a list
|
||||
|
||||
This is used for the filter in the stream (before calling the Listener::on_status)
|
||||
"""
|
||||
quoiListe = []
|
||||
|
||||
for text in array:
|
||||
# Add all combinations
|
||||
# Example for 'oui': ['OUI', 'OUi', 'OuI', 'Oui', 'oUI', 'oUi', 'ouI', 'oui']
|
||||
#
|
||||
# -> Depends on: from itertools import product
|
||||
# -> Problem : Create a too long list (+1000 words, max is 400)
|
||||
# -> Cf. https://developer.twitter.com/en/docs/twitter-api/v1/tweets/filter-realtime/overview
|
||||
#
|
||||
# quoiListe.extend(list(map(''.join, product(*zip(text.upper(), text.lower())))))
|
||||
|
||||
if text.lower() not in quoiListe:
|
||||
# Word in lowercase
|
||||
quoiListe.append(text.lower())
|
||||
if text.upper() not in quoiListe:
|
||||
# Word in uppercase
|
||||
quoiListe.append(text.upper())
|
||||
if text.capitalize() not in quoiListe:
|
||||
# Word capitalized
|
||||
quoiListe.append(text.capitalize())
|
||||
|
||||
return quoiListe
|
||||
|
||||
def createBaseTrigger(lists: list[list]) -> list:
|
||||
"""Merges all given lists into one"""
|
||||
|
@ -261,6 +241,7 @@ def createBaseTrigger(lists: list[list]) -> list:
|
|||
listing.extend(liste)
|
||||
return list(set(listing))
|
||||
|
||||
|
||||
def createBaseAnswers(word: str) -> list:
|
||||
"""Generates default answers for a given word"""
|
||||
irritating_word = [
|
||||
|
@ -270,6 +251,7 @@ def createBaseAnswers(word: str) -> list:
|
|||
]
|
||||
|
||||
return [
|
||||
# Assuming the first element of this list is always word, don't change it
|
||||
word,
|
||||
f"({word})",
|
||||
word.upper(),
|
||||
|
@ -277,47 +259,152 @@ def createBaseAnswers(word: str) -> list:
|
|||
f"{word}...",
|
||||
]
|
||||
|
||||
def start():
|
||||
"""Start the bot"""
|
||||
auth = OAuth1UserHandler(keys["CONSUMER_KEY"], keys["CONSUMER_SECRET"])
|
||||
auth.set_access_token(keys["TOKEN"], keys["TOKEN_SECRET"])
|
||||
|
||||
api = API(auth)
|
||||
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:
|
||||
api.verify_credentials()
|
||||
print(f"Authentification réussie en tant que", end = " ")
|
||||
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)
|
||||
print(f"@{api.verify_credentials().screen_name}.")
|
||||
|
||||
if keys['WHITELIST'] == []:
|
||||
whitelist = "Aucun"
|
||||
else:
|
||||
whitelist = f"@{', @'.join(keys['WHITELIST'])}"
|
||||
print(f"Liste des comptes ignorés : {whitelist}.")
|
||||
return client
|
||||
|
||||
stream = Listener(
|
||||
consumer_key=keys["CONSUMER_KEY"],
|
||||
consumer_secret=keys["CONSUMER_SECRET"],
|
||||
access_token=keys["TOKEN"],
|
||||
access_token_secret=keys["TOKEN_SECRET"],
|
||||
api=api,
|
||||
users=keys["PSEUDOS"],
|
||||
forcelist=keys["FORCELIST"]
|
||||
|
||||
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"],
|
||||
)
|
||||
stream.filter(track = triggerWords, languages = ["fr"], stall_warnings = True, threaded = True)
|
||||
|
||||
# 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 Access Token and Secret sub-heading.
|
||||
TOKEN_SECRET is the Access Token Secret available in the Authentication Tokens section under Access Token and Secret sub-heading.
|
||||
CONSUMER_KEY is the API Key available in the Consumer Keys section.
|
||||
CONSUMER_SECRET is the API Secret Key available in the Consumer Keys section.
|
||||
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
|
||||
--
|
||||
PSEUDO is the PSEUDO of the account you want to listen to snipe.
|
||||
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 !"
|
||||
|
@ -365,62 +452,62 @@ if __name__ == "__main__":
|
|||
# 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"
|
||||
],
|
||||
+ 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"),
|
||||
+ createBaseAnswers("fi"),
|
||||
|
||||
"non": createBaseAnswers("bril"),
|
||||
|
||||
"nan": createBaseAnswers("cy"),
|
||||
|
||||
"hein": createBaseAnswers("deux")
|
||||
+ createBaseAnswers("bécile")
|
||||
+ [
|
||||
"2"
|
||||
],
|
||||
+ createBaseAnswers("bécile")
|
||||
+ [
|
||||
"2"
|
||||
],
|
||||
|
||||
"ci": createBaseAnswers("tron")
|
||||
+ createBaseAnswers("prine"),
|
||||
+ createBaseAnswers("prine"),
|
||||
|
||||
"con": createBaseAnswers("combre")
|
||||
+ createBaseAnswers("gelé")
|
||||
+ createBaseAnswers("pas"),
|
||||
+ createBaseAnswers("gelé")
|
||||
+ createBaseAnswers("pas"),
|
||||
|
||||
"ok": createBaseAnswers("sur glace"),
|
||||
|
||||
"ouais": createBaseAnswers("stern"),
|
||||
|
||||
"comment": createBaseAnswers("tateur")
|
||||
+ createBaseAnswers("tatrice")
|
||||
+ createBaseAnswers("dant Cousteau"),
|
||||
+ createBaseAnswers("tatrice")
|
||||
+ createBaseAnswers("dant Cousteau"),
|
||||
|
||||
"mais": createBaseAnswers("on")
|
||||
+ [
|
||||
"on (-dulation)"
|
||||
],
|
||||
+ [
|
||||
"on (-dulation)"
|
||||
],
|
||||
|
||||
"fort": createBaseAnswers("boyard")
|
||||
+ [
|
||||
"boyard (-ennes)"
|
||||
],
|
||||
+ [
|
||||
"boyard (-ennes)"
|
||||
],
|
||||
|
||||
"coup": createBaseAnswers("teau"),
|
||||
|
||||
"ça": createBaseAnswers("perlipopette")
|
||||
+ createBaseAnswers("von")
|
||||
+ createBaseAnswers("pristi")
|
||||
+ [
|
||||
"pristi (-gnasse)"
|
||||
],
|
||||
+ createBaseAnswers("von")
|
||||
+ createBaseAnswers("pristi")
|
||||
+ [
|
||||
"pristi (-gnasse)"
|
||||
],
|
||||
|
||||
"bon": [
|
||||
createBaseAnswers("jour"),
|
||||
|
@ -428,38 +515,38 @@ if __name__ == "__main__":
|
|||
],
|
||||
|
||||
"qui": createBaseAnswers("wi")
|
||||
+ createBaseAnswers("mono"),
|
||||
+ createBaseAnswers("mono"),
|
||||
|
||||
"sur": createBaseAnswers("prise"),
|
||||
|
||||
"pas": createBaseAnswers("nini")
|
||||
+ createBaseAnswers("steur")
|
||||
+ createBaseAnswers("trimoine")
|
||||
+ createBaseAnswers("té")
|
||||
+ createBaseAnswers("stis"),
|
||||
+ createBaseAnswers("steur")
|
||||
+ createBaseAnswers("trimoine")
|
||||
+ createBaseAnswers("té")
|
||||
+ createBaseAnswers("stis"),
|
||||
|
||||
"ka": createBaseAnswers("pitaine")
|
||||
+ createBaseAnswers("pitulation"),
|
||||
+ createBaseAnswers("pitulation"),
|
||||
|
||||
"fais": createBaseAnswers("rtile"),
|
||||
|
||||
"tant": createBaseAnswers("gente")
|
||||
+ createBaseAnswers("tation"),
|
||||
+ createBaseAnswers("tation"),
|
||||
|
||||
"et": createBaseAnswers("eint")
|
||||
+ createBaseAnswers("ain"),
|
||||
+ createBaseAnswers("ain"),
|
||||
|
||||
"la": createBaseAnswers("vabo")
|
||||
+ createBaseAnswers("vande"),
|
||||
+ createBaseAnswers("vande"),
|
||||
|
||||
"tki": createBaseAnswers("la"),
|
||||
|
||||
"moi": createBaseAnswers("tié")
|
||||
+ createBaseAnswers("sson")
|
||||
+ createBaseAnswers("sissure"),
|
||||
+ createBaseAnswers("sson")
|
||||
+ createBaseAnswers("sissure"),
|
||||
|
||||
"toi": createBaseAnswers("lette")
|
||||
+ createBaseAnswers("ture"),
|
||||
+ createBaseAnswers("ture"),
|
||||
|
||||
"top": createBaseAnswers("inambour"),
|
||||
|
||||
|
@ -468,17 +555,17 @@ if __name__ == "__main__":
|
|||
"ya": createBaseAnswers("hourt"),
|
||||
|
||||
"yo": createBaseAnswers("ghourt")
|
||||
+ createBaseAnswers("yo"),
|
||||
+ createBaseAnswers("yo"),
|
||||
|
||||
"ni": createBaseAnswers("cotine"),
|
||||
|
||||
"re": createBaseAnswers("pas")
|
||||
+ createBaseAnswers("veil")
|
||||
+ createBaseAnswers("tourne"),
|
||||
+ createBaseAnswers("veil")
|
||||
+ createBaseAnswers("tourne"),
|
||||
|
||||
"quand": createBaseAnswers("dide")
|
||||
+ createBaseAnswers("tal")
|
||||
+ createBaseAnswers("didat"),
|
||||
+ createBaseAnswers("tal")
|
||||
+ createBaseAnswers("didat"),
|
||||
|
||||
"sol": createBaseAnswers("itaire"),
|
||||
|
||||
|
@ -490,11 +577,9 @@ if __name__ == "__main__":
|
|||
# List of all the trigger words
|
||||
universalBase = createBaseTrigger(list(base.values()))
|
||||
|
||||
# List of all the triggers words's variations
|
||||
triggerWords = generateWords(universalBase)
|
||||
|
||||
# Loading environment variables
|
||||
keys = load(["TOKEN", "TOKEN_SECRET", "CONSUMER_KEY", "CONSUMER_SECRET", "PSEUDOS", "VERBOSE", "WHITELIST", "FORCELIST"])
|
||||
keys = load(["TOKEN", "TOKEN_SECRET", "CONSUMER_KEY", "CONSUMER_SECRET",
|
||||
"BEARER_TOKEN", "PSEUDOS", "VERBOSE", "WHITELIST", "FORCELIST"])
|
||||
|
||||
# Start the bot
|
||||
start()
|
||||
|
|
Reference in a new issue