diff --git a/database.py b/database.py index f28ef14..6d07697 100644 --- a/database.py +++ b/database.py @@ -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() \ No newline at end of file + 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() diff --git a/twitch_vod.py b/twitch_vod.py index 89d9f27..bd0cfa7 100644 --- a/twitch_vod.py +++ b/twitch_vod.py @@ -1,6 +1,7 @@ import asyncio import json import requests +import database from twitchAPI.twitch import Twitch from twitchAPI.helper import first # Import the explicit VideoType Enum to prevent the AttributeError @@ -86,27 +87,26 @@ async def get_streamer_vods(): print(f"Starting VOD extraction for {USER.display_name}...") vod_generator = TWITCH.get_videos(user_id=USER.id, first=100, video_type=VideoType.ALL) - + all_vods = [] async for v in vod_generator: # FIXED: Resolving game category using the automatic stream markers game_id, game_name = await get_vod_game_name(v.id) - #print(f"{v}") - vod_data = { "id": v.id, "title": v.title, "created_at": str(v.published_at), - "view_count": v.view_count, + "view_count": int(v.view_count), "duration": v.duration, "url": v.url, "thumbnail_url": v.thumbnail_url, - "game_id": game_id, + "game_id": int(game_id), "game_name": game_name, - "stream_id": v.stream_id, - "creator_name": CHANNEL_NAME + "stream_id": str(v.stream_id) if v.stream_id else "0", + "creator_name": CHANNEL_NAME, + "clip_is": False, } all_vods.append(vod_data) print(f"Collected VOD: {v.title} | Category: {game_name} ({v.duration})") @@ -131,14 +131,15 @@ async def get_streamer_clips(): "id": c.id, "title": c.title, "created_at": str(c.created_at), - "view_count": c.view_count, + "view_count": int(c.view_count), "duration": c.duration, "url": c.url, "thumbnail_url": c.thumbnail_url, - "game_id": c.game_id, + "game_id": int(c.game_id), "game_name": game_name, "stream_id": "0", "creator_name": c.creator_name, + "clip_is": True, } all_clips.append(clip_data) print(f"Collected clip: {c.title} | Category: {game_name} ({c.view_count} views)") @@ -155,8 +156,13 @@ async def main(): #if clips_list: # print(f"Top clip: '{clips_list[0]['title']}' (Game: {clips_list[0]['game_name']})") + db = database.Database() # 2. Pull VODs (without categories) vods_list = await get_streamer_vods() + for v in vods_list: + db.insert_video_record(v['id'], v['title'], v['created_at'], v['view_count'], v['duration'], v['url'], v['thumbnail_url'], + v['game_id'], v['game_name'], v['stream_id'], v['creator_name'], v['clip_is']) + #print(f"\nSuccessfully received a list of {len(vods_list)} VODs in main().") if vods_list: print(f"Recent VOD: '{vods_list[0]['title']}'")