76 lines
2.4 KiB
Python
76 lines
2.4 KiB
Python
import subprocess
|
|
import tarfile
|
|
import os
|
|
|
|
from config import load_config, CONFIG_FILE, get_repos
|
|
|
|
DB_PATH = os.path.expanduser(load_config()["db_path"])
|
|
|
|
def get_package_meta(pkgname: str) -> dict | None:
|
|
for dbfile in os.listdir(DB_PATH):
|
|
if not dbfile.endwith(".db"):
|
|
continue
|
|
meta = _search_db(os.path.join(DB_PATH, dbfile), pkgname)
|
|
if meta:
|
|
return meta
|
|
return None
|
|
|
|
def _search_db(dbfile: str, pkgname: str) -> dict | None:
|
|
with tarfile.open(dbfile, "r:gz") as tar:
|
|
for member in tar.getmembers():
|
|
parts = member.name.split("/")
|
|
if len(parts) != 2 or parts[1] != "desc":
|
|
continue
|
|
entry_name = parts[0].rsplit("-", 2)[0]
|
|
if entry_name != pkgname:
|
|
continue
|
|
f = tar.extractfile(member)
|
|
if f:
|
|
return _parse_desc(f.read().decode("utf-8"))
|
|
|
|
def _parse_desc(content: str) -> dict | None:
|
|
meta = {}
|
|
lines = content.strip().split("\n")
|
|
i = 0
|
|
while i < len(lines):
|
|
if lines[i].startswith("%") and lines[i].endswith("%"):
|
|
key = lines[i].strip("%").lower()
|
|
i += 1
|
|
values = []
|
|
while i < len(lines) and lines[i] != "":
|
|
values.append(lines[i])
|
|
i += 1
|
|
meta[key] = values[0] if len(values) == 1 else values
|
|
i += 1
|
|
return meta
|
|
|
|
def ensure_db():
|
|
sync_dir = load_config()["db_path"] / "sync"
|
|
|
|
if not sync_dir.exists() or not any(sync_dir.glob("*.db")):
|
|
print(":: First run detected, syncing databases...")
|
|
_sync_db()
|
|
|
|
def _sync_db():
|
|
sync_dir = load_config()["db_path"] / "sync"
|
|
sync_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
|
|
for repo_name, server in get_repos().items():
|
|
url = resolve_server_url(server, repo_name)
|
|
db_url = f"{url}/{repo_name}.db"
|
|
dest = sync_dir / f"{repo_name}.db"
|
|
|
|
print(f":: Syncing {repo_name}...")
|
|
try:
|
|
subprocess.run(["curl", "-L", "-o", str(dest), db_url], check=True)
|
|
except subprocess.CalledProcessError as e:
|
|
print(f":: Probobly failed to find repo '{repo_name}' at {db_url}")
|
|
|
|
def resolve_server_url(template: str, repo: str, arch: str = load_config()["arch"]) -> str:
|
|
return (template
|
|
.replace("$repo", repo)
|
|
.replace("$arch", arch)
|
|
.replace("${repo}", repo)
|
|
.replace("${arch}", arch))
|