SumGuyV5/python_scripts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update Database to use new table fromat and twitch api

Browse Source
This commit is contained in:
SumGuyV5
2026-07-23 15:32:40 +00:00
parent 81e76b25d0
commit c8d12469e3
2 changed files with 100 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database.py
+83 -103
View File
@@ -2,22 +2,17 @@
import sqlite3 import sqlite3
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from typing import Any
class Database: class Database:
def __init__(self, table: str): def __init__(self):
self.table = table 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.columns = ""
# IMPORTANT Must check if database.db exists before connecting to it. # IMPORTANT Must check if database.db exists before connecting to it.
file_exists = Path("database.db").is_file() file_exists = Path("database.db").is_file()
self.CONN = sqlite3.connect("database.db") self.conn = sqlite3.connect("database.db")
self.CURSOR = self.CONN.cursor() 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"
if not file_exists: if not file_exists:
self.create_database() self.create_database()
@@ -30,123 +25,108 @@ class Database:
pass pass
def create_database(self): 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.""" """Creates a table structured explicitly for Twitch videos properties."""
self.CURSOR.execute( self.cursor.execute("""
""" CREATE TABLE IF NOT EXISTS twitch_videos (
CREATE TABLE IF NOT EXISTS videos ( id TEXT PRIMARY KEY,
id INTEGER PRIMARY KEY,
date TEXT NOT NULL,
title TEXT NOT NULL, title TEXT NOT NULL,
gamename TEXT NOT NULL, created_at TEXT NOT NULL,
downloaded INTEGER NOT NULL, view_count INTEGER NOT NULL,
uploaded_yt INTEGER NOT NULL, duration TEXT NOT NULL,
chats_upload_yt INTEGER 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): def close_database(self):
"""Commit before we close.""" """Commit before we close."""
if self.CONN: if self.conn:
self.CONN.commit() self.conn.commit()
self.CONN.close() self.conn.close()
def __mark_as(self, record_id: int, set_sql: str): def __mark_as(self, record_id: str, set_sql: str):
id = "id" self.cursor.execute(
f"UPDATE twitch_videos SET {set_sql} = 1 WHERE id = ?",
(record_id,))
self.conn.commit()
if self.table == "clips": def mark_as_uploaded_shorts(self, record_id: str):
id = "slug" """Flags a specific row record to uploaded_yt_shorts (1)."""
self.__mark_as(record_id, "uploaded_yt_shorts")
self.CURSOR.execute( def mark_as_uploaded_chats(self, record_id: str):
f"UPDATE {self.table} SET {set_sql} = 1 WHERE {id} = ?", """Flags a specific row record to uploaded_yt_chats (1)."""
(record_id,) self.__mark_as(record_id, "uploaded_yt_chats")
)
self.CONN.commit()
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_yt(self, record_id: str):
"""Flags a specific row record to uploaded_yt (1)."""
self.__mark_as(record_id, "uploaded_yt") self.__mark_as(record_id, "uploaded_yt")
def mark_as_downloaded(self, record_id: int): def mark_as_downloaded(self, record_id: str):
"""Updates the downloaded status to True (1) for a specific record ID.""" """Flags a specific row record to downloaded (1)."""
self.__mark_as(record_id, "downloaded") self.__mark_as(record_id, "downloaded")
def get_unuploaded_shorts_chats(self): def __get_unuploaded(self, set_sql: str) -> list[Any]:
"""Retrieves all clip rows remaining to be chat uploaded.""" """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" def get_unuploaded_shorts(self) -> list[Any]:
if self.table == "clips": """Retrieve all rows that were download but not uploaded_yt_shorts"""
where_sql = "shorts_uploaded_yt" 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") def get_unuploaded_chats(self) -> list[Any]:
return self.CURSOR.fetchall() """Retrieve all rows that were download but not uploaded_yt_chats"""
return self.__get_unuploaded("uploaded_yt_chats")
def get_unuploaded(self): def get_unuploaded(self) -> list[Any]:
"""Retrieves all clip rows remaining to be uploaded.""" """Retrieve all rows that were download but not uploaded_yt"""
self.CURSOR.execute(f"SELECT {self.columns} FROM {self.table} WHERE downloaded = 1 AND uploaded_yt = 0") return self.__get_unuploaded("uploaded_yt")
return self.CURSOR.fetchall()
def get_undownloaded(self): def get_undownloaded(self) -> list[Any]:
"""Retrieves all rows where downloaded status is False (0).""" """Retrieves all rows that were not downloaded"""
self.CURSOR.execute(f"SELECT {self.columns} FROM {self.table} WHERE downloaded = 0") self.cursor.execute(f"SELECT {self.columns} FROM twitch_videos WHERE downloaded = 0")
return self.CURSOR.fetchall() return self.cursor.fetchall()
def insert_videos_record(self, record_id: int, record_date_str: str, title: str, gamename: str): def insert_video_record(
"""Inserts a record with ID, full datetime string, gamename, and title.""" 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: try:
# Parses into a Python datetime object, then drops timezone info to create a clean string dt_obj = datetime.strptime(created_at, "%Y-%m-%dT%H:%M:%SZ")
dt_obj = datetime.strptime(record_date_str, "%Y-%m-%dT%H:%M:%SZ")
clean_datetime = dt_obj.strftime("%Y-%m-%d %H:%M:%S") clean_datetime = dt_obj.strftime("%Y-%m-%d %H:%M:%S")
except (ValueError, TypeError): except (ValueError, TypeError):
clean_datetime = record_date_str clean_datetime = created_at
self.CURSOR.execute( # Explicitly defining columns removes the security risk and column-count bug
f"INSERT OR IGNORE INTO {self.table} ({self.columns}) VALUES (?, ?, ?, ?, ?, ?, ?)", query = """
(record_id, str(clean_datetime), title, gamename, False, False, False), 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: try:
# Parses into a Python datetime object, then drops timezone info to create a clean string self.cursor.execute(query, values)
dt_obj = datetime.strptime(record_date_str, "%Y-%m-%dT%H:%M:%SZ") self.conn.commit()
clean_datetime = dt_obj.strftime("%Y-%m-%d %H:%M:%S") except Exception as e:
except (ValueError, TypeError): # Prevent silent failures if the database connection drops
clean_datetime = record_date_str print(f"Database insertion failed: {e}")
self.conn.rollback()
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()
twitch_vod.py
+14 -8
View File
@@ -1,6 +1,7 @@
import asyncio import asyncio
import json import json
import requests import requests
import database
from twitchAPI.twitch import Twitch from twitchAPI.twitch import Twitch
from twitchAPI.helper import first from twitchAPI.helper import first
# Import the explicit VideoType Enum to prevent the AttributeError # Import the explicit VideoType Enum to prevent the AttributeError
@@ -93,20 +94,19 @@ async def get_streamer_vods():
# FIXED: Resolving game category using the automatic stream markers # FIXED: Resolving game category using the automatic stream markers
game_id, game_name = await get_vod_game_name(v.id) game_id, game_name = await get_vod_game_name(v.id)
#print(f"{v}")
vod_data = { vod_data = {
"id": v.id, "id": v.id,
"title": v.title, "title": v.title,
"created_at": str(v.published_at), "created_at": str(v.published_at),
"view_count": v.view_count, "view_count": int(v.view_count),
"duration": v.duration, "duration": v.duration,
"url": v.url, "url": v.url,
"thumbnail_url": v.thumbnail_url, "thumbnail_url": v.thumbnail_url,
"game_id": game_id, "game_id": int(game_id),
"game_name": game_name, "game_name": game_name,
"stream_id": v.stream_id, "stream_id": str(v.stream_id) if v.stream_id else "0",
"creator_name": CHANNEL_NAME "creator_name": CHANNEL_NAME,
"clip_is": False,
} }
all_vods.append(vod_data) all_vods.append(vod_data)
print(f"Collected VOD: {v.title} | Category: {game_name} ({v.duration})") print(f"Collected VOD: {v.title} | Category: {game_name} ({v.duration})")
@@ -131,14 +131,15 @@ async def get_streamer_clips():
"id": c.id, "id": c.id,
"title": c.title, "title": c.title,
"created_at": str(c.created_at), "created_at": str(c.created_at),
"view_count": c.view_count, "view_count": int(c.view_count),
"duration": c.duration, "duration": c.duration,
"url": c.url, "url": c.url,
"thumbnail_url": c.thumbnail_url, "thumbnail_url": c.thumbnail_url,
"game_id": c.game_id, "game_id": int(c.game_id),
"game_name": game_name, "game_name": game_name,
"stream_id": "0", "stream_id": "0",
"creator_name": c.creator_name, "creator_name": c.creator_name,
"clip_is": True,
} }
all_clips.append(clip_data) all_clips.append(clip_data)
print(f"Collected clip: {c.title} | Category: {game_name} ({c.view_count} views)") print(f"Collected clip: {c.title} | Category: {game_name} ({c.view_count} views)")
@@ -155,8 +156,13 @@ async def main():
#if clips_list: #if clips_list:
# print(f"Top clip: '{clips_list[0]['title']}' (Game: {clips_list[0]['game_name']})") # print(f"Top clip: '{clips_list[0]['title']}' (Game: {clips_list[0]['game_name']})")
db = database.Database()
# 2. Pull VODs (without categories) # 2. Pull VODs (without categories)
vods_list = await get_streamer_vods() 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().") #print(f"\nSuccessfully received a list of {len(vods_list)} VODs in main().")
if vods_list: if vods_list:
print(f"Recent VOD: '{vods_list[0]['title']}'") print(f"Recent VOD: '{vods_list[0]['title']}'")