wcbing/wcbing-apt-repo
1
0
Fork 0
mirror of https://github.com/wcbing/wcbing-apt-repo.git synced 2025-12-28 18:18:31 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: update check_downloader.py

Browse Source
This commit is contained in:
wcbing 2024-12-14 23:04:27 +08:00
parent 769ed03e4b
commit f8f104af1b
1 changed files with 29 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
check_downloader.py
View File

@ -4,41 +4,51 @@ import os
import sqlite3 import sqlite3
def download(url, file_type): def download(url, base_dir):
file_dir = os.path.join(file_type, "/".join(url.split("/")[:-1])) file_dir = os.path.join(base_dir, "/".join(url.split("/")[:-1]))
if not os.path.exists(file_dir): if not os.path.exists(file_dir):
os.makedirs(file_dir) os.makedirs(file_dir)
file_path = os.path.join(file_type, url.split("?")[0]) file_path = os.path.join(base_dir, url.split("?")[0])
# 用 curl 模拟 apt 下载文件，User-Agent 来自 Debian 12 # 用 curl 模拟 apt 下载文件，User-Agent 来自 Debian 12
subprocess.run(["curl", "-H", "User-Agent: Debian APT-HTTP/1.3 (2.6.1)", "-fsLo", file_path, url]) subprocess.run(
[
"curl",
"-H",
"User-Agent: Debian APT-HTTP/1.3 (2.6.1)",
"-fsLo",
file_path,
url,
]
)
def check_download(name, version, url, arch, file_type="deb"): def check_download(name, version, url, arch, base_dir="deb"):
conn = sqlite3.connect(file_type + ".db") conn = sqlite3.connect(base_dir + ".db")
cur = conn.cursor() cur = conn.cursor()
res = cur.execute( res = cur.execute(
"SELECT version, url FROM " + arch + " WHERE name = ?", (name,) "SELECT version, url FROM " + arch + " WHERE name = ?", (name,)
).fetchall() ).fetchall()
if len(res): if len(res):
db_version = res[0][0] local_version = res[0][0]
db_url = res[0][1] local_url = res[0][1]
print(name + ": " + db_version) print(name + ": " + local_version)
if db_version != version: if local_version != version:
print("└ Update: " + db_version + " -> " + version) print("└ Update: " + local_version + " -> " + version)
download(url, file_type) download(url, base_dir)
# wirte to db # wirte to db
cur.execute( cur.execute(
"UPDATE " + arch + " SET version = ?, url = ? WHERE name = ?", "UPDATE " + arch + " SET version = ?, url = ? WHERE name = ?",
(version, url, name), (version, url, name),
) )
# remove old version # remove old version
old_file_path = os.path.join(file_type, db_url.split("?")[0]) if local_url != url: # 针对固定下载链接
if os.path.exists(old_file_path): old_file_path = os.path.join(base_dir, local_url.split("?")[0])
os.remove(old_file_path) if os.path.exists(old_file_path):
os.remove(old_file_path)
else: else:
print(name + "\n└ Add: " + version) print(name + "\n└ Add: " + version)
download(url, file_type) download(url, base_dir)
# wirte to db # wirte to db
cur.execute( cur.execute(
"INSERT INTO " + arch + "(name, version, url) VALUES (?, ?, ?)", "INSERT INTO " + arch + "(name, version, url) VALUES (?, ?, ?)",
@ -57,5 +67,6 @@ if __name__ == "__main__":
elif len(args) == 4: elif len(args) == 4:
check_download(args[1], args[2], args[3], "x86_64") check_download(args[1], args[2], args[3], "x86_64")
else: else:
print("Usage: check.py <name> <version> <url> [arch]\n") print(f"Usage: {args[0]} <package_name> <version> <url> [arch]")
print("arch: x86_64, arm64. default is x86_64") print("options:")
print(" arch: x86_64, arm64. default is x86_64")