135 lines
4.3 KiB
Python
135 lines
4.3 KiB
Python
import io
|
|
import os
|
|
import re
|
|
import shutil
|
|
import tarfile
|
|
import zipfile
|
|
from urllib.parse import urlparse
|
|
##
|
|
import requests
|
|
try:
|
|
import git
|
|
_has_git = True
|
|
except ImportError:
|
|
_has_git = False
|
|
##
|
|
from . import constants
|
|
|
|
|
|
class _Upstream(object):
|
|
def __init__(self, upstream_xml, *args, **kwargs):
|
|
self.xml = upstream_xml
|
|
self.uri = None
|
|
self.proto = None
|
|
self.host = None
|
|
self.port = None
|
|
self.path = None
|
|
self.parsed_uri = None
|
|
self.dest = os.path.abspath(os.path.expanduser(self.xml.find('dest').text))
|
|
os.makedirs(os.path.dirname(self.dest), exist_ok = True, mode = 0o0750)
|
|
|
|
def parse_uri(self):
|
|
self.parsed_uri = urlparse(self.uri)
|
|
self.proto = self.parsed_uri.scheme.lower()
|
|
self.host = self.parsed_uri.hostname.lower()
|
|
self.port = int(getattr(self.parsed_uri, 'port', constants.DEF_PORTS[self.proto]))
|
|
self.path = self.parsed_uri.path
|
|
return(None)
|
|
|
|
|
|
class Git(_Upstream):
|
|
upstreamtype = 'git'
|
|
|
|
def __init__(self, upstream_xml, refresh = False, *args, **kwargs):
|
|
if not _has_git:
|
|
raise RuntimeError('The git module (GitPython) is not installed')
|
|
super().__init__(upstream_xml, *args, **kwargs)
|
|
self.repo = None
|
|
self.remote = None
|
|
self.ref = None
|
|
self.refresh = refresh
|
|
git_xml = self.xml.find('git')
|
|
try:
|
|
self.ref_name = git_xml.find('ref').text
|
|
except AttributeError:
|
|
self.ref_name = 'master'
|
|
try:
|
|
self.ref_type = git_xml.find('refType').text
|
|
except AttributeError:
|
|
self.ref_type = 'branch'
|
|
self.uri = git_xml.find('uri').text
|
|
self.parse_uri()
|
|
|
|
def fetch(self):
|
|
if os.path.isdir(self.dest) and not self.refresh:
|
|
self.repo = git.Repo(self.dest)
|
|
elif os.path.isdir(self.dest):
|
|
shutil.rmtree(self.dest)
|
|
if not self.repo:
|
|
self.repo = git.Repo.clone_from(self.uri, self.dest)
|
|
for r in self.repo.remotes:
|
|
if self.uri in r.urls:
|
|
self.remote = r
|
|
break
|
|
if not self.remote:
|
|
self.remote = self.repo.create_remote('bootbox_ipxe_upstream', self.uri)
|
|
self.remote.pull()
|
|
if self.ref_type in ('branch', 'head'):
|
|
try:
|
|
self.ref = self.repo.branches[self.ref_name]
|
|
except IndexError:
|
|
self.ref = self.repo.create_head(self.ref_name, self.remote.refs[self.ref_name])
|
|
self.ref.set_tracking_branch(self.ref_name)
|
|
self.ref.checkout()
|
|
self.repo.head.reset(self.ref, working_tree = True)
|
|
elif self.ref_type in ('tag', 'commit', 'rev'):
|
|
if self.ref_type == 'tag':
|
|
self.ref = self.repo.tags[self.ref_name]
|
|
else:
|
|
self.ref = self.repo.commit(rev = self.ref_name)
|
|
self.repo.head.reset(self.ref, working_tree = True)
|
|
self.repo.git.clean('-xdf')
|
|
self.remote.pull()
|
|
return(None)
|
|
|
|
|
|
class Archive(_Upstream):
|
|
upstreamtype = 'archive'
|
|
|
|
def __init__(self, upstream_xml, *args, **kwargs):
|
|
super().__init__(upstream_xml, *args, **kwargs)
|
|
self.uri = self.xml.find('archive').text
|
|
self.parse_uri()
|
|
self.suffix = re.sub(r'^.*\.(?P<suffix>(tar(\.((g|x)z|bz2?))?|t((g|x)z|bz2?)|zip))$',
|
|
r'\g<suffix>',
|
|
os.path.basename(self.path))
|
|
self.req = None
|
|
try:
|
|
shutil.rmtree(self.dest)
|
|
except FileNotFoundError:
|
|
pass
|
|
os.makedirs(self.dest, mode = 0o0750, exist_ok = True)
|
|
|
|
def fetch(self):
|
|
self.req = requests.get(self.uri)
|
|
buf = io.BytesIO(self.req.content)
|
|
if not self.req.ok:
|
|
raise RuntimeError('There was an error fetching the source')
|
|
if self.suffix == 'zip':
|
|
handler = zipfile.ZipFile(buf, 'r')
|
|
else:
|
|
handler = tarfile.open(fileobj = buf, mode = 'r')
|
|
handler.extractall(self.dest)
|
|
handler.close()
|
|
return(None)
|
|
|
|
|
|
def upstream_parser(upstream_xml):
|
|
arc = upstream_xml.find('archive')
|
|
gitrepo = upstream_xml.find('git')
|
|
if arc:
|
|
return(Archive(upstream_xml))
|
|
elif gitrepo:
|
|
return(Git(upstream_xml))
|
|
return(None)
|