comictagger/comictaggerlib/cli.py

#!/usr/bin/python
"""ComicTagger CLI functions"""
#
# Copyright 2013 ComicTagger Authors
#
# 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 logging
import os
import sys
from datetime import datetime

from comicapi import utils
from comicapi.comicarchive import ComicArchive, MetaDataStyle
from comicapi.genericmetadata import GenericMetadata
from comictaggerlib import ctversion
from comictaggerlib.cbltransformer import CBLTransformer
from comictaggerlib.ctsettings import ct_ns
from comictaggerlib.filerenamer import FileRenamer, get_rename_dir
from comictaggerlib.graphics import graphics_path
from comictaggerlib.issueidentifier import IssueIdentifier
from comictaggerlib.resulttypes import MultipleMatch, OnlineMatchResults
from comictalker.comictalker import ComicTalker, TalkerError
from comictalker.talker_utils import cleanup_html

logger = logging.getLogger(__name__)


class CLI:
    def __init__(self, config: ct_ns, talkers: dict[str, ComicTalker]) -> None:
        self.config = config
        self.talkers = talkers
        self.batch_mode = False

    def current_talker(self) -> ComicTalker:
        if self.config.Sources_source in self.talkers:
            return self.talkers[self.config.Sources_source]
        logger.error("Could not find the '%s' talker", self.config.Sources_source)
        raise SystemExit(2)

    def actual_issue_data_fetch(self, issue_id: str) -> GenericMetadata:
        # now get the particular issue data
        try:
            ct_md = self.current_talker().fetch_comic_data(issue_id)
        except TalkerError as e:
            logger.exception(f"Error retrieving issue details. Save aborted.\n{e}")
            return GenericMetadata()

        if self.config.Comic_Book_Lover_apply_transform_on_import:
            ct_md = CBLTransformer(ct_md, self.config).apply()

        return ct_md

    def actual_metadata_save(self, ca: ComicArchive, md: GenericMetadata) -> bool:
        if not self.config.Runtime_Options_dryrun:
            for metadata_style in self.config.Runtime_Options_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 self.config.Runtime_Options_quiet:
                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(self, label: str, match_set: MultipleMatch) -> 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, 1):
            print(
                "    {}. {} #{} [{}] ({}/{}) - {}".format(
                    counter,
                    m["series"],
                    m["issue_number"],
                    m["publisher"],
                    m["month"],
                    m["year"],
                    m["issue_title"],
                )
            )
        if self.config.Runtime_Options_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 = self.create_local_metadata(ca)
                ct_md = self.actual_issue_data_fetch(match_set.matches[int(i) - 1]["issue_id"])
                if self.config.Issue_Identifier_clear_metadata_on_import:
                    md = ct_md
                else:
                    notes = (
                        f"Tagged with ComicTagger {ctversion.version} using info from {self.current_talker().name} on"
                        f" {datetime.now():%Y-%m-%d %H:%M:%S}.  [Issue ID {ct_md.issue_id}]"
                    )
                    md.overlay(ct_md.replace(notes=utils.combine_notes(md.notes, notes, "Tagged with ComicTagger")))

                if self.config.Issue_Identifier_auto_imprint:
                    md.fix_publisher()

                self.actual_metadata_save(ca, md)

    def post_process_matches(self, match_results: OnlineMatchResults) -> None:
        # now go through the match results
        if self.config.Runtime_Options_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 self.config.Runtime_Options_summary and not self.config.Runtime_Options_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:
                self.display_match_set_for_choice("Multiple high-confidence matches", match_set)

        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"

                self.display_match_set_for_choice(label, match_set)

    def run(self) -> None:
        if len(self.config.Runtime_Options_files) < 1:
            logger.error("You must specify at least one filename.  Use the -h option for more info")
            return

        match_results = OnlineMatchResults()
        self.batch_mode = len(self.config.Runtime_Options_files) > 1

        for f in self.config.Runtime_Options_files:
            self.process_file_cli(f, match_results)
            sys.stdout.flush()

        self.post_process_matches(match_results)

        if self.config.Runtime_Options_online:
            print(
                f"\nFiles tagged with metadata provided by {self.current_talker().name} {self.current_talker().website}"
            )

    def create_local_metadata(self, ca: ComicArchive) -> GenericMetadata:
        md = GenericMetadata()
        md.set_default_page_list(ca.get_number_of_pages())

        # now, overlay the parsed filename info
        if self.config.Runtime_Options_parse_filename:
            f_md = ca.metadata_from_filename(
                self.config.Filename_Parsing_complicated_parser,
                self.config.Filename_Parsing_remove_c2c,
                self.config.Filename_Parsing_remove_fcbd,
                self.config.Filename_Parsing_remove_publisher,
                self.config.Runtime_Options_split_words,
            )

            md.overlay(f_md)

        for metadata_style in self.config.Runtime_Options_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(self.config.Runtime_Options_metadata)

        return md

    def print(self, ca: ComicArchive) -> None:
        if not self.config.Runtime_Options_type:
            page_count = ca.get_number_of_pages()

            brief = ""

            if self.batch_mode:
                brief = f"{ca.path}: "

            brief += ca.archiver.name() + " archive "

            brief += f"({page_count: >3} pages)"
            brief += "  tags:[ "

            if not (
                ca.has_metadata(MetaDataStyle.CBI)
                or ca.has_metadata(MetaDataStyle.CIX)
                or ca.has_metadata(MetaDataStyle.COMET)
            ):
                brief += "none "
            else:
                if ca.has_metadata(MetaDataStyle.CBI):
                    brief += "CBL "
                if ca.has_metadata(MetaDataStyle.CIX):
                    brief += "CR "
                if ca.has_metadata(MetaDataStyle.COMET):
                    brief += "CoMet "
            brief += "]"

            print(brief)

        if self.config.Runtime_Options_quiet:
            return

        print()

        raw: str | bytes = ""
        if not self.config.Runtime_Options_type or MetaDataStyle.CIX in self.config.Runtime_Options_type:
            if ca.has_metadata(MetaDataStyle.CIX):
                print("--------- ComicRack tags ---------")
                try:
                    if self.config.Runtime_Options_raw:
                        raw = ca.read_raw_cix()
                        if isinstance(raw, bytes):
                            raw = raw.decode("utf-8")
                        print(raw)
                    else:
                        print(ca.read_cix())
                except Exception as e:
                    logger.error("Failed to load metadata for %s: %s", ca.path, e)

        if not self.config.Runtime_Options_type or MetaDataStyle.CBI in self.config.Runtime_Options_type:
            if ca.has_metadata(MetaDataStyle.CBI):
                print("------- ComicBookLover tags -------")
                try:
                    if self.config.Runtime_Options_raw:
                        raw = ca.read_raw_cbi()
                        if isinstance(raw, bytes):
                            raw = raw.decode("utf-8")
                        print(raw)
                    else:
                        print(ca.read_cbi())
                except Exception as e:
                    logger.error("Failed to load metadata for %s: %s", ca.path, e)

        if not self.config.Runtime_Options_type or MetaDataStyle.COMET in self.config.Runtime_Options_type:
            if ca.has_metadata(MetaDataStyle.COMET):
                print("----------- CoMet tags -----------")
                try:
                    if self.config.Runtime_Options_raw:
                        raw = ca.read_raw_comet()
                        if isinstance(raw, bytes):
                            raw = raw.decode("utf-8")
                        print(raw)
                    else:
                        print(ca.read_comet())
                except Exception as e:
                    logger.error("Failed to load metadata for %s: %s", ca.path, e)

    def delete(self, ca: ComicArchive) -> None:
        for metadata_style in self.config.Runtime_Options_type:
            style_name = MetaDataStyle.name[metadata_style]
            if ca.has_metadata(metadata_style):
                if not self.config.Runtime_Options_dryrun:
                    if not ca.remove_metadata(metadata_style):
                        print(f"{ca.path}: Tag removal seemed to fail!")
                    else:
                        print(f"{ca.path}: Removed {style_name} tags.")
                else:
                    print(f"{ca.path}: dry-run. {style_name} tags not removed")
            else:
                print(f"{ca.path}: This archive doesn't have {style_name} tags to remove.")

    def copy(self, ca: ComicArchive) -> None:
        for metadata_style in self.config.Runtime_Options_type:
            dst_style_name = MetaDataStyle.name[metadata_style]
            if not self.config.Runtime_Options_overwrite and ca.has_metadata(metadata_style):
                print(f"{ca.path}: Already has {dst_style_name} tags. Not overwriting.")
                return
            if self.config.Commands_copy == metadata_style:
                print(f"{ca.path}: Destination and source are same: {dst_style_name}. Nothing to do.")
                return

            src_style_name = MetaDataStyle.name[self.config.Commands_copy]
            if ca.has_metadata(self.config.Commands_copy):
                if not self.config.Runtime_Options_dryrun:
                    try:
                        md = ca.read_metadata(self.config.Commands_copy)
                    except Exception as e:
                        md = GenericMetadata()
                        logger.error("Failed to load metadata for %s: %s", ca.path, e)

                    if self.config.Comic_Book_Lover_apply_transform_on_bulk_operation == MetaDataStyle.CBI:
                        md = CBLTransformer(md, self.config).apply()

                    if not ca.write_metadata(md, metadata_style):
                        print(f"{ca.path}: Tag copy seemed to fail!")
                    else:
                        print(f"{ca.path}: Copied {src_style_name} tags to {dst_style_name}.")
                else:
                    print(f"{ca.path}: dry-run.  {src_style_name} tags not copied")
            else:
                print(f"{ca.path}: This archive doesn't have {src_style_name} tags to copy.")

    def save(self, ca: ComicArchive, match_results: OnlineMatchResults) -> None:
        if not self.config.Runtime_Options_overwrite:
            for metadata_style in self.config.Runtime_Options_type:
                if ca.has_metadata(metadata_style):
                    print(f"{ca.path}: Already has {MetaDataStyle.name[metadata_style]} tags. Not overwriting.")
                    return

        if self.batch_mode:
            print(f"Processing {ca.path}...")

        md = self.create_local_metadata(ca)
        if md.issue is None or md.issue == "":
            if self.config.Auto_Tag_assume_1_if_no_issue_num:
                md.issue = "1"

        # now, search online
        if self.config.Runtime_Options_online:
            if self.config.Runtime_Options_issue_id is not None:
                # we were given the actual issue ID to search with
                try:
                    ct_md = self.current_talker().fetch_comic_data(self.config.Runtime_Options_issue_id)
                except TalkerError as e:
                    logger.exception(f"Error retrieving issue details. Save aborted.\n{e}")
                    match_results.fetch_data_failures.append(str(ca.path.absolute()))
                    return

                if ct_md is None:
                    logger.error("No match for ID %s was found.", self.config.Runtime_Options_issue_id)
                    match_results.no_matches.append(str(ca.path.absolute()))
                    return

                if self.config.Comic_Book_Lover_apply_transform_on_import:
                    ct_md = CBLTransformer(ct_md, self.config).apply()
            else:
                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

                ii = IssueIdentifier(ca, self.config, self.current_talker())

                def myoutput(text: str) -> None:
                    if self.config.Runtime_Options_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.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 self.config.Runtime_Options_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
                ct_md = self.actual_issue_data_fetch(matches[0]["issue_id"])
                if ct_md.is_empty:
                    match_results.fetch_data_failures.append(str(ca.path.absolute()))
                    return

            if self.config.Issue_Identifier_clear_metadata_on_import:
                md = GenericMetadata()

            notes = (
                f"Tagged with ComicTagger {ctversion.version} using info from {self.current_talker().name} on"
                f" {datetime.now():%Y-%m-%d %H:%M:%S}.  [Issue ID {ct_md.issue_id}]"
            )
            md.overlay(
                ct_md.replace(
                    notes=utils.combine_notes(md.notes, notes, "Tagged with ComicTagger"),
                    description=cleanup_html(ct_md.description, self.config.Sources_remove_html_tables),
                )
            )

            if self.config.Issue_Identifier_auto_imprint:
                md.fix_publisher()

        # ok, done building our metadata. time to save
        if not self.actual_metadata_save(ca, md):
            match_results.write_failures.append(str(ca.path.absolute()))
        else:
            match_results.good_matches.append(str(ca.path.absolute()))

    def rename(self, ca: ComicArchive) -> None:
        original_path = ca.path
        msg_hdr = ""
        if self.batch_mode:
            msg_hdr = f"{ca.path}: "

        md = self.create_local_metadata(ca)

        if md.series is None:
            logger.error(msg_hdr + "Can't rename without series name")
            return

        new_ext = ""  # default
        if self.config.File_Rename_set_extension_based_on_archive:
            new_ext = ca.extension()

        renamer = FileRenamer(
            md,
            platform="universal" if self.config.File_Rename_strict else "auto",
            replacements=self.config.File_Rename_replacements,
        )
        renamer.set_template(self.config.File_Rename_template)
        renamer.set_issue_zero_padding(self.config.File_Rename_issue_number_padding)
        renamer.set_smart_cleanup(self.config.File_Rename_use_smart_string_cleanup)
        renamer.move = self.config.File_Rename_move_to_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",
                self.config.File_Rename_template,
            )
            return
        except Exception:
            logger.exception("Formatter failure: %s metadata: %s", self.config.File_Rename_template, renamer.metadata)
            return

        folder = get_rename_dir(ca, self.config.File_Rename_dir if self.config.File_Rename_move_to_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 self.config.Runtime_Options_dryrun:
            # rename the file
            try:
                ca.rename(utils.unique_file(full_path))
            except OSError:
                logger.exception("Failed to rename comic archive: %s", ca.path)
        else:
            suffix = " (dry-run, no change)"

        print(f"renamed '{original_path.name}' -> '{new_name}' {suffix}")

    def export(self, ca: ComicArchive) -> None:
        msg_hdr = ""
        if self.batch_mode:
            msg_hdr = f"{ca.path}: "

        if ca.is_zip():
            logger.error(msg_hdr + "Archive is already a zip file.")
            return

        filename_path = ca.path
        new_file = filename_path.with_suffix(".cbz")

        if self.config.Runtime_Options_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 self.config.Runtime_Options_dryrun:
            if ca.export_as_zip(new_file):
                export_success = True
                if self.config.Runtime_Options_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 self.config.Runtime_Options_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 self.config.Runtime_Options_delete_after_zip_export and delete_success:
                msg += " (Original deleted) "
        else:
            msg += "Archive failed to export!"

        print(msg)

    def process_file_cli(self, filename: str, match_results: OnlineMatchResults) -> None:
        if not os.path.lexists(filename):
            logger.error("Cannot find %s", filename)
            return

        ca = ComicArchive(filename, str(graphics_path / "nocover.png"))

        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 (
            self.config.Commands_delete
            or self.config.Commands_copy
            or self.config.Commands_save
            or self.config.Commands_rename
        ):
            logger.error("This archive is not writable")
            return

        if self.config.Commands_print:
            self.print(ca)

        elif self.config.Commands_delete:
            self.delete(ca)

        elif self.config.Commands_copy is not None:
            self.copy(ca)

        elif self.config.Commands_save:
            self.save(ca, match_results)

        elif self.config.Commands_rename:
            self.rename(ca)

        elif self.config.Commands_export_to_zip:
            self.export(ca)