feat Lucky Homepage插件

This commit is contained in:
thsrite
2025-01-06 13:09:54 +08:00
parent a38a11252e
commit 052ba0f949
4 changed files with 717 additions and 8 deletions

37
docs/Lucky.md Normal file
View File

@@ -0,0 +1,37 @@
# Lucky HomePage自定义API
![img.png](../img/HomePage/img.png)
HomePage services.yaml配置
```angular2html
- Media:
- Lucky:
icon: /icons/icon/lucky.png
href: http://lucky_ip:lucky_port
ping: http://lucky_ip:lucky_port
# server: unraid
# container: lucky
showStats: true
widget:
type: customapi
url: http://MoviePilot_IP:NGINX_PORT/api/v1/plugin/Lucky/lucky?apikey=api_token
method: GET
mappings:
- field: enabled_cnt
label: 启用配置数量
- field: closed_cnt
label: 关闭配置数量
- field: ipaddr
label: 公网ip地址
- field: expire_time
label: 证书过期时间
# - field: connections
# label: 链接数
# - field: trafficIn
# label: 流量In
# - field: trafficOut
# label: 流量Out
```
### HomePage自定义API文档
https://gethomepage.dev/latest/widgets/services/customapi/#custom-request-body

View File

@@ -770,5 +770,18 @@
"v1.1": "支持交互命令手动同步单个剧集 /as 媒体库名 剧集名。",
"v1.0": "同步剧演员信息到集演员信息。"
}
},
"Lucky": {
"name": "Lucky",
"description": "Lucky HomePage自定义API。",
"labels": "工具",
"version": "1.0",
"icon": "Lucky_A.png",
"author": "thsrite",
"level": 1,
"v2": true,
"history": {
"v1.0": "Lucky HomePage自定义API。"
}
}
}

View File

@@ -6,7 +6,6 @@ from app.db.subscribe_oper import SubscribeOper
from app.helper.directory import DirectoryHelper
from app.plugins import _PluginBase
from typing import Any, List, Dict, Tuple, Optional
from app.schemas import NotificationType
from app import schemas
from app.utils.string import StringUtils
from app.utils.system import SystemUtils
@@ -114,13 +113,6 @@ class HomePage(_PluginBase):
"""
拼装插件配置页面需要返回两块数据1、页面配置2、数据结构
"""
# 编历 NotificationType 枚举,生成消息类型选项
MsgTypeOptions = []
for item in NotificationType:
MsgTypeOptions.append({
"title": item.value,
"value": item.name
})
return [
{
'component': 'VForm',

667
plugins/lucky/__init__.py Normal file
View File

@@ -0,0 +1,667 @@
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"
# 插件作者
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')
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()
logging.info(
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 {
'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",
"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('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