import pandas as pd import requests from bs4 import BeautifulSoup import os import re import random from dotenv import load_dotenv # For local testing with a .env file from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM import torch import gradio as gr import time # --- Configuration --- load_dotenv() # Loads HF_TOKEN and TMDB_API_KEY from .env for local testing # SECRETS - These will be read from Hugging Face Space Secrets when deployed TMDB_API_KEY = os.environ.get("TMDB_API_KEY") HF_TOKEN = os.environ.get("HF_TOKEN") # Essential for gated models like ALLaM MODEL_NAME = "ALLaM-AI/ALLaM-7B-Instruct-preview" # Target ALLaM model BASE_TMDB_URL = "https://api.themoviedb.org/3" POSTER_BASE_URL = "https://image.tmdb.org/t/p/w500" NUM_RECOMMENDATIONS_TO_DISPLAY = 5 MIN_RATING_FOR_SEED = 3.5 MIN_VOTE_COUNT_TMDB = 100 # Minimum votes on TMDB for a movie to be considered # --- Global Variables for Data & Model (Load once) --- df_profile_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 = None llm_tokenizer = None # --- Helper Functions --- def clean_html(raw_html): if pd.isna(raw_html) or raw_html is None: return "" text = str(raw_html) text = re.sub(r'', '\n', text) # Convert
to newlines soup = BeautifulSoup(text, "html.parser") return soup.get_text(separator=" ", strip=True) def get_movie_uri_map(dfs_dict): uri_map = {} 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']) uri_map[uri] = (str(row['Name']), year) processed_uris.add(uri) except ValueError: # Silently skip if year is not a valid integer for URI mapping pass return uri_map def load_all_data(): global df_profile_global, df_watchlist_global, df_reviews_global, df_diary_global global df_ratings_global, df_watched_global, uri_to_movie_map_global, all_watched_titles_global global watchlist_titles_global, favorite_film_details_global, seed_movies_global try: # Assumes CSV files are in the root of the Hugging Face Space df_profile_global = pd.read_csv("profile.csv") # df_comments_global = pd.read_csv("comments.csv") # Not directly used in recs logic 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") _df_watched_log = pd.read_csv("watched.csv") # Raw log of watched films except FileNotFoundError as e: print(f"CRITICAL ERROR: CSV file not found: {e}. Ensure all CSVs are uploaded to the HF Space root.") return False # Indicate failure to load data dfs_for_uri_map = { "reviews.csv": df_reviews_global, "diary.csv": df_diary_global, "ratings.csv": df_ratings_global, "watched.csv": _df_watched_log, "watchlist.csv": df_watchlist_global } uri_to_movie_map_global = get_movie_uri_map(dfs_for_uri_map) 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) consolidated = df_reviews_global[['Letterboxd URI', 'Name', 'Year', 'Review Rating', 'Review Text', 'Watched Date']].copy() consolidated.rename(columns={'Review Rating': 'Rating'}, inplace=True) 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) 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) watched_log_subset = _df_watched_log[['Letterboxd URI', 'Name', 'Year']].copy() 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).astype(bool) consolidated['Review Text'] = consolidated['Review Text'].fillna('').apply(clean_html) consolidated['Year'] = pd.to_numeric(consolidated['Year'], errors='coerce').astype('Int64') consolidated.dropna(subset=['Name', 'Year'], inplace=True) # Ensure essential fields are present consolidated.drop_duplicates(subset=['Name', 'Year'], keep='first', inplace=True) df_watched_global = consolidated all_watched_titles_global = set(zip(df_watched_global['Name'].astype(str), df_watched_global['Year'].astype(int))) 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 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 favorite_film_details_global = [] if df_profile_global is not None and 'Favorite Films' in df_profile_global.columns and not df_profile_global.empty: 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] 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}) seed_movies_global.extend(favorite_film_details_global) if not df_watched_global.empty: # Ensure df_watched_global is not empty 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: seed_movies_global.append({ 'name': row['Name'], 'year': row['Year'], 'rating': row['Rating'], 'review_text': row['Review Text'], 'uri': row['Letterboxd URI'] }) if seed_movies_global: # Only process if seed_movies_global is not empty temp_df = pd.DataFrame(seed_movies_global) if not temp_df.empty: temp_df.drop_duplicates(subset=['name', 'year'], keep='first', inplace=True) seed_movies_global = temp_df.to_dict('records') else: seed_movies_global = [] random.shuffle(seed_movies_global) return True def initialize_llm(): global llm_pipeline, llm_tokenizer if llm_pipeline is None: # Proceed only if pipeline is not already initialized print(f"Attempting to initialize LLM: {MODEL_NAME}") if not HF_TOKEN: print("CRITICAL ERROR: HF_TOKEN environment variable not set. Cannot access gated model.") return # Stop initialization if token is missing try: llm_tokenizer = AutoTokenizer.from_pretrained( MODEL_NAME, trust_remote_code=True, token=HF_TOKEN, use_fast=False # Using slow tokenizer as per previous debugging for SentencePiece ) print(f"Tokenizer for {MODEL_NAME} loaded.") model = AutoModelForCausalLM.from_pretrained( MODEL_NAME, torch_dtype=torch.float16, device_map="auto", # Automatically map to available device load_in_8bit=True, # Enable 8-bit quantization; requires bitsandbytes trust_remote_code=True, token=HF_TOKEN ) print(f"Model {MODEL_NAME} loaded.") if llm_tokenizer.pad_token is None: print("Tokenizer pad_token is None, setting to eos_token.") llm_tokenizer.pad_token = llm_tokenizer.eos_token if model.config.pad_token_id is None: # Also update model config if needed model.config.pad_token_id = model.config.eos_token_id print(f"Model config pad_token_id set to: {model.config.pad_token_id}") llm_pipeline = pipeline( "text-generation", model=model, tokenizer=llm_tokenizer ) print(f"LLM pipeline for {MODEL_NAME} initialized successfully.") except Exception as e: print(f"ERROR during LLM initialization ({MODEL_NAME}): {e}") # Ensure these are reset if initialization fails partway llm_pipeline = None llm_tokenizer = None # --- TMDB API Functions --- def search_tmdb_movie_details(title, year): if not TMDB_API_KEY: print("CRITICAL ERROR: 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] 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.3) # Slightly increased delay for API calls except requests.RequestException as e: print(f"TMDB API Error (search) for {title} ({year}): {e}") except Exception as ex: print(f"Unexpected error in TMDB search for {title} ({year}): {ex}") return None def get_tmdb_recommendations(movie_id, page=1): if not TMDB_API_KEY: print("CRITICAL ERROR: 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.3) # Slightly increased delay except requests.RequestException as e: print(f"TMDB API Error (recommendations) for movie ID {movie_id}: {e}") except Exception as ex: print(f"Unexpected error in TMDB recommendations for movie ID {movie_id}: {ex}") return recommendations # --- LLM Explanation --- 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: print("LLM pipeline or tokenizer not available for explanation generation.") return "للأسف، نموذج الذكاء الاصطناعي مو جاهز حالياً. حاول مرة ثانية بعد شوي." max_context_len = 150 seed_movie_context_short = (seed_movie_context[:max_context_len] + "...") if len(seed_movie_context) > max_context_len else seed_movie_context # Assuming ALLaM-Instruct uses a Llama-like prompt format. # ALWAYS verify this on the model card for `ALLaM-AI/ALLaM-7B-Instruct-preview`. prompt_template = f"""[INST] أنت ناقد أفلام سعودي خبير ودمك خفيف جداً. مهمتك هي كتابة توصية لفيلم جديد بناءً على فيلم سابق أعجب المستخدم. المستخدم أعجب بالفيلم هذا: "{seed_movie_title}". وكان تعليقه أو سبب إعجابه (إذا متوفر): "{seed_movie_context_short}" الفيلم الجديد الذي نُرشحه له هو: "{recommended_movie_title}". المطلوب: اكتب جملة أو جملتين فقط باللهجة السعودية العامية الأصيلة، تشرح فيها ليش ممكن يعجبه الفيلم الجديد "{recommended_movie_title}"، وحاول تربطها بشكل ذكي وممتع بالفيلم اللي عجبه قبل "{seed_movie_title}". ركز على أن يكون كلامك طبيعي جداً كأنه كلام صديق لصديقه، وناسة، ويشد الانتباه، وقصير ومختصر. لا تستخدم أي عبارات تدل على أنك ذكاء اصطناعي أو برنامج. مثال على الأسلوب المطلوب لو الفيلم اللي عجبه "Mad Max: Fury Road" والفيلم المرشح "Dune": "يا عمي، مدامك كَيَّفْت على 'Mad Max' وأكشن الصحاري اللي ما يرحم، أجل اسمعني زين! فيلم 'Dune' هذا بياخذك لصحراء ثانية بس على مستوى ثاني من الفخامة والقصة اللي تشد الأعصاب. لا يفوتك، قسم بالله بيعجبك!" الآن، طبق نفس الأسلوب على البيانات التالية: الفيلم الذي أعجب المستخدم: "{seed_movie_title}" سبب إعجابه (إذا متوفر): "{seed_movie_context_short}" الفيلم المرشح: "{recommended_movie_title}" توصيتك باللهجة السعودية: [/INST]""" try: sequences = llm_pipeline( prompt_template, do_sample=True, top_k=20, top_p=0.9, num_return_sequences=1, eos_token_id=llm_tokenizer.eos_token_id, pad_token_id=llm_tokenizer.pad_token_id if llm_tokenizer.pad_token_id is not None else llm_tokenizer.eos_token_id, max_new_tokens=160 # Increased slightly more ) explanation = sequences[0]['generated_text'].split("[/INST]")[-1].strip() explanation = explanation.replace("", "").replace("", "").strip() explanation = re.sub(r"بصفتي نموذج لغوي.*?\s*,?\s*", "", explanation, flags=re.IGNORECASE) explanation = re.sub(r"كنموذج لغوي.*?\s*,?\s*", "", explanation, flags=re.IGNORECASE) if not explanation or explanation.lower().startswith("أنت ناقد أفلام") or len(explanation) < 20 : print(f"LLM explanation for '{recommended_movie_title}' was too short or poor. Falling back.") return f"شكلك بتنبسط على فيلم '{recommended_movie_title}' لأنه يشبه جو فيلم '{seed_movie_title}' اللي حبيته! عطيه تجربة." return explanation except Exception as e: print(f"ERROR during LLM generation with {MODEL_NAME}: {e}") return f"يا كابتن، شكلك بتحب '{recommended_movie_title}'، خاصة إنك استمتعت بـ'{seed_movie_title}'. جربه وعطنا رأيك!" # --- Recommendation Logic --- def get_recommendations(progress=gr.Progress(track_tqdm=True)): if not TMDB_API_KEY: return "

