import os import re import json import time import datetime as dt from typing import Any, Dict, List, Optional, Tuple, Iterator import requests import gradio as gr # --------------------------- # CONFIG # --------------------------- TZ = "Europe/Sarajevo" API_BASE = "https://v1.hockey.api-sports.io" APISPORTS_KEY = (os.getenv("APISPORTS_KEY") or "").strip() GEMINI_API_KEY = (os.getenv("GEMINI_API_KEY") or "").strip() GEMINI_MODEL = (os.getenv("GEMINI_MODEL") or "gemini-1.5-flash-001").strip() ADMIN_PIN = (os.getenv("ADMIN_PIN") or "").strip() # Dataset persistence (Secrets) HF_TOKEN = (os.getenv("HF_TOKEN") or "").strip() DATASET_REPO = (os.getenv("DATASET_REPO") or "").strip() # e.g. "username/plexbet-storage" (private dataset) FAV_MAX_ODDS = 1.80 LOW_ODDS_MIN = 1.10 LOW_ODDS_MAX = 1.44 HIGH_ODDS_MIN = 1.45 HIGH_ODDS_MAX = 1.80 H2H_LAST_N = 5 FINISHED_STATUS = {"FT", "AOT", "AP"} # finished statuses RESULT_CACHE_TTL_SEC = 10 * 3600 # 10 hours # Quality tuning TOP_RANK = 3 BOTTOM_RANK = 10 MIN_RANK_GAP_FOR_STRONG = 6 MIN_GDPG_GAP_FOR_STRONG = 0.35 MIN_GFP_GAP_FOR_STRONG = 0.30 MIN_GAPG_GAP_FOR_STRONG = 0.30 ALLOW_STAT_OVERRIDE = True OVERRIDE_ONLY_IF_ODDS_NOT_EXTREME = True TARGET_LEAGUES = [ ("Norway", ["EliteHockey", "EHL"]), ("Norway", ["Division 1"]), ("Italy", ["IHL"]), ("Czech-Republic", ["Extraliga"]), ("Czech-Republic", ["Chance Liga", "Maxa", "1 Liga"]), ("Sweden", ["SHL"]), ("Finland", ["Liiga"]), ("United-Kingdom", ["Elite League", "EIHL"]), ("Switzerland", ["National League"]), ("Switzerland", ["Swiss League"]), ("Slovakia", ["Tipos", "Extraliga"]), ("Germany", ["DEL"]), ("Germany", ["DEL2", "DEL 2"]), ("Austria", ["ICE", "ICE Hockey League", "Alps Hockey League"]), ("Denmark", ["Metal Ligaen"]), ("France", ["Ligue Magnus"]), ("Hungary", ["Erste Liga"]), ("Belarus", ["Extraleague", "Extraliga"]), ("Russia", ["KHL"]), ("Russia", ["VHL"]), ("Russia", ["MHL"]), ("Kazakhstan", ["Championship"]), ] # --------------------------- # LOCAL PATHS (ephemeral) but synced to dataset if configured # --------------------------- LOCAL_DIR = "./data" os.makedirs(LOCAL_DIR, exist_ok=True) CACHE_DIR = os.path.join(LOCAL_DIR, "cache") os.makedirs(CACHE_DIR, exist_ok=True) HISTORY_PATH = os.path.join(LOCAL_DIR, "bet_history.json") def _dataset_enabled() -> bool: return bool(HF_TOKEN and DATASET_REPO) def _relpath_in_repo(local_path: str) -> str: rel = os.path.relpath(local_path, LOCAL_DIR).replace("\\", "/") return f"data/{rel}" def _hub_download(relpath: str) -> Optional[str]: if not _dataset_enabled(): return None try: from huggingface_hub import hf_hub_download fp = hf_hub_download( repo_id=DATASET_REPO, repo_type="dataset", filename=relpath, token=HF_TOKEN, ) return fp except Exception: return None def _hub_upload(local_file: str, relpath: str) -> bool: if not _dataset_enabled(): return False try: from huggingface_hub import HfApi api = HfApi(token=HF_TOKEN) api.upload_file( path_or_fileobj=local_file, path_in_repo=relpath, repo_id=DATASET_REPO, repo_type="dataset", commit_message=f"Update {relpath}", ) return True except Exception: return False def atomic_write_json(path: str, obj: Any) -> None: os.makedirs(os.path.dirname(path), exist_ok=True) tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(obj, f, ensure_ascii=False) os.replace(tmp, path) if _dataset_enabled(): rel = _relpath_in_repo(path) up_tmp = path + ".upload" with open(up_tmp, "w", encoding="utf-8") as f: json.dump(obj, f, ensure_ascii=False) _hub_upload(up_tmp, rel) try: os.remove(up_tmp) except Exception: pass def read_json(path: str, default: Any): if os.path.isfile(path): try: with open(path, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass if _dataset_enabled(): rel = _relpath_in_repo(path) fp = _hub_download(rel) if fp and os.path.isfile(fp): try: with open(fp, "r", encoding="utf-8") as f: data = json.load(f) os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w", encoding="utf-8") as out: json.dump(data, out, ensure_ascii=False) return data except Exception: return default return default # --------------------------- # TTL cache for API responses # --------------------------- class TTLCache: def __init__(self, ttl_sec: int = 60): self.ttl = ttl_sec self.store: Dict[str, Tuple[float, Any]] = {} def get(self, key: str): v = self.store.get(key) if not v: return None ts, val = v if time.time() - ts > self.ttl: self.store.pop(key, None) return None return val def set(self, key: str, val: Any): self.store[key] = (time.time(), val) cache = TTLCache(60) floor_cache = TTLCache(6 * 3600) # --------------------------- # Helpers # --------------------------- _non = re.compile(r"[^a-z0-9]+") def norm(s: str) -> str: s = (s or "").lower().strip() s = _non.sub(" ", s) s = re.sub(r"\s+", " ", s).strip() return s def jaccard(a: str, b: str) -> float: A = set(norm(a).split()) B = set(norm(b).split()) if not A or not B: return 0.0 return len(A & B) / len(A | B) def to_float(x: Any) -> Optional[float]: try: if x is None: return None if isinstance(x, (int, float)): return float(x) s = str(x).strip().replace(",", ".") return float(s) except Exception: return None def now_ts() -> int: return int(time.time()) def date_str_for_choice(choice: str) -> str: base = dt.date.today() if choice == "sutra": base = base + dt.timedelta(days=1) return base.strftime("%Y-%m-%d") def try_acquire_lock(lock_path: str) -> bool: try: fd = os.open(lock_path, os.O_CREAT | os.O_EXCL | os.O_WRONLY) os.close(fd) return True except FileExistsError: return False except Exception: return False def release_lock(lock_path: str) -> None: try: if os.path.isfile(lock_path): os.remove(lock_path) except Exception: pass def check_admin_pin(pin: str) -> bool: if not ADMIN_PIN: return False return (pin or "").strip() == ADMIN_PIN def stable_int_id(s: str) -> int: h = 2166136261 for ch in s.encode("utf-8", errors="ignore"): h ^= ch h = (h * 16777619) & 0xFFFFFFFF return int(h) # --------------------------- # API GET # --------------------------- def api_get(path: str, params: Dict[str, Any]) -> Dict[str, Any]: if not APISPORTS_KEY: raise RuntimeError("Nedostaje APISPORTS_KEY (Settings -> Secrets).") key = f"{path}|{json.dumps(params, sort_keys=True, ensure_ascii=False)}" cached = cache.get(key) if cached is not None: return cached url = API_BASE + path headers = {"x-apisports-key": APISPORTS_KEY} r = requests.get(url, headers=headers, params=params, timeout=25) data = r.json() if isinstance(data, dict) and data.get("errors"): raise RuntimeError(f"API errors: {data['errors']}") cache.set(key, data) return data # --------------------------- # API-Sports calls # --------------------------- def get_leagues_by_country(country: str) -> List[Dict[str, Any]]: data = api_get("/leagues", {"country": country}) return data.get("response") or [] def pick_league(country_leagues: List[Dict[str, Any]], aliases: List[str]) -> Optional[Dict[str, Any]]: best = None best_score = 0.0 for lg in country_leagues: name = str(lg.get("name") or "") for a in aliases: sc = jaccard(name, a) if sc > best_score: best_score = sc best = lg return best if best_score >= 0.30 else None def pick_current_season(lg: Dict[str, Any]) -> Optional[int]: seasons = lg.get("seasons") or [] for s in seasons: if isinstance(s, dict) and s.get("current") is True and isinstance(s.get("season"), int): return s["season"] vals = [] for s in seasons: if isinstance(s, dict) and isinstance(s.get("season"), int): vals.append(s["season"]) return max(vals) if vals else None def get_games_for_date(league_id: int, season: int, date_str: str) -> List[Dict[str, Any]]: data = api_get("/games", {"league": league_id, "season": season, "date": date_str, "timezone": TZ}) return data.get("response") or [] def get_standings(league_id: int, season: int) -> Any: data = api_get("/standings", {"league": league_id, "season": season}) return data.get("response") def get_odds_for_game(game_id: int) -> Any: data = api_get("/odds", {"game": game_id}) resp = data.get("response") or [] return resp[0] if resp else None def get_h2h(home_id: int, away_id: int, league_id: int) -> List[Dict[str, Any]]: data = api_get("/games/h2h", {"h2h": f"{home_id}-{away_id}", "league": league_id, "timezone": TZ}) return data.get("response") or [] # --------------------------- # Standings parse -> team table (+ rank) # --------------------------- def flatten_standings(resp: Any) -> List[Dict[str, Any]]: rows = [] def walk(x: Any): if isinstance(x, list): for it in x: walk(it) elif isinstance(x, dict): team = x.get("team") if isinstance(team, dict) and isinstance(team.get("id"), int): rows.append(x) else: for v in x.values(): walk(v) walk(resp) return rows def extract_team_stats(row: Dict[str, Any]) -> Tuple[Optional[float], Optional[float], Optional[float], Optional[int]]: rank = None for k in ["rank", "position", "pos"]: if isinstance(row.get(k), int): rank = row.get(k) break played = None for path in [("games", "played"), ("games", "played", "total"), ("played",)]: cur = row ok = True for k in path: if isinstance(cur, dict) and k in cur: cur = cur[k] else: ok = False break if ok: v = to_float(cur) if v is not None and v > 0: played = v break gf = None ga = None for path in [("goals", "for", "total"), ("goals", "for"), ("goals_for",), ("gf",)]: cur = row ok = True for k in path: if isinstance(cur, dict) and k in cur: cur = cur[k] else: ok = False break if ok: v = to_float(cur) if v is not None: gf = v break for path in [("goals", "against", "total"), ("goals", "against"), ("goals_against",), ("ga",)]: cur = row ok = True for k in path: if isinstance(cur, dict) and k in cur: cur = cur[k] else: ok = False break if ok: v = to_float(cur) if v is not None: ga = v break return played, gf, ga, rank def build_team_table(standings_resp: Any) -> Dict[int, Dict[str, Any]]: table: Dict[int, Dict[str, Any]] = {} for r in flatten_standings(standings_resp): tid = r.get("team", {}).get("id") if not isinstance(tid, int): continue played, gf, ga, rank = extract_team_stats(r) table[tid] = {"played": played, "gf": gf, "ga": ga, "rank": rank} return table # --------------------------- # Strength helpers # --------------------------- def team_rates(row: Dict[str, Any]) -> Tuple[Optional[float], Optional[float], Optional[float]]: played = row.get("played") gf = row.get("gf") ga = row.get("ga") if not (isinstance(played, (int, float)) and played and played > 0): return None, None, None if gf is None or ga is None: return None, None, None gfpg = float(gf) / float(played) gapg = float(ga) / float(played) gdpg = gfpg - gapg return gfpg, gapg, gdpg def team_strength_index(row: Dict[str, Any]) -> Optional[float]: gfpg, gapg, gdpg = team_rates(row) if gfpg is None: return None rank = row.get("rank") rank_bonus = 0.0 if isinstance(rank, int) and rank > 0: rank_bonus = max(0.0, 0.28 - (min(rank, 25) / 25) * 0.28) return (0.90 * gdpg) + (0.12 * gfpg) - (0.08 * gapg) + rank_bonus def favorite_quality_bonus(fav_row: Dict[str, Any], dog_row: Dict[str, Any]) -> Tuple[int, str]: bonus = 0 reasons = [] fav_rank = fav_row.get("rank") dog_rank = dog_row.get("rank") fav_gfpg, fav_gapg, fav_gdpg = team_rates(fav_row) dog_gfpg, dog_gapg, dog_gdpg = team_rates(dog_row) if isinstance(fav_rank, int) and isinstance(dog_rank, int): rank_gap = dog_rank - fav_rank if fav_rank <= TOP_RANK and dog_rank >= BOTTOM_RANK: bonus += 4 reasons.append(f"Top-vs-bottom (rank {fav_rank} vs {dog_rank})") elif rank_gap >= MIN_RANK_GAP_FOR_STRONG: bonus += 2 reasons.append(f"Velika razlika u tabeli (gap {rank_gap})") if fav_gfpg is not None and dog_gapg is not None and fav_gapg is not None and dog_gfpg is not None: if (fav_gfpg - dog_gfpg) >= MIN_GFP_GAP_FOR_STRONG: bonus += 1 reasons.append("Favorit daje znatno više golova/meč") if (dog_gapg - fav_gapg) >= MIN_GAPG_GAP_FOR_STRONG: bonus += 1 reasons.append("Autsajder prima znatno više golova/meč") if fav_gdpg is not None and dog_gdpg is not None: if (fav_gdpg - dog_gdpg) >= MIN_GDPG_GAP_FOR_STRONG: bonus += 2 reasons.append("Velika razlika u GD/meč") bonus = max(0, min(8, bonus)) return bonus, ("; ".join(reasons) if reasons else "") def favorite_is_plausible(fav_row: Dict[str, Any], dog_row: Dict[str, Any]) -> Tuple[bool, str]: fav_rank = fav_row.get("rank") dog_rank = dog_row.get("rank") fav_gfpg, fav_gapg, fav_gdpg = team_rates(fav_row) dog_gfpg, dog_gapg, dog_gdpg = team_rates(dog_row) if fav_gfpg is None or dog_gfpg is None: return True, "" if isinstance(fav_rank, int) and isinstance(dog_rank, int): if fav_rank - dog_rank >= 6: if (dog_gdpg is not None and fav_gdpg is not None) and (dog_gdpg - fav_gdpg) >= 0.25: return False, "Kvote kažu favorit, ali tabela+GD govore suprotno (zamka)" if fav_gdpg is not None and dog_gdpg is not None: if (dog_gdpg - fav_gdpg) >= 0.45: return False, "Favorit po kvoti je znatno slabiji po GD/meč" return True, "" def maybe_override_favorite_by_stats( home_id: int, away_id: int, home_odds: float, away_odds: float, home_row: Dict[str, Any], away_row: Dict[str, Any], ) -> Tuple[bool, float, float, int, int, str]: if home_odds <= away_odds: fav_is_home = True fav_odds = home_odds dog_odds = away_odds fav_id = home_id dog_id = away_id else: fav_is_home = False fav_odds = away_odds dog_odds = home_odds fav_id = away_id dog_id = home_id if not ALLOW_STAT_OVERRIDE: return fav_is_home, fav_odds, dog_odds, fav_id, dog_id, "" if OVERRIDE_ONLY_IF_ODDS_NOT_EXTREME: if min(home_odds, away_odds) < 1.35 or max(home_odds, away_odds) > 3.50: return fav_is_home, fav_odds, dog_odds, fav_id, dog_id, "" hs = team_strength_index(home_row) as_ = team_strength_index(away_row) if hs is None or as_ is None: return fav_is_home, fav_odds, dog_odds, fav_id, dog_id, "" if fav_is_home and (as_ - hs) >= 0.35: if away_odds <= FAV_MAX_ODDS: return False, away_odds, home_odds, away_id, home_id, "Favorit korigovan po statistici" if (not fav_is_home) and (hs - as_) >= 0.35: if home_odds <= FAV_MAX_ODDS: return True, home_odds, away_odds, home_id, away_id, "Favorit korigovan po statistici" return fav_is_home, fav_odds, dog_odds, fav_id, dog_id, "" # --------------------------- # Too-even filter # --------------------------- def is_too_even_matchup( fav_odds: float, dog_odds: float, fav: Dict[str, Any], dog: Dict[str, Any], ) -> Tuple[bool, str]: fav_rank = fav.get("rank") dog_rank = dog.get("rank") if isinstance(fav_rank, int) and isinstance(dog_rank, int): if fav_rank <= 3 and dog_rank <= 3: return True, f"Top matchup (rank {fav_rank} vs {dog_rank})" fav_played = fav.get("played") dog_played = dog.get("played") if not (isinstance(fav_played, (int, float)) and isinstance(dog_played, (int, float)) and fav_played > 0 and dog_played > 0): return False, "" fav_gf = fav.get("gf") fav_ga = fav.get("ga") dog_gf = dog.get("gf") dog_ga = dog.get("ga") if any(v is None for v in [fav_gf, fav_ga, dog_gf, dog_ga]): return False, "" fav_gfpg = float(fav_gf) / float(fav_played) fav_gapg = float(fav_ga) / float(fav_played) dog_gfpg = float(dog_gf) / float(dog_played) dog_gapg = float(dog_ga) / float(dog_played) fav_gdpg = fav_gfpg - fav_gapg dog_gdpg = dog_gfpg - dog_gapg p_f = 1.0 / float(fav_odds) p_d = 1.0 / float(dog_odds) p_gap = p_f - p_d close_attack = abs(fav_gfpg - dog_gfpg) <= 0.25 close_def = abs(fav_gapg - dog_gapg) <= 0.25 close_gd = abs(fav_gdpg - dog_gdpg) <= 0.20 small_prob_gap = p_gap <= 0.10 if (close_attack and close_def and close_gd) or (close_gd and small_prob_gap): return True, "Previše izjednačeno (slični GF/GA/GD i mala razlika u snazi)" return False, "" # --------------------------- # Odds parser # --------------------------- def parse_moneyline(odds_obj: Dict[str, Any], home_name: str, away_name: str) -> Tuple[Optional[float], Optional[float]]: if not isinstance(odds_obj, dict): return None, None hn = norm(home_name) an = norm(away_name) home_vals = [] away_vals = [] bookmakers = odds_obj.get("bookmakers") or [] for bm in bookmakers: bets = bm.get("bets") or [] for bet in bets: bet_name = norm(str(bet.get("name") or "")) if not any(k in bet_name for k in ["winner", "match", "moneyline", "1x2", "home away"]): continue values = bet.get("values") or [] for v in values: label = norm(str(v.get("value") or "")) odd = to_float(v.get("odd")) if odd is None: continue if label in ("home", "1") or (hn and (label == hn or hn in label or label in hn)): home_vals.append(odd) elif label in ("away", "2") or (an and (label == an or an in label or label in an)): away_vals.append(odd) def median(xs: List[float]) -> Optional[float]: if not xs: return None xs = sorted(xs) m = len(xs) // 2 return xs[m] if len(xs) % 2 else (xs[m - 1] + xs[m]) / 2.0 return median(home_vals), median(away_vals) # --------------------------- # H2H + floor helpers # --------------------------- def extract_total_scores(g: Dict[str, Any]) -> Tuple[Optional[int], Optional[int]]: scores = g.get("scores") or {} def pick(side: str) -> Optional[int]: cur = scores.get(side) if isinstance(cur, dict): v = cur.get("total") vv = to_float(v) return int(vv) if vv is not None else None vv = to_float(cur) return int(vv) if vv is not None else None return pick("home"), pick("away") def extract_game_date(g: Dict[str, Any]) -> Optional[dt.datetime]: s = g.get("date") if isinstance(s, str) and s: try: return dt.datetime.fromisoformat(s.replace("Z", "+00:00")) except Exception: return None gg = g.get("game", {}) if isinstance(gg, dict) and isinstance(gg.get("date"), str): try: return dt.datetime.fromisoformat(gg["date"].replace("Z", "+00:00")) except Exception: return None return None def h2h_fav_scored_rate(h2h_games: List[Dict[str, Any]], fav_is_home: bool, last_n: int) -> Optional[float]: if not h2h_games: return None items = [] for g in h2h_games: d = extract_game_date(g) or dt.datetime.min items.append((d, g)) items.sort(key=lambda x: x[0], reverse=True) picked = [g for _, g in items[:last_n]] if not picked: return None ok = 0 valid = 0 for g in picked: hs, as_ = extract_total_scores(g) if hs is None or as_ is None: continue valid += 1 fav_goals = hs if fav_is_home else as_ if fav_goals >= 1: ok += 1 return (ok / valid) if valid else None def get_status_short(g: Dict[str, Any]) -> Optional[str]: st = g.get("status") if isinstance(st, dict): sh = st.get("short") if isinstance(sh, str): return sh if isinstance(g.get("status"), str): return g["status"] return None def get_team_goals_in_game(g: Dict[str, Any], team_id: int) -> Optional[int]: teams = g.get("teams") or {} home = teams.get("home") or {} away = teams.get("away") or {} hid = home.get("id") aid = away.get("id") hs, as_ = extract_total_scores(g) if hs is None or as_ is None: return None if hid == team_id: return hs if aid == team_id: return as_ return None def season_floor_rate_1plus(league_id: int, season: int, team_id: int, date_limit: str) -> Optional[float]: ck = f"floor|{league_id}|{season}|{team_id}|{date_limit}" cached = floor_cache.get(ck) if cached is not None: return cached data = api_get("/games", { "league": league_id, "season": season, "team": team_id, "date": date_limit, "timezone": TZ }) games = data.get("response") or [] valid = 0 ok = 0 for g in games: sh = get_status_short(g) if sh is None or sh not in FINISHED_STATUS: continue tg = get_team_goals_in_game(g, team_id) if tg is None: continue valid += 1 if tg >= 1: ok += 1 rate = (ok / valid) if valid > 0 else None floor_cache.set(ck, rate) return rate def expected_goals_fav(gfpg: Optional[float], gapg: Optional[float]) -> Optional[float]: if gfpg is None or gapg is None: return None return 0.55 * gfpg + 0.45 * gapg # --------------------------- # Safety formula # --------------------------- def calc_safety( fav_odds: float, dog_odds: float, fav_gf: Optional[float], fav_played: Optional[float], dog_ga: Optional[float], dog_played: Optional[float], h2h_rate: Optional[float], floor_rate: Optional[float], ) -> Tuple[int, str, Optional[float]]: p_f = (1.0 / fav_odds) p_d = (1.0 / dog_odds) if dog_odds > 0 else 0.0 p = p_f / (p_f + p_d) if (p_f + p_d) > 0 else 0.5 base = 60 + 40 * (p - 0.5) * 1.35 base = max(55, min(92, base)) gfpg = (fav_gf / fav_played) if (fav_gf is not None and fav_played and fav_played > 0) else None gapg = (dog_ga / dog_played) if (dog_ga is not None and dog_played and dog_played > 0) else None eg = expected_goals_fav(gfpg, gapg) bonus = 0.0 parts = [] parts.append(f"GF_fav/meč≈{gfpg:.2f}" if gfpg is not None else "GF_fav/meč n/a") parts.append(f"GA_dog/meč≈{gapg:.2f}" if gapg is not None else "GA_dog/meč n/a") if eg is not None: parts.append(f"EG_fav≈{eg:.2f}") if eg >= 2.60: bonus += 3.0 elif eg >= 2.30: bonus += 1.5 elif eg < 2.00: bonus -= 2.0 else: parts.append("EG_fav n/a") bonus -= 1.0 if h2h_rate is not None: parts.append(f"H2H 1+≈{h2h_rate*100:.0f}%") bonus += max(0.0, min(8.0, (h2h_rate - 0.55) * 18.0)) else: parts.append("H2H n/a") if floor_rate is not None: parts.append(f"Sezona 1+≈{floor_rate*100:.0f}%") if floor_rate >= 0.90: bonus += 3.0 elif floor_rate >= 0.80: bonus += 2.0 elif floor_rate >= 0.70: bonus += 1.0 elif floor_rate < 0.60: bonus -= 2.0 else: parts.append("Floor n/a") safety = int(round(max(50.0, min(97.0, base + bonus)))) return safety, " | ".join(parts), eg # --------------------------- # Gemini (optional) # --------------------------- def gemini_explain(summary: Dict[str, Any]) -> str: fallback = "AI rezime: (Gemini nije podešen) — zaključak na osnovu kvota + EG_fav + floor + H2H." if not GEMINI_API_KEY: return fallback url = f"https://generativelanguage.googleapis.com/v1beta/models/{GEMINI_MODEL}:generateContent?key={GEMINI_API_KEY}" prompt = ( "Ti si profesionalni analitičar sportskog klađenja (hokej). Ne izmišljaj podatke.\n" "Napiši KRATAK rezime (1-2 rečenice) zašto favorit ima veliku šansu da da bar 1 gol u meču.\n" f"{json.dumps(summary, ensure_ascii=False)}" ) body = {"contents": [{"role": "user", "parts": [{"text": prompt}]}]} try: r = requests.post(url, json=body, timeout=25) if r.status_code != 200: return fallback data = r.json() cands = data.get("candidates") or [] if not cands: return fallback content = cands[0].get("content") or {} parts = content.get("parts") or [] if not parts: return fallback text = parts[0].get("text") return text.strip() if isinstance(text, str) and text.strip() else fallback except Exception: return fallback # --------------------------- # Cache helpers (+ manual + delete) # --------------------------- def cache_file_path(day_choice: str, date_str: str) -> str: return os.path.join(CACHE_DIR, f"result_{day_choice}_{date_str}.json") def lock_file_path(day_choice: str, date_str: str) -> str: return os.path.join(CACHE_DIR, f"result_{day_choice}_{date_str}.lock") def ensure_cache_container(day_choice: str, date_str: str) -> Dict[str, Any]: path = cache_file_path(day_choice, date_str) data = read_json(path, None) if isinstance(data, dict): data["day_choice"] = day_choice data["date_str"] = date_str data["expires_at"] = max(int(data.get("expires_at") or 0), now_ts() + RESULT_CACHE_TTL_SEC) if not isinstance(data.get("raw_picks"), list): data["raw_picks"] = [] if not isinstance(data.get("manual_picks"), list): data["manual_picks"] = [] if not isinstance(data.get("output_md"), str): data["output_md"] = "" atomic_write_json(path, data) return data data = { "day_choice": day_choice, "date_str": date_str, "created_at": now_ts(), "expires_at": now_ts() + RESULT_CACHE_TTL_SEC, "output_md": "", "raw_picks": [], "manual_picks": [], } atomic_write_json(path, data) return data def read_persistent_cache(day_choice: str, date_str: str) -> Optional[Dict[str, Any]]: path = cache_file_path(day_choice, date_str) data = read_json(path, None) if not isinstance(data, dict): return None exp = data.get("expires_at") if not isinstance(exp, int): return None if now_ts() >= exp: return None if not isinstance(data.get("raw_picks"), list): data["raw_picks"] = [] if not isinstance(data.get("manual_picks"), list): data["manual_picks"] = [] if not isinstance(data.get("output_md"), str): data["output_md"] = "" return data def write_persistent_cache(day_choice: str, date_str: str, output_md: str, raw_picks: List[Dict[str, Any]]) -> None: container = ensure_cache_container(day_choice, date_str) container["created_at"] = now_ts() container["expires_at"] = now_ts() + RESULT_CACHE_TTL_SEC container["output_md"] = output_md container["raw_picks"] = raw_picks atomic_write_json(cache_file_path(day_choice, date_str), container) def all_cached_picks(container: Dict[str, Any]) -> List[Dict[str, Any]]: raw = container.get("raw_picks") if isinstance(container.get("raw_picks"), list) else [] man = container.get("manual_picks") if isinstance(container.get("manual_picks"), list) else [] return list(raw) + list(man) def pick_key(date_str: str, p: Dict[str, Any]) -> str: try: return make_bet_id(date_str, int(p.get("league_id", -1)), int(p.get("game_id", 0))) except Exception: return f"{date_str}|X|{p.get('match','')}" def pick_label(p: Dict[str, Any]) -> str: fav = p.get("favorite", "") match_ = p.get("match", "") league = p.get("league", "") odds = p.get("fav_odds", "") return f"{fav} — {match_} | {league} | kvota {odds}" def dropdown_update_from_cache(day_choice: str) -> Any: date_str = date_str_for_choice(day_choice) c = read_persistent_cache(day_choice, date_str) if not c: return gr.update(choices=[], value=None) picks = all_cached_picks(c) choices = [(pick_label(p), pick_key(date_str, p)) for p in picks] return gr.update(choices=choices, value=(choices[0][1] if choices else None)) def manual_section_md(manual_picks: List[Dict[str, Any]]) -> str: if not manual_picks: return "" lines = [] lines.append("\n---\n## ✍️ Ručno dodani mečevi\n") for i, p in enumerate(manual_picks, start=1): lines.append( f"**R{i}.** {p.get('match','')} | fav: **{p.get('favorite','')}** " f"(kvota {p.get('fav_odds','')}) | sigurnost {p.get('safety','')}% | liga: {p.get('league','')}\n" ) return "\n".join(lines) # --------------------------- # NEW: Render cache output from lists (fix for delete) # --------------------------- def render_cached_markdown(day_choice: str, date_str: str, raw: List[Dict[str, Any]], manual: List[Dict[str, Any]]) -> str: # Sort picks by safety desc (keep original info) def safety_val(p: Dict[str, Any]) -> int: try: return int(p.get("safety", 0)) except Exception: return 0 raw = [p for p in (raw or []) if isinstance(p, dict)] manual = [p for p in (manual or []) if isinstance(p, dict)] raw_sorted = sorted(raw, key=safety_val, reverse=True) low = [p for p in raw_sorted if float(p.get("fav_odds", 99)) <= LOW_ODDS_MAX][:5] high = [p for p in raw_sorted if HIGH_ODDS_MIN <= float(p.get("fav_odds", 0)) <= HIGH_ODDS_MAX][:5] lines: List[str] = [] lines.append( f"## Preporuke ({day_choice} — {date_str})\n" f"- Favorit kvota ≤ **{FAV_MAX_ODDS}**\n" f"- **Top 10:** 5 tipova (≤{LOW_ODDS_MAX:.2f}) + 5 tipova ({HIGH_ODDS_MIN:.2f}–{HIGH_ODDS_MAX:.2f})\n" f"- ✅ Prikaz je iz cache-a (bez ponovnog API poziva)\n" ) def render_pick(idx: int, p: Dict[str, Any]) -> str: ai_txt = p.get("ai_txt", "") if not isinstance(ai_txt, str) or not ai_txt.strip(): ai_txt = "(AI rezime nije sačuvan u cache-u)" return ( f"### Opklada {idx}\n" f"- **Sigurnost:** {p.get('safety','')}%\n" f"- **Meč:** {p.get('match','')} \n" f"- **Liga:** {p.get('league','')}\n" f"- **Favorit (kvota):** {p.get('favorite','')} ({p.get('fav_odds','')}) | **Autsajder:** {p.get('underdog','')} ({p.get('dog_odds','')})\n" f"- **Bucket:** {p.get('bucket','')}\n" f"- **Signal:** {p.get('signal','')}\n" f"- **AI rezime:** {ai_txt}\n" f"- ✅ **Igrati golove na:** **{p.get('favorite','')}** (da da bar 1 gol u meču)\n" ) lines.append(f"---\n## Grupa 1 — Favorit kvota ≤ {LOW_ODDS_MAX:.2f}\n") if not low: lines.append("_Nema dovoljno mečeva u ovoj kvota zoni._\n") else: idx = 1 for p in low: lines.append(render_pick(idx, p)) idx += 1 lines.append(f"---\n## Grupa 2 — Favorit kvota {HIGH_ODDS_MIN:.2f}–{HIGH_ODDS_MAX:.2f}\n") if not high: lines.append("_Nema dovoljno mečeva u ovoj kvota zoni._\n") else: idx = 6 # second group starts at 6 in your format for p in high: lines.append(render_pick(idx, p)) idx += 1 if manual: lines.append("\n---\n## ✍️ Ručno dodani mečevi\n") for i, p in enumerate(sorted(manual, key=safety_val, reverse=True), start=1): lines.append( f"**R{i}.** {p.get('match','')} | fav: **{p.get('favorite','')}** " f"(kvota {p.get('fav_odds','')}) | sigurnost {p.get('safety','')}% | liga: {p.get('league','')}\n" ) lines.append( "---\n" "### Objašnjenje\n" "- **EG_fav** = 0.55×(GF_favorit/meč) + 0.45×(GA_autsajder/meč)\n" "- **Sezona 1+ gol** = % završenih mečeva u sezoni (FT/AOT/AP) gdje favorit daje bar 1 gol\n" "- **Q bonus** = dodatni bodovi kad tabela i GF/GA jasno govore da je favorit mnogo jači\n" ) return "\n".join(lines) # --------------------------- # History helpers # --------------------------- def read_history() -> List[Dict[str, Any]]: data = read_json(HISTORY_PATH, []) return data if isinstance(data, list) else [] def write_history(records: List[Dict[str, Any]]) -> None: atomic_write_json(HISTORY_PATH, records) def make_bet_id(date_str: str, league_id: int, game_id: int) -> str: return f"{date_str}|{league_id}|{game_id}|FAV_1PLUS_GOAL" def normalize_result(val: str) -> str: v = (val or "").strip().upper() if v in {"WIN", "W", "✅"}: return "WIN" if v in {"LOSS", "LOSE", "L", "❌"}: return "LOSS" return "PENDING" def history_stats(rows: List[Dict[str, Any]]) -> str: total = len(rows) win = sum(1 for r in rows if r.get("result") == "WIN") loss = sum(1 for r in rows if r.get("result") == "LOSS") pend = total - win - loss wr = (win / (win + loss) * 100) if (win + loss) > 0 else 0.0 return ( f"📌 Ukupno opklada: {total}\n\n" f"✅ Dobitne (WIN): {win}\n\n" f"❌ Izgubljene (LOSS): {loss}\n\n" f"⏳ Na čekanju (PENDING): {pend}\n\n" f"⭐ Win rate: {wr:.1f}%" ) def render_history_markdown(rows: List[Dict[str, Any]]) -> str: if not rows: return "_Nema snimljenih opklada za ovaj datum._" def result_icon(res: str) -> str: r = (res or "PENDING").strip().upper() if r == "WIN": return "✅" if r == "LOSS": return "❌" return "⏳" lines = ["### Opklade za izabrani datum\n"] for i, r in enumerate(rows, start=1): res = r.get("result", "PENDING") icon = result_icon(res) lines.append( f"**{i}.** {r.get('match','')} | fav: **{r.get('favorite','')}** " f"(kvota {r.get('fav_odds','')}) | sigurnost {r.get('safety','')}% | " f"rezultat: **{res} {icon}**\n" ) return "\n".join(lines) # --------------------------- # Compute recommendations # --------------------------- def compute_recommendations(day_choice: str, progress=gr.Progress(track_tqdm=False)) -> Tuple[str, List[Dict[str, Any]]]: date_str = date_str_for_choice(day_choice) progress(0.02, desc="Mapiranje liga…") resolved: List[Tuple[int, int, str]] = [] for i, (country, aliases) in enumerate(TARGET_LEAGUES, start=1): leagues = get_leagues_by_country(country) best = pick_league(leagues, aliases) if best: league_id = best.get("id") season = pick_current_season(best) name = best.get("name") if isinstance(league_id, int) and isinstance(season, int): resolved.append((league_id, season, str(name))) progress(0.02 + 0.18 * (i / len(TARGET_LEAGUES)), desc=f"Mapiranje liga… ({i}/{len(TARGET_LEAGUES)})") if not resolved: return ("❌ Ne mogu mapirati nijednu ligu (provjeri API key / plan).", []) progress(0.22, desc="Učitavam mečeve i računam…") picks: List[Dict[str, Any]] = [] total_leagues = len(resolved) for li, (league_id, season, league_name) in enumerate(resolved, start=1): progress(0.22 + 0.50 * (li / total_leagues), desc=f"Obrađujem ligu: {league_name} ({li}/{total_leagues})") games = get_games_for_date(league_id, season, date_str) if not games: continue standings_resp = get_standings(league_id, season) team_table = build_team_table(standings_resp) for g in games: game_id = g.get("id") if isinstance(g.get("id"), int) else None teams = g.get("teams") or {} home = teams.get("home") or {} away = teams.get("away") or {} home_id = home.get("id") away_id = away.get("id") home_name = str(home.get("name") or "Home") away_name = str(away.get("name") or "Away") start_time = str(g.get("date") or "") if not (isinstance(game_id, int) and isinstance(home_id, int) and isinstance(away_id, int)): continue odds_obj = get_odds_for_game(game_id) if not odds_obj: continue home_odds, away_odds = parse_moneyline(odds_obj, home_name, away_name) if home_odds is None or away_odds is None: continue home_odds = float(home_odds) away_odds = float(away_odds) home_row = team_table.get(int(home_id), {}) away_row = team_table.get(int(away_id), {}) fav_is_home, fav_odds, dog_odds, fav_id, dog_id, override_reason = maybe_override_favorite_by_stats( home_id=int(home_id), away_id=int(away_id), home_odds=home_odds, away_odds=away_odds, home_row=home_row, away_row=away_row, ) if fav_odds > FAV_MAX_ODDS: continue if fav_id == int(home_id): fav_team = home_name dog_team = away_name fav_row = home_row dog_row = away_row else: fav_team = away_name dog_team = home_name fav_row = away_row dog_row = home_row plausible, _pl_reason = favorite_is_plausible(fav_row, dog_row) if not plausible: continue too_even, _reason = is_too_even_matchup( fav_odds=fav_odds, dog_odds=dog_odds, fav=fav_row, dog=dog_row, ) if too_even: continue fav_played, fav_gf = fav_row.get("played"), fav_row.get("gf") dog_played, dog_ga = dog_row.get("played"), dog_row.get("ga") h2h_games = get_h2h(int(home_id), int(away_id), league_id) h2h_rate = h2h_fav_scored_rate(h2h_games, fav_is_home=fav_is_home, last_n=H2H_LAST_N) floor_rate = season_floor_rate_1plus(league_id, season, fav_id, date_str) safety, signal, eg = calc_safety( fav_odds=fav_odds, dog_odds=dog_odds, fav_gf=fav_gf, fav_played=fav_played, dog_ga=dog_ga, dog_played=dog_played, h2h_rate=h2h_rate, floor_rate=floor_rate, ) q_bonus, q_reason = favorite_quality_bonus(fav_row, dog_row) if q_bonus > 0: safety = int(max(50, min(97, safety + q_bonus))) if q_reason: signal = f"{signal} | Q+{q_bonus}: {q_reason}" if override_reason: signal = f"{signal} | {override_reason}" bucket = "1.45–1.80" if fav_odds >= HIGH_ODDS_MIN else "1.10–1.44" picks.append({ "date_str": date_str, "day_choice": day_choice, "league": league_name, "league_id": int(league_id), "season": int(season), "game_id": int(game_id), "start": start_time, "match": f"{home_name} vs {away_name}", "favorite": fav_team, "underdog": dog_team, "fav_odds": round(float(fav_odds), 2), "dog_odds": round(float(dog_odds), 2), "safety": int(safety), "eg_fav": None if eg is None else round(float(eg), 2), "floor_rate": None if floor_rate is None else round(float(floor_rate), 3), "h2h_rate": None if h2h_rate is None else round(float(h2h_rate), 3), "bucket": bucket, "signal": signal, }) if not picks: return ("⚠️ Nema tipova za izabrani dan (odds možda nisu dostupni ili su mečevi previše izjednačeni).", []) picks.sort(key=lambda x: x["safety"], reverse=True) low = [p for p in picks if float(p["fav_odds"]) <= LOW_ODDS_MAX][:5] high = [p for p in picks if HIGH_ODDS_MIN <= float(p["fav_odds"]) <= HIGH_ODDS_MAX][:5] top10 = low + high out_lines: List[str] = [] out_lines.append( f"## Preporuke ({day_choice} — {date_str})\n" f"- Favorit kvota ≤ **{FAV_MAX_ODDS}**\n" f"- **Top 10:** 5 tipova (≤{LOW_ODDS_MAX:.2f}) + 5 tipova ({HIGH_ODDS_MIN:.2f}–{HIGH_ODDS_MAX:.2f})\n" f"- ✅ Favorit se potvrđuje i po **tabeli + GF/GA** (Q bonus)\n" ) def render_pick(idx: int, p: Dict[str, Any]) -> str: summary = { "liga": p["league"], "mec": p["match"], "favorit": p["favorite"], "kvota_favorit": p["fav_odds"], "kvota_autsajder": p["dog_odds"], "EG_fav": p["eg_fav"], "Sezona_1plus": None if p["floor_rate"] is None else round(float(p["floor_rate"]) * 100, 0), "H2H_1plus": None if p["h2h_rate"] is None else round(float(p["h2h_rate"]) * 100, 0), } ai_txt = gemini_explain(summary) p["ai_txt"] = ai_txt # store to cache so it doesn't re-call on cached view return ( f"### Opklada {idx}\n" f"- **Sigurnost:** {p['safety']}%\n" f"- **Meč:** {p['match']} \n" f"- **Liga:** {p['league']}\n" f"- **Favorit (kvota):** {p['favorite']} ({p['fav_odds']}) | **Autsajder:** {p['underdog']} ({p['dog_odds']})\n" f"- **Bucket:** {p['bucket']}\n" f"- **Signal:** {p['signal']}\n" f"- **AI rezime:** {ai_txt}\n" f"- ✅ **Igrati golove na:** **{p['favorite']}** (da da bar 1 gol u meču)\n" ) out_lines.append(f"---\n## Grupa 1 — Favorit kvota ≤ {LOW_ODDS_MAX:.2f}\n") idx = 1 if not low: out_lines.append("_Nema dovoljno mečeva u ovoj kvota zoni._\n") else: for p in low: out_lines.append(render_pick(idx, p)) idx += 1 out_lines.append(f"---\n## Grupa 2 — Favorit kvota {HIGH_ODDS_MIN:.2f}–{HIGH_ODDS_MAX:.2f}\n") if not high: out_lines.append("_Nema dovoljno mečeva u ovoj kvota zoni._\n") else: for p in high: out_lines.append(render_pick(idx, p)) idx += 1 out_lines.append( "---\n" "### Objašnjenje\n" "- **EG_fav** = 0.55×(GF_favorit/meč) + 0.45×(GA_autsajder/meč)\n" "- **Sezona 1+ gol** = % završenih mečeva u sezoni (FT/AOT/AP) gdje favorit daje bar 1 gol\n" "- **Q bonus** = dodatni bodovi kad tabela i GF/GA jasno govore da je favorit mnogo jači\n" ) return ("\n".join(out_lines), top10) # --------------------------- # Cache runner with progress # --------------------------- def run_analysis_cached(day_choice: str, progress=gr.Progress(track_tqdm=False)) -> Iterator[Tuple[str, str, List[Dict[str, Any]], Any]]: if not APISPORTS_KEY: yield ("❌ Greška: nedostaje APISPORTS_KEY u Secrets.", "", [], gr.update(choices=[], value=None)) return date_str = date_str_for_choice(day_choice) yield ("⏳ Provjeravam cache…", "", [], gr.update(choices=[], value=None)) cached = read_persistent_cache(day_choice, date_str) if cached: manual = cached.get("manual_picks") if isinstance(cached.get("manual_picks"), list) else [] raw = cached.get("raw_picks") if isinstance(cached.get("raw_picks"), list) else [] merged = list(raw) + list(manual) dd = dropdown_update_from_cache(day_choice) # IMPORTANT FIX: render from lists, not from stale output_md output_md = render_cached_markdown(day_choice, date_str, raw, manual) yield ("✅ (CACHE) Rezultat učitan iz cache-a.", output_md, merged, dd) return lockp = lock_file_path(day_choice, date_str) if not try_acquire_lock(lockp): for i in range(15): yield (f"⏳ Analiza je već u toku (drugi korisnik). Čekam cache… ({i+1}/15)", "", [], gr.update(choices=[], value=None)) time.sleep(1) cached2 = read_persistent_cache(day_choice, date_str) if cached2: manual = cached2.get("manual_picks") if isinstance(cached2.get("manual_picks"), list) else [] raw = cached2.get("raw_picks") if isinstance(cached2.get("raw_picks"), list) else [] merged = list(raw) + list(manual) dd = dropdown_update_from_cache(day_choice) # IMPORTANT FIX output_md = render_cached_markdown(day_choice, date_str, raw, manual) yield ("✅ (CACHE) Rezultat je upravo kreiran i učitan.", output_md, merged, dd) return yield ("⚠️ Analiza je još u toku na drugom zahtjevu. Pokušaj opet za minut.", "", [], gr.update(choices=[], value=None)) return try: yield ("⏳ Cache nije pronađen — pokrećem analizu…", "", [], gr.update(choices=[], value=None)) output_md, auto_picks = compute_recommendations(day_choice, progress=progress) # preserve manual picks if exist container = ensure_cache_container(day_choice, date_str) manual = container.get("manual_picks") if isinstance(container.get("manual_picks"), list) else [] if auto_picks and isinstance(output_md, str) and output_md.strip(): yield ("⏳ Upisujem rezultat u cache…", output_md + manual_section_md(manual), auto_picks + manual, gr.update()) write_persistent_cache(day_choice, date_str, output_md, auto_picks) cached3 = read_persistent_cache(day_choice, date_str) or {} manual3 = cached3.get("manual_picks") if isinstance(cached3.get("manual_picks"), list) else [] raw3 = cached3.get("raw_picks") if isinstance(cached3.get("raw_picks"), list) else [] merged3 = list(raw3) + list(manual3) dd = dropdown_update_from_cache(day_choice) # After analysis we can still show cached-render so it stays consistent with deletions output_final = render_cached_markdown(day_choice, date_str, raw3, manual3) yield ("✅ Analiza završena.", output_final, merged3, dd) finally: release_lock(lockp) # --------------------------- # Manual add (PIN) -> cache + history + update dropdown # --------------------------- def add_manual_match( day_choice: str, league: str, match: str, favorite: str, underdog: str, fav_odds: float, dog_odds: float, safety: int, notes: str, pin: str, current_output: str, picks_state: List[Dict[str, Any]], ) -> Tuple[str, str, List[Dict[str, Any]], Any]: if not check_admin_pin(pin): return "❌ Pogrešan PIN (ili ADMIN_PIN nije podešen).", current_output, picks_state, gr.update() date_str = date_str_for_choice(day_choice) league = (league or "").strip() match = (match or "").strip() favorite = (favorite or "").strip() underdog = (underdog or "").strip() if not (league and match and favorite and underdog): return "❌ Popuni: Liga, Meč, Favorit, Autsajder.", current_output, picks_state, gr.update() fo = to_float(fav_odds) do = to_float(dog_odds) if fo is None or do is None or fo <= 1.0 or do <= 1.0: return "❌ Neispravne kvote.", current_output, picks_state, gr.update() if fo > FAV_MAX_ODDS: return f"❌ Favorit kvota mora biti ≤ {FAV_MAX_ODDS}.", current_output, picks_state, gr.update() safety = int(max(50, min(97, int(safety)))) bucket = "1.45–1.80" if fo >= HIGH_ODDS_MIN else "1.10–1.44" manual_id = stable_int_id(f"{date_str}|{league}|{match}|{favorite}|{underdog}|{fo}|{do}") game_id = -abs(manual_id) pick = { "date_str": date_str, "day_choice": day_choice, "league": league, "league_id": -1, "season": -1, "game_id": int(game_id), "start": "", "match": match, "favorite": favorite, "underdog": underdog, "fav_odds": round(float(fo), 2), "dog_odds": round(float(do), 2), "safety": safety, "eg_fav": None, "floor_rate": None, "h2h_rate": None, "bucket": bucket, "signal": f"MANUAL | {notes}".strip() if notes else "MANUAL", "manual": True, } # cache container = ensure_cache_container(day_choice, date_str) manual_list = container.get("manual_picks") if isinstance(container.get("manual_picks"), list) else [] if any(isinstance(x, dict) and x.get("game_id") == pick["game_id"] for x in manual_list): dd = dropdown_update_from_cache(day_choice) return "⚠️ Taj ručni meč već postoji u cache-u.", current_output, picks_state, dd manual_list.append(pick) container["manual_picks"] = manual_list container["expires_at"] = now_ts() + RESULT_CACHE_TTL_SEC atomic_write_json(cache_file_path(day_choice, date_str), container) # history immediately bid = make_bet_id(date_str, int(pick["league_id"]), int(pick["game_id"])) hist = read_history() existing_ids = {r.get("id") for r in hist if isinstance(r, dict)} if bid not in existing_ids: hist.append({ "id": bid, "date_str": date_str, "day_choice": day_choice, "league": league, "match": match, "favorite": favorite, "fav_odds": round(float(fo), 2), "bucket": bucket, "safety": safety, "eg_fav": None, "floor_rate": None, "h2h_rate": None, "result": "PENDING", "marked_at": None, "created_at": now_ts(), "notes": notes or "", }) write_history(hist) # update state (keep as before) new_state = list(picks_state or []) if not any(isinstance(x, dict) and x.get("game_id") == pick["game_id"] for x in new_state): new_state.append(pick) # IMPORTANT: render from cache lists so UI stays consistent cached_now = read_persistent_cache(day_choice, date_str) or {} raw_now = cached_now.get("raw_picks") if isinstance(cached_now.get("raw_picks"), list) else [] manual_now = cached_now.get("manual_picks") if isinstance(cached_now.get("manual_picks"), list) else [] output_updated = render_cached_markdown(day_choice, date_str, raw_now, manual_now) dd = dropdown_update_from_cache(day_choice) return "✅ Ručni meč dodat u cache i upisan u Rezultate (PENDING).", output_updated, new_state, dd # --------------------------- # Delete from cache (PIN) - FIXED: also rebuild cache output from lists # --------------------------- def delete_pick_from_cache( day_choice: str, pick_id: str, pin: str, current_output: str, picks_state: List[Dict[str, Any]], ) -> Tuple[str, str, List[Dict[str, Any]], Any]: if not check_admin_pin(pin): return "❌ Pogrešan PIN (ili ADMIN_PIN nije podešen).", current_output, picks_state, gr.update() date_str = date_str_for_choice(day_choice) pick_id = (pick_id or "").strip() if not pick_id: dd = dropdown_update_from_cache(day_choice) return "❌ Izaberi meč za brisanje.", current_output, picks_state, dd container = read_persistent_cache(day_choice, date_str) if not container: dd = dropdown_update_from_cache(day_choice) return "⚠️ Cache nije pronađen (pokreni analizu prvo).", current_output, picks_state, dd raw = container.get("raw_picks") if isinstance(container.get("raw_picks"), list) else [] man = container.get("manual_picks") if isinstance(container.get("manual_picks"), list) else [] def keep(p: Dict[str, Any]) -> bool: return pick_key(date_str, p) != pick_id raw2 = [p for p in raw if isinstance(p, dict) and keep(p)] man2 = [p for p in man if isinstance(p, dict) and keep(p)] removed = (len(raw) - len(raw2)) + (len(man) - len(man2)) if removed <= 0: dd = dropdown_update_from_cache(day_choice) return "⚠️ Nisam našao taj meč u cache-u (možda je već obrisan).", current_output, picks_state, dd # Persist lists container["raw_picks"] = raw2 container["manual_picks"] = man2 container["expires_at"] = now_ts() + RESULT_CACHE_TTL_SEC # IMPORTANT FIX: rebuild output_md from updated lists new_output_md = render_cached_markdown(day_choice, date_str, raw2, man2) container["output_md"] = new_output_md atomic_write_json(cache_file_path(day_choice, date_str), container) # update state (remove from current state list) new_state = [] for p in (picks_state or []): if isinstance(p, dict) and pick_key(date_str, p) != pick_id: new_state.append(p) dd = dropdown_update_from_cache(day_choice) return f"✅ Obrisano iz cache-a: {removed} meč.", new_output_md, new_state, dd # --------------------------- # History ops # --------------------------- def save_picks_to_history(picks: List[Dict[str, Any]], pin: str) -> str: if not check_admin_pin(pin): return "❌ Pogrešan PIN (ili ADMIN_PIN nije podešen u Secrets)." if not picks: return "Nema preporuka za snimiti (prvo pokreni analizu)." hist = read_history() existing = {r.get("id") for r in hist if isinstance(r, dict) and r.get("id")} added = 0 for p in picks: try: bid = make_bet_id(p.get("date_str",""), int(p.get("league_id", -1)), int(p.get("game_id", 0))) except Exception: continue if bid in existing: continue hist.append({ "id": bid, "date_str": p.get("date_str"), "day_choice": p.get("day_choice"), "league": p.get("league"), "match": p.get("match"), "favorite": p.get("favorite"), "fav_odds": p.get("fav_odds"), "bucket": p.get("bucket"), "safety": p.get("safety"), "eg_fav": p.get("eg_fav"), "floor_rate": p.get("floor_rate"), "h2h_rate": p.get("h2h_rate"), "result": "PENDING", "marked_at": None, "created_at": now_ts(), "notes": p.get("signal","") or "", }) existing.add(bid) added += 1 write_history(hist) return f"✅ Sačuvano u rezultati: {added} novih opklada. Ukupno u istoriji: {len(hist)}." def load_history_simple(date_str: str): date_str = (date_str or "").strip() or dt.date.today().strftime("%Y-%m-%d") hist = read_history() rows = [r for r in hist if isinstance(r, dict) and r.get("date_str") == date_str] md = render_history_markdown(rows) stats = history_stats(rows) choices = [] for r in rows: bid = r.get("id") fav = r.get("favorite", "") match_ = r.get("match", "") if bid: choices.append((f"{fav} — {match_}", bid)) dd = gr.update(choices=choices, value=(choices[0][1] if choices else None)) return md, stats, dd def set_bet_result_simple(date_str: str, bet_id: str, new_result: str, notes: str, pin: str): if not check_admin_pin(pin): return "❌ Pogrešan PIN (ili ADMIN_PIN nije podešen u Secrets).", gr.update(), gr.update(), gr.update() date_str = (date_str or "").strip() or dt.date.today().strftime("%Y-%m-%d") bet_id = (bet_id or "").strip() if not bet_id: return "❌ Izaberi opkladu.", gr.update(), gr.update(), gr.update() hist = read_history() found = False for r in hist: if isinstance(r, dict) and r.get("id") == bet_id: r["result"] = normalize_result(new_result) r["notes"] = notes or "" r["marked_at"] = now_ts() if r["result"] in ("WIN", "LOSS") else None found = True break if not found: return "❌ Ne mogu naći opkladu (probaj ponovo učitati).", gr.update(), gr.update(), gr.update() write_history(hist) rows = [r for r in hist if isinstance(r, dict) and r.get("date_str") == date_str] md = render_history_markdown(rows) stats = history_stats(rows) choices = [] for r in rows: bid = r.get("id") fav = r.get("favorite", "") match_ = r.get("match", "") if bid: choices.append((f"{fav} — {match_}", bid)) dd = gr.update( choices=choices, value=(bet_id if any(v == bet_id for _, v in choices) else (choices[0][1] if choices else None)) ) return "✅ Sačuvano.", md, stats, dd # --------------------------- # UI # --------------------------- with gr.Blocks(title="PlexBet — Hokej favorit 1+ gol") as demo: gr.Markdown( "## PlexBet — Hokej: Top 10 tipova (favorit da bar 1 gol)\n" f"- Filter: favorit kvota ≤ **{FAV_MAX_ODDS}**\n" "- 📌**NAPOMENA:** Da bi ostvarili zaradu morate pratit i poštovati pravila sistema koji igramo.\n" ) picks_state = gr.State([]) with gr.Tabs(): with gr.Tab("Analiza"): with gr.Row(): day_choice = gr.Radio( choices=["danas", "sutra"], value="danas", label="Za koje mečeve želiš analizu?" ) btn = gr.Button("Pokreni analizu", variant="secondary") status = gr.Markdown() output = gr.Markdown() gr.Markdown("### Admin (PIN)") admin_pin_1 = gr.Textbox(label="PIN (samo admin)", type="password", placeholder="Unesi ADMIN PIN") with gr.Row(): save_btn = gr.Button("Sačuvaj ove preporuke u rezultati", variant="primary") save_status = gr.Markdown() # Dropdown for delete gr.Markdown("---\n## 🗑️ Obriši meč iz cache-a (PIN)") del_dd = gr.Dropdown(choices=[], value=None, label="Izaberi meč za brisanje (iz cache-a)") with gr.Row(): del_pin = gr.Textbox(label="PIN (samo admin)", type="password", placeholder="Unesi ADMIN PIN") del_btn = gr.Button("Obriši iz cache-a", variant="stop") del_status = gr.Markdown() # Analysis click returns dropdown too btn.click(fn=run_analysis_cached, inputs=[day_choice], outputs=[status, output, picks_state, del_dd]) save_btn.click(fn=save_picks_to_history, inputs=[picks_state, admin_pin_1], outputs=[save_status]) del_btn.click( fn=delete_pick_from_cache, inputs=[day_choice, del_dd, del_pin, output, picks_state], outputs=[del_status, output, picks_state, del_dd], ) # ---- Manual add ---- gr.Markdown("---\n## ✍️ Ručno dodaj meč (PIN)") with gr.Row(): man_league = gr.Textbox(label="Liga", placeholder="npr. KHL / SHL / DEL ...") man_match = gr.Textbox(label="Meč", placeholder="npr. Team A vs Team B") with gr.Row(): man_fav = gr.Textbox(label="Favorit", placeholder="Team A") man_dog = gr.Textbox(label="Autsajder", placeholder="Team B") with gr.Row(): man_fav_odds = gr.Number(label="Kvota favorita", value=1.50) man_dog_odds = gr.Number(label="Kvota autsajdera", value=3.50) man_safety = gr.Slider(label="Sigurnost (%)", minimum=50, maximum=97, value=90, step=1) man_notes = gr.Textbox(label="Napomena (opciono)", placeholder="Zašto dodaješ ovaj meč ručno...") with gr.Row(): man_pin = gr.Textbox(label="PIN (samo admin)", type="password", placeholder="Unesi ADMIN PIN") man_add_btn = gr.Button("Dodaj meč", variant="primary") man_add_status = gr.Markdown() man_add_btn.click( fn=add_manual_match, inputs=[day_choice, man_league, man_match, man_fav, man_dog, man_fav_odds, man_dog_odds, man_safety, man_notes, man_pin, output, picks_state], outputs=[man_add_status, output, picks_state, del_dd], ) with gr.Tab("Rezultati"): gr.Markdown("### Rezultati\nUčitaj datum → izaberi opkladu (Favorit — Meč) → WIN/LOSS → sačuvaj (PIN).") with gr.Row(): hist_date = gr.Textbox(value=dt.date.today().strftime("%Y-%m-%d"), label="Datum (Godina-Mjesec-Dan) | Primjer: 2026-03-15") load_btn = gr.Button("Učitaj") history_md = gr.Markdown() history_stats_md = gr.Markdown() bet_id_dd = gr.Dropdown(choices=[], value=None, label="Izaberi opkladu (Favorit — Meč)") with gr.Row(): new_result = gr.Dropdown(choices=["WIN", "LOSS", "PENDING"], value="WIN", label="Novi rezultat") notes = gr.Textbox(label="Napomena (opciono)") admin_pin_2 = gr.Textbox(label="PIN (samo admin)", type="password", placeholder="Unesi ADMIN PIN") save_one_btn = gr.Button("Sačuvaj rezultat", variant="primary") save_one_status = gr.Markdown() load_btn.click(fn=load_history_simple, inputs=[hist_date], outputs=[history_md, history_stats_md, bet_id_dd]) save_one_btn.click( fn=set_bet_result_simple, inputs=[hist_date, bet_id_dd, new_result, notes, admin_pin_2], outputs=[save_one_status, history_md, history_stats_md, bet_id_dd], ) if __name__ == "__main__": demo.launch(server_name="0.0.0.0", server_port=7860, ssr_mode=False)