File: processors.py

package info (click to toggle)
python-rebulk 3.3.0-4
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 752 kB
  • sloc: python: 7,497; makefile: 3
file content (107 lines) | stat: -rw-r--r-- 3,388 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
Processor functions
"""
from logging import getLogger

from .utils import IdentitySet

from .rules import Rule, RemoveMatch

log = getLogger(__name__).log

DEFAULT = '__default__'

POST_PROCESS = -2048
PRE_PROCESS = 2048


def _default_conflict_solver(match, conflicting_match):
    """
    Default conflict solver for matches, shorter matches if they conflicts with longer ones

    :param conflicting_match:
    :type conflicting_match:
    :param match:
    :type match:
    :return:
    :rtype:
    """
    if len(conflicting_match.initiator) < len(match.initiator):
        return conflicting_match
    if len(match.initiator) < len(conflicting_match.initiator):
        return match
    return None


class ConflictSolver(Rule):
    """
    Remove conflicting matches.
    """
    priority = PRE_PROCESS

    consequence = RemoveMatch

    @property
    def default_conflict_solver(self):
        """
        Default conflict solver to use.
        """
        return _default_conflict_solver

    def when(self, matches, context):
        # pylint:disable=too-many-nested-blocks
        to_remove_matches = IdentitySet()

        public_matches = [match for match in matches if not match.private]
        public_matches.sort(key=len)

        for match in public_matches:
            conflicting_matches = matches.conflicting(match)

            if conflicting_matches:
                # keep the match only if it's the longest
                conflicting_matches = [conflicting_match for conflicting_match in conflicting_matches if
                                       not conflicting_match.private]
                conflicting_matches.sort(key=len)

                for conflicting_match in conflicting_matches:
                    conflict_solvers = [(self.default_conflict_solver, False)]

                    if match.conflict_solver:
                        conflict_solvers.append((match.conflict_solver, False))
                    if conflicting_match.conflict_solver:
                        conflict_solvers.append((conflicting_match.conflict_solver, True))

                    for conflict_solver, reverse in reversed(conflict_solvers):
                        if reverse:
                            to_remove = conflict_solver(conflicting_match, match)
                        else:
                            to_remove = conflict_solver(match, conflicting_match)
                        if to_remove == DEFAULT:
                            continue
                        if to_remove and to_remove not in to_remove_matches:
                            both_matches = [match, conflicting_match]
                            both_matches.remove(to_remove)
                            to_keep = both_matches[0]

                            if to_keep not in to_remove_matches:
                                log(self.log_level, "Conflicting match %s will be removed in favor of match %s",
                                    to_remove, to_keep)

                                to_remove_matches.add(to_remove)
                        break
        return to_remove_matches


class PrivateRemover(Rule):
    """
    Removes private matches rule.
    """
    priority = POST_PROCESS

    consequence = RemoveMatch

    def when(self, matches, context):
        return [match for match in matches if match.private]