about summary refs log blame commit diff stats
path: root/pkgs/by-name/up/update-vim-plugins/update_vim_plugins/plugin.py
blob: 8334ad5398cd6e461cd634a1995fc3c6aabac42a (plain) (tree)




















































































































































































                                                                                                                                                                                                           
import logging
import os
import urllib

import requests
import jsonpickle
from datetime import datetime, date
from dateparser import parse

from .nix import GitSource, License, Source, UrlSource
from .spec import PluginSpec, RepositoryHost


logger = logging.getLogger(__name__)


class VimPlugin:
    """Abstract base class for vim plugins."""

    name: str
    owner: str
    repo: str
    version: date
    source: Source
    description: str = "No description"
    homepage: str
    license: License
    source_line: str
    checked: date = datetime.now().date()

    def to_nix(self):
        """Return the nix expression for this plugin."""
        meta = f'with lib; {{ description = "{self.description}"; homepage = "{self.homepage}"; license = with licenses; [ {self.license.value} ]; }}'
        return f'/* Generated from: {self.source_line} */ {self.name} = buildVimPlugin {{ pname = "{self.name}";  version = "{self.version}"; src = {self.source.get_nix_expression()}; meta = {meta}; }};'

    def to_json(self):
        """Serizalize the plugin to json"""
        return jsonpickle.encode(self)

    def to_markdown(self):
        link = f"[{self.source_line}]({self.homepage})"
        version = f"{self.version}"
        package_name = f"{self.name}"
        checked = f"{self.checked}"

        return f"| {link} | {version} | `{package_name}` | {checked} |"

    def __lt__(self, o: object) -> bool:
        if not isinstance(o, VimPlugin):
            return False

        return self.name.lower() < o.name.lower()

    def __repr__(self):
        """Return the representation of this plugin."""
        return f"VimPlugin({self.name!r}, {self.version.strftime('%Y-%m-%d')})"


def _get_github_token():
    token = os.environ.get("GITHUB_TOKEN")
    if token is None:
        # NOTE: This should never use more than the free api requests <2023-12-09>
        pass
        # logger.warning("GITHUB_TOKEN environment variable not set")
    return token


class GitHubPlugin(VimPlugin):
    def __init__(self, plugin_spec: PluginSpec) -> None:
        """Initialize a GitHubPlugin."""

        full_name = f"{plugin_spec.owner}/{plugin_spec.repo}"
        repo_info = self._api_call(f"repos/{full_name}")
        default_branch = plugin_spec.branch or repo_info["default_branch"]
        api_callback = self._api_call(f"repos/{full_name}/commits/{default_branch}")
        latest_commit = api_callback["commit"]
        sha = api_callback["sha"]

        self.name = plugin_spec.name
        self.owner = plugin_spec.owner
        self.version = parse(latest_commit["committer"]["date"]).date()
        self.source = UrlSource(f"https://github.com/{full_name}/archive/{sha}.tar.gz")
        self.description = (repo_info.get("description") or "").replace('"', '\\"')
        self.homepage = repo_info["html_url"]
        self.license = plugin_spec.license or License.from_spdx_id((repo_info.get("license") or {}).get("spdx_id"))
        self.source_line = plugin_spec.line

    def _api_call(self, path: str, token: str | None = _get_github_token()):
        """Call the GitHub API."""
        url = f"https://api.github.com/{path}"
        headers = {"Content-Type": "application/json"}
        if token is not None:
            headers["Authorization"] = f"token {token}"
        response = requests.get(url, headers=headers)
        if response.status_code != 200:
            raise RuntimeError(f"GitHub API call failed: {response.text}")
        return response.json()


class GitlabPlugin(VimPlugin):
    def __init__(self, plugin_spec: PluginSpec) -> None:
        """Initialize a GitlabPlugin."""

        full_name = urllib.parse.quote(f"{plugin_spec.owner}/{plugin_spec.repo}", safe="")
        repo_info = self._api_call(f"projects/{full_name}")
        default_branch = plugin_spec.branch or repo_info["default_branch"]
        api_callback = self._api_call(f"projects/{full_name}/repository/branches/{default_branch}")
        latest_commit = api_callback["commit"]
        sha = latest_commit["id"]

        self.name = plugin_spec.name
        self.owner = plugin_spec.owner
        self.version = parse(latest_commit["created_at"]).date()
        self.source = UrlSource(f"https://gitlab.com/api/v4/projects/{full_name}/repository/archive.tar.gz?sha={sha}")
        self.description = (repo_info.get("description") or "").replace('"', '\\"')
        self.homepage = repo_info["web_url"]
        self.license = plugin_spec.license or License.from_spdx_id(repo_info.get("license", {}).get("key"))
        self.source_line = plugin_spec.line

    def _api_call(self, path: str) -> dict:
        """Call the Gitlab API."""
        url = f"https://gitlab.com/api/v4/{path}"
        response = requests.get(url)
        if response.status_code != 200:
            raise RuntimeError(f"Gitlab API call failed: {response.text}")
        return response.json()


def _get_sourcehut_token():
    token = os.environ.get("SOURCEHUT_TOKEN")
    if token is None:
        # NOTE: This should never use more than the free requests <2023-12-09>
        pass
        # logger.warning("SOURCEHUT_TOKEN environment variable not set")
    return token


class SourceHutPlugin(VimPlugin):
    def __init__(self, plugin_spec: PluginSpec) -> None:
        """Initialize a SourceHutPlugin."""

        repo_info = self._api_call(f"~{plugin_spec.owner}/repos/{plugin_spec.repo}")
        if plugin_spec.branch is None:
            commits = self._api_call(f"~{plugin_spec.owner}/repos/{plugin_spec.repo}/log")
        else:
            commits = self._api_call(f"~{plugin_spec.owner}/repos/{plugin_spec.repo}/log/{plugin_spec.branch}")
        latest_commit = commits["results"][0]
        sha = latest_commit["id"]

        self.name = plugin_spec.name
        self.owner = plugin_spec.owner
        self.version = parse(latest_commit["timestamp"]).date()
        self.description = (repo_info.get("description") or "").replace('"', '\\"')
        self.homepage = f"https://git.sr.ht/~{plugin_spec.owner}/{plugin_spec.repo}"
        self.source = GitSource(self.homepage, sha)
        self.license = plugin_spec.license or License.UNKNOWN  # cannot be determined via API
        self.source_line = plugin_spec.line

    def _api_call(self, path: str, token: str | None = _get_sourcehut_token()):
        """Call the SourceHut API."""

        url = f"https://git.sr.ht/api/{path}"
        headers = {"Content-Type": "application/json"}
        if token is not None:
            headers["Authorization"] = f"token {token}"
        response = requests.get(url, headers=headers)
        if response.status_code != 200:
            raise RuntimeError(f"SourceHut API call failed: {response.json()}")
        return response.json()


def plugin_from_spec(plugin_spec: PluginSpec) -> VimPlugin:
    """Initialize a VimPlugin."""

    if plugin_spec.repository_host == RepositoryHost.GITHUB:
        return GitHubPlugin(plugin_spec)
    elif plugin_spec.repository_host == RepositoryHost.GITLAB:
        return GitlabPlugin(plugin_spec)
    elif plugin_spec.repository_host == RepositoryHost.SOURCEHUT:
        return SourceHutPlugin(plugin_spec)
    else:
        raise NotImplementedError(f"Unsupported source: {plugin_spec.repository_host}")