-
Notifications
You must be signed in to change notification settings - Fork 2
/
Copy pathperpetual_access.py
112 lines (91 loc) · 3.47 KB
/
perpetual_access.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
# coding: utf-8
from cached_property import cached_property
from datetime import datetime
from collections import OrderedDict
from app import db
from package_input import PackageInput
class PerpetualAccess(db.Model):
__tablename__ = "jump_perpetual_access"
issn_l = db.Column(db.Text, primary_key=True)
start_date = db.Column(db.DateTime)
end_date = db.Column(db.DateTime)
package_id = db.Column(db.Text, db.ForeignKey("jump_account_package.package_id"), primary_key=True)
package = None
@cached_property
def journal_metadata(self):
from openalex import MissingJournalMetadata, all_journal_metadata
meta = all_journal_metadata.get(self.issn_l)
if not meta:
meta = MissingJournalMetadata(issn_l=self.issn_l)
return meta
@cached_property
def issns(self):
return self.journal_metadata.issns
@cached_property
def display_issns(self):
return self.journal_metadata.display_issns
@cached_property
def display_issn_l(self):
return self.journal_metadata.display_issn_l
@cached_property
def title(self):
return self.journal_metadata.title
@cached_property
def publisher(self):
return self.journal_metadata.publisher
@cached_property
def display_start_date(self):
if not self.start_date:
return None
return self.start_date.isoformat()[0:10]
@cached_property
def display_end_date(self):
if not self.end_date:
return None
return self.end_date.isoformat()[0:10]
def to_dict(self):
return OrderedDict([
("issn_l_prefixed", self.display_issn_l),
("issn_l", self.issn_l),
("issns", self.display_issns),
("title", self.title),
("publisher", self.publisher),
("start_date", self.display_start_date),
("end_date", self.display_end_date),
])
class PerpetualAccessInput(db.Model, PackageInput):
__tablename__ = "jump_perpetual_access_input"
package_id = db.Column(db.Text, db.ForeignKey("jump_account_package.package_id"), primary_key=True)
issn = db.Column(db.Text, primary_key=True)
start_date = db.Column(db.DateTime)
end_date = db.Column(db.DateTime)
def import_view_name(self):
return "jump_perpetual_access_view"
def destination_table(self):
return PerpetualAccess.__tablename__
def file_type_label(self):
return "perpetual-access"
def issn_columns(self):
return ["issn"]
def csv_columns(self):
return {
"start_date": {
"normalize": lambda date, warn_if_blank=False: self.normalize_date(date, default=datetime(1970, 1, 1), warn_if_blank=warn_if_blank),
"name_snippets": ["start", "begin"],
"required": True
},
"end_date": {
"normalize": lambda date, warn_if_blank=False: self.normalize_date(date, default=datetime(1970, 12, 31), warn_if_blank=warn_if_blank),
"name_snippets": ["end"],
"required": True
},
"issn": {
"normalize": self.normalize_issn,
"name_snippets": ["issn"],
"excluded_name_snippets": ["online", "e-", "eissn"],
"required": True,
"warn_if_blank": True,
}
}
def clear_caches(self, my_package):
super(PerpetualAccessInput, self).clear_caches(my_package)