Les choses sérieuses commencent

erminig/controllers/evezh/abstract.py

@@ -0,0 +1,14 @@
+from abc import ABC, abstractmethod
+
+
+class UpstreamSource(ABC):
+    MAX_RETRIES = 2
+    RETRY_DELAY = 2  # secondes
+
+    def __init__(self, name, config):
+        self.name = name
+        self.config = config
+
+    @abstractmethod
+    def get_latest(self):
+        pass

erminig/controllers/evezh/check.py

@@ -0,0 +1,122 @@
+import json
+from pathlib import Path
+import yaml
+
+
+from erminig.controllers.evezh.parsers.github import GitHubSource
+from erminig.controllers.evezh.parsers.http import HttpSource
+from erminig.controllers.evezh.parsers.sourceforge import SourceForgeRSS
+from erminig.handlers.versions import handle_new_version
+from erminig.models.db import ErminigDB
+from erminig.models import upstreams, versions
+
+
+def load_state(path):
+    if Path(path).exists():
+        return json.loads(Path(path).read_text())
+    return {}
+
+
+def save_state(path, state):
+    Path(path).write_text(json.dumps(state, indent=2))
+
+
+def sync_db(config_path):
+    config = get_config(config_path)
+    print(config)
+
+    with ErminigDB() as db:
+        for proj in config:
+            name = proj["name"]
+            type_ = "http"
+            if "github" in proj:
+                type_ = "github"
+                url = proj["github"]
+            elif "sourceforge" in proj:
+                type_ = "sourceforge"
+                url = proj["sourceforge"]
+            else:
+                url = proj.get("url", "")
+
+            pattern = proj.get("pattern", "")
+            file = proj.get("file", None)
+            upstreams.upsert_upstream(db, name, type_, url, pattern, file)
+
+    print("Synchronisation terminée.")
+
+
+def make_source(name, config):
+    if "github" in config:
+        return GitHubSource(name, config)
+    elif "sourceforge" in config:
+        return SourceForgeRSS(name, config)
+    else:
+        return HttpSource(name, config)
+
+
+def apply_template(name, template_name, templates):
+    tpl = templates.get(template_name)
+    if not tpl:
+        raise ValueError(f"Template '{template_name}' non trouvé.")
+    return {
+        "url": tpl["url"].replace("@NAME@", name),
+        "pattern": tpl["pattern"].replace("@NAME@", name),
+    }
+
+
+def get_config(path):
+    with open(path) as f:
+        full = yaml.safe_load(f)
+
+    templates = full.get("templates", {})
+    projects = full.get("projects", [])
+
+    resolved = []
+
+    for proj in projects:
+        name = proj["name"]
+        new_proj = proj.copy()
+
+        if "template" in proj:
+            tpl = apply_template(name, proj["template"], templates)
+            new_proj["url"] = tpl["url"]
+            new_proj["pattern"] = tpl["pattern"]
+
+        resolved.append(new_proj)
+
+    return resolved
+
+
+def check_versions(config_path, state=None):
+    results = []
+    with ErminigDB() as db:
+        for row in upstreams.get_all_upstreams(db):
+            proj = dict(row)
+            if proj["type"] == "github":
+                proj["github"] = proj["url"]
+            elif proj["type"] == "sourceforge":
+                proj["sourceforge"] = proj["url"]
+            else:
+                proj["url"] = proj["url"]
+            upstream_id = proj["id"]
+            name = proj["name"]
+            source = make_source(name, proj)
+            latest = source.get_latest()
+            if latest:
+                results.append(latest)
+                version = latest["version"]
+                url = latest["url"]
+                handle_new_version(db, upstream_id, name, version, url)
+            elif state and name in state:
+                print(f"[{name}] Serveur HS. On garde l’ancienne version.")
+                results.append(
+                    {
+                        "name": name,
+                        "version": state[name]["version"],
+                        "url": state[name]["url"],
+                    }
+                )
+            else:
+                print(f"[{name}] Aucune version détectée.")
+
+    return results

erminig/controllers/evezh/parsers/github.py

