Mutable and Immutable CardDB variants based on click options

fftcg/__init__.py

@@ -1,7 +1,7 @@
 from .book import Book
-from .carddb import CardDB
+from .carddb import CardDB, RWCardDB
 from .language import Language
 from .opus import Opus
 from .ttsdeck import TTSDeck
 
-__all__ = ["Book", "CardDB", "Language", "Opus", "TTSDeck"]
+__all__ = ["Book", "CardDB", "RWCardDB", "Language", "Opus", "TTSDeck"]

fftcg/carddb.py

@@ -1,93 +1,121 @@
 from __future__ import annotations
 
+import io
 import json
 import pickle
 import zipfile
+from os import PathLike
+from typing import IO
+
+import requests
 
 from .card import Card
 from .cards import Cards
 from .code import Code
 from .language import API_LANGS
-from .utils import CARDDB_FILE_NAME
 
 
 class CardDB:
-    __instance: CardDB = None
+    _instance: CardDB = None
-    __cards: dict[Code, Card]
+    _cards: dict[Code, Card]
-    __face_to_url: dict[str, str]
+    _face_to_url: dict[str, str]
 
-    __DB_FILE_NAME = "cards.pickle"
+    _DB_FILE_NAME = "cards.pickle"
-    __MAPPING_FILE_NAME = "face_to_url.json"
+    _MAPPING_FILE_NAME = "face_to_url.json"
 
-    def __new__(cls) -> CardDB:
+    def __new__(cls, *more) -> CardDB:
-        if CardDB.__instance is None:
+        if CardDB._instance is None:
-            CardDB.__instance = object.__new__(cls)
+            CardDB._instance = object.__new__(CardDB)
-            CardDB.__instance.__cards = {}
-            CardDB.__instance.__face_to_url = {}
 
-        return CardDB.__instance
+        return CardDB._instance
+
+    def __init__(self, db_url: str = None):
+        if db_url is not None:
+            res = requests.get(db_url, stream=True)
+            if not res.ok:
+                raise ValueError("Invalid URL given to CardDB!")
+
+            self._load(io.BytesIO(res.content))
+
+    def _load(self, db: PathLike[str] | IO[bytes]):
+        try:
+            # unpickle db file
+            with zipfile.ZipFile(db, "r") as zip_file:
+                # cards db
+                with zip_file.open(CardDB._DB_FILE_NAME, "r") as file:
+                    self._cards = pickle.load(file)
+
+                # face_to_url mapping
+                with zip_file.open(CardDB._MAPPING_FILE_NAME, "r") as file:
+                    self._face_to_url = json.load(file)
+
+        except FileNotFoundError:
+            self._cards = {}
+            self._face_to_url = {}
 
     def __contains__(self, item: Code) -> bool:
-        return item in self.__cards
+        return item in self._cards
 
     def __getitem__(self, code: Code) -> Card:
-        return self.__cards[code]
+        return self._cards[code]
 
     def get_face_url(self, face: str) -> str:
-        if face in self.__face_to_url:
+        if face in self._face_to_url:
-            return self.__face_to_url[face]
+            return self._face_to_url[face]
         else:
             return face
 
-    def __pickle(self) -> None:
+    def save(self) -> None:
-        with zipfile.ZipFile(CARDDB_FILE_NAME, "w", compression=zipfile.ZIP_LZMA) as zip_file:
+        return
+
+    def update(self, cards: Cards) -> None:
+        return
+
+    def upload_prompt(self) -> None:
+        return
+
+
+class RWCardDB(CardDB):
+    __db_path: PathLike[str]
+
+    def __new__(cls, *more) -> RWCardDB:
+        if CardDB._instance is None:
+            CardDB._instance = object.__new__(RWCardDB)
+
+        return CardDB._instance
+
+    def __init__(self, db_path: PathLike[str] = None):
+        super().__init__(None)
+
+        if db_path is not None:
+            self.__db_path = db_path
+            self._load(self.__db_path)
+
+    def save(self) -> None:
+        with zipfile.ZipFile(self.__db_path, "w", compression=zipfile.ZIP_LZMA) as zip_file:
             # cards db
-            with zip_file.open(CardDB.__DB_FILE_NAME, "w") as file:
+            with zip_file.open(CardDB._DB_FILE_NAME, "w") as file:
-                pickle.dump(self.__cards, file)
+                pickle.dump(self._cards, file)
 
             # face_to_url mapping
