import json import time import urllib.parse import urllib.request from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path DATA_FILE = Path("giveaway-data.json") DEFAULT_DATA = { "active": False, "startAt": 0, "requiredRarity": "any", "minMessages": 10, "participants": [], "streamers": [], "users": [], } TWITCH_CLIENT_ID = "50azs0evb4cpv3i5pwfm67aqt8alje" TWITCH_CLIENT_SECRET = "c6a19ki9qnxjbu3mm7nj8k1w6nslzl" TWITCH_CHANNEL_LOGIN = "obn4l" _TWITCH_TOKEN = {"token": "", "expires_at": 0} TG_BOT_TOKEN = "8588784310:AAG4H41x39kX8bskkBWxB9lyVAfTjHAnR1g" TG_CHANNEL_ID = "-1002436946838" def load_data(): if DATA_FILE.exists(): try: return json.loads(DATA_FILE.read_text(encoding="utf-8")) except Exception: return DEFAULT_DATA.copy() return DEFAULT_DATA.copy() def save_data(data): tmp = DATA_FILE.with_suffix(".tmp") tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(DATA_FILE) def twitch_configured() -> bool: return bool(TWITCH_CLIENT_ID and TWITCH_CLIENT_SECRET) def twitch_get_app_token() -> str: now = int(time.time()) if _TWITCH_TOKEN["token"] and now < _TWITCH_TOKEN["expires_at"]: return _TWITCH_TOKEN["token"] params = { "client_id": TWITCH_CLIENT_ID, "client_secret": TWITCH_CLIENT_SECRET, "grant_type": "client_credentials", } data = urllib.parse.urlencode(params).encode() req = urllib.request.Request("https://id.twitch.tv/oauth2/token", data=data) with urllib.request.urlopen(req, timeout=20) as resp: payload = json.loads(resp.read().decode()) token = payload.get("access_token", "") expires_in = int(payload.get("expires_in", 0)) _TWITCH_TOKEN["token"] = token _TWITCH_TOKEN["expires_at"] = now + max(60, expires_in - 60) return token def twitch_api_get(endpoint: str, params: dict) -> dict: token = twitch_get_app_token() query = urllib.parse.urlencode(params, doseq=True) url = f"https://api.twitch.tv/helix/{endpoint}?{query}" req = urllib.request.Request(url) req.add_header("Client-Id", TWITCH_CLIENT_ID) req.add_header("Authorization", f"Bearer {token}") with urllib.request.urlopen(req, timeout=20) as resp: return json.loads(resp.read().decode()) def twitch_api_get_user(token: str, endpoint: str, params: dict) -> dict: query = urllib.parse.urlencode(params, doseq=True) url = f"https://api.twitch.tv/helix/{endpoint}?{query}" req = urllib.request.Request(url) req.add_header("Client-Id", TWITCH_CLIENT_ID) req.add_header("Authorization", f"Bearer {token}") with urllib.request.urlopen(req, timeout=20) as resp: return json.loads(resp.read().decode()) def check_twitch_follow(user_token: str) -> bool: user_data = twitch_api_get_user(user_token, "users", {}).get("data", []) if not user_data: return False user_id = user_data[0]["id"] target_data = twitch_api_get("users", {"login": [TWITCH_CHANNEL_LOGIN]}).get("data", []) if not target_data: return False target_id = target_data[0]["id"] follows = twitch_api_get_user(user_token, "users/follows", { "from_id": user_id, "to_id": target_id, }).get("data", []) return len(follows) > 0 def check_telegram_member(telegram_user_id: int) -> bool: if not TG_BOT_TOKEN or not TG_CHANNEL_ID: return False params = urllib.parse.urlencode({ "chat_id": TG_CHANNEL_ID, "user_id": telegram_user_id, }) url = f"https://api.telegram.org/bot{TG_BOT_TOKEN}/getChatMember?{params}" with urllib.request.urlopen(url, timeout=20) as resp: payload = json.loads(resp.read().decode()) if not payload.get("ok"): return False status = (payload.get("result") or {}).get("status", "") return status in {"member", "administrator", "creator"} def parse_twitch_login(raw: str) -> str | None: raw = (raw or "").strip() if not raw: return None if "twitch.tv" in raw: try: url = urllib.parse.urlparse(raw) path = url.path.strip("/").split("/") if not path: return None return path[0].lower() except Exception: return None return raw.lower() def get_streamers_info(streamers: list[dict]) -> list[dict]: if not streamers: return [] if not twitch_configured(): return [ { "login": s["login"], "url": s.get("url"), "telegram": s.get("telegram"), "displayName": s.get("login"), "avatarUrl": "", "isLive": False, "viewers": 0, "game": "", "title": "", } for s in streamers ] logins = [s["login"] for s in streamers] users = twitch_api_get("users", {"login": logins}).get("data", []) streams = twitch_api_get("streams", {"user_login": logins}).get("data", []) users_by_login = {u["login"].lower(): u for u in users} streams_by_login = {s["user_login"].lower(): s for s in streams} result = [] for s in streamers: login = s["login"] u = users_by_login.get(login, {}) st = streams_by_login.get(login, {}) result.append({ "login": login, "url": s.get("url") or f"https://www.twitch.tv/{login}", "telegram": s.get("telegram"), "displayName": u.get("display_name") or login, "avatarUrl": u.get("profile_image_url") or "", "isLive": bool(st), "viewers": int(st.get("viewer_count", 0)) if st else 0, "game": st.get("game_name") or "", "title": st.get("title") or "", }) return result class Handler(SimpleHTTPRequestHandler): def guess_type(self, path): ctype = super().guess_type(path) if ctype in {"text/html", "text/css", "application/javascript"}: return f"{ctype}; charset=utf-8" return ctype def _send_json(self, code, payload): body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) def do_GET(self): if self.path.startswith("/api/state"): data = load_data() self._send_json(200, { "active": data.get("active", False), "startAt": data.get("startAt", 0), "requiredRarity": data.get("requiredRarity", "any"), "minMessages": data.get("minMessages", 10), }) return if self.path.startswith("/api/participants"): data = load_data() self._send_json(200, {"participants": data.get("participants", [])}) return if self.path.startswith("/api/streamers"): data = load_data() streamers = data.get("streamers", []) try: info = get_streamers_info(streamers) self._send_json(200, {"streamers": info}) except Exception as e: self._send_json(500, {"error": str(e)}) return if self.path.startswith("/api/users"): data = load_data() self._send_json(200, {"users": data.get("users", [])}) return if self.path.startswith("/api/check/twitch"): self._send_json(405, {"ok": False}) return if self.path.startswith("/api/check/telegram"): self._send_json(405, {"ok": False}) return super().do_GET() def do_POST(self): length = int(self.headers.get("Content-Length", "0")) raw = self.rfile.read(length) if length > 0 else b"{}" try: payload = json.loads(raw.decode("utf-8")) except Exception: payload = {} if self.path.startswith("/api/admin/start"): data = load_data() start_delay = int(payload.get("startDelaySec", 30)) data["active"] = True data["startAt"] = int(time.time() * 1000) + start_delay * 1000 data["requiredRarity"] = payload.get("requiredRarity", "any") data["minMessages"] = int(payload.get("minMessages", 10)) data["participants"] = [] save_data(data) self._send_json(200, {"ok": True}) return if self.path.startswith("/api/admin/stop"): data = load_data() data["active"] = False save_data(data) self._send_json(200, {"ok": True}) return if self.path.startswith("/api/admin/settings"): data = load_data() data["requiredRarity"] = payload.get("requiredRarity", "any") data["minMessages"] = int(payload.get("minMessages", 10)) save_data(data) self._send_json(200, {"ok": True}) return if self.path.startswith("/api/join"): data = load_data() if not data.get("active", False): self._send_json(409, {"ok": False, "error": "giveaway_not_active"}) return user = payload.get("user") or {} conditions = payload.get("conditions") or {} msg_count = int(conditions.get("messages", 0)) cond_ok = ( bool(conditions.get("twitch")) and bool(conditions.get("telegram")) and msg_count >= int(data.get("minMessages", 10)) and ( data.get("requiredRarity", "any") == "any" or conditions.get("rarity") == data.get("requiredRarity") ) ) if not cond_ok: self._send_json(400, {"ok": False, "error": "conditions_failed"}) return user_id = user.get("id") or user.get("login") or user.get("name") if not user_id: self._send_json(400, {"ok": False, "error": "missing_user"}) return if any(p.get("id") == user_id for p in data.get("participants", [])): self._send_json(200, {"ok": True, "already": True}) return data["participants"].append({ "id": user_id, "name": user.get("name"), "login": user.get("login"), "conditions": { "twitch": bool(conditions.get("twitch")), "telegram": bool(conditions.get("telegram")), "messages": msg_count, "rarity": conditions.get("rarity"), }, "joinedAt": int(time.time() * 1000), }) save_data(data) self._send_json(200, {"ok": True}) return if self.path.startswith("/api/users"): data = load_data() user = payload.get("user") or {} login = (user.get("login") or "").lower() if not login: self._send_json(400, {"ok": False, "error": "missing_login"}) return users = data.get("users", []) now_ms = int(time.time() * 1000) found = None for u in users: if u.get("login") == login: found = u break if not found: found = { "login": login, "displayName": user.get("displayName") or user.get("name") or login, "avatarUrl": user.get("avatarUrl") or "", "joinedAt": now_ms, } users.append(found) found["displayName"] = user.get("displayName") or user.get("name") or found.get("displayName") found["avatarUrl"] = user.get("avatarUrl") or found.get("avatarUrl") found["lastSeen"] = now_ms data["users"] = users save_data(data) self._send_json(200, {"ok": True}) return if self.path.startswith("/api/check/twitch"): token = payload.get("token") or "" if not token: self._send_json(400, {"ok": False, "error": "missing_token"}) return try: ok = check_twitch_follow(token) self._send_json(200, {"ok": ok}) except Exception as e: self._send_json(500, {"ok": False, "error": str(e)}) return if self.path.startswith("/api/check/telegram"): telegram_user_id = payload.get("telegramUserId") if not telegram_user_id: self._send_json(400, {"ok": False, "error": "missing_user"}) return try: ok = check_telegram_member(int(telegram_user_id)) self._send_json(200, {"ok": ok}) except Exception as e: self._send_json(500, {"ok": False, "error": str(e)}) return if self.path.startswith("/api/streamers"): data = load_data() login = parse_twitch_login(payload.get("login") or payload.get("url")) if not login: self._send_json(400, {"ok": False, "error": "invalid_url"}) return streamers = data.get("streamers", []) if any(s.get("login") == login for s in streamers): self._send_json(200, {"ok": True, "already": True}) return streamers.append({ "login": login, "url": payload.get("url") or f"https://www.twitch.tv/{login}", "telegram": payload.get("telegram", ""), }) data["streamers"] = streamers save_data(data) self._send_json(200, {"ok": True}) return if self.path.startswith("/api/streamers/delete"): data = load_data() login = parse_twitch_login(payload.get("login") or payload.get("url")) if not login: self._send_json(400, {"ok": False, "error": "invalid_login"}) return before = len(data.get("streamers", [])) data["streamers"] = [s for s in data.get("streamers", []) if s.get("login") != login] save_data(data) self._send_json(200, {"ok": True, "removed": before - len(data.get("streamers", []))}) return self._send_json(404, {"ok": False, "error": "not_found"}) def run(): server = ThreadingHTTPServer(("", 8000), Handler) print("Server started on http://localhost:8000") server.serve_forever() if __name__ == "__main__": run()