~seb128/langpack-o-matic/import-extra-log

#!/usr/bin/env python

# this is part of langpack-o-matic, by Martin Pitt <martin.pitt@canonical.com>
#
# (C) 2005, 2011 Canonical Ltd.
#
# Update all language-pack-%PKGNAME% source packages (they must be in the state of
# the previous upload). If a locale does not yet have a -base package, it gets
# created.
#
# After this script finishes successfully, there will be a file
# 'updated-packages' which contains the source package directory names of all
# packages that were updated.
#
# Usage: import <archive> <distro-release> <target dir>

import os
import sys
import subprocess
import tempfile
import shutil
import optparse
import logging
import urllib
import json

DEFAULT_MIRROR = 'http://archive.ubuntu.com/ubuntu'

# change working directory to the directory of this script
os.chdir(os.path.dirname(sys.argv[0]))

# import our own libraries
sys.path.append('lib')
import pkg_classify
import localeinfo
import macros
import makepkg
import static_translations

# global variables
distribution = None
locinfo = localeinfo.SupportedLocales()
macros_map = {}  # class -> locale -> LangpackMacros

locale_count = {}  # locale -> #translated strings
pot_count = {}  # domain -> #translatable strings
pot_priority = {}  # domain -> priority

static_tar_dir = None


def parse_argv():
    '''Parse command line options.

    Return (options, args) pair.
    '''
    optparser = optparse.OptionParser('%prog <translation tarball> <distro release> <target dir>')
    optparser.add_option('--mirror', default=DEFAULT_MIRROR,
                         metavar='URL', help='Archive mirror URL')
    optparser.add_option('-s', '--no-static', action='store_false',
                         dest='static', default=True,
                         help='Disable inclusion of static translations (GNOME help)')
    optparser.add_option('-u', '--update', action='store_true',
                         help='tarball only has updated translations, update '
                         'existing packages (will not create new packages)')
    optparser.add_option('--no-classes', action='store_false',
                         dest='classes', default=True,
                         help='Disable splitting by classes (GNOME/KDE/common)')
    optparser.add_option('--class', dest='custom_class',
                         help='build custom class from a package list (needs --pkglist too)')
    optparser.add_option('-t', '--treshold', type='int', metavar='PERCENT',
                         help='Only build language packs that cover at least '
                         'the given percentage of all translatable strings '
                         '(ignored when updating already existing packages)')
    optparser.add_option('-p', '--min-priority', type='int', metavar='PRIORITY',
                         help='Only include domains with at least given priority')
    optparser.add_option('--pkglist',
                         help='file with source package names for --class')
    optparser.add_option('--distribution', default='ubuntu',
                         metavar='NAME', help='Distribution name (default: ubuntu)')
    optparser.add_option('-v', '--verbose', action='store_true', default=False,
                         help='Verbose logging')

    (opts, args) = optparser.parse_args()

    if len(args) != 3:
        optparser.error('incorrect number of arguments; use --help for a short help')

    if opts.custom_class and not opts.pkglist:
        optparser.error('need to specify --pkglist with --class')

    if opts.verbose:
        logging.basicConfig(level=logging.DEBUG, format='%(levelname)s: %(message)s')
    else:
        logging.basicConfig(level=logging.WARNING, format='%(levelname)s: %(message)s')

    return opts, args


def get_custom_class_domains(map_file, packages):
    '''Return set of domains corresponding to packages list'''

    # read domain map.txt
    pkg_domains = {}
    with open(map_file) as f:
        for line in f:
            f = line.split()
            if len(f) != 2:
                continue
            pkg_domains.setdefault(f[0], []).append(f[1])

    domains = set()
    with open(packages) as f:
        for line in f:
            pkg = line.strip()
            if not pkg:
                continue
            if pkg in pkg_domains:
                domains.update(pkg_domains[pkg])
            else:
                logging.debug('package %s is in --pkglist, but not in mapping.txt', pkg)
    return domains


def get_current_macros(cls, locale, version):
    '''Return a LangpackMacros object for the given class and locale.

    The LangpackMacros objects are cached for performace reasons.'''

    loc_map = macros_map.setdefault(cls, {})
    if locale not in loc_map:
        loc_map[locale] = macros.LangpackMacros(distribution, locale, cls, release, version)
    return loc_map[locale]


def package_updated(pkg):
    '''Check if the given package has already been updated (i. e. it appears in
    updated-packages).'''

    if not os.path.isfile('updated-packages'):
        return False
    for p in open('updated-packages'):
        if p.strip() == pkg.strip():
            return True
    return False


