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}")