import base64
import json
import os
import uuid
from datetime import datetime
from pathlib import Path
import fcntl
import pandas as pd
import pytz
import streamlit as st
from datasets import Dataset, load_dataset, Features, Value
from huggingface_hub import CommitScheduler
# File paths as constants
USERS_JSON = 'leaders/users.json'
MATCHES_JSON = 'matches.json'
OUTCOMES_JSON = 'match_outcomes.json'
OUTCOMES = 'outcomes/match_outcomes.json'
PLAYERS_JSON = 'players.json'
image_path = 'ipl_image.png'
PREDICTIONS_FOLDER = Path("predictions")
PREDICTIONS_FOLDER.mkdir(parents=True, exist_ok=True)
users_file = Path("leaders") / f"users.json"
USERS_FOLDER = users_file.parent
USERS_FOLDER.mkdir(parents=True, exist_ok=True)
outcomes_file = Path("outcomes") / f"match_outcomes.json"
OUTCOMES_FOLDER = outcomes_file.parent
OUTCOMES_FOLDER.mkdir(parents=True, exist_ok=True)
# Initialize CommitScheduler
scheduler = CommitScheduler(
repo_id="DIS_IPL_Preds",
repo_type="dataset",
folder_path=PREDICTIONS_FOLDER, # Local folder where predictions are saved temporarily
path_in_repo="predictions", # Path in dataset repo where predictions will be saved
every=720, # Push every 240 minutes (4 hours)
)
# Initialize CommitScheduler
scheduler = CommitScheduler(
repo_id="DIS_IPL_Leads",
repo_type="dataset",
folder_path=USERS_FOLDER, # Local folder where users are saved temporarily
path_in_repo="leaders", # Path in dataset repo where predictions will be saved
every=720, # Push every 240 minutes (4 hours)
)
# Initialize CommitScheduler
scheduler = CommitScheduler(
repo_id="DIS_IPL_Outcomes",
repo_type="dataset",
folder_path=OUTCOMES_FOLDER, # Local folder where users are saved temporarily
path_in_repo="outcomes", # Path in dataset repo where predictions will be saved
every=720, # Push every 240 minutes (4 hours)
)
def load_data(file_path):
"""
Load data from a JSON or CSV file with better error handling.
Args:
file_path (str): The path to the file to load.
Returns:
dict or pd.DataFrame: The loaded data. Returns empty dict/DataFrame on error.
"""
try:
if file_path.endswith('.json'):
with open(file_path, 'r', encoding='utf-8') as file:
try:
return json.load(file)
except json.JSONDecodeError:
st.error(f"Invalid JSON in {file_path}. Loading empty data.")
return {}
elif file_path.endswith('.csv'):
try:
return pd.read_csv(file_path)
except Exception as e:
st.error(f"Error reading CSV {file_path}: {e}")
return pd.DataFrame()
except FileNotFoundError:
if file_path.endswith('.json'):
return {}
elif file_path.endswith('.csv'):
return pd.DataFrame()
except Exception as e:
st.error(f"Unexpected error loading {file_path}: {e}")
if file_path.endswith('.json'):
return {}
return pd.DataFrame()
def get_base64_of_image(path):
with open(path, "rb") as image_file:
return base64.b64encode(image_file.read()).decode()
# Get today's date in IST to load today's match
def get_current_date_ist():
tz_IST = pytz.timezone('Asia/Kolkata')
datetime_ist = datetime.now(tz_IST)
return datetime_ist.strftime('%Y-%m-%d')
# Function to get matches for today
def get_today_matches():
today = get_current_date_ist()
matches = load_data(MATCHES_JSON)
today_matches = [match for match in matches if match['date'] == today]
return today_matches
# Function to check if prediction submission is allowed
def is_submission_allowed(match_id):
matches = load_data(MATCHES_JSON) # This loads matches correctly with IST times
for match in matches:
if match["match_id"] == match_id:
# Parse the match start time in IST
tz_IST = pytz.timezone('Asia/Kolkata')
match_datetime_str = f'{match["date"]} {match["time"]}'
# The match time string is like "2024-03-21 7:30 PM"
match_datetime = datetime.strptime(match_datetime_str, "%Y-%m-%d %I:%M %p")
match_datetime = tz_IST.localize(match_datetime) # Set the timezone to IST
# Get the current time in IST
current_datetime = datetime.now(tz_IST)
if current_datetime > match_datetime:
return False
else:
return True
return False # If match_id not found, default to False
# Submit prediction function
def submit_prediction(
user_name,
match_id,
predicted_winner,
predicted_motm,
bid_points,
max_bid_points
):
# Validation for user selection
if user_name == "Select a user...":
st.warning("Please select a valid user.")
return
# Check if prediction submission is allowed for the match
if not is_submission_allowed(match_id):
st.error("Prediction submission time has passed. Predictions can't be submitted after match start.")
return
if bid_points > max_bid_points or bid_points < 100:
st.error(f"Oops, Invalid bid! 🙈 You must bid at least 100 points and not exceed the 20% limit of your total points. Maximum allowed bid points: {max_bid_points}.")
# st.error(f"Oops! 🙈 Looks like you're going overboard with your bid points! Your bid points cannot exceed your total points. Maximum allowed bid points: {max_bid_points} 😱")
return
prediction_id = uuid.uuid4().hex
prediction_time = datetime.now().strftime('%Y-%m-%d')
prediction_data = {
'prediction_id': prediction_id,
'user_name': user_name,
'match_id': match_id,
'predicted_winner': predicted_winner,
'predicted_motm': predicted_motm,
'bid_points': bid_points,
'prediction_date': prediction_time # Include the prediction time
}
# Construct the filename to include match_id for easier retrieval
prediction_file_name = f"prediction_{match_id}_{user_name}.json"
prediction_file = PREDICTIONS_FOLDER / prediction_file_name
# Load existing predictions for the user and match, if any
existing_predictions = []
if prediction_file.exists():
with prediction_file.open("r") as file:
for line in file:
existing_predictions.append(json.loads(line.strip()))
# Update existing prediction if it exists for the same user and match
prediction_updated = False
for existing_prediction in existing_predictions:
if existing_prediction['user_name'] == user_name and existing_prediction['match_id'] == match_id:
existing_prediction.update(prediction_data)
prediction_updated = True
break
# Save the updated predictions back to the file
with scheduler.lock:
if not prediction_updated:
# Append the new prediction if it doesn't already exist
with prediction_file.open("a") as file:
file.write(json.dumps(prediction_data))
file.write("\n")
else:
with prediction_file.open("w") as file:
for prediction in existing_predictions:
file.write(json.dumps(prediction))
file.write("\n")
st.success("Prediction submitted successfully!")
def get_user_total_points(user_name):
# users_dataset = load_dataset("Jay-Rajput/DIS_IPL_Leads", split="train")
# users = users_dataset.to_dict()
users = load_users(USERS_JSON)
return users.get(user_name, {}).get('points')
def calculate_max_bid_points(user_name):
total_points = get_user_total_points(user_name)
max_bid_points = int(total_points * 0.20) # 20% of total points
return max_bid_points
def load_users(users_json_path):
"""Load users data with file locking to prevent corruption"""
try:
if not os.path.exists(users_json_path):
return {}
with open(users_json_path, 'r', encoding='utf-8') as f:
# Acquire shared lock for reading
fcntl.flock(f, fcntl.LOCK_SH)
try:
data = json.load(f)
# Validate data structure
if not all(isinstance(v, dict) and 'points' in v for v in data.values()):
st.error("Invalid users data structure")
return {}
return data
except json.JSONDecodeError:
st.error("Invalid JSON in users file")
return {}
finally:
fcntl.flock(f, fcntl.LOCK_UN)
except Exception as e:
st.error(f"Error loading users: {e}")
return {}
def user_selection_and_prediction():
try:
users_data = load_users(USERS_JSON) # Use our improved load_users function
if not users_data:
st.error("Failed to load users data. Please try again later.")
return
users = list(users_data.keys())
user_name = st.selectbox("Select User", ["Select a user..."] + users)
max_bid_points = None
if user_name != "Select a user...":
max_bid_points = calculate_max_bid_points(user_name)
st.write(f"Maximum bid points you can submit: {max_bid_points}")
matches = get_today_matches()
if matches:
match_choice = st.selectbox(
"Select Today's Match",
matches,
format_func=lambda match: f"{match['teams'][0]} vs {match['teams'][1]}"
)
match_id = match_choice['match_id']
teams = match_choice['teams']
predicted_winner = st.selectbox("Predicted Winner", teams)
player_list = load_data(PLAYERS_JSON)
predicted_motm = ""
if predicted_winner in player_list:
players = player_list[predicted_winner]
predicted_motm = st.selectbox("Predicted Man of the Match", players)
bid_points = st.number_input("Bid Points", min_value=0, value=100, format="%d")
if st.button("Submit Prediction"):
submit_prediction(
user_name,
match_id,
predicted_winner,
predicted_motm,
bid_points,
max_bid_points
)
else:
st.write("No matches are scheduled for today.")
except Exception as e:
st.error(f"Error in prediction submission: {e}")
def display_predictions():
if st.button("Show Predictions"):
all_predictions = []
# Check if the directory exists
if not os.path.exists(PREDICTIONS_FOLDER):
st.write("No predictions directory found.")
return
# List all JSON files in the directory
for filename in os.listdir(PREDICTIONS_FOLDER):
if filename.endswith('.json'):
file_path = os.path.join(PREDICTIONS_FOLDER, filename)
# Read each JSON file and append its contents to the list
with open(file_path, 'r') as file:
prediction = json.load(file)
all_predictions.append(prediction)
# Convert the list of dictionaries to a DataFrame
predictions_df = pd.DataFrame(all_predictions)
if not predictions_df.empty:
predictions_df['prediction_date'] = predictions_df.apply(lambda x: datetime.strptime(x['prediction_date'], '%Y-%m-%d'), axis=1)
# Filter for today's predictions
today_str = datetime.now().strftime('%Y-%m-%d')
todays_predictions = predictions_df[predictions_df['prediction_date'] == today_str]
# Remove the 'prediction_id' column if it exists
if 'prediction_id' in todays_predictions.columns:
todays_predictions = todays_predictions.drop(columns=['prediction_id', 'prediction_date'])
st.dataframe(todays_predictions, hide_index=True)
else:
st.write("No predictions for today's matches yet.")
def display_leaderboard():
if st.button("Show Leaderboard"):
try:
# Load users data using our centralized function
users_data = []
data = load_users(USERS_JSON)
for user, user_info in data.items():
points = user_info.get("points", 0)
last_5_results = " ".join(user_info.get("last_5_results", ["⚪"] * 5))
users_data.append({
'User': user,
'Points': points,
"Last 5 Bids": last_5_results
})
if not users_data:
st.warning("No user data found!")
return
leaderboard = pd.DataFrame(users_data)
# Sort DataFrame by points in descending order
leaderboard = leaderboard.sort_values(by='Points', ascending=False)
# Add a 'Rank' column starting from 1
leaderboard['Rank'] = range(1, len(leaderboard) + 1)
# Select and order the columns for display
leaderboard = leaderboard[['Rank', 'User', 'Points', 'Last 5 Bids']]
st.dataframe(leaderboard, hide_index=True)
except Exception as e:
st.error(f"Failed to load leaderboard data: {str(e)}")
# Streamlit UI
encoded_image = get_base64_of_image(image_path)
custom_css = f"""
"""
# Apply custom CSS
st.markdown(custom_css, unsafe_allow_html=True)
# Use the custom class in a div with your title
st.markdown('', unsafe_allow_html=True)
st.write("🏆 Predict, Compete, and Win 🏏 - Where Every Guess Counts! 🏆")
user_guide_content = """
### 📘 User Guide
#### Submitting Predictions
- **Match Selection**: Choose the match you want to predict from today's available matches.
- **Team and Player Prediction**: Select the team you predict will win and the "Man of the Match".
- **Bid Points**: Enter the number of points you wish to bid on your prediction. Remember, the maximum you can bid is capped at **20% of your total points**.
#### Scoring System
- **Winning Team Prediction**:
- ✅ **Correct Prediction**: You earn **2000 points** plus your bid amount.
- ❌ **Incorrect Prediction**: You lose **200 points** plus your bid amount.
- **Man of the Match Prediction**:
- ✅ **Correct Prediction**: You earn **an additional 500 points**.
- ❌ **Incorrect Prediction**: No penalty.
- **No Prediction Submitted**:
- ❌ **You lose 1000 points** automatically for not submitting a prediction.
#### Bid Point Constraints
- You cannot bid more than 20% of your current total points.
- Bid points will be doubled if your prediction is correct, and deducted if incorrect.
#### Rules for Submission
- **Predictions must be submitted before the match starts**.
- **Only one prediction per match is allowed**.
- **Review your prediction carefully before submission, as it cannot be changed once submitted**.
#### 🔴🟢⚪ Match Performance Tracking
- After each match, your last **5 predictions will be tracked** and displayed on the leaderboard:
- 🟢 **Green** → Correct prediction.
- 🔴 **Red** → Wrong prediction.
- ⚪ **White** → No prediction submitted.
🚀 **Compete, strategize, and climb the leaderboard!**
"""
# User Guide as an expander
with st.expander("User Guide 📘"):
st.markdown(user_guide_content)
with st.expander("Submit Prediction 📝"):
user_selection_and_prediction()
with st.expander("Predictions 🔍"):
display_predictions()
with st.expander("Leaderboard 🏆"):
display_leaderboard()
############################# Admin Panel ##################################
ADMIN_PASSPHRASE = "admin123"
def fetch_latest_predictions(match_id):
# Load the dataset. Adjust "split" to "train" or appropriate if "predictions" is a configuration.
dataset = load_dataset("Jay-Rajput/DIS_IPL_Preds", split="train")
# Convert the dataset to a pandas DataFrame
df = pd.DataFrame(dataset)
# Ensure the DataFrame is not empty and contains the required columns
if not df.empty and {'user_name', 'match_id'}.issubset(df.columns):
# Filter rows by 'match_id'
filtered_df = df[df['match_id'] == match_id]
# Drop duplicate rows based on 'user_name'
unique_df = filtered_df.drop_duplicates(subset=['user_name'])
return unique_df
else:
return pd.DataFrame()
def load_and_process_users():
"""Load and process users data from Hugging Face"""
try:
# Load dataset from Hugging Face
users = load_dataset("Jay-Rajput/DIS_IPL_Leads", split="train")
users_df = users.to_dict()
# Process into the correct format
users_dict = {}
for user_name, user_data in users_df.items():
users_dict[user_name] = user_data[0] # Extract the first (and only) item
# Save to local JSON file for faster access
with open(USERS_JSON, 'w', encoding='utf-8') as f:
json.dump(users_dict, f, ensure_ascii=False, indent=4)
return users_dict
except Exception as e:
st.error(f"Error loading users data: {e}")
return {}
def update_leaderboard_and_outcomes(match_id, winning_team, man_of_the_match, outcome_only=False):
try:
# First load and process the latest users data
users_dict = load_and_process_users()
# Rest of your existing code for updating outcomes...
features = Features({
'match_id': Value('string'),
'man_of_the_match': Value('string'),
'winning_team': Value('string'),
})
# Load existing match outcomes
outcomes = load_dataset("Jay-Rajput/DIS_IPL_Outcomes", split="train")
outcomes_df = pd.DataFrame(outcomes)
# Update or add the match outcome
outcome_exists = False
for idx, outcome in outcomes_df.iterrows():
if outcome['match_id'] == match_id:
outcomes_df.at[idx, 'winning_team'] = winning_team
outcomes_df.at[idx, 'man_of_the_match'] = man_of_the_match
outcome_exists = True
break
if not outcome_exists:
new_outcome = {"match_id": match_id, "winning_team": winning_team, "man_of_the_match": man_of_the_match}
outcomes_df = pd.concat([outcomes_df, pd.DataFrame([new_outcome])], ignore_index=True)
outcomes = Dataset.from_pandas(outcomes_df)
if not outcome_only:
# Load predictions
predictions = fetch_latest_predictions(match_id)
# Update user points based on prediction accuracy
users_with_predictions = set(predictions['user_name'])
for user_name, user_data in users_dict.items():
if user_name in users_with_predictions:
prediction = predictions[predictions['user_name'] == user_name].iloc[0]
predicted_winner = prediction['predicted_winner']
predicted_motm = prediction['predicted_motm']
bid_points = prediction['bid_points']
# Update points based on prediction accuracy
if predicted_winner == winning_team:
user_data['points'] += 2000 + bid_points
result_indicator = "🟢"
if predicted_motm == man_of_the_match:
user_data['points'] += 500
else:
user_data['points'] -= 200 + bid_points
result_indicator = "🔴"
else:
user_data['points'] -= 1000
result_indicator = "⚪"
user_data['points'] = max(user_data['points'], 0)
# Update last 5 results
if "last_5_results" not in user_data:
user_data["last_5_results"] = []
user_data["last_5_results"].insert(0, result_indicator)
user_data["last_5_results"] = user_data["last_5_results"][:5]
# Save updated users data back to Hugging Face
users_dataset = Dataset.from_dict({k: [v] for k, v in users_dict.items()})
users_dataset.push_to_hub("Jay-Rajput/DIS_IPL_Leads")
# Also save locally
with open(USERS_JSON, 'w', encoding='utf-8') as f:
json.dump(users_dict, f, ensure_ascii=False, indent=4)
# Save outcomes
outcomes.to_json(OUTCOMES)
outcomes.push_to_hub("Jay-Rajput/DIS_IPL_Outcomes", split="train")
return True
except Exception as e:
st.error(f"Error updating leaderboard: {e}")
return False
# Function to fetch matches for a given date
def fetch_matches_by_date(matches, selected_date):
return [match for match in matches if datetime.strptime(match['date'], '%Y-%m-%d').date() == selected_date]
with st.sidebar:
expander = st.expander("Admin Panel", expanded=False)
admin_pass = expander.text_input("Enter admin passphrase:", type="password", key="admin_pass")
if admin_pass == ADMIN_PASSPHRASE:
expander.success("Authenticated")
all_matches = load_data(MATCHES_JSON)
match_outcomes = load_dataset("Jay-Rajput/DIS_IPL_Outcomes", split="train")
submitted_match_ids = [outcome["match_id"] for outcome in match_outcomes]
# Filter matches to those that do not have outcomes submitted yet
matches_without_outcomes = [match for match in all_matches if match["match_id"] not in submitted_match_ids]
# If matches are available, let the admin select one
if matches_without_outcomes:
# Optional: Allow the admin to filter matches by date
selected_date = expander.date_input("Select Match Date", key="match_date")
if selected_date:
filtered_matches = fetch_matches_by_date(matches_without_outcomes, selected_date)
else:
filtered_matches = matches_without_outcomes
if filtered_matches:
match_selection = expander.selectbox("Select Match", filtered_matches, format_func=lambda match: f"{match['teams'][0]} vs {match['teams'][1]} (Match ID: {match['match_id']})", key="match_selection")
selected_match_id = match_selection['match_id']
teams = match_selection['teams']
# Let admin select the winning team
winning_team = expander.selectbox("Winning Team", teams, key="winning_team")
# Fetch and display players for the selected winning team
player_list = load_data(PLAYERS_JSON)
if winning_team in player_list:
players = player_list[winning_team]
man_of_the_match = expander.selectbox("Man of the Match", players, key="man_of_the_match")
else:
players = []
man_of_the_match = expander.text_input("Man of the Match (Type if not listed)", key="man_of_the_match_fallback")
# Add checkbox for outcome only submission
outcome_only = expander.checkbox("Submit Outcome Only", key="outcome_only_checkbox")
if expander.button("Submit Match Outcome", key="submit_outcome"):
if outcome_only:
# Submit match outcome without updating user scores
update_leaderboard_and_outcomes(selected_match_id, winning_team, man_of_the_match, outcome_only=True)
expander.success("Match outcome submitted!")
else:
# Submit match outcome and update user scores
update_leaderboard_and_outcomes(selected_match_id, winning_team, man_of_the_match)
expander.success("Match outcome submitted and leaderboard updated!")
st.experimental_rerun()
else:
expander.write("No matches available for the selected date.")
else:
expander.write("No matches are available for today.")
else:
if admin_pass: # Show error only if something was typed
expander.error("Not authenticated")