def write_po(locale, domain, pkgdir, contents):
    '''Write file contents to pkgdir/data/locale/LC_MESSAGES/domain.po.'''

    logging.debug('Copying %s/%s into package %s', locale, domain, pkgdir)
    try:
        os.makedirs(pkgdir + '/data/' + locale + '/LC_MESSAGES')
    except:
        pass
    dest = '%s/data/%s/LC_MESSAGES/%s.po' % (pkgdir, locale, domain)
    if locale.startswith('en_'):
        # many languages legitimagely have identical strings, such as fr
        # or pt_BR using the same string as English, but pt does not. So only
        # used msgequal for English.
        msgequal = subprocess.Popen(['bin/msgequal', '-', dest],
                                    stdin=subprocess.PIPE)
        msgequal.communicate(contents)
        assert msgequal.returncode == 0
    else:
        f = open(dest, 'w')
        f.write(contents)
        f.close()


def read_po(locale, domain, pkgdir):
    '''Read file contents from pkgdir/data/locale/domain.po.

    Return None if the file does not exist. Strips off surrounding white
    space.'''

    try:
        return open('%s/data/%s/LC_MESSAGES/%s.po' % (pkgdir, locale, domain)).read().strip()
    except:
        return None


def normalize_po(contents):
    '''Return PO contents in a canonical format suitable for comparison.

    Return (normalized, num_strings).
    '''
    if contents is None:
        return (None, 0)

    msgfmt = subprocess.Popen(['/usr/bin/msgfmt', '--statistics', '-o', '-', '-'],
                              stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE)
    try:
        (out, err) = msgfmt.communicate(contents)
    except OSError as e:
        logging.warning('msgfmt failed with OSError: %s, not normalizing', str(e))
        return (contents, 0)
    if msgfmt.returncode:
        logging.warning('msgfmt failed with code %i, not normalizing', msgfmt.returncode)
        return (contents, 0)

    num_strings = int(err.split()[0])

    msgunfmt = subprocess.Popen(['/usr/bin/msgunfmt', '-'],
                                stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                stderr=subprocess.PIPE)
    if msgunfmt.returncode:
        raise Exception('msgunfmt returned with exit code ' + str(msgunfmt.returncode))
    (out, err) = msgunfmt.communicate(out)

    # remove X-Launchpad-Export-Date:
    export_date = out.find('\n"X-Launchpad-Export-Date: ')
    if export_date >= 0:
        out = out[:export_date] + out[out.index('\n', export_date + 1):]
    return (out, num_strings)


