mirror of
https://git.vectorsigma.ru/public/tubearchivist.git
synced 2026-08-04 23:39:18 +00:00
move AppSettings config to ES
This commit is contained in:
Simon
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user