mirrors/pferd
1
0
Fork 0
mirror of https://github.com/Garmelon/PFERD.git synced 2023-12-21 10:23:01 +01:00
Code Issues Packages Projects Releases Wiki Activity

Explain crawling and partially explain downloading

Browse Source
This commit is contained in:
Joscha 2021-05-22 20:39:57 +00:00
parent 9cb2b68f09
commit 62f0f7bfc5
2 changed files with 22 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
PFERD/crawler.py
View File

@ -220,9 +220,14 @@ class Crawler(ABC):
)
async def crawl(self, path: PurePath) -> Optional[CrawlToken]:
log.explain_topic(f"Decision: Crawl {path}")
if self._transformer.transform(path) is None:
log.explain("Answer: No")
return None
log.explain("Answer: Yes")
desc = f"[bold bright_cyan]Crawling[/] {escape(str(path))}"
return CrawlToken(self._limiter, desc)
@ -233,26 +238,32 @@ class Crawler(ABC):
redownload: Optional[Redownload] = None,
on_conflict: Optional[OnConflict] = None,
) -> Optional[DownloadToken]:
log.explain_topic(f"Decision: Download {path}")
transformed_path = self._transformer.transform(path)
if transformed_path is None:
log.explain("Answer: No")
return None
fs_token = await self._output_dir.download(transformed_path, mtime, redownload, on_conflict)
if fs_token is None:
log.explain("Answer: No")
return None
log.explain("Answer: Yes")
desc = f"[bold bright_cyan]Downloading[/] {escape(str(path))}"
return DownloadToken(self._limiter, fs_token, desc)
async def _cleanup(self) -> None:
log.explain_topic("Decision: Clean up files?")
log.explain_topic("Decision: Clean up files")
if self.error_free:
log.explain("No warnings or errors occurred during this run")
log.explain("Cleaning up files")
log.explain("Answer: Yes")
await self._output_dir.cleanup()
else:
log.explain("Warnings or errors occurred during this run")
log.explain("Not cleaning up files")
log.explain("Answer: No")
async def run(self) -> None:
"""

10
PFERD/transformer.py
View File

@ -317,16 +317,22 @@ class Transformer:
for i, line in enumerate(rules.split("\n")):
line = line.strip()
if line:
self._rules.append(parse_rule(Line(line, i)))
rule = parse_rule(Line(line, i))
self._rules.append((line, rule))
def transform(self, path: PurePath) -> Optional[PurePath]:
for rule in self._rules:
for i, (line, rule) in enumerate(self._rules):
log.explain(f"Testing rule {i}: {line}")
result = rule.transform(path)
if isinstance(result, PurePath):
log.explain(f"Match! Transformed to {result}")
return result
elif result: # Exclamation mark
log.explain("Match! Ignored")
return None
else:
continue
log.explain("No rule matched, path is unchanged")
return path