Upstream mikf/gallery-dl state of the touched files before 8462417350
This commit is contained in:
upstream committed
commit
b5186d86fa
2 files changed
+1025
No files matched your search
@@ -0,0 +1,575 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2014-2026 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.
|
||||
|
||||
"""Downloader module for http:// and https:// URLs"""
|
||||
|
||||
import time
|
||||
import mimetypes
|
||||
from requests.exceptions import RequestException, ConnectionError, Timeout
|
||||
from .common import DownloaderBase
|
||||
from .. import text, util, output, exception
|
||||
from ssl import SSLError
|
||||
FLAGS = util.FLAGS
|
||||
|
||||
|
||||
class HttpDownloader(DownloaderBase):
|
||||
scheme = "http"
|
||||
|
||||
def __init__(self, job):
|
||||
DownloaderBase.__init__(self, job)
|
||||
extractor = job.extractor
|
||||
self.downloading = False
|
||||
|
||||
self.adjust_extension = self.config("adjust-extensions", True)
|
||||
self.chunk_size = self.config("chunk-size", 32768)
|
||||
self.metadata = extractor.config("http-metadata")
|
||||
self.progress = self.config("progress", 3.0)
|
||||
self.validate = self.config("validate", True)
|
||||
self.validate_html = self.config("validate-html", True)
|
||||
self.headers = self.config("headers")
|
||||
self.minsize = self.config("filesize-min")
|
||||
self.maxsize = self.config("filesize-max")
|
||||
self.retries = self.config("retries", extractor._retries)
|
||||
self.retry_codes = self.config("retry-codes", extractor._retry_codes)
|
||||
self.timeout = self.config("timeout", extractor._timeout)
|
||||
self.verify = self.config("verify", extractor._verify)
|
||||
self.mtime = self.config("mtime", True)
|
||||
self.rate = self.config("rate")
|
||||
interval_429 = self.config("sleep-429")
|
||||
|
||||
if self.config("consume-content", False):
|
||||
self.release_conn = self._release_conn_impl
|
||||
else:
|
||||
# this resets the underlying TCP connection, and therefore
|
||||
# if the program makes another request to the same domain,
|
||||
# a new connection (either TLS or plain TCP) must be made
|
||||
self.release_conn = lambda resp: resp.close()
|
||||
|
||||
if self.retries < 0:
|
||||
self.retries = float("inf")
|
||||
if self.minsize:
|
||||
minsize = text.parse_bytes(self.minsize)
|
||||
if not minsize:
|
||||
self.log.warning(
|
||||
"Invalid minimum file size (%r)", self.minsize)
|
||||
self.minsize = minsize
|
||||
if self.maxsize:
|
||||
maxsize = text.parse_bytes(self.maxsize)
|
||||
if not maxsize:
|
||||
self.log.warning(
|
||||
"Invalid maximum file size (%r)", self.maxsize)
|
||||
self.maxsize = maxsize
|
||||
if isinstance(self.chunk_size, str):
|
||||
chunk_size = text.parse_bytes(self.chunk_size)
|
||||
if not chunk_size:
|
||||
self.log.warning(
|
||||
"Invalid chunk size (%r)", self.chunk_size)
|
||||
chunk_size = 32768
|
||||
self.chunk_size = chunk_size
|
||||
if self.rate:
|
||||
func = util.build_selection_func(self.rate, 0, text.parse_bytes)
|
||||
if rmax := func.args[1] if hasattr(func, "args") else func():
|
||||
if rmax < self.chunk_size:
|
||||
# reduce chunk_size to allow for one iteration each second
|
||||
self.chunk_size = rmax
|
||||
self.rate = func
|
||||
self.receive = self._receive_rate
|
||||
else:
|
||||
self.log.warning("Invalid rate limit (%r)", self.rate)
|
||||
self.rate = False
|
||||
if self.progress is not None:
|
||||
self.receive = self._receive_rate
|
||||
if self.progress < 0.0:
|
||||
self.progress = 0.0
|
||||
if interval_429 is None:
|
||||
self.interval_429 = extractor._interval_429
|
||||
else:
|
||||
try:
|
||||
self.interval_429 = util.build_duration_func_ex(interval_429)
|
||||
except Exception as exc:
|
||||
self.log.error("Invalid 'sleep-429' value '%s' (%s: %s)",
|
||||
interval_429, exc.__class__.__name__, exc)
|
||||
self.interval_429 = extractor._interval_429
|
||||
|
||||
def download(self, url, pathfmt):
|
||||
try:
|
||||
return self._download_impl(url, pathfmt)
|
||||
except Exception as exc:
|
||||
if self.downloading:
|
||||
output.stderr_write("\n")
|
||||
self.log.traceback(exc)
|
||||
raise
|
||||
finally:
|
||||
# remove file from incomplete downloads
|
||||
if self.downloading and not self.part:
|
||||
util.remove_file(pathfmt.temppath)
|
||||
|
||||
def _download_impl(self, url, pathfmt):
|
||||
response = None
|
||||
tries = code = 0
|
||||
msg = ""
|
||||
|
||||
metadata = self.metadata
|
||||
kwdict = pathfmt.kwdict
|
||||
expected_status = kwdict.get(
|
||||
"_http_expected_status", ())
|
||||
adjust_extension = kwdict.get(
|
||||
"_http_adjust_extension", self.adjust_extension)
|
||||
|
||||
if self.part and not metadata:
|
||||
pathfmt.part_enable(self.partdir)
|
||||
|
||||
while True:
|
||||
if FLAGS.DOWNLOAD is not None:
|
||||
return FLAGS.process("DOWNLOAD")
|
||||
|
||||
if tries:
|
||||
if response:
|
||||
self.release_conn(response)
|
||||
response = None
|
||||
|
||||
self.log.warning("%s (%s/%s)", msg, tries, self.retries+1)
|
||||
if tries > self.retries:
|
||||
return False
|
||||
|
||||
if code == 429 and self.interval_429:
|
||||
s = self.interval_429(tries)
|
||||
time.sleep(s if s > tries else tries)
|
||||
else:
|
||||
time.sleep(tries)
|
||||
code = 0
|
||||
|
||||
tries += 1
|
||||
file_header = None
|
||||
|
||||
# collect HTTP headers
|
||||
headers = {"Accept": "*/*"}
|
||||
# file-specific headers
|
||||
if extra := kwdict.get("_http_headers"):
|
||||
headers.update(extra)
|
||||
# general headers
|
||||
if self.headers:
|
||||
headers.update(self.headers)
|
||||
# partial content
|
||||
if file_size := pathfmt.part_size():
|
||||
headers["Range"] = f"bytes={file_size}-"
|
||||
|
||||
# connect to (remote) source
|
||||
try:
|
||||
response = self.session.request(
|
||||
kwdict.get("_http_method", "GET"), url,
|
||||
stream=True,
|
||||
headers=headers,
|
||||
data=kwdict.get("_http_data"),
|
||||
timeout=self.timeout,
|
||||
proxies=self.proxies,
|
||||
verify=self.verify,
|
||||
)
|
||||
except ConnectionError as exc:
|
||||
try:
|
||||
reason = exc.args[0].reason
|
||||
cls = reason.__class__.__name__
|
||||
pre, _, err = str(reason.args[-1]).partition(":")
|
||||
msg = f"{cls}: {(err or pre).lstrip()}"
|
||||
except Exception:
|
||||
msg = str(exc)
|
||||
continue
|
||||
except Timeout as exc:
|
||||
msg = str(exc)
|
||||
continue
|
||||
except Exception as exc:
|
||||
self.log.warning(exc)
|
||||
return False
|
||||
|
||||
# check response
|
||||
code = response.status_code
|
||||
if code == 200 or code in expected_status: # OK
|
||||
offset = 0
|
||||
size = response.headers.get("Content-Length")
|
||||
elif code == 206: # Partial Content
|
||||
offset = file_size
|
||||
size = response.headers["Content-Range"].rpartition("/")[2]
|
||||
elif code == 416 and file_size: # Range Not Satisfiable
|
||||
self._release_conn_impl(response)
|
||||
break
|
||||
else:
|
||||
msg = f"'{code} {response.reason}' for '{url}'"
|
||||
|
||||
challenge = util.detect_challenge(response)
|
||||
if challenge is not None:
|
||||
self.log.warning(challenge)
|
||||
|
||||
if code in self.retry_codes or 500 <= code < 600:
|
||||
continue
|
||||
retry = kwdict.get("_http_retry")
|
||||
if retry and retry(response):
|
||||
continue
|
||||
self.release_conn(response)
|
||||
self.log.warning(msg)
|
||||
return False
|
||||
|
||||
# check for invalid responses
|
||||
if self.validate and \
|
||||
(validate := kwdict.get("_http_validate")) is not None:
|
||||
try:
|
||||
result = validate(response)
|
||||
except Exception:
|
||||
self.release_conn(response)
|
||||
raise
|
||||
if isinstance(result, str):
|
||||
url = result
|
||||
tries -= 1
|
||||
continue
|
||||
if not result:
|
||||
self.release_conn(response)
|
||||
self.log.warning("Invalid response")
|
||||
return False
|
||||
if self.validate_html and response.headers.get(
|
||||
"content-type", "").startswith("text/html") and \
|
||||
pathfmt.extension not in ("html", "htm"):
|
||||
if response.history:
|
||||
self.log.warning("HTTP redirect to '%s'", response.url)
|
||||
else:
|
||||
self.log.warning("HTML response")
|
||||
return False
|
||||
|
||||
# check file size
|
||||
size = text.parse_int(size, None)
|
||||
if size is not None:
|
||||
if not size:
|
||||
self.release_conn(response)
|
||||
self.log.warning("Empty file")
|
||||
return False
|
||||
if self.minsize and size < self.minsize:
|
||||
self.release_conn(response)
|
||||
self.log.warning(
|
||||
"File size smaller than allowed minimum (%s < %s)",
|
||||
size, self.minsize)
|
||||
pathfmt.temppath = ""
|
||||
return True
|
||||
if self.maxsize and size > self.maxsize:
|
||||
self.release_conn(response)
|
||||
self.log.warning(
|
||||
"File size larger than allowed maximum (%s > %s)",
|
||||
size, self.maxsize)
|
||||
pathfmt.temppath = ""
|
||||
return True
|
||||
|
||||
build_path = False
|
||||
|
||||
# set missing filename extension from MIME type
|
||||
if not pathfmt.extension:
|
||||
pathfmt.set_extension(self._find_extension(response))
|
||||
build_path = True
|
||||
|
||||
# set metadata from HTTP headers
|
||||
if metadata:
|
||||
kwdict[metadata] = util.extract_headers(response)
|
||||
build_path = True
|
||||
|
||||
# build and check file path
|
||||
if build_path:
|
||||
pathfmt.build_path()
|
||||
if pathfmt.exists():
|
||||
pathfmt.temppath = ""
|
||||
# release the connection back to pool by explicitly
|
||||
# calling .close()
|
||||
# see https://requests.readthedocs.io/en/latest/user
|
||||
# /advanced/#body-content-workflow
|
||||
# when the image size is on the order of megabytes,
|
||||
# re-establishing a TLS connection will typically be faster
|
||||
# than consuming the whole response
|
||||
response.close()
|
||||
return True
|
||||
if self.part and metadata:
|
||||
pathfmt.part_enable(self.partdir)
|
||||
metadata = False
|
||||
|
||||
content = response.iter_content(self.chunk_size)
|
||||
|
||||
validate_sig = kwdict.get("_http_signature")
|
||||
validate_ext = (adjust_extension and
|
||||
pathfmt.extension in SIGNATURE_CHECKS)
|
||||
|
||||
# check filename extension against file header
|
||||
if not offset and (validate_ext or validate_sig):
|
||||
try:
|
||||
file_header = next(
|
||||
content if response.raw.chunked
|
||||
else response.iter_content(16), b"")
|
||||
except (RequestException, SSLError) as exc:
|
||||
msg = str(exc)
|
||||
continue
|
||||
if validate_sig:
|
||||
result = validate_sig(file_header)
|
||||
if result is not True:
|
||||
self.release_conn(response)
|
||||
self.log.warning(
|
||||
result or "Invalid file signature bytes")
|
||||
return False
|
||||
if validate_ext and self._adjust_extension(
|
||||
pathfmt, file_header) and pathfmt.exists():
|
||||
pathfmt.temppath = ""
|
||||
response.close()
|
||||
return True
|
||||
|
||||
# set open mode
|
||||
if not offset:
|
||||
mode = "w+b"
|
||||
if file_size:
|
||||
self.log.debug("Unable to resume partial download")
|
||||
else:
|
||||
mode = "r+b"
|
||||
self.log.debug("Resuming download at byte %d", offset)
|
||||
|
||||
# download content
|
||||
self.downloading = True
|
||||
with pathfmt.open(mode) as fp:
|
||||
if fp is None:
|
||||
# '.part' file no longer exists
|
||||
break
|
||||
if file_header:
|
||||
fp.write(file_header)
|
||||
offset += len(file_header)
|
||||
elif offset:
|
||||
if adjust_extension and \
|
||||
pathfmt.extension in SIGNATURE_CHECKS:
|
||||
self._adjust_extension(pathfmt, fp.read(16))
|
||||
fp.seek(offset)
|
||||
|
||||
self.out.start(pathfmt.path)
|
||||
try:
|
||||
self.receive(fp, content, size, offset)
|
||||
except (RequestException, SSLError) as exc:
|
||||
msg = str(exc)
|
||||
output.stderr_write("\n")
|
||||
continue
|
||||
except exception.StopExtraction:
|
||||
response.close()
|
||||
return False
|
||||
except exception.ControlException:
|
||||
response.close()
|
||||
raise
|
||||
|
||||
# check file size
|
||||
if size and (fsize := fp.tell()) < size:
|
||||
if (segmented := kwdict.get("_http_segmented")) and \
|
||||
segmented is True or segmented == fsize:
|
||||
tries -= 1
|
||||
msg = "Resuming segmented download"
|
||||
output.stdout_write("\r")
|
||||
else:
|
||||
msg = f"file size mismatch ({fsize} < {size})"
|
||||
output.stderr_write("\n")
|
||||
continue
|
||||
|
||||
break
|
||||
|
||||
self.downloading = False
|
||||
if self.mtime:
|
||||
if "_http_lastmodified" in kwdict:
|
||||
kwdict["_mtime_http"] = kwdict["_http_lastmodified"]
|
||||
else:
|
||||
kwdict["_mtime_http"] = response.headers.get("Last-Modified")
|
||||
else:
|
||||
kwdict["_mtime_http"] = None
|
||||
|
||||
return True
|
||||
|
||||
def _release_conn_impl(self, response):
|
||||
"""Release connection back to pool by consuming response body"""
|
||||
try:
|
||||
for _ in response.iter_content(self.chunk_size):
|
||||
pass
|
||||
except (RequestException, SSLError) as exc:
|
||||
output.stderr_write("\n")
|
||||
self.log.debug(
|
||||
"Unable to consume response body (%s: %s); "
|
||||
"closing the connection anyway", exc.__class__.__name__, exc)
|
||||
response.close()
|
||||
|
||||
def receive(self, fp, content, bytes_total, bytes_start):
|
||||
write = fp.write
|
||||
for data in content:
|
||||
if FLAGS.DOWNLOAD is not None:
|
||||
return FLAGS.process("DOWNLOAD")
|
||||
write(data)
|
||||
|
||||
def _receive_rate(self, fp, content, bytes_total, bytes_start):
|
||||
rate = self.rate() if self.rate else None
|
||||
write = fp.write
|
||||
progress = self.progress
|
||||
|
||||
bytes_downloaded = 0
|
||||
time_start = time.monotonic()
|
||||
|
||||
for data in content:
|
||||
if FLAGS.DOWNLOAD is not None:
|
||||
return FLAGS.process("DOWNLOAD")
|
||||
time_elapsed = time.monotonic() - time_start
|
||||
bytes_downloaded += len(data)
|
||||
|
||||
write(data)
|
||||
|
||||
if progress is not None:
|
||||
if time_elapsed > progress:
|
||||
self.out.progress(
|
||||
bytes_total,
|
||||
bytes_start + bytes_downloaded,
|
||||
int(bytes_downloaded / time_elapsed),
|
||||
)
|
||||
|
||||
if rate is not None:
|
||||
time_expected = bytes_downloaded / rate
|
||||
if time_expected > time_elapsed:
|
||||
time.sleep(time_expected - time_elapsed)
|
||||
|
||||
def _find_extension(self, response):
|
||||
"""Get filename extension from MIME type"""
|
||||
mtype = response.headers.get("Content-Type", "image/jpeg")
|
||||
mtype = mtype.partition(";")[0].lower()
|
||||
|
||||
if "/" not in mtype:
|
||||
mtype = "image/" + mtype
|
||||
|
||||
if mtype in MIME_TYPES:
|
||||
return MIME_TYPES[mtype]
|
||||
|
||||
if ext := mimetypes.guess_extension(mtype, strict=False):
|
||||
return ext[1:]
|
||||
|
||||
self.log.warning("Unknown MIME type '%s'", mtype)
|
||||
return "bin"
|
||||
|
||||
def _adjust_extension(self, pathfmt, file_header):
|
||||
"""Check filename extension against file header"""
|
||||
if not SIGNATURE_CHECKS[pathfmt.extension](file_header):
|
||||
for ext, check in SIGNATURE_CHECKS.items():
|
||||
if check(file_header):
|
||||
self.log.debug(
|
||||
"Adjusting filename extension of '%s' to '%s'",
|
||||
pathfmt.filename, ext)
|
||||
pathfmt.set_extension(ext)
|
||||
pathfmt.build_path()
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
MIME_TYPES = {
|
||||
"image/jpeg" : "jpg",
|
||||
"image/jpg" : "jpg",
|
||||
"image/png" : "png",
|
||||
"image/gif" : "gif",
|
||||
"image/bmp" : "bmp",
|
||||
"image/x-bmp" : "bmp",
|
||||
"image/x-ms-bmp": "bmp",
|
||||
"image/webp" : "webp",
|
||||
"image/avif" : "avif",
|
||||
"image/heic" : "heic",
|
||||
"image/heif" : "heif",
|
||||
"image/svg+xml" : "svg",
|
||||
"image/ico" : "ico",
|
||||
"image/icon" : "ico",
|
||||
"image/x-icon" : "ico",
|
||||
"image/vnd.microsoft.icon" : "ico",
|
||||
"image/x-photoshop" : "psd",
|
||||
"application/x-photoshop" : "psd",
|
||||
"image/vnd.adobe.photoshop": "psd",
|
||||
|
||||
"video/webm": "webm",
|
||||
"video/ogg" : "ogg",
|
||||
"video/mp4" : "mp4",
|
||||
"video/m4v" : "m4v",
|
||||
"video/x-m4v": "m4v",
|
||||
"video/quicktime": "mov",
|
||||
|
||||
"audio/wav" : "wav",
|
||||
"audio/x-wav": "wav",
|
||||
"audio/webm" : "webm",
|
||||
"audio/ogg" : "ogg",
|
||||
"audio/mpeg" : "mp3",
|
||||
"audio/aac" : "aac",
|
||||
"audio/x-aac": "aac",
|
||||
|
||||
"application/vnd.apple.mpegurl": "m3u8",
|
||||
"application/x-mpegurl" : "m3u8",
|
||||
"application/dash+xml" : "mpd",
|
||||
|
||||
"application/zip" : "zip",
|
||||
"application/x-zip": "zip",
|
||||
"application/x-zip-compressed": "zip",
|
||||
"application/rar" : "rar",
|
||||
"application/x-rar": "rar",
|
||||
"application/x-rar-compressed": "rar",
|
||||
"application/x-7z-compressed" : "7z",
|
||||
|
||||
"application/pdf" : "pdf",
|
||||
"application/x-pdf": "pdf",
|
||||
"application/x-shockwave-flash": "swf",
|
||||
|
||||
"text/html": "html",
|
||||
|
||||
"application/ogg": "ogg",
|
||||
# https://www.iana.org/assignments/media-types/model/obj
|
||||
"model/obj": "obj",
|
||||
"application/octet-stream": "bin",
|
||||
}
|
||||
|
||||
|
||||
def _signature_html(s):
|
||||
s = s[:14].lstrip()
|
||||
return s and b"<!doctype html".startswith(s.lower())
|
||||
|
||||
|
||||
# https://en.wikipedia.org/wiki/List_of_file_signatures
|
||||
SIGNATURE_CHECKS = {
|
||||
"jpg" : lambda s: s[0:3] == b"\xFF\xD8\xFF",
|
||||
"png" : lambda s: s[0:8] == b"\x89PNG\r\n\x1A\n",
|
||||
"gif" : lambda s: s[0:6] in (b"GIF87a", b"GIF89a"),
|
||||
"bmp" : lambda s: s[0:2] == b"BM",
|
||||
"webp": lambda s: (s[0:4] == b"RIFF" and
|
||||
s[8:12] == b"WEBP"),
|
||||
"avif": lambda s: s[4:11] == b"ftypavi" and s[11] in b"fs",
|
||||
"heic": lambda s: (s[4:10] == b"ftyphe" and s[10:12] in (
|
||||
b"ic", b"im", b"is", b"ix", b"vc", b"vm", b"vs")),
|
||||
"svg" : lambda s: s[0:5] == b"<?xml",
|
||||
"ico" : lambda s: s[0:4] == b"\x00\x00\x01\x00",
|
||||
"cur" : lambda s: s[0:4] == b"\x00\x00\x02\x00",
|
||||
"psd" : lambda s: s[0:4] == b"8BPS",
|
||||
"mp4" : lambda s: (s[4:8] == b"ftyp" and s[8:11] in (
|
||||
b"mp4", b"avc", b"iso")),
|
||||
"m4v" : lambda s: s[4:11] == b"ftypM4V",
|
||||
"mov" : lambda s: s[4:12] == b"ftypqt ",
|
||||
"webm": lambda s: s[0:4] == b"\x1A\x45\xDF\xA3",
|
||||
"ogg" : lambda s: s[0:4] == b"OggS",
|
||||
"wav" : lambda s: (s[0:4] == b"RIFF" and
|
||||
s[8:12] == b"WAVE"),
|
||||
"mp3" : lambda s: (s[0:3] == b"ID3" or
|
||||
s[0:2] in (b"\xFF\xFB", b"\xFF\xF3", b"\xFF\xF2")),
|
||||
"aac" : lambda s: s[0:2] in (b"\xFF\xF9", b"\xFF\xF1"),
|
||||
"m3u8": lambda s: s[0:7] == b"#EXTM3U",
|
||||
"mpd" : lambda s: b"<MPD" in s,
|
||||
"zip" : lambda s: s[0:4] in (b"PK\x03\x04", b"PK\x05\x06", b"PK\x07\x08"),
|
||||
"rar" : lambda s: s[0:6] == b"Rar!\x1A\x07",
|
||||
"7z" : lambda s: s[0:6] == b"\x37\x7A\xBC\xAF\x27\x1C",
|
||||
"pdf" : lambda s: s[0:5] == b"%PDF-",
|
||||
"swf" : lambda s: s[0:3] in (b"CWS", b"FWS"),
|
||||
"html": _signature_html,
|
||||
"htm" : _signature_html,
|
||||
"blend": lambda s: s[0:7] == b"BLENDER",
|
||||
# unfortunately the Wavefront .obj format doesn't have a signature,
|
||||
# so we check for the existence of Blender's comment
|
||||
"obj" : lambda s: s[0:11] == b"# Blender v",
|
||||
# Celsys Clip Studio Paint format
|
||||
# https://github.com/rasensuihei/cliputils/blob/master/README.md
|
||||
"clip": lambda s: s[0:8] == b"CSFCHUNK",
|
||||
# check 'bin' files against all other file signatures
|
||||
"bin" : lambda s: False,
|
||||
}
|
||||
|
||||
__downloader__ = HttpDownloader
|
||||
@@ -0,0 +1,450 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2018-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.
|
||||
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import Mock, MagicMock, patch
|
||||
|
||||
import re
|
||||
import logging
|
||||
import os.path
|
||||
import binascii
|
||||
import tempfile
|
||||
import threading
|
||||
import http.server
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
from gallery_dl import downloader, extractor, output, config, path # noqa E402
|
||||
from gallery_dl.downloader.http import MIME_TYPES, SIGNATURE_CHECKS # noqa E402
|
||||
|
||||
|
||||
class MockDownloaderModule(Mock):
|
||||
__downloader__ = "mock"
|
||||
|
||||
|
||||
class FakeJob():
|
||||
|
||||
def __init__(self):
|
||||
self.extractor = extractor.find("generic:https://example.org/")
|
||||
self.extractor.initialize()
|
||||
self.pathfmt = path.PathFormat(self.extractor)
|
||||
self.out = output.NullOutput()
|
||||
self.get_logger = logging.getLogger
|
||||
|
||||
|
||||
class TestDownloaderModule(unittest.TestCase):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
# allow import of ytdl downloader module without youtube_dl installed
|
||||
cls._orig_ytdl = sys.modules.get("youtube_dl")
|
||||
sys.modules["youtube_dl"] = MagicMock()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
if cls._orig_ytdl:
|
||||
sys.modules["youtube_dl"] = cls._orig_ytdl
|
||||
else:
|
||||
del sys.modules["youtube_dl"]
|
||||
|
||||
def setUp(self):
|
||||
downloader._cache.clear()
|
||||
|
||||
def tearDown(self):
|
||||
downloader._cache.clear()
|
||||
|
||||
def test_find(self):
|
||||
cls = downloader.find("http")
|
||||
self.assertEqual(cls.__name__, "HttpDownloader")
|
||||
self.assertEqual(cls.scheme , "http")
|
||||
|
||||
cls = downloader.find("https")
|
||||
self.assertEqual(cls.__name__, "HttpDownloader")
|
||||
self.assertEqual(cls.scheme , "http")
|
||||
|
||||
cls = downloader.find("text")
|
||||
self.assertEqual(cls.__name__, "TextDownloader")
|
||||
self.assertEqual(cls.scheme , "text")
|
||||
|
||||
cls = downloader.find("ytdl")
|
||||
self.assertEqual(cls.__name__, "YoutubeDLDownloader")
|
||||
self.assertEqual(cls.scheme , "ytdl")
|
||||
|
||||
self.assertEqual(downloader.find("ftp"), None)
|
||||
self.assertEqual(downloader.find("foo"), None)
|
||||
self.assertEqual(downloader.find(1234) , None)
|
||||
self.assertEqual(downloader.find(None) , None)
|
||||
|
||||
@patch("builtins.__import__")
|
||||
def test_cache(self, import_module):
|
||||
import_module.return_value = MockDownloaderModule()
|
||||
downloader.find("http")
|
||||
downloader.find("text")
|
||||
downloader.find("ytdl")
|
||||
self.assertEqual(import_module.call_count, 3)
|
||||
downloader.find("http")
|
||||
downloader.find("text")
|
||||
downloader.find("ytdl")
|
||||
self.assertEqual(import_module.call_count, 3)
|
||||
|
||||
@patch("builtins.__import__")
|
||||
def test_cache_http(self, import_module):
|
||||
import_module.return_value = MockDownloaderModule()
|
||||
downloader.find("http")
|
||||
downloader.find("https")
|
||||
self.assertEqual(import_module.call_count, 1)
|
||||
|
||||
@patch("builtins.__import__")
|
||||
def test_cache_https(self, import_module):
|
||||
import_module.return_value = MockDownloaderModule()
|
||||
downloader.find("https")
|
||||
downloader.find("http")
|
||||
self.assertEqual(import_module.call_count, 1)
|
||||
|
||||
|
||||
class TestDownloaderConfig(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
config.clear()
|
||||
|
||||
def tearDown(self):
|
||||
config.clear()
|
||||
|
||||
def test_default_http(self):
|
||||
job = FakeJob()
|
||||
extr = job.extractor
|
||||
dl = downloader.find("http")(job)
|
||||
|
||||
self.assertEqual(dl.adjust_extension, True)
|
||||
self.assertEqual(dl.chunk_size, 32768)
|
||||
self.assertEqual(dl.metadata, None)
|
||||
self.assertEqual(dl.progress, 3.0)
|
||||
self.assertEqual(dl.validate, True)
|
||||
self.assertEqual(dl.headers, None)
|
||||
self.assertEqual(dl.minsize, None)
|
||||
self.assertEqual(dl.maxsize, None)
|
||||
self.assertEqual(dl.mtime, True)
|
||||
self.assertEqual(dl.rate, None)
|
||||
self.assertEqual(dl.part, True)
|
||||
self.assertEqual(dl.partdir, None)
|
||||
|
||||
self.assertIs(dl.interval_429, extr._interval_429)
|
||||
self.assertIs(dl.retry_codes, extr._retry_codes)
|
||||
self.assertIs(dl.retries, extr._retries)
|
||||
self.assertIs(dl.timeout, extr._timeout)
|
||||
self.assertIs(dl.proxies, extr._proxies)
|
||||
self.assertIs(dl.verify, extr._verify)
|
||||
|
||||
def test_config_http(self):
|
||||
config.set((), "rate", 42)
|
||||
config.set((), "mtime", False)
|
||||
config.set((), "headers", {"foo": "bar"})
|
||||
config.set(("downloader",), "retries", -1)
|
||||
config.set(("downloader", "http"), "filesize-min", "10k")
|
||||
config.set(("extractor", "generic"), "verify", False)
|
||||
config.set(("extractor", "generic", "example.org"), "timeout", 10)
|
||||
config.set(("extractor", "generic", "http"), "part", False)
|
||||
config.set(
|
||||
("extractor", "generic", "example.org", "http"), "headers", {})
|
||||
|
||||
job = FakeJob()
|
||||
dl = downloader.find("http")(job)
|
||||
|
||||
self.assertEqual(dl.headers, {"foo": "bar"})
|
||||
self.assertEqual(dl.minsize, 10240)
|
||||
self.assertEqual(dl.retries, float("inf"))
|
||||
self.assertEqual(dl.timeout, 10)
|
||||
self.assertEqual(dl.verify, False)
|
||||
self.assertEqual(dl.mtime, False)
|
||||
self.assertEqual(dl.rate(), 42)
|
||||
self.assertEqual(dl.part, False)
|
||||
|
||||
|
||||
class TestDownloaderBase(unittest.TestCase):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.dir = tempfile.TemporaryDirectory()
|
||||
cls.fnum = 0
|
||||
config.set((), "base-directory", cls.dir.name)
|
||||
cls.job = FakeJob()
|
||||
|
||||
@classmethod
|
||||
def tearDownClass(cls):
|
||||
cls.dir.cleanup()
|
||||
config.clear()
|
||||
|
||||
@classmethod
|
||||
def _prepare_destination(cls, content=None, part=True, extension=None):
|
||||
name = f"file-{cls.fnum}"
|
||||
cls.fnum += 1
|
||||
|
||||
kwdict = {
|
||||
"category" : "test",
|
||||
"subcategory": "test",
|
||||
"filename" : name,
|
||||
"extension" : extension,
|
||||
}
|
||||
|
||||
pathfmt = cls.job.pathfmt
|
||||
pathfmt.set_directory(kwdict)
|
||||
pathfmt.set_filename(kwdict)
|
||||
pathfmt.build_path()
|
||||
|
||||
if content:
|
||||
mode = "wb" if isinstance(content, bytes) else "w"
|
||||
with pathfmt.open(mode) as fp:
|
||||
fp.write(content)
|
||||
|
||||
return pathfmt
|
||||
|
||||
def _run_test(self, url, input, output,
|
||||
extension, expected_extension=None):
|
||||
pathfmt = self._prepare_destination(input, extension=extension)
|
||||
success = self.downloader.download(url, pathfmt)
|
||||
|
||||
# test successful download
|
||||
self.assertTrue(success, f"downloading '{url}' failed")
|
||||
|
||||
# test content
|
||||
mode = "rb" if isinstance(output, bytes) else "r"
|
||||
with pathfmt.open(mode) as fp:
|
||||
content = fp.read()
|
||||
self.assertEqual(content, output)
|
||||
|
||||
# test filename extension
|
||||
self.assertEqual(
|
||||
pathfmt.extension,
|
||||
expected_extension,
|
||||
content[0:16],
|
||||
)
|
||||
self.assertEqual(
|
||||
os.path.splitext(pathfmt.realpath)[1][1:],
|
||||
expected_extension,
|
||||
)
|
||||
|
||||
|
||||
class TestHTTPDownloader(TestDownloaderBase):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
TestDownloaderBase.setUpClass()
|
||||
cls.downloader = downloader.find("http")(cls.job)
|
||||
|
||||
host = "127.0.0.1"
|
||||
port = 0 # select random not-in-use port
|
||||
|
||||
try:
|
||||
server = http.server.HTTPServer((host, port), HttpRequestHandler)
|
||||
except OSError as exc:
|
||||
raise unittest.SkipTest(
|
||||
f"cannot spawn local HTTP server ({exc})")
|
||||
|
||||
host, port = server.server_address
|
||||
cls.address = f"http://{host}:{port}"
|
||||
threading.Thread(target=server.serve_forever, daemon=True).start()
|
||||
|
||||
def _run_test(self, ext, input, output,
|
||||
extension, expected_extension=None):
|
||||
TestDownloaderBase._run_test(
|
||||
self, f"{self.address}/{ext}", input, output,
|
||||
extension, expected_extension)
|
||||
|
||||
def tearDown(self):
|
||||
self.downloader.minsize = self.downloader.maxsize = None
|
||||
|
||||
def test_http_download(self):
|
||||
self._run_test("jpg", None, DATA["jpg"], "jpg", "jpg")
|
||||
self._run_test("png", None, DATA["png"], "png", "png")
|
||||
self._run_test("gif", None, DATA["gif"], "gif", "gif")
|
||||
|
||||
def test_http_offset(self):
|
||||
self._run_test("jpg", DATA["jpg"][:123], DATA["jpg"], "jpg", "jpg")
|
||||
self._run_test("png", DATA["png"][:12] , DATA["png"], "png", "png")
|
||||
self._run_test("gif", DATA["gif"][:1] , DATA["gif"], "gif", "gif")
|
||||
|
||||
def test_http_extension(self):
|
||||
self._run_test("jpg", None, DATA["jpg"], None, "jpg")
|
||||
self._run_test("png", None, DATA["png"], None, "png")
|
||||
self._run_test("gif", None, DATA["gif"], None, "gif")
|
||||
|
||||
def test_http_adjust_extension(self):
|
||||
self._run_test("jpg", None, DATA["jpg"], "png", "jpg")
|
||||
self._run_test("png", None, DATA["png"], "gif", "png")
|
||||
self._run_test("gif", None, DATA["gif"], "jpg", "gif")
|
||||
|
||||
def test_http_filesize_min(self):
|
||||
url = f"{self.address}/gif"
|
||||
pathfmt = self._prepare_destination(None, extension=None)
|
||||
self.downloader.minsize = 100
|
||||
with self.assertLogs(self.downloader.log, "WARNING"):
|
||||
success = self.downloader.download(url, pathfmt)
|
||||
self.assertTrue(success)
|
||||
self.assertEqual(pathfmt.temppath, "")
|
||||
|
||||
def test_http_filesize_max(self):
|
||||
url = f"{self.address}/jpg"
|
||||
pathfmt = self._prepare_destination(None, extension=None)
|
||||
self.downloader.maxsize = 100
|
||||
with self.assertLogs(self.downloader.log, "WARNING"):
|
||||
success = self.downloader.download(url, pathfmt)
|
||||
self.assertTrue(success)
|
||||
self.assertEqual(pathfmt.temppath, "")
|
||||
|
||||
def test_http_empty(self):
|
||||
url = f"{self.address}/~NUL"
|
||||
pathfmt = self._prepare_destination(None, extension=None)
|
||||
with self.assertLogs(self.downloader.log, "WARNING") as log_info:
|
||||
success = self.downloader.download(url, pathfmt)
|
||||
self.assertFalse(success)
|
||||
self.assertEqual(log_info.output[0],
|
||||
"WARNING:downloader.http:Empty file")
|
||||
|
||||
|
||||
class TestTextDownloader(TestDownloaderBase):
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
TestDownloaderBase.setUpClass()
|
||||
cls.downloader = downloader.find("text")(cls.job)
|
||||
|
||||
def test_text_download(self):
|
||||
self._run_test("text:foobar", None, "foobar", "txt", "txt")
|
||||
|
||||
def test_text_offset(self):
|
||||
self._run_test("text:foobar", "foo", "foobar", "txt", "txt")
|
||||
|
||||
def test_text_empty(self):
|
||||
self._run_test("text:", None, "", "txt", "txt")
|
||||
|
||||
|
||||
class HttpRequestHandler(http.server.BaseHTTPRequestHandler):
|
||||
|
||||
def do_GET(self):
|
||||
try:
|
||||
output = DATA[self.path[1:]]
|
||||
except KeyError:
|
||||
self.send_response(404)
|
||||
self.wfile.write(self.path.encode())
|
||||
return
|
||||
|
||||
headers = {"Content-Length": len(output)}
|
||||
|
||||
if "Range" in self.headers:
|
||||
status = 206
|
||||
|
||||
match = re.match(r"bytes=(\d+)-", self.headers["Range"])
|
||||
start = int(match[1])
|
||||
|
||||
headers["Content-Range"] = \
|
||||
f"bytes {start}-{len(output) - 1}/{len(output)}"
|
||||
output = output[start:]
|
||||
else:
|
||||
status = 200
|
||||
|
||||
self.send_response(status)
|
||||
for key, value in headers.items():
|
||||
self.send_header(key, value)
|
||||
self.end_headers()
|
||||
self.wfile.write(output)
|
||||
|
||||
|
||||
SAMPLES = {
|
||||
("jpg" , binascii.a2b_base64(
|
||||
"/9j/4AAQSkZJRgABAQEASABIAAD/2wBDAAEBAQEBAQEBAQEBAQEBAQEBAQEBAQEB"
|
||||
"AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQH/2wBDAQEB"
|
||||
"AQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEBAQEB"
|
||||
"AQEBAQEBAQEBAQEBAQH/wAARCAABAAEDAREAAhEBAxEB/8QAFAABAAAAAAAAAAAA"
|
||||
"AAAAAAAACv/EABQQAQAAAAAAAAAAAAAAAAAAAAD/xAAUAQEAAAAAAAAAAAAAAAAA"
|
||||
"AAAA/8QAFBEBAAAAAAAAAAAAAAAAAAAAAP/aAAwDAQACEQMRAD8AfwD/2Q==")),
|
||||
("png" , binascii.a2b_base64(
|
||||
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAAAAAA6fptVAAAACklEQVQIHWP4DwAB"
|
||||
"AQEANl9ngAAAAABJRU5ErkJggg==")),
|
||||
("gif" , binascii.a2b_base64(
|
||||
"R0lGODdhAQABAIAAAP///////ywAAAAAAQABAAACAkQBADs=")),
|
||||
("bmp" , b"BM"),
|
||||
("webp", b"RIFF????WEBP"),
|
||||
("avif", b"????ftypavif"),
|
||||
("avif", b"????ftypavis"),
|
||||
("heic", b"????ftypheic"),
|
||||
("heic", b"????ftypheim"),
|
||||
("heic", b"????ftypheis"),
|
||||
("heic", b"????ftypheix"),
|
||||
("svg" , b"<?xml"),
|
||||
("htm" , b"<!DOCTYPE html><html>...</html>"),
|
||||
("htm" , b" \n \n\r\t\n <!DOCTYPE html><html>...</html>"),
|
||||
("ico" , b"\x00\x00\x01\x00"),
|
||||
("cur" , b"\x00\x00\x02\x00"),
|
||||
("psd" , b"8BPS"),
|
||||
("mp4" , b"????ftypmp4"),
|
||||
("mp4" , b"????ftypavc1"),
|
||||
("mp4" , b"????ftypiso3"),
|
||||
("m4v" , b"????ftypM4V"),
|
||||
("mov" , b"????ftypqt "),
|
||||
("webm", b"\x1A\x45\xDF\xA3"),
|
||||
("ogg" , b"OggS"),
|
||||
("wav" , b"RIFF????WAVE"),
|
||||
("mp3" , b"ID3"),
|
||||
("mp3" , b"\xFF\xFB"),
|
||||
("mp3" , b"\xFF\xF3"),
|
||||
("mp3" , b"\xFF\xF2"),
|
||||
("aac" , b"\xFF\xF9"),
|
||||
("aac" , b"\xFF\xF1"),
|
||||
("m3u8", b"#EXTM3U\n#EXT-X-STREAM-INF:PROGRAM-ID=1, BANDWIDTH=200000"),
|
||||
("mpd" , b'<MPD xmlns="urn:mpeg:dash:schema:mpd:2011"'),
|
||||
("zip" , b"PK\x03\x04"),
|
||||
("zip" , b"PK\x05\x06"),
|
||||
("zip" , b"PK\x07\x08"),
|
||||
("rar" , b"Rar!\x1A\x07"),
|
||||
("rar" , b"\x52\x61\x72\x21\x1A\x07"),
|
||||
("7z" , b"\x37\x7A\xBC\xAF\x27\x1C"),
|
||||
("pdf" , b"%PDF-"),
|
||||
("swf" , b"FWS"),
|
||||
("swf" , b"CWS"),
|
||||
("blend", b"BLENDER-v303RENDH"),
|
||||
("obj" , b"# Blender v3.2.0 OBJ File: 'foo.blend'"),
|
||||
("clip", b"CSFCHUNK\x00\x00\x00\x00"),
|
||||
("~NUL", b""),
|
||||
}
|
||||
|
||||
|
||||
DATA = {}
|
||||
|
||||
for ext, content in SAMPLES:
|
||||
if ext not in DATA:
|
||||
DATA[ext] = content
|
||||
|
||||
for idx, (_, content) in enumerate(SAMPLES):
|
||||
DATA[f"S{idx:>02}"] = content
|
||||
|
||||
|
||||
# reverse mime types mapping
|
||||
MIME_TYPES = {
|
||||
ext: mtype
|
||||
for mtype, ext in MIME_TYPES.items()
|
||||
}
|
||||
|
||||
|
||||
def generate_tests():
|
||||
def generate_test(idx, ext, content):
|
||||
def test(self):
|
||||
self._run_test(f"S{idx:>02}", None, content, "bin", ext)
|
||||
test.__name__ = f"test_http_ext_{idx:>02}_{ext}"
|
||||
return test
|
||||
|
||||
for idx, (ext, content) in enumerate(SAMPLES):
|
||||
if ext[0].isalnum():
|
||||
test = generate_test(idx, ext, content)
|
||||
setattr(TestHTTPDownloader, test.__name__, test)
|
||||
|
||||
|
||||
generate_tests()
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in new issue
Block a user