Spaces:
Sleeping
Sleeping
import pandas as pd | |
import requests | |
from bs4 import BeautifulSoup | |
import os | |
import re | |
import random | |
import io # No longer needed for CSV data, but keep for other potential uses | |
from dotenv import load_dotenv | |
from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM | |
import torch | |
import gradio as gr | |
from sklearn.feature_extraction.text import TfidfVectorizer | |
from sklearn.metrics.pairwise import cosine_similarity | |
import time # For adding slight delays if TMDB API rate limits are hit | |
# --- Configuration --- | |
load_dotenv() # Load environment variables from .env file for local testing | |
TMDB_API_KEY = os.environ.get("TMDB_API_KEY") | |
HF_TOKEN = os.environ.get("HF_TOKEN") | |
MODEL_NAME = "mistralai/Mistral-7B-Instruct-v0.1" | |
BASE_TMDB_URL = "https://api.themoviedb.org/3" | |
POSTER_BASE_URL = "https://image.tmdb.org/t/p/w500" | |
NUM_RECOMMENDATIONS_TO_GENERATE = 20 # Generate more initially | |
NUM_RECOMMENDATIONS_TO_DISPLAY = 5 # Display top 5 | |
MIN_RATING_FOR_SEED = 3.5 | |
MIN_VOTE_COUNT_TMDB = 100 # Min votes on TMDB for a movie to be considered | |
# --- Global Variables for Data (Load once) --- | |
df_profile_global = None | |
df_comments_global = None | |
df_watchlist_global = None | |
df_reviews_global = None | |
df_diary_global = None | |
df_ratings_global = None | |
df_watched_global = None # This will be a consolidated df | |
uri_to_movie_map_global = {} | |
all_watched_titles_global = set() | |
watchlist_titles_global = set() | |
favorite_film_details_global = [] | |
seed_movies_global = [] | |
# LLM Pipeline (Load once) | |
llm_pipeline = None | |
llm_tokenizer = None | |
# --- Helper Functions --- | |
def clean_html(raw_html): | |
if pd.isna(raw_html) or raw_html is None: | |
return "" | |
# Add space before tags to handle cases like </b>text | |
text = str(raw_html) | |
text = re.sub(r'<br\s*/?>', '\n', text) # Convert <br> to newlines | |
soup = BeautifulSoup(text, "html.parser") | |
return soup.get_text(separator=" ", strip=True) | |
def get_movie_uri_map(dfs_dict): | |
"""Creates a map from Letterboxd URI to (Name, Year).""" | |
uri_map = {} | |
# Order of preference for names/years if URIs are duplicated across files | |
# (though Name/Year should ideally be consistent for the same URI) | |
df_priority = ['reviews.csv', 'diary.csv', 'ratings.csv', 'watched.csv', 'watchlist.csv'] | |
processed_uris = set() | |
for df_name in df_priority: | |
df = dfs_dict.get(df_name) | |
if df is not None and 'Letterboxd URI' in df.columns and 'Name' in df.columns and 'Year' in df.columns: | |
for _, row in df.iterrows(): | |
uri = row['Letterboxd URI'] | |
if pd.notna(uri) and uri not in processed_uris: | |
if pd.notna(row['Name']) and pd.notna(row['Year']): | |
try: | |
year = int(row['Year']) # Ensure year is int | |
uri_map[uri] = (str(row['Name']), year) | |
processed_uris.add(uri) | |
except ValueError: | |
# print(f"Warning: Could not parse year for {row['Name']} in {df_name}. Skipping URI map entry.") | |
pass # Or handle as an error/log | |
return uri_map | |
def load_all_data(): | |
global df_profile_global, df_comments_global, df_watchlist_global, df_reviews_global | |
global df_diary_global, df_ratings_global, df_watched_global, uri_to_movie_map_global | |
global all_watched_titles_global, watchlist_titles_global, favorite_film_details_global, seed_movies_global | |
# --- Load DataFrames from CSV files --- | |
# IMPORTANT: Ensure these CSV files are uploaded to your Hugging Face Space root. | |
try: | |
df_profile_global = pd.read_csv("profile.csv") | |
df_comments_global = pd.read_csv("comments.csv") | |
df_watchlist_global = pd.read_csv("watchlist.csv") | |
df_reviews_global = pd.read_csv("reviews.csv") | |
df_diary_global = pd.read_csv("diary.csv") | |
df_ratings_global = pd.read_csv("ratings.csv") | |
# The 'watched.csv' you provided seems to be a log similar to diary, but without ratings. | |
# We'll primarily use diary, reviews, and ratings for watched history with ratings. | |
_df_watched_log = pd.read_csv("watched.csv") # Raw watched log | |
except FileNotFoundError as e: | |
print(f"ERROR: CSV file not found: {e}. Please ensure all CSV files are uploaded to the HF Space.") | |
return False # Indicate failure | |
dfs_for_uri_map = { | |
"reviews.csv": df_reviews_global, | |
"diary.csv": df_diary_global, | |
"ratings.csv": df_ratings_global, | |
"watched.csv": _df_watched_log, # from watched.csv | |
"watchlist.csv": df_watchlist_global | |
} | |
uri_to_movie_map_global = get_movie_uri_map(dfs_for_uri_map) | |
# --- Consolidate Watched History --- | |
# Combine diary, reviews, and ratings to get a comprehensive view of watched movies and their ratings/reviews | |
# Standardize column names for easier merging | |
df_diary_global.rename(columns={'Rating': 'Diary Rating'}, inplace=True) | |
df_reviews_global.rename(columns={'Rating': 'Review Rating', 'Review': 'Review Text'}, inplace=True) | |
df_ratings_global.rename(columns={'Rating': 'Simple Rating'}, inplace=True) | |
# Merge based on Letterboxd URI, Name, and Year (if URI is missing, try Name/Year) | |
# Start with reviews as it's richest | |
consolidated = df_reviews_global[['Letterboxd URI', 'Name', 'Year', 'Review Rating', 'Review Text', 'Watched Date']].copy() | |
consolidated.rename(columns={'Review Rating': 'Rating'}, inplace=True) | |
# Merge diary | |
diary_subset = df_diary_global[['Letterboxd URI', 'Name', 'Year', 'Diary Rating', 'Watched Date']].copy() | |
diary_subset.rename(columns={'Diary Rating': 'Rating_diary', 'Watched Date': 'Watched Date_diary'}, inplace=True) | |
consolidated = pd.merge(consolidated, diary_subset, on=['Letterboxd URI', 'Name', 'Year'], how='outer', suffixes=('', '_diary')) | |
consolidated['Rating'] = consolidated['Rating'].fillna(consolidated['Rating_diary']) | |
consolidated['Watched Date'] = consolidated['Watched Date'].fillna(consolidated['Watched Date_diary']) | |
consolidated.drop(columns=['Rating_diary', 'Watched Date_diary'], inplace=True) | |
# Merge simple ratings | |
ratings_subset = df_ratings_global[['Letterboxd URI', 'Name', 'Year', 'Simple Rating']].copy() | |
ratings_subset.rename(columns={'Simple Rating': 'Rating_simple'}, inplace=True) | |
consolidated = pd.merge(consolidated, ratings_subset, on=['Letterboxd URI', 'Name', 'Year'], how='outer', suffixes=('', '_simple')) | |
consolidated['Rating'] = consolidated['Rating'].fillna(consolidated['Rating_simple']) | |
consolidated.drop(columns=['Rating_simple'], inplace=True) | |
# Add movies from the raw watched.csv if they aren't already there (they won't have ratings from this source) | |
watched_log_subset = _df_watched_log[['Letterboxd URI', 'Name', 'Year']].copy() | |
# Add a 'Watched' column to mark these, and merge, filling NaNs appropriately | |
watched_log_subset['from_watched_log'] = True | |
consolidated = pd.merge(consolidated, watched_log_subset, on=['Letterboxd URI', 'Name', 'Year'], how='outer') | |
consolidated['from_watched_log'] = consolidated['from_watched_log'].fillna(False) | |
# Clean up and fill NAs | |
consolidated['Review Text'] = consolidated['Review Text'].fillna('').apply(clean_html) | |
consolidated['Year'] = pd.to_numeric(consolidated['Year'], errors='coerce').astype('Int64') # Handle potential non-numeric years | |
consolidated.dropna(subset=['Name', 'Year'], inplace=True) # Movies must have a name and year | |
consolidated.drop_duplicates(subset=['Name', 'Year'], keep='first', inplace=True) | |
df_watched_global = consolidated | |
# Populate all_watched_titles_global (Name, Year) tuples | |
all_watched_titles_global = set(zip(df_watched_global['Name'].astype(str), df_watched_global['Year'].astype(int))) | |
# Add from raw watched log as well | |
for _, row in _df_watched_log.iterrows(): | |
if pd.notna(row['Name']) and pd.notna(row['Year']): | |
try: | |
all_watched_titles_global.add((str(row['Name']), int(row['Year']))) | |
except ValueError: | |
pass | |
# --- Process Watchlist --- | |
if df_watchlist_global is not None: | |
watchlist_titles_global = set() | |
for _, row in df_watchlist_global.iterrows(): | |
if pd.notna(row['Name']) and pd.notna(row['Year']): | |
try: | |
watchlist_titles_global.add((str(row['Name']), int(row['Year']))) | |
except ValueError: | |
pass | |
# --- Process Favorite Films --- | |
favorite_film_details_global = [] | |
if df_profile_global is not None and 'Favorite Films' in df_profile_global.columns: | |
fav_uris_str = df_profile_global.iloc[0]['Favorite Films'] | |
if pd.notna(fav_uris_str): | |
fav_uris = [uri.strip() for uri in fav_uris_str.split(',')] | |
for uri in fav_uris: | |
if uri in uri_to_movie_map_global: | |
name, year = uri_to_movie_map_global[uri] | |
# Try to find rating and review from consolidated watched data | |
match = df_watched_global[(df_watched_global['Name'] == name) & (df_watched_global['Year'] == year)] | |
rating = match['Rating'].iloc[0] if not match.empty and pd.notna(match['Rating'].iloc[0]) else None | |
review = match['Review Text'].iloc[0] if not match.empty and match['Review Text'].iloc[0] else "" | |
favorite_film_details_global.append({'name': name, 'year': year, 'rating': rating, 'review_text': review, 'uri': uri}) | |
# --- Identify Seed Movies --- | |
# Start with favorites | |
seed_movies_global.extend(favorite_film_details_global) | |
# Add other highly-rated movies (non-favorites) | |
highly_rated_df = df_watched_global[df_watched_global['Rating'] >= MIN_RATING_FOR_SEED] | |
favorite_uris = {fav['uri'] for fav in favorite_film_details_global if 'uri' in fav} | |
for _, row in highly_rated_df.iterrows(): | |
if row['Letterboxd URI'] not in favorite_uris: # Avoid duplicates if already in favorites | |
seed_movies_global.append({ | |
'name': row['Name'], | |
'year': row['Year'], | |
'rating': row['Rating'], | |
'review_text': row['Review Text'], | |
'uri': row['Letterboxd URI'] | |
}) | |
# Remove duplicates based on name and year, preferring entries with more info (e.g., from favorites) | |
temp_df = pd.DataFrame(seed_movies_global) | |
temp_df.drop_duplicates(subset=['name', 'year'], keep='first', inplace=True) | |
seed_movies_global = temp_df.to_dict('records') | |
random.shuffle(seed_movies_global) # Shuffle to get variety if we pick a subset | |
return True # Indicate success | |
def initialize_llm(): | |
global llm_pipeline, llm_tokenizer | |
if llm_pipeline is None: | |
print(f"Initializing LLM: {MODEL_NAME}") | |
try: | |
llm_tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME, trust_remote_code=True) | |
# For CPU, bfloat16 might not be supported, try float32 or default | |
# Adding device_map="auto" and load_in_8bit=True for potentially better memory management on CPU | |
# For Spaces CPU, bitsandbytes might not be ideal. Try without quantization first if issues arise. | |
# Remove load_in_8bit if it causes issues on standard CPU Space. | |
model = AutoModelForCausalLM.from_pretrained( | |
MODEL_NAME, | |
torch_dtype=torch.float16, # Use float16 for faster inference and less memory | |
device_map="auto", # Automatically maps to available device (CPU or GPU if available) | |
# load_in_8bit=True, # Quantization - might need bitsandbytes | |
trust_remote_code=True, | |
token=HF_TOKEN if HF_TOKEN else None | |
) | |
llm_pipeline = pipeline( | |
"text-generation", | |
model=model, | |
tokenizer=llm_tokenizer, | |
torch_dtype=torch.float16, | |
device_map="auto" | |
) | |
print("LLM Initialized Successfully.") | |
except Exception as e: | |
print(f"Error initializing LLM: {e}") | |
llm_pipeline = None # Ensure it's None if initialization fails | |
# --- TMDB API Functions --- | |
def search_tmdb_movie_details(title, year): | |
if not TMDB_API_KEY: | |
print("TMDB API Key not configured.") | |
return None | |
try: | |
search_url = f"{BASE_TMDB_URL}/search/movie" | |
params = {'api_key': TMDB_API_KEY, 'query': title, 'year': year, 'language': 'en-US'} | |
response = requests.get(search_url, params=params) | |
response.raise_for_status() | |
results = response.json().get('results', []) | |
if results: | |
movie = results[0] | |
# Fetch genres using the /movie/{movie_id} endpoint to get full genre names | |
movie_details_url = f"{BASE_TMDB_URL}/movie/{movie['id']}" | |
details_params = {'api_key': TMDB_API_KEY, 'language': 'en-US'} | |
details_response = requests.get(movie_details_url, params=details_params) | |
details_response.raise_for_status() | |
movie_full_details = details_response.json() | |
return { | |
'id': movie.get('id'), | |
'title': movie.get('title'), | |
'year': str(movie.get('release_date', ''))[:4], | |
'overview': movie.get('overview'), | |
'poster_path': POSTER_BASE_URL + movie.get('poster_path') if movie.get('poster_path') else "https://via.placeholder.com/500x750.png?text=No+Poster", | |
'genres': [genre['name'] for genre in movie_full_details.get('genres', [])], | |
'vote_average': movie.get('vote_average'), | |
'vote_count': movie.get('vote_count'), | |
'popularity': movie.get('popularity') | |
} | |
time.sleep(0.2) # Small delay to respect API rate limits | |
except requests.RequestException as e: | |
print(f"Error searching TMDB for {title} ({year}): {e}") | |
except Exception as ex: | |
print(f"Unexpected error in search_tmdb_movie_details for {title} ({year}): {ex}") | |
return None | |
def get_tmdb_recommendations(movie_id, page=1): | |
if not TMDB_API_KEY: | |
print("TMDB API Key not configured.") | |
return [] | |
recommendations = [] | |
try: | |
rec_url = f"{BASE_TMDB_URL}/movie/{movie_id}/recommendations" | |
params = {'api_key': TMDB_API_KEY, 'page': page, 'language': 'en-US'} | |
response = requests.get(rec_url, params=params) | |
response.raise_for_status() | |
results = response.json().get('results', []) | |
for movie in results: | |
if movie.get('vote_count', 0) >= MIN_VOTE_COUNT_TMDB: | |
recommendations.append({ | |
'id': movie.get('id'), | |
'title': movie.get('title'), | |
'year': str(movie.get('release_date', ''))[:4] if movie.get('release_date') else "N/A", | |
'overview': movie.get('overview'), | |
'poster_path': POSTER_BASE_URL + movie.get('poster_path') if movie.get('poster_path') else "https://via.placeholder.com/500x750.png?text=No+Poster", | |
'vote_average': movie.get('vote_average'), | |
'vote_count': movie.get('vote_count'), | |
'popularity': movie.get('popularity') | |
}) | |
time.sleep(0.2) # Small delay | |
except requests.RequestException as e: | |
print(f"Error getting TMDB recommendations for movie ID {movie_id}: {e}") | |
except Exception as ex: | |
print(f"Unexpected error in get_tmdb_recommendations for movie ID {movie_id}: {ex}") | |
return recommendations | |
# --- LLM Explanation Generation --- | |
def generate_saudi_explanation(recommended_movie_title, seed_movie_title, seed_movie_context=""): | |
global llm_pipeline, llm_tokenizer | |
if llm_pipeline is None or llm_tokenizer is None: | |
return "للأسف، نموذج الذكاء الاصطناعي مو جاهز الحين. حاول مرة ثانية بعد شوي." | |
# Truncate long context to avoid overly long prompts | |
max_context_len = 200 | |
if len(seed_movie_context) > max_context_len: | |
seed_movie_context_short = seed_movie_context[:max_context_len] + "..." | |
else: | |
seed_movie_context_short = seed_movie_context | |
prompt_template = f"""<s>[INST] أنت ناقد أفلام سعودي خبير ودمك خفيف. المستخدم أعجب بالفيلم "{seed_movie_title}". | |
سبب إعجابه بالفيلم الأول (إذا متوفر): "{seed_movie_context_short}" | |
بناءً على ذلك، نُرشح له فيلم "{recommended_movie_title}". | |
اكتب جملة أو جملتين باللهجة السعودية العامية، تشرح ليش ممكن يعجبه الفيلم الجديد "{recommended_movie_title}"، مع ربطها بالفيلم اللي عجبه "{seed_movie_title}". خلي كلامك وناسة ويشد الواحد وما يكون طويل. لا تذكر أبداً أنك نموذج لغوي أو ذكاء اصطناعي. | |
مثال للأسلوب المطلوب (لو الفيلم اللي عجبه "Mad Max: Fury Road" والفيلم المرشح "Dune"): | |
"يا طويل العمر، شفت كيف 'Mad Max: Fury Road' عجّبك بجوّه الصحراوي والأكشن اللي ما يوقّف؟ أجل اسمع، 'Dune' بيوديك لصحراء ثانية بس أعظم وأفخم، وقصة تحبس الأنفاس! شد حيلك وشوفه." | |
الآن، الفيلم الذي أعجب المستخدم هو: "{seed_movie_title}" | |
سبب إعجابه بالفيلم الأول (إذا متوفر): "{seed_movie_context_short}" | |
الفيلم المرشح: "{recommended_movie_title}" | |
اشرح باللهجة السعودية: [/INST]""" | |
try: | |
sequences = llm_pipeline( | |
prompt_template, | |
do_sample=True, | |
top_k=10, | |
num_return_sequences=1, | |
eos_token_id=llm_tokenizer.eos_token_id, | |
max_new_tokens=100 # Limit output length | |
) | |
explanation = sequences[0]['generated_text'].split("[/INST]")[-1].strip() | |
# Further clean up if the model repeats parts of the prompt or adds unwanted prefixes | |
explanation = re.sub(r"^اشرح باللهجة السعودية:\s*", "", explanation, flags=re.IGNORECASE) | |
explanation = explanation.replace("<s>", "").replace("</s>", "").strip() | |
if not explanation or explanation.lower().startswith("أنت ناقد أفلام"): # Fallback if generation is poor | |
return f"شكلك بتنبسط على فيلم '{recommended_movie_title}' لأنه يشبه جو فيلم '{seed_movie_title}' اللي حبيته! عطيه تجربة." | |
return explanation | |
except Exception as e: | |
print(f"Error during LLM generation: {e}") | |
return f"يا كابتن، شكلك بتحب '{recommended_movie_title}'، خاصة إنك استمتعت بـ'{seed_movie_title}'. جربه وعطنا رأيك!" | |
# --- Recommendation Logic --- | |
def get_recommendations_for_salman(progress=gr.Progress()): | |
if not TMDB_API_KEY: | |
return "<p style='color:red; text-align:right;'>خطأ: مفتاح TMDB API مو موجود. الرجاء إضافته كـ Secret في Hugging Face Space.</p>" | |
if not all([df_profile_global is not None, df_watched_global is not None, seed_movies_global]): | |
return "<p style='color:red; text-align:right;'>خطأ: فشل في تحميل بياناتك. تأكد من رفع ملفات CSV بشكل صحيح.</p>" | |
if llm_pipeline is None: | |
initialize_llm() # Attempt to initialize if not already | |
if llm_pipeline is None: | |
return "<p style='color:red; text-align:right;'>خطأ: فشل في تهيئة نموذج الذكاء الاصطناعي. حاول تحديث الصفحة.</p>" | |
progress(0.1, desc="جمعنا أفلامك المفضلة واللي قيمتها عالي...") | |
potential_recs = {} # Store as {tmdb_id: {'movie_info': ..., 'seed_movie': ..., 'seed_context': ...}} | |
# Limit the number of seed movies to process to avoid excessive API calls / long processing | |
seeds_to_process = seed_movies_global[:30] if len(seed_movies_global) > 30 else seed_movies_global | |
for i, seed_movie in enumerate(seeds_to_process): | |
progress(0.1 + (i / len(seeds_to_process)) * 0.4, desc=f"نبحث عن توصيات بناءً على: {seed_movie['name']}") | |
seed_tmdb_details = search_tmdb_movie_details(seed_movie['name'], seed_movie['year']) | |
if seed_tmdb_details and seed_tmdb_details.get('id'): | |
tmdb_recs = get_tmdb_recommendations(seed_tmdb_details['id']) | |
for rec in tmdb_recs: | |
rec_tuple = (str(rec['title']), int(rec['year'])) # (Name, Year) | |
# Ensure rec_tuple elements are of correct type before comparison | |
if rec.get('id') and rec_tuple not in all_watched_titles_global and rec_tuple not in watchlist_titles_global: | |
if rec['id'] not in potential_recs: # Add if new, prioritizing first seed | |
potential_recs[rec['id']] = { | |
'movie_info': rec, | |
'seed_movie_title': seed_movie['name'], | |
'seed_movie_context': seed_movie.get('review_text', '') or seed_movie.get('comment_text', '') | |
} | |
# Simple content-based similarity as a fallback or supplement (Optional, can be complex) | |
# For now, primarily TMDB-based | |
if not potential_recs: | |
return "<p style='text-align:right;'>ما لقينا توصيات جديدة لك حالياً. يمكن شفت كل شيء رهيب! 😉</p>" | |
# Sort recommendations (e.g., by popularity or a mix, or just randomize for now) | |
# Let's sort by TMDB popularity for now to get some generally well-regarded films | |
sorted_recs_list = sorted(potential_recs.values(), key=lambda x: x['movie_info'].get('popularity', 0), reverse=True) | |
final_recommendations_data = [] | |
# Take top N distinct recommendations | |
displayed_ids = set() | |
for rec_data in sorted_recs_list: | |
if len(final_recommendations_data) >= NUM_RECOMMENDATIONS_TO_DISPLAY: | |
break | |
if rec_data['movie_info']['id'] not in displayed_ids: | |
final_recommendations_data.append(rec_data) | |
displayed_ids.add(rec_data['movie_info']['id']) | |
if not final_recommendations_data: | |
return "<p style='text-align:right;'>ما لقينا توصيات جديدة لك حالياً بعد الفلترة. يمكن شفت كل شيء رهيب! 😉</p>" | |
output_html = "<div>" # Main container | |
progress(0.6, desc="نجهز لك الشرح باللغة العامية...") | |
for i, rec_data in enumerate(final_recommendations_data): | |
progress(0.6 + (i / len(final_recommendations_data)) * 0.4, desc=f"نكتب شرح لفيلم: {rec_data['movie_info']['title']}") | |
explanation = generate_saudi_explanation( | |
rec_data['movie_info']['title'], | |
rec_data['seed_movie_title'], | |
rec_data['seed_movie_context'] | |
) | |
poster_url = rec_data['movie_info']['poster_path'] | |
if not poster_url or "placeholder.com" in poster_url: # Use a default if no poster | |
poster_url = f"https://via.placeholder.com/300x450.png?text={rec_data['movie_info']['title'].replace(' ', '+')}" | |
output_html += f""" | |
<div style="display: flex; flex-direction: row-reverse; align-items: flex-start; margin-bottom: 25px; border-bottom: 1px solid #ddd; padding-bottom:15px; background-color: #f9f9f9; border-radius: 8px; padding: 15px;"> | |
<img src="{poster_url}" alt="{rec_data['movie_info']['title']}" style="width: 150px; max-width:30%; height: auto; margin-left: 20px; border-radius: 5px; box-shadow: 2px 2px 5px rgba(0,0,0,0.1);"> | |
<div style="text-align: right; direction: rtl; flex-grow: 1;"> | |
<h3 style="margin-top:0; color: #c70039;">{rec_data['movie_info']['title']} ({rec_data['movie_info']['year']})</h3> | |
<p style="font-size: 1.1em; color: #333; line-height: 1.6;">{explanation}</p> | |
<p style="font-size: 0.9em; color: #777; margin-top: 10px;"><em>يا وحش، رشحنا لك هذا الفيلم لأنك حبيت: <strong style="color:#555;">{rec_data['seed_movie_title']}</strong></em></p> | |
</div> | |
</div> | |
""" | |
output_html += "</div>" | |
return gr.HTML(output_html) | |
# --- Gradio Interface --- | |
css = """ | |
body { font-family: 'Tajawal', sans-serif; } | |
.gradio-container { font-family: 'Tajawal', sans-serif !important; direction: rtl; } | |
footer { display: none !important; } | |
.gr-button { background-color: #c70039 !important; color: white !important; font-size: 1.2em !important; padding: 10px 20px !important; border-radius: 8px !important; } | |
.gr-button:hover { background-color: #a3002f !important; } | |
.gr-input { text-align: right !important; } | |
.gr-output { text-align: right !important; } | |
h1, h3 { color: #900c3f !important; } | |
""" | |
# Load data once when the script starts | |
data_loaded_successfully = load_all_data() | |
if data_loaded_successfully: | |
print("All user data loaded and preprocessed successfully.") | |
# Initialize LLM after data loading to ensure it happens on app startup if data is present | |
initialize_llm() | |
else: | |
print("Failed to load user data. The app might not function correctly.") | |
with gr.Blocks(theme=gr.themes.Soft(primary_hue="red", secondary_hue="pink"), css=css) as iface: | |
gr.Markdown( | |
""" | |
<div style="text-align: center;"> | |
<h1 style="color: #c70039; font-size: 2.5em;">🎬 رفيقك السينمائي 🍿</h1> | |
<p style="font-size: 1.2em; color: #555;">يا هلا بك يا سلمان! اضغط الزر تحت وخلنا نعطيك توصيات أفلام على كيف كيفك، مع شرح بالعامية ليش ممكن تدخل مزاجك.</p> | |
</div> | |
""" | |
) | |
recommend_button = gr.Button("يا سلمان، عطني توصيات أفلام!") | |
with gr.Column(): | |
output_recommendations = gr.HTML(label="توصياتك النارية 🔥") | |
recommend_button.click( | |
fn=get_recommendations_for_salman, | |
inputs=[], | |
outputs=[output_recommendations] | |
) | |
gr.Markdown( | |
""" | |
<div style="text-align: center; margin-top: 30px; font-size: 0.9em; color: #777;"> | |
<p>تم تطوير هذا النظام بواسطة الذكاء الاصطناعي مع لمسة شخصية من بياناتك في ليتربوكسد.</p> | |
<p>استمتع بالمشاهدة! 🎥</p> | |
</div> | |
""" | |
) | |
if __name__ == "__main__": | |
iface.launch(debug=True) # debug=True for local testing, remove for HF |