def _task(**kw):
    # Remove any dupes. `UNIQUE` constraint introduced in migration 504.
    dupes = (ADT.objects.values_list('addon', 'device_type')
                        .annotate(c=Count('id')).filter(c__gt=1))
    for addon, device_type, total in dupes:
        devices = ADT.objects.filter(addon_id=addon, device_type=device_type)
        for d in devices[:total - 1]:
            d.delete()

    # Remove stale device types.
    devices = ADT.objects.all()
    for chunk in chunked(devices, 50):
        for device in chunk:
            try:
                device.addon
            except ObjectDoesNotExist:
                device.delete()

    # `DEVICE_MOBILE` -> `DEVICE_MOBILE` and `DEVICE_GAIA`.
    devices = ADT.objects.filter(device_type=mkt.DEVICE_MOBILE.id)

    for chunk in chunked(devices, 50):
        for device in chunk:
            if mkt.DEVICE_GAIA in device.addon.device_types:
                continue
            device.id = None
            device.device_type = mkt.DEVICE_GAIA.id
            device.save()
            device.addon.save()

    def handle(self, *args, **options):
        app_ids = (FeedApp.objects.filter(color__isnull=True)
                                  .values_list('id', flat=True))
        coll_ids = (FeedCollection.objects.filter(color__isnull=True)
                                          .values_list('id', flat=True))

        for chunk in chunked(app_ids, 100):
            _migrate_collection_colors.delay(chunk, 'app')

        for chunk in chunked(coll_ids, 100):
            _migrate_collection_colors.delay(chunk, 'collection')

    def handle(self, *args, **options):
        ids = (CommunicationNote.objects
                                .filter(note_type=comm.REVIEWER_COMMENT)
                                .values_list('id', flat=True))

        for log_chunk in chunked(ids, 100):
            _fix_developer_version_notes.delay(ids)

def convert(directory, delete=False):
    print 'Converting icons in %s' % directory

    pks = []
    k = 0
    for path, names, filenames in walk_storage(directory):
        for filename in filenames:
            old = os.path.join(path, filename)
            pre, ext = os.path.splitext(old)
            if (pre[-3:] in size_suffixes or ext not in extensions):
                continue

            if not storage.size(old):
                print 'Icon %s is empty, ignoring.' % old
                continue

            for size, size_suffix in zip(sizes, size_suffixes):
                new = '%s%s%s' % (pre, size_suffix, '.png')
                if os.path.exists(new):
                    continue
                resize_image(old, new, (size, size), remove_src=False)

            if ext != '.png':
                pks.append(os.path.basename(pre))

            if delete:
                storage.delete(old)

            k += 1
            if not k % 1000:
                print "... converted %s" % k

    for chunk in chunked(pks, 100):
        Webapp.objects.filter(pk__in=chunk).update(icon_type='image/png')

def dump_user_installs_cron():
    """
    Sets up tasks to do user install dumps.
    """
    chunk_size = 100
    # Get valid users to dump.
    user_ids = set(Installed.objects.filter(user__enable_recommendations=True)
                   .values_list('user', flat=True))

    # Clean up the path where we'll store the individual json files from each
    # user installs dump (which are in users/ in DUMPED_USERS_PATH).
    path_to_cleanup = os.path.join(settings.DUMPED_USERS_PATH, 'users')
    task_log.info('Cleaning up path {0}'.format(path_to_cleanup))
    try:
        for dirpath, dirnames, filenames in walk_storage(
                path_to_cleanup, storage=private_storage):
            for filename in filenames:
                private_storage.delete(os.path.join(dirpath, filename))
    except OSError:
        # Ignore if the directory does not exist.
        pass

    grouping = []
    for chunk in chunked(user_ids, chunk_size):
        grouping.append(dump_user_installs.subtask(args=[chunk]))

    post = zip_users.subtask(immutable=True)
    ts = chord(grouping, post)
    ts.apply_async()

