Coverage for portality / crosswalks / journal_form.py: 86%
400 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-04 09:41 +0100
1from werkzeug.datastructures import MultiDict
3from portality import models, lcc
4from portality.datasets import licenses
5from portality.forms.utils import expanded2compact
6from portality.models import Account
7from portality.lib import dates
8from builtins import ValueError
10class JournalGenericXWalk(object):
11 """
12 ~~Journal:Crosswalk->Journal:Form~~
13 ~~->Journal:Model~~
14 """
15 @classmethod
16 def forminfo2multidict(cls, forminfo):
17 formdata = expanded2compact(forminfo,
18 join_lists={"keywords": ",", "subject": ","},
19 repeat_lists=["preservation_service_library", "language"]
20 )
21 return MultiDict(formdata)
23 @classmethod
24 def is_new_editor_group(cls, form, old):
25 old_eg = old.editor_group
26 new_eg = form.editor_group.data
27 return old_eg != new_eg and new_eg is not None and new_eg != ""
29 @classmethod
30 def is_new_editor(cls, form, old):
31 old_ed = old.editor
32 new_ed = form.editor.data
33 return old_ed != new_ed and new_ed is not None and new_ed != ""
35 @classmethod
36 def form_diff(cls, a_formdata, b_formdata):
38 def _serialise(x):
39 if isinstance(x, list):
40 return "; ".join(sorted([_serialise(y) for y in x]))
41 elif isinstance(x, dict):
42 kvs = []
43 for k, v in x.items():
44 kvs.append(k + ":" + str(v))
45 kvs.sort()
46 return ", ".join(kvs)
47 else:
48 return x
50 diff = {}
51 for k, v in b_formdata.items():
52 if k in a_formdata:
53 if isinstance(a_formdata[k], list):
54 for entry in a_formdata[k]:
55 if entry not in v:
56 diff[k] = {"a" : _serialise(a_formdata[k]), "b" : _serialise(v)}
57 break
58 for entry in v:
59 if entry not in a_formdata[k]:
60 diff[k] = {"a": _serialise(a_formdata[k]), "b": _serialise(v)}
61 break
63 elif isinstance(a_formdata[k], dict):
64 pass # No instances of this in our current form, so leaving it out
65 else:
66 if a_formdata[k] != v:
67 diff[k] = {"a" : a_formdata[k], "b" : v}
69 else:
70 diff[k] = {"a" : "", "b": _serialise(v)}
72 return diff
74 @classmethod
75 def form2bibjson(cls, form, bibjson):
76 # pre-strip all the form content whitespace
77 for field in form:
78 if hasattr(field.data, "strip"):
79 field.data = field.data.strip()
81 if form.alternative_title.data:
82 bibjson.alternative_title = form.alternative_title.data
84 for apc_record in form.apc_charges.data:
85 if not apc_record["apc_currency"] and not apc_record["apc_max"]:
86 continue
87 bibjson.add_apc(apc_record["apc_currency"], apc_record["apc_max"])
89 if form.apc_url.data:
90 bibjson.apc_url = form.apc_url.data
92 if form.preservation_service.data:
93 pres_services = [e for e in form.preservation_service.data if e not in ["national_library", "none", "other"]]
94 if "other" in form.preservation_service.data and form.preservation_service_other.data:
95 pres_services.append(form.preservation_service_other.data)
96 bibjson.set_preservation(pres_services, None)
97 if "national_library" in form.preservation_service.data and form.preservation_service_library.data:
98 libs = [x for x in form.preservation_service_library.data if x]
99 for lib in libs:
100 bibjson.add_preservation_library(lib)
101 if "none" in form.preservation_service.data:
102 bibjson.has_preservation = False
104 if form.preservation_service_url.data:
105 bibjson.preservation_url = form.preservation_service_url.data
107 if form.copyright_author_retains.data:
108 bibjson.author_retains_copyright = form.copyright_author_retains.data == 'y'
110 if form.copyright_url.data:
111 bibjson.copyright_url = form.copyright_url.data
113 if form.publisher_name.data:
114 bibjson.publisher_name = form.publisher_name.data
115 if form.publisher_country.data:
116 bibjson.publisher_country = form.publisher_country.data
118 if form.deposit_policy.data:
119 dep_services = [e for e in form.deposit_policy.data if e not in ["none", "other"]]
120 if "other" in form.deposit_policy.data and form.deposit_policy_other.data:
121 dep_services.append(form.deposit_policy_other.data)
122 if dep_services:
123 bibjson.deposit_policy = dep_services
124 else:
125 bibjson.has_deposit_policy = False
127 if form.review_process.data or form.review_url.data:
128 processes = None
129 if form.review_process.data:
130 processes = [e for e in form.review_process.data if e not in ["other"]]
131 if "other" in form.review_process.data and form.review_process_other.data:
132 processes.append(form.review_process_other.data)
133 bibjson.set_editorial_review(processes, form.review_url.data, form.editorial_board_url.data)
135 if form.pissn.data:
136 bibjson.pissn = form.pissn.data
138 if form.eissn.data:
139 bibjson.eissn = form.eissn.data
141 if form.institution_name.data:
142 bibjson.institution_name = form.institution_name.data
143 if form.institution_country.data:
144 bibjson.institution_country = form.institution_country.data
146 if form.keywords.data:
147 bibjson.keywords = form.keywords.data
149 if form.language.data:
150 norm_language = [x for x in form.language.data if x != ""]
151 if norm_language:
152 bibjson.language = norm_language # select multiple field - gives a list back
154 lurl = form.license_terms_url.data
155 if lurl:
156 bibjson.license_terms_url = lurl
157 if form.license.data:
158 for ltype in form.license.data:
159 by, nc, nd, sa = None, None, None, None
160 if ltype in licenses:
161 by = licenses[ltype]['BY']
162 nc = licenses[ltype]['NC']
163 nd = licenses[ltype]['ND']
164 sa = licenses[ltype]['SA']
165 lurl = licenses[ltype]["url"]
166 elif form.license_attributes.data:
167 by = True if 'BY' in form.license_attributes.data else False
168 nc = True if 'NC' in form.license_attributes.data else False
169 nd = True if 'ND' in form.license_attributes.data else False
170 sa = True if 'SA' in form.license_attributes.data else False
171 bibjson.add_license(ltype, url=lurl, by=by, nc=nc, nd=nd, sa=sa)
173 # FIXME: this is not quite what we planned
174 if form.license_display.data:
175 bibjson.article_license_display = ["Embed"] if form.license_display.data == "y" else ["No"]
177 if form.license_display_example_url.data:
178 bibjson.article_license_display_example_url = form.license_display_example_url.data
180 if form.boai.data:
181 bibjson.boai = form.boai.data == "y"
183 if form.oa_start.data:
184 bibjson.oa_start = form.oa_start.data
186 if form.oa_statement_url.data:
187 bibjson.oa_statement_url = form.oa_statement_url.data
189 if form.journal_url.data:
190 bibjson.journal_url = form.journal_url.data
192 if form.aims_scope_url.data:
193 bibjson.aims_scope_url = form.aims_scope_url.data
195 if form.author_instructions_url.data:
196 bibjson.author_instructions_url = form.author_instructions_url.data
198 if form.waiver_url.data:
199 bibjson.waiver_url = form.waiver_url.data
201 if form.persistent_identifiers.data:
202 schemes = [e for e in form.persistent_identifiers.data if e not in ["none", "other"]]
203 if "other" in form.persistent_identifiers.data and form.persistent_identifiers_other.data:
204 schemes.append(form.persistent_identifiers_other.data)
205 if len(schemes) > 0:
206 bibjson.pid_scheme = schemes
207 elif "none" in form.persistent_identifiers.data:
208 bibjson.has_pid_scheme = False
210 if form.plagiarism_detection.data:
211 has_detection = form.plagiarism_detection.data == "y"
212 bibjson.set_plagiarism_detection(form.plagiarism_url.data, has_detection)
214 if form.publication_time_weeks.data:
215 bibjson.publication_time_weeks = form.publication_time_weeks.data
217 if form.other_charges_url.data:
218 bibjson.other_charges_url = form.other_charges_url.data
220 if form.title.data:
221 bibjson.title = form.title.data
223 if form.apc.data:
224 bibjson.has_apc = form.apc.data == "y"
226 if form.has_other_charges.data:
227 has_other = form.has_other_charges.data == "y"
228 bibjson.has_other_charges = has_other
230 if form.has_waiver.data:
231 has_waiver = form.has_waiver.data == "y"
232 bibjson.has_waiver = has_waiver
234 if form.deposit_policy_url.data:
235 bibjson.deposit_policy_url = form.deposit_policy_url.data
237 # continuations information
238 if getattr(form, "continues", None):
239 bibjson.replaces = form.continues.data
240 if getattr(form, "continued_by", None):
241 bibjson.is_replaced_by = form.continued_by.data
242 if getattr(form, "discontinued_date", None):
243 bibjson.discontinued_date = form.discontinued_date.data
245 # subject information
246 if getattr(form, "subject", None):
247 new_subjects = []
248 incoming = form.subject.data
249 if not isinstance(incoming, list):
250 incoming = [x.strip() for x in form.subject.data.split(",")]
251 for code in incoming:
252 sobj = {"scheme": 'LCC', "term": lcc.lookup_code(code), "code": code}
253 new_subjects.append(sobj)
254 bibjson.subject = new_subjects
256 s2o = getattr(form, "s2o", None)
257 if s2o is not None and (s2o.data == "y" or s2o.data is True):
258 bibjson.add_label("s2o")
260 mirror = getattr(form, "mirror", None)
261 if mirror is not None and (mirror.data == "y" or mirror.data is True):
262 bibjson.add_label("mirror")
264 ojc = getattr(form, "ojc", None)
265 if ojc is not None and (ojc.data == "y" or ojc.data is True):
266 bibjson.add_label("ojc")
268 @classmethod
269 def form2admin(cls, form, obj):
270 if getattr(form, "notes", None):
271 for formnote in form.notes.data:
272 if formnote["note"]:
273 note_date = formnote["note_date"]
274 note_id = formnote["note_id"]
275 obj.add_note(formnote["note"], date=note_date, id=note_id,
276 author_id=formnote["note_author_id"])
278 if getattr(form, "flags", None):
279 for flag in form.flags.data:
280 if flag.get("flag_note") is None or flag.get("flag_note") == "":
281 continue
283 flag_date = flag["flag_created_date"]
284 try:
285 if flag.get("flag_deadline") == "" or flag.get("flag_deadline") is None:
286 flag_deadline = dates.far_in_the_future()
287 else:
288 flag_deadline = dates.parse(flag.get("flag_deadline"), format=dates.FMT_DATE_STD)
289 except:
290 # FIXME: why is there validation in the crosswalk?
291 raise ValueError("Flag deadline must be a valid date in BigEnd format (ie. YYYY-MM-DD)")
292 flag_assigned_to = flag["flag_assignee"]
293 flag_author = flag["flag_setter"]
294 flag_id = flag["flag_note_id"]
295 flag_note = flag["flag_note"]
296 obj.add_note(flag_note, date=flag_date, id=flag_id,
297 author_id=flag_author, assigned_to=flag_assigned_to, deadline=flag_deadline)
299 if getattr(form, 'owner', None):
300 owner = form.owner.data
301 if owner:
302 owner = owner.strip()
303 obj.set_owner(owner)
305 if getattr(form, 'editor_group', None):
306 editor_group = form.editor_group.data
307 if editor_group:
308 editor_group = editor_group.strip()
309 obj.set_editor_group(editor_group)
311 if getattr(form, "editor", None):
312 editor = form.editor.data
313 if editor:
314 editor = editor.strip()
315 obj.set_editor(editor)
317 if getattr(form, "last_full_review", None):
318 lfr = form.last_full_review.data
319 if lfr:
320 lfr = lfr.strip()
321 obj.last_full_review = lfr
323 @classmethod
324 def bibjson2form(cls, bibjson, forminfo):
325 from portality.models import JournalLikeBibJSON
326 from portality.forms.application_forms import ApplicationFormFactory
327 from portality.datasets import get_currency_code
329 assert isinstance(bibjson, JournalLikeBibJSON)
331 forminfo['alternative_title'] = bibjson.alternative_title
333 # when we crosswalk the apc currency we also have to check that it is current
334 # so that we only present information to the form is the form is capable of
335 # presenting it on to the user
336 forminfo["apc_charges"] = []
337 for apc_record in bibjson.apc:
338 currency = apc_record.get("currency")
339 check = get_currency_code(currency, fail_if_not_found=True)
340 if check is None:
341 continue
343 forminfo["apc_charges"].append({
344 "apc_currency" : apc_record.get("currency"),
345 "apc_max" : apc_record.get("price")
346 })
348 forminfo["apc_url"] = bibjson.apc_url
350 pres_choices = [x for x, y in ApplicationFormFactory.choices_for("preservation_service")]
351 if bibjson.preservation_services is not None:
352 forminfo["preservation_service"] = [x for x in bibjson.preservation_services if x in pres_choices]
353 forminfo["preservation_service_other"] = " ".join([x for x in bibjson.preservation_services if x not in pres_choices])
354 if len(forminfo["preservation_service_other"]) > 0:
355 forminfo["preservation_service"].append("other")
356 if "preservation_service" not in forminfo:
357 forminfo["preservation_service"] = []
358 if bibjson.preservation_library:
359 forminfo["preservation_service_library"] = bibjson.preservation_library
360 forminfo["preservation_service"].append("national_library")
361 if bibjson.has_preservation is False and len(forminfo["preservation_service"]) == 0:
362 forminfo["preservation_service"].append("none")
364 forminfo["preservation_service_url"] = bibjson.preservation_url
365 if bibjson.author_retains_copyright is not None:
366 forminfo["copyright_author_retains"] = "y" if bibjson.author_retains_copyright else "n"
367 forminfo["copyright_url"] = bibjson.copyright_url
369 forminfo["publisher_name"] = bibjson.publisher_name
370 forminfo["publisher_country"] = bibjson.publisher_country
372 dep_choices = [x for x, y in ApplicationFormFactory.choices_for("deposit_policy")]
373 if bibjson.deposit_policy:
374 forminfo["deposit_policy"] = [x for x in bibjson.deposit_policy if x in dep_choices]
375 forminfo["deposit_policy_other"] = " ".join([x for x in bibjson.deposit_policy if x not in dep_choices])
376 if len(forminfo["deposit_policy_other"]) > 0:
377 forminfo["deposit_policy"].append("other")
378 if "deposit_policy" not in forminfo:
379 forminfo["deposit_policy"] = []
380 if bibjson.has_deposit_policy is False and len(forminfo["deposit_policy"]) == 0:
381 forminfo["deposit_policy"].append("none")
383 review_choices = [x for x, y in ApplicationFormFactory.choices_for("review_process")]
384 if bibjson.editorial_review_process:
385 forminfo["review_process"] = [x for x in bibjson.editorial_review_process if x in review_choices]
386 forminfo["review_process_other"] = " ".join([x for x in bibjson.editorial_review_process if x not in review_choices])
387 if len(forminfo["review_process_other"]) > 0:
388 forminfo["review_process"].append("other")
390 forminfo["review_url"] = bibjson.editorial_review_url
391 forminfo["pissn"] = bibjson.pissn
392 forminfo["eissn"] = bibjson.eissn
394 if bibjson.institution_name:
395 forminfo["institution_name"] = bibjson.institution_name
396 if bibjson.institution_country:
397 forminfo["institution_country"] = bibjson.institution_country
399 forminfo['keywords'] = bibjson.keywords # fixme: all keywords are being rendered as one single item
400 forminfo['language'] = bibjson.language
402 license_attributes = []
403 ltypes = []
404 for l in bibjson.licenses:
405 ltypes.append(l.get("type"))
406 if l.get("type") == "Publisher's own license":
407 if l.get("BY"): license_attributes.append("BY")
408 if l.get("SA"): license_attributes.append("SA")
409 if l.get("NC"): license_attributes.append("NC")
410 if l.get("ND"): license_attributes.append("ND")
411 forminfo["license_attributes"] = license_attributes
412 forminfo["license"] = ltypes
414 if bibjson.article_license_display is not None and len(bibjson.article_license_display) > 0:
415 forminfo["license_display"] = "y" if "Embed" in bibjson.article_license_display else "n"
416 forminfo["license_display_example_url"] = bibjson.article_license_display_example_url
417 if bibjson.boai is not None:
418 forminfo["boai"] = 'y' if bibjson.boai else 'n'
419 forminfo["license_terms_url"] = bibjson.license_terms_url
420 if bibjson.oa_start:
421 forminfo["oa_start"] = bibjson.oa_start
422 forminfo["oa_statement_url"] = bibjson.oa_statement_url
423 forminfo["journal_url"] = bibjson.journal_url
424 forminfo["aims_scope_url"] = bibjson.aims_scope_url
425 forminfo["editorial_board_url"] = bibjson.editorial_board_url
426 forminfo["author_instructions_url"] = bibjson.author_instructions_url
427 forminfo["waiver_url"] = bibjson.waiver_url
429 pid_choices = [x for x, y in ApplicationFormFactory.choices_for("persistent_identifiers")]
430 if bibjson.pid_scheme:
431 forminfo["persistent_identifiers"] = [x for x in bibjson.pid_scheme if x in pid_choices]
432 forminfo["persistent_identifiers_other"] = " ".join([x for x in bibjson.pid_scheme if x not in pid_choices])
433 if len(forminfo["persistent_identifiers_other"]) > 0:
434 forminfo["persistent_identifiers"].append("other")
435 if bibjson.has_pid_scheme is False: # distinct from None
436 forminfo["persistent_identifiers"] = ["none"]
438 if bibjson.plagiarism_detection is not None:
439 forminfo["plagiarism_detection"] = "y" if bibjson.plagiarism_detection else "n"
440 forminfo["plagiarism_url"] = bibjson.plagiarism_url
441 forminfo["publication_time_weeks"] = bibjson.publication_time_weeks
442 forminfo["other_charges_url"] = bibjson.other_charges_url
443 forminfo['title'] = bibjson.title
445 # FIXME: these translations don't handle partial records (i.e. drafts)
446 # we may want to add some methods to allow the settedness of these fields to be checked
447 if bibjson.has_apc is not None:
448 forminfo["apc"] = "y" if bibjson.has_apc else "n"
449 if bibjson.has_other_charges is not None:
450 forminfo["has_other_charges"] = "y" if bibjson.has_other_charges else "n"
451 if bibjson.has_waiver is not None:
452 forminfo["has_waiver"] = "y" if bibjson.has_waiver else "n"
454 forminfo["deposit_policy_url"] = bibjson.deposit_policy_url
456 # continuation information
457 forminfo["continues"] = bibjson.replaces
458 forminfo["continued_by"] = bibjson.is_replaced_by
459 forminfo["discontinued_date"] = bibjson.discontinued_date
461 # subject classifications
462 forminfo['subject'] = []
463 for s in bibjson.subject:
464 if "code" in s:
465 forminfo['subject'].append(s['code'])
467 # labels
468 forminfo['s2o'] = "s2o" in bibjson.labels
469 forminfo['mirror'] = "mirror" in bibjson.labels
470 forminfo['ojc'] = "ojc" in bibjson.labels
472 @classmethod
473 def admin2form(cls, obj, forminfo):
474 forminfo['notes'] = []
475 for n in obj.ordered_notes_except_flags:
476 author_id = n.get('author_id', '')
477 note_author_name = f'{Account.get_name_safe(author_id)} ({author_id})' if author_id else ''
478 note_obj = {'note': n['note'], 'note_date': n['date'], 'note_id': n['id'],
479 'note_author': note_author_name,
480 'note_author_id': author_id,
481 }
482 forminfo['notes'].append(note_obj)
484 forminfo["flags"] = []
485 if obj.flags:
486 # display only the newest flag
487 flag = obj.flags[0]
488 author_id = flag["author_id"]
489 flag_setter = f'{Account.get_name_safe(author_id)} ({author_id})' if author_id else ''
490 deadline = flag["flag"].get("deadline")
491 flag_deadline = (
492 deadline if (deadline and deadline != dates.far_in_the_future()) else ""
493 )
494 flag_assignee = flag["flag"]["assigned_to"]
495 flag_obj = {"flag_created_date": flag["date"], "flag_note": flag["note"],
496 "flag_note_id": flag["id"], "flag_setter": flag_setter, "flag_assignee": flag_assignee,
497 "flag_deadline": flag_deadline }
498 forminfo['flags'].append(flag_obj)
500 forminfo['owner'] = obj.owner
501 if obj.editor_group is not None:
502 forminfo['editor_group'] = obj.editor_group
503 if obj.editor is not None:
504 forminfo['editor'] = obj.editor
506 if getattr(obj, "last_full_review", None):
507 forminfo["last_full_review"] = obj.last_full_review
510class JournalFormXWalk(JournalGenericXWalk):
512 @classmethod
513 def form2obj(cls, form):
514 journal = models.Journal()
515 bibjson = journal.bibjson()
517 # first do the generic crosswalk to bibjson
518 cls.form2bibjson(form, bibjson)
520 # then do the admin fields
521 cls.form2admin(form, journal)
523 return journal
525 @classmethod
526 def obj2form(cls, obj):
527 forminfo = {}
528 bibjson = obj.bibjson()
530 cls.bibjson2form(bibjson, forminfo)
531 cls.admin2form(obj, forminfo)
533 return forminfo