118 lines
3.6 KiB
Python
118 lines
3.6 KiB
Python
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")
|