mirror of
https://git.vectorsigma.ru/public/tubearchivist.git
synced 2026-08-05 02:39:25 +00:00
remove dependency on redis json
This commit is contained in:
@@ -238,8 +238,5 @@ class ReleaseVersion:
|
|||||||
|
|
||||||
def get_update(self) -> dict:
|
def get_update(self) -> dict:
|
||||||
"""return new version dict if available"""
|
"""return new version dict if available"""
|
||||||
message = RedisArchivist().get_message(self.NEW_KEY)
|
message = RedisArchivist().get_message_dict(self.NEW_KEY)
|
||||||
if not message.get("status"):
|
|
||||||
return {}
|
|
||||||
|
|
||||||
return message
|
return message
|
||||||
|
|||||||
@@ -198,7 +198,7 @@ class CookieView(ApiBaseView):
|
|||||||
"""handle get request"""
|
"""handle get request"""
|
||||||
# pylint: disable=unused-argument
|
# pylint: disable=unused-argument
|
||||||
config = AppConfig().config
|
config = AppConfig().config
|
||||||
valid = RedisArchivist().get_message("cookie:valid")
|
valid = RedisArchivist().get_message_dict("cookie:valid")
|
||||||
response = {"cookie_enabled": config["downloads"]["cookie_import"]}
|
response = {"cookie_enabled": config["downloads"]["cookie_import"]}
|
||||||
response.update(valid)
|
response.update(valid)
|
||||||
|
|
||||||
|
|||||||
@@ -40,15 +40,15 @@ class RedisArchivist(RedisBase):
|
|||||||
def set_message(
|
def set_message(
|
||||||
self,
|
self,
|
||||||
key: str,
|
key: str,
|
||||||
message: dict,
|
message: dict | str,
|
||||||
path: str = ".",
|
|
||||||
expire: bool | int = False,
|
expire: bool | int = False,
|
||||||
save: bool = False,
|
save: bool = False,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""write new message to redis"""
|
"""write new message to redis"""
|
||||||
self.conn.execute_command(
|
to_write = (
|
||||||
"JSON.SET", self.NAME_SPACE + key, path, json.dumps(message)
|
json.dumps(message) if isinstance(message, dict) else message
|
||||||
)
|
)
|
||||||
|
self.conn.execute_command("SET", self.NAME_SPACE + key, to_write)
|
||||||
|
|
||||||
if expire:
|
if expire:
|
||||||
if isinstance(expire, bool):
|
if isinstance(expire, bool):
|
||||||
@@ -67,8 +67,24 @@ class RedisArchivist(RedisBase):
|
|||||||
except redis.exceptions.ResponseError:
|
except redis.exceptions.ResponseError:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def get_message(self, key: str) -> dict:
|
def get_message_str(self, key: str) -> str | None:
|
||||||
"""get message dict from redis"""
|
"""get message string"""
|
||||||
|
reply = self.conn.execute_command("GET", self.NAME_SPACE + key)
|
||||||
|
return reply
|
||||||
|
|
||||||
|
def get_message_dict(self, key: str) -> dict:
|
||||||
|
"""get message dict"""
|
||||||
|
reply = self.conn.execute_command("GET", self.NAME_SPACE + key)
|
||||||
|
if not reply:
|
||||||
|
return {}
|
||||||
|
|
||||||
|
return json.loads(reply)
|
||||||
|
|
||||||
|
def get_message(self, key: str) -> dict | None:
|
||||||
|
"""
|
||||||
|
get message dict from redis
|
||||||
|
old json get message, only used for migration, to be removed later
|
||||||
|
"""
|
||||||
reply = self.conn.execute_command("JSON.GET", self.NAME_SPACE + key)
|
reply = self.conn.execute_command("JSON.GET", self.NAME_SPACE + key)
|
||||||
if reply:
|
if reply:
|
||||||
return json.loads(reply)
|
return json.loads(reply)
|
||||||
@@ -91,7 +107,7 @@ class RedisArchivist(RedisBase):
|
|||||||
if not all_matches:
|
if not all_matches:
|
||||||
return []
|
return []
|
||||||
|
|
||||||
return [self.get_message(i) for i in all_matches]
|
return [self.get_message_dict(i) for i in all_matches]
|
||||||
|
|
||||||
def del_message(self, key: str) -> bool:
|
def del_message(self, key: str) -> bool:
|
||||||
"""delete key from redis"""
|
"""delete key from redis"""
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ from common.src.ta_redis import RedisArchivist
|
|||||||
from django.core.management.base import BaseCommand, CommandError
|
from django.core.management.base import BaseCommand, CommandError
|
||||||
from django.utils import dateformat
|
from django.utils import dateformat
|
||||||
from django_celery_beat.models import CrontabSchedule, PeriodicTasks
|
from django_celery_beat.models import CrontabSchedule, PeriodicTasks
|
||||||
|
from redis.exceptions import ResponseError
|
||||||
from task.models import CustomPeriodicTask
|
from task.models import CustomPeriodicTask
|
||||||
from task.src.config_schedule import ScheduleBuilder
|
from task.src.config_schedule import ScheduleBuilder
|
||||||
from task.src.task_manager import TaskManager
|
from task.src.task_manager import TaskManager
|
||||||
@@ -54,7 +55,14 @@ class Command(BaseCommand):
|
|||||||
def _mig_app_settings(self) -> None:
|
def _mig_app_settings(self) -> None:
|
||||||
"""update from v0.4.10 to v0.5.0, migrate application settings"""
|
"""update from v0.4.10 to v0.5.0, migrate application settings"""
|
||||||
self.stdout.write("[MIGRATION] move appconfig to ES")
|
self.stdout.write("[MIGRATION] move appconfig to ES")
|
||||||
config = RedisArchivist().get_message("config")
|
try:
|
||||||
|
config = RedisArchivist().get_message("config")
|
||||||
|
except ResponseError:
|
||||||
|
self.stdout.write(
|
||||||
|
self.style.SUCCESS(" Redis does not support JSON decoding")
|
||||||
|
)
|
||||||
|
return
|
||||||
|
|
||||||
if not config:
|
if not config:
|
||||||
self.stdout.write(
|
self.stdout.write(
|
||||||
self.style.SUCCESS(" no config values to migrate")
|
self.style.SUCCESS(" no config values to migrate")
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ from http import cookiejar
|
|||||||
from io import StringIO
|
from io import StringIO
|
||||||
|
|
||||||
import yt_dlp
|
import yt_dlp
|
||||||
|
from appsettings.src import config as ConfigApp
|
||||||
from common.src.env_settings import EnvironmentSettings
|
from common.src.env_settings import EnvironmentSettings
|
||||||
from common.src.ta_redis import RedisArchivist
|
from common.src.ta_redis import RedisArchivist
|
||||||
|
|
||||||
@@ -91,7 +92,7 @@ class CookieHandler:
|
|||||||
|
|
||||||
def get(self):
|
def get(self):
|
||||||
"""get cookie io stream"""
|
"""get cookie io stream"""
|
||||||
cookie = RedisArchivist().get_message("cookie")
|
cookie = RedisArchivist().get_message_str("cookie")
|
||||||
self.cookie_io = StringIO(cookie)
|
self.cookie_io = StringIO(cookie)
|
||||||
return self.cookie_io
|
return self.cookie_io
|
||||||
|
|
||||||
@@ -116,8 +117,7 @@ class CookieHandler:
|
|||||||
def set_cookie(self, cookie):
|
def set_cookie(self, cookie):
|
||||||
"""set cookie str and activate in config"""
|
"""set cookie str and activate in config"""
|
||||||
RedisArchivist().set_message("cookie", cookie, save=True)
|
RedisArchivist().set_message("cookie", cookie, save=True)
|
||||||
path = ".downloads.cookie_import"
|
ConfigApp.AppConfig().update_config({"downloads.cookie_import": True})
|
||||||
RedisArchivist().set_message("config", True, path=path, save=True)
|
|
||||||
self.config["downloads"]["cookie_import"] = True
|
self.config["downloads"]["cookie_import"] = True
|
||||||
print("cookie: activated and stored in Redis")
|
print("cookie: activated and stored in Redis")
|
||||||
|
|
||||||
@@ -126,9 +126,7 @@ class CookieHandler:
|
|||||||
"""revoke cookie"""
|
"""revoke cookie"""
|
||||||
RedisArchivist().del_message("cookie")
|
RedisArchivist().del_message("cookie")
|
||||||
RedisArchivist().del_message("cookie:valid")
|
RedisArchivist().del_message("cookie:valid")
|
||||||
RedisArchivist().set_message(
|
ConfigApp.AppConfig().update_config({"downloads.cookie_import": False})
|
||||||
"config", False, path=".downloads.cookie_import"
|
|
||||||
)
|
|
||||||
print("cookie: revoked")
|
print("cookie: revoked")
|
||||||
|
|
||||||
def validate(self):
|
def validate(self):
|
||||||
|
|||||||
@@ -4,9 +4,8 @@ functionality:
|
|||||||
- handle threads and locks
|
- handle threads and locks
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from common.src.ta_redis import RedisArchivist, TaskRedis
|
from common.src.ta_redis import TaskRedis
|
||||||
from task.celery import app as celery_app
|
from task.celery import app as celery_app
|
||||||
from task.src.task_config import TASK_CONFIG
|
|
||||||
|
|
||||||
|
|
||||||
class TaskManager:
|
class TaskManager:
|
||||||
@@ -96,20 +95,13 @@ class TaskCommand:
|
|||||||
|
|
||||||
return message
|
return message
|
||||||
|
|
||||||
def stop(self, task_id, message_key):
|
def stop(self, task_id):
|
||||||
"""
|
"""
|
||||||
send stop signal to task_id,
|
send stop signal to task_id,
|
||||||
needs to be implemented in task to take effect
|
needs to be implemented in task to take effect
|
||||||
"""
|
"""
|
||||||
print(f"[task][{task_id}]: received STOP signal.")
|
print(f"[task][{task_id}]: received STOP signal.")
|
||||||
handler = TaskRedis()
|
TaskRedis().set_command(task_id, "STOP")
|
||||||
|
|
||||||
task = handler.get_single(task_id)
|
|
||||||
if not task["name"] in TASK_CONFIG:
|
|
||||||
raise ValueError
|
|
||||||
|
|
||||||
handler.set_command(task_id, "STOP")
|
|
||||||
RedisArchivist().set_message(message_key, "STOP", path=".command")
|
|
||||||
|
|
||||||
def kill(self, task_id):
|
def kill(self, task_id):
|
||||||
"""send kill signal to task_id"""
|
"""send kill signal to task_id"""
|
||||||
|
|||||||
@@ -103,8 +103,7 @@ class TaskIDView(ApiBaseView):
|
|||||||
message = {"message": "task can not be stopped"}
|
message = {"message": "task can not be stopped"}
|
||||||
return Response(message, status=400)
|
return Response(message, status=400)
|
||||||
|
|
||||||
message_key = self._build_message_key(task_conf, task_id)
|
TaskCommand().stop(task_id)
|
||||||
TaskCommand().stop(task_id, message_key)
|
|
||||||
if command == "kill":
|
if command == "kill":
|
||||||
if not task_conf.get("api_stop"):
|
if not task_conf.get("api_stop"):
|
||||||
message = {"message": "task can not be killed"}
|
message = {"message": "task can not be killed"}
|
||||||
@@ -114,10 +113,6 @@ class TaskIDView(ApiBaseView):
|
|||||||
|
|
||||||
return Response({"message": "command sent"})
|
return Response({"message": "command sent"})
|
||||||
|
|
||||||
def _build_message_key(self, task_conf, task_id):
|
|
||||||
"""build message key to forward command to notification"""
|
|
||||||
return f"message:{task_conf.get('group')}:{task_id.split('-')[0]}"
|
|
||||||
|
|
||||||
|
|
||||||
class ScheduleListView(ApiBaseView):
|
class ScheduleListView(ApiBaseView):
|
||||||
"""resolves to /api/task/schedule/
|
"""resolves to /api/task/schedule/
|
||||||
|
|||||||
@@ -102,7 +102,7 @@ class VideoProgressView(ApiBaseView):
|
|||||||
"""get progress for a single video"""
|
"""get progress for a single video"""
|
||||||
user_id = request.user.id
|
user_id = request.user.id
|
||||||
key = f"{user_id}:progress:{video_id}"
|
key = f"{user_id}:progress:{video_id}"
|
||||||
video_progress = RedisArchivist().get_message(key)
|
video_progress = RedisArchivist().get_message_dict(key)
|
||||||
position = video_progress.get("position", 0)
|
position = video_progress.get("position", 0)
|
||||||
|
|
||||||
self.response = {
|
self.response = {
|
||||||
|
|||||||
Reference in New Issue
Block a user