path: root/gd_diff.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
#
#  This file is part of gns-deb-diff.
#
#  gns-deb-diff is under the Public Domain. See
#  <https://creativecommons.org/publicdomain/zero/1.0>

import argparse
import json
import os
import re
import shlex
import sys
import xmlrpc.client

import requests

from os import path
from subprocess import run, PIPE
from urllib.parse import urljoin
from xmlrpc.client import ServerProxy, MultiCall, Fault, ProtocolError

from bs4 import BeautifulSoup
from pkg_resources import resource_string

from gns_deb_diff._version import __version__

# list of recognized fields.
field_list = [
    'Change-Type',
    'Changed-From-Debian',
]

# urls
sv_bzr_http = 'http://bzr.savannah.gnu.org'
sv_bzr_gns = '/'.join(['bzr://bzr.savannah.gnu.org', 'gnewsense'])
gns_wiki = 'http://gnewsense.org'

# fmt
readme_link_fmt = '/'.join([sv_bzr_http, 'lh', 'gnewsense',
                            'packages-{}', '{}', 'annotate',
                            'head:', 'debian', 'README.gNewSense'])
readme_url_fmt = '/'.join([sv_bzr_gns, 'packages-{}', '{}','debian',
                           'README.gNewSense'])


def read_file(fpath):
    """Read file `f` and return its content.

    """
    try:
        f = open(fpath, 'r')
    except FileNotFoundError as e:
        print('Error opening \'{}\' \n Error Info:\n  {}'.format(fpath, e),
              file=sys.stderr)
        sys.exit(1)

    return f.read()


def write_file(fpath, content):
    """Write `content` to file `fpath`.

    """
    try:
      f = open(fpath, 'w')
      f.write(content)
      f.close()
    except IOError as e:
        print('Error creating and writing content to {}\n {}'.format(
            fpath, e))
        exit(1)


def execute(cmd, out=None, err=None):
    """Run `cmd`. Returns an instance of `subprocess.CompletedProcess`

    `cmd` must be a string containing the command to run.
    """
    cmd = shlex.split(cmd)

    try:
        completed_process = run(cmd, stdout=out, stderr=err)
    except (FileNotFoundError, OSError, ValueError) as e:
        print("Error running '%s'\n Error Info:\n %r" % (cmd[0], e),
              file=sys.stderr)
        sys.exit(1)

    return completed_process


def read_packages(pkgs_file):
    """Return list contaning of package names from `pkgs_file`.

    """
    pkgs = read_file(pkgs_file).split('\n')
    # sanitize
    pkgs_iter = map(lambda x: x.strip(), pkgs)

    return list(pkgs_iter)


def get_packages(release):
    """Return newline separated list of packages for `release`.

    List of packages is slurped from
        http://bzr.savannah.gnu.org/lh/gnewsense/packages-`release`
    """
    url = 'http://bzr.savannah.gnu.org/lh/gnewsense/packages-{}/'
    req = url.format(release)

    try:
        res = requests.get(req)
    except ConnectionError as ce:
        print('ERROR: Problem GETting {} \n{}'.format(req, ce))
        sys.exit(1)

    if res.status_code != 200:
        print('{}: Error GETting {}'.format(res.status_code, req))
        sys.exit(1)

    html_forest = BeautifulSoup(res.text, 'html.parser')

    pkgs = '' # newline separated list of pkgs.
    for td in html_forest.find_all('td', class_='autcell'):
        pkgs += td.a.string.strip() + '\n'

    return pkgs


def config_dir():
    """Return the gns-deb-diff config directory.

    As a side effect, the directory is created if it does not exist.

    """
    cd = os.path.join(os.getenv('HOME'), '.config', 'gns-deb-diff')

    if not os.path.isdir(cd):
        os.makedirs(cd)

    return cd


def config_file():
    return os.path.join(config_dir(), 'config')


def read_config_file():
    """Return config as a Python Object; False when config does not \
    exist.

    """
    cf = config_file()

    if not os.path.isfile(cf):
        return False

    return json.load(open(cf, 'r'))


def pkgs_dir():
    """Return the `pkgs` directory.

    As a side effect, the directory is created if it does not exist.
    """
    pd = os.path.join(config_dir(), 'pkgs')
    if not os.path.isdir(pd):
        os.mkdir(pd)

    return pd


