diff --git a/tubearchivist/appsettings/config.json b/tubearchivist/appsettings/config.json deleted file mode 100644 index 184721b4..00000000 --- a/tubearchivist/appsettings/config.json +++ /dev/null @@ -1,30 +0,0 @@ -{ - "subscriptions": { - "channel_size": 50, - "live_channel_size": 50, - "shorts_channel_size": 50, - "auto_start": false - }, - "downloads": { - "limit_speed": false, - "sleep_interval": 3, - "autodelete_days": false, - "format": false, - "format_sort": false, - "add_metadata": false, - "add_thumbnail": false, - "subtitle": false, - "subtitle_source": false, - "subtitle_index": false, - "comment_max": false, - "comment_sort": "top", - "cookie_import": false, - "throttledratelimit": false, - "extractor_lang": false, - "integrate_ryd": false, - "integrate_sponsorblock": false - }, - "application": { - "enable_snapshot": true - } -} diff --git a/tubearchivist/appsettings/src/config.py b/tubearchivist/appsettings/src/config.py index bd7b2d06..b44d037a 100644 --- a/tubearchivist/appsettings/src/config.py +++ b/tubearchivist/appsettings/src/config.py @@ -4,104 +4,119 @@ Functionality: - load config variables into redis """ -import json from random import randint from time import sleep +from typing import Literal, TypedDict import requests +from common.src.es_connect import ElasticWrap from common.src.ta_redis import RedisArchivist from django.conf import settings +class SubscriptionsConfigType(TypedDict): + """describes subscriptions config""" + + channel_size: int + live_channel_size: int + shorts_channel_size: int + auto_start: bool + + +class DownloadsConfigType(TypedDict): + """describes downloads config""" + + limit_speed: int + sleep_interval: int + autodelete_days: int + format: str | bool + format_sort: str | bool + add_metadata: bool + add_thumbnail: bool + subtitle: str | bool + subtitle_source: Literal["user", "auto"] | bool + subtitle_index: bool + comment_max: str | bool + comment_sort: Literal["top", "new"] + cookie_import: bool + throttledratelimit: int + extractor_lang: str | bool + integrate_ryd: bool + integrate_sponsorblock: bool + + +class ApplicationConfigType(TypedDict): + """describes application config""" + + enable_snapshot: bool + + +class AppConfigType(TypedDict): + """combined app config type""" + + subscriptions: SubscriptionsConfigType + downloads: DownloadsConfigType + application: ApplicationConfigType + + class AppConfig: """handle application variables""" + ES_PATH = "ta_config/_doc/appsettings" + CONFIG_DEFAULTS: AppConfigType = { + "subscriptions": { + "channel_size": 50, + "live_channel_size": 50, + "shorts_channel_size": 50, + "auto_start": False, + }, + "downloads": { + "limit_speed": False, + "sleep_interval": 3, + "autodelete_days": False, + "format": False, + "format_sort": False, + "add_metadata": False, + "add_thumbnail": False, + "subtitle": False, + "subtitle_source": False, + "subtitle_index": False, + "comment_max": False, + "comment_sort": "top", + "cookie_import": False, + "throttledratelimit": False, + "extractor_lang": False, + "integrate_ryd": False, + "integrate_sponsorblock": False, + }, + "application": {"enable_snapshot": True}, + } + def __init__(self): self.config = self.get_config() - def get_config(self): - """get config from default file or redis if changed""" - config = self.get_config_redis() - if not config: - config = self.get_config_file() + def get_config(self) -> AppConfigType: + """get config from ES""" + response, status_code = ElasticWrap(self.ES_PATH).get() + if not status_code == 200: + return self.CONFIG_DEFAULTS - return config + return response["_source"] - def get_config_file(self): - """read the defaults from config.json""" - with open("appsettings/config.json", "r", encoding="utf-8") as f: - config_file = json.load(f) + def update_config(self, key, value): + """update single config value""" + key_map = key.split(".") + self._validate_key(key_map) + self.config[key_map[0]][key_map[1]] = value + response, status_code = ElasticWrap(self.ES_PATH).post(self.config) + if not status_code == 200: + print(response) - return config_file - - @staticmethod - def get_config_redis(): - """read config json set from redis to overwrite defaults""" - for i in range(10): - try: - config = RedisArchivist().get_message("config") - if not list(config.values())[0]: - return False - - return config - - except Exception: # pylint: disable=broad-except - print(f"... Redis connection failed, retry [{i}/10]") - sleep(3) - - raise ConnectionError("failed to connect to redis") - - def update_config(self, form_post): - """update config values from settings form""" - updated = [] - for key, value in form_post.items(): - if not value and not isinstance(value, int): - continue - - if value in ["0", 0]: - to_write = False - elif value == "1": - to_write = True - else: - to_write = value - - config_dict, config_value = key.split("_", maxsplit=1) - self.config[config_dict][config_value] = to_write - updated.append((config_value, to_write)) - - RedisArchivist().set_message("config", self.config, save=True) - return updated - - def load_new_defaults(self): - """check config.json for missing defaults""" - default_config = self.get_config_file() - redis_config = self.get_config_redis() - - # check for customizations - if not redis_config: - config = self.get_config() - RedisArchivist().set_message("config", config) - return False - - needs_update = False - - for key, value in default_config.items(): - # missing whole main key - if key not in redis_config: - redis_config.update({key: value}) - needs_update = True - continue - - # missing nested values - for sub_key, sub_value in value.items(): - if sub_key not in redis_config[key].keys(): - redis_config[key].update({sub_key: sub_value}) - needs_update = True - - if needs_update: - RedisArchivist().set_message("config", redis_config) - - return needs_update + def _validate_key(self, key_map: list[str]) -> None: + """raise valueerror on invalid key""" + exists = self.CONFIG_DEFAULTS.get(key_map[0], {}).get(key_map[1]) # type: ignore # noqa: E501 + if exists is None: + raise ValueError(f"trying to access invalid config key: {key_map}") class ReleaseVersion: diff --git a/tubearchivist/config/management/commands/ta_startup.py b/tubearchivist/config/management/commands/ta_startup.py index f7e93796..2ea6597f 100644 --- a/tubearchivist/config/management/commands/ta_startup.py +++ b/tubearchivist/config/management/commands/ta_startup.py @@ -9,21 +9,18 @@ from datetime import datetime from random import randint from time import sleep -from appsettings.src.config import AppConfig, ReleaseVersion +from appsettings.src.config import ReleaseVersion from appsettings.src.index_setup import ElasitIndexWrap from appsettings.src.snapshot import ElasticSnapshot from common.src.env_settings import EnvironmentSettings from common.src.es_connect import ElasticWrap from common.src.helper import clear_dl_cache from common.src.ta_redis import RedisArchivist -from django.conf import settings from django.core.management.base import BaseCommand, CommandError from django.utils import dateformat from django_celery_beat.models import CrontabSchedule, PeriodicTasks from task.models import CustomPeriodicTask from task.src.config_schedule import ScheduleBuilder -from task.src.notify import Notifications -from task.src.task_config import TASK_CONFIG from task.src.task_manager import TaskManager from task.tasks import version_check @@ -44,30 +41,41 @@ class Command(BaseCommand): def handle(self, *args, **options): """run all commands""" self.stdout.write(TOPIC) - self._sync_redis_state() self._make_folders() self._clear_redis_keys() self._clear_tasks() self._clear_dl_cache() self._version_check() - self._mig_index_setup() - self._mig_snapshot_check() - self._mig_schedule_store() - self._mig_custom_playlist() - self._mig_add_missing_timestamp() + self._index_setup() + self._snapshot_check() self._create_default_schedules() self._update_schedule_tz() - def _sync_redis_state(self): - """make sure redis gets new config.json values""" - self.stdout.write("[1] set new config.json values") - needs_update = AppConfig().load_new_defaults() - if needs_update: + def _mig_app_settings(self) -> None: + """update from v0.4.10 to v0.5.0, migrate application settings""" + self.stdout.write("[MIGRATION] move appconfig to ES") + config = RedisArchivist().get_message("config") + if not config: self.stdout.write( - self.style.SUCCESS(" ✓ new config values set") + self.style.SUCCESS(" no config values to migrate") ) - else: - self.stdout.write(self.style.SUCCESS(" no new config values")) + return + + path = "ta_config/_doc/appsettings" + response, status_code = ElasticWrap(path).post(config) + + if status_code in [200, 201]: + self.stdout.write( + self.style.SUCCESS(" ✓ migrated appconfig to ES") + ) + RedisArchivist().del_message("config") + return + + message = " 🗙 failed to migrate app config" + self.stdout.write(self.style.ERROR(message)) + self.stdout.write(response) + sleep(60) + raise CommandError(message) def _make_folders(self): """make expected cache folders""" @@ -160,197 +168,22 @@ class Command(BaseCommand): self.style.SUCCESS(" ✓ send initial version check task") version_check.delay() - def _mig_index_setup(self): + def _index_setup(self): """migration: validate index mappings""" - self.stdout.write("[MIGRATION] validate index mappings") + self.stdout.write("[7] validate index mappings") ElasitIndexWrap().setup() - def _mig_snapshot_check(self): + def _snapshot_check(self): """migration setup snapshots""" - self.stdout.write("[MIGRATION] setup snapshots") + self.stdout.write("[8] setup snapshots") ElasticSnapshot().setup() - def _mig_schedule_store(self): - """ - update from 0.4.7 to 0.4.8 - migrate schedule task store to CustomCronSchedule - """ - self.stdout.write("[MIGRATION] migrate schedule store") - config = AppConfig().config - current_schedules = config.get("scheduler") - if not current_schedules: - self.stdout.write( - self.style.SUCCESS(" no schedules to migrate") - ) - return - - self._mig_update_subscribed(current_schedules) - self._mig_download_pending(current_schedules) - self._mig_check_reindex(current_schedules) - self._mig_thumbnail_check(current_schedules) - self._mig_run_backup(current_schedules) - self._mig_version_check() - - del config["scheduler"] - RedisArchivist().set_message("config", config, save=True) - - def _mig_update_subscribed(self, current_schedules): - """create update_subscribed schedule""" - task_name = "update_subscribed" - update_subscribed_schedule = current_schedules.get(task_name) - if update_subscribed_schedule: - self._create_task(task_name, update_subscribed_schedule) - - self._create_notifications(task_name, current_schedules) - - def _mig_download_pending(self, current_schedules): - """create download_pending schedule""" - task_name = "download_pending" - download_pending_schedule = current_schedules.get(task_name) - if download_pending_schedule: - self._create_task(task_name, download_pending_schedule) - - self._create_notifications(task_name, current_schedules) - - def _mig_check_reindex(self, current_schedules): - """create check_reindex schedule""" - task_name = "check_reindex" - check_reindex_schedule = current_schedules.get(task_name) - if check_reindex_schedule: - task_config = {} - days = current_schedules.get("check_reindex_days") - if days: - task_config.update({"days": days}) - - self._create_task( - task_name, - check_reindex_schedule, - task_config=task_config, - ) - - self._create_notifications(task_name, current_schedules) - - def _mig_thumbnail_check(self, current_schedules): - """create thumbnail_check schedule""" - thumbnail_check_schedule = current_schedules.get("thumbnail_check") - if thumbnail_check_schedule: - self._create_task("thumbnail_check", thumbnail_check_schedule) - - def _mig_run_backup(self, current_schedules): - """create run_backup schedule""" - run_backup_schedule = current_schedules.get("run_backup") - if run_backup_schedule: - task_config = False - rotate = current_schedules.get("run_backup_rotate") - if rotate: - task_config = {"rotate": rotate} - - self._create_task( - "run_backup", run_backup_schedule, task_config=task_config - ) - - def _mig_version_check(self): - """create version_check schedule""" - version_check_schedule = { - "minute": randint(0, 59), - "hour": randint(0, 23), - "day_of_week": "*", - } - self._create_task("version_check", version_check_schedule) - - def _create_task(self, task_name, schedule, task_config=False): - """create task""" - description = TASK_CONFIG[task_name].get("title") - schedule, _ = CrontabSchedule.objects.get_or_create(**schedule) - schedule.timezone = settings.TIME_ZONE - schedule.save() - - task, _ = CustomPeriodicTask.objects.get_or_create( - crontab=schedule, - name=task_name, - description=description, - task=task_name, - ) - if task_config: - task.task_config = task_config - task.save() - - self.stdout.write( - self.style.SUCCESS(f" ✓ new task created: '{task}'") - ) - - def _create_notifications(self, task_name, current_schedules): - """migrate notifications of task""" - notifications = current_schedules.get(f"{task_name}_notify") - if not notifications: - return - - urls = [i.strip() for i in notifications.split()] - if not urls: - return - - self.stdout.write( - self.style.SUCCESS(f" ✓ migrate notifications: '{urls}'") - ) - handler = Notifications(task_name) - for url in urls: - handler.add_url(url) - - def _mig_custom_playlist(self): - """add playlist_type for migration from v0.4.6 to v0.4.7""" - self.stdout.write("[MIGRATION] custom playlist") - data = { - "query": { - "bool": {"must_not": [{"exists": {"field": "playlist_type"}}]} - }, - "script": {"source": "ctx._source['playlist_type'] = 'regular'"}, - } - path = "ta_playlist/_update_by_query" - response, status_code = ElasticWrap(path).post(data=data) - if status_code == 200: - updated = response.get("updated", 0) - if updated: - self.stdout.write( - self.style.SUCCESS( - f" ✓ {updated} playlist_type updated in ta_playlist" - ) - ) - else: - self.stdout.write( - self.style.SUCCESS( - " no playlist_type needed updating in ta_playlist" - ) - ) - return - - message = " 🗙 ta_playlist playlist_type update failed" - self.stdout.write(self.style.ERROR(message)) - self.stdout.write(response) - sleep(60) - raise CommandError(message) - - def _mig_add_missing_timestamp(self) -> None: - """ - add missing timestamp for versioncheck - migrate from v0.4.8 to v0.4.9 - """ - version_tasks = CustomPeriodicTask.objects.filter(name="version_check") - if not version_tasks.exists(): - return - - version_task = version_tasks.first() - if not version_task.last_run_at: - self.style.SUCCESS(" ✓ send initial version check task") - version_check.delay() - version_task.last_run_at = dateformat.make_aware(datetime.now()) - version_task.save() - def _create_default_schedules(self) -> None: """ create default schedules for new installations needs to be called after _mig_schedule_store """ - self.stdout.write("[7] create initial schedules") + self.stdout.write("[9] create initial schedules") init_has_run = CustomPeriodicTask.objects.filter( name="version_check" ).exists() @@ -401,6 +234,7 @@ class Command(BaseCommand): def _update_schedule_tz(self) -> None: """update timezone for Schedule instances""" + self.stdout.write("[9] validate schedules TZ") tz = EnvironmentSettings.TZ to_update = CrontabSchedule.objects.exclude(timezone=tz)