<commit_before>import pendulum

from share.disambiguation.criteria import MatchByOneToMany
from share.exceptions import MergeRequired
from share.models import AbstractCreativeWork
from share.management.commands import BaseShareCommand
from share.models import IngestJob
from share.tasks.jobs import IngestJobConsumer
from share.util import ensure_iterable


class Command(BaseShareCommand):
    MAX_RETRIES = 10

    def add_arguments(self, parser):
        parser.add_argument('--dry', action='store_true', help='Should the script actually make changes? (In a transaction)')
        parser.add_argument('--commit', action='store_true', help='Should the script actually commit?')
        parser.add_argument('--noninteractive', action='store_true', help='Should the script merge objects without asking?')
        parser.add_argument('--from', type=lambda d: pendulum.from_format(d, '%Y-%m-%d'), help='Only consider jobs on or after this date')
        parser.add_argument('--until', type=lambda d: pendulum.from_format(d, '%Y-%m-%d'), help='Only consider jobs on or before this date')

    def handle(self, *args, **options):
        dry_run = options.get('dry')
        interactive = not options.get('noninteractive')

        qs = IngestJob.objects.filter(
            error_type='MergeRequired',
            status=IngestJob.STATUS.failed,
            source_config__source__canonical=True
        ).order_by('date_created')
        if options.get('from'):
            qs = qs.filter(date_modified__gte=options.get('from'))
        if options.get('until'):
            qs = qs.filter(date_modified__lte=options.get('until'))

        for job in qs.select_related('suid', 'source_config'):
            with self.rollback_unless_commit(options.get('commit')):
                self.stdout.write('\n\nTrying job {!r}'.format(job), style_func=self.style.HTTP_INFO)
                self._try_job(job, dry_run, interactive)

    def _try_job(self, job, dry_run, interactive):
        for i in range(self.MAX_RETRIES):
            self.stdout.write('Attempt {} of {}:'.format(i + 1, self.MAX_RETRIES))
            try:
                IngestJobConsumer().consume(job_id=job.id, exhaust=False)
            except MergeRequired as e:
                (_, *dupe_sets) = e.args

                if not self._fix_merge_error(dupe_sets, dry_run, interactive):
                    return

                if dry_run:
                    return
                continue
            except Exception as e:
                self.stdout.write('Failed in a way we cant fix:', style_func=self.style.ERROR)
                self.stdout.write(str(e))
                return
            self.stdout.write('Success!', style_func=self.style.SUCCESS)
            return
        self.stdout.write('Failed to fix after {} tries'.format(self.MAX_RETRIES), style_func=self.style.ERROR)

    def _fix_merge_error(self, dupe_sets, dry_run, interactive):
        for dupes in dupe_sets:
            if len(dupes) < 2:
                continue
            model = dupes[0]._meta.concrete_model
            if any(d._meta.concrete_model is not model for d in dupes):
                raise ValueError('Cannot merge across tables: {}'.format(dupes))

            if not self._merge_together(model, dupes, dry_run, interactive):
                return False
        return True

    def _merge_together(self, model, dupes, dry_run, interactive):
        criteria = [
            c for c in ensure_iterable(model.matching_criteria)
            if isinstance(c, MatchByOneToMany)
        ]

        if len(criteria) != 1:
            self.stdout.write('Cannot automatically fix errors on {}, need a MatchByOneToMany criterion'.format(model), style_func=self.style.ERROR)
            return False
<commit_msg>Handle non-list dupes<commit_after>import pendulum

from share.disambiguation.criteria import MatchByOneToMany
from share.exceptions import MergeRequired
from share.models import AbstractCreativeWork
from share.management.commands import BaseShareCommand
from share.models import IngestJob
from share.tasks.jobs import IngestJobConsumer
from share.util import ensure_iterable


class Command(BaseShareCommand):
    MAX_RETRIES = 10

    def add_arguments(self, parser):
        parser.add_argument('--dry', action='store_true', help='Should the script actually make changes? (In a transaction)')
        parser.add_argument('--commit', action='store_true', help='Should the script actually commit?')
        parser.add_argument('--noninteractive', action='store_true', help='Should the script merge objects without asking?')
        parser.add_argument('--from', type=lambda d: pendulum.from_format(d, '%Y-%m-%d'), help='Only consider jobs on or after this date')
        parser.add_argument('--until', type=lambda d: pendulum.from_format(d, '%Y-%m-%d'), help='Only consider jobs on or before this date')

    def handle(self, *args, **options):
        dry_run = options.get('dry')
        interactive = not options.get('noninteractive')

        qs = IngestJob.objects.filter(
            error_type='MergeRequired',
            status=IngestJob.STATUS.failed,
            source_config__source__canonical=True
        ).order_by('date_created')
        if options.get('from'):
            qs = qs.filter(date_modified__gte=options.get('from'))
        if options.get('until'):
            qs = qs.filter(date_modified__lte=options.get('until'))

        for job in qs.select_related('suid', 'source_config'):
            with self.rollback_unless_commit(options.get('commit')):
                self.stdout.write('\n\nTrying job {!r}'.format(job), style_func=self.style.HTTP_INFO)
                self._try_job(job, dry_run, interactive)

    def _try_job(self, job, dry_run, interactive):
        for i in range(self.MAX_RETRIES):
            self.stdout.write('Attempt {} of {}:'.format(i + 1, self.MAX_RETRIES))
            try:
                IngestJobConsumer().consume(job_id=job.id, exhaust=False)
            except MergeRequired as e:
                (_, *dupe_sets) = e.args

                if not self._fix_merge_error(dupe_sets, dry_run, interactive):
                    return

                if dry_run:
                    return
                continue
            except Exception as e:
                self.stdout.write('Failed in a way we cant fix:', style_func=self.style.ERROR)
                self.stdout.write(str(e))
                return
            self.stdout.write('Success!', style_func=self.style.SUCCESS)
            return
        self.stdout.write('Failed to fix after {} tries'.format(self.MAX_RETRIES), style_func=self.style.ERROR)

    def _fix_merge_error(self, dupe_sets, dry_run, interactive):
        for dupes in dupe_sets:
            if len(dupes) < 2:
                continue
            model = list(dupes)[0]._meta.concrete_model
            if any(d._meta.concrete_model is not model for d in dupes):
                raise ValueError('Cannot merge across tables: {}'.format(dupes))

            if not self._merge_together(model, dupes, dry_run, interactive):
                return False
        return True

    def _merge_together(self, model, dupes, dry_run, interactive):
        criteria = [
            c for c in ensure_iterable(model.matching_criteria)
            if isinstance(c, MatchByOneToMany)
        ]

        if len(criteria) != 1:
            self.stdout.write('Cannot automatically fix errors on {}, need a MatchByOneToMany criterion'.format(model), style_func=self.style.ERROR)
            return False

{'fraction_non_alphanumeric': 0.10744568097509274, 'fraction_numerical': 0.001589825119236884, 'mean_word_length': 2.61888782358581, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1458471', 'n_tokens_mistral': 2184, 'n_tokens_neox': 2091, 'n_words': 508}