diff --git a/netkan/netkan/spacedock_adder.py b/netkan/netkan/spacedock_adder.py index 04d7154..69f514f 100644 --- a/netkan/netkan/spacedock_adder.py +++ b/netkan/netkan/spacedock_adder.py @@ -6,8 +6,11 @@ from collections import defaultdict, deque import logging from typing import Dict, Deque, Any, List, Optional, Type, TYPE_CHECKING +import urllib.parse import git from ruamel.yaml import YAML +from github.Repository import Repository +from github import Github from .cli.common import Game from .github_pr import GitHubPR @@ -29,9 +32,10 @@ class SpaceDockAdder: PR_BODY_TEMPLATE = Template(read_text('netkan', 'sd_adder_pr_body_template.md')) USER_TEMPLATE = Template('[$username]($user_url)') TITLE_TEMPLATE = Template('Add $name from $site_name') + GITHUB_PATH_PATTERN = re.compile(r'^/([^/]+)/([^/]+)') _info: Dict[str, Any] - def __init__(self, message: Message, nk_repo: NetkanRepo, game: Game, github_pr: Optional[GitHubPR] = None) -> None: + def __init__(self, message: Message, nk_repo: NetkanRepo, game: Game, github_pr: GitHubPR) -> None: self.message = message self.nk_repo = nk_repo self.game = game @@ -52,13 +56,13 @@ def try_add(self) -> bool: netkan = self.make_netkan(self.info) # Create .netkan file or quit if already there - netkan_path = self.nk_repo.nk_path(netkan.get('identifier', '')) + netkan_path = self.nk_repo.nk_path(netkan[0].get('identifier', '')) if netkan_path.exists(): # Already exists, we are done return True # Create and checkout branch - branch_name = f"add/{netkan.get('identifier')}" + branch_name = f"add/{netkan[0].get('identifier')}" with self.nk_repo.change_branch(branch_name): # Create file netkan_path.write_text(self.yaml_dump(netkan)) @@ -102,30 +106,40 @@ def _pr_body(info: Dict[str, Any]) -> str: logging.error('Failed to generate pull request body from %s', info) raise exc - def yaml_dump(self, obj: Dict[str, Any]) -> str: + def yaml_dump(self, objs: List[Dict[str, Any]]) -> str: sio = io.StringIO() - self.yaml.dump(obj, sio) + self.yaml.dump_all(objs, sio) return sio.getvalue() @staticmethod def sd_download_url(info: Dict[str, Any]) -> str: return f"https://spacedock.info/mod/{info.get('id', '')}/{info.get('name', '')}/download" - def make_netkan(self, info: Dict[str, Any]) -> Dict[str, Any]: + def make_netkan(self, info: Dict[str, Any]) -> List[Dict[str, Any]]: + netkans = [] ident = re.sub(r'[\W_]+', '', info.get('name', '')) + gh_repo = self.get_github_repo(info.get('source_link', '')) + if gh_repo is not None: + gh_netkan = self.make_github_netkan(ident, gh_repo, info) + if gh_netkan is not None: + netkans.append(gh_netkan) + netkans.append(self.make_spacedock_netkan(ident, info)) + return netkans + + def make_spacedock_netkan(self, ident: str, info: Dict[str, Any]) -> Dict[str, Any]: mod: Optional[ModAnalyzer] = None props: Dict[str, Any] = {} url = SpaceDockAdder.sd_download_url(info) try: mod = ModAnalyzer(ident, url, self.game) props = mod.get_netkan_properties() if mod else {} - except Exception as exc: # pylint: disable=broad-except + except Exception as exc: # pylint: disable=broad-except # Tell Discord about the problem and move on logging.error('%s failed to analyze %s from %s', self.__class__.__name__, ident, url, exc_info=exc) vref_props = {'$vref': props.pop('$vref')} if '$vref' in props else {} return { - 'spec_version': 'v1.18', + 'spec_version': 'v1.34', 'identifier': ident, '$kref': f"#/ckan/spacedock/{info.get('id', '')}", **(vref_props), @@ -134,6 +148,65 @@ def make_netkan(self, info: Dict[str, Any]) -> Dict[str, Any]: 'x_via': f"Automated {info.get('site_name')} CKAN submission" } + def get_github_repo(self, source_link: str) -> Optional[Repository]: + url_parse = urllib.parse.urlparse(source_link) + if url_parse.netloc == 'github.com': + match = self.GITHUB_PATH_PATTERN.match(url_parse.path) + if match: + repo_name = '/'.join(match.groups()) + g = Github(self.github_pr.token) + try: + return g.get_repo(repo_name) + except Exception as exc: # pylint: disable=broad-except + # Tell Discord about the problem and move on + logging.error('%s failed to get the GitHub repository from SpaceDock source url %s', + self.__class__.__name__, source_link, exc_info=exc) + return None + return None + + def make_github_netkan(self, ident: str, gh_repo: Repository, info: Dict[str, Any]) -> Optional[Dict[str, Any]]: # pylint: disable=too-many-locals + mod: Optional[ModAnalyzer] = None + props: Dict[str, Any] = {} + try: + latest_release = gh_repo.get_latest_release() + except: # pylint: disable=broad-except,bare-except + logging.warning('No releases found on GitHub for %s, omitting GitHub section', ident) + return None + tag_name = latest_release.tag_name + digit = re.search(r"\d", tag_name) + version_find = '' + if digit: + version_find = tag_name[:digit.start()] + assets = latest_release.assets + if len(assets) == 0: + logging.warning('Release for %s has no assets, omitting GitHub section', ident) + return None + url = assets[0].browser_download_url + try: + mod = ModAnalyzer(ident, url, self.game) + props = mod.get_netkan_properties() if mod else {} + except Exception as exc: # pylint: disable=broad-except + # Tell Discord about the problem and move on + logging.error('%s failed to analyze %s from %s', + self.__class__.__name__, ident, url, exc_info=exc) + vref_props = {'$vref': props.pop('$vref')} if '$vref' in props else {} + netkan = { + 'spec_version': 'v1.34', + 'identifier': ident, + '$kref': f"#/ckan/github/{gh_repo.full_name}", + **(vref_props), + 'license': info.get('license', '').strip().replace(' ', '-'), + **(props), + } + + if version_find != '': + netkan['x_netkan_version_edit'] = { + 'find': f'^{version_find}', + 'replace': '', + 'strict': 'false' + } + return netkan + @property def delete_attrs(self) -> DeleteMessageBatchRequestEntryTypeDef: return { diff --git a/netkan/netkan/status.py b/netkan/netkan/status.py index 6bcd851..95e68f1 100644 --- a/netkan/netkan/status.py +++ b/netkan/netkan/status.py @@ -47,7 +47,7 @@ class Meta: release_date = UTCDateTimeAttribute(null=True) success = BooleanAttribute() frozen = BooleanAttribute(default=False) - resources: 'MapAttribute[str, Any]' = MapAttribute(default={}) + resources: 'MapAttribute[str, Any]' = MapAttribute() def mod_attrs(self) -> Dict[str, Any]: attributes = {}