def run():
    for chunk in chunked(Webapp.objects.all(), 50):
        for app in chunk:
            try:
                generate_image_assets.delay(app)
            except Exception:
                pass

    def handle(self, *args, **options):
        task = tasks.get(options.get('task'))
        if not task:
            raise CommandError('Unknown task provided. Options are: %s'
                               % ', '.join(tasks.keys()))
        qs = Webapp.objects.all()
        if 'qs' in task:
            qs = qs.filter(*task['qs'])
        pks = qs.values_list('pk', flat=True).order_by('-last_updated')
        if 'pre' in task:
            # This is run in process to ensure its run before the tasks.
            pks = task['pre'](pks)
        if pks:
            kw = task.get('kwargs', {})
            # All the remaining tasks go in one group.
            grouping = []
            for chunk in chunked(pks, 100):
                grouping.append(
                    task['method'].subtask(args=[chunk], kwargs=kw))

            # Add the post task on to the end.
            post = None
            if 'post' in task:
                post = task['post'].subtask(args=[], kwargs=kw, immutable=True)
                ts = chord(grouping, post)
            else:
                ts = group(grouping)
            ts.apply_async()

    def handle(self, *args, **kwargs):

        ids = []

        ids.extend(
            FeedCollectionMembership.objects.values_list('group', flat=True))
        ids.extend(
            FeedCollection.objects.values_list('name', flat=True))
        ids.extend(
            FeedCollection.objects.values_list('description', flat=True))
        ids.extend(
            FeedShelfMembership.objects.values_list('group', flat=True))
        ids.extend(
            FeedShelf.objects.values_list('description', flat=True))
        ids.extend(
            FeedShelf.objects.values_list('name', flat=True))
        ids.extend(
            FeedApp.objects.values_list('description', flat=True))
        ids.extend(
            FeedApp.objects.values_list('pullquote_text', flat=True))
        ids.extend(
            Version.objects.values_list('releasenotes', flat=True))
        ids.extend(
            Webapp.objects.values_list('description', flat=True))
        ids.extend(
            Webapp.objects.values_list('privacy_policy', flat=True))
        ids.extend(
            Geodata.objects.values_list('banner_message', flat=True))

        # Filter out any None's.
        ids = filter(None, ids)

        for chunk in chunked(ids, 100):
            update_translations.delay(chunk)

def run():
    """Delete duplicate image assets."""
    for chunk in chunked(Webapp.objects.all(), 50):
        for app in chunk:
            for slug in SIZE_SLUGS:
                assets = ImageAsset.objects.filter(addon=app, slug=slug)
                for asset in assets[1:]:
                    asset.delete()

    def handle(self, *args, **kwargs):
        qs = Webapp.objects.filter(_geodata__restricted=False)

        if kwargs['app']:
            qs = qs.filter(pk=kwargs['app'])

        apps = qs.values_list('id', flat=True)
        for chunk in chunked(apps, 100):
            fix_excluded_regions.delay(chunk)

    def handle(self, *args, **options):
        applog_ids = AppLog.objects.values_list('activity_log', flat=True)

        ids = (ActivityLog.objects.filter(
            pk__in=list(applog_ids), action__in=mkt.LOG_REVIEW_QUEUE)
            .order_by('created').values_list('id', flat=True))

        for log_chunk in chunked(ids, 100):
            _migrate_activity_log.delay(ids)

