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 | """
Shared helpers for online-source transforms.
Both `MetronApiTransform` and `ComicVineApiTransform` need to convert
list-of-dicts collections into comicbox's `SimpleNamedDictField` shape
(`{name: {extra_fields}}`), parse credit/role lists, and build
identifier sub-dicts. Putting the shared logic here keeps the per-
source transforms readable.
"""
from __future__ import annotations
import re
from collections.abc import Iterable, Mapping
from typing import Any
from comicbox.identifiers.identifiers import create_identifier
def named_dict(
items: Iterable[Mapping[str, Any]] | None,
*,
key: str = "name",
) -> dict[str, dict]:
"""
Convert a list of `{name, ...}` dicts into a `{name: {}}` dict.
Used for SimpleNamedDictField targets: characters, teams, story
arcs, locations, universes, etc.
"""
if not items:
return {}
out: dict[str, dict] = {}
for item in items:
if not item:
continue
name = item.get(key)
if not name:
continue
out.setdefault(str(name), {})
return out
def named_dict_with_id(
items: Iterable[Mapping[str, Any]] | None,
source: str,
id_type: str = "issue",
) -> dict[str, dict]:
"""
Like `named_dict`, but each entry carries an `identifiers` sub-dict.
The id field of each item maps to comicbox's IdentifiedNameSchema
`identifiers.<source>.{key, url}`. Used where the upstream provides
an id alongside the name (most named collections on Metron and CV).
"""
if not items:
return {}
out: dict[str, dict] = {}
for item in items:
if not item:
continue
name = item.get("name")
if not name:
continue
entry: dict = {}
if (item_id := item.get("id")) is not None:
entry["identifiers"] = {source: build_identifier(source, id_type, item_id)}
out.setdefault(str(name), entry)
return out
def build_identifier(source: str, id_type: str, value: Any) -> dict[str, str]:
"""Build the {key, url} sub-dict for one identifier."""
s = str(value)
if not s:
return {}
try:
identifier = create_identifier(source, s, id_type=id_type)
except Exception:
return {"key": s}
# `create_identifier` already returns a {key, url?} dict.
return dict(identifier) or {"key": s}
_ROLE_SPLIT_RE = re.compile(r"\s*,\s*")
def parse_creator_roles(raw: str | None) -> list[str]:
"""
Split a comma-string of role names ("writer, penciler") into a list.
ComicVine returns creator roles as a comma-separated string;
mokkari returns them as a list of `{id, name}` dicts so this is
only used for the CV path. Empty / None inputs produce [].
"""
if not raw:
return []
return [r.strip() for r in _ROLE_SPLIT_RE.split(raw) if r.strip()]
def _extract_role_names(roles_raw: Any, *, role_is_string: bool) -> Iterable[str]:
"""Yield role-name strings from either CV's comma-string form or mokkari's list-of-dicts."""
if role_is_string:
yield from parse_creator_roles(roles_raw)
return
if not roles_raw:
return
for role_item in roles_raw:
if isinstance(role_item, Mapping) and (name := role_item.get("name")):
yield str(name)
def credits_to_cb(
credits_list: Iterable[Mapping[str, Any]] | None,
*,
creator_key: str,
role_key: str,
role_is_string: bool = False,
source: str | None = None,
creator_id_type: str = "creator",
) -> dict[str, dict]:
"""
Convert a list of credits into comicbox's nested `credits` shape.
The shape is `{<creator>: {roles: {<role>: {}}, identifiers: {...}}}`.
`role_is_string=True` means `role` is a comma-separated string (CV).
Otherwise `role` is a list of `{id, name}` dicts (mokkari).
`source` and `creator_id_type` populate the per-creator identifiers
sub-dict from the upstream creator id; pass None to skip.
"""
if not credits_list:
return {}
out: dict[str, dict] = {}
for credit in credits_list:
if not credit:
continue
creator_name = credit.get(creator_key)
if not creator_name:
continue
entry: dict = out.setdefault(str(creator_name), {})
roles_dict: dict = entry.setdefault("roles", {})
for role in _extract_role_names(
credit.get(role_key), role_is_string=role_is_string
):
roles_dict.setdefault(role, {})
if (
source
and (creator_id := credit.get("id")) is not None
and "identifiers" not in entry
):
entry["identifiers"] = {
source: build_identifier(source, creator_id_type, creator_id)
}
return out
def named_block(parent: Mapping[str, Any], key: str) -> dict[str, str] | None:
"""
Pull `{key: {name: <name>}}` out of a nested {id, name} dict, or None.
Used for publisher / imprint / similar single-named blocks where the
upstream returns a small `{id, name}` object but comicbox stores it
as `{name: ...}` only.
"""
nested = parent.get(key) or {}
if name := nested.get("name"):
return {"name": name}
return None
def reprints_to_cb(
reprints: Iterable[Mapping[str, Any]] | None,
*,
source: str,
) -> list[dict]:
"""
Convert mokkari's `reprints` list into comicbox's ReprintSchema list.
mokkari `Reprint` has `id` (Metron issue id) and `issue` (display
string like "Series Name #N"). We pass the issue string through;
the metron id goes into `identifiers.<source>` per ReprintSchema.
"""
if not reprints:
return []
out: list[dict] = []
for r in reprints:
if not r:
continue
entry: dict = {}
if issue := r.get("issue"):
entry["issue"] = str(issue)
if (rid := r.get("id")) is not None:
entry["identifiers"] = {source: build_identifier(source, "issue", rid)}
if entry:
out.append(entry)
return out
|