dylan/gander
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

UPDATE: Refactored backend routes to improve readability

Browse Source
This commit is contained in:
Dylan De Faoite
2025-02-05 21:04:05 +00:00
parent 4c79c80f3d
commit a8af24e256
10 changed files with 228 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
web_server/utils/recommendation_utils.py
View File

@@ -2,7 +2,7 @@ from database.database import Database
from typing import Optional, List
def user_recommendation_category(user_id: int) -> Optional[int]:
def get_user_preferred_category(user_id: int) -> Optional[int]:
"""
Queries user_preferences database to find users favourite streaming category and returns the category
"""
@@ -23,9 +23,9 @@ def followed_categories_recommendations(user_id: int) -> Optional[List[dict]]:
"""
with Database() as db:
streams = db.fetchall("""
SELECT stream_id, title, username, num_viewers, category_name
SELECT u.user_id, title, u.username, num_viewers, category_name
FROM streams
JOIN users ON users.user_id = streams.user_id
JOIN users u ON streams.user_id = u.user_id
JOIN categories ON streams.category_id = categories.category_id
WHERE categories.category_id IN (SELECT category_id FROM followed_categories WHERE user_id = ?)
ORDER BY num_viewers DESC
@@ -34,15 +34,15 @@ def followed_categories_recommendations(user_id: int) -> Optional[List[dict]]:
return streams
def recommendations_based_on_category(category_id: int) -> Optional[List[dict]]:
def get_streams_based_on_category(category_id: int) -> Optional[List[dict]]:
"""
Queries stream database to get top 25 most viewed streams based on given category
"""
with Database() as db:
streams = db.fetchall("""
SELECT streams.stream_id, title, username, num_viewers, category_name
SELECT u.user_id, title, username, num_viewers, category_name
FROM streams
JOIN users ON users.user_id = streams.user_id
JOIN users u ON streams.user_id = u.user_id
JOIN categories ON streams.category_id = categories.category_id
WHERE categories.category_id = ?
ORDER BY num_viewers DESC
@@ -51,41 +51,37 @@ def recommendations_based_on_category(category_id: int) -> Optional[List[dict]]:
return streams
def default_recommendations() -> Optional[List[dict]]:
def get_highest_view_streams(no_streams: int) -> Optional[List[dict]]:
"""
Return a list of 25 recommended live streams by number of viewers
Return a list of live streams by number of viewers
"""
with Database() as db:
data = db.fetchall("""
SELECT streams.stream_id, title, username, num_viewers, category_name
SELECT u.user_id, username, title, num_viewers, category_name
FROM streams
JOIN users ON users.user_id = streams.user_id
JOIN users u ON streams.user_id = u.user_id
JOIN categories ON streams.category_id = categories.category_id
WHERE isLive = 1
ORDER BY num_viewers DESC
LIMIT 25;
""")
LIMIT ?;
""", (no_streams,))
return data
def category_recommendations() -> Optional[List[dict]]:
def get_highest_view_categories(no_categories: int) -> Optional[List[dict]]:
"""
Returns a list of the top 5 most popular live categories
Returns a list of top 5 most popular categories
"""
with Database() as db:
categories = db.fetchall("""
SELECT categories.category_id, categories.category_name
SELECT categories.category_id, categories.category_name, SUM(streams.num_viewers) AS total_viewers
FROM streams
JOIN categories ON streams.category_id = categories.category_id
WHERE streams.isLive = 1
GROUP BY categories.category_name
ORDER BY SUM(streams.num_viewers) DESC
LIMIT 5;
""")
LIMIT ?;
""", (no_categories,))
return categories
def user_category_recommendations(user_id: int) -> Optional[List[dict]]:
def get_user_category_recommendations(user_id: int) -> Optional[List[dict]]:
"""
Queries user_preferences database to find users top 5 favourite streaming category and returns the category
"""

131
web_server/utils/stream_utils.py
View File

@@ -3,37 +3,38 @@ from typing import Optional
import sqlite3, os, subprocess
from time import sleep
from typing import Optional, List
from datetime import datetime
def streamer_live_status(user_id: int) -> dict:
def get_streamer_live_status(user_id: int):
"""
Returns boolean on whether the given streamer is live
"""
with Database() as db:
is_live = db.fetchone("""
SELECT isLive
FROM streams
WHERE user_id = ?
ORDER BY stream_id DESC
LIMIT 1;
SELECT is_live
FROM users
WHERE user_id = ?;
""", (user_id,))
return is_live
def followed_live_streams(user_id: int) -> Optional[List[dict]]:
def get_followed_live_streams(user_id: int) -> Optional[List[dict]]:
"""
Searches for streamers who the user followed which are currently live
Returns a list of live streams with the streamer's user id, stream title, and number of viewers
"""
with Database() as db:
live_streams = db.fetchall("""
SELECT user_id, stream_id, title, num_viewers
FROM streams
WHERE user_id IN (SELECT followed_id FROM follows WHERE user_id = ?)
AND stream_id = (SELECT MAX(stream_id) FROM streams WHERE user_id = streams.user_id)
AND isLive = 1;
""", (user_id,))
SELECT users.user_id, streams.title, streams.num_viewers, users.username
FROM streams JOIN users
ON streams.user_id = users.user_id
WHERE users.user_id IN
(SELECT followed_id FROM follows WHERE user_id = ?)
AND users.is_live = 1;
""", (user_id,))
return live_streams
def followed_streamers(user_id: int) -> Optional[List[dict]]:
def get_followed_streamers(user_id: int) -> Optional[List[dict]]:
"""
Returns a list of streamers who the user follows
"""
@@ -45,62 +46,54 @@ def followed_streamers(user_id: int) -> Optional[List[dict]]:
""", (user_id,))
return followed_streamers
def user_stream(user_id: int, stream_id: int) -> dict:
def get_vod(vod_id: int) -> dict:
"""
Returns data of a streamers selected stream
Returns data of a streamers vod
"""
with Database() as db:
stream = db.fetchone("""
SELECT u.username, s.user_id, s.title, s.start_time, s.num_viewers, c.category_name
FROM streams AS s
JOIN categories AS c ON s.category_id = c.category_id
JOIN users AS u ON s.user_id = u.user_id
WHERE u.user_id = ?
AND s.stream_id = ?
""", (user_id, stream_id))
return stream
vod = db.fetchone("""SELECT * FROM vods WHERE vod_id = ?;""", (vod_id,))
return vod
def streamer_most_recent_stream(user_id: int) -> Optional[dict]:
def get_latest_vod(user_id: int):
"""
Returns data of the most recent stream by a streamer
"""
with Database() as db:
most_recent_stream = db.fetchone("""
SELECT s.stream_id, u.username, s.user_id, s.title, s.start_time, s.num_viewers, c.category_name
FROM streams AS s
JOIN categories AS c ON s.category_id = c.category_id
JOIN users AS u ON s.user_id = u.user_id
WHERE u.user_id = ?
AND s.stream_id = (SELECT MAX(stream_id) FROM streams WHERE user_id = ?)
""", (user_id, user_id))
return most_recent_stream
latest_vod = db.fetchone("""SELECT * FROM vods WHERE user_id = ? ORDER BY vod_id DESC LIMIT 1;""", (user_id,))
return latest_vod
def streamer_data(streamer_id: int) -> Optional[dict]:
def get_user_vods(user_id: int):
"""
Returns data of all vods by a streamer
"""
with Database() as db:
vods = db.fetchall("""SELECT * FROM vods WHERE user_id = ?;""", (user_id,))
return vods
def get_streamer_data(user_id: int) -> Optional[dict]:
"""
Returns information about the streamer
"""
with Database() as db:
data = db.fetchone("""
SELECT username, bio, num_followers, is_partnered FROM users
WHERE user_id = ?
""", (streamer_id,))
WHERE user_id = ?;
""", (user_id,))
return data
def generate_thumbnail(user_id: int) -> None:
"""
Returns the thumbnail of a stream
Generates the thumbnail of a stream
"""
db = Database()
username = db.fetchone("""SELECT * FROM users WHERE user_id = ?""", (user_id,))
db.close_connection()
with Database() as db:
username = db.fetchone("""SELECT * FROM users WHERE user_id = ?""", (user_id,))
if not username:
return None
if not os.path.exists(f"stream_data/thumbnails/"):
os.makedirs(f"stream_data/thumbnails/")
subprocess.Popen(["ls", "-lR"])
thumbnail_command = [
"ffmpeg",
@@ -115,16 +108,56 @@ def generate_thumbnail(user_id: int) -> None:
]
subprocess.run(thumbnail_command)
def stream_tags(stream_id: int) -> Optional[List[str]]:
def get_stream_tags(user_id: int) -> Optional[List[str]]:
"""
Given a stream return tags associated with the stream
Given a stream return tags associated with the user's stream
"""
with Database() as db:
tags = db.fetchall("""
SELECT tag_name
FROM tags
JOIN stream_tags ON tags.tag_id = stream_tags.tag_id
WHERE stream_id = ?
""", (stream_id,))
tags = [tag['tag_name'] for tag in tags] if tags else None
WHERE user_id = ?;
""", (user_id,))
return tags
def get_vod_tags(vod_id: int):
"""
Given a vod return tags associated with the vod
"""
with Database() as db:
tags = db.fetchall("""
SELECT tag_name
FROM tags
JOIN vod_tags ON tags.tag_id = vod_tags.tag_id
WHERE vod_id = ?;
""", (vod_id,))
return tags
def transfer_stream_to_vod(user_id: int):
"""
Deletes stream from stream table and moves it to VoD table
TODO: Add functionaliy to save stream permanently
"""
with Database() as db:
stream = db.fetchone("""
SELECT * FROM streams WHERE user_id = ?;
""", (user_id,))
if not stream:
return None
## TODO: calculate length in seconds, currently using temp value
db.execute("""
INSERT INTO vods (user_id, title, datetime, category_id, length, views)
VALUES (?, ?, ?, ?, ?, ?);
""", (stream["user_id"], stream["title"], stream["datetime"], stream["category_id"], 10, stream["num_viewers"]))
db.execute("""
DELETE FROM streams WHERE user_id = ?;
""", (user_id,))
return True