Compare commits

...

47 commits

Author SHA1 Message Date
61a1d52bf8
fix thread detection 2022-08-11 14:18:06 +02:00
83956d2d47
fix private mode account safety 2022-08-11 14:05:17 +02:00
30be0904d0
check condition 2022-08-11 13:58:29 +02:00
c0c00a0759
fix rules sometimes updating without changes 2022-08-11 13:52:20 +02:00
dc28c29c6f
fix private account detection 2022-08-11 13:50:42 +02:00
945844712f
remove threaded 2022-08-11 12:38:29 +02:00
bda4d8ccd4
split in multiples lines 2022-08-11 12:27:44 +02:00
43c962c3e0
add verboses 2022-08-11 12:25:39 +02:00
805ab1bb57
Don't answer on private accounts 2022-08-11 12:20:44 +02:00
b83944959b
fix verbose bool 2022-08-09 02:54:16 +02:00
eda6804d38
add some verbose 2022-08-09 02:49:07 +02:00
8c57391b54
fix nonetype 2022-08-09 02:35:42 +02:00
f351e477e3
fix nonetype 2022-08-09 02:32:34 +02:00
6eab4a41b3
update comment 2022-08-09 02:20:15 +02:00
b1c5abe1f9
reduce spam of twitter API 2022-08-09 02:17:28 +02:00
68144b56ca
fr tweets 2022-08-09 02:03:02 +02:00
a1effc03c7
reorder following twitter website 2022-08-09 00:35:14 +02:00
c8db7f6436
add comment 2022-08-08 09:59:20 +02:00
d23b9b57c3
fix rules 2022-08-08 09:58:23 +02:00
30615db063
clean rules first 2022-08-08 09:48:25 +02:00
1eb5f84cc0
fix rules 2022-08-08 09:41:09 +02:00
f595dcd83d
better handling of invalid tweets 2022-08-08 09:36:10 +02:00
dcda3d107e
handle request errors 2022-08-08 09:09:21 +02:00
3b5c78f539
fix errror handling 2022-08-08 09:04:17 +02:00
a0032708af
update comment padding 2022-08-08 02:51:58 +02:00
95f2fe551d
fix newline 2022-08-08 02:49:39 +02:00
2d8ef4e0ce
update words list 2022-08-08 02:48:54 +02:00
7fad7d09c6
move word list to the bottom 2022-08-08 02:44:48 +02:00
11f5180346
add BEARER_TOKEN in docker examples 2022-08-08 02:36:40 +02:00
2110a4d961
enable tweets 2022-08-08 02:32:28 +02:00
e347f7fcb7
fix bug with hashtag tweets 2022-08-08 02:29:40 +02:00
d1ee920d7a
newlines 2022-08-08 01:59:42 +02:00
9300961afa
fix bug in verbose 2022-08-08 01:54:50 +02:00
2513236f6c
* fix crash
* complete migration
2022-08-08 01:20:35 +02:00
a40866e736
don't check the tweet date and doublecheck for retweets 2022-08-07 22:06:41 +02:00
52525cab51
follow pep8 2022-08-07 21:55:48 +02:00
36792c1949
order imports 2022-08-07 21:52:40 +02:00
46037106e6
convert stream entirely to the API v2 2022-08-07 21:50:59 +02:00
ed6e72270c
rule creator 2022-08-07 21:14:23 +02:00
f99ab71369
track on user and not on keyword basis 2022-08-07 20:16:07 +02:00
4e437ad675
add problematic todo 2022-08-07 17:32:42 +02:00
c0b5c168a7
update lookup functions 2022-08-07 17:24:42 +02:00
a2e118c38f
update permission 2022-08-07 17:14:35 +02:00
86853437a9
update comments 2022-08-07 17:05:47 +02:00
0874d5211d
wip: add bearer token, migrate from Stream to StreamingClient, add some TODOs to complete 2022-08-07 16:47:10 +02:00
756016ae8d
more precise readme, add bearer token 2022-08-07 16:46:06 +02:00
78bfe13fa3
add bearer token 2022-08-07 16:45:55 +02:00
3 changed files with 407 additions and 282 deletions

View file

@ -1,5 +1,6 @@
TOKEN=
TOKEN_SECRET=
CONSUMER_KEY=
CONSUMER_SECRET=
BEARER_TOKEN=
TOKEN=
TOKEN_SECRET=
PSEUDO=

View file

@ -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
View file

@ -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("")
+ createBaseAnswers("stis"),
+ createBaseAnswers("steur")
+ createBaseAnswers("trimoine")
+ createBaseAnswers("")
+ 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()