def install_po_auto_class(locale, domain, contents, data_version,
                          include_static, update_mode):
    '''Install translation file for automatically classified packages (default).

    locale: Target locale
    domain: Translation domain
    data_version: version number of the PO export
    contents: The actual translation data (PO file contents)
    update_mode: delta tarball mode, do not build new packages

    There is a magic domain None which just causes the class-less base package
    to be created, but nothing installed into it. This is important since the
    common package must always exist. If e. g. just -gnome exist, its
    dependencies are unsatisfyable, and it's missing the extra tarball and
    locales. (LP #422760, #335307)
    '''
    try:
        if domain and classifier:
            cls = classifier.classify_domain(domain)
        else:
            cls = ''
    except KeyError:
        logging.warning('unknown translation domain: %s', domain)
        return
    logging.debug('The domain is classified as %s', cls)

    if cls == 'kde' and release_version >= '12.10':
        logging.debug('Skipping KDE language pack for release %s', release)
        return

    macr = get_current_macros(cls, locale, data_version)

    # workaround for Rosetta exported files without translations
    if contents:
        (ncontents, num_strings) = normalize_po(contents)
        if not ncontents or not num_strings:
            return
        # update #translated strings
        locale_count[locale] = locale_count.get(locale, 0) + num_strings

    update_pkg = macr.subst_string(target_dir + '/sources-update/language-pack-%PKGNAME%')
    base_pkg = macr.subst_string(target_dir + '/sources-base/language-pack-%PKGNAME%-base')

    # if base package does not exist, create it
    if not os.path.isdir(base_pkg):
        if update_mode:
            logging.debug('Skipping %s/%s in update mode, %s does not already exist',
                          locale, domain, base_pkg)
            return

        # determine name of tarball with extra files
        extra_tar = 'extra-files/%s-%s.tar' % (cls, macr['PKGCODE'])
        if not os.path.isfile(extra_tar):
            extra_tar = extra_tar + '.gz'
            if not os.path.isfile(extra_tar):
                extra_tar = None

        # add locales to extra tarball of base package
        locale_tar = None
        if cls == '':
            logging.debug('Creating locale tarball')
            locale_tar = locinfo.create_locale_tar(macr['PKGCODE'])
            if extra_tar is not None:
                raise Exception('Not yet implemented: tarball merging (locale+extra.tar)')
            else:
                extra_tar = locale_tar

        makepkg.make_pkg('skel-base', base_pkg, macr, extra_tar)

        if locale_tar is not None:
            os.unlink(locale_tar)

        # add static translations
        if include_static:
            if not os.path.isdir(static_tar_dir):
                logging.debug('Downloading and preparing static translations...')
                # lazily download static tars, so that we don't have to do it when
                # building updates only
                os.mkdir(static_tar_dir)
                if distribution == 'ubuntu':
                    # we don't always rebuild everything in a release, so
                    # search for tarballs in previous releases too
                    tarballs = static_translations.get_static_translation_tarballs('ubuntu', 'trusty', release)
                else:
                    tarballs = static_translations.get_static_translation_tarballs(distribution, release, release)
                static_translations.create_static_tarballs(tarballs, static_tar_dir)

            static_tar = os.path.join(static_tar_dir, macr['PKGNAME'] + '.tar')
            if os.path.exists(static_tar):
                logging.debug('Adding static tarball %s', static_tar)
                shutil.move(static_tar, os.path.join(base_pkg, 'data', 'static.tar'))

        # sanity check: we just created -base, so the update package should not
        # be present
        if os.path.isdir(update_pkg):
            raise Exception('Inconsistency: just created fresh base, but update package already exists')

        # determine %BASEVERDEP% macro (needs to be postponed until here, since we
        # know where the -base package is, and which version it has, and it is
        # not created yet)
        if 'BASEVERDEP' not in macr:
            macr['BASEVERDEP'] = ' (>= %s)' % makepkg.get_pkg_version(base_pkg)

        # Create an empty update package
        makepkg.make_pkg('skel-update', update_pkg, macr)
    else:
        # determine %BASEVERDEP% macro (needs to be postponed until here, since we
        # know where the -base package is, and which version it has, and is not
        # updated yet)
        if 'BASEVERDEP' not in macr:
            macr['BASEVERDEP'] = ' (>= %s)' % makepkg.get_pkg_version(base_pkg)

    if domain is None:
        return

    # ensure that we always have the common package
    if macr['CLASS']:
        common_base_pkg = macr.subst_string(target_dir + '/sources-base/language-pack-%PKGCODE%-base')
        if not os.path.isdir(common_base_pkg) and not update_mode:
            logging.debug('Creating common package for %s', macr['PKGCODE'])
            install_po_auto_class(locale, None, None, data_version, options.static, False)

    # prefer to change the base package if we already changed it
    if package_updated(base_pkg):
        write_po(locale, domain, base_pkg, contents)
    else:
        if ncontents != normalize_po(read_po(locale, domain, base_pkg))[0] and \
                ncontents != normalize_po(read_po(locale, domain, update_pkg))[0]:
            if not package_updated(update_pkg):
                # if we have an extra tarball, do not install it if the same
                # version is already in the base package
                # XXX: deactivated for now, it costs lots of time, does not
                # respect locales, and is useless ATM
                # if extra_tar:
                #     base_extra_tar = os.path.join(base_pkg, 'data',
                #         os.path.basename(extra_tar))
                #     if os.path.isfile(base_extra_tar) and \
                #         filecmp(extra_tar, base_extra_tar):
                #         extra_tar = None

                makepkg.make_pkg('skel-update', update_pkg, macr)

            write_po(locale, domain, update_pkg, contents)


def install_po_custom_class(locale, domain, contents, data_version, update_mode):
    '''Install translation file for custom class

    locale: Target locale
    domain: Translation domain
    contents: The actual translation data (PO file contents)
    data_version: version number of the PO export
    update_mode: delta tarball mode, do not build new packages
    '''
    macr = get_current_macros(options.custom_class, locale, data_version)
    pkg = macr.subst_string(target_dir + '/sources-%CLASS%/language-pack-%PKGNAME%')
    if not os.path.isdir(pkg):
        if update_mode:
            logging.debug('Skipping %s/%s in update mode, %s does not already exist',
                          locale, domain, pkg)
            return
        else:
            makepkg.make_pkg('skel-customclass', pkg, macr)
    else:
        if not package_updated(pkg):
            makepkg.make_pkg('skel-customclass', pkg, macr)

    write_po(locale, domain, pkg, contents)

    # update #translated strings
    msgfmt = subprocess.Popen(['msgfmt', '--statistics', '-o', '/dev/null', '-'],
                              stdin=subprocess.PIPE, stderr=subprocess.PIPE,
                              universal_newlines=True, env={})
    out = msgfmt.communicate(contents)[1]
    locale_count[locale] = locale_count.get(locale, 0) + int(out.split()[0])


