38 lines
5.3 KiB
Python
38 lines
5.3 KiB
Python
#!/usr/bin/env python3
|
||
import hashlib,hmac,json,os,pathlib,time,uuid,urllib.request,re
|
||
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
|
||
ROOT=pathlib.Path(os.environ.get('LIVING_LIGHTHOUSE_STATE','/var/lib/guanghu-living-lighthouse'));MAP=pathlib.Path(os.environ.get('LIVING_LIGHTHOUSE_MAP','/opt/guanghu-living-lighthouse/navigation.json'));KEY=os.environ.get('LIVING_LIGHTHOUSE_SIGNING_KEY','').encode();API=os.environ.get('OPENLUX_API_KEY','');HOST=os.environ.get('LIVING_LIGHTHOUSE_BIND','127.0.0.1');PORT=int(os.environ.get('LIVING_LIGHTHOUSE_PORT','8035'))
|
||
def require(v,e):
|
||
if not v:raise ValueError(e)
|
||
def read_json(h,limit=20000):
|
||
n=int(h.headers.get('content-length','0'));require(0<n<=limit,'body_size');v=json.loads(h.rfile.read(n));require(isinstance(v,dict),'body_object');return v
|
||
def sign(body):
|
||
payload=json.dumps(body,ensure_ascii=False,sort_keys=True,separators=(',',':'));return {'payload':payload,'sha256':hashlib.sha256(payload.encode()).hexdigest(),'hmac_sha256':hmac.new(KEY,payload.encode(),hashlib.sha256).hexdigest()}
|
||
def ignite(x):
|
||
require(KEY and x.get('schema')=='guanghu.linguistic-echo/v1' and x.get('scope')=='ENTERPRISE_PUBLIC' and x.get('session_id'),'echo_identity');h=x.get('human_language',{});p=x.get('persona_echo',{});now=int(time.time()*1000);require(h.get('human_id') and len(h.get('sha256',''))==64 and isinstance(h.get('occurred_at'),int),'human_language');require(p.get('persona_id') and len(p.get('sha256',''))==64 and isinstance(p.get('occurred_at'),int),'persona_echo');require(h['occurred_at']<=p['occurred_at'] and now-h['occurred_at']<=600000 and now-p['occurred_at']<=600000,'echo_time');body={'schema':'guanghu.living-lamp-receipt/v1','lamp_id':str(uuid.uuid4()),'state':'LIT','scope':'ENTERPRISE_PUBLIC','session_id':x['session_id'],'human_id':h['human_id'],'persona_id':p['persona_id'],'human_event_sha256':h['sha256'],'persona_echo_sha256':p['sha256'],'echo_kernel':'ECHO-KERNEL-0001','lit_at':now,'expires_at':now+900000,'reality_authority_granted':False};s=sign(body);d=ROOT/'lamps';d.mkdir(parents=True,exist_ok=True);f=d/(body['lamp_id']+'.json');f.write_text(json.dumps(s));os.chmod(f,0o600);return s
|
||
def lamp(lamp_id,session):
|
||
f=ROOT/'lamps'/(str(lamp_id)+'.json');require(f.is_file(),'lamp_unknown_or_dark');s=json.loads(f.read_text());require(hmac.compare_digest(s['hmac_sha256'],hmac.new(KEY,s['payload'].encode(),hashlib.sha256).hexdigest()),'lamp_integrity');b=json.loads(s['payload']);require(b['state']=='LIT' and b['session_id']==session and int(time.time()*1000)<=b['expires_at'],'lamp_dark_or_expired');return b
|
||
def model_pick(query,routes):
|
||
if not API:return None
|
||
body={'model':os.environ.get('OPENLUX_MODEL','qwen-flash'),'stream':False,'temperature':0,'response_format':{'type':'json_object'},'messages':[{'role':'system','content':'你是光湖企业灯塔的有界问路器官。只能从候选routes选择一个id；不确定输出null。只输出JSON字段id。不得发明路径或授予权限。'},{'role':'user','content':json.dumps({'query':query,'routes':routes[:80]},ensure_ascii=False)}]};req=urllib.request.Request(os.environ.get('OPENLUX_API_URL','https://api.openlux.ai/v1')+'/chat/completions',data=json.dumps(body).encode(),headers={'authorization':'Bearer '+API,'content-type':'application/json'},method='POST')
|
||
try:
|
||
with urllib.request.urlopen(req,timeout=30) as r:v=json.loads(r.read());choice=json.loads(v['choices'][0]['message']['content']);return choice.get('id')
|
||
except:return None
|
||
def ask(x):
|
||
b=lamp(x.get('lamp_id'),x.get('session_id'));q=str(x.get('query','')).strip();require(q,'query');m=json.loads(MAP.read_text());routes=m['routes'];hits=[r for r in routes if r['id'].lower()==q.lower()];
|
||
if not hits:hits=[r for r in routes if q.lower() in (' '.join(str(r.get(k,'')) for k in ('id','name','kind','route'))).lower()][:10]
|
||
assisted=False
|
||
if not hits and not re.fullmatch(r'[A-Z0-9∞_.:-]{2,160}',q):
|
||
selected=model_pick(q,routes);hits=[r for r in routes if r['id']==selected];assisted=bool(hits)
|
||
return {'outcome':'PASS','lamp':'LIT','echo_kernel':b['echo_kernel'],'routes':hits,'model_assisted':assisted,'map_sha256':m['sha256']} if hits else {'outcome':'FAIL','lamp':'LIT','state':'DARK_OR_UNKNOWN_DO_NOT_WALK','ask_next':'ENTERPRISE_LIGHTHOUSE_HUMAN_OR_MOTHER_REVIEW','map_sha256':m['sha256']}
|
||
class H(BaseHTTPRequestHandler):
|
||
def log_message(self,*a):pass
|
||
def sendj(self,n,v):b=json.dumps(v,ensure_ascii=False).encode();self.send_response(n);self.send_header('content-type','application/json');self.send_header('content-length',str(len(b)));self.end_headers();self.wfile.write(b)
|
||
def do_GET(self):
|
||
if self.path=='/health':
|
||
m=json.loads(MAP.read_text());return self.sendj(200,{'ok':True,'service':'guanghu-living-lighthouse','system':'TCS-0002∞-LakeEcho-0001','echo_kernel':'ECHO-KERNEL-0001','navigator':'SYS-GLW-LTH-0001','route_count':len(m['routes']),'map_sha256':m['sha256'],'model_provider':'OPENLUX_SMART' if API else None,'machine_may_self_ignite':False,'reality_authority':'NONE'})
|
||
self.sendj(404,{'error':'not_found'})
|
||
def do_POST(self):
|
||
try:self.sendj(200,ignite(read_json(self)) if self.path=='/v1/echo/ignite' else ask(read_json(self)) if self.path=='/v1/ask' else {'error':'not_found'})
|
||
except Exception as e:self.sendj(400,{'error':str(e)[:160]})
|
||
if __name__=='__main__':ThreadingHTTPServer((HOST,PORT),H).serve_forever()
|