Add boss schedule retrieval & parsing

.gitignore

@@ -1,2 +1,3 @@
 __pycache__
-.env
+.env
+boss.yaml

config.py

@@ -7,6 +7,10 @@ class Settings(BaseSettings):
     fernet_key: str
     cookie_httponly: bool = False
     frontend_url: str
+    boss_url: str
+    boss_aes_key: str
+    boss_hmac_key: str
+    
     model_config = SettingsConfigDict(env_file=".env", extra="ignore")
     cookie_secure: bool = False
 

main.py

@@ -5,6 +5,10 @@ import uvicorn
 from fastapi.middleware.cors import CORSMiddleware
 from database import init_db
 from routes import logout
+from routes import boss
+from services.boss_retrieval import process_boss_file
+from contextlib import asynccontextmanager
+import asyncio
 
 app = FastAPI()
 
@@ -17,6 +21,29 @@ app.add_middleware(
     expose_headers=["Set-Cookie"],
 )
 
+async def boss_worker_loop():
+    print("background worker started")
+    while True:
+        try:
+            print("running boss service")
+            process_boss_file()
+        except Exception as e:
+            print(f"worker error: {e}")
+        
+        # TODO: actually sync to what schedule says.
+        # for now its set to 1hr just incase pretendo fuckery happens
+        # (such as rotation ending)
+        await asyncio.sleep(3600)
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    task = asyncio.create_task(boss_worker_loop())
+    yield
+    print("shutdown: cancelling background tasks")
+    task.cancel()
+
+app = FastAPI(lifespan=lifespan)
+
 @app.middleware("http")
 async def force_cors_on_errors(request: Request, call_next):
     response = await call_next(request)
@@ -29,7 +56,7 @@ async def force_cors_on_errors(request: Request, call_next):
 app.include_router(sso.router, prefix="/api/v2/sso")
 app.include_router(user.router, prefix="/api/v1/users")
 app.include_router(logout.router, prefix="/api/v1")
+app.include_router(boss.router, prefix="/api/v1")
 
 if __name__ == '__main__':
-    uvicorn.run("main:app", host="0.0.0.0", port=settings.port, reload=True)
-    init_db()
+    uvicorn.run("main:app", host="0.0.0.0", port=settings.port, reload=True)

routes/boss.py

