mirror of
https://github.com/YuCat-OVO/BaiduNetdisk-Docker
synced 2025-05-10 20:10:04 +08:00
remove python
This commit is contained in:
YuCat
parent
d13097ba39
commit
205e502e00
20
info.json
20
info.json
@ -1,20 +0,0 @@
|
||||
[
|
||||
{
|
||||
"arch": "amd64",
|
||||
"pak": "deb",
|
||||
"url": "https://issuepcdn.baidupcs.com/issue/netdisk/LinuxGuanjia/4.11.5/baidunetdisk_4.11.5_amd64.deb",
|
||||
"version": "4.11.5"
|
||||
},
|
||||
{
|
||||
"arch": "amd64",
|
||||
"pak": "deb",
|
||||
"url": "https://issuepcdn.baidupcs.com/issue/netdisk/LinuxGuanjia/4.17.7/baidunetdisk_4.17.7_amd64.deb",
|
||||
"version": "4.17.7"
|
||||
},
|
||||
{
|
||||
"arch": "arm64",
|
||||
"pak": "deb",
|
||||
"url": "https://issuepcdn.baidupcs.com/issue/netdisk/LinuxGuanjia/4.17.7/baidunetdisk_4.17.7_arm64.deb",
|
||||
"version": "4.17.7"
|
||||
}
|
||||
]
|
||||
@ -1,42 +0,0 @@
|
||||
{
|
||||
"platform": {
|
||||
"linux": {
|
||||
"latest": {
|
||||
"version": "4.17.7",
|
||||
"amd64": "amd64url",
|
||||
"arm64": "arm64url"
|
||||
},
|
||||
"4.17.7": {
|
||||
"amd64": "amd64url",
|
||||
"arm64": "arm64url"
|
||||
},
|
||||
"4.11.5": {
|
||||
"amd64": "amd64url",
|
||||
"arm64": "arm64url"
|
||||
}
|
||||
},
|
||||
"windows": {
|
||||
"latest": {
|
||||
"version": "7.19.0.18",
|
||||
"amd64": "amd64url"
|
||||
},
|
||||
"7.19.0.18": {
|
||||
"amd64": "amd64url"
|
||||
},
|
||||
"6.9.10.1": {
|
||||
"amd64": "amd64url"
|
||||
}
|
||||
},
|
||||
"mac": {
|
||||
"latest": {
|
||||
"version": "4.11.0",
|
||||
"amd64": "amd64url",
|
||||
"arm64": "arm64url"
|
||||
},
|
||||
"4.11.0": {
|
||||
"amd64": "amd64url",
|
||||
"arm64": "arm64url"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,2 +0,0 @@
|
||||
lxml==5.2.1
|
||||
requests==2.31.0
|
||||
@ -1,81 +0,0 @@
|
||||
import asyncio
|
||||
import json
|
||||
import random
|
||||
import re
|
||||
|
||||
import httpx
|
||||
|
||||
|
||||
class APIFetcher:
|
||||
def __init__(self) -> None:
|
||||
uas = [
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36 Edg/121.0.0.0",
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0",
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
|
||||
]
|
||||
self.ua = random.choice(uas)
|
||||
|
||||
async def async_fetcher(self, platform: str, max_attempt: int = 5) -> str | None:
|
||||
url = "https://yun.baidu.com/disk/cmsdata?platform={0}&page=1&num=100".format(
|
||||
platform
|
||||
)
|
||||
for _ in range(max_attempt):
|
||||
try:
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
url,
|
||||
headers={
|
||||
"User-Agent": self.ua,
|
||||
"Referer": "https://yun.baidu.com/disk/version?errmsg=Auth+Login+Sucess&errno=0&ssnerror=0&",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
return response.text
|
||||
except httpx.HTTPError as e:
|
||||
print(f"An error occurred: {e}")
|
||||
print(f"Fetch {0} failed after several attempts", url)
|
||||
return None
|
||||
|
||||
def linux_version(self):
|
||||
doc = self.linux_json
|
||||
version = json.loads(doc)["list"][0]["version"]
|
||||
version = re.search(r"(?P<version>([0-9]{1,2}(\.)?){3,4})", version).group(
|
||||
"version"
|
||||
)
|
||||
|
||||
return version
|
||||
|
||||
def windows_version(self):
|
||||
doc = self.windows_json
|
||||
version = json.loads(doc)["list"][0]["version"]
|
||||
version = re.search(r"(?P<version>([0-9]{1,2}(\.)?){3,4})", version).group(
|
||||
"version"
|
||||
)
|
||||
return version
|
||||
|
||||
def mac_version(self):
|
||||
doc = self.mac_json
|
||||
version = json.loads(doc)["list"][0]["version"]
|
||||
version = re.search(r"(?P<version>([0-9]{1,2}(\.)?){3,4})", version).group(
|
||||
"version"
|
||||
)
|
||||
return version
|
||||
|
||||
async def run(self):
|
||||
L = await asyncio.gather(
|
||||
self.async_fetcher("linux"),
|
||||
self.async_fetcher("windows"),
|
||||
self.async_fetcher("mac"),
|
||||
)
|
||||
# print(L)
|
||||
self.linux_json, self.windows_json, self.mac_json = L
|
||||
print(self.linux_json)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import doctest
|
||||
|
||||
doctest.testmod()
|
||||
af = APIFetcher()
|
||||
af.run()
|
||||
@ -1,146 +0,0 @@
|
||||
import json
|
||||
import re
|
||||
import time
|
||||
|
||||
import requests
|
||||
|
||||
|
||||
class LinkObj:
|
||||
def __init__(self, link: str) -> None:
|
||||
self.link = link
|
||||
self.version, self.pkg = re.search(
|
||||
r"(?P<version>([0-9]{1,2}(\.)?){3,4}(?=[\._/])).*(?P<pkg>(?<=\.)[a-z]{3,4}$)",
|
||||
link,
|
||||
).group("version", "pkg")
|
||||
arch = re.search(
|
||||
r"(?P<arch>amd64|arm64)",
|
||||
link,
|
||||
)
|
||||
if arch:
|
||||
self.arch = arch.group("arch")
|
||||
else:
|
||||
self.arch = "amd64"
|
||||
|
||||
|
||||
class Version:
|
||||
def __init__(self, version, arch_links):
|
||||
self.version = version
|
||||
self.arch_links = arch_links
|
||||
|
||||
def to_dict(self):
|
||||
return self.arch_links.copy()
|
||||
|
||||
|
||||
class Platform:
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
|
||||
def append_json(json_file_name: str, add):
|
||||
"""为Json添加元素"""
|
||||
file = None
|
||||
try:
|
||||
with open(json_file_name, "r") as f:
|
||||
file = json.load(f)
|
||||
# print("file=", file)
|
||||
|
||||
except FileNotFoundError:
|
||||
file = json.loads("")
|
||||
|
||||
file.append(add)
|
||||
|
||||
# print("ufile=", u_file)
|
||||
|
||||
with open(json_file_name, "w") as f:
|
||||
f.write(json.dumps(file, sort_keys=True, indent=4))
|
||||
return
|
||||
|
||||
|
||||
def parse_url_from_json() -> list:
|
||||
link = ""
|
||||
try:
|
||||
with open("url.json", "r") as f:
|
||||
link = json.load(f)
|
||||
except FileNotFoundError:
|
||||
link = []
|
||||
return link
|
||||
|
||||
|
||||
def make_info_json(urls: list):
|
||||
"""保存为Json文件"""
|
||||
# print("urls:", urls)
|
||||
l_url = urls[-3:]
|
||||
# print(l_url)
|
||||
j = []
|
||||
for i in l_url:
|
||||
j.append(
|
||||
{
|
||||
"version": re.findall("(?<=/)((?:[0-9]{1,2}(?:\\.)?){1,3})(?=/)", i)[0],
|
||||
"arch": re.findall("(amd64|arm64|x86_64)", i)[0],
|
||||
"pak": re.findall("(rpm|deb)", i)[0],
|
||||
"url": i,
|
||||
},
|
||||
)
|
||||
# print(j)
|
||||
with open("info.json", "w") as f:
|
||||
f.write(json.dumps(j, sort_keys=True, indent=4))
|
||||
return
|
||||
|
||||
|
||||
def make_url_temple(arch: str):
|
||||
"""按照之前所获取的 url.json,创建模板"""
|
||||
link = ""
|
||||
url_temple = ""
|
||||
try:
|
||||
with open("url.json", "r") as f:
|
||||
link = json.load(f)
|
||||
for i in link:
|
||||
if re.search("amd64", i) is not None and re.search("deb", i) is not None:
|
||||
url_temple = re.sub("amd64", arch, i)
|
||||
url_temple = re.sub(
|
||||
"(?<=[_/])((?:[0-9]{1,2}(?:\\.)?){1,3})(?=[_/])", "{0}", url_temple
|
||||
)
|
||||
break
|
||||
# print(url_temple)
|
||||
except FileNotFoundError:
|
||||
url_temple = "https://issuepcdn.baidupcs.com/issue/netdisk/LinuxGuanjia/{0}/baidunetdisk_{0}_arm64.deb"
|
||||
# print(url_temple)
|
||||
return url_temple
|
||||
|
||||
|
||||
def check_version(version: list, arch: str):
|
||||
"""检测指定版本号是否可用"""
|
||||
url = make_url_temple(arch).format(".".join(str(v) for v in version))
|
||||
# url = "https://httpbin.org/status/200"
|
||||
# print("Getting {0}".format(url))
|
||||
|
||||
max_attempts = 5
|
||||
attempt = 0
|
||||
|
||||
while attempt < max_attempts:
|
||||
try:
|
||||
response = requests.head(
|
||||
url,
|
||||
headers="",
|
||||
)
|
||||
if response.status_code == 200:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
except Exception as e:
|
||||
print(f"An error occurred: {e}")
|
||||
attempt += 1
|
||||
time.sleep(1) # 等待1秒再重试
|
||||
|
||||
if attempt == max_attempts:
|
||||
print("Failed after several attempts")
|
||||
|
||||
return
|
||||
|
||||
|
||||
def main():
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@ -1,138 +0,0 @@
|
||||
import re
|
||||
|
||||
from api_fetcher import APIFetcher
|
||||
|
||||
|
||||
class VersionFinder:
|
||||
def __init__(self) -> None:
|
||||
pass
|
||||
|
||||
class TemplateGenerator:
|
||||
"""模板创建类
|
||||
|
||||
Args: 从网页获取的url数组
|
||||
Returns: 模板数组, 版本区域由{0}替代
|
||||
"""
|
||||
|
||||
def __init__(self, urls: list) -> list[str]:
|
||||
self.linux_link = []
|
||||
self.windows_link = ""
|
||||
self.mac_link = []
|
||||
|
||||
for i in urls:
|
||||
if re.search("(deb|rpm)", i) is not None:
|
||||
self.linux_link.append(i)
|
||||
elif re.search("exe", i) is not None:
|
||||
if self.windows_link != "":
|
||||
continue
|
||||
self.windows_link = i
|
||||
elif re.search("dmg", i) is not None:
|
||||
self.mac_link.append(i)
|
||||
|
||||
def linux(self):
|
||||
arch = ["arm64", "amd64"]
|
||||
url_template = []
|
||||
|
||||
for url in self.linux_link:
|
||||
# print(url)
|
||||
|
||||
version = re.search(
|
||||
r"(?P<version>(?<=[_/])((?:[0-9]{1,2}(?:\.)?){1,3})(?=[_/])).*(?P<pkg>(?<=\.)[a-z]{3,4}$)",
|
||||
url,
|
||||
).group("pkg", "version")
|
||||
|
||||
self.linux_version = version[1]
|
||||
|
||||
if version[0] == "rpm":
|
||||
url_template.append(re.sub(version[1], "{0}", url))
|
||||
else:
|
||||
template = re.sub("amd64", "{0}", url)
|
||||
url_template += [
|
||||
re.sub(version[1], "{0}", template.format(i)) for i in arch
|
||||
]
|
||||
|
||||
return url_template
|
||||
|
||||
def windows(self):
|
||||
self.windows_version = re.search(
|
||||
r"(?P<version>([0-9]{1,2}(\.)?){3,4}(?=\.|_|\/))", self.windows_link
|
||||
).group("version")
|
||||
return re.sub(self.windows_version, "{0}", self.windows_link)
|
||||
|
||||
def mac(self):
|
||||
versions = {}
|
||||
url_template = []
|
||||
for url in self.mac_link:
|
||||
version = re.search(
|
||||
r"(?P<version>(?<=[_])((?:[0-9]{1,2}(?:\.)?){1,3})(?=[_\.]))",
|
||||
url,
|
||||
).group("version")
|
||||
if re.search("arm", url):
|
||||
versions["arm"] = version
|
||||
url_template.append(re.sub(version, "{0}", url))
|
||||
else:
|
||||
versions["amd"] = version
|
||||
url_template.append(re.sub(version, "{0}", url))
|
||||
self.mac_version = versions
|
||||
return url_template
|
||||
|
||||
def __version_compear(a: str, b: str):
|
||||
a_split = a.split(".")
|
||||
b_split = b.split(".")
|
||||
flag = False
|
||||
if len(a) == len(b):
|
||||
length = len(a_split)
|
||||
for i in range(length):
|
||||
if a_split[i] > b_split[i]:
|
||||
flag = True
|
||||
if flag:
|
||||
return a
|
||||
else:
|
||||
return b
|
||||
else:
|
||||
print("The two version numbers have different lengths.")
|
||||
return a
|
||||
|
||||
def version_setter(self, api_version):
|
||||
self.linux_version = self.__version_compear(
|
||||
self.linux_version, api_version.linux_version()
|
||||
)
|
||||
self.windows_version = self.__version_compear(
|
||||
self.windows_version, api_version.windows_version()
|
||||
)
|
||||
self.mac_version = self.__version_compear(
|
||||
self.mac_version, api_version.mac_version()
|
||||
)
|
||||
|
||||
def run(self, urls: list):
|
||||
tg = self.TemplateGenerator(urls)
|
||||
linux_template = tg.linux()
|
||||
windows_template = tg.windows()
|
||||
mac_template = tg.mac()
|
||||
self.linux_version = tg.linux_version
|
||||
self.windows_version = tg.windows_version
|
||||
self.mac_version = tg.mac_version
|
||||
api = APIFetcher()
|
||||
api.run()
|
||||
self.version_setter(api)
|
||||
|
||||
def append_json(self):
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# import doctest
|
||||
|
||||
# doctest.testmod()
|
||||
urls = [
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/yunguanjia/BaiduNetdisk_7.19.0.18.exe",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/yunguanjia/BaiduNetdisk_6.9.10.1.exe",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/MACguanjia/BaiduNetdisk_mac_4.11.0.dmg",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/MACguanjia/BaiduNetdisk_mac_4.11.0_arm64.dmg",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/LinuxGuanjia/4.11.5/baidunetdisk-4.11.5.x86_64.rpm",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/LinuxGuanjia/4.11.5/baidunetdisk_4.11.5_amd64.deb",
|
||||
]
|
||||
vf = VersionFinder()
|
||||
vf.run(urls)
|
||||
print(vf.linux_version)
|
||||
pass
|
||||
@ -1,148 +0,0 @@
|
||||
import json
|
||||
import random
|
||||
import re
|
||||
|
||||
import httpx
|
||||
from lxml import etree
|
||||
|
||||
|
||||
class WebFetcher:
|
||||
def __init__(self) -> None:
|
||||
uas = [
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36 Edg/121.0.0.0",
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0",
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
|
||||
]
|
||||
self.ua = random.choice(uas)
|
||||
|
||||
def fetcher(self, url: str, max_attempt: int = 5) -> str | None:
|
||||
"""网址抓取器
|
||||
|
||||
从给定 url 抓取对应数据, 失败时重试, 默认重试5次
|
||||
|
||||
>>> bdnd = WebFetcher()
|
||||
>>> type(bdnd.fetcher("https://pan.baidu.com/download")) is str
|
||||
True
|
||||
>>> type(bdnd.fetcher("https://1.1.1.1", max_attempt=1)) is str
|
||||
True
|
||||
>>> bdnd.fetcher("https://172.1.1.1", max_attempt=1)
|
||||
An error occurred: timed out
|
||||
|
||||
Args:
|
||||
url: 网址
|
||||
max_attempt: 重试次数
|
||||
|
||||
Returns:
|
||||
成功时返回所获取网页的text数据
|
||||
失败时返回None
|
||||
"""
|
||||
for _ in range(max_attempt):
|
||||
try:
|
||||
response = httpx.get(
|
||||
url,
|
||||
headers={
|
||||
"User-Agent": self.ua,
|
||||
"Referer": "https://pan.baidu.com/disk/main",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
return response.text
|
||||
except httpx.HTTPError as e:
|
||||
print(f"An error occurred: {e}")
|
||||
|
||||
async def async_fetcher(self, url: str, max_attempt: int = 5) -> str | None:
|
||||
for attempt in range(max_attempt):
|
||||
try:
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
url,
|
||||
headers={
|
||||
"User-Agent": self.ua,
|
||||
"Referer": "https://pan.baidu.com/disk/main",
|
||||
},
|
||||
timeout=10,
|
||||
)
|
||||
return response
|
||||
except httpx.HTTPError as e:
|
||||
print(f"An error occurred: {e}")
|
||||
print(f"Fetch {0} failed after several attempts", url)
|
||||
return None
|
||||
|
||||
def fetch_js_url(self) -> list:
|
||||
"""请求百度网盘下载页,获取存在下载链接的js文件
|
||||
|
||||
Returns:
|
||||
成功时设置self.js_url为所获取网页的text数据
|
||||
失败时设置self.js_url为None
|
||||
|
||||
>>> bdnd = WebFetcher()
|
||||
>>> bdnd.fetch_js_url()
|
||||
'https://nd-static.bdstatic.com/m-static/wp-brand/js/chunk-common.8b953c6c.js'
|
||||
"""
|
||||
response = self.fetcher("https://pan.baidu.com/download")
|
||||
if response is not None:
|
||||
try:
|
||||
doc = etree.HTML(response, parser=None)
|
||||
for i in doc.xpath("//script/@src"):
|
||||
if re.search(r"https:\/\/.*chunk-common.*\.js", i):
|
||||
self.js_url = i
|
||||
return i
|
||||
except Exception as e:
|
||||
print(f"An error occurred: {e}, set js_url to default")
|
||||
self.js_url = "https://nd-static.bdstatic.com/m-static/wp-brand/js/chunk-common.8b953c6c.js"
|
||||
return self.js_url
|
||||
else:
|
||||
print("Fetcher return None, set js_url to default")
|
||||
self.js_url = "https://nd-static.bdstatic.com/m-static/wp-brand/js/chunk-common.8b953c6c.js"
|
||||
return self.js_url
|
||||
|
||||
def parse_url_from_js(self) -> list:
|
||||
"""从js文件中提取下载的链接并且去重
|
||||
|
||||
>>> bdnd = WebFetcher()
|
||||
>>> bdnd.fetch_js_url()
|
||||
'https://nd-static.bdstatic.com/m-static/wp-brand/js/chunk-common.8b953c6c.js'
|
||||
>>> len(bdnd.parse_url_from_js())
|
||||
6
|
||||
"""
|
||||
# 获取js数据
|
||||
js_url = self.js_url
|
||||
response = self.fetcher(js_url)
|
||||
|
||||
# 过滤链接
|
||||
original_url = re.findall(
|
||||
r'(?:(?:link|url)(?:_[0-9])?:")(?P<link>https://[a-zA-Z/\.]*?/netdisk/[a-zA-Z10-9/\._-]*?)(?=")',
|
||||
response,
|
||||
)
|
||||
|
||||
# 去重
|
||||
unique_url = []
|
||||
for i in original_url:
|
||||
if i not in unique_url:
|
||||
unique_url.append(i)
|
||||
self.link_url = unique_url
|
||||
return unique_url
|
||||
|
||||
def json_saver(self):
|
||||
try:
|
||||
with open("url.json", "w") as f:
|
||||
f.write(json.dumps(self.link_url, sort_keys=True, indent=4))
|
||||
except Exception as e:
|
||||
print(f"An error occurred: {e}, Please check the permissions.")
|
||||
|
||||
def run(self):
|
||||
doc = WebFetcher()
|
||||
doc.fetch_js_url()
|
||||
doc.parse_url_from_js()
|
||||
doc.json_saver()
|
||||
return self.link_url
|
||||
|
||||
def get_urls(self) -> list:
|
||||
return self.link_url
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import doctest
|
||||
|
||||
doctest.testmod()
|
||||
8
url.json
8
url.json
@ -1,8 +0,0 @@
|
||||
[
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/yunguanjia/BaiduNetdisk_7.19.0.18.exe",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/yunguanjia/BaiduNetdisk_6.9.10.1.exe",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/MACguanjia/BaiduNetdisk_mac_4.11.0.dmg",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/MACguanjia/BaiduNetdisk_mac_4.11.0_arm64.dmg",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/LinuxGuanjia/4.11.5/baidunetdisk-4.11.5.x86_64.rpm",
|
||||
"https://issuepcdn.baidupcs.com/issue/netdisk/LinuxGuanjia/4.11.5/baidunetdisk_4.11.5_amd64.deb"
|
||||
]
|
||||
Loading…
Reference in New Issue
Block a user