Integrate xklb db in Calibre-Web

cps/__init__.py

@@ -123,6 +123,7 @@ def create_app():
     cli_param.init()
 
     ub.init_db(cli_param.settings_path)
+    # xb.init_db(cli_param.xklb_path)
     # pylint: disable=no-member
     encrypt_key, error = config_sql.get_encryption_key(os.path.dirname(cli_param.settings_path))
 

cps/cli.py

@@ -24,7 +24,7 @@
 from .constants import CONFIG_DIR as _CONFIG_DIR
 from .constants import STABLE_VERSION as _STABLE_VERSION
 from .constants import NIGHTLY_VERSION as _NIGHTLY_VERSION
-from .constants import DEFAULT_SETTINGS_FILE, DEFAULT_GDRIVE_FILE
+from .constants import DEFAULT_SETTINGS_FILE, DEFAULT_GDRIVE_FILE, DEFAULT_XKLB_FILE
 
 
 def version_info():
@@ -46,6 +46,7 @@ def __init__(self):
         self.keyfilepath = None
         self.gd_path = None
         self.settings_path = None
+        self.xklb_path = None
         self.logpath = None
 
     def init(self):
@@ -81,13 +82,17 @@ def arg_parser(self):
         self.logpath = args.o or ""
         self.settings_path = args.p or os.path.join(_CONFIG_DIR, DEFAULT_SETTINGS_FILE)
         self.gd_path = args.g or os.path.join(_CONFIG_DIR, DEFAULT_GDRIVE_FILE)
+        self.xklb_path = os.path.join(_CONFIG_DIR, DEFAULT_XKLB_FILE)
 
         if os.path.isdir(self.settings_path):
             self.settings_path = os.path.join(self.settings_path, DEFAULT_SETTINGS_FILE)
 
         if os.path.isdir(self.gd_path):
             self.gd_path = os.path.join(self.gd_path, DEFAULT_GDRIVE_FILE)
 
+        if os.path.isdir(self.xklb_path):
+            self.xklb_path = os.path.join(self.xklb_path, DEFAULT_XKLB_FILE)
+
         # handle and check parameter for ssl encryption
         self.certfilepath = None
         self.keyfilepath = None

cps/constants.py

@@ -67,6 +67,7 @@
 
 DEFAULT_SETTINGS_FILE = "app.db"
 DEFAULT_GDRIVE_FILE = "gdrive.db"
+DEFAULT_XKLB_FILE = "xklb.db"
 
 ROLE_USER               = 0 << 0
 ROLE_ADMIN              = 1 << 0

cps/db.py

@@ -48,6 +48,7 @@
 from flask import flash
 
 from . import logger, ub, isoLanguages
+from .constants import DEFAULT_XKLB_FILE
 from .pagination import Pagination
 from .string_helper import strip_whitespaces
 
@@ -612,6 +613,7 @@ def check_valid_db(cls, config_calibre_dir, app_db_path, config_calibre_uuid):
         if not config_calibre_dir:
             return False, False
         dbpath = os.path.join(config_calibre_dir, "metadata.db")
+        xklb_db_path = os.path.join(config_calibre_dir, DEFAULT_XKLB_FILE)
         if not os.path.exists(dbpath):
             return False, False
         try:
@@ -623,6 +625,7 @@ def check_valid_db(cls, config_calibre_dir, app_db_path, config_calibre_uuid):
             with check_engine.begin() as connection:
                 connection.execute(text("attach database '{}' as calibre;".format(dbpath)))
                 connection.execute(text("attach database '{}' as app_settings;".format(app_db_path)))
+                connection.execute(text("attach database '{}' as xklb;".format(xklb_db_path)))
                 local_session = scoped_session(sessionmaker())
                 local_session.configure(bind=connection)
                 database_uuid = local_session().query(Library_Id).one_or_none()
@@ -651,6 +654,8 @@ def setup_db(cls, config_calibre_dir, app_db_path):
             cls.config.invalidate()
             return None
 
