#!/usr/bin/env python3 """ 🎯 LaoPatana VIP Lottery Sender v5.2.1 Queries PostgreSQL directly for predictions — no more HTML scraping! Runs daily at 18:00 via cron """ import subprocess import os import json import requests from datetime import datetime def _read_secret(path, default=""): try: with open(path) as f: return f.read().strip() except Exception: return default TOKEN = os.environ.get("VIP_TELEGRAM_BOT_TOKEN") or _read_secret("/etc/nami-harness/vip_telegram_token") CHANNEL = "-1003736959465" DB_ENV = {"PGPASSWORD": os.environ.get("PGPASSWORD") or _read_secret("/etc/nami-harness/postgres_password")} PSQL = ["psql", "-h", "127.0.0.1", "-U", "postgres", "-d", "laopatana_stat_lab", "-t", "-A", "-F", "|"] def db_query(sql): """Run a psql query and return stripped output""" try: r = subprocess.run(PSQL + ["-c", sql], capture_output=True, text=True, env=DB_ENV, timeout=15) if r.returncode != 0: print(f"⚠️ DB error: {r.stderr[:200]}") return None return r.stdout.strip() except Exception as e: print(f"❌ DB unreachable: {e}") return None def fetch_vip_data(): """Fetch all VIP prediction data from DB""" data = { "time": datetime.now().strftime("%d/%m/%Y %H:%M"), "ok": False, } # 1. Get latest locked prediction pred_rows = db_query(""" SELECT p.id, p.target_draw_date::date, rv.config_json->>'engineVersion' as engine_ver FROM predictions p JOIN rule_versions rv ON p.rule_version_id = rv.id WHERE p.status = 'locked' ORDER BY p.created_at DESC LIMIT 1 """) if not pred_rows: data["error"] = "No locked prediction found" return data parts = pred_rows.split("|") if len(parts) < 3: data["error"] = "Bad prediction row" return data pred_id = parts[0] data["target_date"] = parts[1] data["engine"] = f"Engine {parts[2]}" if parts[2] else "AI Engine" # 2. Get prediction items (non-rejected only) items_raw = db_query(f""" SELECT bet_type, number, rank, score::numeric(5,4), category FROM prediction_items WHERE prediction_id = '{pred_id}' AND is_rejected = false ORDER BY bet_type, rank """) if not items_raw: data["error"] = "No prediction items" return data # Group by bet_type by_type = {} for line in items_raw.split("\n"): cols = line.split("|") if len(cols) < 5: continue bt, num, rank, score, cat = cols[0], cols[1], cols[2], cols[3], cols[4] if bt not in by_type: by_type[bt] = [] by_type[bt].append({"number": num, "rank": int(rank), "score": float(score), "category": cat}) # 1D: top 3 data["1d"] = [x["number"] for x in by_type.get("1d", [])[:3]] # 2D main: top 5 data["2d_main"] = [x["number"] for x in by_type.get("2d", []) if x["category"] == "main"][:5] # 2D secondary: top 5 data["2d_secondary"] = [x["number"] for x in by_type.get("2d", []) if x["category"] == "secondary"][:5] # 3D: top 5 data["3d"] = [x["number"] for x in by_type.get("3d", []) if x["category"] == "main"][:5] # 3. Latest draw draw_raw = db_query(""" SELECT lao_last4, lao_last2, lao_last3, draw_date::date FROM draws WHERE status = 'drawn' AND lao_last2 IS NOT NULL ORDER BY draw_date DESC LIMIT 1 """) if draw_raw: cols = draw_raw.split("|") data["latest_4d"] = cols[0] if len(cols) > 0 else "?" data["latest_2d"] = cols[1] if len(cols) > 1 else "?" data["latest_3d"] = cols[2] if len(cols) > 2 else "?" data["latest_date"] = cols[3] if len(cols) > 3 else "?" # 4. Hit rate hit_raw = db_query(""" SELECT COUNT(*) FILTER (WHERE win_streak > 0) as hits, COUNT(*) as total FROM predictions WHERE status = 'evaluated' """) if hit_raw: cols = hit_raw.split("|") hits, total = int(cols[0]), int(cols[1]) data["hit_rate"] = f"{(hits/total*100):.1f}% ({hits}/{total})" if total > 0 else "N/A" data["ok"] = True return data def format_message(data): """Format VIP message in HTML""" if not data.get("ok"): err = data.get("error", "Unknown error") return f"❌ VIP Engine — {err}\n🔧 ตรวจสอบ Dashboard" today = datetime.now().strftime("%d/%m/%Y") msg = f"""🎯 LaoPatana VIP — {today} 🔮 {data.get('engine', 'AI Engine')} 📊 Hit Rate: {data.get('hit_rate', '?')} 📅 งวด: {data.get('target_date', '?')} ━━━━━━━━━━━━━━━━ """ # 1D Running msg += "🔥 1D วิ่ง\n " msg += " ".join(data.get("1d", [])) if data.get("1d") else "—" msg += "\n\n" # 2D Main msg += "🎲 2D หลัก\n " msg += " ".join(data.get("2d_main", [])) if data.get("2d_main") else "—" msg += "\n" # 2D Secondary if data.get("2d_secondary"): msg += "🎲 2D รอง\n " msg += " ".join(data["2d_secondary"]) msg += "\n" msg += "\n" # 3D Exact msg += "🎰 3D Exact\n " msg += " ".join(data.get("3d", [])) if data.get("3d") else "—" msg += "\n" msg += f""" ━━━━━━━━━━━━━━━━ 📜 งวดที่แล้ว: {data.get('latest_4d', '?')} ({data.get('latest_date', '?')}) 2D: {data.get('latest_2d', '?')} | 3D: {data.get('latest_3d', '?')} 🧬 Bonus Signals: digit_bias, miroshark, ensemble, gap_overdue, consecutive_pair, reverse_consecutive_pair, double_digit 📊 ดูสถิติเต็มที่ Dashboard ⚠️ การพนันมีความเสี่ยง โปรดใช้วิจารณญาณ 🌸 ส่งโดยนามิ — @namiByPao_bot""" return msg def send_to_vip(message): """Send message to VIP channel""" url = f"https://api.telegram.org/bot{TOKEN}/sendMessage" try: resp = requests.post(url, json={ "chat_id": CHANNEL, "text": message, "parse_mode": "HTML", "disable_web_page_preview": True }, timeout=15) return resp.json().get("ok", False) except Exception as e: print(f"❌ Send failed: {e}") return False def main(): print(f"[{datetime.now()}] 🎯 VIP Lottery Sender v5.2.1 (DB mode)") data = fetch_vip_data() if data.get("ok"): print(f" Engine: {data.get('engine', '?')}") print(f" 1D: {data.get('1d', [])}") print(f" 2D Main: {data.get('2d_main', [])}") print(f" 2D Sec: {data.get('2d_secondary', [])}") print(f" 3D: {data.get('3d', [])}") print(f" Hit Rate: {data.get('hit_rate', '?')}") print(f" Latest: {data.get('latest_4d', '?')} ({data.get('latest_date', '?')})") else: print(f" ❌ Error: {data.get('error', 'Unknown')}") message = format_message(data) ok = send_to_vip(message) if ok: print(f" ✅ Sent to VIP channel successfully!") else: print(f" ❌ Failed to send") return ok if __name__ == "__main__": main()