232 lines
9.1 KiB
Python
Executable File
232 lines
9.1 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"
|
|
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() |