from web.database import query_db class StatsService: @staticmethod def get_recent_matches(limit=5): sql = """ SELECT m.match_id, m.start_time, m.map_name, m.score_team1, m.score_team2, m.winner_team, p.username as mvp_name FROM fact_matches m LEFT JOIN dim_players p ON m.mvp_uid = p.uid ORDER BY m.start_time DESC LIMIT ? """ return query_db('l2', sql, [limit]) @staticmethod def get_matches(page=1, per_page=20, map_name=None, date_from=None, date_to=None): offset = (page - 1) * per_page args = [] where_clauses = ["1=1"] if map_name: where_clauses.append("map_name = ?") args.append(map_name) if date_from: where_clauses.append("start_time >= ?") args.append(date_from) if date_to: where_clauses.append("start_time <= ?") args.append(date_to) where_str = " AND ".join(where_clauses) sql = f""" SELECT m.match_id, m.start_time, m.map_name, m.score_team1, m.score_team2, m.winner_team, m.duration FROM fact_matches m WHERE {where_str} ORDER BY m.start_time DESC LIMIT ? OFFSET ? """ args.extend([per_page, offset]) matches = query_db('l2', sql, args) # Enrich matches with Avg ELO, Party info, and Our Team Result if matches: match_ids = [m['match_id'] for m in matches] placeholders = ','.join('?' for _ in match_ids) # Fetch ELO elo_sql = f""" SELECT match_id, AVG(group_origin_elo) as avg_elo FROM fact_match_teams WHERE match_id IN ({placeholders}) AND group_origin_elo > 0 GROUP BY match_id """ elo_rows = query_db('l2', elo_sql, match_ids) elo_map = {row['match_id']: row['avg_elo'] for row in elo_rows} # Fetch Max Party Size party_sql = f""" SELECT match_id, MAX(cnt) as max_party FROM ( SELECT match_id, match_team_id, COUNT(*) as cnt FROM fact_match_players WHERE match_id IN ({placeholders}) AND match_team_id > 0 GROUP BY match_id, match_team_id ) GROUP BY match_id """ party_rows = query_db('l2', party_sql, match_ids) party_map = {row['match_id']: row['max_party'] for row in party_rows} # --- New: Determine "Our Team" Result --- # Logic: Check if any player from `active_roster` played in these matches. # Use WebService to get the active roster from web.services.web_service import WebService import json lineups = WebService.get_lineups() active_roster_ids = [] if lineups: try: # Load IDs and ensure they are all strings for DB comparison consistency raw_ids = json.loads(lineups[0]['player_ids_json']) active_roster_ids = [str(uid) for uid in raw_ids] except: pass # If no roster, we can't determine "Our Result" if not active_roster_ids: result_map = {} else: roster_placeholders = ','.join('?' for _ in active_roster_ids) # We cast steam_id_64 to TEXT to ensure match even if stored as int our_result_sql = f""" SELECT mp.match_id, mp.team_id, m.winner_team, COUNT(*) as our_count FROM fact_match_players mp JOIN fact_matches m ON mp.match_id = m.match_id WHERE mp.match_id IN ({placeholders}) AND CAST(mp.steam_id_64 AS TEXT) IN ({roster_placeholders}) GROUP BY mp.match_id, mp.team_id """ # Combine args: match_ids + roster_ids combined_args = match_ids + active_roster_ids our_rows = query_db('l2', our_result_sql, combined_args) # Map match_id -> result ('win', 'loss', 'draw', 'mixed') result_map = {} match_sides = {} match_winners = {} for r in our_rows: mid = r['match_id'] if mid not in match_sides: match_sides[mid] = {} match_sides[mid][r['team_id']] = r['our_count'] match_winners[mid] = r['winner_team'] for mid, sides in match_sides.items(): winner = match_winners.get(mid) if not winner: result_map[mid] = 'draw' continue our_on_winner = sides.get(winner, 0) loser = 2 if winner == 1 else 1 our_on_loser = sides.get(loser, 0) if our_on_winner > 0 and our_on_loser == 0: result_map[mid] = 'win' elif our_on_loser > 0 and our_on_winner == 0: result_map[mid] = 'loss' elif our_on_winner > 0 and our_on_loser > 0: result_map[mid] = 'mixed' else: result_map[mid] = None # Convert to dict to modify matches = [dict(m) for m in matches] for m in matches: m['avg_elo'] = elo_map.get(m['match_id'], 0) m['max_party'] = party_map.get(m['match_id'], 1) m['our_result'] = result_map.get(m['match_id']) # Count total for pagination count_sql = f"SELECT COUNT(*) as cnt FROM fact_matches WHERE {where_str}" total = query_db('l2', count_sql, args[:-2], one=True)['cnt'] return matches, total @staticmethod def get_match_detail(match_id): sql = "SELECT * FROM fact_matches WHERE match_id = ?" return query_db('l2', sql, [match_id], one=True) @staticmethod def get_match_players(match_id): sql = """ SELECT mp.*, p.username, p.avatar_url FROM fact_match_players mp LEFT JOIN dim_players p ON mp.steam_id_64 = p.steam_id_64 WHERE mp.match_id = ? ORDER BY mp.team_id, mp.rating DESC """ return query_db('l2', sql, [match_id]) @staticmethod def get_match_rounds(match_id): sql = "SELECT * FROM fact_rounds WHERE match_id = ? ORDER BY round_num" return query_db('l2', sql, [match_id]) @staticmethod def get_players(page=1, per_page=20, search=None, sort_by='rating_desc'): offset = (page - 1) * per_page args = [] where_clauses = ["1=1"] if search: # Force case-insensitive search where_clauses.append("(LOWER(username) LIKE LOWER(?) OR steam_id_64 LIKE ?)") args.append(f"%{search}%") args.append(f"%{search}%") where_str = " AND ".join(where_clauses) # Sort mapping order_clause = "rating DESC" # Default logic (this query needs refinement as L2 dim_players doesn't store avg rating) # Wait, dim_players only has static info. We need aggregated stats. # Ideally, we should fetch from L3 for player list stats. # But StatsService is for L2. # For the Player List, we usually want L3 data (Career stats). # I will leave the detailed stats logic for FeatureService or do a join here if necessary. # For now, just listing players from dim_players. sql = f""" SELECT * FROM dim_players WHERE {where_str} LIMIT ? OFFSET ? """ args.extend([per_page, offset]) players = query_db('l2', sql, args) total = query_db('l2', f"SELECT COUNT(*) as cnt FROM dim_players WHERE {where_str}", args[:-2], one=True)['cnt'] return players, total @staticmethod def get_player_info(steam_id): sql = "SELECT * FROM dim_players WHERE steam_id_64 = ?" return query_db('l2', sql, [steam_id], one=True) @staticmethod def get_daily_match_counts(days=365): # Return list of {date: 'YYYY-MM-DD', count: N} sql = """ SELECT date(start_time, 'unixepoch') as day, COUNT(*) as count FROM fact_matches WHERE start_time > strftime('%s', 'now', ?) GROUP BY day ORDER BY day """ # sqlite modifier for 'now' needs format like '-365 days' modifier = f'-{days} days' rows = query_db('l2', sql, [modifier]) return rows @staticmethod def get_players_by_ids(steam_ids): if not steam_ids: return [] placeholders = ','.join('?' for _ in steam_ids) sql = f"SELECT * FROM dim_players WHERE steam_id_64 IN ({placeholders})" return query_db('l2', sql, steam_ids) @staticmethod def get_player_basic_stats(steam_id): # Calculate stats from fact_match_players # Prefer calculating from sums (kills/deaths) for K/D accuracy # AVG(adr) is used as damage_total might be missing in some sources sql = """ SELECT AVG(rating) as rating, SUM(kills) as total_kills, SUM(deaths) as total_deaths, AVG(kd_ratio) as avg_kd, AVG(kast) as kast, AVG(adr) as adr, COUNT(*) as matches_played FROM fact_match_players WHERE steam_id_64 = ? """ row = query_db('l2', sql, [steam_id], one=True) if row and row['matches_played'] > 0: res = dict(row) # Calculate K/D: Sum Kills / Sum Deaths kills = res.get('total_kills') or 0 deaths = res.get('total_deaths') or 0 if deaths > 0: res['kd'] = kills / deaths else: res['kd'] = kills # If 0 deaths, K/D is kills (or infinity, but kills is safer for display) # Fallback to avg_kd if calculation failed (e.g. both 0) but avg_kd exists if res['kd'] == 0 and res['avg_kd'] and res['avg_kd'] > 0: res['kd'] = res['avg_kd'] # ADR validation if res['adr'] is None: res['adr'] = 0.0 return res return None @staticmethod def get_shared_matches(steam_ids): # Find matches where ALL steam_ids were present if not steam_ids or len(steam_ids) < 1: return [] placeholders = ','.join('?' for _ in steam_ids) count = len(steam_ids) # We need to know which team the players were on to determine win/loss # Assuming they were on the SAME team for "shared experience" # If count=1, it's just match history # Query: Get matches where all steam_ids are present # Also join to get team_id to check if they were on the same team (optional but better) # For simplicity in v1: Just check presence in the match. # AND check if the player won. # We need to return: match_id, map_name, score, result (Win/Loss) # "Result" is relative to the lineup. # If they were on the winning team, it's a Win. sql = f""" SELECT m.match_id, m.start_time, m.map_name, m.score_team1, m.score_team2, m.winner_team, MAX(mp.team_id) as player_team_id -- Just take one team_id (assuming same) FROM fact_matches m JOIN fact_match_players mp ON m.match_id = mp.match_id WHERE mp.steam_id_64 IN ({placeholders}) GROUP BY m.match_id HAVING COUNT(DISTINCT mp.steam_id_64) = ? ORDER BY m.start_time DESC LIMIT 20 """ args = list(steam_ids) args.append(count) rows = query_db('l2', sql, args) results = [] for r in rows: # Determine if Win # winner_team in DB is 'Team 1' or 'Team 2' usually, or the team name. # fact_matches.winner_team stores the NAME of the winner? Or 'team1'/'team2'? # Let's check how L2_Builder stores it. Usually it stores the name. # But fact_match_players.team_id stores the name too. # Logic: If m.winner_team == mp.team_id, then Win. is_win = (r['winner_team'] == r['player_team_id']) # If winner_team is NULL or empty, it's a draw? if not r['winner_team']: result_str = 'Draw' elif is_win: result_str = 'Win' else: result_str = 'Loss' res = dict(r) res['is_win'] = is_win # Boolean for styling res['result_str'] = result_str # Text for display results.append(res) return results @staticmethod def get_player_trend(steam_id, limit=20): # We need party_size: count of players with same match_team_id in the same match # Using a correlated subquery for party_size sql = """ SELECT * FROM ( SELECT m.start_time, mp.rating, mp.kd_ratio, mp.adr, m.match_id, m.map_name, mp.is_win, mp.match_team_id, (SELECT COUNT(*) FROM fact_match_players p2 WHERE p2.match_id = mp.match_id AND p2.match_team_id = mp.match_team_id AND p2.match_team_id > 0 -- Ensure we don't count 0 (solo default) as a massive party ) as party_size FROM fact_match_players mp JOIN fact_matches m ON mp.match_id = m.match_id WHERE mp.steam_id_64 = ? ORDER BY m.start_time DESC LIMIT ? ) ORDER BY start_time ASC """ return query_db('l2', sql, [steam_id, limit]) @staticmethod def get_live_matches(): # Query matches started in last 2 hours with no winner # Assuming we have a way to ingest live matches. # For now, this query is 'formal' but will likely return empty on static dataset. sql = """ SELECT m.match_id, m.map_name, m.score_team1, m.score_team2, m.start_time FROM fact_matches m WHERE m.winner_team IS NULL AND m.start_time > strftime('%s', 'now', '-2 hours') """ return query_db('l2', sql)