tpeetz/kontor
1
0
Fork 0
Code Issues 6 Pull Requests 2 Actions Packages Projects Releases 2 Wiki Activity

add missing endpoints

Browse Source
This commit is contained in:
Thomas Peetz
2026-05-17 19:52:00 +02:00
parent eeb2f23e71
commit f1abf03b35
17 changed files with 440 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kontor-scripts/Makefile
-15
View File
@@ -1,15 +0,0 @@
clean:
find . -name '*.py[co]' -delete
test:
python -m pytest \
-v \
--cov=kontor \
--cov-report=term \
--cov-report=html:coverage-report \
tests/
docker: clean
docker build -t kontor:latest .
kontor-scripts/api.py
+166
View File
@@ -0,0 +1,166 @@
from dataclasses import dataclass
import logging
import logging.config
from logging import Logger
from pathlib import Path
from typing import Dict, List, Optional
from uuid import UUID
from platformdirs import PlatformDirs
import requests
import yaml
MAPPING: Dict[str, str] = {
"sport": "api/tysc/sports",
"player": "api/tysc/players",
"team": "api/tysc/teams",
"field_position": "api/tysc/positions",
"rooster": "api/tysc/roosters",
"vendor": "api/tysc/vendors",
"card_set": "api/tysc/cardsets",
"card": "api/tysc/cards",
"artist": "api/comics/artists",
"publisher": "api/comics/publishers",
"worktype": "api/comics/worktypes",
"comic": "api/comics/comics",
"volume": "api/comics/volumes",
"story_arc": "api/comics/storyarcs",
"issue": "api/comics/issues",
"issue_work": "api/comics/issueworks",
"article": "",
"bookshelf_publisher": "",
"book": "",
"author": "",
"article_author": "",
"book_author": "",
"media_article": "",
"media_video": "api/media/videos",
"media_file": "api/media/files",
"media_actor": "api/media/actors",
"media_actor_file": "api/media/actorfiles",
"profile": "",
"permission": "",
"assignment": "",
"token": "",
"mail_account": "",
}
@dataclass
class Login:
"""
Dataclass to store login information.
"""
email: str
password: str
@dataclass
class Server:
"""
Dataclass to represent a Kontor-API instance.
"""
name: str
url: str
token: str
token_type: str
timeout: int
def login(self, login: Login, log: Logger):
"""
get token from server by calling login endpoint.
"""
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, timeout=self.timeout)
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, table: str, param: Optional[str] = None):
if not param:
url: str = f"{self.url}/{MAPPING[table]}"
else:
url: str = f"{self.url}/{MAPPING[table]}?{param}"
headers: Dict[str, str] = {"Authorization": f"Bearer {self.token}"}
response = requests.get(url, headers=headers, timeout=self.timeout)
log.info(f"Status: {response.status_code}")
data = response.json()
return data
def update(self, log: Logger, table: str, item_id: UUID, file_info: dict):
url: str = f"{self.url}/{MAPPING[table]}/{item_id}"
headers: Dict[str, str] = {"Authorization": f"Bearer {self.token}"}
update = requests.put(
url, headers=headers, json=file_info, timeout=self.timeout
)
log.info(f"Status: {update.status_code}")
return update
@dataclass
class ApiConfig:
"""
Dataclass to define required contents of configuration file.
"""
login: Login
server: List[Server]
def get_logger(level, config: str):
"""
get Logger according to given log level by verbosity.
"""
dirs = PlatformDirs(config)
logging_config = Path(dirs.user_config_dir, "logging-config.yaml")
log_config = None
with open(logging_config, "rt", encoding="utf-8") 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:
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.debug(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
kontor-scripts/download.py
+7 -7
View File
@@ -2,7 +2,6 @@
download files with URLs from DB
"""
import os
import re
import subprocess
import sys
@@ -10,16 +9,17 @@ from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser
from datetime import datetime
from enum import Enum, auto
from pathlib import Path
from typing import Any, Dict
from typing import Dict
from logging import Logger
from uuid import UUID
import requests
from config import get_api_config, get_logger
from api import Server, get_api_config, get_logger
parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter)
parser.add_argument("--verbose", "-v", action="count", default=0)
parser.add_argument("--config", "-c", default="kontor-docker")
parser.add_argument("--config", "-c", default="kontor-api")
parser.add_argument("--dir", "-d", default="/data/media")
parser.add_argument("--limit", "-l", type=int, help="maximum number of links to check")
parser.add_argument("--tool", "-t", default="yt-dlp")
@@ -148,15 +148,15 @@ if __name__ == "__main__":
match download_status:
case FileStatus.DOWNLOADED:
rename_file(item)
update_status(file_id, item, api_data)
update_status(file_id, item, server=server, log=log)
case FileStatus.RENAMED:
log.info("update status")
update_status(file_id, item, api_data)
update_status(file_id, item, server=server, log=log)
case FileStatus.UNKNOWN:
download_file(link, item, args.dir)
rename_file(item)
log.info(f"{item}")
update_status(file_id, item, api_data)
update_status(file_id, item, server=server, log=log)
log.warning(f"processed {mediafile_index}/{entries_count}")
if args.limit and args.limit <= mediafile_index:
break
kontor-scripts/sync.py
+2 -98
View File
@@ -1,14 +1,6 @@
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
from api import get_logger, get_api_config
parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter)
@@ -19,99 +11,11 @@ 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)
data = server.request(logger, "media_file")
logger.info(len(data))
logger.info("kontor.sync finished")