def get_translatable_counts(release):
    '''Get number of translatable strings per language'''

    global pot_count, pot_priority

    url = os.path.join(
        os.environ.get('TRANSLATION_STATS_URL', 'http://people.canonical.com/~people-l10n/data/ubuntu-l10n/'),
        '%s_%s_potemplate-stats.json' % (distribution, release.split('-')[0]))

    f = urllib.urlopen(url)
    data = json.load(f)
    f.close()

    for i in data:
        if i['enabled'] and i['languagepack']:
            pot_count[i['translation_domain']] = i['total']
            pot_priority[i['translation_domain']] = i['priority']


def discard_languages(langs, custom_class=None):
    '''Remove built packages for given languages

    When custom_class is given, only remove packages for that.
    '''
    if not langs:
        return

    packages = []
    with open('updated-packages') as f:
        for pkg in f:
            packages.append(pkg.strip())
    for pkg in list(packages):
        if custom_class and 'sources-%s/' % custom_class not in pkg:
            continue
        for l in langs:
            if pkg.endswith('-' + l) or '-%s-' % l in pkg:
                logging.debug('Discarding %s', pkg)
                shutil.rmtree(pkg)
                packages.remove(pkg)

    with open('updated-packages.cleaned', 'w') as f:
        for pkg in packages:
            f.write(pkg + '\n')
    os.rename('updated-packages.cleaned', 'updated-packages')


#
# main
#

options, args = parse_argv()
(archive_fname, release, target_dir) = args
distribution = options.distribution

if not os.path.isdir(target_dir):
    sys.stderr.write('Target directory does not exist\n')
    sys.exit(1)

release_version = get_current_macros('', 'en_GB.UTF-8', 'invalid').subst_string('%RELEASEVERSION%')
assert release_version, 'no release version for ' + release

