-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy patharchive.py
197 lines (169 loc) · 6.52 KB
/
archive.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
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
import requests
from pydantic import BaseModel
from script_house.utils import JsonUtils
from lanraragi_api.Config import assert_use_untested_functions
from lanraragi_api.base.base import BaseAPICall
from lanraragi_api.base.category import Category
ARCHIVE_TAG_VALUES_SET = "ONLY_VALUES"
class Archive(BaseModel):
arcid: str
isnew: str
extension: str
pagecount: int
progress: int
# k1:v1, k2:v21, k2:v22, v3, v4
# allow duplicate keys, only values
tags: str
lastreadtime: int
title: str
def __tags_to_dict(self) -> dict[str, list[str]]:
tags = self.tags.split(',')
ans = {}
for t in tags:
if t == '':
continue
t = t.strip()
if ':' in t:
kv = t.split(':')
k = kv[0]
v = kv[1]
if k not in ans:
ans[k] = []
ans[k].append(v)
else:
k = ARCHIVE_TAG_VALUES_SET
if k not in ans:
ans[k] = []
ans[k].append(t)
return ans
def __dict_to_tags(self, json: dict[str, list[str]]):
"""
The function will modify the object
"""
tags = ""
modified: bool = False
for k in json:
for v in json[k]:
modified = True
if k == ARCHIVE_TAG_VALUES_SET:
tags += f"{v},"
else:
tags += f"{k}:{v},"
if modified:
tags = tags[:-1]
self.tags = tags
def get_artists(self) -> list[str]:
return self.__tags_to_dict()['artist']
def set_artists(self, artists: list[str]):
json = self.__tags_to_dict()
json['artist'] = artists
self.__dict_to_tags(json)
def remove_artists(self):
json = self.__tags_to_dict()
json['artist'] = []
self.__dict_to_tags(json)
def has_artists(self) -> bool:
return "artist" in self.tags
class ArchiveAPI(BaseAPICall):
"""
Everything dealing with Archives.
"""
def get_all_archives(self) -> list[Archive]:
resp = requests.get(f"{self.server}/api/archives", params={'key': self.key},
headers=self.build_headers())
list = JsonUtils.to_obj(resp.text)
return [JsonUtils.to_obj(JsonUtils.to_str(o), Archive) for o in list]
def get_untagged_archives(self) -> list[str]:
"""
Get archives that don't have any tags recorded. This follows the same
rules as the Batch Tagging filter and will include archives that have
parody:, date_added:, series: or artist: tags.
:return: list of archive IDs
"""
resp = requests.get(f"{self.server}/api/archives/untagged", params={'key': self.key},
headers=self.build_headers())
return JsonUtils.to_obj(resp.text)
def get_archive_metadata(self, id: str) -> Archive:
"""
Get Metadata (title, tags) for a given Archive.
:param id: ID of the Archive to process.
:return:
"""
resp = requests.get(f"{self.server}/api/archives/{id}/metadata", params={'key': self.key},
headers=self.build_headers())
return JsonUtils.to_obj(resp.text, Archive)
def get_archive_categories(self, id: str) -> list[Category]:
"""
Get all the Categories which currently refer to this Archive ID.
:param id: ID of the Archive to process.
:return:
"""
resp = requests.get(f"{self.server}/api/archives/{id}/categories", params={'key': self.key},
headers=self.build_headers())
clist = JsonUtils.to_obj(resp.text)["categories"]
return [JsonUtils.to_obj(JsonUtils.to_str(c), Category) for c in clist]
def get_archive_tankoubons(self, id: str) -> list[str]:
"""
Get all the Tankoubons which currently refer to this Archive ID.
Tankoubon: 単行本
:param id: ID of the Archive to process.
:return: list of tankoubon ids
"""
resp = requests.get(f"{self.server}/api/archives/{id}/tankoubons", params={'key': self.key},
headers=self.build_headers())
return JsonUtils.to_obj(resp.text)['tankoubons']
def get_archive_thumbnail(self):
# TODO: used not so often
pass
def download_archive(self, id: str) -> bytes:
"""
Download an Archive from the server.
:param id: ID of the Archive to download.
:return: bytes representing the archive. You can write it to a file.
"""
resp = requests.get(f"{self.server}/api/archives/{id}/download", params={'key': self.key},
headers=self.build_headers())
return resp.content
def extract_archive(self):
# TODO: used not so often
pass
def clear_archive_new_flag(self, id: str) -> dict:
"""
Clears the "New!" flag on an archive.
:param id: ID of the Archive to process.
:return: operation result
"""
# TODO: untested
assert_use_untested_functions()
resp = requests.delete(f"{self.server}/api/archives/{id}/isnew", params={'key': self.key},
headers=self.build_headers())
return JsonUtils.to_obj(resp.text)
def update_reading_progression(self):
# TODO: used not so often
pass
def update_thumbnail(self):
# TODO: used not so often
pass
def update_archive_metadata(self, id: str, archive: Archive) -> dict:
"""
Update tags and title for the given Archive. Data supplied to the server through
this method will <b>overwrite</b> the previous data.
:param archive: the Archive whose tags and title will be updated
:param id: ID of the Archive to process.
:return: operation result
"""
resp = requests.put(f"{self.server}/api/archives/{id}/metadata", params={
'key': self.key,
'title': archive.title,
'tags': archive.tags
}, headers=self.build_headers())
return JsonUtils.to_obj(resp.text)
def delete_archive(self, id: str) -> dict:
"""
Delete both the archive metadata and the file stored on the server.
:param id: ID of the Archive to process.
:return: operation result
"""
# This function is not implemented on purpose. Just because it is
# too dangerous.
raise NotImplementedError