@@ -0,0 +1,76 @@
+import re
+import requests
+from erminig.config import Config
+from erminig.controllers.evezh.abstract import UpstreamSource
+from erminig.system.retry import retry_on_failure
+
+
+class GitHubSource(UpstreamSource):
+
+    @retry_on_failure()
+    def get_latest(self):
+        repo = self.config["github"]
+        file_template = self.config.get("file")
+
+        latest = self._get_latest_release(repo)
+        if not latest:
+            latest = self._get_latest_tag(repo)
+        if not latest:
+            print(f"[{self.name}] Aucune version détectée.")
+            return None
+
+        version = self.normalize_version(latest["tag"])
+        url = latest.get("url")
+
+        if not url and file_template:
+            filename = file_template.replace("${version}", version)
+            url = f"https://github.com/{repo}/releases/download/{latest['tag']}/{filename}"
+            print(f"[{self.name}] Fallback URL : {url}")
+
+        print(url)
+        return {"name": self.name, "version": version, "url": url or ""}
+
+    def _github_headers(self):
+        headers = {}
+        if Config.GITHUB_TOKEN:
+            headers["Authorization"] = f"token {Config.GITHUB_TOKEN}"
+        return headers
+
+    def _get_latest_release(self, repo):
+        r = requests.get(
+            f"https://api.github.com/repos/{repo}/releases",
+            headers=self._github_headers(),
+        )
+        r.raise_for_status()
+        data = r.json()
+        if not data:
+            return None
+
+        release = data[0]
+        for asset in release.get("assets", []):
+            if asset["browser_download_url"].endswith(".tar.gz"):
+                return {
+                    "tag": release["tag_name"],
+                    "url": asset["browser_download_url"],
+                }
+        return {"tag": release["tag_name"]}
+
+    def _get_latest_tag(self, repo):
+        r = requests.get(
+            f"https://api.github.com/repos/{repo}/tags", headers=self._github_headers()
+        )
+        r.raise_for_status()
+        tags = r.json()
+        if not tags:
+            return None
+        url = f"https://github.com/{repo}/archive/refs/tags/{tags[0]["name"]}.tar.gz"
+        return {"tag": tags[0]["name"], "url": url}
+
+    def normalize_version(self, tag):
+        # Exemples : v2.7.1 → 2.7.1, R_2_7_1 → 2.7.1, expat-2.7.1 → 2.7.1
+        tag = tag.strip()
+        if tag.lower().startswith(("v", "r_", "r")):
+            tag = re.sub(r"^[vVrR_]+", "", tag)
+        tag = tag.replace("_", ".")
+        match = re.search(r"(\d+\.\d+(?:\.\d+)?)", tag)
+        return match.group(1) if match else tag

erminig/controllers/evezh/parsers/http.py

@@ -0,0 +1,45 @@
+import re
+import requests
+from erminig.controllers.evezh.abstract import UpstreamSource
+from erminig.system.retry import retry_on_failure
+
+
+class HttpSource(UpstreamSource):
+
+    @retry_on_failure()
+    def get_latest(self):
+        base_url = self.config["url"]
+        pattern = self.config["pattern"]
+        file_pattern = self.config.get("file")
+
+        response = requests.get(
+            base_url, timeout=10
+        )  # timeout pour éviter de bloquer éternellement
+        response.raise_for_status()
+        html = response.text
+
+        matches = re.findall(pattern, html)
+        if not matches:
+            print(f"[{self.name}] Aucun match avec pattern : {pattern}")
+            return None
+
+        latest_dir = sorted(set(matches), key=self.version_key)[-1]
+        version = self.extract_version(latest_dir)
+
+        if file_pattern:
+            filename = file_pattern.replace("${version}", version)
+            url = f"{base_url}{latest_dir}{filename}"
+        else:
+            url = f"{base_url}{latest_dir}"
+
+        print(url)
+
+        return {"name": self.name, "version": version, "url": url}
+
+    def extract_version(self, path):
+        match = re.search(r"([0-9]+\.[0-9]+(?:\.[0-9]+)?)", path)
+        return match.group(1) if match else path
+
+    def version_key(self, ver):
+        nums = re.findall(r"\d+", ver)
+        return list(map(int, nums))

erminig/controllers/evezh/parsers/sourceforge.py

@@ -0,0 +1,33 @@
+import re
+import requests
+import xml.etree.ElementTree as ET
+from erminig.controllers.evezh.abstract import UpstreamSource
+from erminig.system.retry import retry_on_failure
+
+
+class SourceForgeRSS(UpstreamSource):
+
+    @retry_on_failure()
+    def get_latest(self):
+        rss_url = self.config["sourceforge"]
+        r = requests.get(rss_url, timeout=10)
+        r.raise_for_status()
+        root = ET.fromstring(r.text)
+        items = root.findall(".//item")
+
+        versions = []
+        for item in items:
+            title = item.findtext("title") or ""
+            match = re.search(r"([0-9]+\.[0-9]+(?:\.[0-9]+)?)", title)
+            if match:
+                version = match.group(1)
+                link = item.findtext("link")
+                versions.append((version, link))
+
+        if not versions:
+            print(f"[{self.name}] Aucune version trouvée via RSS.")
+            return None
+
+        latest = sorted(versions, key=lambda x: list(map(int, x[0].split("."))))[-1]
+        print(latest[1])
+        return {"name": self.name, "version": latest[0], "url": latest[1]}

