import json from datetime import datetime, timedelta from app.modules.emby import Emby from app.core.config import settings from app.plugins import _PluginBase from app.log import logger from typing import List, Tuple, Dict, Any, Optional import pytz from apscheduler.triggers.cron import CronTrigger from apscheduler.schedulers.background import BackgroundScheduler class DiagParamAdjust(_PluginBase): # 插件名称 plugin_name = "诊断参数调整" # 插件描述 plugin_desc = "Emby专用插件|暂时性解决emby字幕偏移问题，需要emby安装Diagnostics插件。" # 插件图标 plugin_icon = "Themeengine_A.png" # 插件版本 plugin_version = "1.0" # 插件作者 plugin_author = "jeblove" # 作者主页 author_url = "https://github.com/jeblove" # 插件配置项ID前缀 plugin_config_prefix = "dpa_" # 加载顺序 plugin_order = 14 # 可使用的用户级别 auth_level = 1 # 私有属性 _enabled: bool = False _offset = True _onlyonce = False _base_url = None _endpoint = None _api_key = None _search_text = None _replace_text = None _cron = None # 请求接口 _url = "[HOST]emby/EncodingDiagnostics/DiagnosticOptions?api_key=[APIKEY]" # 定时器 _scheduler: Optional[BackgroundScheduler] = None def init_plugin(self, config: dict = None): # 停止现有任务 self.stop_service() if config: self._enabled = config.get("enabled") self._offset = config.get('offset') self._onlyonce = config.get("onlyonce") self._search_text = config.get("search") self._replace_text = config.get("replace") self._cron = config.get("cron") if self._onlyonce: self._scheduler = BackgroundScheduler(timezone=settings.TZ) logger.info(f"诊断参数调整服务启动，立刻运行一次") self._scheduler.add_job(func=self.run, trigger='date', run_date=datetime.now(tz=pytz.timezone(settings.TZ)) + timedelta(seconds=3), name="诊断参数调整") # 关闭一次性开关 self._onlyonce = False self.update_config({ "enabled": self._enabled, "offset": self._offset, "onlyonce": False, "search": self._search_text, "replace": self._replace_text, "cron": self._cron }) # 启动任务 # self.run() if self._scheduler.get_jobs(): self._scheduler.print_jobs() self._scheduler.start() def get_state(self) -> bool: return self._enabled @staticmethod def get_command() -> List[Dict[str, Any]]: pass def get_api(self) -> List[Dict[str, Any]]: pass def get_service(self) -> List[Dict[str, Any]]: """ 注册插件公共服务 [{ "id": "服务ID", "name": "服务名称", "trigger": "触发器：cron/interval/date/CronTrigger.from_crontab()", "func": self.xxx, "kwargs": {} # 定时器参数 }] """ if self._enabled and self._cron: return [{ "id": "DiagParamAdjust", "name": "诊断参数调整定时服务", "trigger": CronTrigger.from_crontab(self._cron), "func": self.run, "kwargs": {} }] def get_form(self) -> Tuple[List[dict], Dict[str, Any]]: """ 拼装插件配置页面，需要返回两块数据：1、页面配置；2、数据结构 """ return [ { 'component': 'VForm', 'content': [ { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, 'md': 4 }, 'content': [ { 'component': 'VSwitch', 'props': { 'model': 'enabled', 'label': '启用插件', } } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 4 }, 'content': [ { 'component': 'VSwitch', 'props': { 'model': 'offset', 'label': '字幕偏移用途', } } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 4 }, 'content': [ { 'component': 'VSwitch', 'props': { 'model': 'onlyonce', 'label': '立即运行一次', } } ] } ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, 'md': 6 }, 'content': [ { 'component': 'VTextField', 'props': { 'model': 'search', 'label': '搜索文本' } } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 6 }, 'content': [ { 'component': 'VTextField', 'props': { 'model': 'replace', 'label': '替换文本' } } ] } ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, 'md': 6 }, 'content': [ { 'component': 'VTextField', 'props': { 'model': 'cron', 'label': '检测执行周期', 'placeholder': '*/5 * * * *' } } ] }, ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, }, 'content': [ { 'component': 'VAlert', 'props': { 'type': 'info', 'variant': 'tonal', 'text': '暂时性解决emby字幕偏移问题，请在默认参数的基础上修改至适合。\n 此替换文本参数应用于emby-Diagnostics-Parameter Adjustment。\n 默认参数用于修改ffmpeg中字幕覆盖在视频上的位置。\n 方案来源于https://opve.cn/archives/983.html', 'style': 'white-space: pre-line;' } } ] } ] } ] } ],{ "enabled": False, "offset": True, "onlyonce": False, "search": "x=(W-w)/2:y=(H-h):repeatlast=0", "replace": "x=W/4:y=h/4:repeatlast=0", "cron": "*/5 * * * *" } def detect(self): ''' 检测是否存在目标参数 :return True: 存在; False: 不存在 ''' logger.info('字幕偏移修正，检测目标参数') replaceText = "" try: # req_url = "[HOST]emby/EncodingDiagnostics/DiagnosticOptions?api_key=[APIKEY]" res = Emby().get_data(self._url) result = res.json() data = result['Object']['CommandLineOptions'] replaceText = data['ReplaceText'] except json.JSONDecodeError as j: logger.error('服务停止，Emby请安装【Diagnostics】插件') return None except KeyError: # 以装插件，未设置过该参数 # logger.info('目标参数为空') pass # 符合所有情况 if 'repeatlast' in replaceText \ and 'x=(W-w)/2:y=(H-h):repeatlast=0' in data['SearchText'] \ and result['Object']['TranscodingOptions']['DisableHardwareSubtitleOverlay']==True: return True return False def setOptions(self): data = { "CommandLineOptions": { "SearchText": self._search_text, "ReplaceText": self._replace_text }, "TranscodingOptions": { "DisableHardwareSubtitleOverlay": True } } data = json.dumps(data) headers = { 'Content-Type': 'application/octet-stream' } res = Emby().post_data(self._url, data, headers) if res.status_code // 100 == 2: logger.info('参数设置成功') return True else: logger.error('参数设置失败 {}'.format(res.status_code)) def run(self): # 字幕偏移修正，则带检测 if self._offset: state = self.detect() if state: logger.info('参数正常，无需修正') return True elif state==None: return None self.setOptions() def get_page(self) -> List[dict]: pass def stop_service(self): """ 退出插件 """ try: if self._scheduler: self._scheduler.remove_all_jobs() if self._scheduler.running: self._scheduler.shutdown() self._scheduler = None except Exception as e: logger.error("退出插件失败：%s" % str(e))