~launchpad-pqm/launchpad/devel

# Copyright 2009 Canonical Ltd.  This software is licensed under the

# GNU Affero General Public License version 3 (see the file LICENSE).

"""Remove specific translation messages from the database."""

__metaclass__ = type

__all__ = [

    'process_options',

    'RemoveTranslations',

    'remove_translations',

    ]

import logging

from optparse import (

    Option,

    OptionValueError,

    )

from zope.component import getUtility

from lp.services.database.postgresql import drop_tables

from lp.services.database.sqlbase import (

    cursor,

    sqlvalues,

    )

from lp.services.scripts.base import (

    LaunchpadScript,

    LaunchpadScriptFailure,

    )

from lp.translations.interfaces.translationmessage import (

    RosettaTranslationOrigin,

    )

def process_bool_option(value):

    """Validation and conversion for Boolean argument."""

    value = value.lower()

    bool_representations = {

        'true': True,

        '1': True,

        'false': False,

        '0': False,

        }

    if value not in bool_representations:

        raise OptionValueError("Invalid boolean value: %s" % value)

    return bool_representations[value]

def get_id(identifier, lookup_function=None):

    """Look up id of object identified by a string.

    Raises `OptionValueError` if the option's value appears invalid.

    :param identifier: String identifying an object.  If entirely

        numeric, taken as id.  Otherwise, passed to lookup_function.

    :param lookup_function: Callback that will take `identifier` as

        its argument and return a numeric object id.  If no object

        has the given identifier, may raise a `LookUpError` or return

        None.

    :return: Numeric object id, or None if no identifier is given.

    """

    if identifier is None or identifier == '':

        return None

    elif isinstance(identifier, basestring) and identifier == '':

        return None

    elif isinstance(identifier, int):

        return identifier

    elif identifier.isdigit():

        return int(identifier)

    elif lookup_function is None:

        raise OptionValueError("Expected numeric id, got '%s'." % identifier)

    else:

        try:

            result = lookup_function(identifier)

        except LookupError:

            raise OptionValueError("'%s' not found." % identifier)

    if result is None:

        raise OptionValueError("'%s' not found." % identifier)

    return result

def get_person_id(name):

    """`get_id` helper.  Look up person by name."""

    # XXX sinzui 2010-10-04 bug=654537: Account and EmailAddress cause cyclic

    # imports because they are not in the lp tree.

    from lp.registry.interfaces.person import IPersonSet

    person = getUtility(IPersonSet).getByName(name)

    if person is None:

        return None

    return person.id

def get_origin(name):

    """`get_id` helper.  Look up `RosettaTranslationOrigin` by name."""

    try:

        return getattr(RosettaTranslationOrigin, name).value

    except AttributeError:

        return None

def process_origin_option(value):

    """Validation and conversion for `RosettaTranslationsOrigin`."""

    return get_id(value, get_origin)

def process_person_option(value):

    """Validation and conversion for `Person`."""

    return get_id(value, get_person_id)

# Options that need special processing.

OPTIONS_TO_PROCESS = {

    'submitter': process_person_option,

    'reviewer': process_person_option,

    'origin': process_origin_option,

    'is_current_ubuntu': process_bool_option,

    'is_current_upstream': process_bool_option,

    }

def process_options(options):

    """Process options that need special processing."""

    for option_name, process_func in OPTIONS_TO_PROCESS.items():

        option_value = getattr(options, option_name)

        if option_value is not None:

            setattr(options, option_name, process_func(option_value))

def is_nonempty_list(list_option):

    """Is list_option a non-empty a nonempty list of option values?"""

    return list_option is not None and len(list_option) > 0

def is_nonempty_string(string_option):

    """Is string_option a non-empty option value?"""

    return string_option is not None and string_option != ''

def compose_language_match(language_code):

    """Compose SQL condition for matching a language in the deletion query.

    :param: Language code to match.

    :return: SQL condition in string form.

    """

    return 'Language.code = %s' % sqlvalues(language_code)

