from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser from dataclasses import dataclass import logging import logging.config from logging import Logger from pathlib import Path from typing import Dict, List from platformdirs import PlatformDirs import requests import yaml parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter) parser.add_argument("--verbose", "-v", action="count", default=0) parser.add_argument("--config", "-c", default="kontor-api") parser.add_argument("--dry-run", "-m", action="store_true") parser.add_argument("--cleanup", "-d", action="store_true") args = parser.parse_args() @dataclass class Login: email: str password: str @dataclass class Server: name: str url: str token: str token_type: str def login(self, login: Login, log: Logger): if not self.token: log.info("Call login first") login_url = f"{self.url}/login" login_data = {} login_data["email"] = login.email login_data["password"] = login.password response = requests.post(login_url, json=login_data) status = response.status_code log.info(f"Status: {status}") if status != 200: log.fatal("authentication failed") return data = response.json() log.debug(f"got data: {data}") token = data["access_token"] token_type = data["token_type"] self.token = str(token) self.token_type = str(token_type) def request(self, log: Logger): url: str = f"{self.url}/api/media/files" headers: Dict[str, str] = {"Authorization": f"Bearer {self.token}"} response = requests.get(url, headers=headers) log.info(f"Status: {response.status_code}") data = response.json() return data @dataclass class ApiConfig: login: Login server: List[Server] def get_logger(level, config: str): dirs = PlatformDirs(config) logging_config = Path(dirs.user_config_dir, "logging-config.yaml") log_config = None with open(logging_config, "rt") as f: log_config = yaml.safe_load(f.read()) logging.config.dictConfig(log_config) logger = logging.getLogger("development") if level is not None: match level: case 0: logger.setLevel(logging.CRITICAL) case 1: logger.setLevel(logging.INFO) case 2: logger.setLevel(logging.DEBUG) case _: logger.setLevel(logging.INFO) return logger def get_api_config(log: Logger, config: str) -> ApiConfig: # api_data: Dict[str, Any] = {} dirs = PlatformDirs(config) api_config = Path(dirs.user_config_dir, "api.yaml") with open(api_config, "rt") as f: api_data = yaml.safe_load(f.read()) servers = [Server(**server) for server in api_data['server']] login = Login(**(api_data["login"])) apiConfig = ApiConfig(server=servers, login=login) log.info(apiConfig) if not api_data: log.fatal("API configuration is missing") return apiConfig for server in apiConfig.server: server.login(apiConfig.login, log) with open(api_config, "w") as f: yaml.dump(api_data, f) return apiConfig if __name__== "__main__": logger = get_logger(args.verbose, "kontor") logger.info("kontor.sync started") apiConfig = get_api_config(logger, args.config) for server in apiConfig.server: data = server.request(logger) logger.info(len(data)) logger.info("kontor.sync finished")