File size: 3,860 Bytes
17e971c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | import json
from typing import Optional, Union, Tuple, List, Any, Dict
from .base import ServerNode, LocalNode
import public
class _RsyncAPIBase:
def has_rsync_perm(self) -> bool:
raise NotImplementedError()
def is_setup_rsync(self) -> bool:
raise NotImplementedError()
def add_module(self, path: str, name: str, password: str, add_white_ips: List[str]) -> Tuple[Optional[dict], str]:
raise NotImplementedError()
def add_send_task(self, sou):
pass
def get_secretkey(self, ip_type: str = "local_ip") -> Tuple[str, str]:
pass
def check_receiver_conn(self, secret_key: str, work_type: int) -> Tuple[Dict, str]:
pass
class BtLocalRsyncAPI(LocalNode, _RsyncAPIBase):
@classmethod
def new_by_id(cls, node_id: int) -> Optional['BtLocalRsyncAPI']:
node_data = public.M('node').where('id=?', (node_id,)).find()
if not node_data:
return None
if node_data["api_key"] == "local" and node_data["app_key"] == "local":
return BtLocalRsyncAPI()
return None
@staticmethod
def _plugin_func(func_name: str, **kwargs) -> Any:
from panelPlugin import panelPlugin
return panelPlugin().a(public.to_dict_obj({
"name": "rsync",
"s": func_name,
**kwargs,
}))
def has_rsync_perm(self) -> bool:
from panelPlugin import panelPlugin
res = panelPlugin().a(public.to_dict_obj({"name": "rsync"}))
if not res["status"]:
return False
return True
def is_setup_rsync(self) -> bool:
from panelPlugin import panelPlugin
res = panelPlugin().get_soft_find(public.to_dict_obj({"sName": "rsync"}))
try:
return res["setup"]
except:
return False
def add_module(self, path: str, name: str, password: str, add_white_ips: List[str]) -> Tuple[Optional[dict], str]:
res = self._plugin_func("add_module", **{
"path": path,
"mName": name,
"password": password,
"add_white_ips": json.dumps(add_white_ips)
})
return res, ""
class BtRsyncAPI(ServerNode, _RsyncAPIBase):
def _plugin_api_func(self, func_name: str, **kwargs) -> Tuple[Any, str]:
return self._request("/plugin", "a", pdata={
"name": "rsync",
"s": func_name,
**kwargs
})
@classmethod
def new_by_id(cls, node_id: int) -> Optional['BtRsyncAPI']:
node_data = public.M('node').where('id=?', (node_id,)).find()
if not node_data:
return None
if node_data["api_key"] == "local" and node_data["app_key"] == "local":
return None
if node_data['lpver']:
return None
return BtRsyncAPI(node_data["address"], node_data["api_key"], "")
def has_rsync_perm(self) -> bool:
data, err = self._request("/plugin", "a", pdata={"name": "rsync"})
if err:
return False
return data["status"]
def is_setup_rsync(self) -> bool:
data, err = self._request("/plugin", "get_soft_find", pdata={"sName": "rsync"})
if err:
return False
try:
return data["setup"]
except:
return False
def add_module(self, path: str, name: str, password: str, add_white_ips: List[str]) -> Tuple[Optional[dict], str]:
return self._plugin_api_func("add_module", **{
"path": path,
"mName": name,
"password": password,
"add_white_ips": json.dumps(add_white_ips)
})
def get_rsync_api_node(node_id: int) -> Optional[Union['BtRsyncAPI', 'BtLocalRsyncAPI']]:
srv = BtLocalRsyncAPI.new_by_id(node_id)
if srv:
return srv
return BtRsyncAPI.new_by_id(node_id)
|