mirror of
https://github.com/mikf/gallery-dl.git
synced 2024-11-23 11:12:40 +01:00
d1bb7fae2d
makedirs in those places is only meant to create missing directories, anything should be considered an error
336 lines
11 KiB
Python
336 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
# Copyright 2021-2022 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.
|
|
|
|
"""Filesystem path handling"""
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
import functools
|
|
from . import util, formatter, exception
|
|
|
|
WINDOWS = util.WINDOWS
|
|
|
|
|
|
class PathFormat():
|
|
EXTENSION_MAP = {
|
|
"jpeg": "jpg",
|
|
"jpe" : "jpg",
|
|
"jfif": "jpg",
|
|
"jif" : "jpg",
|
|
"jfi" : "jpg",
|
|
}
|
|
|
|
def __init__(self, extractor):
|
|
config = extractor.config
|
|
kwdefault = config("keywords-default")
|
|
|
|
filename_fmt = config("filename")
|
|
try:
|
|
if filename_fmt is None:
|
|
filename_fmt = extractor.filename_fmt
|
|
elif isinstance(filename_fmt, dict):
|
|
self.filename_conditions = [
|
|
(util.compile_expression(expr),
|
|
formatter.parse(fmt, kwdefault).format_map)
|
|
for expr, fmt in filename_fmt.items() if expr
|
|
]
|
|
self.build_filename = self.build_filename_conditional
|
|
filename_fmt = filename_fmt.get("", extractor.filename_fmt)
|
|
|
|
self.filename_formatter = formatter.parse(
|
|
filename_fmt, kwdefault).format_map
|
|
except Exception as exc:
|
|
raise exception.FilenameFormatError(exc)
|
|
|
|
directory_fmt = config("directory")
|
|
try:
|
|
if directory_fmt is None:
|
|
directory_fmt = extractor.directory_fmt
|
|
elif isinstance(directory_fmt, dict):
|
|
self.directory_conditions = [
|
|
(util.compile_expression(expr), [
|
|
formatter.parse(fmt, kwdefault).format_map
|
|
for fmt in fmts
|
|
])
|
|
for expr, fmts in directory_fmt.items() if expr
|
|
]
|
|
self.build_directory = self.build_directory_conditional
|
|
directory_fmt = directory_fmt.get("", extractor.directory_fmt)
|
|
|
|
self.directory_formatters = [
|
|
formatter.parse(dirfmt, kwdefault).format_map
|
|
for dirfmt in directory_fmt
|
|
]
|
|
except Exception as exc:
|
|
raise exception.DirectoryFormatError(exc)
|
|
|
|
self.kwdict = {}
|
|
self.directory = self.realdirectory = \
|
|
self.filename = self.extension = self.prefix = \
|
|
self.path = self.realpath = self.temppath = ""
|
|
self.delete = False
|
|
|
|
extension_map = config("extension-map")
|
|
if extension_map is None:
|
|
extension_map = self.EXTENSION_MAP
|
|
self.extension_map = extension_map.get
|
|
|
|
restrict = config("path-restrict", "auto")
|
|
replace = config("path-replace", "_")
|
|
if restrict == "auto":
|
|
restrict = "\\\\|/<>:\"?*" if WINDOWS else "/"
|
|
elif restrict == "unix":
|
|
restrict = "/"
|
|
elif restrict == "windows":
|
|
restrict = "\\\\|/<>:\"?*"
|
|
elif restrict == "ascii":
|
|
restrict = "^0-9A-Za-z_."
|
|
self.clean_segment = self._build_cleanfunc(restrict, replace)
|
|
|
|
remove = config("path-remove", "\x00-\x1f\x7f")
|
|
self.clean_path = self._build_cleanfunc(remove, "")
|
|
|
|
strip = config("path-strip", "auto")
|
|
if strip == "auto":
|
|
strip = ". " if WINDOWS else ""
|
|
elif strip == "unix":
|
|
strip = ""
|
|
elif strip == "windows":
|
|
strip = ". "
|
|
self.strip = strip
|
|
|
|
basedir = extractor._parentdir
|
|
if not basedir:
|
|
basedir = config("base-directory")
|
|
sep = os.sep
|
|
if basedir is None:
|
|
basedir = "." + sep + "gallery-dl" + sep
|
|
elif basedir:
|
|
basedir = util.expand_path(basedir)
|
|
altsep = os.altsep
|
|
if altsep and altsep in basedir:
|
|
basedir = basedir.replace(altsep, sep)
|
|
if basedir[-1] != sep:
|
|
basedir += sep
|
|
basedir = self.clean_path(basedir)
|
|
self.basedirectory = basedir
|
|
|
|
@staticmethod
|
|
def _build_cleanfunc(chars, repl):
|
|
if not chars:
|
|
return util.identity
|
|
elif isinstance(chars, dict):
|
|
def func(x, table=str.maketrans(chars)):
|
|
return x.translate(table)
|
|
elif len(chars) == 1:
|
|
def func(x, c=chars, r=repl):
|
|
return x.replace(c, r)
|
|
else:
|
|
return functools.partial(
|
|
re.compile("[" + chars + "]").sub, repl)
|
|
return func
|
|
|
|
def open(self, mode="wb"):
|
|
"""Open file and return a corresponding file object"""
|
|
try:
|
|
return open(self.temppath, mode)
|
|
except FileNotFoundError:
|
|
os.makedirs(self.realdirectory)
|
|
return open(self.temppath, mode)
|
|
|
|
def exists(self):
|
|
"""Return True if the file exists on disk"""
|
|
if self.extension and os.path.exists(self.realpath):
|
|
return self.check_file()
|
|
return False
|
|
|
|
@staticmethod
|
|
def check_file():
|
|
return True
|
|
|
|
def _enum_file(self):
|
|
num = 1
|
|
try:
|
|
while True:
|
|
self.prefix = str(num) + "."
|
|
self.set_extension(self.extension, False)
|
|
os.stat(self.realpath) # raises OSError if file doesn't exist
|
|
num += 1
|
|
except OSError:
|
|
pass
|
|
return False
|
|
|
|
def set_directory(self, kwdict):
|
|
"""Build directory path and create it if necessary"""
|
|
self.kwdict = kwdict
|
|
sep = os.sep
|
|
|
|
segments = self.build_directory(kwdict)
|
|
if segments:
|
|
self.directory = directory = self.basedirectory + self.clean_path(
|
|
sep.join(segments) + sep)
|
|
else:
|
|
self.directory = directory = self.basedirectory
|
|
|
|
if WINDOWS:
|
|
# Enable longer-than-260-character paths
|
|
directory = os.path.abspath(directory)
|
|
if directory.startswith("\\\\"):
|
|
directory = "\\\\?\\UNC\\" + directory[2:]
|
|
else:
|
|
directory = "\\\\?\\" + directory
|
|
|
|
# abspath() in Python 3.7+ removes trailing path separators (#402)
|
|
if directory[-1] != sep:
|
|
directory += sep
|
|
|
|
self.realdirectory = directory
|
|
|
|
def set_filename(self, kwdict):
|
|
"""Set general filename data"""
|
|
self.kwdict = kwdict
|
|
self.temppath = self.prefix = ""
|
|
|
|
ext = kwdict["extension"]
|
|
kwdict["extension"] = self.extension = self.extension_map(ext, ext)
|
|
|
|
if self.extension:
|
|
self.build_path()
|
|
else:
|
|
self.filename = ""
|
|
|
|
def set_extension(self, extension, real=True):
|
|
"""Set filename extension"""
|
|
extension = self.extension_map(extension, extension)
|
|
if real:
|
|
self.extension = extension
|
|
self.kwdict["extension"] = self.prefix + extension
|
|
self.build_path()
|
|
|
|
def fix_extension(self, _=None):
|
|
"""Fix filenames without a given filename extension"""
|
|
if not self.extension:
|
|
self.set_extension("", False)
|
|
if self.path[-1] == ".":
|
|
self.path = self.path[:-1]
|
|
self.temppath = self.realpath = self.realpath[:-1]
|
|
return True
|
|
|
|
def build_filename(self, kwdict):
|
|
"""Apply 'kwdict' to filename format string"""
|
|
try:
|
|
return self.clean_path(self.clean_segment(
|
|
self.filename_formatter(kwdict)))
|
|
except Exception as exc:
|
|
raise exception.FilenameFormatError(exc)
|
|
|
|
def build_filename_conditional(self, kwdict):
|
|
try:
|
|
for condition, fmt in self.filename_conditions:
|
|
if condition(kwdict):
|
|
break
|
|
else:
|
|
fmt = self.filename_formatter
|
|
return self.clean_path(self.clean_segment(fmt(kwdict)))
|
|
except Exception as exc:
|
|
raise exception.FilenameFormatError(exc)
|
|
|
|
def build_directory(self, kwdict):
|
|
"""Apply 'kwdict' to directory format strings"""
|
|
segments = []
|
|
append = segments.append
|
|
strip = self.strip
|
|
|
|
try:
|
|
for fmt in self.directory_formatters:
|
|
segment = fmt(kwdict).strip()
|
|
if strip:
|
|
# remove trailing dots and spaces (#647)
|
|
segment = segment.rstrip(strip)
|
|
if segment:
|
|
append(self.clean_segment(segment))
|
|
return segments
|
|
except Exception as exc:
|
|
raise exception.DirectoryFormatError(exc)
|
|
|
|
def build_directory_conditional(self, kwdict):
|
|
segments = []
|
|
append = segments.append
|
|
strip = self.strip
|
|
|
|
try:
|
|
for condition, formatters in self.directory_conditions:
|
|
if condition(kwdict):
|
|
break
|
|
else:
|
|
formatters = self.directory_formatters
|
|
for fmt in formatters:
|
|
segment = fmt(kwdict).strip()
|
|
if strip:
|
|
segment = segment.rstrip(strip)
|
|
if segment:
|
|
append(self.clean_segment(segment))
|
|
return segments
|
|
except Exception as exc:
|
|
raise exception.DirectoryFormatError(exc)
|
|
|
|
def build_path(self):
|
|
"""Combine directory and filename to full paths"""
|
|
self.filename = filename = self.build_filename(self.kwdict)
|
|
self.path = self.directory + filename
|
|
self.realpath = self.realdirectory + filename
|
|
if not self.temppath:
|
|
self.temppath = self.realpath
|
|
|
|
def part_enable(self, part_directory=None):
|
|
"""Enable .part file usage"""
|
|
if self.extension:
|
|
self.temppath += ".part"
|
|
else:
|
|
self.set_extension("part", False)
|
|
if part_directory:
|
|
self.temppath = os.path.join(
|
|
part_directory,
|
|
os.path.basename(self.temppath),
|
|
)
|
|
|
|
def part_size(self):
|
|
"""Return size of .part file"""
|
|
try:
|
|
return os.stat(self.temppath).st_size
|
|
except OSError:
|
|
pass
|
|
return 0
|
|
|
|
def finalize(self):
|
|
"""Move tempfile to its target location"""
|
|
if self.delete:
|
|
self.delete = False
|
|
os.unlink(self.temppath)
|
|
return
|
|
|
|
if self.temppath != self.realpath:
|
|
# Move temp file to its actual location
|
|
while True:
|
|
try:
|
|
os.replace(self.temppath, self.realpath)
|
|
except FileNotFoundError:
|
|
# delayed directory creation
|
|
os.makedirs(self.realdirectory)
|
|
continue
|
|
except OSError:
|
|
# move across different filesystems
|
|
shutil.copyfile(self.temppath, self.realpath)
|
|
os.unlink(self.temppath)
|
|
break
|
|
|
|
mtime = self.kwdict.get("_mtime")
|
|
if mtime:
|
|
util.set_mtime(self.realpath, mtime)
|