def add_bool_match(conditions, expression, match_value):

    """Add match for tri-state Boolean to SQL conditions.

    :param conditions: Set of SQL condition clauses to add to.

    :param expression: Variable or other SQL expression to match on.

    :param match_value: If given, the Boolean value to match.  If left

        as None, no condition is added.

    """

    if match_value is None:

        return

    if match_value:

        match = expression

    else:

        match = 'NOT (%s)' % expression

    conditions.add(match)

class RemoveTranslations(LaunchpadScript):

    """Remove specific `TranslationMessage`s from the database.

    The script accepts a wide range of options to specify exactly which

    messages need deleting.  It will refuse to run if the options are so

    non-specific that the command is more likely to be a mistake than a

    valid use case.  In borderline cases, it may be persuaded to run

    using a "force" option.

    """

    description = "Delete matching translation messages from the database."

    loglevel = logging.INFO

    my_options = [

        Option(

            '-s', '--submitter', dest='submitter',

            help="Submitter match: delete only messages with this "

                "submitter."),

        Option(

            '-r', '--reviewer', dest='reviewer',

            help="Reviewer match: delete only messages with this reviewer."),

        Option(

            '-x', '--reject-license', action='store_true',

            dest='reject_license',

            help="Match submitters who rejected the license agreement."),

        Option(

            '-i', '--id', action='append', dest='ids', type='int',

            help="ID of message to delete.  May be specified multiple "

                "times."),

        Option(

            '-p', '--potemplate', dest='potemplate', type='int',

            help="Template id match.  Delete only messages in this "

                "template."),

        Option(

            '-l', '--language', dest='language',

            help="Language match.  Deletes (default) or spares (with -L) "

                 "messages in this language."),

        Option(

            '-L', '--not-language', action='store_true', dest='not_language',

            help="Invert language match: spare messages in given language."),

        Option(

            '-C', '--is-current-ubuntu', dest='is_current_ubuntu',

            help="Match on is_current_ubuntu value (True or False)."),

        Option(

            '-I', '--is-current-upstream', dest='is_current_upstream',

            help="Match on is_current_upstream value (True or False)."),

        Option(

            '-m', '--msgid', dest='msgid',

            help="Match on (singular) msgid text."),

        Option(

            '-o', '--origin', dest='origin',

            help="Origin match: delete only messages with this origin code."),

        Option(

            '-f', '--force', action='store_true', dest='force',

            help="Override safety check on moderately unsafe action."),

        Option(

            '-d', '--dry-run', action='store_true', dest='dry_run',

            help="Go through the motions, but don't really delete."),

        ]

    def add_my_options(self):

        """See `LaunchpadScript`."""

        self.parser.add_options(self.my_options)

    def _check_constraints_safety(self):

        """Are these options to the deletion script sufficiently safe?

        :return: Boolean approval and output message.  All disapprovals come

            with an explanation; some approvals come with an informational

            message.

        """

        if is_nonempty_list(self.options.ids):

            return (True, None)

        if is_nonempty_string(self.options.submitter):

            return (True, None)

        if is_nonempty_string(self.options.reviewer):

            return (True, None)

        forced = self.options.force

        if is_nonempty_string(self.options.potemplate) and forced:

            return (

                True,

                "Safety override in effect.  Deleting translations for "

                "template %s." % self.options.potemplate)

        if self.options.reject_license:

            if self.options.is_current_upstream == False:

                # "Remove non-is_current_upstream messages submitted by users

                # who rejected the license."

                return (True, None)

            rosettaweb_key = RosettaTranslationOrigin.ROSETTAWEB.value

            if self.options.origin == rosettaweb_key:

                # "Remove messages submitted directly in Launchpad by

                # users who rejected the license."

                return (True, None)

            if forced:

                return (

                    True,

                    "Safety override in effect.  Removing translations "

                    "by users who rejected the license, regardless of "

                    "origin.")

            return (

                False,

                "To delete the translations by users who "

                "rejected the translations license, specify at least "

                "--origin=ROSETTAWEB or --is-imported=False.")

        return (

            False,

            "Refusing unsafe deletion.  Use matching options to constrain "

            "deletion to a safe subset.")

    def main(self):

        """See `LaunchpadScript`."""

        process_options(self.options)

        (result, message) = self._check_constraints_safety()

        if not result:

            raise LaunchpadScriptFailure(message)

        if message is not None:

            self.logger.warn(message)

        if self.options.dry_run:

            self.logger.info("Dry run only.  Not really deleting.")

        remove_translations(logger=self.logger,

            submitter=self.options.submitter,

            reject_license=self.options.reject_license,

            reviewer=self.options.reviewer,

            ids=self.options.ids,

            potemplate=self.options.potemplate,

            language_code=self.options.language,

            not_language=self.options.not_language,

            is_current_ubuntu=self.options.is_current_ubuntu,

            is_current_upstream=self.options.is_current_upstream,

            msgid_singular=self.options.msgid,

            origin=self.options.origin)

        if self.options.dry_run:

            if self.txn is not None:

                self.txn.abort()

        else:

            self.txn.commit()

