#!/usr/bin/env python3 # Планета Жопа — онлайн: aiohttp раздаёт игру, релеит позы дронов, и АВТОРИТЕТНО симулирует стадо # сфинктеров (бродят + разбегаются от дронов). Позиции стада одинаковы у всех игроков (сервер — источник истины). import json, asyncio, os, math, random, mimetypes from aiohttp import web, WSMsgType mimetypes.add_type("audio/ogg", ".ogg") # чтобы саундтрек отдавался с корректным MIME HERE = os.path.dirname(os.path.abspath(__file__)) INDEX = os.path.join(HERE, "serve", "index.html") players = {} # ws -> {"id","name","dx","dz"} (dx,dz = last drone x,z) _next = [0] # ---- terrain (ported 1:1 from the client groundH so heights/land match) ---- def _hash(x, y): h = math.sin(x*127.1 + y*311.7)*43758.5453 return h - math.floor(h) def _vnoise(x, y): xi = math.floor(x); yi = math.floor(y); xf = x-xi; yf = y-yi u = xf*xf*(3-2*xf); v = yf*yf*(3-2*yf) a = _hash(xi, yi); b = _hash(xi+1, yi); c = _hash(xi, yi+1); d = _hash(xi+1, yi+1) return a*(1-u)*(1-v) + b*u*(1-v) + c*(1-u)*v + d*u*v def _fbm(x, y): s = 0.0; a = 0.5; f = 1.0 for _ in range(5): s += a*_vnoise(x*f, y*f); f *= 2.03; a *= 0.5 return s TERR = 600.0; WATER_Y = 9.0 def groundH(x, z): n = _fbm(x*0.010+11, z*0.010+7) rim = min(1.0, math.hypot(x, z)/TERR) mtn = rim**2.2*78 + rim**6*40 return (n-0.5)*30 + mtn + _fbm(x*0.045, z*0.045)*5 + _fbm(x*0.02, z*0.02)*9 - 6 def on_land(x, z): return math.hypot(x, z) < TERR*0.88 and groundH(x, z) > WATER_Y + 1.0 # ---- herd ---- NHERD = 80 FLEE_R = 42.0 herd = [] def init_herd(): rnd = random.Random(20260709) tries = 0 while len(herd) < NHERD and tries < 6000: tries += 1 x = (rnd.random()-0.5)*TERR*1.7; z = (rnd.random()-0.5)*TERR*1.7 if on_land(x, z): herd.append({"x": x, "z": z, "vx": 0.0, "vz": 0.0, "w": rnd.random()*6.283, "sc": 0.8 + rnd.random()*0.9, "col": rnd.randint(0, 3), "age": 1 if rnd.random() < 0.42 else 0}) # skin tone 0-3, age young/old init_herd() def herd_payload(): return [[round(h["x"], 1), round(h["z"], 1), round(math.atan2(h["vz"], h["vx"]), 2), h["sc"], h["col"], h["age"]] for h in herd] async def herd_loop(): dt = 0.1 while True: drones = [(p["dx"], p["dz"]) for p in players.values() if p.get("dx") is not None] for h in herd: h["w"] += (random.random()-0.5)*0.6 # wander heading drift ax = math.cos(h["w"])*2.2; az = math.sin(h["w"])*2.2 # gentle wander accel for dx, dz in drones: # flee any nearby drone ddx = h["x"]-dx; ddz = h["z"]-dz; d2 = ddx*ddx + ddz*ddz if 1.0 < d2 < FLEE_R*FLEE_R: dd = math.sqrt(d2); s = (1 - dd/FLEE_R)*90.0 ax += ddx/dd*s; az += ddz/dd*s h["vx"] = h["vx"]*0.82 + ax*dt; h["vz"] = h["vz"]*0.82 + az*dt sp = math.hypot(h["vx"], h["vz"]); mx = 26.0 if sp > mx: h["vx"] *= mx/sp; h["vz"] *= mx/sp nx = h["x"] + h["vx"]*dt; nz = h["z"] + h["vz"]*dt if on_land(nx, nz): h["x"] = nx; h["z"] = nz else: # hit water/mountain -> turn back h["vx"] *= -0.6; h["vz"] *= -0.6; h["w"] += 3.14159 await broadcast(json.dumps({"t": "herd", "h": herd_payload()})) await asyncio.sleep(dt) async def broadcast(payload, exclude=None): dead = [] for w in list(players): if w is exclude or w.closed: continue try: await w.send_str(payload) except Exception: dead.append(w) for w in dead: players.pop(w, None) # ---- ЖОПО-СВО: ЗАБЕГ — server-authoritative FPV strike-drone race ---- # Круг из чекпоинтов-«рубежей». Сервер знает позы всех дронов (dx,dz,dy) => он и судит # проходы (античит бесплатно) и держит единый для всех таймер/старт/финиш. LAPS = 2 NCP = 8 CP_R = 28.0 # горизонтальный радиус прохода рубежа CP_VR = 60.0 # вертикальный допуск (высоту держать трудно — даём люфт) TIME_LIMIT = 240.0 race_cps = [] def init_race(): rnd = random.Random(770077) for i in range(NCP): ang = 2*math.pi*i/NCP + (rnd.random()-0.5)*0.55 rad = TERR*(0.46 + rnd.random()*0.26) x = math.cos(ang)*rad; z = math.sin(ang)*rad for _ in range(24): # подтянуть внутрь, пока не суша if on_land(x, z): break rad *= 0.93; x = math.cos(ang)*rad; z = math.sin(ang)*rad gy = groundH(x, z) + 26.0 + rnd.random()*14.0 race_cps.append({"x": round(x,1), "z": round(z,1), "y": round(gy,1)}) init_race() race = {"st": "idle", "clock": 0.0, "cd": 0.0, "results": []} prog = {} # pid -> {"cp","lap","fin","t"} def _pname(pid): for q in players.values(): if q["id"] == pid: return q["name"] return "дрон-%d" % pid def race_snapshot(full=False): m = {"t": "race", "st": race["st"], "cd": round(race["cd"],1), "clock": round(race["clock"],1), "laps": LAPS, "ncp": NCP, "stand": [{"id": pid, "name": _pname(pid), "cp": pr["cp"], "lap": pr["lap"], "fin": pr["fin"], "t": round(pr["t"],2)} for pid, pr in prog.items()]} if full or race["st"] in ("countdown", "racing"): m["cps"] = race_cps if race["st"] == "done": m["results"] = race["results"] return json.dumps(m) def _posed(): return [p for p in players.values() if p.get("dx") is not None] def start_race(): prog.clear() for p in players.values(): prog[p["id"]] = {"cp": 0, "lap": 0, "fin": False, "t": 0.0} race["st"] = "countdown"; race["cd"] = 4.0; race["clock"] = 0.0; race["results"] = [] async def race_loop(): dt = 0.1 while True: st = race["st"] if st == "idle": if _posed(): start_race() elif st == "countdown": race["cd"] -= dt if race["cd"] <= 0: race["st"] = "racing"; race["clock"] = 0.0 elif st == "racing": race["clock"] += dt for p in players.values(): pid = p["id"] pr = prog.get(pid) if pr is None: prog[pid] = pr = {"cp": 0, "lap": 0, "fin": False, "t": 0.0} if pr["fin"] or p.get("dx") is None: continue cp = race_cps[pr["cp"] % NCP] ddx = p["dx"]-cp["x"]; ddz = p["dz"]-cp["z"] dy = (p.get("dy") if p.get("dy") is not None else cp["y"]) - cp["y"] if ddx*ddx + ddz*ddz < CP_R*CP_R and abs(dy) < CP_VR: pr["cp"] += 1 if pr["cp"] % NCP == 0: pr["lap"] += 1 if pr["lap"] >= LAPS: pr["fin"] = True; pr["t"] = race["clock"] race["results"].append({"id": pid, "name": _pname(pid), "t": round(race["clock"],2)}) posed = _posed() done = posed and all(prog.get(p["id"], {}).get("fin") for p in posed) if done or race["clock"] > TIME_LIMIT or not posed: race["st"] = "done"; race["cd"] = 14.0 elif st == "done": race["cd"] -= dt if race["cd"] <= 0: race["st"] = "idle" await broadcast(race_snapshot()) await asyncio.sleep(dt) sprays = [] # Source-style decals shared by everyone (FIFO cap) SPRAY_CAP = 60 async def ws_handler(request): ws = web.WebSocketResponse(heartbeat=25) await ws.prepare(request) _next[0] += 1; pid = _next[0] players[ws] = {"id": pid, "name": "дрон-%d" % pid, "dx": None, "dz": None} await ws.send_str(json.dumps({"t": "welcome", "id": pid, "n": len(players)})) await ws.send_str(json.dumps({"t": "herd", "h": herd_payload()})) await ws.send_str(json.dumps({"t": "sprays", "l": sprays})) # existing sprays for the new joiner await ws.send_str(race_snapshot(full=True)) # current race state for the new joiner try: async for msg in ws: if msg.type == WSMsgType.TEXT: try: d = json.loads(msg.data) except Exception: continue if d.get("t") == "state" and "s" in d: s = d["s"]; players[ws]["dx"] = s.get("x"); players[ws]["dz"] = s.get("z"); players[ws]["dy"] = s.get("y") nm = str(d.get("n", players[ws]["name"]))[:18]; players[ws]["name"] = nm await broadcast(json.dumps({"t": "peer", "id": pid, "n": nm, "s": s}), exclude=ws) elif d.get("t") == "spray": sp = {"x": d.get("x", 0), "z": d.get("z", 0), "r": round(float(d.get("r", 0)), 2)} g = d.get("g") if isinstance(g, str) and g: sp["g"] = os.path.basename(g)[:64] # curated gallery file only (no path traversal, no raw uploads) else: sp["i"] = int(d.get("i", 0)) % 8 sprays.append(sp) if len(sprays) > SPRAY_CAP: sprays.pop(0) await broadcast(json.dumps({"t": "spray", **sp}), exclude=ws) elif msg.type == WSMsgType.ERROR: break finally: players.pop(ws, None) await broadcast(json.dumps({"t": "leave", "id": pid})) return ws async def jopa_event(request): # мост от TShock-плагина JopaBridge: событие Terraria -> новость всем веб-игрокам Планеты Жопа try: d = await request.json() except Exception: d = {} kind = str(d.get("kind", ""))[:24] msg = str(d.get("msg", ""))[:200] if msg: await broadcast(json.dumps({"t": "news", "kind": kind, "msg": msg})) return web.json_response({"ok": True}) async def index(request): return web.FileResponse(INDEX) async def health(request): return web.json_response({"ok": True, "players": len(players), "herd": len(herd), "sprays": len(sprays)}) async def sprays_list(request): d = os.path.join(HERE, "serve", "sprays") try: files = sorted(f for f in os.listdir(d) if f.lower().endswith((".png", ".jpg", ".jpeg", ".webp", ".gif"))) except Exception: files = [] return web.json_response(files) async def asset(request): # раздаём статику из serve/ (саундтрек и пр.) — только белый список расширений, без выхода из папки fn = os.path.basename(request.match_info.get("fn") or request.path) p = os.path.join(HERE, "serve", fn) ext = os.path.splitext(fn)[1].lower() CT = {".ogg": "audio/ogg", ".mp3": "audio/mpeg", ".wav": "audio/wav", ".m4a": "audio/mp4", ".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp", ".gif": "image/gif"} if os.path.isfile(p) and ext in CT: with open(p, "rb") as fh: data = fh.read() return web.Response(body=data, content_type=CT[ext], headers={"Cache-Control": "public, max-age=86400"}) return web.Response(status=404) async def on_start(app): app["herd"] = asyncio.create_task(herd_loop()) app["race"] = asyncio.create_task(race_loop()) app = web.Application() app.router.add_get("/", index) app.router.add_get("/ws", ws_handler) app.router.add_get("/health", health) app.router.add_post("/jopa_event", jopa_event) # мост Terraria(JopaBridge) -> Планета Жопа app.router.add_get("/sprays_list", sprays_list) app.router.add_get("/zhopa_theme.ogg", asset) # саундтрек (относительный путь из index.html) app.router.add_get("/zhopa_theme.mp3", asset) app.router.add_get("/media/{fn}", asset) # общий путь для будущих ассетов app.router.add_static("/sprays/", os.path.join(HERE, "serve", "sprays")) app.on_startup.append(on_start) if __name__ == "__main__": web.run_app(app, host="127.0.0.1", port=8790)