poikilos
/
EnlivenMinetest
mirror of https://github.com/poikilos/EnlivenMinetest
1
0
Fork 0
Code Issues 2 Projects Releases Wiki Activity
Browse Source

Move API description to metadata to deprecate GiteaRepo subclass.

master
poikilos 3 years ago
parent
ed9806b2b0
commit
d29016227a
1 changed files with 239 additions and 126 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 305
      utilities/enissue.py

305
utilities/enissue.py
View File

@ -62,6 +62,71 @@ def error(*args, **kwargs):
# https://api.github.com/repos/poikilos/EnlivenMinetest/issues/475/timeline # https://api.github.com/repos/poikilos/EnlivenMinetest/issues/475/timeline
verbose = False verbose = False
default_options = {
# 'repo_url': "https://api.github.com/repos/poikilos/EnlivenMinetest",
'repo_url': "https://github.com/poikilos/EnlivenMinetest",
}
sites_users_repos_meta = {
'https://api.github.com': {
'poikilos': {
'EnlivenMinetest': {
'repository_id': "80873867"
}
}
}
}
'''
# ^ Get repo metadata for known repos such as via:
site_users_repos_meta = sites_users_repos_meta.get(instance_url)
if site_users_repos_meta is not None:
user_repos_meta = site_users_repos_meta.get(remote_user)
if user_repos_meta is not None:
repo_meta = user_repos_meta.get(repo_name)
if repo_meta is not None:
if repository_id is None:
repository_id = repo_meta.get('repository_id')
'''
github_defaults = {
'repository_id': "80873867",
'instance_url': "https://api.github.com",
'api_repo_url_fmt': "{instance_url}/repos/{ru}/{rn}",
'api_issue_url_fmt': "{instance_url}/repos/{ru}/{rn}/issues/{issue_no}",
'search_issues_url_fmt': "{instance_url}/search/issues?q=repo%3A{ru}/{rn}+",
# 'base_query_fmt': "?q=repo%3A{ru}/{rn}+",
'search_results_key': "items",
'page_size': 30,
'c_issues_name_fmt': "issues_page={p}{q}.json",
'c_issue_name_fmt': "{issue_no}.json",
'default_query': {'state':'open'},
'hide_events': ['renamed', 'assigned'],
'api_comments_url_fmt': "{instance_url}/repos/{ru}/{rn}/issues/comments",
}
gitea_defaults = {
'repository_id': None,
'api_repo_url_fmt': "{instance_url}/api/v1/repos/{ru}/{rn}",
'api_issue_url_fmt': "{instance_url}/api/v1/repos/{ru}/{rn}/issues/{issue_no}",
'search_issues_url_fmt': "{instance_url}/api/v1/search/issues?q=repo%3A{ru}/{rn}+",
# 'base_query_fmt': "?q=repo%3A{ru}/{rn}+", # TODO: Change for Gitea ??
'search_results_key': "items", # TODO: Change for Gitea ??
'page_size': 30, # TODO: Change for Gitea ??
'c_issues_name_fmt': "issues_page={p}{q}.json",
'c_issue_name_fmt': "{issue_no}.json",
'default_query': {'state':'open'}, # TODO: Change for Gitea ??
'hide_events': ['renamed', 'assigned'],
'api_comments_url_fmt': "{instance_url}/api/v1/repos/{ru}/{rn}/issues/comments",
}
# API documentation:
# https://docs.gitea.io/en-us/api-usage/ says:
# > API Reference guide is auto-generated by swagger and available on: https://gitea.your.host/api/swagger or on [gitea demo instance](https://try.gitea.io/api/swagger)
# > The OpenAPI document is at: https://gitea.your.host/swagger.v1.json
apis = {}
apis["GitHub"] = github_defaults
apis["Gitea"] = gitea_defaults
def debug(msg): def debug(msg):
@ -133,6 +198,7 @@ modes = {
"examples": [" page 2", " page 2 --closed"] "examples": [" page 2", " page 2 --closed"]
}, },
"<#>": { "<#>": {
"parent": "issue",
"help": "Specify an issue number to see details.", "help": "Specify an issue number to see details.",
"examples": [" 1"] "examples": [" 1"]
}, },
@ -147,7 +213,8 @@ match_all_labels = []
def toSubQueryValue(value): def toSubQueryValue(value):
''' '''
Convert the value to one that will fit in a Convert the value to one that will fit in a
key+urlencoded(colon)+value string for GitHub queries. key+urlencoded(colon)+value="+" string (can end in plus, so leave
it on the end to easily add more terms later) for GitHub queries.
This function is copied to multiple scripts so they have no This function is copied to multiple scripts so they have no
dependencies: dependencies:
@ -188,27 +255,20 @@ class Repo:
profile = os.environ['HOME'] profile = os.environ['HOME']
os_user = os.environ.get('USER') os_user = os.environ.get('USER')
def __init__( def __init__(
self, self,
remote_user="poikilos", options,
repo_name="EnlivenMinetest",
repository_id="80873867",
instance_url="https://api.github.com",
api_repo_url_fmt="{instance_url}/repos/{ru}/{rn}",
api_issue_url_fmt="{instance_url}/repos/{ru}/{rn}/issues/{issue_no}",
search_issues_url_fmt="{instance_url}/search/issues?q=repo%3A{ru}/{rn}+",
search_results_key="items",
page_size=30,
c_issues_name_fmt="issues_page={p}{q}.json",
c_issue_name_fmt="{issue_no}.json",
default_query={'state':'open'},
hide_events=['renamed', 'assigned'],
caches_path=None, caches_path=None,
api_comments_url_fmt="{instance_url}/repos/{ru}/{rn}/issues/comments",
): ):
''' '''
Keyword arguments: Keyword arguments:
options -- The options dict have any of the following keys (any
that aren't set will be detected based on the URL--if
there is an api name that corresponds to your site's
API in the apis global dict):
repo_url -- This is required. It can be an API or web URL
as long as it ends with username/reponame (except where
there is no username in the URL).
remote_user -- The repo user may be used in api_repo_url_fmt. remote_user -- The repo user may be used in api_repo_url_fmt.
repo_name -- The repo name may be used in api_repo_url_fmt. repo_name -- The repo name may be used in api_repo_url_fmt.
api_repo_url_fmt -- The format string where {ru} is where a repo api_repo_url_fmt -- The format string where {ru} is where a repo
@ -245,7 +305,81 @@ class Repo:
"~/.cache/enissue/poikilos/EnlivenMinetest/issues". To set "~/.cache/enissue/poikilos/EnlivenMinetest/issues". To set
it later, use the setCachesPath method. it later, use the setCachesPath method.
''' '''
repo_url = options.get('repo_url')
debug("* using URL {}".format(repo_url))
if repo_url is None:
raise ValueError("repo_url is required (API or web URL).")
if repo_url.endswith(".git"):
repo_url = repo_url[:-4]
urlParts = repo_url.split("/")
remote_user = urlParts[-2]
self.api_id = options.get('api_id')
if urlParts[-2] == "repo.or.cz":
# Such as https://repo.or.cz/minetest_treasurer.git
remote_user = "almikes@aol.com" # Wuzzy2
self.api_id = "git_instaweb"
repo_name = urlParts[-1]
if self.api_id is None:
if len(urlParts) > 2:
if "github.com" in urlParts[2]:
# [0] is https:
# [1] is '' (because of //)
self.api_id = "GitHub"
debug("* detected GitHub in {}".format(urlParts))
else:
debug("* no api detected in {}[2]".format(urlParts))
else:
debug("* no generic urlParts were found in "
"".format(urlParts))
else:
debug("* using specified API: {}".format(self.api_id))
if self.api_id is None:
self.api_id = "Gitea"
error(" * assuming API is {}".format(self.api_id))
if self.api_id is None:
raise RuntimeError("api_id is not set")
api_meta = apis.get(self.api_id)
if api_meta is None:
raise NotImplementedError("{} api_id is not implemented"
"".format(self.api_id))
for k, v in api_meta.items():
# Set it to the default if it is None:
if options.get(k) is None:
options[k] = v
debug("* constructing {} Repo".format(self.api_id))
debug(" * detected remote_user \"{}\" in url"
"".format(remote_user))
debug(" * detected repo_name \"{}\" in url"
"".format(repo_name))
if self.api_id == "Gitea":
instance_url = "/".join(urlParts[:-2])
debug(" * detected Gitea url " + instance_url)
else:
instance_url = options.get('instance_url')
if instance_url is None:
raise NotImplementedError("Detecting the instance_url"
" is not implemented for"
" {}".format(self.api_id))
debug(" * detected instance_url " + instance_url)
# NOTE: self.instance_url is set by super __init__ below.
# base_query_fmt = options['base_query_fmt']
# search_issues_url_fmt = \
# "{instance_url}/api/v1/repos/issues/search"+base_query_fmt
self.repository_id = options.get('repository_id')
site_users_repos_meta = sites_users_repos_meta.get(instance_url)
if site_users_repos_meta is not None:
user_repos_meta = site_users_repos_meta.get(remote_user)
if user_repos_meta is not None:
repo_meta = user_repos_meta.get(repo_name)
if repo_meta is not None:
if self.repository_id is None:
self.repository_id = \
repo_meta.get('repository_id')
self.instance_url = instance_url self.instance_url = instance_url
self.rateLimitFmt = ("You may be able to view the issues" self.rateLimitFmt = ("You may be able to view the issues"
" at the html_url, and a login may be" " at the html_url, and a login may be"
" required. The URL \"{}\" is not" " required. The URL \"{}\" is not"
@ -258,25 +392,26 @@ class Repo:
self.repo_name = repo_name self.repo_name = repo_name
self.setCachesPath(caches_path) self.setCachesPath(caches_path)
self.search_results_key = search_results_key self.search_results_key = options.get('search_results_key')
self.page = None self.page = options.get('page')
self.repository_id = repository_id self.c_issue_name_fmt = options['c_issue_name_fmt']
self.c_issue_name_fmt = c_issue_name_fmt self.api_repo_url_fmt = options['api_repo_url_fmt']
self.api_repo_url_fmt = api_repo_url_fmt self.api_issue_url_fmt = options['api_issue_url_fmt']
self.api_issue_url_fmt = api_issue_url_fmt
self.repo_url = self.api_repo_url_fmt.format( self.repo_url = self.api_repo_url_fmt.format(
instance_url=instance_url, instance_url=instance_url,
ru=remote_user, ru=remote_user,
rn=repo_name, rn=repo_name,
) )
self.search_issues_url = search_issues_url_fmt.format( self.search_issues_url_fmt = \
options.get('search_issues_url_fmt')
self.search_issues_url = self.search_issues_url_fmt.format(
instance_url=instance_url, instance_url=instance_url,
ru=remote_user, ru=remote_user,
rn=repo_name, rn=repo_name,
) )
self.api_comments_url_fmt = api_comments_url_fmt self.api_comments_url_fmt = options['api_comments_url_fmt']
self.comments_url = api_comments_url_fmt.format( self.comments_url = self.api_comments_url_fmt.format(
instance_url=instance_url, instance_url=instance_url,
ru=remote_user, ru=remote_user,
rn=repo_name, rn=repo_name,
@ -284,15 +419,16 @@ class Repo:
self.issues_url = self.repo_url + "/issues" self.issues_url = self.repo_url + "/issues"
self.labels_url = self.repo_url + "/labels" self.labels_url = self.repo_url + "/labels"
self.page_size = page_size self.page_size = options['page_size']
self.log_prefix = "@ " self.log_prefix = "@ "
self.c_issues_name_fmt = c_issues_name_fmt self.c_issues_name_fmt = options['c_issues_name_fmt']
self.label_ids = [] # all label ids in the repo self.label_ids = [] # all label ids in the repo
self.labels = [] # all labels in the repo self.labels = [] # all labels in the repo
self.default_query = default_query self.default_query = options['default_query']
self.hide_events = hide_events self.hide_events = options['hide_events']
self.issues = None self.issues = None
self.last_query_s = None
def setCachesPath(self, caches_path): def setCachesPath(self, caches_path):
''' '''
@ -450,6 +586,8 @@ class Repo:
else: else:
debug(" There was no custom query.") debug(" There was no custom query.")
self.last_query_s = query_s
if os.path.isfile(c_path): if os.path.isfile(c_path):
# See <https://stackoverflow.com/questions/7430928/ # See <https://stackoverflow.com/questions/7430928/
@ -511,7 +649,10 @@ class Repo:
response = request.urlopen(query_s) response = request.urlopen(query_s)
except HTTPError as e: except HTTPError as e:
if "Error 410" in str(e): if "Error 410" in str(e):
return None, "The issue was deleted." return (
None,
"The issue was apparently deleted (response 410)."
)
msg = str(e) + ": " + self.rateLimitFmt.format(query_s) msg = str(e) + ": " + self.rateLimitFmt.format(query_s)
return None, msg return None, msg
response_s = decode_safe(response.read()) response_s = decode_safe(response.read())
@ -575,6 +716,10 @@ class Repo:
c_path = None c_path = None
if "?" in url: if "?" in url:
raise NotImplementedError("getCachedJsonDict can't query") raise NotImplementedError("getCachedJsonDict can't query")
# If this is implemented, use:
# c_issues_name_fmt
# Since even the first page should have "page" or something
# to denote there are potentially multiple pages.
''' '''
elif url.startswith(self.comments_url): elif url.startswith(self.comments_url):
# This code is not necessary since it startswith # This code is not necessary since it startswith
@ -798,6 +943,7 @@ class Repo:
cmts_data = self.getCachedJsonDict( cmts_data = self.getCachedJsonDict(
issue_data["comments_url"], issue_data["comments_url"],
refresh=refresh, refresh=refresh,
quiet=True,
) )
data = cmts_data data = cmts_data
@ -1039,77 +1185,25 @@ class Repo:
return {'issue':matching_issue, 'count':match_count} return {'issue':matching_issue, 'count':match_count}
class GiteaRepo(Repo):
def __init__(
self,
repo_url,
# instance_url="https://api.github.com",
# repository_id="80873867",
# api_repo_url_fmt="{instance_url}/repos/{ru}/{rn}",
# api_issue_url_fmt="{instance_url}/repos/{ru}/{rn}/issues/{issue_no}",
# search_issues_url_fmt="{instance_url}/search/issues?q=repo%3A{ru}/{rn}+",
# search_results_key="items",
page_size=30,
# c_issues_name_fmt="issues_page={p}{q}.json",
# c_issue_name_fmt="{issue_no}.json",
default_query={'state':'open'},
hide_events=['renamed', 'assigned'],
caches_path=None,
# api_comments_url_fmt="{instance_url}/repos/{ru}/{rn}/issues/comments",
):
if repo_url.endswith(".git"):
repo_url = repo_url[:-4]
urlParts = repo_url.split("/")
remote_user = urlParts[-2]
repo_name = urlParts[-1]
debug("* constructing GiteaRepo")
debug(" * detected remote_user \"{}\" in url"
"".format(remote_user))
debug(" * detected repo_name \"{}\" in url"
"".format(repo_name))
instance_url = "/".join(urlParts[:-2])
debug(" * detected Gitea url " + instance_url)
# NOTE: self.instance_url is set by super __init__ below.
base_query_fmt = "?q=repo%3A{ru}/{rn}+"
search_issues_url_fmt = \
"{instance_url}/api/v1/repos/issues/search"+base_query_fmt
Repo.__init__(
self,
remote_user=remote_user,
repo_name=repo_name,
repository_id=None,
instance_url=instance_url,
api_repo_url_fmt="{instance_url}/api/v1/repos/{ru}/{rn}",
api_issue_url_fmt="{instance_url}/api/v1/repos/{ru}/{rn}/issues/{issue_no}",
search_issues_url_fmt=search_issues_url_fmt,
search_results_key="items", # TODO: Change for Gitea ??
page_size=page_size, # TODO: Change for Gitea ??
default_query=default_query, # TODO: Change for Gitea ??
hide_events=hide_events,
caches_path=caches_path,
api_comments_url_fmt="{instance_url}/api/v1/repos/{ru}/{rn}/issues/comments",
)
# API documentation:
# https://docs.gitea.io/en-us/api-usage/ says:
# > API Reference guide is auto-generated by swagger and available on: https://gitea.your.host/api/swagger or on [gitea demo instance](https://try.gitea.io/api/swagger)
# > The OpenAPI document is at: https://gitea.your.host/swagger.v1.json
def main(): def main():
global verbose global verbose
mode = None mode = None
repo = Repo()
prev_arg = None prev_arg = None
issue_no = None issue_no = None
state = repo.default_query.get('state') state = None
options = {} options = {}
for k,v in default_options.items():
options[k] = v
search_terms = [] search_terms = []
SEARCH_COMMANDS = ['find', 'AND'] SEARCH_COMMANDS = ['find', 'AND'] # CLI-only commands
caches_path = None caches_path = None
logic = {} logic = {} # CLI-only values
save_key = None save_key = None
collect_options = ['--repo-url'] # Repo init data
collect_logic = ['--copy-meta-to', '--db-type', '--db-user', collect_logic = ['--copy-meta-to', '--db-type', '--db-user',
'--db-password', '--cache-base'] '--db-password', '--cache-base']
# ^ CLI arguments override default_options. For example:
# - repo_url is initially set to default_options['repo_url']
for i in range(1, len(sys.argv)): for i in range(1, len(sys.argv)):
arg = sys.argv[i] arg = sys.argv[i]
isValue = False isValue = False
@ -1129,7 +1223,7 @@ def main():
try: try:
i = int(arg) i = int(arg)
if prev_arg == "page": if prev_arg == "page":
repo.page = i options['page'] = i
isValue = True isValue = True
else: else:
if issue_no is not None: if issue_no is not None:
@ -1160,7 +1254,9 @@ def main():
usage() usage()
exit(0) exit(0)
elif arg in collect_logic: elif arg in collect_logic:
save_key = arg.strip("-") save_key = arg.strip("-").replace("-", "_")
elif arg in collect_options:
save_option = arg.strip("-").replace("-", "_")
elif arg.startswith("--"): elif arg.startswith("--"):
usage() usage()
error("Error: The argument \"{}\" is not valid" error("Error: The argument \"{}\" is not valid"
@ -1189,6 +1285,9 @@ def main():
elif save_key is not None: elif save_key is not None:
logic[save_key] = arg logic[save_key] = arg
save_key = None save_key = None
elif save_option is not None:
options[save_option] = arg
save_option = None
elif arg != "page": elif arg != "page":
mode = "list" mode = "list"
match_all_labels.append(arg) match_all_labels.append(arg)
@ -1197,6 +1296,10 @@ def main():
# It is not a command that will determine meaning for the # It is not a command that will determine meaning for the
# next var. # next var.
prev_arg = None prev_arg = None
debug("options: {}".format(options))
repo = Repo(options)
if mode is None: if mode is None:
if len(match_all_labels) > 1: if len(match_all_labels) > 1:
mode = "list" mode = "list"
@ -1205,6 +1308,9 @@ def main():
if save_key is not None: if save_key is not None:
raise ValueError("--{} requires a space then a value." raise ValueError("--{} requires a space then a value."
"".format(save_key)) "".format(save_key))
if save_option is not None:
raise ValueError("--{} requires a space then a value."
"".format(save_option))
caches_path = logic.get('cache-base') caches_path = logic.get('cache-base')
valid_modes = ["issue"] valid_modes = ["issue"]
print("command metadata: {}".format(logic)) print("command metadata: {}".format(logic))
@ -1243,7 +1349,9 @@ def main():
# TODO: get labels another way, and make this conditional: # TODO: get labels another way, and make this conditional:
# if mode == "list": # if mode == "list":
msg = None msg = None
if (mode != "issue") and (state != repo.default_query.get('state')): if (mode != "issue"):
query = None
if (state != repo.default_query.get('state')):
query = { query = {
'state': state 'state': state
} }
@ -1280,7 +1388,9 @@ def main():
" operation will be attempted without it." " operation will be attempted without it."
" Success will depend on your database type and" " Success will depend on your database type and"
" settings.") " settings.")
dstRepo = GiteaRepo(dstRepoUrl) dstRepo = Repo({
'repo_url': dstRepoUrl,
})
# print("* rewriting Gitea issue {}...".format(issue_no)) # print("* rewriting Gitea issue {}...".format(issue_no))
sys.exit(0) # Change based on return of the method. sys.exit(0) # Change based on return of the method.
@ -1318,8 +1428,11 @@ def main():
refresh=False, refresh=False,
) )
# ^ Never refresh, since that would already have been done. # ^ Never refresh, since that would already have been done.
if state != "open": state_msg = repo.default_query.get('state')
print("(showing {} issue(s))".format(state.upper())) if state_msg is None:
state_msg = repo.last_query_s
if state_msg != "open":
print("(showing {} issue(s))".format(state_msg.upper()))
# ^ such as CLOSED # ^ such as CLOSED
else: else:
debug("* There is no matching_issue; matching manually...") debug("* There is no matching_issue; matching manually...")

Write Preview
Loading…
Cancel
Save