# unpack translation tarball
contentdirbase = tempfile.mkdtemp()
try:
    # extract tarball to a temporary dir
    if archive_fname[-4:] == '.tar':
        result = os.spawnlp(os.P_WAIT, 'tar', 'tar', '-C', contentdirbase, '-xf',
                            os.path.abspath(archive_fname))
    elif archive_fname[-7:] == '.tar.gz':
        result = os.spawnlp(os.P_WAIT, 'tar', 'tar', '-C', contentdirbase, '-xzf',
                            os.path.abspath(archive_fname))
    elif archive_fname[-8:] == '.tar.bz2':
        result = os.spawnlp(os.P_WAIT, 'tar', 'tar', '-C', contentdirbase, '--bzip2 -xf',
                            os.path.abspath(archive_fname))
    elif archive_fname[-9:] == '.tar.lzma':
        result = os.spawnlp(os.P_WAIT, 'tar', 'tar', '-C', contentdirbase, '--lzma -xf',
                            os.path.abspath(archive_fname))
    else:
        sys.stderr.write('Unknown tar format')
        result = 1
    if result != 0:
        sys.stderr.write('Error executing tar, aborting')
        sys.exit(1)

    toplevel_dirs = os.listdir(contentdirbase)
    if len(toplevel_dirs) != 1:
        raise Exception('Archive does not contain a single top level directory')

    content_dir = os.path.join(contentdirbase, toplevel_dirs[0])
    static_tar_dir = os.path.join(contentdirbase, 'static-tars')

    # read time stamp
    timestamp_file = os.path.join(content_dir, 'timestamp.txt')
    if not os.path.exists(timestamp_file):
        raise Exception('Archive does not contain a timestamp')
    data_version = open(timestamp_file).read().strip()
    os.unlink(timestamp_file)

    # initialize domain map
    map_file = os.path.join(content_dir, 'mapping.txt')
    if not os.path.exists(map_file):
        raise Exception('Archive does not contain a domain map file (mapping.txt)')
    classifier = None
    if options.custom_class:
        custom_class_domains = get_custom_class_domains(map_file, options.pkglist)
    elif options.classes:
        classifier = pkg_classify.PackageClassificator(release, map_file,
                                                       options.mirror)
    os.unlink(map_file)

    try:
        get_translatable_counts(release)  # need pot_priority for updates, too!
    except ValueError:
        logging.warning('Translations stats missing for %s', release)
        # hack until http://people.canonical.com/~people-l10n/data/ubuntu-l10n/ exists
        # for RTM 15.04
        orig_distribution = distribution
        distribution = 'ubuntu'
        get_translatable_counts(release.replace('15.04', 'xenial'))
        distribution = orig_distribution

    if not options.update:
        total_translatable = 0
        counted_domains = set()

        def count_domain(domain):
            global total_translatable, counted_domains
            if domain in counted_domains:
                return
            counted_domains.add(domain)
            try:
                total_translatable += pot_count[domain]
            except KeyError:
                logging.warning('translation stats are missing domain %s', domain)
    else:
        def count_domain(domain):
            # not necessary in update mode
            pass

    # Process every .po file
    lastlocale = ''
    valid_locale = {}
    for root, dirs, files in os.walk(content_dir):
        for f in files:
            logging.debug('Considering %s', f)
            if not f.endswith('.po'):
                continue

            file = os.path.join(root, f)
            comp = file.split(os.sep)[-3:]

            # Verify that we have locale/LC_MESSAGES/domain.po
            if len(comp) < 3 or comp[2] == '':
                continue
            if comp[1] != 'LC_MESSAGES':
                raise IOError('Invalid file: %s' % file)
            (domain, ext) = os.path.splitext(comp[2])
            if ext != '.po':
                raise IOError('Unknown file type: %s' % file)

            # Ignore ISO files
            if domain.startswith('iso_'):
                continue

            # Verify that we have a known locale
            locale = comp[0]
            if locale != lastlocale:
                logging.debug('--------------------------------')
                logging.debug('Processing locale %s...', locale)
            lastlocale = locale

            # check language
            lang = (locale.split('_')[0]).split('@')[0]
            if not (locinfo.known_language(lang) and locinfo.language_locales(lang)):
                logging.warning('Skipping unknown language: %s', locale)
                continue

            # check country, if present
            noat = locale.split('@')[0]
            if noat.find('_') >= 0:
                (lang, country) = noat.split('_')
                # XXX: hack: ignore invalid/obsolete per-country locales from LP
                if lang not in ('zh', 'en') and noat != 'pt_BR':
                    logging.warning('Skipping obsolete locale: %s', locale)
                    continue
                if not locinfo.known_country(country):
                    logging.warning('Skipping unknown country: %s', locale)
                    continue
            if noat == 'zh':
                # split into zh_* now
                logging.warning('Skipping obsolete locale: %s', locale)
                continue

            # check minimal priority
            if options.min_priority is not None:
                try:
                    if pot_priority[domain] < options.min_priority and pot_priority[domain] > 0:
                        logging.debug('Skipping domain %s with too low priority %i' %
                                      (domain, pot_priority[domain]))
                        continue
                except KeyError:
                    logging.warning('domain %s has no priority' % domain)

            # Everything is fine, install it
            with open(file) as f:
                po_data = f.read()
            if options.custom_class:
                if domain in custom_class_domains:
                    install_po_custom_class(locale, domain, po_data,
                                            data_version, options.update)
                    count_domain(domain)
            else:
                install_po_auto_class(locale, domain, po_data, data_version,
                                      options.static, options.update)
                count_domain(domain)

    if not options.update:
        # translate locale_count into per-language counts; take the dominant locale
        # as determining the coverage
        lang_count = {}
        for locale, count in locale_count.items():
            lang = (locale.split('_')[0]).split('@')[0]
            if lang_count.setdefault(lang, 0) < count:
                lang_count[lang] = count

        # show stats and discard packages which don't reach the treshold
        print('Translated strings per language (%i translatable in total):' % total_translatable)
        discarded = set()
        for l in sorted(lang_count):
            if total_translatable:
                pct = lang_count[l] * 100 // total_translatable
                if options.treshold is not None and pct < options.treshold:
                    if l == 'ckb':
                        status = ' special-cased'
                    else:
                        discarded.add(l)
                        status = ' discarded'
                else:
                    status = ''
                print('  %s\t%i (%i%%)%s' % (l, lang_count[l], pct, status))
            else:
                print('  %s\t%i' % (l, lang_count[l]))

        discard_languages(discarded, options.custom_class)
finally:
    shutil.rmtree(contentdirbase, True)