#!/usr/bin/env python
# Copyright 2011 Corentin Chary <corentin.chary@gmail.com>
# Copyright 2020-2023 src_prepare group
# Distributed under the terms of the GNU General Public License v2
import gzip
import logging
import os
import re
import shutil
import sys
import tarfile
import urllib
from difflib import unified_diff
from tempfile import mkstemp
from BeautifulSoup import BeautifulSoup, SoupStrainer
from gentoolkit.query import Query
from portage.exception import AmbiguousPackageName
logger = logging.getLogger(__name__)
# From portage-janitor
def guess_indent_values(before):
rindent = -1
indent = -1
tab = False
def guess_for_tags(tags):
for tag in tags:
for i in [0, 2, 4, 6, 8, 12, 16]:
if "\n%s<%s" % (" " * i, tag) in before:
return i, False
for i in [0, 1, 2]:
if "\n%s<%s" % ("\t" * i, tag) in before:
return i, True
return -1, False
rindent, tab = guess_for_tags(
["herd", "maintainer", "longdescription", "use", "upstream"]
)
if rindent == -1:
rindent = 2
rindent_str = ("\t" if tab else " ") * rindent
indent, tab = guess_for_tags(["watch", "name", "email"])
if indent == -1:
indent = rindent * 2 if rindent else 4
if rindent and rindent_str == "\t":
tab = True
indent_str = ("\t" if tab else " ") * indent
return rindent_str, indent_str
def get_watch_data(package):
deb_url, deb_type = get_deb_url(package.name)
if deb_type == "source":
return handle_source(deb_url)
if deb_type == "diff":
return handle_diff(deb_url)
def handle_diff(deb_url):
_, temp_deb = mkstemp()
logger.info(" Downloading debian diff %s...", deb_url)
urllib.urlretrieve(deb_url, temp_deb)
watch_data = ""
fp = gzip.open(temp_deb, "rb")
for line in fp:
if re.match(r"\+\+\+ .+?/debian/watch", line):
fp.readline() # diff lines, don't care
cur_line = fp.readline()
while cur_line.startswith("+"):
watch_data += cur_line[1:]
cur_line = fp.readline()
fp.close()
os.unlink(temp_deb)
return watch_data
def handle_source(deb_url):
_, temp_deb = mkstemp()
temp_dir = os.path.dirname(temp_deb)
logger.info(" Downloading debian source %s...", deb_url)
urllib.urlretrieve(deb_url, temp_deb)
tar = tarfile.open(temp_deb)
watch_data = None
try:
tar.extract("debian/watch", temp_dir)
except KeyError:
pass
else:
debian_path = os.path.join(temp_dir, "debian")
watch_path = os.path.join(debian_path, "watch")
watch_data = open(os.path.join(watch_path)).read()
shutil.rmtree(debian_path)
os.unlink(temp_deb)
return watch_data
def get_deb_url(name):
deb_url = None
deb_type = None
while not deb_url:
url = "http://packages.debian.org/source/unstable/%s" % name
opened = urllib.urlopen(url)
content = opened.read()
for link in BeautifulSoup(content, parseOnlyThese=SoupStrainer("a")):
if re.match(r"[^\s]+\.debian\.tar\.(?:gz|bz2)", link.text):
deb_url = link["href"]
deb_type = "source"
break
if re.match(r"[^\s]+\.diff\.gz", link.text):
deb_url = link["href"]
deb_type = "diff"
break
if not deb_url:
logger.error(" Cannot get package from %s" % url)
name = input(" Package name in Debian: ")
return deb_url, deb_type
def patch_metadata(package, watch_data, diff=False):
logger.info(" Patching metadata file")
metadata_path = package.metadata.metadata_path
with open(metadata_path) as fp:
original = fp.read()
rindent, indent = guess_indent_values(original)
data = original
# clean watch_data
watch_data = "\n".join(
[line for line in watch_data.split("\n") if not line.startswith("#")]
) # comments
watch_data = watch_data.replace("\\\n", "") # remove backslashes
watch_tags = []
for watch_line in watch_data.split("\n"): # there can be multiple lines
watch_line = " ".join(watch_line.split()) # remove extra spaces and \n
version_parse = re.match(r"version=(\d+?)", watch_line)
if version_parse:
version = version_parse.group(1)
continue
if not watch_line: # skip empty lines
continue
# parse watch_line
result = re.match(r'(?:opts=(?:"([^"]+?)"|([^\s]+?)) )?(.*)', watch_line)
opts_quote, opts, url = result.groups()
opts = opts_quote or opts
if opts:
# clean opts, skip useless ones
valid = ("uversionmangle", "versionmangle", "downloadurlmangle")
cleaned_opts = []
for opt in opts.split(","):
opt_name, opt_value = opt.split("=", 1)
if opt_name in valid:
if opt_name == "uversionmangle":
opt_name = "versionmangle"
cleaned_opts.append('%s="%s"' % (opt_name, opt_value))
opts = " ".join(cleaned_opts)
# clean url from useless stuff. Just keep <base> [<filepattern>]
url_search = re.search(r"^([^\s]+)(?: ([^\s]*\([^\s]+\)[^\s]*))?", url)
url = " ".join([x for x in url_search.groups() if x is not None])
if opts:
watch_tag = '%s<watch version="%s" %s>%s</watch>' % (
indent,
version,
opts,
url,
)
else:
watch_tag = '%s<watch version="%s">%s</watch>' % (indent, version, url)
watch_tags.append(watch_tag)
watch_tags = "\n".join(watch_tags)
if "<upstream>" in data:
data = data.replace("<upstream>", "<upstream>\n%s" % watch_tags, 1)
else:
rep = "%s<upstream>\n%s\n%s</upstream>\n</pkgmetadata>" % (
rindent,
watch_tags,
rindent,
)
data = data.replace("</pkgmetadata>", rep, 1)
if not diff:
return data
else:
# Generate clean a/category/package/metadata.xml path
n = metadata_path.find(package.category)
if n != -1:
metadata_path = metadata_path[n:]
res = unified_diff(
original.splitlines(True),
data.splitlines(True),
fromfile=os.path.join("a/", metadata_path),
tofile=os.path.join("b/", metadata_path),
)
return "".join([x for x in res])
def process_package(query, diff=False):
try:
matches = Query(query).smart_find(
in_installed=True,
in_porttree=True,
in_overlay=True,
include_masked=True,
show_progress=False,
no_matches_fatal=False,
)
except AmbiguousPackageName:
logger.error(" Ambiguous package name")
return None
if len(matches) == 0:
logger.error(" Package not found")
return None
matches = sorted(matches)
package = matches.pop()
if "9999" in package.version and len(matches) > 0:
package = matches.pop()
watch_data = get_watch_data(package)
if watch_data is None:
logger.error(" No watch file found")
else:
return patch_metadata(package, watch_data, diff=diff)
def main():
import optparse
p = optparse.OptionParser(
usage="usage: %prog <package> [<package> [...]]",
)
p.add_option(
"-d",
"--diff",
action="store_true",
dest="diff",
default=False,
help="Outputs a diff",
)
opts, packages = p.parse_args()
logging.basicConfig(stream=sys.stderr, level=logging.INFO, format="%(message)s")
for package in packages:
logger.info("Processing %s..." % package)
result = process_package(package, opts.diff)
if result:
sys.stdout.write(result)
if __name__ == "__main__":
main()