128 lines
4.0 KiB
Python
128 lines
4.0 KiB
Python
from datetime import datetime
|
|
import bcrypt
|
|
import requests
|
|
|
|
from lib.database.database import db_run
|
|
from models import Message
|
|
from models.user import UserCredentials
|
|
from models.movie import MovieIn, Movie
|
|
from models.card import Card
|
|
from lib.settings import settings
|
|
from lib.logger import logger
|
|
|
|
|
|
def add_user(user: UserCredentials):
|
|
hashed_password = bcrypt.hashpw(
|
|
user.password.encode("utf-8"), bcrypt.gensalt()
|
|
).decode("utf-8")
|
|
sql = f"INSERT INTO users (username, password) VALUES ('{user.username}', '{hashed_password}')"
|
|
try:
|
|
db_run(sql)
|
|
except Exception as e:
|
|
return Message(message=f"User already exists. Error: {e}", message_type="error")
|
|
|
|
return Message(message="User added successfully", message_type="success")
|
|
|
|
|
|
def remove_user(user_id: str):
|
|
logger.debug(f"Removing user {user_id}")
|
|
sql = f"DELETE FROM users WHERE id = '{user_id}'"
|
|
|
|
try:
|
|
db_run(sql)
|
|
except Exception as e:
|
|
return Message(message=f"User not found. Error: {e}", message_type="error")
|
|
|
|
return Message(message="User removed successfully", message_type="success")
|
|
|
|
|
|
def add_movie(movie: MovieIn):
|
|
logger.info(f"Adding movie {movie.imdb_id}")
|
|
logger.debug(f"Requesting movie data from OMDB: {movie.imdb_id}")
|
|
request_url = "https://omdbapi.com/"
|
|
request_params = {
|
|
"apikey": settings.OMDB_API_KEY,
|
|
"i": movie.imdb_id,
|
|
}
|
|
|
|
response = requests.get(request_url, params=request_params)
|
|
if response.status_code != 200:
|
|
logger.error(f"Failed to add movie {movie.imdb_id}")
|
|
return Message(message="Failed to add movie", message_type="error")
|
|
movie_data = response.json()
|
|
logger.debug(f"Movie data received from OMDB: {movie_data}")
|
|
if "Released" in movie_data:
|
|
movie_data["Released"] = datetime.strptime(movie_data["Released"], "%d %b %Y")
|
|
|
|
new_movie = Movie(
|
|
name=movie_data["Title"],
|
|
imdb_id=movie_data["imdbID"],
|
|
actors=[actor.strip() for actor in movie_data["Actors"].split(",")],
|
|
release_date=movie_data["Released"],
|
|
plot=movie.plot,
|
|
poster_url=movie_data["Poster"],
|
|
showtime=movie.showtime,
|
|
)
|
|
logger.debug(f"Movie data: {new_movie.model_dump()}")
|
|
sql = "INSERT INTO movies (name, imdb_id, actors, release_date, plot, showtime, poster_url) VALUES (?, ?, ?, ?, ?, ?, ?)"
|
|
try:
|
|
db_run(
|
|
sql,
|
|
(
|
|
new_movie.name,
|
|
new_movie.imdb_id,
|
|
new_movie.actors,
|
|
new_movie.release_date,
|
|
new_movie.plot,
|
|
new_movie.showtime,
|
|
new_movie.poster_url,
|
|
),
|
|
)
|
|
except Exception as e:
|
|
logger.error(f"Failed to add movie {movie.imdb_id}. Error: {e}")
|
|
return Message(message="Failed to add movie", message_type="error")
|
|
|
|
logger.info(f"Movie {movie.imdb_id} added successfully")
|
|
return Message(message="Movie added successfully", message_type="success")
|
|
|
|
|
|
def set_movie_watched(movie_id: str):
|
|
sql = f"UPDATE movies SET is_watched = TRUE WHERE id = '{movie_id}'"
|
|
try:
|
|
db_run(sql)
|
|
except Exception as e:
|
|
return Message(
|
|
message=f"Failed to set movie as watched. Error: {e}", message_type="error"
|
|
)
|
|
|
|
return Message(message="Movie set as watched successfully", message_type="success")
|
|
|
|
|
|
def add_card(card: Card):
|
|
sql = """
|
|
INSERT INTO cards (title, description, point_value)
|
|
VALUES (?, ?, ?)
|
|
"""
|
|
values = (
|
|
card.title,
|
|
card.description,
|
|
card.point_value,
|
|
)
|
|
try:
|
|
db_run(sql, values)
|
|
return {"message": "Card added successfully"}
|
|
except Exception as e:
|
|
return {"message": f"Error adding card: {e}"}
|
|
|
|
|
|
def remove_card(card_id: str):
|
|
sql = f"""
|
|
DELETE FROM cards WHERE id = '{card_id}'
|
|
"""
|
|
|
|
try:
|
|
db_run(sql)
|
|
return {"message": "Card removed successfully"}
|
|
except Exception as e:
|
|
return {"message": f"Error removing card: {e}"}
|