-            with zip_file.open(CardDB.__MAPPING_FILE_NAME, "w") as file:
+            with zip_file.open(CardDB._MAPPING_FILE_NAME, "w") as file:
-                file.write(json.dumps(self.__face_to_url, indent=2).encode("utf-8"))
+                file.write(json.dumps(self._face_to_url, indent=2).encode("utf-8"))
-
-    def __unpickle(self) -> None:
-        # unpickle db file
-        self.__cards.clear()
-        self.__face_to_url.clear()
-        try:
-            with zipfile.ZipFile(CARDDB_FILE_NAME, "r") as zip_file:
-                # cards db
-                with zip_file.open(CardDB.__DB_FILE_NAME, "r") as file:
-                    self.__cards |= pickle.load(file)
-
-                # face_to_url mapping
-                with zip_file.open(CardDB.__MAPPING_FILE_NAME, "r") as file:
-                    self.__face_to_url |= json.load(file)
-
-        except FileNotFoundError:
-            pass
-
-    def load(self) -> None:
-        self.__unpickle()
 
     def update(self, cards: Cards) -> None:
         for card in cards:
-            self.__cards[card.code] = card
+            self._cards[card.code] = card
-
-        self.__pickle()
 
     def upload_prompt(self) -> None:
         faces = list(set([
             card[lang].face
-            for card in self.__cards.values()
+            for card in self._cards.values()
             for lang in API_LANGS
             if card[lang].face
         ]))
         faces.sort()
 
         for face in faces:
-            if face not in self.__face_to_url:
+            if face not in self._face_to_url:
                 face_url = input(f"Upload '{face}' and paste URL: ")
                 if face_url:
-                    self.__face_to_url[face] = face_url
+                    self._face_to_url[face] = face_url
-
-        self.__pickle()

fftcg/utils.py

@@ -10,7 +10,6 @@ GRID = Grid((10, 7))  # default in TTsim: 10 columns, 7 rows
 RESOLUTION = Grid((429, 600))  # default in TTsim: 480x670 pixels per card
 DECKS_DIR_NAME = "decks"  # name of decks directory
 IMAGES_DIR_NAME = "images"  # name of images directory
-CARDDB_FILE_NAME = "carddb.zip"  # name of card db file
 # card back URL (image by Aurik)
 CARD_BACK_URL = "http://cloud-3.steamusercontent.com/ugc/948455238665576576/85063172B8C340602E8D6C783A457122F53F7843/"
 

fftcgtool.py

@@ -8,13 +8,8 @@ import click
 
 import fftcg
 
-# constants
-OUT_DIR_NAME = "out"  # name of output directory
-
 
 class LanguageParamType(click.ParamType):
-    name = "lang"
-
     def convert(self, value, param, ctx) -> fftcg.Language:
         if isinstance(value, fftcg.Language):
             return value
@@ -38,23 +33,53 @@ LANGUAGE = LanguageParamType()
     type=LANGUAGE,
     default="en",
     help="language for imported objects",
+    metavar="LANG",
 )
 @click.option(
-    "-s", "--stdout",
+    "-z", "--zip",
-    is_flag=True,
+    type=click.File("wb"),
-    help="print the deck files in a zip archive to stdout, skip creating JSONs on disk",
+    help="wrap deck files into a zip archive, skip creating individual JSONs",
+    metavar="FILE",
+)
+@click.option(
+    "-o", "--output",
+    type=click.Path(
+        allow_dash=False,
+        dir_okay=True,
+        file_okay=False,
+    ),
+    help="use specified output directory instead of ./out",
+    default="out",
+    metavar="DIR",
+)
+@click.option(
+    "-u", "--db-url",
+    type=str,
+    help="load immutable CardDB from URL instead of local, overrides -f",
+    metavar="URL",
+)
+@click.option(
+    "-f", "--db-file",
+    type=click.Path(
+        allow_dash=False,
+        dir_okay=False,
+        file_okay=True,
+    ),
+    default="carddb.zip",
+    help="use specified CardDB file instead of ./out/carddb.zip",
+    metavar="FILE",
 )
 @click.pass_context
-def main(ctx, verbose, language, stdout) -> None:
+def main(ctx, **kwargs) -> None:
     """Imports FFTCG cards for TT-Sim."""
 
     ctx.ensure_object(dict)
-    ctx.obj['LANG'] = language
+    ctx.obj["language"] = kwargs["language"]
 
     # set up logging
-    if verbose == 0:
+    if kwargs["verbose"] == 0:
         verbose = logging.WARN
-    elif verbose == 1:
+    elif kwargs["verbose"] == 1:
         verbose = logging.INFO
     else:
         verbose = logging.DEBUG
