-
Notifications
You must be signed in to change notification settings - Fork 16
/
preservation.py
130 lines (100 loc) · 3.74 KB
/
preservation.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
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
from portality.dao import DomainObject
from copy import deepcopy
from portality.lib import dates
class PreservationState(DomainObject):
"""~~Preservation:Model~~
Model class for preservation feature"""
__type__ = "preserve"
@property
def status(self):
return self.data.get("status")
@status.setter
def status(self, value):
self.data["status"] = value
@property
def filename(self):
return self.data.get("filename")
@property
def owner(self):
return self.data.get("owner")
@property
def background_task_id(self):
return self.data.get("background_task_id")
@background_task_id.setter
def background_task_id(self, value):
self.data["background_task_id"] = value
@property
def created_timestamp(self):
if "created_date" not in self.data:
return None
return dates.parse(self.data["created_date"])
@property
def error(self):
return self.data.get("error")
@property
def error_details(self):
return self.data.get("error_details")
@property
def articles_info(self):
return self.data.get("articles_info", {})
def initiated(self, owner, filename, status="initiated"):
self.data["filename"] = filename
self.data["owner"] = owner
self.data["articles_info"] = {}
self.status = status
def validated(self):
self.status = "validated"
def pending(self):
self.status = "pending"
def uploaded_to_ia(self):
self.status = "uploaded"
def failed(self, message, details=None):
self.status = "failed"
self.data["error"] = message
if details is not None:
self.data["error_details"] = details
def partial(self):
self.status = "partial"
def successful_articles(self, articles_list):
if articles_list is not None and len(articles_list) > 0:
self.data["articles_info"]["successful_articles"] = ", ".join(articles_list)
def unowned_articles(self, articles_list):
if articles_list is not None and len(articles_list) > 0:
self.data["articles_info"]["unowned_articles"] = ", ".join(articles_list)
def no_identifier_articles(self, articles_list):
if articles_list is not None and len(articles_list) > 0:
self.data["articles_info"]["no_identifier_articles"] = ", ".join(articles_list)
def unbagged_articles(self, articles_list):
if articles_list is not None and len(articles_list) > 0:
self.data["articles_info"]["unbagged_articles"] = ", ".join(articles_list)
def not_found_articles(self, articles_list):
if articles_list is not None and len(articles_list) > 0:
self.data["articles_info"]["not_found_articles"] = ", ".join(articles_list)
def no_files_articles(self, articles_list):
if articles_list is not None and len(articles_list) > 0:
self.data["articles_info"]["no_files_articles"] = ", ".join(articles_list)
@classmethod
def by_owner(cls, owner, size=10):
q = OwnerFileQuery(owner)
res = cls.query(q=q.query())
rs = [PreservationState(**r.get("_source")) for r in res.get("hits", {}).get("hits", [])]
return rs
class OwnerFileQuery(object):
base_query = {
"query": {
"bool": {
"must": []
}
},
"sort": [
{"created_date": "desc"}
],
"size": 10
}
def __init__(self, owner, size=10):
self._query = deepcopy(self.base_query)
owner_term = {"match": {"owner.exact": owner}}
self._query["query"]["bool"]["must"].append(owner_term)
self._query["size"] = size
def query(self):
return self._query