def email_devs(request):
    form = DevMailerForm(request.POST or None)
    preview = EmailPreviewTopic(topic='email-devs')
    if preview.filter().count():
        preview_csv = reverse('zadmin.email_preview_csv',
                              args=[preview.topic])
    else:
        preview_csv = None
    if request.method == 'POST' and form.is_valid():
        data = form.cleaned_data
        qs = (AddonUser.objects.filter(role__in=(mkt.AUTHOR_ROLE_DEV,
                                                 mkt.AUTHOR_ROLE_OWNER))
                               .exclude(user__email=None))

        if data['recipients'] in ('payments', 'desktop_apps'):
            qs = qs.exclude(addon__status=mkt.STATUS_DELETED)
        else:
            qs = qs.filter(addon__status__in=mkt.LISTED_STATUSES)

        if data['recipients'] in ('payments', 'payments_region_enabled',
                                  'payments_region_disabled'):
            qs = qs.exclude(addon__premium_type__in=(mkt.ADDON_FREE,
                                                     mkt.ADDON_OTHER_INAPP))
            if data['recipients'] == 'payments_region_enabled':
                qs = qs.filter(addon__enable_new_regions=True)
            elif data['recipients'] == 'payments_region_disabled':
                qs = qs.filter(addon__enable_new_regions=False)
        elif data['recipients'] in ('apps', 'free_apps_region_enabled',
                                    'free_apps_region_disabled'):
            if data['recipients'] == 'free_apps_region_enabled':
                qs = qs.filter(addon__enable_new_regions=True)
            elif data['recipients'] == 'free_apps_region_disabled':
                qs = qs.filter(addon__enable_new_regions=False)
        elif data['recipients'] == 'desktop_apps':
            qs = (qs.filter(
                addon__addondevicetype__device_type=mkt.DEVICE_DESKTOP.id))
        else:
            raise NotImplementedError('If you want to support emailing other '
                                      'types of developers, do it here!')
        if data['preview_only']:
            # Clear out the last batch of previewed emails.
            preview.filter().delete()
        total = 0
        for emails in chunked(set(qs.values_list('user__email', flat=True)),
                              100):
            total += len(emails)
            tasks.admin_email.delay(emails, data['subject'], data['message'],
                                    preview_only=data['preview_only'],
                                    preview_topic=preview.topic)
        msg = 'Emails queued for delivery: %s' % total
        if data['preview_only']:
            msg = '%s (for preview only, emails not sent!)' % msg
        messages.success(request, msg)
        return redirect('zadmin.email_devs')
    return render(request, 'zadmin/email-devs.html',
                  dict(form=form, preview_csv=preview_csv))

    def handle(self, *args, **kw):
        from mkt.webapps.models import Webapp

        # Get apps.
        apps = Webapp.objects.filter(iarc_info__isnull=False)
        ids = kw.get("apps")
        if ids:
            apps = apps.filter(id__in=(int(id.strip()) for id in ids.split(",")))

        for chunk in chunked(apps.values_list("id", flat=True), 100):
            refresh_iarc_ratings.delay(chunk)

def manifest_revalidation(request):
    if request.method == 'POST':
        # Collect the apps to revalidate.
        qs = Q(is_packaged=False, status=mkt.STATUS_PUBLIC,
               disabled_by_user=False)
        webapp_pks = Webapp.objects.filter(qs).values_list('pk', flat=True)

        for pks in chunked(webapp_pks, 100):
            update_manifests.delay(list(pks), check_hash=False)

        messages.success(request, 'Manifest revalidation queued')

    return render(request, 'zadmin/manifest.html')

    def handle(self, *args, **options):
        webapps = (Webapp.objects.filter(app_payment_accounts__isnull=False)
                                 .no_transforms()
                                 .select_related('app_payment_accounts'))
        for chunk in chunked(webapps, 50):
            for app in chunk:
                generic_product = get_generic_product(app)
                if not generic_product:
                    continue
                print 'Found public_id', generic_product['public_id']

                if not options['dry_run']:
                    print 'Saving app', app
                    app.solitude_public_id = generic_product['public_id']
                    app.save()

    def handle(self, *args, **kw):
        from mkt.webapps.models import Webapp

        # Get apps.
        if waffle.switch_is_active('iarc-upgrade-v2'):
            apps = Webapp.objects.filter(iarc_cert__isnull=False)
        else:
            apps = Webapp.objects.filter(iarc_info__isnull=False)
        ids = kw.get('apps')
        if ids:
            apps = apps.filter(
                id__in=(int(id.strip()) for id in ids.split(',')))

        for chunk in chunked(apps.values_list('id', flat=True), 100):
            refresh_iarc_ratings.delay(chunk)

