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

FEAT UPDATE: Updated Utils to use new database. Changed route arguments. Added unfollow feature

Browse Source
This commit is contained in:
JustIceO7
2025-01-31 02:00:31 +00:00
parent be53d4072e
commit 908870f751
9 changed files with 151 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

97
web_server/utils/recommendation_utils.py
View File

@@ -1,12 +1,12 @@
from database.database import Database
from typing import Optional, List, Tuple
from typing import Optional, List
def user_recommendation_category(user_id: int) -> Optional[int]:
"""
Queries user_preferences database to find users favourite streaming category and returns the category
"""
db = Database()
data = db.fetchone("""
with Database() as db:
data = db.fetchone("""
SELECT category_id
FROM user_preferences
WHERE user_id = ?
@@ -15,66 +15,60 @@ def user_recommendation_category(user_id: int) -> Optional[int]:
""", (user_id,))
return data
#TODO Needs to be reworked to get categories instead of streams of categories (below can be done in another function - get_streams_by_category)
def followed_categories_recommendations(user_id : int):
def followed_categories_recommendations(user_id : int) -> Optional[List[dict]]:
"""
Returns top 25 streams given a users category following
"""
db = Database()
# TODO: Change this to do what the function says
categories = db.fetchall("""
SELECT s.stream_id, s.title, u.username, s.num_viewers, c.category_name
FROM streams AS s
JOIN users AS u ON u.user_id = s.user_id
JOIN categories AS c ON s.category_id = c.category_id
JOIN followed_categories AS f ON s.category_id = c.category_id
WHERE f.user_id = ?
ORDER BY s.num_viewers DESC
LIMIT 25;
with Database() as db:
streams = db.fetchall("""
SELECT stream_id, title, username, num_viewers, category_name
FROM streams
JOIN users ON users.user_id = streams.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
LIMIT 25;
""", (user_id,))
return streams
""", (user_id,))
return categories
#TODO Needs to be reworked to get categories instead of streams of categories
def recommendations_based_on_category(category_id: int) -> Optional[List[Tuple[int, str, int]]]:
def recommendations_based_on_category(category_id: int) -> Optional[List[dict]]:
"""
Queries stream database to get top 25 most viewed streams based on given category and returns
(user_id, title, username, num_viewers, category_name)
Queries stream database to get top 25 most viewed streams based on given category
"""
db = Database()
data = db.fetchall("""
SELECT streams.category_id, streams.user_id, streams.title, users.username, streams.num_viewers, categories.category_name
FROM streams
JOIN users ON users.user_id = streams.user_id
JOIN categories ON streams.category_id = categories.category_id
WHERE categories.category_id = ?
ORDER BY num_viewers DESC
LIMIT 25
""", (category_id,))
return data
with Database() as db:
streams = db.fetchall("""
SELECT streams.stream_id, title, username, num_viewers, category_name
FROM streams
JOIN users ON users.user_id = streams.user_id
JOIN categories ON streams.category_id = categories.category_id
WHERE categories.category_id = ?
ORDER BY num_viewers DESC
LIMIT 25
""", (category_id,))
return streams
def default_recommendations():
def default_recommendations() -> Optional[List[dict]]:
"""
Return a list of 25 recommended live streams by number of viewers
(user_id, title, username, num_viewers, category_name)
"""
db = Database()
data = db.fetchall("""
SELECT stream_id, users.user_id, title, username, num_viewers, category_name
FROM streams
JOIN users ON users.user_id = streams.user_id
JOIN categories ON streams.category_id = categories.category_id
ORDER BY num_viewers DESC
LIMIT 25;
""")
return data
with Database() as db:
data = db.fetchall("""
SELECT streams.stream_id, title, username, num_viewers, category_name
FROM streams
JOIN users ON users.user_id = streams.user_id
JOIN categories ON streams.category_id = categories.category_id
ORDER BY num_viewers DESC
LIMIT 25;
""")
return data
def category_recommendations():
def category_recommendations() -> Optional[List[dict]]:
"""
Returns a list of the top 5 most popular categories
Returns a list of the top 5 most popular live categories
"""
db = Database()
categories = db.fetchall("""
with Database() as db:
categories = db.fetchall("""
SELECT categories.category_id, categories.category_name
FROM streams
JOIN categories ON streams.category_id = categories.category_id
@@ -83,5 +77,4 @@ def category_recommendations():
ORDER BY SUM(streams.num_viewers) DESC
LIMIT 5;
""")
return categories
return categories

40
web_server/utils/stream_utils.py
View File

@@ -2,13 +2,14 @@ from database.database import Database
from typing import Optional
import sqlite3, os, subprocess
from time import sleep
from typing import Optional, List
def streamer_live_status(user_id: int) -> dict:
"""
Returns boolean on whether the given streamer is live
"""
db = Database()
is_live = db.fetchone("""
with Database() as db:
is_live = db.fetchone("""
SELECT isLive
FROM streams
WHERE user_id = ?
@@ -18,12 +19,12 @@ def streamer_live_status(user_id: int) -> dict:
return is_live
def followed_live_streams(user_id: int) -> list[dict]:
def followed_live_streams(user_id: int) -> Optional[List[dict]]:
"""
Searches for streamers who the user followed which are currently live
"""
db = Database()
live_streams = db.fetchall("""
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 = ?)
@@ -32,12 +33,12 @@ def followed_live_streams(user_id: int) -> list[dict]:
""", (user_id,))
return live_streams
def followed_streamers(user_id: int) -> list[dict]:
def followed_streamers(user_id: int) -> Optional[List[dict]]:
"""
Returns a list of streamers who the user follows
"""
db = Database()
followed_streamers = db.fetchall("""
with Database() as db:
followed_streamers = db.fetchall("""
SELECT user_id, username
FROM users
WHERE user_id IN (SELECT followed_id FROM follows WHERE user_id = ?);
@@ -48,8 +49,8 @@ def streamer_most_recent_stream(user_id: int) -> dict:
"""
Returns data of the most recent stream by a streamer
"""
db = Database()
most_recent_stream = db.fetchone("""
with Database() as db:
most_recent_stream = db.fetchone("""
SELECT * FROM streams
WHERE user_id = ?
AND stream_id = (SELECT MAX(stream_id) FROM streams WHERE user_id = ?)
@@ -60,8 +61,8 @@ def user_stream(user_id: int, stream_id: int) -> dict:
"""
Returns data of a streamers selected stream
"""
db = Database()
stream = db.fetchone("""
with Database() as db:
stream = db.fetchone("""
SELECT * FROM streams
WHERE user_id = ?
AND stream_id = ?
@@ -96,4 +97,17 @@ def generate_thumbnail(user_id: int) -> None:
f"stream_data/thumbnails/{username['username']}.jpg"
]
subprocess.run(thumbnail_command)
subprocess.run(thumbnail_command)
def stream_tags(stream_id: int) -> Optional[List[str]]:
"""
Given a stream return tags associated with the 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
return tags

49
web_server/utils/user_utils.py
View File

@@ -9,15 +9,15 @@ load_dotenv()
serializer = URLSafeTimedSerializer(getenv("AUTH_SECRET_KEY"))
def get_user_id(username: str) -> int:
def get_user_id(username: str) -> Optional[int]:
"""
Returns user_id associated with given username
"""
db = Database()
data = db.fetchone("""
with Database() as db:
data = db.fetchone("""
SELECT user_id
FROM users
WHERE username = ?;
WHERE username = ?
""", (username,))
return data['user_id'] if data else None
@@ -25,8 +25,8 @@ def get_username(user_id: str) -> Optional[str]:
"""
Returns username associated with given user_id
"""
db = Database()
data = db.fetchone("""
with Database() as db:
data = db.fetchone("""
SELECT username
FROM user
WHERE user_id = ?
@@ -37,8 +37,8 @@ def is_user_partner(user_id: int) -> bool:
"""
Returns True if user is a partner, else False
"""
db = Database()
data = db.fetchone("""
with Database() as db:
data = db.fetchone("""
SELECT is_partnered
FROM users
WHERE user_id = ?
@@ -49,8 +49,8 @@ def is_subscribed(user_id: int, streamer_id: int) -> bool:
"""
Returns True if user is subscribed to a streamer, else False
"""
db = Database()
result = db.fetchone("""
with Database() as db:
result = db.fetchone("""
SELECT 1
FROM subscribes
WHERE user_id = ?
@@ -63,8 +63,8 @@ def is_following(user_id: int, followed_id: int) -> bool:
"""
Returns where a user is following another
"""
db = Database()
result = db.fetchone("""
with Database() as db:
result = db.fetchone("""
SELECT 1
FROM follows
WHERE user_id = ?
@@ -72,12 +72,25 @@ def is_following(user_id: int, followed_id: int) -> bool:
""", (user_id, followed_id))
return bool(result)
def unfollow(user_id: int, followed_id: int) -> bool:
"""
Unfollows follow_id user from user_id user
"""
with Database() as db:
db.execute("""
DELETE FROM follows
WHERE user_id = ?
AND followed_id = ?
""", (user_id, followed_id))
return True
return False
def subscription_expiration(user_id: int, subscribed_id: int) -> int:
"""
Returns the amount of time left until user subscription to a streamer ends
"""
db = Database()
data = db.fetchone("""
with Database() as db:
data = db.fetchone("""
SELECT expires
FROM subscriptions
WHERE user_id = ?
@@ -92,19 +105,19 @@ def subscription_expiration(user_id: int, subscribed_id: int) -> int:
return 0
def verify_token(token: str):
def verify_token(token: str) -> Optional[str]:
"""
Given a token verifies token and decodes the token into an email
"""
email = serializer.loads(token, salt='1', max_age=3600)
return email if email else False
def reset_password(new_password: str, email: str):
def reset_password(new_password: str, email: str) -> bool:
"""
Given email and new password reset the password for a given user
"""
db = Database()
db.execute("""
with Database() as db:
db.execute("""
UPDATE users
SET password = ?
WHERE email = ?

20
web_server/utils/utils.py
View File

@@ -1,29 +1,30 @@
from database.database import Database
from typing import Optional, List
def categories():
def categories() -> Optional[List[dict]]:
"""
Returns all possible streaming categories
"""
db = Database()
all_categories = db.fetchall("SELECT * FROM categories")
with Database() as db:
all_categories = db.fetchall("SELECT * FROM categories")
return all_categories
def tags():
def tags() -> Optional[List[dict]]:
"""
Returns all possible streaming tags
"""
db = Database()
all_tags = db.fetchall("SELECT * FROM tags")
with Database() as db:
all_tags = db.fetchall("SELECT * FROM tags")
return all_tags
def most_popular_category():
def most_popular_category() -> Optional[List[dict]]:
"""
Returns the most popular category based on live stream viewers
"""
db = Database()
category = db.fetchone("""
with Database() as db:
category = db.fetchone("""
SELECT categories.category_id, categories.category_name
FROM streams
JOIN categories ON streams.category_id = categories.category_id
@@ -34,4 +35,3 @@ def most_popular_category():
""")
return category