خطأ: مفتاح TMDB API مو موجود أو غير صحيح. الرجاء التأكد من إضافته كـ Secret بشكل صحيح في إعدادات الـ Space.

" if not all([df_profile_global is not None, df_watched_global is not None, seed_movies_global is not None]): # seed_movies_global can be empty list return "

خطأ: فشل في تحميل بيانات المستخدم. تأكد من رفع ملفات CSV بشكل صحيح.

" if llm_pipeline is None: # Ensure LLM is ready initialize_llm() # Try to initialize if it wasn't at startup if llm_pipeline is None: return "

خطأ: فشل في تهيئة نموذج الذكاء الاصطناعي. تأكد من وجود HF_TOKEN صحيح وأن لديك صلاحية الوصول للنموذج المحدد.

" if not seed_movies_global: # Check if seed_movies list is empty after loading return "

ما لقينا أفلام مفضلة أو مقيمة تقييم عالي كفاية عشان نبني عليها توصيات. حاول تقيّم بعض الأفلام!

" progress(0.1, desc="نجمع أفلامك المفضلة...") potential_recs = {} # Limit number of seeds to process to avoid excessive API calls / long processing seeds_to_process = seed_movies_global[:20] if len(seed_movies_global) > 20 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.get('name', 'فيلم غير معروف')}") seed_tmdb_details = search_tmdb_movie_details(seed_movie.get('name'), seed_movie.get('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: try: # Ensure year is a valid integer for tuple creation year_val = int(rec['year']) if rec.get('year') and str(rec['year']).isdigit() else None if year_val is None: continue # Skip if year is invalid rec_tuple = (str(rec['title']), year_val) 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 potential_recs[rec['id']] = { 'movie_info': rec, 'seed_movie_title': seed_movie.get('name'), 'seed_movie_context': seed_movie.get('review_text', '') or seed_movie.get('comment_text', '') } except (ValueError, TypeError) as e: # print(f"Skipping recommendation due to data issue: {rec.get('title')} - {e}") continue if not potential_recs: return "

ما لقينا توصيات جديدة لك حالياً بناءً على أفلامك المفضلة. يمكن شفت كل شيء رهيب! 😉

" # Sort recommendations by TMDB popularity sorted_recs_list = sorted(potential_recs.values(), key=lambda x: x['movie_info'].get('popularity', 0), reverse=True) final_recommendations_data = [] 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 "

ما لقينا توصيات جديدة لك حالياً بعد الفلترة. يمكن شفت كل شيء رهيب! 😉

" output_html = "
" # Main container with some padding 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'] # Fallback for missing posters if not poster_url or "No+Poster" in poster_url or "placeholder.com" in poster_url : poster_url = f"https://via.placeholder.com/300x450.png?text={requests.utils.quote(rec_data['movie_info']['title'])}" output_html += f"""
{rec_data['movie_info']['title']}

{rec_data['movie_info']['title']} ({rec_data['movie_info']['year']})

{explanation}

السبب: حبيّت فيلم {rec_data['seed_movie_title']}

""" output_html += "
" return gr.HTML(output_html) # --- Gradio Interface --- css_theme = """ body { font-family: 'Tajawal', sans-serif; } .gradio-container { font-family: 'Tajawal', sans-serif !important; direction: rtl; max-width: 900px !important; margin: auto !important; } footer { display: none !important; } .gr-button { background-color: #c70039 !important; color: white !important; font-size: 1.2em !important; padding: 12px 24px !important; border-radius: 8px !important; font-weight: bold; } .gr-button:hover { background-color: #a3002f !important; box-shadow: 0 2px 5px rgba(0,0,0,0.2); } h1 { color: #900c3f !important; } .gr-html-output h3 { color: #c70039 !important; } /* Style h3 within the HTML output specifically */ """ # Attempt to load data and LLM at startup data_loaded_successfully = load_all_data() if data_loaded_successfully: print("User data loaded successfully.") # LLM initialization will be attempted when the Gradio app starts, # or on the first click if it failed at startup. # initialize_llm() # Call it here to attempt loading at startup else: print("CRITICAL: Failed to load user data. App functionality will be limited.") # It's better to initialize LLM once the app blocks are defined, # or trigger it on first use if it's very resource-intensive at startup. # For Spaces, startup initialization is fine. with gr.Blocks(theme=gr.themes.Soft(primary_hue="red", secondary_hue="pink", font=[gr.themes.GoogleFont("Tajawal"), "sans-serif"]), css=css_theme) as iface: gr.Markdown( """

🎬 رفيقك السينمائي 🍿

يا هلا بك! اضغط الزر تحت وخلنا نعطيك توصيات أفلام على كيف كيفك، مع شرح بالعامية ليش ممكن تدخل مزاجك.

""" ) recommend_button = gr.Button("عطني توصيات أفلام جديدة!") with gr.Column(elem_id="recommendation-output-column"): # Added elem_id for potential specific styling output_recommendations = gr.HTML(label="👇 توصياتك النارية وصلت 👇") # Initialize LLM when the Blocks context is active, after data loading attempt if data_loaded_successfully: initialize_llm() recommend_button.click(fn=get_recommendations, inputs=None, outputs=[output_recommendations], show_progress="full") gr.Markdown( """

نتمنى لك مشاهدة ممتعة مع رفيقك السينمائي! 🎥✨

""" ) if __name__ == "__main__": # Print warnings if critical secrets are missing when running locally if not TMDB_API_KEY: print("\nCRITICAL WARNING: TMDB_API_KEY environment variable is NOT SET.") print("TMDB API calls will fail. Please set it in your .env file or system environment.\n") if not HF_TOKEN: print("\nCRITICAL WARNING: HF_TOKEN environment variable is NOT SET.") print(f"LLM initialization for gated models like {MODEL_NAME} will fail. Please set it.\n") iface.launch(debug=True) # debug=True for local testing, set to False for production