+        xklb_db_path = os.path.join(config_calibre_dir, DEFAULT_XKLB_FILE)
+
         try:
             cls.engine = create_engine('sqlite://',
                                        echo=False,
@@ -660,6 +665,7 @@ def setup_db(cls, config_calibre_dir, app_db_path):
             with cls.engine.begin() as connection:
                 connection.execute(text("attach database '{}' as calibre;".format(dbpath)))
                 connection.execute(text("attach database '{}' as app_settings;".format(app_db_path)))
+                connection.execute(text("attach database '{}' as xklb;".format(xklb_db_path)))
 
             conn = cls.engine.connect()
             # conn.text_factory = lambda b: b.decode(errors = 'ignore') possible fix for #1302

cps/xb.py

@@ -0,0 +1,104 @@
+from . import constants, logger
+from sqlalchemy import create_engine, Column, Integer, String, ForeignKey, Text
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import relationship, sessionmaker, scoped_session
+
+log = logger.create()
+
+Base = declarative_base()
+
+# define the Media table 
+class Media(Base):
+    __tablename__ = 'media'
+    id = Column(Integer, primary_key=True)
+    playlists_id = Column(Integer)
+    size = Column(Integer)
+    duration = Column(Float)
+    time_created = Column(DateTime)
+    time_modified = Column(DateTime)
+    time_deleted = Column(DateTime)
+    time_downloaded = Column(DateTime)
+    fps = Column(Float)
+    view_count = Column(Integer)
+    path = Column(String)
+    webpath = Column(String)
+    extractor_id = Column(String)
+    title = Column(String)
+    uploader = Column(String)
+    live_status = Column(String)
+    error = Column(String)
+    time_uploaded = Column(DateTime)
+    width = Column(Integer)
+    height = Column(Integer)
+    type = Column(String)
+    video_codecs = Column(String)
+    audio_codecs = Column(String)
+    subtitle_codecs = Column(String)
+    video_count = Column(Integer)
+    audio_count = Column(Integer)
+    language = Column(String)
+    subtitle_count = Column(Integer)
+    download_attempts = Column(Integer)
+
+    # Relationships
+    captions = relationship("Caption", back_populates="media")
+
+    def __repr__(self):
+        return f"<Media(title='{self.title}', path='{self.path}')>"
+
+# define the Caption table
+class Caption(Base):
+    __tablename__ = 'captions'
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    media_id = Column(Integer, ForeignKey('media.id'))
+    time = Column(Integer)
+    text = Column(Text)
+
+    # Relationships to Media
+    media = relationship("Media", back_populates="captions")
+
+# define the mapping table for book ids from metadata.db to media ids in xklb.db
+class BookMediaMapping(Base):
+    __tablename__ = 'book_media_map'
+    book_id = Column(Integer, primary_key=True)
+    media_id = Column(Integer, ForeignKey('media.id'))
+    media = relationship("Media")
+
+# create the engine and session maker
+engine = create_engine(constants.XB_DB_PATH)
+Session = scoped_session(sessionmaker(bind=engine))
+Base.metadata.create_all(engine)
+
+# function to map book ids to media ids
+def add_book_media_mapping(book_id, media_id):
+    session = Session()
+
+    try:
+        new_mapping = BookMediaMapping(book_id=book_id, media_id=media_id)
+        session.add(new_mapping)
+        session.commit()
+    except Exception as e:
+        log.error(f"Error adding book-media mapping: {e}")
+        session.rollback()
+    finally:
+        session.close()
+
+# function to get a specific book (which maps to a media entry) for a given caption
+def get_book_for_caption(caption):
+    session = Session()
+    try:
+        media_entry = session.query(Caption).filter(Caption.id == caption).first()
+        if not media_entry:
+            log.error(f"No media found for caption id: {caption}")
+            return None
+
+        book_entry = session.query(BookMediaMapping).filter(BookMediaMapping.media_id == media_entry.media_id).first()
+        if not book_entry:
+            log.error(f"No book mapping found for media id: {media_entry.media_id}")
+            return None
+
+        return book_entry.book_id
+    except Exception as e:
+        log.error(f"Error getting book for caption: {e}")
+    finally:
+        session.close()