mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-07-09 21:37:47 +08:00
117 lines
3.5 KiB
Python
117 lines
3.5 KiB
Python
from enum import Enum
|
|
from typing import Optional, Union
|
|
|
|
from app.core.cache import TTLCache
|
|
from app.helper.locale import LocaleHelper
|
|
from app.schemas.types import ProgressKey
|
|
|
|
|
|
class ProgressHelper:
|
|
"""
|
|
处理进度辅助类
|
|
"""
|
|
|
|
def __init__(self, key: Union[ProgressKey, str]) -> None:
|
|
if isinstance(key, Enum):
|
|
key = key.value
|
|
self._key = key
|
|
self._progress = TTLCache(region="progress", maxsize=1024, ttl=24 * 60 * 60)
|
|
|
|
def __reset(self) -> None:
|
|
"""
|
|
重置进度
|
|
"""
|
|
self._progress[self._key] = {
|
|
"enable": False,
|
|
"value": 0,
|
|
"text": "请稍候...",
|
|
"data": {}
|
|
}
|
|
|
|
def start(self) -> None:
|
|
"""
|
|
开始进度
|
|
"""
|
|
self.__reset()
|
|
current = self._progress.get(self._key)
|
|
if not current:
|
|
return
|
|
current['enable'] = True
|
|
self._progress[self._key] = current
|
|
|
|
def end(
|
|
self,
|
|
text: Optional[str] = "",
|
|
data: Optional[dict] = None,
|
|
value: Optional[Union[float, int]] = 100,
|
|
) -> None:
|
|
"""
|
|
结束进度
|
|
"""
|
|
current = self._progress.get(self._key)
|
|
if not current:
|
|
return
|
|
if data is not None:
|
|
if not current.get('data'):
|
|
current['data'] = {}
|
|
current['data'].update(data)
|
|
current["enable"] = False
|
|
if value is not None:
|
|
current["value"] = max(min(float(value), 100), 0)
|
|
current["text"] = text or ""
|
|
self._progress[self._key] = current
|
|
|
|
def update(
|
|
self,
|
|
value: Optional[Union[float, int]] = None,
|
|
text: Optional[str] = None,
|
|
data: Optional[dict] = None,
|
|
) -> None:
|
|
"""
|
|
更新进度
|
|
"""
|
|
current = self._progress.get(self._key)
|
|
if not current or not current.get('enable'):
|
|
return
|
|
if value is not None:
|
|
current['value'] = max(min(float(value), 100), 0)
|
|
if text is not None:
|
|
current['text'] = text
|
|
if data is not None:
|
|
if not current.get('data'):
|
|
current['data'] = {}
|
|
current['data'].update(data)
|
|
self._progress[self._key] = current
|
|
|
|
def get(self, locale: Optional[str] = None) -> Optional[dict]:
|
|
"""
|
|
获取当前进度，并按语言补充前端展示字段。
|
|
|
|
:param locale: 目标语言，未传入时使用当前请求上下文语言
|
|
:return: 当前进度字典
|
|
"""
|
|
current = self._progress.get(self._key)
|
|
if not current:
|
|
return current
|
|
|
|
detail = current.copy()
|
|
text = detail.get("text")
|
|
if isinstance(text, str):
|
|
detail["text_i18n"] = LocaleHelper.translate_text(text, locale=locale)
|
|
|
|
data = detail.get("data")
|
|
if isinstance(data, dict):
|
|
localized_data = data.copy()
|
|
error = localized_data.get("error")
|
|
message = localized_data.get("message")
|
|
if isinstance(error, str):
|
|
localized_data["error_i18n"] = LocaleHelper.translate_text(
|
|
error, locale=locale
|
|
)
|
|
if isinstance(message, str):
|
|
localized_data["message_i18n"] = LocaleHelper.translate_text(
|
|
message, locale=locale
|
|
)
|
|
detail["data"] = localized_data
|
|
return detail
|