erminig/controllers/govel/build.py

@@ -0,0 +1,33 @@
+import subprocess
+from erminig.system.security import check_root, check_user_exists, run_as_user
+
+check_root
+check_user_exists("pak")
+
+
+@run_as_user("pak")
+def run_build_function(pakva_path):
+    """
+    Exécute la fonction build() du fichier Pakva donné.
+    """
+    try:
+        result = subprocess.run(
+            f"""
+            set -e
+            source "{pakva_path}"
+            build
+            """,
+            shell=True,
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            executable="/bin/bash",
+            text=True,
+        )
+        print(f"[BUILD] Succès : {pakva_path.name}")
+        print(result.stdout)
+        return True
+    except subprocess.CalledProcessError as e:
+        print(f"[BUILD] Échec : {pakva_path.name}")
+        print(e.stderr)
+        return False

erminig/controllers/govel/pakva.py

@@ -0,0 +1,103 @@
+from pathlib import Path
+from erminig.config import Config
+from erminig.system.security import run_as_user
+
+
+class Pakva:
+
+    def __init__(self, name, version, archive):
+        self.name = name
+        self.version = version
+        self.archive = archive
+        self.path = Config.GOVEL_DIR / name[0] / name / "Pakva"
+
+    @run_as_user("pak")
+    def save(self):
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with open(self.path, "w") as f:
+            f.write(
+                f"""\
+name="{self.name}"
+version="{self.version}"
+revision=1
+source=("{self.archive}")
+
+build() {{
+    echo "Build function not implemented"
+}}
+
+pak() {{
+    echo "Packaging function not implemented"
+}}               
+"""
+            )
+
+    @run_as_user("pak")
+    def update_version(self, version, archive, reset_revision=False):
+        if not self.path.exists():
+            raise FileNotFoundError(f"Aucun fichier Pakva pour {self.name}")
+
+        with open(self.path, "r") as f:
+            lines = f.readlines()
+
+        new_lines = []
+        updated_version = False
+        updated_source = False
+
+        for line in lines:
+            if line.startswith("version=") and not updated_version:
+                new_lines.append(f'version="{version}"\n')
+                updated_version = True
+            elif line.startswith("source=") and not updated_source:
+                new_lines.append(f'source=("{archive}")\n')
+                updated_source = True
+            elif reset_revision and line.startswith("revision="):
+                new_lines.append("revision=1\n")
+            else:
+                new_lines.append(line)
+
+        # Ajout si jamais les champs n'existaient pas
+        if not updated_version:
+            new_lines.insert(1, f'version="{version}"\n')
+        if not updated_source:
+            new_lines.append(f'source=("{archive}")\n')
+        if reset_revision and not any(line.startswith("revision=") for line in lines):
+            new_lines.insert(2, "revision=1\n")
+
+        with open(self.path, "w") as f:
+            f.writelines(new_lines)
+
+        self.version = version
+        self.archive = archive
+        print(f"[Pakva] Version mise à jour pour {self.name} -> {version}")
+
+    @classmethod
+    def read(cls, path):
+        """
+        Lit un fichier Pakva et retourne un objet Pakva.
+        """
+        path = Path(path)
+        if not path.exists():
+            raise FileNotFoundError(f"Fichier Pakva introuvable : {path}")
+
+        with path.open() as f:
+            lines = f.readlines()
+
+        name = None
+        version = None
+        archive = None
+
+        for line in lines:
+            if line.startswith("name="):
+                name = line.split("=")[1].strip().strip('"')
+            elif line.startswith("version="):
+                version = line.split("=")[1].strip().strip('"')
+            elif line.startswith("source=("):
+                archive = line.split("(")[1].split(")")[0].strip().strip('"')
+
+        obj = cls(name, version, archive)
+        obj.path = path
+        return obj
+
+    def __repr__(self):
+        return f"<Pakva {self.name}-{self.version}>"