convert simple-update-robot to python

gitee.py

+#!/usr/bin/python3
+"""
+This is a simple script to query that contact person for specific package
+"""
+
+import urllib
+import urllib.request
+import argparse
+import yaml
+import re
+import os.path
+import json
+
+
+class Gitee:
+    def __init__(self):
+        self.secret = open(os.path.expanduser("~/.gitee_token.json"), "r")
+        self.token = json.load(self.secret)
+
+        self.headers = {'User-Agent':'Mozilla/5.0 (Windows NT 6.1; WOW 64; rv:23.0) Gecko/20100101 Firefox/23.0'}
+        self.specfile_url_template = "https://gitee.com/src-openeuler/{package}/raw/master/{specfile}"
+        self.yamlfile_url_template = "https://gitee.com/src-openeuler/{package}/raw/master/{package}.yaml"
+        self.advisor_url_template = "https://gitee.com/openeuler/openEuler-Advisor/raw/master/upstream-info/{package}.yaml"
+        self.specfile_exception_url = "https://gitee.com/openeuler/openEuler-Advisor/raw/master/helper/specfile_exceptions.yaml"
+
+    def get_gitee(self, url, headers=None):
+        if headers == None:
+            req = urllib.request.Request(url=url, headers=self.headers)
+        else:
+            req = urllib.request.Request(url=url, headers=headers)
+        u = urllib.request.urlopen(req)
+        return u.read().decode("utf-8")
+
+    def get_gitee_json(self, url):
+        resp = self.get_gitee(url)
+        return json.loads(resp)
+
+    def get_spec_exception(self):
+        resp = self.get_gitee(self.specfile_exception_url)
+        exps = yaml.load(resp, Loader=yaml.Loader)
+        return exps
+
+    def get_spec(self, pkg):
+        exp = self.get_spec_exception()
+        if pkg in exp:
+            dir_name = exp[pkg]["dir"]
+            file_name = exp[pkg]["file"]
+            specurl = self.specfile_url_template.format(package=pkg, specfile=dir_name + "/" + file_name)
+        else:
+            specurl = self.specfile_url_template.format(package=pkg, specfile=pkg+".spec")
+
+        return self.get_gitee(specurl)
+
+    def get_yaml(self, pkg):
+        yamlurl = self.advisor_url_template.format(package=pkg)
+        resp = self.get_gitee(yamlurl)
+        if re.match("Not found", resp):
+            yamlurl = self.yamlfile_url_template.format(package=pkg)
+            resp = self.get_gitee(yamlurl)
+            if re.match("Not found", resp):
+                print("Cannot find upstream metadata")
+                return False
+            else:
+                return resp
+        else:
+            return False
+
+
+class openEuler_TC:
+    def __init__(self):
+        self.gitee = Gitee()
+        self.list_url = "https://gitee.com/api/v5/repos/openeuler/community/pulls?access_token={token}&state=open&sort=created&direction=desc&page=1&per_page=100"
+        self.desc_url = "https://gitee.com/api/v5/repos/openeuler/community/pulls/{number}/comments?access_token={token}&page=1&per_page=100"
+        self.tc_members = ["myeuler", "cynthia_xh", "shinwell_hu", "dream0819", "hanjun-guo", "xiexiuqi", "zhanghai_lucky"]
+
+
+    def get_prs(self):
+        url = self.list_url.format(token=self.token["access_token"])
+        return self.gitee.get_gitee(url)
+
+    def get_pr_comments(self, number):
+        url = self.desc_url.format(number=number, token=self.token["access_token"])
+        return self.gitee.get_gitee(url)
+
+    def filter_out_tc(self, users):
+        return [x for x in self.tc_members if x in users]
+        
+if __name__ == "__main__":
+    par = argparse.ArgumentParser()
+
+    args = par.parse_args()
+
+    oe_tc = openEuler_TC()
+    PRs = oe_tc.get_prs()
+    for pr in PRs:
+        print("URL: https://gitee.com/openeuler/community/pulls/{number}".format(number=pr["number"]))
+        print("Title: "+pr["title"])
+        comm = oe_tc.get_pr_comments(pr["number"])
+        users = []
+        for c in comm:
+           users.append(c["user"]["login"]) 
+        tc = oe_tc.filter_out_tc(users)
+        print("Currently involved TC members: " + ", ".join(tc))
+        print("\n")
+
+

known-issues/hyphen.yaml