def mk_pkgs_list(release):
    """Get pkgs for release and write to disk.

    It gets written to `~/.config/pkgs/release`.

    """
    pkgs = get_packages(release)
    pkgs_file = os.path.join(pkgs_dir(), release)
    write_file(pkgs_file, pkgs)

    return pkgs_file


def readmes_dir(release):
    """Return readmes directory for `release`.

    As a side effect, the directory is created if it does not exist.
    """
    rd = os.path.join(config_dir(), 'readmes')
    if not os.path.isdir(rd):
        os.mkdir(rd)

    rd_release = os.path.join(rd, release)
    if not os.path.isdir(rd_release):
        os.mkdir(rd_release)

    return rd_release


def wiki_page_dir(release):
    """Get wiki page directory for `release`.
    """
    wd = os.path.join(config_dir(), 'wiki-page')

    if not os.path.isdir(wd):
        os.mkdir(wd)

    wd_release = os.path.join(wd, release)
    if not os.path.isdir(wd_release):
        os.mkdir(wd_release)

    return wd_release


def wiki_page_path(release):
    """Returns the path to file that contains wiki page for `release`.

    """
    wd_release = wiki_page_dir(release)
    return os.path.join(wd_release, 'wiki.page')


def configured_p():
    """Returns True if gns-deb-diff is configured; False otherwise.
    """
    if os.path.isfile(config_file()):
        return True
    else:
        return False


def configure():
    """Configure gns-deb-diff.
    """
    # prompt username and password.
    config = {}
    config['user'] = input('gNewSense wiki username: ')
    config['pass'] = input('gNewSense wiki password: ')

    json.dump(config, open(config_file(), 'w'))
    os.chmod(config_file(), mode=0o600)


def save_gns_readme(content, release, pkg):
    """Save README.gNewsense locally.

    :param str content:
        Content of the README.gNewsense file.
    :param str release:
        Release name.
    :param str pkg:
        Package name.
    """
    # create gns_readme dir. for pkg.
    gns_readme_dir = path.join(readmes_dir(release), pkg, 'debian')

    try:
        os.makedirs(gns_readme_dir, exist_ok=True)
    except Exception as e:
        print("Error creating directory '%s'\n Error Info:\n %r" %
              (gns_readme_dir, e), file=sys.stderr)
        sys.exit(1)

    gns_readme = path.join(gns_readme_dir, 'README.gNewSense')
    write_file(gns_readme, content)
    print('Saved {}'.format(gns_readme))


def slurp_gns_readme(release, pkg):
    """Read and save the README.gNewSense for `pkg` in `release`.

    """
    readme_url = readme_url_fmt.format(release, pkg)
    cmd = 'bzr cat {}'.format(readme_url)
    cp = execute(cmd, out=PIPE, err=PIPE)

    if(cp.returncode == 0):
        save_gns_readme(cp.stdout.decode(), release, pkg)
        return True
    else:
        print("README.gNewSense not found for package {}".format(pkg),
              file=sys.stderr)
        return False


def slurp_all_gns_readmes(release, pkgs):
    """Read and save all README.gNewSense for `pkgs` in `release`.

    Returns list of packages in `pkgs` that does not have README.gNewSense.
    """
    pkgs_noreadmes = []
    for pkg in pkgs:
        slurped = slurp_gns_readme(release, pkg)

        if(not slurped):
            pkgs_noreadmes.append(pkg)

    return pkgs_noreadmes


def read_gns_readme(release, pkg):
    """Returns content of README.gNewSense for `pkg`.

    If `README.gNewSense` does not exists for `pkg`, None is returned.

    """
    readme_path = path.join(readmes_dir(release), pkg, 'debian',
                            'README.gNewSense')

    if not path.isfile(readme_path):
        return None

    readme_content = read_file(readme_path)
    return readme_content


def slurp_fields_from_readme(content):
    """Returns dict containing fields slurped from `content`.

    - If a field is not defined or if its value is empty in the
    `content`, then its corresponding value in the dict will be None.

    """
    field_values = {}
    for field in field_list:
        pattern = r'{}:[ ]*(.+)'.format(field)
        field_pattern = re.compile(pattern)
        field_match = field_pattern.search(content)

        if (field_match and
            field_match.group(1) and
            field_match.group(1).strip()):
            field_values[field] = field_match.group(1).strip()
        else:
            field_values[field] = None

    return field_values


