#!/usr/bin/env python3 # -*- encoding: utf-8 -*- from pathlib import Path # from collections import defaultdict from typing import Optional, Tuple import gallery_dl import gallery_dl.extractor import gallery_dl.extractor.common import gallery_dl.job import gallery_dl.util from .downloader.cache import get_path_for_caching def find_extractor(link) -> Optional[gallery_dl.extractor.common.Extractor]: extractor: gallery_dl.extractor.common.Extractor = None try: extractor = gallery_dl.extractor.find(link) except gallery_dl.exception.NotFoundError: pass return extractor def find_extractor_category_subcategory(link) -> Optional[Tuple[str, str]]: extractor: gallery_dl.extractor.common.Extractor = find_extractor(link) if extractor is None: return None else: return (type(extractor).category, type(extractor).subcategory) def find_extractor_archival_key(link) -> Optional[str]: extractor: gallery_dl.extractor.common.Extractor = find_extractor(link) if extractor is None: return None else: return str(get_path_for_caching(link, Path(''))) # job = gallery_dl.job.SimulationJob(extractor) # job.archive = FakeArchive('', extractor) # job.run() # uses the network # return next(iter(job.archive.captured_keys), None) class FakeArchive(gallery_dl.util.DownloadArchive): def __init__(self, path, extractor): super().__init__(':memory:', extractor) self.captured_keys = list() def check(self, kwdict): return super().check(kwdict) def add(self, kwdict): self.captured_keys.append(kwdict.get( "_archive_key") or self.keygen(kwdict)) return super().add(kwdict)