diff --git a/app/plugins/modules/customwordimport.py b/app/plugins/modules/customwordimport.py new file mode 100644 index 00000000..372c9a9f --- /dev/null +++ b/app/plugins/modules/customwordimport.py @@ -0,0 +1,310 @@ +import time +from datetime import datetime, timedelta +from urllib.parse import urlsplit +import ruamel.yaml + +import pytz +from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.triggers.cron import CronTrigger + +from threading import Event +from app.plugins.modules._base import _IPluginModule +from app.utils import RequestUtils +from config import Config +from app.utils.types import MediaType +from app.media import Media +from app.helper import WordsHelper + + + +class CustomWordImport(_IPluginModule): + # 插件名称 + module_name = "自定义识别词导入" + # 插件描述 + module_desc = "从github导入自定义识别词规则。" + # 插件图标 + module_icon = "text-recognition.png" + # 主题色 + module_color = "#a6cce0" + # 插件版本 + module_version = "1.0" + # 插件作者 + module_author = "linyuan213" + # 作者主页 + author_url = "https://github.com/linyuan213" + # 插件配置项ID前缀 + module_config_prefix = "customwordimport_" + # 加载顺序 + module_order = 22 + # 可使用的用户级别 + auth_level = 1 + + # 私有属性 + _scheduler = None + + # 设置开关 + _enabled = False + # 任务执行间隔 + _cron = None + _status = None + _github_path = None + _default_path = 'https://github.com/linyuan0213/MediaRecognitionRule' + _onlyonce = False + _notify = False + _file_list = ['common.yaml', 'tv.yaml', 'movie.yaml', 'anime.yaml'] + _github_raw_url = 'https://raw.githubusercontent.com' + # 退出事件 + _event = Event() + + _media = None + _wordshelper = None + + @staticmethod + def get_fields(): + return [ + # 同一板块 + { + 'type': 'div', + 'content': [ + # 同一行 + [ + { + 'title': '是否启用插件', + 'required': "", + 'tooltip': '开启后会根据周期定时导入识别词', + 'type': 'switch', + 'id': 'enabled', + }, + { + 'title': '立即运行一次', + 'required': "", + 'tooltip': '打开后立即运行一次', + 'type': 'switch', + 'id': 'onlyonce', + }, + { + 'title': '运行时通知', + 'required': "", + 'tooltip': '运行任务后会发送通知（需要打开插件消息通知）', + 'type': 'switch', + 'id': 'notify', + }, + ] + ] + }, + { + 'type': 'div', + 'content': [ + # 同一行 + [ + { + 'title': '导入周期', + 'required': "", + 'tooltip': '设置自动备份时间周期，支持5位cron表达式', + 'type': 'text', + 'content': [ + { + 'id': 'cron', + 'placeholder': '0 0 0 ? *', + } + ] + }, + { + 'title': 'github 地址', + 'required': "", + 'tooltip': 'github 地址（默认地址 https://github.com/linyuan0213/MediaRecognitionRule）', + 'type': 'text', + 'content': [ + { + 'id': 'github_path', + 'placeholder': 'https://github.com/linyuan0213/MediaRecognitionRule', + } + ] + } + ] + ] + }, + { + 'type': 'div', + 'content': [ + # 同一行 + [ + { + 'title': '是否启用识别词', + 'required': "", + 'tooltip': '开启后会启用识别词', + 'type': 'switch', + 'id': 'status', + } + ] + ] + }, + + ] + + def init_config(self, config=None): + # 读取配置 + if config: + self._enabled = config.get("enabled") + self._cron = config.get("cron") + self._status = config.get("status") + self._github_path = config.get("github_path") + self._notify = config.get("notify") + self._onlyonce = config.get("onlyonce") + self._media = Media() + self._wordshelper = WordsHelper() + + # 停止现有任务 + self.stop_service() + + # 启动服务 + if self._enabled or self._onlyonce: + self._scheduler = BackgroundScheduler(timezone=Config().get_timezone()) + + # 运行一次 + if self._onlyonce: + self.info(f"自定义识别词导入服务启动，立即运行一次") + self._scheduler.add_job(self.__custom_word_import, 'date', + run_date=datetime.now(tz=pytz.timezone(Config().get_timezone())) + timedelta( + seconds=3)) + # 关闭一次性开关 + self._onlyonce = False + self.update_config({ + "enabled": self._enabled, + "cron": self._cron, + "status": self._status, + "github_path": self._github_path, + "notify": self._notify, + "onlyonce": self._onlyonce, + }) + + # 周期运行 + if self._cron: + self.info(f"定时备份服务启动，周期：{self._cron}") + self._scheduler.add_job(self.__custom_word_import, + CronTrigger.from_crontab(self._cron)) + + # 启动任务 + if self._scheduler.get_jobs(): + self._scheduler.print_jobs() + self._scheduler.start() + + def __custom_word_import(self): + """ + 自动导入 + """ + self.info(f"当前时间 {time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))} 开始备份") + + ua = Config().get_config('app').get('user_agent') + github_path = self._github_path or self._default_path + + self.info(f"github url {github_path} ") + + split_url = urlsplit(github_path) + url_path = split_url.path + + for file_name in self._file_list: + download_url = f'{self._github_raw_url}{url_path}/master/{file_name}' + self.info(f'开始下载规则：{download_url}') + headers = { + "user-agent": ua, + "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,ja;q=0.7" + } + res = RequestUtils(headers=headers).get_res(download_url) + if res.status_code != 200: + return + yaml = ruamel.yaml.YAML() + custom_word_dict = yaml.load(res.text) + if not custom_word_dict: + continue + media_type = None + gtype = None + group_id = -1 + if 'tv' in file_name: + media_type = MediaType.TV + gtype = 2 + if 'movie' in file_name: + media_type = MediaType.MOVIE + gtype = 1 + if 'anime' in file_name: + media_type = MediaType.ANIME + gtype = 2 + + for tmdb_id, rules in custom_word_dict.items(): + self.info(f'开始导入：{tmdb_id}') + if media_type: + tmdb_info = self._media.get_tmdb_info(media_type, tmdb_id) + if not tmdb_info: + continue + + if not self._wordshelper.is_custom_word_group_existed(tmdbid=tmdb_id, gtype=gtype): + self._wordshelper.insert_custom_word_groups(title=tmdb_info.get("name"), + year=tmdb_info.get( + "first_air_date")[0:4], + gtype=2, + tmdbid=tmdb_id, + season_count=tmdb_info.get("number_of_seasons")) + + custom_word_groups = self._wordshelper.get_custom_word_groups(tmdbid=tmdb_id, gtype=gtype) + if custom_word_groups: + group_id = custom_word_groups[0].ID + else: + group_id = -1 + + for import_word_info in rules: + replaced = import_word_info.get("replaced") + replace = import_word_info.get("replace") + front = import_word_info.get("front") + back = import_word_info.get("back") + offset = import_word_info.get("offset") + whelp = import_word_info.get("help") + wtype = int(import_word_info.get("type")) + season = import_word_info.get("season") + regex = 1 + # 屏蔽, 替换, 替换+集偏移 + if wtype in [1, 2, 3]: + if self._wordshelper.is_custom_words_existed(replaced=replaced): + self.info(f"识别词已存在\n（被替换词：{replaced}）") + continue + # 集偏移 + elif wtype == 4: + if self._wordshelper.is_custom_words_existed(front=front, back=back): + self.info(f"识别词已存在\n（前后定位词：{front}@{back}") + continue + self._wordshelper.insert_custom_word(replaced=replaced, + replace=replace, + front=front, + back=back, + offset=offset, + wtype=wtype, + gid=group_id, + season=season, + enabled=1 if self._status else 0, + regex=regex, + whelp=whelp if whelp else "") + + self.info('自定义识别词导入任务完成') + # 发送通知 + if self._notify: + next_run_time = self._scheduler.get_jobs()[0].next_run_time.strftime('%Y-%m-%d %H:%M:%S') + self.send_message(title="【自定义识别词导入任务完成】", + text=f"自定义识别词导入{'成功' if True else '失败'}\n" + f"下次导入时间: {next_run_time}") + + def stop_service(self): + """ + 退出插件 + """ + try: + if self._scheduler: + self._scheduler.remove_all_jobs() + if self._scheduler.running: + self._event.set() + self._scheduler.shutdown() + self._event.clear() + self._scheduler = None + except Exception as e: + print(str(e)) + + def get_state(self): + return self._enabled and self._cron diff --git a/version.py b/version.py index 1fbb60b1..80ef637d 100644 --- a/version.py +++ b/version.py @@ -1 +1 @@ -APP_VERSION = 'v3.2.3' +APP_VERSION = 'v3.2.4' diff --git a/web/static/img/plugins/text-recognition.png b/web/static/img/plugins/text-recognition.png new file mode 100644 index 00000000..74ec7c2e Binary files /dev/null and b/web/static/img/plugins/text-recognition.png differ