Mirrors/ArchiveBox
2
0
Fork 0
mirror of https://github.com/ArchiveBox/ArchiveBox synced 2024-11-29 23:50:22 +00:00
Code Issues Projects Releases Packages Wiki Activity

better type checking of latest output methods

Browse source
This commit is contained in:
Nick Sweeting 2019-03-27 18:25:17 -04:00
parent d2a34f2602
commit 8b50fee0f5
1 changed files with 14 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
archivebox/schema.py
View file

@ -14,12 +14,14 @@ class ArchiveError(Exception):
LinkDict = Dict[str, Any] LinkDict = Dict[str, Any]
ArchiveOutput = Union[str, Exception, None]
@dataclass(frozen=True) @dataclass(frozen=True)
class ArchiveResult: class ArchiveResult:
cmd: List[str] cmd: List[str]
pwd: Optional[str] pwd: Optional[str]
cmd_version: Optional[str] cmd_version: Optional[str]
output: Union[str, Exception, None] output: ArchiveOutput
status: str status: str
start_ts: datetime start_ts: datetime
end_ts: datetime end_ts: datetime
@ -211,31 +213,26 @@ class Link:
domain(self.url), domain(self.url),
)) ))
def latest_outputs(self, status: str=None) -> Dict[str, Optional[str]]: def latest_outputs(self, status: str=None) -> Dict[str, ArchiveOutput]:
"""get the latest output that each archive method produced for link""" """get the latest output that each archive method produced for link"""
latest = { ARCHIVE_METHODS = (
'title': None, 'title', 'favicon', 'wget', 'warc', 'pdf',
'favicon': None, 'screenshot', 'dom', 'git', 'media', 'archive_org',
'wget': None, )
'warc': None, latest: Dict[str, ArchiveOutput] = {}
'pdf': None, for archive_method in ARCHIVE_METHODS:
'screenshot': None,
'dom': None,
'git': None,
'media': None,
'archive_org': None,
}
for archive_method in latest.keys():
# get most recent succesful result in history for each archive method # get most recent succesful result in history for each archive method
history = self.history.get(archive_method) or [] history = self.history.get(archive_method) or []
history = filter(lambda result: result.output, reversed(history)) history = list(filter(lambda result: result.output, reversed(history)))
if status is not None: if status is not None:
history = filter(lambda result: result.status == status, history) history = list(filter(lambda result: result.status == status, history))
history = list(history) history = list(history)
if history: if history:
latest[archive_method] = history[0].output latest[archive_method] = history[0].output
else:
latest[archive_method] = None
return latest return latest