@@ -0,0 +1,79 @@
+from fastapi import APIRouter, Request, Response
+import yaml
+import json
+from datetime import datetime, timedelta
+
+router = APIRouter()
+
+#ripped straight from splatcord. sorry not sorry?
+MAP_DATA = {
+    0: {"it-IT": "Periferia urbana", "de-DE": "Dekabahnstation", "en-US": "Urchin Underpass", "es-ES": "Parque Viaducto", "ja-JP": "デカライン高架下"},
+    1: {"it-IT": "Magazzino", "de-DE": "Kofferfisch-Lager", "en-US": "Walleye Warehouse", "es-ES": "Almacén Rodaballo", "ja-JP": "ハコフグ倉庫"},
+    2: {"it-IT": "Raffineria", "de-DE": "Bohrinsel Nautilus", "en-US": "Saltspray Rig", "es-ES": "Plataforma Gaviota", "ja-JP": "シオノメ油田"},
+    3: {"it-IT": "Centro commerciale", "de-DE": "Arowana-Center", "en-US": "Arowana Mall", "es-ES": "Plazuela del Calamar", "ja-JP": "アロワナモール"},
+    4: {"it-IT": "Pista Polposkate", "de-DE": "Punkasius-Skatepark", "en-US": "Blackbelly Skatepark", "es-ES": "Parque Lubina", "ja-JP": "Ｂバスパーク"},
+    5: {"it-IT": "Campeggio Totan", "de-DE": "Camp Schützenfisch", "en-US": "Camp Triggerfish", "es-ES": "Campamento Arowana", "ja-JP": "モンガラキャンプ場"},
+    6: {"it-IT": "Porto Polpo", "de-DE": "Heilbutt-Hafen", "en-US": "Port Mackerel", "es-ES": "Puerto Jurel", "ja-JP": "ホッケふ頭"},
+    7: {"it-IT": "Serra di alghe", "de-DE": "Tümmlerkuppel", "en-US": "Kelp Dome", "es-ES": "Jardín botánico", "ja-JP": "モズク農園"},
+    8: {"it-IT": "Torri cittadine", "de-DE": "Muränentürme", "en-US": "Moray Towers", "es-ES": "Torres Merluza", "ja-JP": "タチウオパーキング"},
+    9: {"it-IT": "Molo Mollusco", "de-DE": "Blauflossen-Depot", "en-US": "Bluefin Depot", "es-ES": "Mina costera", "ja-JP": "ネギトロ炭鉱"},
+    10: {"it-IT": "Ponte Sgombro", "de-DE": "Makrelenbrücke", "en-US": "Hammerhead Bridge", "es-ES": "Puente Salmón", "ja-JP": "マサバ海峡大橋"},
+    11: {"it-IT": "Cime sogliolose", "de-DE": "Schollensiedlung", "en-US": "Flounder Heights", "es-ES": "Complejo Medusa", "ja-JP": "ヒラメが丘団地"},
+    12: {"it-IT": "Museo di Cefalò", "de-DE": "Pinakoithek", "en-US": "Museum d'Alfonsino", "es-ES": "Museo del Pargo", "ja-JP": "キンメダイ美術館"},
+    13: {"it-IT": "Acciugames", "de-DE": "Anchobit Games HQ", "en-US": "Ancho-V Games", "es-ES": "Estudios Esturión", "ja-JP": "アンチョビットゲームズ"},
+    14: {"it-IT": "Miniera d'Orata", "de-DE": "Steinköhler-Grube", "en-US": "Piranha Pit", "es-ES": "Cantera Tintorera", "ja-JP": "ショッツル鉱山"},
+    15: {"it-IT": "Villanguilla", "de-DE": "Mahi-Mahi Resort", "en-US": "Mahi-Mahi Resort", "es-ES": "Spa Cala Bacalao", "ja-JP": "マヒマヒリゾート＆スパ"}
+}
+
+RULE_NAMES = {
+    "cVar": "SplatZones",
+    "cVgl": "Rainmaker",
+    "cVlf": "TowerControl",
+    "cPnt": "TurfWar"
+}
+
+@router.api_route('/boss', methods=['GET', 'POST'])
+async def boss_rotation(request: Request):
+    try:
+        with open("boss.yaml", "r") as f:
+            yaml_data = yaml.safe_load(f)
+        
+        base_time_str = yaml_data.get("ByamlInfo", {}).get("BaseByamlStartTime", "2026-02-06T06:00:00.0000000Z")
+        base_time_str = base_time_str.replace('Z', '+00:00')
+        current_time = datetime.fromisoformat(base_time_str)
+
+        rotations = {}
+
+        for phase in yaml_data.get("Phases", []):
+            ts_ms = str(int(current_time.timestamp() * 1000))
+            turf_stages = []
+            for s in phase.get("RegularStages", []):
+                mid = s.get("MapID")
+                turf_stages.append({"mapID": mid, "translatedNames": MAP_DATA.get(mid, {})})
+
+            ranked_stages = []
+            for s in phase.get("GachiStages", []):
+                mid = s.get("MapID")
+                ranked_stages.append({"mapID": mid, "translatedNames": MAP_DATA.get(mid, {})})
+
+            rotations[ts_ms] = {
+                "turfStages": turf_stages,
+                "rankedStages": ranked_stages,
+                "rankedMode": RULE_NAMES.get(phase.get("GachiRule"), "SplatZones")
+            }
+
+            current_time += timedelta(hours=2)
+
+        response_data = {
+            "nintendo": {
+                "notice": "Nintendo Network has been shut down. Thanks for your interest."
+            },
+            "pretendo": {
+                "rotations": rotations
+            }
+        }
+        
+        return Response(content=json.dumps(response_data), media_type="application/json")
+
+    except Exception as e:
+        return Response(content=json.dumps({"error": str(e)}), status_code=500, media_type="application/json")

