103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# Copyright 2014-2025 Mike Fährmann
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License version 2 as
|
|
# published by the Free Software Foundation.
|
|
|
|
"""Common classes and constants used by downloader modules."""
|
|
|
|
import os
|
|
from .. import config, util
|
|
_config = config._config
|
|
|
|
|
|
class DownloaderBase():
|
|
"""Base class for downloaders"""
|
|
scheme = ""
|
|
|
|
def __init__(self, job):
|
|
extractor = job.extractor
|
|
self.log = job.get_logger("downloader." + self.scheme)
|
|
|
|
if opts := self._extractor_config(extractor):
|
|
self.opts = opts
|
|
self.config = self.config_opts
|
|
|
|
self.out = job.out
|
|
self.session = extractor.session
|
|
self.part = self.config("part", True)
|
|
self.partdir = self.config("part-directory")
|
|
|
|
if self.partdir:
|
|
if isinstance(self.partdir, dict):
|
|
self.partdir = [
|
|
(util.compile_filter(expr) if expr else util.true,
|
|
util.expand_path(pdir))
|
|
for expr, pdir in self.partdir.items()
|
|
]
|
|
else:
|
|
self.partdir = util.expand_path(self.partdir)
|
|
os.makedirs(self.partdir, exist_ok=True)
|
|
|
|
proxies = self.config("proxy", util.SENTINEL)
|
|
if proxies is util.SENTINEL:
|
|
self.proxies = extractor._proxies
|
|
else:
|
|
self.proxies = util.build_proxy_map(proxies, self.log)
|
|
|
|
def config(self, key, default=None):
|
|
"""Interpolate downloader config value for 'key'"""
|
|
return config.interpolate(("downloader", self.scheme), key, default)
|
|
|
|
def config_opts(self, key, default=None, conf=_config):
|
|
if key in conf:
|
|
return conf[key]
|
|
value = self.opts.get(key, util.SENTINEL)
|
|
if value is not util.SENTINEL:
|
|
return value
|
|
return config.interpolate(("downloader", self.scheme), key, default)
|
|
|
|
def _extractor_config(self, extractor):
|
|
path = extractor._cfgpath
|
|
if not isinstance(path, list):
|
|
return self._extractor_opts(path[1], path[2])
|
|
|
|
opts = {}
|
|
for cat, sub in reversed(path):
|
|
if popts := self._extractor_opts(cat, sub):
|
|
opts.update(popts)
|
|
return opts
|
|
|
|
def _extractor_opts(self, category, subcategory):
|
|
cfg = config.get(("extractor",), category)
|
|
if not cfg:
|
|
return None
|
|
|
|
if copts := cfg.get(self.scheme):
|
|
if subcategory in cfg:
|
|
try:
|
|
if sopts := cfg[subcategory].get(self.scheme):
|
|
opts = copts.copy()
|
|
opts.update(sopts)
|
|
return opts
|
|
except Exception:
|
|
self._report_config_error(subcategory, cfg[subcategory])
|
|
return copts
|
|
|
|
if subcategory in cfg:
|
|
try:
|
|
return cfg[subcategory].get(self.scheme)
|
|
except Exception:
|
|
self._report_config_error(subcategory, cfg[subcategory])
|
|
|
|
return None
|
|
|
|
def _report_config_error(self, subcategory, value):
|
|
config.log.warning("Subcategory '%s' set to '%s' instead of object",
|
|
subcategory, util.json_dumps(value).strip('"'))
|
|
|
|
def download(self, url, pathfmt):
|
|
"""Write data from 'url' into the file specified by 'pathfmt'"""
|