def mkt_gc(**kw):
    """Site-wide garbage collections."""
    log.info('Collecting data to delete')
    logs = (ActivityLog.objects.filter(created__lt=days_ago(90))
            .exclude(action__in=mkt.LOG_KEEP).values_list('id', flat=True))

    for chunk in chunked(logs, 100):
        chunk.sort()
        log.info('Deleting log entries: %s' % str(chunk))
        delete_logs.delay(chunk)

    # Clear oauth nonce rows. These expire after 10 minutes but we're just
    # clearing those that are more than 1 day old.
    Nonce.objects.filter(created__lt=days_ago(1)).delete()

    # Delete the dump apps over 30 days.
    _remove_stale_files(os.path.join(settings.DUMPED_APPS_PATH, 'tarballs'),
                        settings.DUMPED_APPS_DAYS_DELETE,
                        'Deleting old tarball: {0}',
                        storage=public_storage)

    # Delete the dumped user installs over 30 days. Those are using private
    # storage.
    _remove_stale_files(os.path.join(settings.DUMPED_USERS_PATH, 'tarballs'),
                        settings.DUMPED_USERS_DAYS_DELETE,
                        'Deleting old tarball: {0}',
                        storage=private_storage)

    # Delete old files in select directories under TMP_PATH.
    _remove_stale_files(os.path.join(settings.TMP_PATH, 'preview'),
                        settings.TMP_PATH_DAYS_DELETE,
                        'Deleting TMP_PATH file: {0}',
                        storage=private_storage)
    _remove_stale_files(os.path.join(settings.TMP_PATH, 'icon'),
                        settings.TMP_PATH_DAYS_DELETE,
                        'Deleting TMP_PATH file: {0}',
                        storage=private_storage)

    # Delete stale FileUploads.
    for fu in FileUpload.objects.filter(created__lte=days_ago(90)):
        log.debug(u'[FileUpload:{uuid}] Removing file: {path}'
                  .format(uuid=fu.uuid, path=fu.path))
        if fu.path:
            try:
                private_storage.delete(fu.path)
            except OSError:
                pass
        fu.delete()

def hide_disabled_files():
    # If an add-on or a file is disabled, it should be moved to
    # GUARDED_ADDONS_PATH so it's not publicly visible.
    #
    # We ignore deleted versions since we hide those files when deleted and
    # also due to bug 980916.
    ids = (File.objects
           .filter(version__deleted=False)
           .filter(Q(status=mkt.STATUS_DISABLED) |
                   Q(version__addon__status=mkt.STATUS_DISABLED) |
                   Q(version__addon__disabled_by_user=True))
           .values_list('id', flat=True))
    for chunk in chunked(ids, 300):
        qs = File.objects.filter(id__in=chunk)
        qs = qs.select_related('version')
        for f in qs:
            f.hide_disabled_file()

def run():
    """
    Retroactively award theme reviewer points for all the theme
    reviewers done since the Great Theme Migration to amo up to
    when we started recording points.
    """
    start_date = datetime.date(2013, 8, 27)

    # Get theme reviews that are approves and rejects from before we started
    # awarding.
    approve = '"action": %s' % rvw.ACTION_APPROVE
    reject = '"action": %s' % rvw.ACTION_REJECT
    al = ActivityLog.objects.filter(
        Q(_details__contains=approve) | Q(_details__contains=reject),
        action=mkt.LOG.THEME_REVIEW.id, created__lte=start_date)

    for chunk in chunked(al, 50):
        # Review and thou shall receive.
        _batch_award_points.delay(chunk)

def dump_user_installs_cron():
    """
    Sets up tasks to do user install dumps.
    """
    chunk_size = 100
    # Get valid users to dump.
    user_ids = set(Installed.objects.filter(user__enable_recommendations=True)
                   .values_list('user', flat=True))

    # Remove old dump data before running.
    user_dir = os.path.join(settings.DUMPED_USERS_PATH, 'users')
    if os.path.exists(user_dir):
        shutil.rmtree(user_dir)

    grouping = []
    for chunk in chunked(user_ids, chunk_size):
        grouping.append(dump_user_installs.subtask(args=[chunk]))

    post = zip_users.subtask(immutable=True)
    ts = chord(grouping, post)
    ts.apply_async()