File size: 3,594 Bytes
bc2882e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 |
# backend/utils/db.py
from simple_salesforce import Salesforce
import os
from datetime import datetime
from dotenv import load_dotenv
# Load environment variables from .env file
load_dotenv()
# Salesforce credentials
SF_USERNAME = os.getenv("SF_USERNAME")
SF_PASSWORD = os.getenv("SF_PASSWORD")
SF_SECURITY_TOKEN = os.getenv("SF_SECURITY_TOKEN")
SF_CLIENT_ID = os.getenv("SF_CLIENT_ID")
SF_CLIENT_SECRET = os.getenv("SF_CLIENT_SECRET")
SF_DOMAIN = os.getenv("SF_DOMAIN", "login") # Default to login
# Validate credentials
if not all([SF_USERNAME, SF_PASSWORD, SF_SECURITY_TOKEN, SF_CLIENT_ID, SF_CLIENT_SECRET]):
raise ValueError("Missing Salesforce credentials in environment variables")
# Initialize Salesforce client
sf = Salesforce(
username=SF_USERNAME,
password=SF_PASSWORD,
security_token=SF_SECURITY_TOKEN,
client_id=SF_CLIENT_ID,
client_secret=SF_CLIENT_SECRET,
domain=SF_DOMAIN
)
def update_leaderboard(player_name: str, drs_out: bool = False, runs: float = 0):
"""
Update the player's leaderboard stats in User_Profile__c.
"""
try:
query = f"SELECT Id, Name__c, Matches__c, LBW_Dismissals__c, Score__c FROM User_Profile__c WHERE Name__c = '{player_name}'"
result = sf.query(query)
if result["totalSize"] > 0:
player = result["records"][0]
player_id = player["Id"]
sf.User_Profile__c.update(player_id, {
"Matches__c": player["Matches__c"] + 1,
"LBW_Dismissals__c": player["LBW_Dismissals__c"] + 1 if drs_out else player["LBW_Dismissals__c"],
"Score__c": player["Score__c"] + runs
})
else:
sf.User_Profile__c.create({
"Name__c": player_name,
"Matches__c": 1,
"LBW_Dismissals__c": 1 if drs_out else 0,
"Score__c": runs
})
except Exception as e:
raise Exception(f"Failed to update leaderboard: {str(e)}")
def get_leaderboard():
"""
Retrieve the leaderboard sorted by score from User_Profile__c.
"""
try:
query = "SELECT Id, Name__c, Matches__c, LBW_Dismissals__c, Score__c FROM User_Profile__c ORDER BY Score__c DESC"
result = sf.query(query)
return result["records"]
except Exception as e:
raise Exception(f"Failed to retrieve leaderboard: {str(e)}")
def store_drs_result(video_url: str, verdict: str, speed: float, replay_link: str, player_name: str) -> str:
"""
Store DRS analysis result in Match__c and DRS_Result__c.
Returns the Match__c record ID.
"""
try:
# Find player (User_Profile__c)
query = f"SELECT Id FROM User_Profile__c WHERE Name__c = '{player_name}'"
result = sf.query(query)
if result["totalSize"] == 0:
raise Exception(f"Player {player_name} not found")
player_id = result["records"][0]["Id"]
# Create Match__c record
match = sf.Match__c.create({
"Video_URL__c": video_url,
"Match_Date__c": datetime.now().isoformat(),
"Player__c": player_id,
"Match_Type__c": "Friendly",
"Result__c": verdict
})
match_id = match["id"]
# Create DRS_Result__c record
sf.DRS_Result__c.create({
"Verdict__c": verdict,
"Speed__c": speed,
"Replay_Link__c": replay_link,
"Match__c": match_id
})
return match_id
except Exception as e:
raise Exception(f"Failed to store DRS result: {str(e)}") |