def get_wiki_page_data(release):
    """Returns data needed to generate the gNewSense Debian Diff table.

    """
    # get packages for release.
    pkgs_file = mk_pkgs_list(release)
    pkgs = read_packages(pkgs_file)

    # get readmes for release.
    pkgs_noreadmes = slurp_all_gns_readmes(release, pkgs)

    # go through each pkg's readme and slurp the fields.
    table_data = {}
    for pkg in pkgs:
        readme_content = read_gns_readme(release, pkg)
        if readme_content:
            table_data[pkg] = slurp_fields_from_readme(readme_content)

    return pkgs_noreadmes, table_data


def construct_table_row(release, pkg, change, reason):
    """Return a table row in moinmoin wiki markup.

    """
    if change is None:
        change = ' '
    if reason is None:
        reason = ' '

    more_info_link = readme_link_fmt.format(release, pkg)

    return '||{}||{}||{}||[[{}|more_info]]||'.format(pkg, change, reason,
                                                   more_info_link)


def generate_wiki_table(release):
    """Generate and return the gNewSense Debian Diff table as a string.
    """
    pkgs_noreadmes, table_data = get_wiki_page_data(release)

    wiki_table = ''
    for pkg, fields in table_data.items():
        change = fields['Change-Type']
        reason = fields['Changed-From-Debian']
        wiki_table += construct_table_row(release, pkg, change, reason) + '\n'

    return pkgs_noreadmes, wiki_table


def gns_wiki_header():
    """Return gNewSense wiki header."""
    header = resource_string(__name__, 'gns_deb_diff/data/wiki-header.txt')
    return header.decode()


def generate_wiki_page(release):
    """Generate and return the gNewSense Debian Diff wiki page.
    """
    pkgs_noreadmes, wiki_table = generate_wiki_table(release)
    wiki_page = gns_wiki_header() + '\n' + wiki_table

    return pkgs_noreadmes, wiki_page


def read_wiki_page(release):
    """Read wiki page for `release` from disk.
    """
    wp_file = wiki_page_path(release)

    if not path.isfile(wp_file):
        return None

    return read_file(wp_file)


def write_wiki_page(release, content):
    """Write wiki page `content` to `release`' last.rev file.

    """
    wp_file = wiki_page_path(release)
    write_file(wp_file, content)


def get_wiki_mc(url, user, passwd): # pragma: no cover
    """Return instance of `xmlprc.client.MultiCall` object.

    """
    url = urljoin(url, '?action=xmlrpc2')
    conn = ServerProxy(url, allow_none=True)

    try:
        token = conn.getAuthToken(user, passwd)
    except Fault as f:
        print(f.faultString)
        exit(1)
    except ProtocolError as p:
        print('Protocal Error: {} {}'.format(p.errcode, p.errmsg))
        exit(1)

    mc = MultiCall(conn)
    mc.applyAuthToken(token)

    return mc


def push_wiki_page(url, user, passwd, version, content):
    """Push `release`' wiki page to wiki at `uri`.

    """
    def process_result(r, f=None):
        if(r == 'SUCCESS'):
            print('auth success.')
        elif(r == True):
            print('wiki page updated.')
        elif(f and f.faultCode == 'INVALID'):
            print('auth failure')
        elif(f and f.faultCode == 1):
            print('wiki page not updated.')


    page = '/'.join(['Documentation', str(version), 'DifferencesWithDebian'])
    mc = get_wiki_mc(url, user, passwd)
    mc.putPage(page, content)

    results = mc()
    try:
        for r in results:
            process_result(r)
    except Fault as f:
        process_result(None, f)


def make_push(args):
    """make wiki page and push it.
    """
    release = args.release
    version = args.version

    # read previously generated wiki page for release
    old_wiki_page = read_wiki_page(release)

    # freshly generate wiki page
    pkgs_noreadmes, wiki_page = generate_wiki_page(release)

    if old_wiki_page == wiki_page:
        print('no changes.')
        return

    # configure if needed.
    if not configured_p():
        configure()

    # read configuration.
    config = read_config_file()

    write_wiki_page(release, wiki_page)
    push_wiki_page(gns_wiki, config['user'], config['pass'], version, wiki_page)

    return config, pkgs_noreadmes, old_wiki_page, wiki_page


def get_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('--version', action='version', version=__version__)
    parser.add_argument('release', help='gNewSense release name')
    parser.add_argument('version', help='gNewSense version number',
                        type=int)

    return parser.parse_args()


def main():
    args = get_args()
    make_push(args)