AJout synchro vers yuno
This commit is contained in:
@@ -0,0 +1,119 @@
|
||||
"""Logique partagee entre sync-apps.py (CLI) et preview.py (endpoint /__catalog)
|
||||
pour recuperer et normaliser la liste des apps installees sur le serveur
|
||||
YunoHost via /etc/yunohost/portal/<domaine>.json.
|
||||
"""
|
||||
import json
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||||
CONFIG_PATH = PROJECT_ROOT / "src" / "config.json"
|
||||
ENV_PATH = PROJECT_ROOT / ".env"
|
||||
|
||||
PALETTE = [
|
||||
"#1db9c3", "#ff5a1f", "#2094f3", "#ffc230", "#00a651", "#e6721a",
|
||||
"#b71c1c", "#f39c12", "#0d3b66", "#3b8a8c", "#568d34", "#609926",
|
||||
"#d24939", "#0066cc", "#8e44ad", "#16a085", "#c0392b", "#2c3e50",
|
||||
]
|
||||
|
||||
|
||||
def load_env():
|
||||
env = {}
|
||||
if not ENV_PATH.exists():
|
||||
return env
|
||||
for line in ENV_PATH.read_text(encoding="utf-8").splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#") or "=" not in line:
|
||||
continue
|
||||
key, value = line.split("=", 1)
|
||||
env[key.strip()] = value.strip()
|
||||
return env
|
||||
|
||||
|
||||
def sync_target(env, arg_target=None, arg_domain=None):
|
||||
"""Determine (ssh_target, domain, self_app_id) depuis .env ou arguments CLI."""
|
||||
sync_ssh_user = env.get("SYNC_SSH_USER", "root")
|
||||
sync_ssh_host = env.get("SYNC_SSH_HOST") or env.get("REMOTE_HOST")
|
||||
portal_domain = env.get("PORTAL_DOMAIN", "")
|
||||
self_app_id = env.get("REMOTE_USER", "")
|
||||
|
||||
ssh_target = arg_target or (f"{sync_ssh_user}@{sync_ssh_host}" if sync_ssh_host else None)
|
||||
domain = arg_domain or portal_domain
|
||||
return ssh_target, domain, self_app_id
|
||||
|
||||
|
||||
def color_of(name):
|
||||
h = 0
|
||||
for ch in name:
|
||||
h = (h * 31 + ord(ch)) & 0xFFFFFFFF
|
||||
return PALETTE[h % len(PALETTE)]
|
||||
|
||||
|
||||
def initials_of(name):
|
||||
words = name.strip().split()
|
||||
if len(words) == 1:
|
||||
return words[0][:2].upper()
|
||||
return (words[0][0] + words[1][0]).upper()
|
||||
|
||||
|
||||
def normalize_url(url):
|
||||
return url.replace("https://", "").replace("http://", "").rstrip("/").lower()
|
||||
|
||||
|
||||
def known_urls(config):
|
||||
urls = set()
|
||||
for app in config.get("pinned", []):
|
||||
urls.add(normalize_url(app["url"]))
|
||||
for group in config.get("groups", []):
|
||||
for app in group["apps"]:
|
||||
urls.add(normalize_url(app["url"]))
|
||||
for app in config.get("ungrouped", []):
|
||||
urls.add(normalize_url(app["url"]))
|
||||
return urls
|
||||
|
||||
|
||||
def fetch_remote_apps(ssh_target, domain):
|
||||
cmd = ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", ssh_target, f"cat /etc/yunohost/portal/{domain}.json"]
|
||||
result = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if result.returncode != 0:
|
||||
raise RuntimeError(result.stderr.strip() or "echec SSH")
|
||||
return json.loads(result.stdout)
|
||||
|
||||
|
||||
def build_app_list(remote, domain, self_app_id, exclude_urls=frozenset()):
|
||||
"""Transforme la reponse brute YunoHost en liste d'apps normalisees
|
||||
(id/name/url/color/initials/icon), triee par nom, en excluant l'app
|
||||
elle-meme et celles dont l'URL est deja connue (exclude_urls)."""
|
||||
remote_apps = remote.get("apps", {})
|
||||
apps = []
|
||||
for perm_key, info in remote_apps.items():
|
||||
if not perm_key.endswith(".main"):
|
||||
continue
|
||||
app_id = perm_key[: -len(".main")]
|
||||
if app_id == self_app_id:
|
||||
continue
|
||||
|
||||
url = info["url"]
|
||||
if not url.startswith("http"):
|
||||
url = "https://" + url
|
||||
if not url.endswith("/"):
|
||||
url += "/"
|
||||
|
||||
if normalize_url(url) in exclude_urls:
|
||||
continue
|
||||
|
||||
name = info.get("label", app_id)
|
||||
app = {
|
||||
"id": app_id,
|
||||
"name": name,
|
||||
"url": url,
|
||||
"color": color_of(name),
|
||||
"initials": initials_of(name),
|
||||
}
|
||||
logo = info.get("logo")
|
||||
if logo:
|
||||
app["icon"] = f"https://{domain}{logo}"
|
||||
apps.append(app)
|
||||
|
||||
apps.sort(key=lambda a: a["name"].lower())
|
||||
return apps
|
||||
Reference in New Issue
Block a user