Init

src/sources/gitea.py

@@ -0,0 +1,128 @@
+
+from sources import CSSourceGitea, CSItem, CSInventoryItem
+from sources.util import cache_cheatsheet, get_datestring
+import requests
+from pathlib import Path
+
+
+def process_gitea(item: CSInventoryItem, outdir: str) -> list[CSItem] | None:
+    source: CSSourceGitea = item.source
+    commit_hash = get_release_commit_sha(source.base_url, source.owner, source.repo, source.tag)
+    asserts = list_release_assets(source.base_url, source.owner, source.repo, source.tag)
+    
+    asserts = filter(lambda a: a[1].endswith(".pdf"), asserts)
+
+    res = []
+
+    for a in asserts:
+        res_url = a[0]
+        if item.cache:
+            cache_url = cache_cheatsheet(a[0], outdir)
+            if cache_url:
+                res_url = cache_url
+            else:
+                continue
+
+        name = Path(a[1]).stem
+
+        res.append(CSItem(
+            url = res_url,
+            date=get_datestring(),
+            commit=commit_hash[:10] if commit_hash else "",
+            author=item.author if item.author else source.owner,
+            title=f"{name}",
+            id=item.id,
+            git_repo=f"{source.base_url}/{source.owner}/{source.repo}" if not source.hide_repo else "",
+            git_repo_type="Gitea"
+        ))
+
+    return res
+
+def get_release_commit_sha(base_url, owner, repo, tag_name, token=None):
+    """
+    Resolve the commit SHA for a Gitea release tag.
+
+    :param base_url: e.g. "https://gitea.example.com"
+    :param owner: repo owner
+    :param repo: repository name
+    :param tag_name: release tag (e.g. "v1.2.3")
+    :param token: optional API token
+    :return: commit SHA (str)
+    """
+
+    headers = {}
+    if token:
+        headers["Authorization"] = f"token {token}"
+
+    session = requests.Session()
+    session.headers.update(headers)
+
+    # 1) List tags and find the matching tag
+    tags_url = f"{base_url}/api/v1/repos/{owner}/{repo}/tags"
+    resp = session.get(tags_url)
+    resp.raise_for_status()
+    tags = resp.json()
+
+    tag = next((t for t in tags if t["name"] == tag_name), None)
+    if not tag:
+        raise ValueError(f"Tag '{tag_name}' not found")
+
+    # Lightweight tags usually already contain the commit SHA
+    commit_sha = tag.get("commit", {}).get("sha")
+    tag_obj_sha = tag.get("id")
+
+    # If commit.sha looks valid, return it
+    if commit_sha:
+        return commit_sha
+
+    # 2) Annotated tag: dereference via /git/tags/{sha}
+    if not tag_obj_sha:
+        raise RuntimeError("Tag object SHA missing; cannot dereference annotated tag")
+
+    git_tag_url = f"{base_url}/api/v1/repos/{owner}/{repo}/git/tags/{tag_obj_sha}"
+    resp = session.get(git_tag_url)
+    resp.raise_for_status()
+    annotated = resp.json()
+
+    # The object pointed to by the tag (usually a commit)
+    target = annotated.get("object", {})
+    if target.get("type") != "commit":
+        raise RuntimeError(f"Tag points to a {target.get('type')} instead of a commit")
+
+    return target.get("sha")
+
+
+def list_release_assets(base_url, owner, repo, tag, token=None):
+    """
+    Return a list of (download_url, filename) for all assets of a Gitea release.
+
+    :param base_url: Gitea host URL, e.g. "https://gitea.example.com"
+    :param owner: repository owner
+    :param repo: repository name
+    :param tag: release tag name
+    :param token: optional API token
+    :returns: list of (download_url, filename) tuples
+    """
+    headers = {}
+    if token:
+        headers["Authorization"] = f"token {token}"
+
+    # 1) Get release by tag
+    rel_url = f"{base_url}/api/v1/repos/{owner}/{repo}/releases/tags/{tag}"
+    rel_resp = requests.get(rel_url, headers=headers)
+    rel_resp.raise_for_status()
+    release = rel_resp.json()
+
+    assets = release.get("assets", [])
+    result = []
+
+    for asset in assets:
+        # Gitea asset info usually contains:
+        # - "browser_download_url" → direct URL
+        # - "name" → filename
+        download_url = asset.get("browser_download_url")
+        filename = asset.get("name")
+        if download_url and filename:
+            result.append((download_url, filename))
+
+    return result