From 0d22af2470a2e63488ce7d7550cb3acdea6ca599 Mon Sep 17 00:00:00 2001 From: Piotr Dec Date: Wed, 30 Apr 2025 00:12:16 +0200 Subject: [PATCH] exporter-ng: argparse removed, reformatted --- exporter-ng.py | 80 ++++++++++++-------------------------------------- 1 file changed, 19 insertions(+), 61 deletions(-) diff --git a/exporter-ng.py b/exporter-ng.py index 16a7ecc..4d00ab5 100644 --- a/exporter-ng.py +++ b/exporter-ng.py @@ -1,16 +1,15 @@ +import json import os import sys -import requests -import json -from timeit import default_timer +from dataclasses import dataclass from datetime import datetime -import argparse +from time import sleep +from typing import List, Optional, Dict, Any + +import requests from dotenv import load_dotenv from pathvalidate import sanitize_filename -from time import sleep -from dataclasses import dataclass -from typing import List, Optional, Dict, Any, Iterator, Tuple -from abc import ABC, abstractmethod + @dataclass class SlackConfig: @@ -30,8 +29,10 @@ class SlackConfig: except KeyError: raise ValueError("Brak SLACK_USER_TOKEN w zmiennych środowiskowych") + class SlackAPI: """Klasa do komunikacji z API Slacka""" + def __init__(self, config: SlackConfig): self.config = config self.headers = {"Authorization": f"Bearer {config.user_token}"} @@ -130,6 +131,7 @@ class SlackAPI: params["channel"] = channel return self.paginated_get("files.list", params, "files") + @dataclass class SlackUser: """Reprezentacja użytkownika Slack""" @@ -189,8 +191,10 @@ class SlackUser: return ", ".join(parts) + class SlackChannel: """Reprezentacja kanału Slack""" + def __init__(self, data: Dict): self.id = data["id"] self.name = data.get("name", "") @@ -231,6 +235,7 @@ class SlackChannel: return " ".join(parts) + @dataclass class SlackFile: """Reprezentacja pliku Slack""" @@ -246,8 +251,10 @@ class SlackFile: url_private=data.get("url_private", "") ) + class SlackMessage: """Reprezentacja wiadomości Slack""" + def __init__(self, data: Dict, users: Dict[str, SlackUser]): self.timestamp = float(data["ts"]) self.text = data.get("text", "[no message content]") @@ -295,8 +302,10 @@ class SlackMessage: return message + "\n\n" + "*" * 24 + "\n\n" + class SlackExporter: """Główna klasa eksportera""" + def __init__(self, config: SlackConfig, output_dir: str): self.api = SlackAPI(config) self.timestamp = datetime.now().strftime("%Y-%m-%d_%H%M%S") @@ -397,14 +406,14 @@ class SlackExporter: data = header + "".join(msg.format(True) for msg in messages) self._save_data(data, f"channel-replies_{channel_id}", as_json) - + def export_channel_files(self, channel_id: Optional[str] = None): """Eksportuje pliki w kanale""" files = [SlackFile.from_dict(f) for f in self.api.get_files(channel_id)] for file in files: filename = f"{file.id}-{sanitize_filename(file.name)}" self.download_file(filename, file.url_private) - + def download_file(self, filename: str, url: str, attempts: int = 10) -> bool: if attempts == 0: return False @@ -422,58 +431,7 @@ class SlackExporter: print(f"Error downloading file {filename}: {e}. {attempts} attempts left.") return self.download_file(filename, url, attempts - 1) -def main(): - parser = argparse.ArgumentParser(description="Eksporter danych ze Slacka") - parser.add_argument("-o", help="Katalog wyjściowy (jeśli pusty, wyświetla na stdout)") - parser.add_argument("--lc", action="store_true", help="Lista wszystkich konwersacji") - parser.add_argument("--lu", action="store_true", help="Lista wszystkich użytkowników") - parser.add_argument("--json", action="store_true", help="Wynik w formacie JSON") - parser.add_argument("-c", action="store_true", help="Historia wszystkich dostępnych konwersacji") - parser.add_argument("--ch", help="Z -c, ogranicza eksport do podanego ID kanału") - parser.add_argument("--fr", help="Z -c, timestamp początku zakresu (Unix)") - parser.add_argument("--to", help="Z -c, timestamp końca zakresu (Unix)") - parser.add_argument("-r", action="store_true", help="Pobierz wątki ze wszystkich konwersacji") - parser.add_argument("--files", action="store_true", help="Pobierz wszystkie pliki") - - args = parser.parse_args() - - if args.files and not args.o: - print("Opcja --files wymaga określenia katalogu wyjściowego (-o)") - sys.exit(1) - - try: - config = SlackConfig.from_env() - exporter = SlackExporter(config) - - if args.o: - exporter.set_output_dir(args.o) - - exporter.load_users() - exporter.load_channels() - - if args.lc: - exporter.export_channel_list(args.json) - - if args.lu: - exporter.export_user_list(args.json) - - if args.c or args.r: - channel_ids = [args.ch] if args.ch else [ch.id for ch in exporter.channels] - for channel_id in channel_ids: - if args.c: - exporter.export_channel_history(channel_id, args.fr, args.to, args.json) - if args.r: - exporter.export_channel_replies(channel_id, args.fr, args.to, args.json) - - if args.files and args.o: - # TODO: Implementacja pobierania plików - print("Funkcja pobierania plików jeszcze nie zaimplementowana") - - except Exception as e: - print(f"Błąd: {e}") - sys.exit(1) if __name__ == "__main__": - # main() config = SlackConfig.from_env() exporter = SlackExporter(config, 'out')