#!/usr/bin/env python3 """Servidor central de desarrollo para Grano POS. Usa solo la biblioteca estándar.""" import errno, json, os, sqlite3, subprocess, tempfile, threading, time from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler from pathlib import Path from urllib.parse import urlparse, parse_qs from urllib.request import Request, urlopen ROOT=Path(__file__).resolve().parent DATA=ROOT/'pos-central.sqlite3' LOCK=threading.Lock() def connect(): db=sqlite3.connect(DATA) db.execute('PRAGMA journal_mode=WAL') db.execute('''CREATE TABLE IF NOT EXISTS events( seq INTEGER PRIMARY KEY AUTOINCREMENT,event_id TEXT UNIQUE NOT NULL, device_id TEXT NOT NULL,created_at TEXT NOT NULL,action TEXT NOT NULL, actor TEXT, payload TEXT NOT NULL)''') db.commit(); return db class Handler(SimpleHTTPRequestHandler): def __init__(self,*args,**kwargs): super().__init__(*args,directory=str(ROOT),**kwargs) def json(self,status,data): body=json.dumps(data,ensure_ascii=False).encode();self.send_response(status) self.send_header('Content-Type','application/json; charset=utf-8');self.send_header('Content-Length',str(len(body))) self.send_header('Cache-Control','no-store');self.end_headers();self.wfile.write(body) def do_GET(self): u=urlparse(self.path) if u.path=='/api/health': return self.json(200,{'ok':True,'time':time.time()}) if u.path=='/api/printers': printers=[] try: result=subprocess.run(['lpstat','-p'],capture_output=True,text=True,timeout=5) for line in result.stdout.splitlines(): if line.startswith('printer '): name=line.split()[1];printers.append({'id':name,'name':name,'status':'idle' if ' is idle' in line else 'available'}) except (FileNotFoundError,subprocess.SubprocessError): pass return self.json(200,{'printers':printers,'default':os.environ.get('POS_PRINTER','')}) if u.path=='/api/terminals': integrated=bool(os.environ.get('POS_PAYMENT_URL','').strip()) return self.json(200,{'terminals':[{'id':'mercado_pago','name':'Mercado Pago Point','manual':True,'integrated':integrated},{'id':'clip','name':'Clip','manual':True,'integrated':integrated},{'id':'bbva','name':'BBVA','manual':True,'integrated':integrated},{'id':'getnet','name':'Getnet','manual':True,'integrated':integrated},{'id':'other','name':'Otra terminal','manual':True,'integrated':integrated}]}) if u.path=='/api/sync': q=parse_qs(u.query);after=max(0,int(q.get('after',['0'])[0]));device=q.get('device',[''])[0] with LOCK: db=connect();rows=db.execute('SELECT seq,event_id,device_id,created_at,action,actor,payload FROM events WHERE seq>? ORDER BY seq LIMIT 500',(after,)).fetchall();db.close() events=[{'seq':r[0],'id':r[1],'device_id':r[2],'time':r[3],'action':r[4],'actor':r[5],'payload':json.loads(r[6])} for r in rows] return self.json(200,{'events':events,'cursor':events[-1]['seq'] if events else after,'device':device}) return super().do_GET() def do_POST(self): path=urlparse(self.path).path if path in ('/api/payment','/api/print','/api/cfdi'): try: length=int(self.headers.get('Content-Length','0')) if length>1_000_000: return self.json(413,{'error':'too_large'}) data=json.loads(self.rfile.read(length) or b'{}') if path=='/api/payment': return self.handle_payment(data) if path=='/api/print': return self.handle_print(data) return self.handle_cfdi(data) except Exception as exc: return self.json(400,{'error':'hardware_request_failed','detail':str(exc)}) if path!='/api/sync': return self.json(404,{'error':'not_found'}) try: length=int(self.headers.get('Content-Length','0')) if length>8_000_000: return self.json(413,{'error':'too_large'}) data=json.loads(self.rfile.read(length));device=str(data['deviceId']);events=data.get('events',[]) with LOCK: db=connect() for e in events: db.execute('INSERT OR IGNORE INTO events(event_id,device_id,created_at,action,actor,payload) VALUES(?,?,?,?,?,?)',(str(e['id']),device,str(e.get('time','')),str(e.get('action','')),str(e.get('user','')),json.dumps(e,ensure_ascii=False))) db.commit();cursor=db.execute('SELECT COALESCE(MAX(seq),0) FROM events').fetchone()[0];db.close() return self.json(200,{'accepted':len(events),'cursor':cursor}) except Exception as exc: return self.json(400,{'error':'invalid_request','detail':str(exc)}) def handle_payment(self,data): amount=round(float(data.get('amount',0)),2) if amount<=0: return self.json(400,{'error':'invalid_amount','detail':'El importe debe ser mayor que cero'}) adapter=os.environ.get('POS_PAYMENT_URL','').strip() if not adapter: return self.json(200,{'status':'approved','reference':f'SIM-{int(time.time())}','simulated':True,'amount':amount}) payload=json.dumps(data,ensure_ascii=False).encode() req=Request(adapter,data=payload,headers={'Content-Type':'application/json'},method='POST') with urlopen(req,timeout=45) as response: result=json.loads(response.read() or b'{}') return self.json(200,result) def handle_print(self,data): content=str(data.get('content','')) if not content: return self.json(400,{'error':'empty_document','detail':'No hay contenido para imprimir'}) printer=str(data.get('printer') or os.environ.get('POS_PRINTER','')).strip() if not printer: return self.json(200,{'status':'queued','simulated':True}) copies=max(1,min(5,int(data.get('copies',1)))) with tempfile.NamedTemporaryFile('w',encoding='utf-8',suffix='.txt',delete=False) as temp: temp.write(content);temp_path=temp.name try: result=subprocess.run(['lp','-d',printer,'-n',str(copies),temp_path],capture_output=True,text=True,timeout=20) if result.returncode: return self.json(502,{'error':'print_failed','detail':result.stderr.strip() or 'La cola de impresión rechazó el trabajo'}) return self.json(200,{'status':'queued','job':result.stdout.strip(),'simulated':False}) finally: Path(temp_path).unlink(missing_ok=True) def handle_cfdi(self,data): adapter=os.environ.get('POS_PAC_URL','').strip() if not adapter: return self.json(503,{'error':'pac_not_configured','detail':'Configura POS_PAC_URL con el servicio seguro de tu PAC'}) payload=json.dumps(data,ensure_ascii=False).encode() req=Request(adapter,data=payload,headers={'Content-Type':'application/json'},method='POST') with urlopen(req,timeout=60) as response: result=json.loads(response.read() or b'{}') return self.json(200,result) if __name__=='__main__': host=os.environ.get('POS_HOST','0.0.0.0');port=int(os.environ.get('POS_PORT','8080')) try: server=ThreadingHTTPServer((host,port),Handler) except OSError as exc: if exc.errno==errno.EADDRINUSE: print(f'El puerto {port} ya está ocupado por otra instancia de Grano POS.') print('Cierra el servidor anterior con Control+C y vuelve a ejecutar este comando.') print(f'Si deseas conservar ambos, usa: POS_PORT={port+1} python3 server.py') raise SystemExit(2) raise print(f'Grano POS disponible en http://localhost:{port}/punto-de-venta-cafeteria.html') server.serve_forever()