summaryrefslogtreecommitdiffstats
path: root/python/fatcat_tools/importers/journal_metadata.py
blob: fc1dfcbdc667376d1ed93f9246a959206e1497c9 (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
from typing import Any, Dict, List, Optional

import fatcat_openapi_client
from fatcat_openapi_client import ApiClient, ContainerEntity

from fatcat_tools.normal import clean_str

from .common import EntityImporter


def or_none(s: Optional[str]) -> Optional[str]:
    if s is None:
        return None
    if len(s) == 0:
        return None
    return s


def truthy(s: Optional[str]) -> Optional[bool]:
    if s is None:
        return None
    s = s.lower()

    if s in ("true", "t", "yes", "y", "1"):
        return True
    elif s in ("false", "f", "no", "n", "0"):
        return False
    else:
        return None


class JournalMetadataImporter(EntityImporter):
    """
    Imports journal metadata ("containers") by ISSN, currently from a custom
    munged JSON format (see ../extra/journal_metadata/).

    See guide for details on the many 'extra' fields used here.
    """

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

        eg_desc = kwargs.get(
            "editgroup_description",
            "Automated import of container-level metadata, by ISSN. Metadata from Internet Archive munging.",
        )
        eg_extra = kwargs.get("editgroup_extra", dict())
        eg_extra["agent"] = eg_extra.get("agent", "fatcat_tools.JournalMetadataImporter")
        super().__init__(api, editgroup_description=eg_desc, editgroup_extra=eg_extra, **kwargs)

    def want(self, raw_record: Any) -> bool:
        if raw_record.get("issnl") and raw_record.get("name"):
            return True
        return False

    def parse_record(self, row: Dict[str, Any]) -> Optional[ContainerEntity]:
        """
        row is a python dict (parsed from JSON).

        returns a ContainerEntity (or None if invalid or couldn't parse)
        """

        if not row.get("name"):
            # Name is required (by schema)
            return None

        extra = dict()
        for key in (
            "issne",
            "issnp",
            "languages",
            "country",
            "urls",
            "abbrev",
            "coden",
            "aliases",
            "original_name",
            "first_year",
            "last_year",
            "platform",
            "default_license",
            "road",
            "mimetypes",
            "sherpa_romeo",
            "kbart",
        ):
            if row.get(key):
                extra[key] = row[key]
        # TODO: not including for now: norwegian, dois/crossref, ia

        extra_doaj = dict()
        if row.get("doaj"):
            if row["doaj"].get("as_of"):
                extra_doaj["as_of"] = row["doaj"]["as_of"]
            if row["doaj"].get("works"):
                extra_doaj["works"] = row["doaj"]["works"]
        if extra_doaj:
            extra["doaj"] = extra_doaj

        extra_ia = dict()
        # TODO: would like an ia.longtail_ia flag
        if row.get("sim"):
            # NB: None case of the .get() here is blech, but othrwise
            # extra['ia'].get('sim') would be false-y, breaking 'any_ia_sim' later on
            extra_ia["sim"] = {
                "year_spans": row["sim"].get("year_spans"),
            }
        if extra_ia:
            extra["ia"] = extra_ia

        name = clean_str(row.get("name"))
        if not name:
            return None

        ce = ContainerEntity(
            issnl=row["issnl"],
            issne=row.get("issne"),
            issnp=row.get("issnp"),
            container_type=None,  # TODO
            name=name,
            publisher=clean_str(row.get("publisher")),
            wikidata_qid=None,  # TODO
            extra=extra,
        )
        return ce

    def try_update(self, ce: ContainerEntity) -> bool:

        existing = None
        try:
            existing = self.api.lookup_container(issn=ce.issnl)
        except fatcat_openapi_client.rest.ApiException as err:
            if err.status != 404:
                raise err

        if not existing:
            # doesn't exist, create it
            return True

        # for now, only update KBART, and only if there is new content
        if not existing.extra:
            existing.extra = dict()
        if ce.extra.get("kbart") and (existing.extra.get("kbart") != ce.extra["kbart"]):
            if not existing.extra.get("kbart"):
                existing.extra["kbart"] = {}
            existing.extra["kbart"].update(ce.extra["kbart"])
            self.api.update_container(self.get_editgroup_id(), existing.ident, existing)
            self.counts["update"] += 1
            return False
        else:
            self.counts["exists"] += 1
            return False

        # if we got this far, it's a bug
        raise NotImplementedError

    def insert_batch(self, batch: List[ContainerEntity]) -> None:
        self.api.create_container_auto_batch(
            fatcat_openapi_client.ContainerAutoBatch(
                editgroup=fatcat_openapi_client.Editgroup(
                    description=self.editgroup_description, extra=self.editgroup_extra
                ),
                entity_list=batch,
            )
        )