Aus irgendeinem Grund habe ich ungefähr ein Jahr lang einen Nachrichtenverlauf von slack erhalten, daher werde ich schreiben, wie er in Python implementiert wird. Ich habe die abgerufene Nachricht reformiert, um die Analyse zu vereinfachen, kann sie jedoch nicht vollständig veröffentlichen, da sie so schlecht ist. Ich würde gerne wieder einen Artikel schreiben, wenn es einen Teil gibt, der veröffentlicht werden kann.
Python hat slackapi / python-slackclient, aber ich habe es diesmal nicht verwendet. Wenn Sie wissen möchten, wie Sie Python-Slackclient implementieren, empfehle ich, einen anderen Artikel als diesen zu lesen.
Client
Das Slack-Token ist eine Instanzvariable, sodass es aus der Umgebungsvariablen abgerufen oder direkt in das Hauptskript geschrieben werden kann. Wenn Sie pipenv verwenden, wird automatisch ".env" angezeigt, sodass der Standardwert der in der Umgebungsvariablen festgelegte Wert ist. Es ist eine Implementierung, die von meiner Entwicklungsumgebung abhängt, aber ich habe sie auch mit Fällen kompatibel gemacht, in denen pipenv nicht verwendet wird (ich möchte sie nicht in der Umgebungsvariablen festlegen).
Das Argument der Anforderungsfunktion enthält method: BaseSlackMethod. Dies liegt jedoch daran, dass im Fall von slack jeder API-Endpunkt als Methode bezeichnet wird. Ich werde die Implementierung von BaseSlackMethod später erklären, aber ich habe BaseSlackMethod zu einer Basisklasse gemacht, damit ich die Anzahl der Klassen für die Methode erhöhen kann. Dadurch wurden die Anforderungsparameter im Code verwaltbar. Sie können sich die Mühe sparen, einzeln zur Referenz zu wechseln. Du hast es geschafft!
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]:
"""API-Anfrage an 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("Datenerfassungsfehler durch Durchhang")
raise err
else:
logger.info("Datenerfassung von Slack abgeschlossen")
return res.json()
Die API-Methode zum Abrufen des Nachrichtenverlaufs lautet gespräche.historie. Lesen Sie die Referenz, um weitere Informationen zu den Anforderungsparametern zu erhalten. Wenn Sie die Parameter wie unten gezeigt in den Code einfügen, ist es einfacher, die Parameter zu verstehen, die von der Methode angefordert werden können. Der Code kann auch eine gute Referenz mit den entsprechenden Kommentaren sein. Vorerst werde ich die wichtigen Parameter für die Erfassung der Geschichte für ein Jahr erläutern. Sie sind "Cursor" und "älteste". Der Cursor ist das nächste Token zum rekursiven Abrufen des Verlaufs. Das älteste gibt das Startdatum und die Uhrzeit der Geschichte als allgemeine Bedeutung an. Der zu beachtende Punkt ist, dass der älteste als Unix-Zeitstempel angegeben werden kann.
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: ist die Basisklasse.
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)
Ich möchte den Verlauf für ein Jahr abrufen, daher verwende ich die Formel "datetime.now () --timedelta (Tage = 365)", um Datum und Uhrzeit vor einem Jahr zu berechnen. Timedelta ist praktisch, da Sie Datum und Uhrzeit ein Jahr später berechnen können, indem Sie Minus in Plus ändern. Danke ~~: bete: Dieses Mal habe ich eine einfache while-Schleife gewählt, weil ich die Geschichte für ein weiteres Jahr rekursiv erfassen muss. Es ist ein beschissenes Einweg-Skript, daher musste ich die if-Anweisung nicht sorgfältig implementieren, um zu sehen, ob next_cursor vorhanden ist, aber ich mochte es nicht, mit KeyError zu enden, also habe ich das getan.
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()
Als ich ein Jahr lang die Historie eines Kanals erhielt, wurden ungefähr 200 Dateien mit mehr als 2000 Zeilen pro Datei erstellt. Erschreckend: schreien:
Reference