services/boss_retrieval.py

@@ -0,0 +1,60 @@
+import subprocess
+import requests
+import oead
+import os
+from config import settings
+
+def process_boss_file():
+    temp_boss = "bosstemp.bin"
+    output_yaml = "boss.yaml"
+    decrypt_script = os.path.join("services", "decrypt.js")
+    
+    if not os.path.exists(decrypt_script):
+        print(f"{decrypt_script} is missing!")
+        return False
+
+    try:
+        boss_url = settings.boss_url
+        res = requests.get(boss_url, timeout=10)
+        res.raise_for_status()
+        
+        with open(temp_boss, "wb") as f:
+            f.write(res.content)
+
+        node_env = os.environ.copy()
+        node_env["BOSS_AES_KEY"] = settings.boss_aes_key
+        node_env["BOSS_HMAC_KEY"] = settings.boss_hmac_key
+
+        result = subprocess.run(
+            ['node', decrypt_script, temp_boss],
+            capture_output=True,
+            text=False,
+            env=node_env
+        )
+        
+        if result.returncode != 0:
+            print(f"decrypt fail! {result.stderr.decode()}")
+            return False
+
+        byml_obj = oead.byml.from_binary(result.stdout)
+        yaml_content = oead.byml.to_text(byml_obj)
+        
+        with open(output_yaml, "w", encoding="utf-8") as f:
+            f.write(yaml_content)
+            
+        print(f"yay! success: {output_yaml}")
+        return True
+
+    except Exception as e:
+        print(f"exception {e}")
+        return False
+
+    finally:
+        print("garbage pickup is running (boss)")
+        if os.path.exists(temp_boss):
+            os.remove(temp_boss)
+            print(f"Removed {temp_boss}")
+            
+        if os.path.exists("boss.byml"):
+            os.remove("boss.byml")
+            print("Removed boss.byml")

services/decrypt.js

@@ -0,0 +1,44 @@
+const fs = require('node:fs');
+const path = require('node:path');
+const crypto = require('node:crypto');
+
+const BOSS_AES_KEY = process.env.BOSS_AES_KEY;
+const BOSS_HMAC_KEY = process.env.BOSS_HMAC_KEY;
+
+async function run() {
+    try {
+        if (!BOSS_AES_KEY || !BOSS_HMAC_KEY) {
+            process.stderr.write("decrypt: missing environment keys\n");
+            process.exit(1);
+        }
+
+        const data = fs.readFileSync(process.argv[2]);
+
+        const IV = Buffer.concat([
+            data.subarray(0x0C, 0x18),
+            Buffer.from([0x00, 0x00, 0x00, 0x01])
+        ]);
+
+        const decipher = crypto.createDecipheriv('aes-128-ctr', Buffer.from(BOSS_AES_KEY, 'hex'), IV);
+        const decrypted = Buffer.concat([decipher.update(data.subarray(0x20)), decipher.final()]);
+
+        const hmac = decrypted.subarray(0, 0x20);
+        const content = decrypted.subarray(0x20);
+
+        const calculatedHmac = crypto.createHmac('sha256', Buffer.from(BOSS_HMAC_KEY, 'hex'))
+            .update(content)
+            .digest();
+
+        if (!calculatedHmac.equals(hmac)) {
+            process.stderr.write("Console HMAC check failed. Is the key correct?\n");
+            process.exit(1);
+        }
+
+        process.stdout.write(content);
+    } catch (err) {
+        process.stderr.write(`decrypt.js: ${err.message}\n`);
+        process.exit(1);
+    }
+}
+
+run();