aadeb07c49
Refactor add_to_path with tests Fix type hints for titles_match Use casefold in get_language Fix using the recursive flag in cli mode Add http status code to ComicVine exceptions Fix parenthesis getting removed when renaming Add more tests
583 lines
22 KiB
Python
583 lines
22 KiB
Python
#!/usr/bin/python
|
|
"""ComicTagger CLI functions"""
|
|
#
|
|
# Copyright 2013 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 argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import sys
|
|
from pprint import pprint
|
|
|
|
from comicapi import utils
|
|
from comicapi.comicarchive import ComicArchive, MetaDataStyle
|
|
from comicapi.genericmetadata import GenericMetadata
|
|
from comictaggerlib.cbltransformer import CBLTransformer
|
|
from comictaggerlib.comicvinetalker import ComicVineTalker, ComicVineTalkerException
|
|
from comictaggerlib.filerenamer import FileRenamer, get_rename_dir
|
|
from comictaggerlib.issueidentifier import IssueIdentifier
|
|
from comictaggerlib.resulttypes import IssueResult, MultipleMatch, OnlineMatchResults
|
|
from comictaggerlib.settings import ComicTaggerSettings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def actual_issue_data_fetch(
|
|
match: IssueResult, settings: ComicTaggerSettings, opts: argparse.Namespace
|
|
) -> GenericMetadata:
|
|
# now get the particular issue data
|
|
try:
|
|
comic_vine = ComicVineTalker(settings.id_series_match_search_thresh)
|
|
comic_vine.wait_for_rate_limit = opts.wait_on_cv_rate_limit
|
|
cv_md = comic_vine.fetch_issue_data(match["volume_id"], match["issue_number"], settings)
|
|
except ComicVineTalkerException:
|
|
logger.exception("Network error while getting issue details. Save aborted")
|
|
return GenericMetadata()
|
|
|
|
if settings.apply_cbl_transform_on_cv_import:
|
|
cv_md = CBLTransformer(cv_md, settings).apply()
|
|
|
|
return cv_md
|
|
|
|
|
|
def actual_metadata_save(ca: ComicArchive, opts: argparse.Namespace, md: GenericMetadata) -> bool:
|
|
if not opts.dryrun:
|
|
for metadata_style in opts.type:
|
|
# write out the new data
|
|
if not ca.write_metadata(md, metadata_style):
|
|
logger.error("The tag save seemed to fail for style: %s!", MetaDataStyle.name[metadata_style])
|
|
return False
|
|
|
|
print("Save complete.")
|
|
logger.info("Save complete.")
|
|
else:
|
|
if opts.terse:
|
|
logger.info("dry-run option was set, so nothing was written")
|
|
print("dry-run option was set, so nothing was written")
|
|
else:
|
|
logger.info("dry-run option was set, so nothing was written, but here is the final set of tags:")
|
|
print("dry-run option was set, so nothing was written, but here is the final set of tags:")
|
|
print(f"{md}")
|
|
return True
|
|
|
|
|
|
def display_match_set_for_choice(
|
|
label: str, match_set: MultipleMatch, opts: argparse.Namespace, settings: ComicTaggerSettings
|
|
) -> None:
|
|
print(f"{match_set.ca.path} -- {label}:")
|
|
|
|
# sort match list by year
|
|
match_set.matches.sort(key=lambda k: k["year"] or 0)
|
|
|
|
for (counter, m) in enumerate(match_set.matches):
|
|
counter += 1
|
|
print(
|
|
" {}. {} #{} [{}] ({}/{}) - {}".format(
|
|
counter,
|
|
m["series"],
|
|
m["issue_number"],
|
|
m["publisher"],
|
|
m["month"],
|
|
m["year"],
|
|
m["issue_title"],
|
|
)
|
|
)
|
|
if opts.interactive:
|
|
while True:
|
|
i = input("Choose a match #, or 's' to skip: ")
|
|
if (i.isdigit() and int(i) in range(1, len(match_set.matches) + 1)) or i == "s":
|
|
break
|
|
if i != "s":
|
|
# save the data!
|
|
# we know at this point, that the file is all good to go
|
|
ca = match_set.ca
|
|
md = create_local_metadata(opts, ca, settings)
|
|
cv_md = actual_issue_data_fetch(match_set.matches[int(i) - 1], settings, opts)
|
|
if opts.overwrite:
|
|
md = cv_md
|
|
else:
|
|
md.overlay(cv_md)
|
|
|
|
if opts.auto_imprint:
|
|
md.fix_publisher()
|
|
|
|
actual_metadata_save(ca, opts, md)
|
|
|
|
|
|
def post_process_matches(
|
|
match_results: OnlineMatchResults, opts: argparse.Namespace, settings: ComicTaggerSettings
|
|
) -> None:
|
|
# now go through the match results
|
|
if opts.show_save_summary:
|
|
if len(match_results.good_matches) > 0:
|
|
print("\nSuccessful matches:\n------------------")
|
|
for f in match_results.good_matches:
|
|
print(f)
|
|
|
|
if len(match_results.no_matches) > 0:
|
|
print("\nNo matches:\n------------------")
|
|
for f in match_results.no_matches:
|
|
print(f)
|
|
|
|
if len(match_results.write_failures) > 0:
|
|
print("\nFile Write Failures:\n------------------")
|
|
for f in match_results.write_failures:
|
|
print(f)
|
|
|
|
if len(match_results.fetch_data_failures) > 0:
|
|
print("\nNetwork Data Fetch Failures:\n------------------")
|
|
for f in match_results.fetch_data_failures:
|
|
print(f)
|
|
|
|
if not opts.show_save_summary and not opts.interactive:
|
|
# just quit if we're not interactive or showing the summary
|
|
return
|
|
|
|
if len(match_results.multiple_matches) > 0:
|
|
print("\nArchives with multiple high-confidence matches:\n------------------")
|
|
for match_set in match_results.multiple_matches:
|
|
display_match_set_for_choice("Multiple high-confidence matches", match_set, opts, settings)
|
|
|
|
if len(match_results.low_confidence_matches) > 0:
|
|
print("\nArchives with low-confidence matches:\n------------------")
|
|
for match_set in match_results.low_confidence_matches:
|
|
if len(match_set.matches) == 1:
|
|
label = "Single low-confidence match"
|
|
else:
|
|
label = "Multiple low-confidence matches"
|
|
|
|
display_match_set_for_choice(label, match_set, opts, settings)
|
|
|
|
|
|
def cli_mode(opts: argparse.Namespace, settings: ComicTaggerSettings) -> None:
|
|
if len(opts.file_list) < 1:
|
|
logger.error("You must specify at least one filename. Use the -h option for more info")
|
|
return
|
|
|
|
match_results = OnlineMatchResults()
|
|
|
|
for f in opts.file_list:
|
|
process_file_cli(f, opts, settings, match_results)
|
|
sys.stdout.flush()
|
|
|
|
post_process_matches(match_results, opts, settings)
|
|
|
|
|
|
def create_local_metadata(opts: argparse.Namespace, ca: ComicArchive, settings: ComicTaggerSettings) -> GenericMetadata:
|
|
md = GenericMetadata()
|
|
md.set_default_page_list(ca.get_number_of_pages())
|
|
|
|
# now, overlay the parsed filename info
|
|
if opts.parse_filename:
|
|
f_md = ca.metadata_from_filename(
|
|
settings.complicated_parser,
|
|
settings.remove_c2c,
|
|
settings.remove_fcbd,
|
|
settings.remove_publisher,
|
|
opts.split_words,
|
|
)
|
|
|
|
md.overlay(f_md)
|
|
|
|
for metadata_style in opts.type:
|
|
if ca.has_metadata(metadata_style):
|
|
try:
|
|
t_md = ca.read_metadata(metadata_style)
|
|
md.overlay(t_md)
|
|
break
|
|
except Exception as e:
|
|
logger.error("Failed to load metadata for %s: %s", ca.path, e)
|
|
|
|
# finally, use explicit stuff
|
|
md.overlay(opts.metadata)
|
|
|
|
return md
|
|
|
|
|
|
def process_file_cli(
|
|
filename: str, opts: argparse.Namespace, settings: ComicTaggerSettings, match_results: OnlineMatchResults
|
|
) -> None:
|
|
batch_mode = len(opts.file_list) > 1
|
|
|
|
ca = ComicArchive(filename, settings.rar_exe_path, ComicTaggerSettings.get_graphic("nocover.png"))
|
|
|
|
if not os.path.lexists(filename):
|
|
logger.error("Cannot find %s", filename)
|
|
return
|
|
|
|
if not ca.seems_to_be_a_comic_archive():
|
|
logger.error("Sorry, but %s is not a comic archive!", filename)
|
|
return
|
|
|
|
if not ca.is_writable() and (opts.delete or opts.copy or opts.save or opts.rename):
|
|
logger.error("This archive is not writable")
|
|
return
|
|
|
|
has = [False, False, False]
|
|
if ca.has_cix():
|
|
has[MetaDataStyle.CIX] = True
|
|
if ca.has_cbi():
|
|
has[MetaDataStyle.CBI] = True
|
|
if ca.has_comet():
|
|
has[MetaDataStyle.COMET] = True
|
|
|
|
if opts.print:
|
|
|
|
if not opts.type:
|
|
page_count = ca.get_number_of_pages()
|
|
|
|
brief = ""
|
|
|
|
if batch_mode:
|
|
brief = f"{ca.path}: "
|
|
|
|
if ca.is_sevenzip():
|
|
brief += "7Z archive "
|
|
elif ca.is_zip():
|
|
brief += "ZIP archive "
|
|
elif ca.is_rar():
|
|
brief += "RAR archive "
|
|
elif ca.is_folder():
|
|
brief += "Folder archive "
|
|
|
|
brief += f"({page_count: >3} pages)"
|
|
brief += " tags:[ "
|
|
|
|
if not (has[MetaDataStyle.CBI] or has[MetaDataStyle.CIX] or has[MetaDataStyle.COMET]):
|
|
brief += "none "
|
|
else:
|
|
if has[MetaDataStyle.CBI]:
|
|
brief += "CBL "
|
|
if has[MetaDataStyle.CIX]:
|
|
brief += "CR "
|
|
if has[MetaDataStyle.COMET]:
|
|
brief += "CoMet "
|
|
brief += "]"
|
|
|
|
print(brief)
|
|
|
|
if opts.terse:
|
|
return
|
|
|
|
print()
|
|
|
|
if not opts.type or MetaDataStyle.CIX in opts.type:
|
|
if has[MetaDataStyle.CIX]:
|
|
print("--------- ComicRack tags ---------")
|
|
try:
|
|
if opts.raw:
|
|
print(ca.read_raw_cix())
|
|
else:
|
|
print(ca.read_cix())
|
|
except Exception as e:
|
|
logger.error("Failed to load metadata for %s: %s", ca.path, e)
|
|
|
|
if not opts.type or MetaDataStyle.CBI in opts.type:
|
|
if has[MetaDataStyle.CBI]:
|
|
print("------- ComicBookLover tags -------")
|
|
try:
|
|
if opts.raw:
|
|
pprint(json.loads(ca.read_raw_cbi()))
|
|
else:
|
|
print(ca.read_cbi())
|
|
except Exception as e:
|
|
logger.error("Failed to load metadata for %s: %s", ca.path, e)
|
|
|
|
if not opts.type or MetaDataStyle.COMET in opts.type:
|
|
if has[MetaDataStyle.COMET]:
|
|
print("----------- CoMet tags -----------")
|
|
try:
|
|
if opts.raw:
|
|
print(ca.read_raw_comet())
|
|
else:
|
|
print(ca.read_comet())
|
|
except Exception as e:
|
|
logger.error("Failed to load metadata for %s: %s", ca.path, e)
|
|
|
|
elif opts.delete:
|
|
for metadata_style in opts.type:
|
|
style_name = MetaDataStyle.name[metadata_style]
|
|
if has[metadata_style]:
|
|
if not opts.dryrun:
|
|
if not ca.remove_metadata(metadata_style):
|
|
print(f"{filename}: Tag removal seemed to fail!")
|
|
else:
|
|
print(f"{filename}: Removed {style_name} tags.")
|
|
else:
|
|
print(f"{filename}: dry-run. {style_name} tags not removed")
|
|
else:
|
|
print(f"{filename}: This archive doesn't have {style_name} tags to remove.")
|
|
|
|
elif opts.copy is not None:
|
|
for metadata_style in opts.type:
|
|
dst_style_name = MetaDataStyle.name[metadata_style]
|
|
if opts.no_overwrite and has[metadata_style]:
|
|
print(f"{filename}: Already has {dst_style_name} tags. Not overwriting.")
|
|
return
|
|
if opts.copy == metadata_style:
|
|
print(f"{filename}: Destination and source are same: {dst_style_name}. Nothing to do.")
|
|
return
|
|
|
|
src_style_name = MetaDataStyle.name[opts.copy]
|
|
if has[opts.copy]:
|
|
if not opts.dryrun:
|
|
try:
|
|
md = ca.read_metadata(opts.copy)
|
|
except Exception as e:
|
|
md = GenericMetadata()
|
|
logger.error("Failed to load metadata for %s: %s", ca.path, e)
|
|
|
|
if settings.apply_cbl_transform_on_bulk_operation and metadata_style == MetaDataStyle.CBI:
|
|
md = CBLTransformer(md, settings).apply()
|
|
|
|
if not ca.write_metadata(md, metadata_style):
|
|
print(f"{filename}: Tag copy seemed to fail!")
|
|
else:
|
|
print(f"{filename}: Copied {src_style_name} tags to {dst_style_name}.")
|
|
else:
|
|
print(f"{filename}: dry-run. {src_style_name} tags not copied")
|
|
else:
|
|
print(f"{filename}: This archive doesn't have {src_style_name} tags to copy.")
|
|
|
|
elif opts.save:
|
|
|
|
if opts.no_overwrite:
|
|
for metadata_style in opts.type:
|
|
if has[metadata_style]:
|
|
print(f"{filename}: Already has {MetaDataStyle.name[metadata_style]} tags. Not overwriting.")
|
|
return
|
|
|
|
if batch_mode:
|
|
print(f"Processing {ca.path}...")
|
|
|
|
md = create_local_metadata(opts, ca, settings)
|
|
if md.issue is None or md.issue == "":
|
|
if opts.assume_issue_one:
|
|
md.issue = "1"
|
|
|
|
# now, search online
|
|
if opts.online:
|
|
if opts.issue_id is not None:
|
|
# we were given the actual ID to search with
|
|
try:
|
|
comic_vine = ComicVineTalker(settings.id_series_match_search_thresh)
|
|
comic_vine.wait_for_rate_limit = opts.wait_on_cv_rate_limit
|
|
cv_md = comic_vine.fetch_issue_data_by_issue_id(opts.issue_id, settings)
|
|
except ComicVineTalkerException:
|
|
logger.exception("Network error while getting issue details. Save aborted")
|
|
match_results.fetch_data_failures.append(str(ca.path.absolute()))
|
|
return
|
|
|
|
if cv_md is None:
|
|
logger.error("No match for ID %s was found.", opts.issue_id)
|
|
match_results.no_matches.append(str(ca.path.absolute()))
|
|
return
|
|
|
|
if settings.apply_cbl_transform_on_cv_import:
|
|
cv_md = CBLTransformer(cv_md, settings).apply()
|
|
else:
|
|
ii = IssueIdentifier(ca, settings)
|
|
|
|
if md is None or md.is_empty:
|
|
logger.error("No metadata given to search online with!")
|
|
match_results.no_matches.append(str(ca.path.absolute()))
|
|
return
|
|
|
|
def myoutput(text: str) -> None:
|
|
if opts.verbose:
|
|
IssueIdentifier.default_write_output(text)
|
|
|
|
# use our overlaid MD struct to search
|
|
ii.set_additional_metadata(md)
|
|
ii.only_use_additional_meta_data = True
|
|
ii.wait_and_retry_on_rate_limit = opts.wait_on_cv_rate_limit
|
|
ii.set_output_function(myoutput)
|
|
ii.cover_page_index = md.get_cover_page_index_list()[0]
|
|
matches = ii.search()
|
|
|
|
result = ii.search_result
|
|
|
|
found_match = False
|
|
choices = False
|
|
low_confidence = False
|
|
|
|
if result == ii.result_no_matches:
|
|
pass
|
|
elif result == ii.result_found_match_but_bad_cover_score:
|
|
low_confidence = True
|
|
found_match = True
|
|
elif result == ii.result_found_match_but_not_first_page:
|
|
found_match = True
|
|
elif result == ii.result_multiple_matches_with_bad_image_scores:
|
|
low_confidence = True
|
|
choices = True
|
|
elif result == ii.result_one_good_match:
|
|
found_match = True
|
|
elif result == ii.result_multiple_good_matches:
|
|
choices = True
|
|
|
|
if choices:
|
|
if low_confidence:
|
|
logger.error("Online search: Multiple low confidence matches. Save aborted")
|
|
match_results.low_confidence_matches.append(MultipleMatch(ca, matches))
|
|
return
|
|
|
|
logger.error("Online search: Multiple good matches. Save aborted")
|
|
match_results.multiple_matches.append(MultipleMatch(ca, matches))
|
|
return
|
|
if low_confidence and opts.abort_on_low_confidence:
|
|
logger.error("Online search: Low confidence match. Save aborted")
|
|
match_results.low_confidence_matches.append(MultipleMatch(ca, matches))
|
|
return
|
|
if not found_match:
|
|
logger.error("Online search: No match found. Save aborted")
|
|
match_results.no_matches.append(str(ca.path.absolute()))
|
|
return
|
|
|
|
# we got here, so we have a single match
|
|
|
|
# now get the particular issue data
|
|
cv_md = actual_issue_data_fetch(matches[0], settings, opts)
|
|
if cv_md.is_empty:
|
|
match_results.fetch_data_failures.append(str(ca.path.absolute()))
|
|
return
|
|
|
|
if opts.overwrite:
|
|
md = cv_md
|
|
else:
|
|
md.overlay(cv_md)
|
|
|
|
if opts.auto_imprint:
|
|
md.fix_publisher()
|
|
|
|
# ok, done building our metadata. time to save
|
|
if not actual_metadata_save(ca, opts, md):
|
|
match_results.write_failures.append(str(ca.path.absolute()))
|
|
else:
|
|
match_results.good_matches.append(str(ca.path.absolute()))
|
|
|
|
elif opts.rename:
|
|
|
|
msg_hdr = ""
|
|
if batch_mode:
|
|
msg_hdr = f"{ca.path}: "
|
|
|
|
md = create_local_metadata(opts, ca, settings)
|
|
|
|
if md.series is None:
|
|
logger.error(msg_hdr + "Can't rename without series name")
|
|
return
|
|
|
|
new_ext = "" # default
|
|
if settings.rename_extension_based_on_archive:
|
|
if ca.is_sevenzip():
|
|
new_ext = ".cb7"
|
|
elif ca.is_zip():
|
|
new_ext = ".cbz"
|
|
elif ca.is_rar():
|
|
new_ext = ".cbr"
|
|
|
|
renamer = FileRenamer(md, platform="universal" if settings.rename_strict else "auto")
|
|
renamer.set_template(settings.rename_template)
|
|
renamer.set_issue_zero_padding(settings.rename_issue_number_padding)
|
|
renamer.set_smart_cleanup(settings.rename_use_smart_string_cleanup)
|
|
renamer.move = settings.rename_move_dir
|
|
|
|
try:
|
|
new_name = renamer.determine_name(ext=new_ext)
|
|
except ValueError:
|
|
logger.exception(
|
|
msg_hdr + "Invalid format string!\n"
|
|
"Your rename template is invalid!\n\n"
|
|
"%s\n\n"
|
|
"Please consult the template help in the settings "
|
|
"and the documentation on the format at "
|
|
"https://docs.python.org/3/library/string.html#format-string-syntax",
|
|
settings.rename_template,
|
|
)
|
|
return
|
|
except Exception:
|
|
logger.exception("Formatter failure: %s metadata: %s", settings.rename_template, renamer.metadata)
|
|
|
|
folder = get_rename_dir(ca, settings.rename_dir if settings.rename_move_dir else None)
|
|
|
|
full_path = folder / new_name
|
|
|
|
if full_path == ca.path:
|
|
print(msg_hdr + "Filename is already good!", file=sys.stderr)
|
|
return
|
|
|
|
suffix = ""
|
|
if not opts.dryrun:
|
|
# rename the file
|
|
ca.rename(utils.unique_file(full_path))
|
|
else:
|
|
suffix = " (dry-run, no change)"
|
|
|
|
print(f"renamed '{os.path.basename(ca.path)}' -> '{new_name}' {suffix}")
|
|
|
|
elif opts.export_to_zip:
|
|
msg_hdr = ""
|
|
if batch_mode:
|
|
msg_hdr = f"{ca.path}: "
|
|
|
|
if ca.is_zip():
|
|
logger.error(msg_hdr + "Archive is already a zip file.")
|
|
return
|
|
|
|
filename_path = pathlib.Path(filename).absolute()
|
|
new_file = filename_path.with_suffix(".cbz")
|
|
|
|
if opts.abort_on_conflict and new_file.exists():
|
|
print(msg_hdr + f"{new_file.name} already exists in the that folder.")
|
|
return
|
|
|
|
new_file = utils.unique_file(new_file)
|
|
|
|
delete_success = False
|
|
export_success = False
|
|
if not opts.dryrun:
|
|
if ca.export_as_zip(new_file):
|
|
export_success = True
|
|
if opts.delete_after_zip_export:
|
|
try:
|
|
filename_path.unlink(missing_ok=True)
|
|
delete_success = True
|
|
except OSError:
|
|
logger.exception(msg_hdr + "Error deleting original archive after export")
|
|
delete_success = False
|
|
else:
|
|
# last export failed, so remove the zip, if it exists
|
|
new_file.unlink(missing_ok=True)
|
|
else:
|
|
msg = msg_hdr + f"Dry-run: Would try to create {os.path.split(new_file)[1]}"
|
|
if opts.delete_after_zip_export:
|
|
msg += " and delete original."
|
|
print(msg)
|
|
return
|
|
|
|
msg = msg_hdr
|
|
if export_success:
|
|
msg += f"Archive exported successfully to: {os.path.split(new_file)[1]}"
|
|
if opts.delete_after_zip_export and delete_success:
|
|
msg += " (Original deleted) "
|
|
else:
|
|
msg += "Archive failed to export!"
|
|
|
|
print(msg)
|