141 lines
5.4 KiB
Python
Executable File
141 lines
5.4 KiB
Python
Executable File
#!/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"
|
|
|
|
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 __del__(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.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 __get_unuploaded(self, set_row: 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")
|
|
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")
|
|
|
|
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")
|
|
|
|
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_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 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 OR IGNORE INTO twitch_videos (
|
|
id, title, created_at, view_count, duration, url, thumbnail_url,
|
|
game_id, game_name, stream_id, creator_name, clip_is
|
|
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
"""
|
|
|
|
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()
|