Files

88 lines
4.8 KiB
Python

"""Persistent, revision-checked hint library for the lab. Text is data, never executable."""
import json
import os
from pathlib import Path
import threading
import uuid
ROOT = Path(__file__).resolve().parent
PATH = ROOT / ".lab-data" / "hints.json"
LOCK = threading.Lock()
PHASES = ["ALL","BRIEFING","ARCADE","BOSS","REVEAL","CELLS","CALIBRATION_TUTORIAL","CALIBRATION","LAUNCH_ARM","COUNTDOWN","DEBRIEF"]
TARGETS = ["room"] + ["P01","P02A","P02B","P03","P04","P05","P06"] + [f"cabinet_{i:02}" for i in range(1,7)]
def defaults():
hints=[]
for id,title,text in [
("P01","Claw controls","Start by investigating the claw machine controls."),
("P02A","First claw reward","Now that the controls work, try retrieving the first token."),
("P02B","Second claw reward","The first token has arrived. The claw still has another reward."),
("P03","Records and sleeves","Look at the records together with their sleeves."),
("P04","Service panel","Compare the symbols on the service panel."),
("P05","Ticket values","Look closely at the values on the tickets."),
("P06","Poster clues","There may be more to the poster when you look at its layers.")]:
hints.append({"id":"default-"+id,"title":title,"text":text,"categories":["Discovery"],"phase":"ARCADE","target":id,"condition":"incomplete","level":1})
for id,title,text,phase,category in [
("tokens","Matching tokens","Each token belongs to one machine. Look for the cabinet name.","ARCADE","Arcade"),
("cells","Power cells","Charge each cell, remove it from its bay, then place it in its matching ship socket.","CELLS","Ship"),
("practice","Practice together","Name the station, repeat the target, hold it steady, press Ready, then Execute.","CALIBRATION_TUTORIAL","Calibration"),
("calibration","Team coordination","Read back the target before pressing Ready. The navigator presses Execute when the team is ready.","CALIBRATION","Calibration"),
("launch","Final launch","Arm the stations in order. Confirm each group is ready before the navigator commits the launch.","LAUNCH_ARM","Finale")]:
hints.append({"id":"default-"+id,"title":title,"text":text,"categories":[category],"phase":phase,"target":"room","condition":"any","level":1})
return {"revision":0,"hints":hints}
def read():
with LOCK:
return _read()
def _read():
if not PATH.exists(): return defaults()
data=json.loads(PATH.read_text(encoding="utf-8"))
if not isinstance(data,dict) or not isinstance(data.get("revision"),int) or not isinstance(data.get("hints"),list):
raise ValueError("Hint file is invalid; it was not overwritten")
return data
class Conflict(ValueError): pass
def validate(hint):
if not isinstance(hint,dict): raise ValueError("Missing hint")
clean={}
for key,limit in (("title",100),("text",1500)):
value=hint.get(key)
if not isinstance(value,str) or not 1<=len(value.strip())<=limit:
raise ValueError(f"{key} must contain 1-{limit} characters")
clean[key]=value.strip()
categories=hint.get("categories",[])
if not isinstance(categories,list) or not 1<=len(categories)<=8 or any(not isinstance(x,str) or not 1<=len(x.strip())<=40 for x in categories):
raise ValueError("Use 1-8 category names, each up to 40 characters")
clean["categories"]=list(dict.fromkeys(x.strip() for x in categories))
for key,allowed in (("phase",PHASES),("target",TARGETS),("condition",["any","incomplete","complete"]),("level",[1,2,3])):
if hint.get(key) not in allowed: raise ValueError("Invalid "+key)
clean[key]=hint[key]
return clean
def save(body):
if not isinstance(body,dict): raise ValueError("Expected a hint object")
hint=validate(body.get("hint"))
with LOCK:
data=_read()
if body.get("revision")!=data["revision"]: raise Conflict("Hints changed on another device. Reload the library, then save your draft again.")
id=body["hint"].get("id","")
if id:
index=next((i for i,h in enumerate(data["hints"]) if h["id"]==id),None)
if index is None: raise ValueError("Hint no longer exists")
data["hints"][index]={"id":id,**hint}
else:
if len(data["hints"])>=1000: raise ValueError("Library is limited to 1000 hints")
data["hints"].append({"id":uuid.uuid4().hex,**hint})
data["revision"]+=1
PATH.parent.mkdir(parents=True,exist_ok=True)
temp=PATH.with_suffix(".tmp")
with temp.open("w",encoding="utf-8") as stream:
json.dump(data,stream,ensure_ascii=False,indent=2)
stream.flush()
os.fsync(stream.fileno())
if PATH.exists():
PATH.with_suffix(".bak").write_bytes(PATH.read_bytes())
temp.replace(PATH)
return data