aboutsummaryrefslogtreecommitdiffstats
path: root/python/fatcat_tools/mergers/releases.py
blob: 1f995b00f342f6431c804877227a3926de274a7d (plain)
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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
import argparse
import os
import sys
from typing import Any, Dict, List, Optional

import fatcat_openapi_client
from fatcat_openapi_client.models import ReleaseEntity, WorkEntity

from fatcat_tools import authenticated_api
from fatcat_tools.importers import JsonLinePusher

from .common import EntityMerger


class ReleaseMerger(EntityMerger):
    """
    Hard merges a set of release entities, redirecting all entities to a single
    primary release. This is different from "grouping" multiple releases under
    a single work.

    A "primary" (which the other releases will redirect to) can be provided, or
    one will be chosen from the set of duplicate releases based on the
    completeness of metadata and other heuristic factors.

    Releases are some of the most complex entities to merge, because of
    the complexity of bibliographic metadata and the number of related entities
    which also need to be updated.

    File, Fileset, and Webcapture entities which currently point to a release
    which gets redirected will be updated to point at the "primary" release.

    Any Work entities which will end up with no releases pointing at them after
    the merging will get redirected to the work corresponding to the "primary"
    release.

    NOTE: the "primary" release will currently (as implemented) *not* get
    updated with metadata from all the redirected releases
    """

    def __init__(self, api: fatcat_openapi_client.ApiClient, **kwargs) -> None:

        eg_desc = (
            kwargs.pop("editgroup_description", None) or "Automated merge of release entities"
        )
        eg_extra = kwargs.get("editgroup_extra", dict())
        eg_extra["agent"] = eg_extra.get("agent", "fatcat_tools.ReleaseMerger")
        self.dry_run_mode: bool = eg_extra.get("dry_run_mode", False)
        super().__init__(api, editgroup_description=eg_desc, editgroup_extra=eg_extra, **kwargs)
        self.entity_type_name = "release"

    def choose_primary_release(
        self, entities: List[ReleaseEntity], existing_redirects: Dict[str, List[str]]
    ) -> str:
        assert entities and len(entities) >= 2

        # want to sort in descending order, so reverse=True
        entities = sorted(
            entities,
            key=lambda a: (
                # number of entities already redirected to this one
                len(existing_redirects[a.ident]),
                # number of file/fileset/webcapture entities (would need to update)
                int(len(a.files or []) + len(a.filesets or []) + len(a.webcaptures or [])),
                # has a strong identifier?
                bool(a.ext_id.doi or a.ext_id.pmid or a.ext_id.arxiv_id),
                # has any identifier?
                bool(a.ext_id),
                # has basic metadata?
                bool(a.release_type),
                bool(a.release_status),
                bool(a.release_year),
                bool(a.container_id),
                # has refs, abstracts, extra stuff?
                bool(a.refs),
                bool(a.abstracts),
            ),
            reverse=True,
        )
        return entities[0].ident

    def try_merge(
        self,
        dupe_ids: List[str],
        primary_id: Optional[str] = None,
        evidence: Optional[Dict[str, Any]] = None,
    ) -> int:

        # TODO: this code is pretty old and has only been partially refactored.
        # Needs more testing and review.
        raise NotImplementedError

        updated_entities = 0
        releases = dict()
        existing_redirects: Dict[str, List[str]] = dict()
        eg_id = self.get_editgroup_id()

        all_ids = dupe_ids.copy()
        if primary_id:
            all_ids.append(primary_id)
        for ident in all_ids:
            releases[ident] = self.api.get_release(ident, expand="files,filesets,webcaptures")
            existing_redirects[ident] = self.api.get_release_redirects(ident)

        if not primary_id:
            primary_id = self.choose_primary_release(
                list(releases.values()), existing_redirects
            )
            dupe_ids = [d for d in dupe_ids if d != primary_id]

        assert primary_id not in dupe_ids

        primary_work_id = releases[primary_id].work_id
        updated_work_ids = []
        redirected_release_ids = []

        # execute all the release redirects
        for release in releases.values():
            if release.ident == primary_id:
                continue

            # file redirects
            for e in release.files:
                assert release.ident in e.release_ids
                e.release_ids.remove(release.ident)
                if primary_id not in e.release_ids:
                    e.release_ids.append(primary_id)
                if not self.dry_run_mode:
                    self.api.update_file(eg_id, e.ident, e)
                updated_entities += 1
                self.counts["updated-files"] += 1

            # fileset redirects
            for e in release.filesets:
                assert release.ident in e.release_ids
                e.release_ids.remove(release.ident)
                if primary_id not in e.release_ids:
                    e.release_ids.append(primary_id)
                if not self.dry_run_mode:
                    self.api.update_fileset(eg_id, e.ident, e)
                updated_entities += 1
                self.counts["updated-filesets"] += 1

            # webcapture redirects
            for e in release.webcaptures:
                assert release.ident in e.release_ids
                e.release_ids.remove(release.ident)
                if primary_id not in e.release_ids:
                    e.release_ids.append(primary_id)
                if not self.dry_run_mode:
                    self.api.update_webcapture(eg_id, e.ident, e)
                updated_entities += 1
                self.counts["updated-webcaptures"] += 1

            # the release redirect itself
            updated_work_ids.append(release.work_id)
            redirected_release_ids.append(release.ident)
            if not self.dry_run_mode:
                self.api.update_release(
                    eg_id,
                    release.ident,
                    ReleaseEntity(redirect=primary_id, edit_extra=evidence),
                )
            updated_entities += 1
            self.counts["updated-releases"] += 1

        # lastly, clean up any merged work entities
        redirected_release_ids = list(set(redirected_release_ids))
        updated_work_ids = list(set(updated_work_ids))
        assert primary_work_id not in updated_work_ids
        for work_id in updated_work_ids:
            work_releases = self.api.get_work_releases(work_id, hide="abstracts,refs")
            rids = set([r.ident for r in work_releases])
            if rids.issubset(redirected_release_ids):
                # all the releases for this work were updated/merged; we should
                # redirect to primary work_id
                # also check we don't have any in-flight edit conflicts
                assert work_id not in self._idents_inflight
                self._idents_inflight.append(work_id)
                if not self.dry_run_mode:
                    self.api.update_work(eg_id, work_id, WorkEntity(redirect=primary_work_id))
                updated_entities += 1
                self.counts["updated-works"] += 1

        return updated_entities


