fix problem when deleting MediaFile with MediaActor relations

kontor-scripts/check_kontor.py

@@ -1,40 +1,27 @@
 """
 Checks the database kontor
 """
-from dataclasses import dataclass
-from enum import Enum, auto
+
 from logging import Logger
-from typing import Dict, List, Optional
+from pathlib import Path
+from typing import Any, Dict, List, Optional
 from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
 from urllib.parse import urlparse
+import click
+from simple_term_menu import TerminalMenu
 
-from api import Option, OptionType, Server, get_api_config, get_logger
+from api import Server, get_api_config, get_logger
 
 
 parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter)
 parser.add_argument("--verbose", "-v", action="count", default=0)
 parser.add_argument("--config", "-c", default="kontor-api")
 parser.add_argument("--dir", "-d", default="/data/media")
+parser.add_argument("--add-dir", "-a", action="append")
 parser.add_argument("--dry-run", "-m", action="store_true")
 parser.add_argument("--server", "-s")
 args = parser.parse_args()
 
-class StatusType(Enum):
-    UNKNOWN = auto()
-    FILE_NAME = auto()
-    FILE_ID = auto()
-    DUPLICATE = auto()
-    CLOUD_LINK = auto()
-    CLOUD_LINK_ID = auto()
-
-class FileStatus:
-    id: str | None = None
-    status_type: StatusType = StatusType.UNKNOWN
-
-    def get_response(self, response: dict):
-        self.status_type = StatusType.FILE_NAME
-        self.id = response['id']
-
 
 def create_item_id_mapping(log: Logger, data_list: List[dict]) -> Dict[str, dict]:
     """
@@ -47,8 +34,32 @@ def create_item_id_mapping(log: Logger, data_list: List[dict]) -> Dict[str, dict
     return item_id_mapping
 
 
-def check_duplicate_links(log: Logger, server: Server):
-    data = server.request(log=logger, table="media_file")
+def remove_file(log: Logger, item_data: Dict[str, Any], media_dirs: List[str]):
+    """
+    Delete file from path in dictionary.
+    """
+    log.debug(item_data)
+    cloud_link = item_data["cloud_link"]
+    for file_dir in media_dirs:
+        log.info("look in %s", file_dir)
+        file_name = Path(cloud_link).name
+        media_file = Path(file_dir, file_name)
+        if media_file.exists():
+            log.info("File to remove %s", media_file.absolute())
+            media_file.unlink(missing_ok=True)
+            break
+        else:
+            log.info("File not found %s", media_file.absolute())
+
+
+def check_duplicate_links(log: Logger, server: Optional[Server], media_dirs: List[str]):
+    """
+    Check if there are MediaFile URLs which only differ in hostname.
+    """
+    if server is None:
+        log.info("no server selected")
+        return
+    data = server.request(log=log, table="media_file")
     mapping = create_item_id_mapping(log=log, data_list=data)
     visited_link_path: Dict[str, str] = {}
     duplicate_link_paths: Dict[str, List[str]] = {}
@@ -60,7 +71,7 @@ def check_duplicate_links(log: Logger, server: Server):
         parsed_url = urlparse(link)
         link_path = parsed_url.path
         if link_path in visited_link_path:
-            log.info("duplicate url path found: %s", link_path)
+            log.debug("duplicate url path found: %s", link_path)
             if link_path in duplicate_link_paths:
                 duplicate_link_paths[link_path].append(file_id)
             else:
@@ -70,32 +81,42 @@ def check_duplicate_links(log: Logger, server: Server):
         else:
             visited_link_path[link_path] = file_id
     log.info("found %s duplicate links", len(duplicate_link_paths.keys()))
-    deletion_list: List[str] = []
-    for key, value in duplicate_link_paths.items():
-        if len(value) == 2:
-            log.info("%s:\n%s - %s\n%s - %s", key, value[0], mapping[value[0]]["url"], value[1], mapping[value[1]]["url"])
-            if mapping[value[0]]["url"].startswith("https://xhamster"):
-                deletion_list.append(value[0])
-            else:
-                deletion_list.append(value[1])
+    for _, value in duplicate_link_paths.items():
+        choices = [mapping[value[0]]["url"], mapping[value[1]]["url"], "Abbruch"]
+        menu = TerminalMenu(
+            choices, title="Choose an link to delete:", multi_select=False
+        )
+        menu_choice = menu.show()
+        if isinstance(menu_choice, int):
+            if menu_choice == 2:
+                break
+            index: int = int(menu_choice)
+            server.delete(log=log, table="media_file", item_id=value[index])
+            remove_file(log, mapping[value[index]], media_dirs)
         else:
-            log.info("found %s links", len(value))
-    for key in deletion_list:
-        log.info("%s - %s", key, mapping[key]["url"])
+            print("selection canceled")
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     logger = get_logger(args.verbose, args.config)
     logger.info("kontor.check_kontor started")
     APICONFIG = get_api_config(logger, args.config)
-    server: Server = APICONFIG.server[0]
+    first_server: Optional[Server] = APICONFIG.get_server(args.server)
+    if not first_server:
+        SystemExit(2)
+    dirs: List[str] = args.add_dir
+    if dirs is None:
+        dirs = [args.dir]
+    else:
+        dirs.insert(0, args.dir)
+    logger.info(dirs)
     logger.info("kontor.check_kontor.check_duplicate_links")
-    check_duplicate_links(logger, server)
-    #logger.info("kontor.check_kontor.update_cloud_link_with_found_files")
-    #update_cloud_link_with_found_files(data_dir, mariadb_conn, args.dry_run)
-    #logger.info("kontor.check_kontor.get_ids_from_column_cloud_link")
-    #get_ids_from_column_cloud_link(link_list, mariadb_cursor)
-    #logger.info('found {} ids in column cloud_link'.format(len(link_list)))
-    #logger.info("kontor.check_kontor.checking_ids_from_cloud_link")
-    #checking_ids_from_cloud_link(link_list, mariadb_cursor)
+    check_duplicate_links(logger, first_server, dirs)
+    # logger.info("kontor.check_kontor.update_cloud_link_with_found_files")
+    # update_cloud_link_with_found_files(data_dir, mariadb_conn, args.dry_run)
+    # logger.info("kontor.check_kontor.get_ids_from_column_cloud_link")
+    # get_ids_from_column_cloud_link(link_list, mariadb_cursor)
+    # logger.info('found {} ids in column cloud_link'.format(len(link_list)))
+    # logger.info("kontor.check_kontor.checking_ids_from_cloud_link")
+    # checking_ids_from_cloud_link(link_list, mariadb_cursor)
     logger.info("kontor.check_kontor finished")