import logging import time from typing import Any, List, Dict, Tuple import requests from app import schemas from app.core.config import settings from app.plugins import _PluginBase from app.utils.string import StringUtils class Lucky(_PluginBase): # 插件名称 plugin_name = "Lucky" # 插件描述 plugin_desc = "Lucky HomePage自定义API。" # 插件图标 plugin_icon = "Lucky_A.png" # 插件版本 plugin_version = "1.0.1" # 插件作者 plugin_author = "thsrite" # 作者主页 author_url = "https://github.com/thsrite" # 插件配置项ID前缀 plugin_config_prefix = "lucky_" # 加载顺序 plugin_order = 30 # 可使用的用户级别 auth_level = 1 # 任务执行间隔 _enabled = False _openToken = None _baseUrl = None _lucky_url = None def init_plugin(self, config: dict = None): if config: self._enabled = config.get("enabled") self._baseUrl = config.get("baseUrl") self._openToken = config.get("openToken") self._lucky_url = f'{self._baseUrl}%s?_=%s&openToken={self._openToken}' def get_rules(self): rule_url = self._lucky_url % ('/api/webservice/rules', int(time.time() * 1000)) rules = [] connections = 0 trafficIn = 0 trafficOut = 0 try: response = requests.get(rule_url, verify=False) # 关闭SSL证书验证 response.raise_for_status() # 如果状态码不是 2xx,抛出异常 if response.json().get('ret') == 0: for rule in response.json().get('ruleList'): if rule.get('ProxyList'): rules += rule.get('ProxyList') if response.json().get('statistics'): for statistic in response.json().get('statistics').values(): if statistic.get('Connections'): connections += statistic.get('Connections') if statistic.get('TrafficIn'): trafficIn += statistic.get('TrafficIn') if statistic.get('TrafficOut'): trafficOut += statistic.get('TrafficOut') except requests.exceptions.RequestException as e: logging.error("An error occurred:", e) return rules, connections, trafficIn, trafficOut def get_ip(self): ip_url = self._lucky_url % ('/api/ddnstasklist', int(time.time() * 1000)) try: response = requests.get(ip_url, verify=False) # 关闭SSL证书验证 response.raise_for_status() # 如果状态码不是 2xx,抛出异常 if response.json().get('ret') == 0: return response.json().get('data')[0].get('IpAddr') or response.json().get('data')[0].get('Ipv4Addr') except requests.exceptions.RequestException as e: logging.error("An error occurred:", e) return None def get_ssl(self): ssl_url = self._lucky_url % ('/api/ssl', int(time.time() * 1000)) try: response = requests.get(ssl_url, verify=False) # 关闭SSL证书验证 response.raise_for_status() # 如果状态码不是 2xx,抛出异常 if response.json().get('ret') == 0: return response.json().get('list')[0].get('CertsInfo')[0].get('NotAfterTime') except requests.exceptions.RequestException as e: logging.error("An error occurred:", e) return None def get_state(self) -> bool: return self._enabled def lucky(self, apikey: str) -> Any: """ 订阅、剩余空间等信息 """ if apikey != settings.API_TOKEN: return schemas.Response(success=False, message="API密钥错误") rules, connections, trafficIn, trafficOut = self.get_rules() enabled_cnt = 0 closed_cnt = 0 for rule in rules: if rule.get('Enable'): enabled_cnt += 1 else: closed_cnt += 1 ipaddr = self.get_ip() expire_time = self.get_ssl() if expire_time: expire_time = expire_time.split(' ')[0] logging.info( f"Proxy Rules Total: {len(rules)}\n" f"Proxy Rules Enabled: {enabled_cnt}\n" f"Proxy Rules Closed: {closed_cnt}\n" f"Connections: {connections}\n" f"TrafficIn: {trafficIn}\n" f"TrafficOut: {trafficOut}\n" f"Lucky IP: {ipaddr}\n" f"SSL Expire Time: {expire_time}\n") return { 'total_cnt': len(rules), 'enabled_cnt': enabled_cnt, 'closed_cnt': closed_cnt, 'ipaddr': ipaddr, 'expire_time': expire_time, 'connections': connections, 'trafficIn': StringUtils.str_filesize(trafficIn), 'trafficOut': StringUtils.str_filesize(trafficOut) } @staticmethod def get_command() -> List[Dict[str, Any]]: pass def get_api(self) -> List[Dict[str, Any]]: """ 获取插件API [{ "path": "/xx", "endpoint": self.xxx, "methods": ["GET", "POST"], "summary": "API说明" }] """ return [{ "path": "/lucky", "endpoint": self.lucky, "methods": ["GET"], "summary": "Lucky HomePage自定义API", "description": "Lucky", }] 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': 6 }, 'content': [ { 'component': 'VSwitch', 'props': { 'model': 'enabled', 'label': '启用插件', } } ] }, ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, 'md': 6 }, 'content': [ { 'component': 'VTextField', 'props': { 'model': 'baseUrl', 'label': 'Lucky地址', 'placeholder': 'http://localhost:16601 (结尾没有/)' } } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 6 }, 'content': [ { 'component': 'VTextField', 'props': { 'model': 'openToken', 'label': 'openToken', } } ] } ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, }, 'content': [ { 'component': 'VAlert', 'props': { 'type': 'success', 'variant': 'tonal' }, 'content': [ { 'component': 'a', 'props': { 'href': 'https://raw.githubusercontent.com/thsrite/MoviePilot-Plugins/main/docs/Lucky.md', 'target': '_blank' }, 'text': '需自行前往Lucky设置开启OpenToken并重启Lucky。' } ] } ] } ] }, { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, }, 'content': [ { 'component': 'VAlert', 'props': { 'type': 'info', 'variant': 'tonal', 'text': '如安装完启用插件后,HomePage提示404,重启MoviePilot即可。' } } ] } ] } ] } ], { "enabled": False, "openToken": "", } def get_page(self) -> List[dict]: dict = self.lucky(settings.API_TOKEN) # 拼装页面 return [ { 'component': 'VRow', 'content': [ { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3, 'sm': 6 }, 'content': [ { 'component': 'VCard', 'props': { 'variant': 'tonal', }, 'content': [ { 'component': 'VCardText', 'props': { 'class': 'd-flex align-center', }, 'content': [ { 'component': 'div', 'content': [ { 'component': 'span', 'props': { 'class': 'text-caption' }, 'text': '总配置数量' }, { 'component': 'div', 'props': { 'class': 'd-flex align-center flex-wrap' }, 'content': [ { 'component': 'span', 'props': { 'class': 'text-h6' }, 'text': dict.get('total_cnt') } ] } ] } ] } ] }, ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3, 'sm': 6 }, 'content': [ { 'component': 'VCard', 'props': { 'variant': 'tonal', }, 'content': [ { 'component': 'VCardText', 'props': { 'class': 'd-flex align-center', }, 'content': [ { 'component': 'div', 'content': [ { 'component': 'span', 'props': { 'class': 'text-caption' }, 'text': '启用配置数量' }, { 'component': 'div', 'props': { 'class': 'd-flex align-center flex-wrap' }, 'content': [ { 'component': 'span', 'props': { 'class': 'text-h6' }, 'text': dict.get('enabled_cnt') } ] } ] } ] } ] }, ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3, 'sm': 6 }, 'content': [ { 'component': 'VCard', 'props': { 'variant': 'tonal', }, 'content': [ { 'component': 'VCardText', 'props': { 'class': 'd-flex align-center', }, 'content': [ { 'component': 'div', 'content': [ { 'component': 'span', 'props': { 'class': 'text-caption' }, 'text': '关闭配置数量' }, { 'component': 'div', 'props': { 'class': 'd-flex align-center flex-wrap' }, 'content': [ { 'component': 'span', 'props': { 'class': 'text-h6' }, 'text': dict.get('closed_cnt') } ] } ] } ] } ] }, ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3, 'sm': 6 }, 'content': [ { 'component': 'VCard', 'props': { 'variant': 'tonal', }, 'content': [ { 'component': 'VCardText', 'props': { 'class': 'd-flex align-center', }, 'content': [ { 'component': 'div', 'content': [ { 'component': 'span', 'props': { 'class': 'text-caption' }, 'text': '公网ip地址' }, { 'component': 'div', 'props': { 'class': 'd-flex align-center flex-wrap' }, 'content': [ { 'component': 'span', 'props': { 'class': 'text-h6' }, 'text': dict.get('ipaddr') } ] } ] } ] } ] }, ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3, 'sm': 6 }, 'content': [ { 'component': 'VCard', 'props': { 'variant': 'tonal', }, 'content': [ { 'component': 'VCardText', 'props': { 'class': 'd-flex align-center', }, 'content': [ { 'component': 'div', 'content': [ { 'component': 'span', 'props': { 'class': 'text-caption' }, 'text': '证书过期日期' }, { 'component': 'div', 'props': { 'class': 'd-flex align-center flex-wrap' }, 'content': [ { 'component': 'span', 'props': { 'class': 'text-h6' }, 'text': dict.get('expire_time') } ] } ] } ] } ] }, ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3, 'sm': 6 }, 'content': [ { 'component': 'VCard', 'props': { 'variant': 'tonal', }, 'content': [ { 'component': 'VCardText', 'props': { 'class': 'd-flex align-center', }, 'content': [ { 'component': 'div', 'content': [ { 'component': 'span', 'props': { 'class': 'text-caption' }, 'text': '链接数' }, { 'component': 'div', 'props': { 'class': 'd-flex align-center flex-wrap' }, 'content': [ { 'component': 'span', 'props': { 'class': 'text-h6' }, 'text': dict.get('connections') } ] } ] } ] } ] }, ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3, 'sm': 6 }, 'content': [ { 'component': 'VCard', 'props': { 'variant': 'tonal', }, 'content': [ { 'component': 'VCardText', 'props': { 'class': 'd-flex align-center', }, 'content': [ { 'component': 'div', 'content': [ { 'component': 'span', 'props': { 'class': 'text-caption' }, 'text': '流量In' }, { 'component': 'div', 'props': { 'class': 'd-flex align-center flex-wrap' }, 'content': [ { 'component': 'span', 'props': { 'class': 'text-h6' }, 'text': dict.get('trafficIn') } ] } ] } ] } ] } ] }, { 'component': 'VCol', 'props': { 'cols': 12, 'md': 3, 'sm': 6 }, 'content': [ { 'component': 'VCard', 'props': { 'variant': 'tonal', }, 'content': [ { 'component': 'VCardText', 'props': { 'class': 'd-flex align-center', }, 'content': [ { 'component': 'div', 'content': [ { 'component': 'span', 'props': { 'class': 'text-caption' }, 'text': '流量Out' }, { 'component': 'div', 'props': { 'class': 'd-flex align-center flex-wrap' }, 'content': [ { 'component': 'span', 'props': { 'class': 'text-h6' }, 'text': dict.get('trafficOut') } ] } ] } ] } ] } ] }, ] }] def stop_service(self): """ 退出插件 """ pass