def warn_about_deleting_current_messages(cur, from_text, where_text, logger):

    # Deleting currently used translations is a bit harmful. Log

    # them so that we have a clue which messages might have to be

    # translated again. Note that this script tries to find

    # another translation that becomes current -- but only in one

    # situation: If we delete a shared translation which is current

    # in Ubuntu, a shared translation which is current in upstream

    # becomes the current Ubuntu translation. In other cases (deleting

    # a diverged translation, deleting a shared translation which

    # is current upstream) we do not attempt to find another current

    # message.

    if logger is not None and logger.getEffectiveLevel() <= logging.WARN:

        query = """

            SELECT

                TranslationMessage.id, TranslationMessage.is_current_upstream,

                TranslationMessage.is_current_ubuntu

            FROM %s

            WHERE %s AND (

                TranslationMessage.is_current_upstream OR

                TranslationMessage.is_current_ubuntu)

            """ % (from_text, where_text)

        cur.execute(query)

        rows = cur.fetchall()

        if cur.rowcount > 0:

            logger.warn(

                'Deleting messages currently in use:')

            for (id, is_current_upstream, is_current_ubuntu) in rows:

                current = []

                if is_current_upstream:

                    current.append('upstream')

                if is_current_ubuntu:

                    current.append('Ubuntu')

                logger.warn(

                    'Message %i is a current translation in %s'

                    % (id, ' and '.join(current)))

