-
Notifications
You must be signed in to change notification settings - Fork 4
/
dup_core.py
166 lines (129 loc) · 5.77 KB
/
dup_core.py
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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
"""Detect and handle duplicates in the library."""
import logging
from typing import Optional
import sqlalchemy
from sqlalchemy.orm.session import Session
import moe
from moe import config
from moe.library import Album, Extra, LibItem, Track
from moe.query import query
__all__ = ["DuplicateError", "get_duplicates", "resolve_duplicates"]
log = logging.getLogger("moe.dup")
class DuplicateError(Exception):
"""Duplicate items could not be resolved."""
class Hooks:
"""Duplicate plugin hook specifications."""
@staticmethod
@moe.hookspec
def resolve_dup_items(session: Session, item_a: LibItem, item_b: LibItem):
"""Resolve two duplicate items.
A resolution should come in one of two forms:
1. The field(s) triggering the duplicate detection is altered on one or both
of the items so that they are no longer duplicates.
2. One of the items is removed from the library entirely using
:meth:`~moe.remove.remove_item`.
Important:
Duplicates are determined by each item's
:meth:`~moe.library.lib_item.LibItem.is_unique` method. This behavior can be
extended by the various ``is_unique_*`` hooks.
Note:
As a plugin that may have introduced new ``unique`` constraints via one of
the ``is_unique_*`` hooks, you only need to be concerned about resolving
that particular constraint. You may also just rely on the default duplicate
resolution method of Moe, which depending on the UI in use, may be
sufficient for your plugin's needs. For example, the CLI will offer a
'duplicate resolution' prompt to the user anytime a duplicate is detected.
This hook allows you to implement your own duplicate resolution methods in
addition to what's offered by default.
Args:
session: Library db session.
item_a: First item.
item_b: Second item.
See Also:
* The :meth:`~moe.library.album.Hooks.is_unique_album` hook.
* The :meth:`~moe.library.extra.Hooks.is_unique_extra` hook.
* The :meth:`~moe.library.track.Hooks.is_unique_track` hook.
"""
@moe.hookimpl(hookwrapper=True)
def edit_changed_items(session: Session, items: list[LibItem]):
"""Check for and resolve duplicates when items are edited."""
yield # run all `edit_changed_items` hook implementations
albums = [item for item in items if isinstance(item, Album)] # resolve albums first
tracks = [item for item in items if isinstance(item, Track)]
extras = [item for item in items if isinstance(item, Extra)]
resolve_duplicates(session, albums) # type: ignore
resolve_duplicates(session, tracks) # type: ignore
resolve_duplicates(session, extras) # type: ignore
@moe.hookimpl(hookwrapper=True)
def edit_new_items(session: Session, items: list[LibItem]):
"""Check for and resolve duplicates when items are added to the library."""
yield # run all `edit_new_items` hook implementations
albums = [item for item in items if isinstance(item, Album)] # resolve albums first
tracks = [item for item in items if isinstance(item, Track)]
extras = [item for item in items if isinstance(item, Extra)]
resolve_duplicates(session, albums) # type: ignore
resolve_duplicates(session, tracks) # type: ignore
resolve_duplicates(session, extras) # type: ignore
def resolve_duplicates(session: Session, items: list[LibItem]):
"""Search for and resolve any duplicates of items in ``items``."""
log.debug(f"Checking for duplicate items. [{items=!r}]")
resolved_items = []
for item in items:
if _is_removed(item):
continue
dup_items = get_duplicates(session, item, items)
dup_items += get_duplicates(session, item)
for dup_item in dup_items:
if dup_item in resolved_items or _is_removed(item):
continue
log.debug(
f"Resolving duplicate items. [item_a={item!r}, item_b={dup_item!r}]"
)
config.CONFIG.pm.hook.resolve_dup_items(
session=session, item_a=item, item_b=dup_item
)
if (
not item.is_unique(dup_item)
and not _is_removed(item)
and not _is_removed(dup_item)
):
raise DuplicateError(
"Duplicate items could not be resolved. "
f"[item_a={item!r}, item_b={dup_item!r}]"
)
if dup_items:
resolved_items.append(item)
if resolved_items:
log.debug(f"Resolved duplicate items. [{resolved_items=!r}]")
else:
log.debug("No duplicate items found.")
def _is_removed(item):
"""Check whether or not an item is removed from the library."""
insp = sqlalchemy.inspect(item)
if insp.deleted or insp.transient:
return True
return False
def get_duplicates(
session: Session, item: LibItem, others: Optional[list[LibItem]] = None
) -> list[LibItem]:
"""Returns items considered duplicates of ``item``.
Args:
session: Library db session.
item: Library item to get duplicates of.
others: Items to compare against. If not given, will query the database
and compare against all items in the library.
Returns:
Any items considered a duplicate as defined by
:meth:`~moe.library.lib_item.LibItem.is_unique`.
"""
dup_items = []
if not others:
others = query(session, "*", type(item).__name__.lower())
for other in others:
if (
item is not other
and type(item) == type(other)
and not item.is_unique(other)
):
dup_items.append(other)
return dup_items