Update Database to use new table fromat and twitch api
This commit is contained in:
+85
-105
@@ -2,22 +2,17 @@
|
||||
import sqlite3
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
class Database:
|
||||
def __init__(self, table: str):
|
||||
self.table = table
|
||||
self.columns = ""
|
||||
def __init__(self):
|
||||
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"
|
||||
|
||||
# IMPORTANT Must check if database.db exists before connecting to it.
|
||||
file_exists = Path("database.db").is_file()
|
||||
|
||||
self.CONN = sqlite3.connect("database.db")
|
||||
self.CURSOR = self.CONN.cursor()
|
||||
|
||||
if self.table == "videos":
|
||||
self.columns = "id, date, title, gamename, downloaded, uploaded_yt, chats_upload_yt"
|
||||
elif self.table == "clips":
|
||||
self.columns = "slug, date, title, gamename, clip_by, view_count, downloaded, uploaded_yt, shorts_upload_yt"
|
||||
self.conn = sqlite3.connect("database.db")
|
||||
self.cursor = self.conn.cursor()
|
||||
|
||||
if not file_exists:
|
||||
self.create_database()
|
||||
@@ -30,123 +25,108 @@ class Database:
|
||||
pass
|
||||
|
||||
def create_database(self):
|
||||
"""Creates a table structured explicitly for Twitch clips properties."""
|
||||
self.CURSOR.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS clips (
|
||||
slug TEXT PRIMARY KEY,
|
||||
date TEXT NOT NULL,
|
||||
title TEXT NOT NULL,
|
||||
gamename TEXT NOT NULL,
|
||||
clip_by TEXT NOT NULL,
|
||||
view_count INTEGER NOT NULL,
|
||||
downloaded INTEGER NOT NULL,
|
||||
uploaded_yt INTEGER NOT NULL,
|
||||
shorts_upload_yt INTEGER NOT NULL
|
||||
)
|
||||
"""
|
||||
)
|
||||
|
||||
"""Creates a table structured explicitly for Twitch videos properties."""
|
||||
self.CURSOR.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS videos (
|
||||
id INTEGER PRIMARY KEY,
|
||||
date TEXT NOT NULL,
|
||||
title TEXT NOT NULL,
|
||||
gamename TEXT NOT NULL,
|
||||
downloaded INTEGER NOT NULL,
|
||||
uploaded_yt INTEGER NOT NULL,
|
||||
chats_upload_yt INTEGER NOT NULL
|
||||
)
|
||||
"""
|
||||
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 INTEGER 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()
|
||||
""")
|
||||
|
||||
self.conn.commit()
|
||||
|
||||
def close_database(self):
|
||||
"""Commit before we close."""
|
||||
if self.CONN:
|
||||
self.CONN.commit()
|
||||
self.CONN.close()
|
||||
if self.conn:
|
||||
self.conn.commit()
|
||||
self.conn.close()
|
||||
|
||||
def __mark_as(self, record_id: int, set_sql: str):
|
||||
id = "id"
|
||||
|
||||
if self.table == "clips":
|
||||
id = "slug"
|
||||
def __mark_as(self, record_id: str, set_sql: str):
|
||||
self.cursor.execute(
|
||||
f"UPDATE twitch_videos SET {set_sql} = 1 WHERE id = ?",
|
||||
(record_id,))
|
||||
self.conn.commit()
|
||||
|
||||
self.CURSOR.execute(
|
||||
f"UPDATE {self.table} SET {set_sql} = 1 WHERE {id} = ?",
|
||||
(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_shorts_chats(self, record_id: int):
|
||||
"""Flags a specific row record to uploaded (1)."""
|
||||
|
||||
set_sql = "chats_upload_yt"
|
||||
if self.table == "clips":
|
||||
set_sql = "shorts_uploaded_yt"
|
||||
|
||||
self.__mark_as(record_id, set_sql)
|
||||
|
||||
def mark_as_uploaded(self, record_id: int):
|
||||
"""Flags a specific row record to uploaded (1)."""
|
||||
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: int):
|
||||
"""Updates the downloaded status to True (1) for a specific record ID."""
|
||||
|
||||
def mark_as_downloaded(self, record_id: str):
|
||||
"""Flags a specific row record to downloaded (1)."""
|
||||
self.__mark_as(record_id, "downloaded")
|
||||
|
||||
def get_unuploaded_shorts_chats(self):
|
||||
"""Retrieves all clip rows remaining to be chat uploaded."""
|
||||
def __get_unuploaded(self, set_sql: 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_sql} = 0")
|
||||
return self.cursor.fetchall()
|
||||
|
||||
where_sql = "chats_upload_yt"
|
||||
if self.table == "clips":
|
||||
where_sql = "shorts_uploaded_yt"
|
||||
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")
|
||||
|
||||
self.CURSOR.execute(f"SELECT {self.columns} FROM {self.table} WHERE downloaded = 1 AND uploaded_yt = 1 AND {where_sql} = 0")
|
||||
return self.CURSOR.fetchall()
|
||||
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):
|
||||
"""Retrieves all clip rows remaining to be uploaded."""
|
||||
self.CURSOR.execute(f"SELECT {self.columns} FROM {self.table} WHERE downloaded = 1 AND uploaded_yt = 0")
|
||||
return self.CURSOR.fetchall()
|
||||
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):
|
||||
"""Retrieves all rows where downloaded status is False (0)."""
|
||||
self.CURSOR.execute(f"SELECT {self.columns} FROM {self.table} WHERE downloaded = 0")
|
||||
return self.CURSOR.fetchall()
|
||||
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 insert_videos_record(self, record_id: int, record_date_str: str, title: str, gamename: str):
|
||||
"""Inserts a record with ID, full datetime string, gamename, and title."""
|
||||
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:
|
||||
# Parses into a Python datetime object, then drops timezone info to create a clean string
|
||||
dt_obj = datetime.strptime(record_date_str, "%Y-%m-%dT%H:%M:%SZ")
|
||||
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 = record_date_str
|
||||
clean_datetime = created_at
|
||||
|
||||
self.CURSOR.execute(
|
||||
f"INSERT OR IGNORE INTO {self.table} ({self.columns}) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
(record_id, str(clean_datetime), title, gamename, False, False, False),
|
||||
# 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
|
||||
)
|
||||
self.CONN.commit()
|
||||
|
||||
def insert_clips_record(self, slug: str, record_date_str: str, title: str, gamename: str, clip_by: str, views: int):
|
||||
"""Cleans up ISO-8601 strings into full datetime structures for the database."""
|
||||
try:
|
||||
# Parses into a Python datetime object, then drops timezone info to create a clean string
|
||||
dt_obj = datetime.strptime(record_date_str, "%Y-%m-%dT%H:%M:%SZ")
|
||||
clean_datetime = dt_obj.strftime("%Y-%m-%d %H:%M:%S")
|
||||
except (ValueError, TypeError):
|
||||
clean_datetime = record_date_str
|
||||
|
||||
self.CURSOR.execute(
|
||||
f"INSERT OR IGNORE INTO {self.table} ({self.columns}) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||
(slug, str(clean_datetime), title, gamename, clip_by, views, False, False, False),
|
||||
)
|
||||
self.CONN.commit()
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user