+---
+version_control: github
+src_repo: hunspell/hyphen
+tag_prefix: "^v"
+seperator: "."
+last_query:
+  time_stamp: 2020-06-01 05:35:19.023933660 +00:00
+  raw_data: ''
+query_type: git-ls

simple-update-robot.py

+#!/usr/bin/python3
+# process
+# 1. get URL to download updated version
+#    so far we know the URL in spec is not reliable
+# 2. Change Version to new one
+# 3. Change Source or Source0 if needed
+# 4. Update %changelog
+# 5. try rpmbuild -bb
+# 6. fork on gitee
+# 7. git clone, git add, git commit, git push
+# 8. PR on gitee
+
+from pyrpm.spec import Spec, replace_macros
+import yaml
+import argparse
+import gitee
+import sys
+import subprocess
+import os.path
+import re
+import datetime
+
+def download_source_url(spec, o_ver, n_ver):
+    source = replace_macros(spec.sources[0], spec).replace(o_ver, n_ver) 
+    if re.match(r"%{.*?}", source):
+        print("Extra macros in URL which failed to be expanded")
+        return False
+    elif source.startswith("http") or source.startswith("ftp"):
+        fn = os.path.basename(source)
+        subprocess.call(["curl", "-L", source, "-o", fn])
+        return fn
+    else:
+        print("Not valid URL for Source code")
+        return False
+
+def download_upstream_url(gt, repo, o_ver, n_ver):
+    upstream_yaml = gt.get_yaml(repo)
+    if not upstream_yaml:
+        return False
+
+    rp_yaml = yaml.loads(upstream_yaml, Loader=yaml.Loader)
+    if rp_yaml["version_control"] == "github":
+        url = "https://github.com/{rp}/archive/{nv}.tar.gz".format(rp=rp_yaml["src_repo"], nv=n_ver)     
+        fn = "{rp}.{nv}.tar.gz".format(rp=repo, nv=n_ver)
+        subprocess.call(["curl", "-L", url, "-o", fn])
+        return fn
+    else:
+        print("Handling {vc} is still under developing".format(vc=rp_yaml["version_control"]))
+        return False
+
+def create_spec(repo, spec_str, o_ver, n_ver, src_fn=None):
+    fn = open(repo+".spec", "w")
+    in_changelog = False
+    for l in spec_str.splitlines():
+        if l.startswith("Release:"):
+            fn.write("Release:\t\t0\n")
+            continue
+        if l.startswith("Source:") or l.startswith("Source0:"):
+            if src_fn:
+                fn.write("Source:	{src_fn}\n".format(src_fn=src_fn))
+            else:
+                fn.write(l+"\n")
+            continue
+        if not in_changelog:
+            nl = l.replace(o_ver, n_ver)
+        else:
+            nl = l
+        fn.write(nl+"\n")
+
+        if nl.startswith("%changelog"):
+            in_changelog = True
+            d = datetime.date.today()
+            fn.write(d.strftime("* %a %b %d %Y SimpleUpdate Robot <tc@openeuler.org>\n"))
+            fn.write("- Update to version {ver}\n".format(ver=n_ver))
+            fn.write("\n")
+    fn.close()
+
+if __name__ == "__main__":
+    pars = argparse.ArgumentParser()
+    pars.add_argument("pkg", type=str, help="The package to be upgraded")
+    pars.add_argument("-o", "--old_version", type=str, help="Current upstream version of package")
+    pars.add_argument("-n", "--new_version", type=str, help="New upstream version of package will be upgrade to")
+    args = pars.parse_args()
+
+    gt = gitee.Gitee()
+    spec_string= gt.get_spec(args.pkg)
+
+    spec = Spec.from_string(spec_string)
+    if len(spec.patches) >= 1:
+        print("I'm too naive to handle complicated package.")
+        print("This package has multiple in-house patches.")
+        sys.exit(1)
+
+    source_file = download_source_url(spec, args.old_version, args.new_version)
+    if source_file:
+        print(source_file)
+    else:
+        source_file = download_upstream_url(gt, args.pkg, args.old_version, args.new_version)
+        if source_file:
+            print(source_file)
+        else:
+            print("Failed to download the latest source code.")
+            sys.exit(1)
+
+    create_spec(args.pkg, spec_string, args.old_version, args.new_version)
+
+"""
+
+"""

upstream-info/hyphen.yaml

-version_control: github
-src_repo:  hunspell/hyphen
-tag_prefix: ^v
-seperator: .