def run_merge_releases(args: argparse.Namespace) -> None:
    em = ReleaseMerger(args.api, edit_batch_size=args.batch_size, dry_run_mode=args.dry_run)
    JsonLinePusher(em, args.json_file).run()


def main() -> None:
    """
    Invoke like:

        python3 -m fatcat_tools.mergers.releases [options]
    """
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "--host-url", default="http://localhost:9411/v0", help="connect to this host/port"
    )
    parser.add_argument("--batch-size", help="size of batch to send", default=50, type=int)
    parser.add_argument(
        "--editgroup-description-override",
        help="editgroup description override",
        default=None,
        type=str,
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="don't actually commit merges, just count what would have been",
    )
    parser.set_defaults(
        auth_var="FATCAT_AUTH_API_TOKEN",
    )
    subparsers = parser.add_subparsers()

    sub_merge_releases = subparsers.add_parser("merge-releases")
    sub_merge_releases.set_defaults(func=run_merge_releases)
    sub_merge_releases.add_argument(
        "json_file",
        help="source of merge lines to process (or stdin)",
        default=sys.stdin,
        type=argparse.FileType("r"),
    )

    args = parser.parse_args()
    if not args.__dict__.get("func"):
        print("tell me what to do!")
        sys.exit(-1)

    # allow editgroup description override via env variable (but CLI arg takes
    # precedence)
    if not args.editgroup_description_override and os.environ.get(
        "FATCAT_EDITGROUP_DESCRIPTION"
    ):
        args.editgroup_description_override = os.environ.get("FATCAT_EDITGROUP_DESCRIPTION")

    args.api = authenticated_api(
        args.host_url,
        # token is an optional kwarg (can be empty string, None, etc)
        token=os.environ.get(args.auth_var),
    )
    args.func(args)


if __name__ == "__main__":
    main()