Pour une raison quelconque, j'ai reçu un historique des messages de slack pendant environ un an, donc j'écrirai comment l'implémenter en Python. J'ai réformé le message récupéré pour le rendre plus facile à analyser, mais je ne peux pas le publier dans son intégralité car il est si mauvais. Je voudrais réécrire un article s'il y a une partie qui peut être publiée.
Python a slackapi / python-slackclient, mais je ne l'ai pas utilisé cette fois. Si vous voulez savoir comment implémenter en utilisant python-slackclient, je vous recommande de lire autre chose que cet article.
Client
Le jeton de slack est une variable d'instance afin qu'il puisse être obtenu à partir de la variable d'environnement ou écrit directement dans le script principal. Si vous utilisez pipenv, il lira automatiquement .env, donc la valeur par défaut est celle définie dans la variable d'environnement. C'est une implémentation qui dépend de mon environnement de développement, mais je l'ai également rendue compatible avec les cas où pipenv n'est pas utilisé (je ne veux pas le définir dans la variable d'environnement).
Il y a method: BaseSlackMethod dans l'argument de la fonction de requête, mais c'est parce que dans le cas de slack, chaque point de terminaison d'API est appelé method. J'expliquerai l'implémentation de BaseSlackMethod plus tard, mais j'ai fait de BaseSlackMethod une classe de base afin que je puisse augmenter le nombre de classes pour la méthode. Cela rendait les paramètres de requête gérables dans le code. Vous pouvez éviter d'aller à la référence une par une. Tu l'as fait!
src/slack/client.py
import os
from dataclasses import dataclass
from typing import Any, ClassVar, Dict
import requests
from src.log import get_logger
from src.slack.exceptions import SlackRequestError
from src.slack.types import Headers
from src.slack.methods.base import BaseSlackMethod
SLACK_API_TOKEN = os.getenv("SLACK_API_TOKEN", "")
logger = get_logger(__name__)
@dataclass
class SlackClient:
api_url: ClassVar[str] = "https://slack.com/api"
token: str = SLACK_API_TOKEN
def _get_headers(self, headers: Headers) -> Headers:
"""Get headers
Args:
headers (Headers)
Returns:
Headers
"""
final_headers = {
"Content-Type": "application/x-www-form-urlencoded;charset=utf-8",
}
if self.token:
final_headers["Authorization"] = f"Bearer {self.token}"
final_headers.update(headers)
return final_headers
def request(
self, method: BaseSlackMethod, headers: Dict[str, Any] = None,
) -> Dict[Any, Any]:
"""Demande d'API à Slack
Args:
method (BaseSlackMethod)
headers (Dict[str, Any], optional): Defaults to None.
Raises:
SlackRequestError
err
Returns:
Dict[Any, Any]: response body
"""
if not isinstance(headers, dict):
headers = {}
headers = self._get_headers(headers)
url = f"{self.api_url}/{method.endpoint}"
try:
res = requests.get(url, headers=headers, params=method.params)
if res.ok is False:
raise SlackRequestError(res.text)
except Exception as err:
logger.error(err)
logger.error("Échec de l'acquisition de données de Slack")
raise err
else:
logger.info("Acquisition de données terminée depuis Slack")
return res.json()
La méthode API pour obtenir l'historique des messages est conversations.history. Lisez la référence pour plus d'informations sur les paramètres de demande. En déposant les paramètres dans le code comme indiqué ci-dessous, il est plus facile de comprendre les paramètres qui peuvent être demandés par méthode. Le code peut également être une bonne référence avec les commentaires appropriés. Pour le moment, j'expliquerai les paramètres importants pour acquérir l'historique pendant un an. Ce sont «curseur» et «le plus ancien». Le curseur est le prochain jeton pour la récupération récursive de l'historique. le plus ancien spécifie la date et l'heure de début de l'histoire comme signification générale. Le point à noter est que le plus ancien peut être spécifié comme horodatage Unix.
src/slack/methods/conversation.py
import os
from datetime import datetime
from dataclasses import dataclass, asdict
from typing import ClassVar, Optional
from src.slack.types import SlackParams
SLACK_CHANNEL_ID = os.getenv("SLACK_CHANNEL_ID", "")
@dataclass
class ConversationsHistory:
endpoint: ClassVar[str] = "conversations.history"
channel: str = SLACK_CHANNEL_ID
cursor: Optional[str] = None
inclusive: bool = False
limit: int = 100
latest: float = datetime.now().timestamp()
oldest: float = 0
@property
def params(self) -> SlackParams:
self_dict = asdict(self)
if self.cursor is None:
del self_dict["cursor"]
return self_dict
: arrow_down_small: est la classe de base.
src/slack/methods/base.py
from dataclasses import dataclass, asdict
from typing import ClassVar
from src.slack.types import SlackParams
@dataclass
class BaseSlackMethod:
endpoint: ClassVar[str] = ""
@property
def params(self) -> SlackParams:
return asdict(self)
Je veux obtenir l'historique d'un an, j'utilise donc la formule datetime.now () --timedelta (days = 365) pour calculer la date et l'heure il y a un an. Timedelta est pratique car vous pouvez calculer la date et l'heure un an plus tard en remplaçant moins par plus. Merci ~~: priez:
Cette fois, j'ai adopté une boucle while simple car je dois acquérir de manière récursive l'histoire pour une autre année. C'est un script jetable de merde, donc je n'ai pas eu à implémenter l'instruction if avec soin pour voir s'il y avait next_cursor, mais je n'aimais pas finir par KeyError, alors j'ai fait ça.
src/slack/__main__.py
from datetime import datetime, timedelta
from src.utils import save_to_file
from src.slack.client import SlackClient
from src.slack.methods.conversation import ConversationsHistory
def main() -> None:
tmp_oldest = datetime.now() - timedelta(days=365)
oldest = tmp_oldest.timestamp()
method = ConversationsHistory(inclusive=True, oldest=oldest)
client = SlackClient()
count = 1
while True:
res = client.request(method)
save_to_file(res, f"outputs/tests/sample{count}.json")
if (
"response_metadata" in res
and "next_cursor" in res["response_metadata"]
):
method.cursor = res["response_metadata"]["next_cursor"]
count += 1
else:
break
if __name__ == "__main__":
main()
Lorsque j'ai eu l'historique d'un canal pendant un an, environ 200 fichiers ont été créés avec plus de 2000 lignes par fichier. Terrifiant: crier:
Reference