@@ -66,31 +91,39 @@ def main(ctx, verbose, language, stdout) -> None:
 
     logger = logging.getLogger(__name__)
     logger.info("fftcgtool started.")
-    logger.debug(f"args: {verbose = }, {language = }, {stdout = }")
+    logger.debug(f"{kwargs = }")
 
     # output directory
-    if not os.path.exists(OUT_DIR_NAME):
+    if not os.path.exists(kwargs["output"]):
-        os.mkdir(OUT_DIR_NAME)
+        os.mkdir(kwargs["output"])
 
-    os.chdir(OUT_DIR_NAME)
+    os.chdir(kwargs["output"])
 
     # load the current carddb
-    carddb = fftcg.CardDB()
+    if kwargs["db_url"] is not None:
-    carddb.load()
+        try:
+            fftcg.CardDB(kwargs["db_url"])
+
+        except (ValueError, KeyError, zipfile.BadZipFile) as cause:
+            logger.critical(f"Couldn't initialize CardDB: {cause}")
+            sys.exit(1)
+
+    else:
+        fftcg.RWCardDB(kwargs["db_file"])
 
 
 @main.command()
 @click.option(
-    "-n", "--num_requests",
+    "-n", "--num-requests",
     type=int,
     default=20,
     help="maximum number of concurrent requests",
 )
 @click.argument(
-    "opus_ids",
+    "opus-ids",
     nargs=-1,
     type=str,
-    metavar="[OPUS_ID] ...",
+    metavar="[OPUS-ID] ...",
 )
 @click.pass_context
 def opuses(ctx, opus_ids, num_requests) -> list[fftcg.TTSDeck]:
@@ -101,7 +134,7 @@ def opuses(ctx, opus_ids, num_requests) -> list[fftcg.TTSDeck]:
     """
 
     ctx.ensure_object(dict)
-    language = ctx.obj['LANG'] or fftcg.Language("")
+    language = ctx.obj["language"] or fftcg.Language("")
 
     carddb = fftcg.CardDB()
     decks: list[fftcg.TTSDeck] = []
@@ -115,6 +148,7 @@ def opuses(ctx, opus_ids, num_requests) -> list[fftcg.TTSDeck]:
         decks.extend(opus.elemental_decks)
 
     carddb.upload_prompt()
+    carddb.save()
 
     # create elemental decks for opus
     return decks
@@ -122,10 +156,10 @@ def opuses(ctx, opus_ids, num_requests) -> list[fftcg.TTSDeck]:
 
 @main.command()
 @click.argument(
-    "deck_ids",
+    "deck-ids",
     nargs=-1,
     type=str,
-    metavar="[DECK_ID] ...",
+    metavar="[DECK-ID] ...",
 )
 def ffdecks(deck_ids) -> list[fftcg.TTSDeck]:
     """
@@ -134,7 +168,6 @@ def ffdecks(deck_ids) -> list[fftcg.TTSDeck]:
     DECK_ID: each of the Decks to import
     """
 
-    print(f"{deck_ids = }")
     decks: list[fftcg.TTSDeck] = []
     for deck_id in deck_ids:
         # import a deck
@@ -144,23 +177,20 @@ def ffdecks(deck_ids) -> list[fftcg.TTSDeck]:
 
 
 @main.result_callback()
-def process_decks(decks: list[fftcg.TTSDeck], verbose, language, stdout):
+def finalize(decks: list[fftcg.TTSDeck], **kwargs):
-    # arg needed because it's in this group
-    int(verbose)
-
     # decide what to do with the decks
-    if stdout:
+    if kwargs["zip"] is not None:
-        # print out a zip file
+        if decks:
-        with open(sys.stdout.fileno(), "wb", closefd=False, buffering=0) as raw_stdout:
+            # create zip file
-            with zipfile.ZipFile(raw_stdout, "w", compression=zipfile.ZIP_DEFLATED) as zip_file:
+            with zipfile.ZipFile(kwargs["zip"], "w", compression=zipfile.ZIP_DEFLATED) as zip_file:
                 # put the decks into that zip file
                 for deck in decks:
-                    zip_file.writestr(deck.file_name, deck.get_json(language))
+                    zip_file.writestr(deck.file_name, deck.get_json(kwargs["language"]))
 
     else:
         # save the decks to disk
         for deck in decks:
-            deck.save(language)
+            deck.save(kwargs["language"])
 
         # bye
         print("Done. Put the generated JSON files in your 'Saved Objects' Folder.")