1266 lines
45 KiB
Python
1266 lines
45 KiB
Python
"""A class to represent a single comic, be it file or folder of images"""
|
|
# Copyright 2012-2014 Anthony Beville
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import platform
|
|
import shutil
|
|
import struct
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import zipfile
|
|
from typing import cast
|
|
|
|
import natsort
|
|
import wordninja
|
|
|
|
from comicapi import filenamelexer, filenameparser, utils
|
|
from comicapi.comet import CoMet
|
|
from comicapi.comicbookinfo import ComicBookInfo
|
|
from comicapi.comicinfoxml import ComicInfoXml
|
|
from comicapi.genericmetadata import GenericMetadata, PageType
|
|
|
|
try:
|
|
import py7zr
|
|
|
|
z7_support = True
|
|
except ImportError:
|
|
z7_support = False
|
|
try:
|
|
from unrar.cffi import rarfile
|
|
|
|
rar_support = True
|
|
except ImportError:
|
|
rar_support = False
|
|
|
|
try:
|
|
from PIL import Image
|
|
|
|
pil_available = True
|
|
except ImportError:
|
|
pil_available = False
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
if not pil_available:
|
|
logger.error("PIL unavalable")
|
|
if not rar_support:
|
|
logger.error("unrar-cffi unavailable")
|
|
|
|
|
|
class MetaDataStyle:
|
|
CBI = 0
|
|
CIX = 1
|
|
COMET = 2
|
|
name = ["ComicBookLover", "ComicRack", "CoMet"]
|
|
short_name = ["cbl", "cr", "comet"]
|
|
|
|
|
|
class UnknownArchiver:
|
|
|
|
"""Unknown implementation"""
|
|
|
|
def __init__(self, path: pathlib.Path | str) -> None:
|
|
self.path = pathlib.Path(path)
|
|
|
|
def get_comment(self) -> str:
|
|
return ""
|
|
|
|
def set_comment(self, comment: str) -> bool:
|
|
return False
|
|
|
|
def read_file(self, archive_file: str) -> bytes:
|
|
raise NotImplementedError
|
|
|
|
def remove_file(self, archive_file: str) -> bool:
|
|
return False
|
|
|
|
def write_file(self, archive_file: str, data: bytes) -> bool:
|
|
return False
|
|
|
|
def get_filename_list(self) -> list[str]:
|
|
return []
|
|
|
|
def rebuild(self, exclude_list: list[str]) -> bool:
|
|
return False
|
|
|
|
def copy_from_archive(self, other_archive: UnknownArchiver) -> bool:
|
|
return False
|
|
|
|
|
|
class SevenZipArchiver(UnknownArchiver):
|
|
|
|
"""7Z implementation"""
|
|
|
|
def __init__(self, path: pathlib.Path | str) -> None:
|
|
super().__init__(path)
|
|
|
|
# @todo: Implement Comment?
|
|
def get_comment(self) -> str:
|
|
return ""
|
|
|
|
def set_comment(self, comment: str) -> bool:
|
|
return False
|
|
|
|
def read_file(self, archive_file: str) -> bytes:
|
|
data = b""
|
|
try:
|
|
with py7zr.SevenZipFile(self.path, "r") as zf:
|
|
data = zf.read(archive_file)[archive_file].read()
|
|
except (py7zr.Bad7zFile, OSError) as e:
|
|
logger.error("Error reading 7zip archive [%s]: %s :: %s", e, self.path, archive_file)
|
|
raise
|
|
|
|
return data
|
|
|
|
def remove_file(self, archive_file: str) -> bool:
|
|
return self.rebuild([archive_file])
|
|
|
|
def write_file(self, archive_file: str, data: bytes) -> bool:
|
|
# At the moment, no other option but to rebuild the whole
|
|
# archive w/o the indicated file. Very sucky, but maybe
|
|
# another solution can be found
|
|
files = self.get_filename_list()
|
|
if archive_file in files:
|
|
if not self.rebuild([archive_file]):
|
|
return False
|
|
|
|
try:
|
|
# now just add the archive file as a new one
|
|
with py7zr.SevenZipFile(self.path, "a") as zf:
|
|
zf.writestr(data, archive_file)
|
|
return True
|
|
except (py7zr.Bad7zFile, OSError) as e:
|
|
logger.error("Error writing 7zip archive [%s]: %s :: %s", e, self.path, archive_file)
|
|
return False
|
|
|
|
def get_filename_list(self) -> list[str]:
|
|
try:
|
|
with py7zr.SevenZipFile(self.path, "r") as zf:
|
|
namelist: list[str] = [file.filename for file in zf.list() if not file.is_directory]
|
|
|
|
return namelist
|
|
except (py7zr.Bad7zFile, OSError) as e:
|
|
logger.error("Error listing files in 7zip archive [%s]: %s", e, self.path)
|
|
return []
|
|
|
|
def rebuild(self, exclude_list: list[str]) -> bool:
|
|
"""Zip helper func
|
|
|
|
This recompresses the zip archive, without the files in the exclude_list
|
|
"""
|
|
|
|
try:
|
|
# py7zr treats all archives as if they used solid compression
|
|
# so we need to get the filename list first to read all the files at once
|
|
with py7zr.SevenZipFile(self.path, mode="r") as zin:
|
|
targets = [f for f in zin.getnames() if f not in exclude_list]
|
|
with tempfile.NamedTemporaryFile(dir=os.path.dirname(self.path), delete=False) as tmp_file:
|
|
with py7zr.SevenZipFile(tmp_file.file, mode="w") as zout:
|
|
with py7zr.SevenZipFile(self.path, mode="r") as zin:
|
|
for filename, buffer in zin.read(targets).items():
|
|
zout.writef(buffer, filename)
|
|
|
|
self.path.unlink(missing_ok=True)
|
|
tmp_file.close() # Required on windows
|
|
|
|
shutil.move(tmp_file.name, self.path)
|
|
except (py7zr.Bad7zFile, OSError) as e:
|
|
logger.error("Error rebuilding 7zip file [%s]: %s", e, self.path)
|
|
return False
|
|
return True
|
|
|
|
def copy_from_archive(self, other_archive: UnknownArchiver) -> bool:
|
|
"""Replace the current zip with one copied from another archive"""
|
|
try:
|
|
with py7zr.SevenZipFile(self.path, "w") as zout:
|
|
for filename in other_archive.get_filename_list():
|
|
data = other_archive.read_file(
|
|
filename
|
|
) # This will be very inefficient if other_archive is a 7z file
|
|
if data is not None:
|
|
zout.writestr(data, filename)
|
|
except Exception as e:
|
|
logger.error("Error while copying to 7zip archive [%s]: from %s to %s", e, other_archive.path, self.path)
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
class ZipArchiver(UnknownArchiver):
|
|
|
|
"""ZIP implementation"""
|
|
|
|
def __init__(self, path: pathlib.Path | str) -> None:
|
|
super().__init__(path)
|
|
|
|
def get_comment(self) -> str:
|
|
with zipfile.ZipFile(self.path, "r") as zf:
|
|
comment = zf.comment.decode("utf-8")
|
|
return comment
|
|
|
|
def set_comment(self, comment: str) -> bool:
|
|
with zipfile.ZipFile(self.path, mode="a") as zf:
|
|
zf.comment = bytes(comment, "utf-8")
|
|
return True
|
|
|
|
def read_file(self, archive_file: str) -> bytes:
|
|
with zipfile.ZipFile(self.path, mode="r") as zf:
|
|
try:
|
|
data = zf.read(archive_file)
|
|
except (zipfile.BadZipfile, OSError) as e:
|
|
logger.error("Error reading zip archive [%s]: %s :: %s", e, self.path, archive_file)
|
|
raise
|
|
return data
|
|
|
|
def remove_file(self, archive_file: str) -> bool:
|
|
return self.rebuild([archive_file])
|
|
|
|
def write_file(self, archive_file: str, data: bytes) -> bool:
|
|
# At the moment, no other option but to rebuild the whole
|
|
# zip archive w/o the indicated file. Very sucky, but maybe
|
|
# another solution can be found
|
|
files = self.get_filename_list()
|
|
if archive_file in files:
|
|
if not self.rebuild([archive_file]):
|
|
return False
|
|
|
|
try:
|
|
# now just add the archive file as a new one
|
|
with zipfile.ZipFile(self.path, mode="a", allowZip64=True, compression=zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr(archive_file, data)
|
|
return True
|
|
except (zipfile.BadZipfile, OSError) as e:
|
|
logger.error("Error writing zip archive [%s]: %s :: %s", e, self.path, archive_file)
|
|
return False
|
|
|
|
def get_filename_list(self) -> list[str]:
|
|
try:
|
|
with zipfile.ZipFile(self.path, mode="r") as zf:
|
|
namelist = [file.filename for file in zf.infolist() if not file.is_dir()]
|
|
return namelist
|
|
except (zipfile.BadZipfile, OSError) as e:
|
|
logger.error("Error listing files in zip archive [%s]: %s", e, self.path)
|
|
return []
|
|
|
|
def rebuild(self, exclude_list: list[str]) -> bool:
|
|
"""Zip helper func
|
|
|
|
This recompresses the zip archive, without the files in the exclude_list
|
|
"""
|
|
try:
|
|
with zipfile.ZipFile(
|
|
tempfile.NamedTemporaryFile(dir=os.path.dirname(self.path), delete=False), "w", allowZip64=True
|
|
) as zout:
|
|
with zipfile.ZipFile(self.path, mode="r") as zin:
|
|
for item in zin.infolist():
|
|
buffer = zin.read(item.filename)
|
|
if item.filename not in exclude_list:
|
|
zout.writestr(item, buffer)
|
|
|
|
# preserve the old comment
|
|
zout.comment = zin.comment
|
|
|
|
# replace with the new file
|
|
self.path.unlink(missing_ok=True)
|
|
zout.close() # Required on windows
|
|
|
|
shutil.move(cast(str, zout.filename), self.path)
|
|
|
|
except (zipfile.BadZipfile, OSError) as e:
|
|
logger.error("Error rebuilding zip file [%s]: %s", e, self.path)
|
|
return False
|
|
return True
|
|
|
|
def copy_from_archive(self, other_archive: UnknownArchiver) -> bool:
|
|
"""Replace the current zip with one copied from another archive"""
|
|
try:
|
|
with zipfile.ZipFile(self.path, mode="w", allowZip64=True) as zout:
|
|
for filename in other_archive.get_filename_list():
|
|
data = other_archive.read_file(filename)
|
|
if data is not None:
|
|
zout.writestr(filename, data)
|
|
|
|
# preserve the old comment
|
|
comment = other_archive.get_comment()
|
|
if comment is not None:
|
|
if not self.write_zip_comment(self.path, comment):
|
|
return False
|
|
except Exception as e:
|
|
logger.error("Error while copying to zip archive [%s]: from %s to %s", e, other_archive.path, self.path)
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
def write_zip_comment(self, filename: pathlib.Path | str, comment: str) -> bool:
|
|
"""
|
|
This is a custom function for writing a comment to a zip file,
|
|
since the built-in one doesn't seem to work on Windows and Mac OS/X
|
|
|
|
Fortunately, the zip comment is at the end of the file, and it's
|
|
easy to manipulate. See this website for more info:
|
|
see: http://en.wikipedia.org/wiki/Zip_(file_format)#Structure
|
|
"""
|
|
|
|
# get file size
|
|
statinfo = os.stat(filename)
|
|
file_length = statinfo.st_size
|
|
|
|
try:
|
|
with open(filename, mode="r+b") as file:
|
|
|
|
# the starting position, relative to EOF
|
|
pos = -4
|
|
found = False
|
|
|
|
# walk backwards to find the "End of Central Directory" record
|
|
while (not found) and (-pos != file_length):
|
|
# seek, relative to EOF
|
|
file.seek(pos, 2)
|
|
value = file.read(4)
|
|
|
|
# look for the end of central directory signature
|
|
if bytearray(value) == bytearray([0x50, 0x4B, 0x05, 0x06]):
|
|
found = True
|
|
else:
|
|
# not found, step back another byte
|
|
pos = pos - 1
|
|
|
|
if found:
|
|
|
|
# now skip forward 20 bytes to the comment length word
|
|
pos += 20
|
|
file.seek(pos, 2)
|
|
|
|
# Pack the length of the comment string
|
|
fmt = "H" # one 2-byte integer
|
|
comment_length = struct.pack(fmt, len(comment)) # pack integer in a binary string
|
|
|
|
# write out the length
|
|
file.write(comment_length)
|
|
file.seek(pos + 2, 2)
|
|
|
|
# write out the comment itself
|
|
file.write(comment.encode("utf-8"))
|
|
file.truncate()
|
|
else:
|
|
raise Exception("Could not find the End of Central Directory record!")
|
|
except Exception as e:
|
|
logger.error("Error writing comment to zip archive [%s]: %s", e, self.path)
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
class RarArchiver(UnknownArchiver):
|
|
"""RAR implementation"""
|
|
|
|
def __init__(self, path: pathlib.Path | str, rar_exe_path: str = "rar") -> None:
|
|
super().__init__(path)
|
|
self.rar_exe_path = shutil.which(rar_exe_path) or ""
|
|
|
|
# windows only, keeps the cmd.exe from popping up
|
|
if platform.system() == "Windows":
|
|
self.startupinfo = subprocess.STARTUPINFO() # type: ignore
|
|
self.startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW # type: ignore
|
|
else:
|
|
self.startupinfo = None
|
|
|
|
def get_comment(self) -> str:
|
|
rarc = self.get_rar_obj()
|
|
return rarc.comment.decode("utf-8") if rarc else ""
|
|
|
|
def set_comment(self, comment: str) -> bool:
|
|
if rar_support and self.rar_exe_path:
|
|
try:
|
|
# write comment to temp file
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
tmp_file = pathlib.Path(tmp_dir) / "rar_comment.txt"
|
|
tmp_file.write_text(comment, encoding="utf-8")
|
|
|
|
working_dir = os.path.dirname(os.path.abspath(self.path))
|
|
|
|
# use external program to write comment to Rar archive
|
|
proc_args = [
|
|
self.rar_exe_path,
|
|
"c",
|
|
f"-w{working_dir}",
|
|
"-c-",
|
|
f"-z{tmp_file}",
|
|
str(self.path),
|
|
]
|
|
subprocess.run(
|
|
proc_args,
|
|
startupinfo=self.startupinfo,
|
|
stdout=subprocess.DEVNULL,
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
check=True,
|
|
)
|
|
|
|
if platform.system() == "Darwin":
|
|
time.sleep(1)
|
|
except (subprocess.CalledProcessError, OSError) as e:
|
|
logger.exception("Error writing comment to rar archive [%s]: %s", e, self.path)
|
|
return False
|
|
else:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def read_file(self, archive_file: str) -> bytes:
|
|
|
|
rarc = self.get_rar_obj()
|
|
if rarc is None:
|
|
return b""
|
|
|
|
tries = 0
|
|
while tries < 7:
|
|
try:
|
|
tries = tries + 1
|
|
data: bytes = rarc.open(archive_file).read()
|
|
entries = [(rarc.getinfo(archive_file), data)]
|
|
|
|
if entries[0][0].file_size != len(entries[0][1]):
|
|
logger.info(
|
|
"Error reading rar archive [file is not expected size: %d vs %d] %s :: %s :: tries #%d",
|
|
entries[0][0].file_size,
|
|
len(entries[0][1]),
|
|
self.path,
|
|
archive_file,
|
|
tries,
|
|
)
|
|
continue
|
|
|
|
except OSError as e:
|
|
logger.error("Error reading rar archive [%s]: %s :: %s :: tries #%d", e, self.path, archive_file, tries)
|
|
time.sleep(1)
|
|
except Exception as e:
|
|
logger.error(
|
|
"Unexpected exception reading rar archive [%s]: %s :: %s :: tries #%d",
|
|
e,
|
|
self.path,
|
|
archive_file,
|
|
tries,
|
|
)
|
|
break
|
|
|
|
else:
|
|
# Success. Entries is a list of of tuples: ( rarinfo, filedata)
|
|
if len(entries) == 1:
|
|
return entries[0][1]
|
|
|
|
raise OSError
|
|
|
|
raise OSError
|
|
|
|
def remove_file(self, archive_file: str) -> bool:
|
|
if self.rar_exe_path:
|
|
# use external program to remove file from Rar archive
|
|
result = subprocess.run(
|
|
[self.rar_exe_path, "d", "-c-", self.path, archive_file],
|
|
startupinfo=self.startupinfo,
|
|
stdout=subprocess.DEVNULL,
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
|
|
if platform.system() == "Darwin":
|
|
time.sleep(1)
|
|
if result.returncode != 0:
|
|
logger.error(
|
|
"Error removing file from rar archive [exitcode: %d]: %s :: %s",
|
|
result.returncode,
|
|
self.path,
|
|
archive_file,
|
|
)
|
|
return False
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def write_file(self, archive_file: str, data: bytes) -> bool:
|
|
if self.rar_exe_path:
|
|
archive_path = pathlib.PurePosixPath(archive_file)
|
|
archive_name = archive_path.name
|
|
archive_parent = str(archive_path.parent).lstrip("./")
|
|
|
|
# use external program to write file to Rar archive
|
|
result = subprocess.run(
|
|
[self.rar_exe_path, "a", f"-si{archive_name}", f"-ap{archive_parent}", "-c-", "-ep", self.path],
|
|
input=data,
|
|
startupinfo=self.startupinfo,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
|
|
if platform.system() == "Darwin":
|
|
time.sleep(1)
|
|
if result.returncode != 0:
|
|
logger.error(
|
|
"Error writing rar archive [exitcode: %d]: %s :: %s", result.returncode, self.path, archive_file
|
|
)
|
|
return False
|
|
else:
|
|
return True
|
|
else:
|
|
return False
|
|
|
|
def get_filename_list(self) -> list[str]:
|
|
rarc = self.get_rar_obj()
|
|
tries = 0
|
|
if rar_support and rarc:
|
|
while tries < 7:
|
|
try:
|
|
tries = tries + 1
|
|
namelist = []
|
|
for item in rarc.infolist():
|
|
if item.file_size != 0:
|
|
namelist.append(item.filename)
|
|
|
|
except OSError as e:
|
|
logger.error("Error listing files in rar archive [%s]: %s :: attempt #%d", e, self.path, tries)
|
|
time.sleep(1)
|
|
|
|
else:
|
|
return namelist
|
|
return []
|
|
|
|
def copy_from_archive(self, other_archive: UnknownArchiver) -> bool:
|
|
"""Replace the current archive with one copied from another archive"""
|
|
try:
|
|
with tempfile.TemporaryDirectory() as tmp_dir:
|
|
tmp_path = pathlib.Path(tmp_dir)
|
|
rar_cwd = tmp_path / "rar"
|
|
rar_cwd.mkdir(exist_ok=True)
|
|
rar_path = (tmp_path / self.path.name).with_suffix(".rar")
|
|
|
|
for filename in other_archive.get_filename_list():
|
|
(rar_cwd / filename).parent.mkdir(exist_ok=True, parents=True)
|
|
data = other_archive.read_file(filename)
|
|
if data is not None:
|
|
with open(rar_cwd / filename, mode="w+b") as tmp_file:
|
|
tmp_file.write(data)
|
|
result = subprocess.run(
|
|
[self.rar_exe_path, "a", "-r", "-c-", str(rar_path.absolute()), "."],
|
|
cwd=rar_cwd.absolute(),
|
|
startupinfo=self.startupinfo,
|
|
stdout=subprocess.DEVNULL,
|
|
stdin=subprocess.DEVNULL,
|
|
stderr=subprocess.DEVNULL,
|
|
)
|
|
if result.returncode != 0:
|
|
logger.error("Error while copying to rar archive [exitcode: %d]: %s", result.returncode, self.path)
|
|
return False
|
|
|
|
self.path.unlink(missing_ok=True)
|
|
shutil.move(rar_path, self.path)
|
|
except Exception as e:
|
|
logger.exception("Error while copying to rar archive [%s]: from %s to %s", e, other_archive.path, self.path)
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
def get_rar_obj(self) -> rarfile.RarFile | None:
|
|
if rar_support:
|
|
try:
|
|
rarc = rarfile.RarFile(str(self.path))
|
|
except (OSError, rarfile.RarFileError) as e:
|
|
logger.error("Unable to get rar object [%s]: %s", e, self.path)
|
|
else:
|
|
return rarc
|
|
|
|
return None
|
|
|
|
|
|
class FolderArchiver(UnknownArchiver):
|
|
|
|
"""Folder implementation"""
|
|
|
|
def __init__(self, path: pathlib.Path | str) -> None:
|
|
super().__init__(path)
|
|
self.comment_file_name = "ComicTaggerFolderComment.txt"
|
|
|
|
def get_comment(self) -> str:
|
|
try:
|
|
return self.read_file(self.comment_file_name).decode("utf-8")
|
|
except OSError:
|
|
return ""
|
|
|
|
def set_comment(self, comment: str) -> bool:
|
|
if (self.path / self.comment_file_name).exists() or comment:
|
|
return self.write_file(self.comment_file_name, comment.encode("utf-8"))
|
|
return True
|
|
|
|
def read_file(self, archive_file: str) -> bytes:
|
|
try:
|
|
with open(self.path / archive_file, mode="rb") as f:
|
|
data = f.read()
|
|
except OSError as e:
|
|
logger.error("Error reading folder archive [%s]: %s :: %s", e, self.path, archive_file)
|
|
raise
|
|
|
|
return data
|
|
|
|
def remove_file(self, archive_file: str) -> bool:
|
|
try:
|
|
(self.path / archive_file).unlink(missing_ok=True)
|
|
except OSError as e:
|
|
logger.error("Error removing file for folder archive [%s]: %s :: %s", e, self.path, archive_file)
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
def write_file(self, archive_file: str, data: bytes) -> bool:
|
|
try:
|
|
file_path = self.path / archive_file
|
|
file_path.parent.mkdir(exist_ok=True, parents=True)
|
|
with open(self.path / archive_file, mode="wb") as f:
|
|
f.write(data)
|
|
except OSError as e:
|
|
logger.error("Error writing folder archive [%s]: %s :: %s", e, self.path, archive_file)
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
def get_filename_list(self) -> list[str]:
|
|
filenames = []
|
|
try:
|
|
for root, _dirs, files in os.walk(self.path):
|
|
for f in files:
|
|
filenames.append(os.path.relpath(os.path.join(root, f), self.path).replace(os.path.sep, "/"))
|
|
return filenames
|
|
except OSError as e:
|
|
logger.error("Error listing files in folder archive [%s]: %s", e, self.path)
|
|
return []
|
|
|
|
def copy_from_archive(self, other_archive: UnknownArchiver) -> bool:
|
|
"""Replace the current zip with one copied from another archive"""
|
|
try:
|
|
for filename in other_archive.get_filename_list():
|
|
data = other_archive.read_file(filename)
|
|
if data is not None:
|
|
self.write_file(filename, data)
|
|
|
|
# preserve the old comment
|
|
comment = other_archive.get_comment()
|
|
if comment is not None:
|
|
if not self.set_comment(comment):
|
|
return False
|
|
except Exception:
|
|
logger.exception("Error while copying archive from %s to %s", other_archive.path, self.path)
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
class ComicArchive:
|
|
logo_data = b""
|
|
|
|
class ArchiveType:
|
|
SevenZip, Zip, Rar, Folder, Pdf, Unknown = list(range(6))
|
|
|
|
def __init__(
|
|
self,
|
|
path: pathlib.Path | str,
|
|
rar_exe_path: str = "rar",
|
|
default_image_path: pathlib.Path | str | None = None,
|
|
) -> None:
|
|
self.cbi_md: GenericMetadata | None = None
|
|
self.cix_md: GenericMetadata | None = None
|
|
self.comet_filename: str | None = None
|
|
self.comet_md: GenericMetadata | None = None
|
|
self._has_cbi: bool | None = None
|
|
self._has_cix: bool | None = None
|
|
self._has_comet: bool | None = None
|
|
self.path = pathlib.Path(path).absolute()
|
|
self.page_count: int | None = None
|
|
self.page_list: list[str] = []
|
|
|
|
self.rar_exe_path = shutil.which(rar_exe_path or "rar") or ""
|
|
self.ci_xml_filename = "ComicInfo.xml"
|
|
self.comet_default_filename = "CoMet.xml"
|
|
self.reset_cache()
|
|
self.default_image_path = default_image_path
|
|
|
|
# Use file extension to decide which archive test we do first
|
|
ext = self.path.suffix
|
|
|
|
self.archive_type = self.ArchiveType.Unknown
|
|
self.archiver = UnknownArchiver(self.path)
|
|
|
|
if ext in [".cbr", ".rar"]:
|
|
if self.rar_test():
|
|
self.archive_type = self.ArchiveType.Rar
|
|
self.archiver = RarArchiver(self.path, rar_exe_path=self.rar_exe_path)
|
|
|
|
elif self.zip_test():
|
|
self.archive_type = self.ArchiveType.Zip
|
|
self.archiver = ZipArchiver(self.path)
|
|
else:
|
|
if self.sevenzip_test():
|
|
self.archive_type = self.ArchiveType.SevenZip
|
|
self.archiver = SevenZipArchiver(self.path)
|
|
|
|
elif self.zip_test():
|
|
self.archive_type = self.ArchiveType.Zip
|
|
self.archiver = ZipArchiver(self.path)
|
|
|
|
elif self.rar_test():
|
|
self.archive_type = self.ArchiveType.Rar
|
|
self.archiver = RarArchiver(self.path, rar_exe_path=self.rar_exe_path)
|
|
|
|
elif self.folder_test():
|
|
self.archive_type = self.ArchiveType.Folder
|
|
self.archiver = FolderArchiver(self.path)
|
|
|
|
if not ComicArchive.logo_data and self.default_image_path:
|
|
with open(self.default_image_path, mode="rb") as fd:
|
|
ComicArchive.logo_data = fd.read()
|
|
|
|
def reset_cache(self) -> None:
|
|
"""Clears the cached data"""
|
|
|
|
self._has_cix = None
|
|
self._has_cbi = None
|
|
self._has_comet = None
|
|
self.comet_filename = None
|
|
self.page_count = None
|
|
self.page_list = []
|
|
self.cix_md = None
|
|
self.cbi_md = None
|
|
self.comet_md = None
|
|
|
|
def load_cache(self, style_list: list[int]) -> None:
|
|
for style in style_list:
|
|
self.read_metadata(style)
|
|
|
|
def rename(self, path: pathlib.Path | str) -> None:
|
|
new_path = pathlib.Path(path).absolute()
|
|
if new_path == self.path:
|
|
return
|
|
os.makedirs(new_path.parent, 0o777, True)
|
|
shutil.move(self.path, new_path)
|
|
self.path = new_path
|
|
self.archiver.path = pathlib.Path(path)
|
|
|
|
def sevenzip_test(self) -> bool:
|
|
return z7_support and py7zr.is_7zfile(self.path)
|
|
|
|
def zip_test(self) -> bool:
|
|
return zipfile.is_zipfile(self.path)
|
|
|
|
def rar_test(self) -> bool:
|
|
return rar_support and rarfile.is_rarfile(str(self.path))
|
|
|
|
def folder_test(self) -> bool:
|
|
return self.path.is_dir()
|
|
|
|
def is_sevenzip(self) -> bool:
|
|
return self.archive_type == self.ArchiveType.SevenZip
|
|
|
|
def is_zip(self) -> bool:
|
|
return self.archive_type == self.ArchiveType.Zip
|
|
|
|
def is_rar(self) -> bool:
|
|
return self.archive_type == self.ArchiveType.Rar
|
|
|
|
def is_pdf(self) -> bool:
|
|
return self.archive_type == self.ArchiveType.Pdf
|
|
|
|
def is_folder(self) -> bool:
|
|
return self.archive_type == self.ArchiveType.Folder
|
|
|
|
def is_writable(self, check_rar_status: bool = True) -> bool:
|
|
if self.archive_type == self.ArchiveType.Unknown:
|
|
return False
|
|
|
|
if check_rar_status and self.is_rar() and not self.rar_exe_path:
|
|
return False
|
|
|
|
if not os.access(self.path, os.W_OK):
|
|
return False
|
|
|
|
if (self.archive_type != self.ArchiveType.Folder) and (not os.access(self.path.parent, os.W_OK)):
|
|
return False
|
|
|
|
return True
|
|
|
|
def is_writable_for_style(self, data_style: int) -> bool:
|
|
|
|
if (self.is_rar() or self.is_sevenzip()) and data_style == MetaDataStyle.CBI:
|
|
return False
|
|
|
|
return self.is_writable()
|
|
|
|
def seems_to_be_a_comic_archive(self) -> bool:
|
|
if (self.is_zip() or self.is_rar() or self.is_sevenzip() or self.is_folder()) and (
|
|
self.get_number_of_pages() > 0
|
|
):
|
|
return True
|
|
|
|
return False
|
|
|
|
def read_metadata(self, style: int) -> GenericMetadata:
|
|
|
|
if style == MetaDataStyle.CIX:
|
|
return self.read_cix()
|
|
if style == MetaDataStyle.CBI:
|
|
return self.read_cbi()
|
|
if style == MetaDataStyle.COMET:
|
|
return self.read_comet()
|
|
return GenericMetadata()
|
|
|
|
def write_metadata(self, metadata: GenericMetadata, style: int) -> bool:
|
|
retcode = False
|
|
if style == MetaDataStyle.CIX:
|
|
retcode = self.write_cix(metadata)
|
|
if style == MetaDataStyle.CBI:
|
|
retcode = self.write_cbi(metadata)
|
|
if style == MetaDataStyle.COMET:
|
|
retcode = self.write_comet(metadata)
|
|
return retcode
|
|
|
|
def has_metadata(self, style: int) -> bool:
|
|
if style == MetaDataStyle.CIX:
|
|
return self.has_cix()
|
|
if style == MetaDataStyle.CBI:
|
|
return self.has_cbi()
|
|
if style == MetaDataStyle.COMET:
|
|
return self.has_comet()
|
|
return False
|
|
|
|
def remove_metadata(self, style: int) -> bool:
|
|
retcode = True
|
|
if style == MetaDataStyle.CIX:
|
|
retcode = self.remove_cix()
|
|
elif style == MetaDataStyle.CBI:
|
|
retcode = self.remove_cbi()
|
|
elif style == MetaDataStyle.COMET:
|
|
retcode = self.remove_co_met()
|
|
return retcode
|
|
|
|
def get_page(self, index: int) -> bytes:
|
|
image_data = b""
|
|
|
|
filename = self.get_page_name(index)
|
|
|
|
if filename:
|
|
try:
|
|
image_data = self.archiver.read_file(filename) or b""
|
|
except Exception:
|
|
logger.error("Error reading in page %d. Substituting logo page.", index)
|
|
image_data = ComicArchive.logo_data
|
|
|
|
return image_data
|
|
|
|
def get_page_name(self, index: int) -> str:
|
|
if index is None:
|
|
return ""
|
|
|
|
page_list = self.get_page_name_list()
|
|
|
|
num_pages = len(page_list)
|
|
if num_pages == 0 or index >= num_pages:
|
|
return ""
|
|
|
|
return page_list[index]
|
|
|
|
def get_scanner_page_index(self) -> int | None:
|
|
scanner_page_index = None
|
|
|
|
# make a guess at the scanner page
|
|
name_list = self.get_page_name_list()
|
|
count = self.get_number_of_pages()
|
|
|
|
# too few pages to really know
|
|
if count < 5:
|
|
return None
|
|
|
|
# count the length of every filename, and count occurrences
|
|
length_buckets: dict[int, int] = {}
|
|
for name in name_list:
|
|
fname = os.path.split(name)[1]
|
|
length = len(fname)
|
|
if length in length_buckets:
|
|
length_buckets[length] += 1
|
|
else:
|
|
length_buckets[length] = 1
|
|
|
|
# sort by most common
|
|
sorted_buckets = sorted(length_buckets.items(), key=lambda tup: (tup[1], tup[0]), reverse=True)
|
|
|
|
# statistical mode occurrence is first
|
|
mode_length = sorted_buckets[0][0]
|
|
|
|
# we are only going to consider the final image file:
|
|
final_name = os.path.split(name_list[count - 1])[1]
|
|
|
|
common_length_list = []
|
|
for name in name_list:
|
|
if len(os.path.split(name)[1]) == mode_length:
|
|
common_length_list.append(os.path.split(name)[1])
|
|
|
|
prefix = os.path.commonprefix(common_length_list)
|
|
|
|
if mode_length <= 7 and prefix == "":
|
|
# probably all numbers
|
|
if len(final_name) > mode_length:
|
|
scanner_page_index = count - 1
|
|
|
|
# see if the last page doesn't start with the same prefix as most others
|
|
elif not final_name.startswith(prefix):
|
|
scanner_page_index = count - 1
|
|
|
|
return scanner_page_index
|
|
|
|
def get_page_name_list(self, sort_list: bool = True) -> list[str]:
|
|
if not self.page_list:
|
|
# get the list file names in the archive, and sort
|
|
files: list[str] = self.archiver.get_filename_list()
|
|
|
|
# seems like some archive creators are on Windows, and don't know about case-sensitivity!
|
|
if sort_list:
|
|
|
|
files = cast(list[str], natsort.os_sorted(files))
|
|
|
|
# make a sub-list of image files
|
|
self.page_list = []
|
|
for name in files:
|
|
if (
|
|
os.path.splitext(name)[1].casefold() in [".jpg", ".jpeg", ".png", ".gif", ".webp"]
|
|
and os.path.basename(name)[0] != "."
|
|
):
|
|
self.page_list.append(name)
|
|
|
|
return self.page_list
|
|
|
|
def get_number_of_pages(self) -> int:
|
|
if self.page_count is None:
|
|
self.page_count = len(self.get_page_name_list())
|
|
return self.page_count
|
|
|
|
def read_cbi(self) -> GenericMetadata:
|
|
if self.cbi_md is None:
|
|
raw_cbi = self.read_raw_cbi()
|
|
if raw_cbi:
|
|
self.cbi_md = ComicBookInfo().metadata_from_string(raw_cbi)
|
|
else:
|
|
self.cbi_md = GenericMetadata()
|
|
|
|
self.cbi_md.set_default_page_list(self.get_number_of_pages())
|
|
|
|
return self.cbi_md
|
|
|
|
def read_raw_cbi(self) -> str:
|
|
if not self.has_cbi():
|
|
return ""
|
|
|
|
return self.archiver.get_comment()
|
|
|
|
def has_cbi(self) -> bool:
|
|
if self._has_cbi is None:
|
|
if not self.seems_to_be_a_comic_archive():
|
|
self._has_cbi = False
|
|
else:
|
|
comment = self.archiver.get_comment()
|
|
self._has_cbi = ComicBookInfo().validate_string(comment)
|
|
|
|
return self._has_cbi
|
|
|
|
def write_cbi(self, metadata: GenericMetadata) -> bool:
|
|
if metadata is not None:
|
|
try:
|
|
self.apply_archive_info_to_metadata(metadata)
|
|
cbi_string = ComicBookInfo().string_from_metadata(metadata)
|
|
write_success = self.archiver.set_comment(cbi_string)
|
|
if write_success:
|
|
self._has_cbi = True
|
|
self.cbi_md = metadata
|
|
self.reset_cache()
|
|
return write_success
|
|
except Exception as e:
|
|
logger.error("Error saving CBI! for %s: %s", self.path, e)
|
|
|
|
return False
|
|
|
|
def remove_cbi(self) -> bool:
|
|
if self.has_cbi():
|
|
write_success = self.archiver.set_comment("")
|
|
if write_success:
|
|
self._has_cbi = False
|
|
self.cbi_md = None
|
|
self.reset_cache()
|
|
return write_success
|
|
return True
|
|
|
|
def read_cix(self) -> GenericMetadata:
|
|
if self.cix_md is None:
|
|
raw_cix = self.read_raw_cix()
|
|
if raw_cix:
|
|
self.cix_md = ComicInfoXml().metadata_from_string(raw_cix)
|
|
else:
|
|
self.cix_md = GenericMetadata()
|
|
|
|
# validate the existing page list (make sure count is correct)
|
|
if len(self.cix_md.pages) != 0:
|
|
if len(self.cix_md.pages) != self.get_number_of_pages():
|
|
# pages array doesn't match the actual number of images we're seeing
|
|
# in the archive, so discard the data
|
|
self.cix_md.pages = []
|
|
|
|
if len(self.cix_md.pages) == 0:
|
|
self.cix_md.set_default_page_list(self.get_number_of_pages())
|
|
|
|
return self.cix_md
|
|
|
|
def read_raw_cix(self) -> bytes:
|
|
if not self.has_cix():
|
|
return b""
|
|
try:
|
|
raw_cix = self.archiver.read_file(self.ci_xml_filename) or b""
|
|
except Exception as e:
|
|
logger.error("Error reading in raw CIX! for %s: %s", self.path, e)
|
|
raw_cix = b""
|
|
return raw_cix
|
|
|
|
def write_cix(self, metadata: GenericMetadata) -> bool:
|
|
if metadata is not None:
|
|
try:
|
|
self.apply_archive_info_to_metadata(metadata, calc_page_sizes=True)
|
|
raw_cix = self.read_raw_cix()
|
|
cix_string = ComicInfoXml().string_from_metadata(metadata, xml=raw_cix)
|
|
write_success = self.archiver.write_file(self.ci_xml_filename, cix_string.encode("utf-8"))
|
|
if write_success:
|
|
self._has_cix = True
|
|
self.cix_md = metadata
|
|
self.reset_cache()
|
|
return write_success
|
|
except Exception as e:
|
|
logger.error("Error saving CIX! for %s: %s", self.path, e)
|
|
|
|
return False
|
|
|
|
def remove_cix(self) -> bool:
|
|
if self.has_cix():
|
|
write_success = self.archiver.remove_file(self.ci_xml_filename)
|
|
if write_success:
|
|
self._has_cix = False
|
|
self.cix_md = None
|
|
self.reset_cache()
|
|
return write_success
|
|
return True
|
|
|
|
def has_cix(self) -> bool:
|
|
if self._has_cix is None:
|
|
|
|
if not self.seems_to_be_a_comic_archive():
|
|
self._has_cix = False
|
|
elif self.ci_xml_filename in self.archiver.get_filename_list():
|
|
self._has_cix = True
|
|
else:
|
|
self._has_cix = False
|
|
return self._has_cix
|
|
|
|
def read_comet(self) -> GenericMetadata:
|
|
if self.comet_md is None:
|
|
raw_comet = self.read_raw_comet()
|
|
if raw_comet is None or raw_comet == "":
|
|
self.comet_md = GenericMetadata()
|
|
else:
|
|
self.comet_md = CoMet().metadata_from_string(raw_comet)
|
|
|
|
self.comet_md.set_default_page_list(self.get_number_of_pages())
|
|
# use the coverImage value from the comet_data to mark the cover in this struct
|
|
# walk through list of images in file, and find the matching one for md.coverImage
|
|
# need to remove the existing one in the default
|
|
if self.comet_md.cover_image is not None:
|
|
cover_idx = 0
|
|
for idx, f in enumerate(self.get_page_name_list()):
|
|
if self.comet_md.cover_image == f:
|
|
cover_idx = idx
|
|
break
|
|
if cover_idx != 0:
|
|
del self.comet_md.pages[0]["Type"]
|
|
self.comet_md.pages[cover_idx]["Type"] = PageType.FrontCover
|
|
|
|
return self.comet_md
|
|
|
|
def read_raw_comet(self) -> str:
|
|
raw_comet = ""
|
|
if not self.has_comet():
|
|
raw_comet = ""
|
|
else:
|
|
try:
|
|
raw_bytes = self.archiver.read_file(cast(str, self.comet_filename))
|
|
if raw_bytes:
|
|
raw_comet = raw_bytes.decode("utf-8")
|
|
except OSError as e:
|
|
logger.exception("Error reading in raw CoMet!: %s", e)
|
|
return raw_comet
|
|
|
|
def write_comet(self, metadata: GenericMetadata) -> bool:
|
|
|
|
if metadata is not None:
|
|
if not self.has_comet():
|
|
self.comet_filename = self.comet_default_filename
|
|
|
|
self.apply_archive_info_to_metadata(metadata)
|
|
# Set the coverImage value, if it's not the first page
|
|
cover_idx = int(metadata.get_cover_page_index_list()[0])
|
|
if cover_idx != 0:
|
|
metadata.cover_image = self.get_page_name(cover_idx)
|
|
|
|
comet_string = CoMet().string_from_metadata(metadata)
|
|
write_success = self.archiver.write_file(cast(str, self.comet_filename), comet_string.encode("utf-8"))
|
|
if write_success:
|
|
self._has_comet = True
|
|
self.comet_md = metadata
|
|
self.reset_cache()
|
|
return write_success
|
|
|
|
return False
|
|
|
|
def remove_co_met(self) -> bool:
|
|
if self.has_comet():
|
|
write_success = self.archiver.remove_file(cast(str, self.comet_filename))
|
|
if write_success:
|
|
self._has_comet = False
|
|
self.comet_md = None
|
|
self.reset_cache()
|
|
return write_success
|
|
return True
|
|
|
|
def has_comet(self) -> bool:
|
|
if self._has_comet is None:
|
|
self._has_comet = False
|
|
if not self.seems_to_be_a_comic_archive():
|
|
return self._has_comet
|
|
|
|
# look at all xml files in root, and search for CoMet data, get first
|
|
for n in self.archiver.get_filename_list():
|
|
if os.path.dirname(n) == "" and os.path.splitext(n)[1].casefold() == ".xml":
|
|
# read in XML file, and validate it
|
|
data = ""
|
|
try:
|
|
d = self.archiver.read_file(n)
|
|
if d:
|
|
data = d.decode("utf-8")
|
|
except Exception as e:
|
|
logger.warning("Error reading in Comet XML for validation! from %s: %s", self.path, e)
|
|
if CoMet().validate_string(data):
|
|
# since we found it, save it!
|
|
self.comet_filename = n
|
|
self._has_comet = True
|
|
break
|
|
|
|
return self._has_comet
|
|
|
|
def apply_archive_info_to_metadata(self, md: GenericMetadata, calc_page_sizes: bool = False) -> None:
|
|
md.page_count = self.get_number_of_pages()
|
|
|
|
if calc_page_sizes:
|
|
for index, p in enumerate(md.pages):
|
|
idx = int(p["Image"])
|
|
if pil_available:
|
|
if "ImageSize" not in p or "ImageHeight" not in p or "ImageWidth" not in p:
|
|
data = self.get_page(idx)
|
|
if data:
|
|
try:
|
|
if isinstance(data, bytes):
|
|
im = Image.open(io.BytesIO(data))
|
|
else:
|
|
im = Image.open(io.StringIO(data))
|
|
w, h = im.size
|
|
|
|
p["ImageSize"] = str(len(data))
|
|
p["ImageHeight"] = str(h)
|
|
p["ImageWidth"] = str(w)
|
|
except Exception as e:
|
|
logger.warning("Error decoding image [%s] %s :: image %s", e, self.path, index)
|
|
p["ImageSize"] = str(len(data))
|
|
|
|
else:
|
|
if "ImageSize" not in p:
|
|
data = self.get_page(idx)
|
|
p["ImageSize"] = str(len(data))
|
|
|
|
def metadata_from_filename(
|
|
self,
|
|
complicated_parser: bool = False,
|
|
remove_c2c: bool = False,
|
|
remove_fcbd: bool = False,
|
|
remove_publisher: bool = False,
|
|
split_words: bool = False,
|
|
) -> GenericMetadata:
|
|
|
|
metadata = GenericMetadata()
|
|
|
|
filename = self.path.name
|
|
if split_words:
|
|
filename = " ".join(wordninja.split(self.path.stem)) + self.path.suffix
|
|
|
|
if complicated_parser:
|
|
lex = filenamelexer.Lex(filename)
|
|
p = filenameparser.Parse(
|
|
lex.items, remove_c2c=remove_c2c, remove_fcbd=remove_fcbd, remove_publisher=remove_publisher
|
|
)
|
|
metadata.alternate_number = utils.xlate(p.filename_info["alternate"])
|
|
metadata.issue = utils.xlate(p.filename_info["issue"])
|
|
metadata.issue_count = utils.xlate(p.filename_info["issue_count"])
|
|
metadata.publisher = utils.xlate(p.filename_info["publisher"])
|
|
metadata.series = utils.xlate(p.filename_info["series"])
|
|
metadata.title = utils.xlate(p.filename_info["title"])
|
|
metadata.volume = utils.xlate(p.filename_info["volume"])
|
|
metadata.volume_count = utils.xlate(p.filename_info["volume_count"])
|
|
metadata.year = utils.xlate(p.filename_info["year"])
|
|
|
|
metadata.scan_info = utils.xlate(p.filename_info["remainder"])
|
|
metadata.format = "FCBD" if p.filename_info["fcbd"] else None
|
|
if p.filename_info["annual"]:
|
|
metadata.format = "Annual"
|
|
else:
|
|
fnp = filenameparser.FileNameParser()
|
|
fnp.parse_filename(filename)
|
|
|
|
if fnp.issue:
|
|
metadata.issue = fnp.issue
|
|
if fnp.series:
|
|
metadata.series = fnp.series
|
|
if fnp.volume:
|
|
metadata.volume = utils.xlate(fnp.volume, True)
|
|
if fnp.year:
|
|
metadata.year = utils.xlate(fnp.year, True)
|
|
if fnp.issue_count:
|
|
metadata.issue_count = utils.xlate(fnp.issue_count, True)
|
|
if fnp.remainder:
|
|
metadata.scan_info = fnp.remainder
|
|
|
|
metadata.is_empty = False
|
|
|
|
return metadata
|
|
|
|
def export_as_zip(self, zip_filename: pathlib.Path | str) -> bool:
|
|
if self.archive_type == self.ArchiveType.Zip:
|
|
# nothing to do, we're already a zip
|
|
return True
|
|
|
|
zip_archiver = ZipArchiver(zip_filename)
|
|
return zip_archiver.copy_from_archive(self.archiver)
|