def remove_translations(logger=None, submitter=None, reviewer=None,

                        reject_license=False, ids=None, potemplate=None,

                        language_code=None, not_language=False,

                        is_current_ubuntu=None, is_current_upstream=None,

                        msgid_singular=None, origin=None):

    """Remove specified translation messages.

    :param logger: Optional logger to write output to.

    :param submitter: Delete only messages submitted by this person.

    :param reviewer: Delete only messages reviewed by this person.

    :param reject_license: Delete only messages submitted by persons who

        have rejected the licensing agreement.

    :param ids: Delete only messages with these `TranslationMessage` ids.

    :param potemplate: Delete only messages in this template.

    :param language_code: Language code.  Depending on `not_language`,

        either delete messages in this language or spare messages in this

        language that would otherwise be deleted.

    :param not_language: Whether to spare (True) or delete (False)

        messages in this language.

    :param is_current_ubuntu: Delete only messages with this is_current_ubuntu

        value.

    :param is_current_upstream: Delete only messages with this

        is_current_upstream value.

    :param msgid_singular: Delete only messages with this singular msgid.

    :param origin: Delete only messages with this `TranslationOrigin` code.

    :return: Number of messages deleted.

    """

    joins = set()

    conditions = set()

    if submitter is not None:

        conditions.add(

            'TranslationMessage.submitter = %s' % sqlvalues(submitter))

    if reviewer is not None:

        conditions.add(

            'TranslationMessage.reviewer = %s' % sqlvalues(reviewer))

    if reject_license:

        joins.add('TranslationRelicensingAgreement')

        conditions.add(

            'TranslationMessage.submitter = '

            'TranslationRelicensingAgreement.person')

        conditions.add(

            'NOT TranslationRelicensingAgreement.allow_relicensing')

    if ids is not None:

        conditions.add('TranslationMessage.id IN %s' % sqlvalues(ids))

    if potemplate is not None:

        joins.add('TranslationTemplateItem')

        conditions.add(

            'TranslationTemplateItem.potmsgset '

            ' = TranslationMessage.potmsgset')

        conditions.add(

            'TranslationTemplateItem.potemplate = %s' % sqlvalues(potemplate))

    if language_code is not None:

        joins.add('Language')

        conditions.add('Language.id = TranslationMessage.language')

        language_match = compose_language_match(language_code)

        if not_language:

            conditions.add('NOT (%s)' % language_match)

        else:

            conditions.add(language_match)

    add_bool_match(

        conditions, 'TranslationMessage.is_current_ubuntu', is_current_ubuntu)

    add_bool_match(

        conditions, 'TranslationMessage.is_current_upstream',

        is_current_upstream)

    if msgid_singular is not None:

        joins.add('POTMsgSet')

        conditions.add('POTMsgSet.id = TranslationMessage.potmsgset')

        joins.add('POMsgID')

        conditions.add('POMsgID.id = POTMsgSet.msgid_singular')

        conditions.add('POMsgID.msgid = %s' % sqlvalues(msgid_singular))

    if origin is not None:

        conditions.add('TranslationMessage.origin = %s' % sqlvalues(origin))

    assert len(conditions) > 0, "That would delete ALL translations, maniac!"

    cur = cursor()

    drop_tables(cur, 'temp_doomed_message')

    joins.add('TranslationMessage')

    from_text = ', '.join(joins)

    where_text = ' AND\n    '.join(conditions)

    warn_about_deleting_current_messages(cur, from_text, where_text, logger)

    # Keep track of messages we're going to delete.

    # Don't bother indexing this.  We'd more likely end up optimizing

    # away the operator's "oh-shit-ctrl-c" time than helping anyone.

    query = """

        CREATE TEMP TABLE temp_doomed_message AS

        SELECT TranslationMessage.id, NULL::integer AS imported_message

        FROM %s

        WHERE %s

        """ % (from_text, where_text)

    cur.execute(query)

    # Note which shared messages are masked by the messages we're

    # going to delete.  We'll be making those the current ones.

    query = """

         UPDATE temp_doomed_message

        SET imported_message = Imported.id

        FROM TranslationMessage Doomed, TranslationMessage Imported

        WHERE

            Doomed.id = temp_doomed_message.id AND

            -- Is alternative for the message we're about to delete.

            Imported.potmsgset = Doomed.potmsgset AND

            Imported.language = Doomed.language AND

            Imported.potemplate IS NULL AND

            Doomed.potemplate IS NULL AND

            -- Is used upstream.

            Imported.is_current_upstream IS TRUE AND

            -- Was masked by the message we're about to delete.

            Doomed.is_current_ubuntu IS TRUE AND

            Imported.id <> Doomed.id

            """

    cur.execute(query)

    if logger is not None and logger.getEffectiveLevel() <= logging.DEBUG:

        # Dump sample of doomed messages for debugging purposes.

        cur.execute("""

            SELECT *

            FROM temp_doomed_message

            ORDER BY id

            LIMIT 20

            """)

        rows = cur.fetchall()

        if cur.rowcount > 0:

            logger.debug("Sample of messages to be deleted follows.")

            logger.debug("%10s %10s" % ("[message]", "[unmasks]"))

            for (doomed, unmasked) in rows:

                if unmasked is None:

                    unmasked = '--'

                logger.debug("%10s %10s" % (doomed, unmasked))

    cur.execute("""

        DELETE FROM TranslationMessage

        USING temp_doomed_message

        WHERE TranslationMessage.id = temp_doomed_message.id

        """)

    rows_deleted = cur.rowcount

    if logger is not None:

        if rows_deleted > 0:

            logger.info("Deleting %d message(s)." % rows_deleted)

        else:

            logger.warn("No rows match; not deleting anything.")

    cur.execute("""

        UPDATE TranslationMessage

        SET is_current_ubuntu = TRUE

        FROM temp_doomed_message

        WHERE TranslationMessage.id = temp_doomed_message.imported_message

        """)

    if cur.rowcount > 0 and logger is not None:

        logger.debug("Unmasking %d imported message(s)." % cur.rowcount)

    drop_tables(cur, 'temp_doomed_message')

    return rows_deleted