update database to use sqlalchemy

delete/database.py

@@ -0,0 +1,232 @@
+#!/usr/bin/env python3
+import sqlite3
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+class Database:
+    def __init__(self, db_path: str = "database.db"):
+        self.columns = "id, title, created_at, view_count, duration, url, thumbnail_url, game_id, game_name, stream_id, creator_name, clip_is, downloaded, uploaded_yt, uploaded_yt_chats, uploaded_yt_shorts"
+        self.data = "id, name, date_shorts, date_videos, date_videos_chat"
+
+        file_exists = Path(db_path).is_file()
+
+        self.conn = sqlite3.connect(db_path)
+        self.cursor = self.conn.cursor()
+
+        if not file_exists:
+            self.create_database()
+
+    def __exit__(self):
+        # Destructors are unpredictable in Python; explicitly close when done instead
+        try:
+            self.close_database()
+        except:
+            pass
+
+    def create_database(self):
+        """Creates a table structured explicitly for Twitch videos properties."""
+        self.cursor.execute("""
+        CREATE TABLE IF NOT EXISTS twitch_videos (
+            id TEXT PRIMARY KEY,
+            title TEXT NOT NULL,
+            created_at TEXT NOT NULL,
+            view_count INTEGER NOT NULL,
+            duration TEXT NOT NULL,
+            url TEXT NOT NULL,
+            thumbnail_url TEXT NOT NULL,
+            game_id TEXT NOT NULL,
+            game_name TEXT NOT NULL,
+            stream_id TEXT NOT NULL,
+            creator_name TEXT NOT NULL,
+            clip_is BOOLEAN NOT NULL DEFAULT 0,
+            downloaded BOOLEAN NOT NULL DEFAULT 0,
+            uploaded_yt BOOLEAN NOT NULL DEFAULT 0,
+            uploaded_yt_chats BOOLEAN NOT NULL DEFAULT 0,
+            uploaded_yt_shorts BOOLEAN NOT NULL DEFAULT 0
+        )
+        """)
+
+        self.cursor.execute("""
+        CREATE TABLE IF NOT EXISTS data (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            name TEXT NOT NULL,
+            date_shorts TEXT NOT NULL,
+            date_videos TEXT NOT NULL,
+            date_videos_chat TEXT NOT NULL
+        )
+        """)
+        
+        self.conn.commit()
+
+    def close_database(self):
+        """Commit before we close."""
+        if self.conn:
+            self.conn.commit()
+            self.conn.close()
+
+    def __mark_as(self, record_id: str, set_row: str, mark: str = "1" ):
+        self.cursor.execute(
+            f"UPDATE twitch_videos SET {set_row} = ? WHERE id = ?", 
+            (mark, record_id))
+        self.conn.commit()
+
+    def mark_as_uploaded_shorts(self, record_id: str):
+        """Flags a specific row record to uploaded_yt_shorts (1)."""
+        self.__mark_as(record_id, "uploaded_yt_shorts")
+
+    def mark_as_uploaded_chats(self, record_id: str):
+        """Flags a specific row record to uploaded_yt_chats (1)."""
+        self.__mark_as(record_id, "uploaded_yt_chats")
+
+    def mark_as_uploaded_yt(self, record_id: str):
+        """Flags a specific row record to uploaded_yt (1)."""
+        self.__mark_as(record_id, "uploaded_yt")
+
+    def mark_as_downloaded(self, record_id: str):
+        """Flags a specific row record to downloaded (1)."""
+        self.__mark_as(record_id, "downloaded")
+
+    def unmark_as_download(self, record_id: str):
+        """Flags a specific row record to downloaded (0)."""
+        self.__mark_as(record_id, "downloaded", "0")
+
+    def update_datetime(self, record_id: str, created_at: str):
+        """Updates the Datetime for timezone fix"""
+        self.cursor.execute(
+            f"UPDATE twitch_videos SET created_at = ? WHERE id = ?",
+            (created_at, record_id))
+        self.conn.commit()
+
+    def __get_unuploaded(self, set_row: str, also: str = "") -> list[Any]:
+        """Retrieve all rows that were download but not uploaded"""
+        self.cursor.execute(f"SELECT {self.columns} FROM twitch_videos WHERE downloaded = 1 AND {set_row} = 0 {also}")
+        return self.cursor.fetchall()
+
+    def get_unuploaded_shorts(self) -> list[Any]:
+        """Retrieve all rows that were download but not uploaded_yt_shorts"""
+        return self.__get_unuploaded("uploaded_yt_shorts", "AND clip_is = 1")
+
+    def get_unuploaded_chats(self) -> list[Any]:
+        """Retrieve all rows that were download but not uploaded_yt_chats"""
+        return self.__get_unuploaded("uploaded_yt_chats", "AND clip_is = 0")
+                
+    def get_unuploaded(self) -> list[Any]:
+        """Retrieve all rows that were download but not uploaded_yt"""
+        return self.__get_unuploaded("uploaded_yt")
+
+    def get_undownloaded(self) -> list[Any]:
+        """Retrieves all rows that were not downloaded"""
+        self.cursor.execute(f"SELECT {self.columns} FROM twitch_videos WHERE downloaded = 0")
+        return self.cursor.fetchall()
+
+    def get_download(self) -> list[Any]:
+        """Retrieves all rows that were downloaded"""
+        self.cursor.execute(f"SELECT {self.columns} FROM twitch_videos WHERE downloaded = 1")
+        return self.cursor.fetchall()
+
+    def get_clips(self) -> list[Any]:
+        """Retrieves all rows that are clip_is"""
+        self.cursor.execute(f"SELECT {self.columns} FROM twitch_videos WHERE clip_is = 1")
+        return self.cursor.fetchall()
+
+    def get_vods(self) -> list[Any]:
+        self.cursor.execute(f"SELECT {self.columns} FROM twitch_videos WHERE clip_is = 0")
+        return self.cursor.fetchall()
+
+    def insert_video_record(
+        self, id: str, title: str, created_at: str, view_count: int, duration: str, 
+        url: str, thumbnail_url: str, game_id: int, game_name: str, stream_id: str, 
+        creator_name: str, clip_is: bool
+    ):
+        try:
+            dt_obj = datetime.strptime(created_at, "%Y-%m-%dT%H:%M:%SZ")
+            clean_datetime = dt_obj.strftime("%Y-%m-%d %H:%M:%S")
+        except (ValueError, TypeError):
+            clean_datetime = created_at
+
+        # Explicitly defining columns removes the security risk and column-count bug
+        query = """
+            INSERT INTO twitch_videos (
+                id, title, created_at, view_count, duration, url, thumbnail_url, 
+                game_id, game_name, stream_id, creator_name, clip_is
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(id) DO UPDATE SET
+                title = excluded.title,
+                created_at = excluded.created_at,
+                view_count = excluded.view_count,
+                duration = excluded.duration,
+                url = excluded.url,
+                thumbnail_url = excluded.thumbnail_url,
+                game_id = excluded.game_id,
+                game_name = excluded.game_name,
+                stream_id = excluded.stream_id,
+                creator_name = excluded.creator_name,
+                clip_is = excluded.clip_is
+            WHERE excluded.duration != twitch_videos.duration
+        """
+
+        values = (
+            id, title, clean_datetime, view_count, duration, url, thumbnail_url, 
+            game_id, game_name, stream_id, creator_name, clip_is
+        )
+
+        try:
+            self.cursor.execute(query, values)
+            self.conn.commit()
+        except Exception as e:
+            # Prevent silent failures if the database connection drops
+            print(f"❌ Database insertion failed: {e}")
+            self.conn.rollback()
+
+    def __update_date(self, record_id: str, set_row: str, date: str):
+        """Updates the Datetime for timezone fix"""
+        self.cursor.execute(
+            f"UPDATE data SET {set_row} = ? WHERE id = ?",
+            (date, record_id))
+        self.conn.commit()
+
+    def update_date_shorts(self, record_id: str, date_shorts: str):
+        self.__update_date(record_id, "date_shorts", date_shorts)
+
+    def get_data(self) -> list[Any]:
+        self.cursor.execute(f"SELECT {self.data} FROM data")
+        return self.cursor.fetchall()
+
+    def insert_data_record(self, name: str, date_shorts: str, date_videos: str, date_videos_chat: str):
+        try:
+            dt_obj = datetime.strptime(date_shorts, "%Y-%m-%dT%H:%M:%SZ")
+            clean_date_shorts = dt_obj.strftime("%Y-%m-%d %H:%M:%S")
+        except (ValueError, TypeError):
+            clean_date_shorts = date_shorts
+
+        try:
+            dt_obj = datetime.strptime(date_videos, "%Y-%m-%dT%H:%M:%SZ")
+            clean_date_videos = dt_obj.strftime("%Y-%m-%d %H:%M:%S")
+        except (ValueError, TypeError):
+            clean_date_videos = date_videos
+
+        try:
+            dt_obj = datetime.strptime(date_videos_chat, "%Y-%m-%dT%H:%M:%SZ")
+            clean_date_videos_chat = dt_obj.strftime("%Y-%m-%d %H:%M:%S")
+        except (ValueError, TypeError):
+            clean_date_videos_chat = date_videos_chat
+
+        # Explicitly defining columns removes the security risk and column-count bug
+        query = """
+            INSERT INTO data (
+                name, date_shorts, date_videos, date_videos_chat
+            ) VALUES (?, ?, ?, ?)
+        """
+
+        values = (
+            name, clean_date_shorts, clean_date_videos, clean_date_videos_chat
+        )
+
+        try:
+            self.cursor.execute(query, values)
+            self.conn.commit()
+        except Exception as e:
+            # Prevent silent failures if the database connection drops
+            print(f"❌ Database insertion failed: {e}")
+            self.conn.rollback()