ngn
ab580d6a18
Some checks failed
format / build (push) Failing after 6s
Signed-off-by: ngn <ngn@ngn.tf>
79 lines
2.1 KiB
Python
79 lines
2.1 KiB
Python
from os.path import dirname, basename
|
|
from urllib.parse import urlencode
|
|
import urllib.parse as urlparse
|
|
from .util import validate_url
|
|
import requests as req
|
|
from typing import List
|
|
|
|
|
|
class provider:
|
|
def __init__(self, url: str) -> None:
|
|
parsed = validate_url(url)
|
|
|
|
if parsed is None:
|
|
raise Exception("invalid provider URL: %s" % url)
|
|
|
|
self.host = parsed.hostname
|
|
self.protocol = parsed.scheme
|
|
self.owner = dirname(parsed.path)
|
|
self.repo = basename(parsed.path)
|
|
|
|
if self.owner[0] == "/":
|
|
self.owner = self.owner[1:]
|
|
|
|
if self.repo[0] == "/":
|
|
self.repo = self.repo[1:]
|
|
|
|
if self.owner == "" or self.repo == "":
|
|
raise Exception("invalid provider URL: %s" % url)
|
|
|
|
def url(self, url: str, queries={}) -> str:
|
|
url = url.replace("HOST", self.host)
|
|
url = url.replace("PROTO", self.protocol)
|
|
url = url.replace("OWNER", self.owner)
|
|
url = url.replace("REPO", self.repo)
|
|
|
|
parsed = list(urlparse.urlparse(url))
|
|
|
|
query = dict(urlparse.parse_qsl(parsed[4]))
|
|
query.update(queries)
|
|
|
|
parsed[4] = urlencode(query)
|
|
url = urlparse.urlunparse(parsed)
|
|
|
|
return url
|
|
|
|
# send a HTTP GET request to the provider
|
|
def GET(self, url: str, queries={}, code=200) -> dict:
|
|
url = self.url(url, queries=queries)
|
|
|
|
# res = req.get(url, headers={
|
|
# "User-Agent": "ups (https://git.ngn.tf/ngn/ups)",
|
|
# })
|
|
|
|
res = req.get(url)
|
|
|
|
if res.status_code != code:
|
|
raise Exception(
|
|
"expected %d from %s, received %d"
|
|
% (code, url, res.status_code)
|
|
)
|
|
|
|
return res.json()
|
|
|
|
# check if the provider is available
|
|
def check(self) -> bool:
|
|
return False
|
|
|
|
# get all the commits until the target commit
|
|
def until(self, commit: str) -> List[str]:
|
|
return []
|
|
|
|
# get last "count" commits
|
|
def last(self, count=1) -> List[str]:
|
|
return []
|
|
|
|
# download the patch for a commit
|
|
def download(self, commit: str, file: str) -> None:
|
|
return False
|