julioandClaude Opus 5.5
35d443a358
- comum.py: ler_json (Origin/JSON/tamanho), gravar_atomico e token (CARINHA_TOKEN ou seguranca.json), substituindo 6 copias nos modulos - eventos.py: hub SSE em /eventos com heartbeat e Last-Event-ID; modulos publicam expressao, telemetria, conversa, reacao, ia, motores - /estado agrega o snapshot inicial; /comando unifica gesto/rosto/cor/reacao - /parar-tudo desliga o PCA9685, para reacoes e trava motores ate /rearmar - braco: lease (ttl_ms) e interpolacao com limite de velocidade no servidor - braco e diagnostico nao disputam o PCA9685 (409 para o segundo dono) - fila limitada (20) com validade de 10 s; POST /face usa prefixo "rosto" - /app/config passa a checar Origin e JSON; versao vem de version.json - figurinhas automaticas do Gemini limitadas a 6 por hora - rosto entende a ordem "calma" enviada pelo E-Stop Token so e exigido quando configurado (docs/SEGURANCA.md). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
57 lines
3.0 KiB
Python
57 lines
3.0 KiB
Python
"""Configuração privada do Gemini; respostas nunca incluem a chave."""
|
|
import json,os,re,threading,time
|
|
from pathlib import Path
|
|
import gemini
|
|
from comum import gravar_atomico, ler_json
|
|
|
|
class ConfigAPI:
|
|
def __init__(self,path):
|
|
self.path=Path(path);self.lock=threading.RLock();self.ultimo_teste=None
|
|
try:
|
|
d=json.loads(self.path.read_text(encoding='utf-8'))
|
|
self.validar(d)
|
|
if d.get('chave'):os.environ['GEMINI_API_KEY']=d['chave']
|
|
gemini.MODELO=d['modelo']
|
|
except (OSError,ValueError,KeyError,TypeError):pass
|
|
@staticmethod
|
|
def validar(d):
|
|
if not isinstance(d,dict) or set(d)-{'modelo','chave'}:raise ValueError('Campos de API inválidos')
|
|
if not isinstance(d.get('modelo'),str) or not re.fullmatch(r'[a-zA-Z0-9._-]{3,100}',d['modelo']):raise ValueError('Nome do modelo inválido')
|
|
if 'chave' in d and (not isinstance(d['chave'],str) or not 20<=len(d['chave'])<=250 or not re.fullmatch(r'[A-Za-z0-9._-]+',d['chave'])):raise ValueError('Chave inválida')
|
|
def status(self):
|
|
with self.lock:return dict(provedor='Gemini',modelo=gemini.MODELO,configurada=bool(os.environ.get('GEMINI_API_KEY')),ultimo_teste=self.ultimo_teste)
|
|
def salvar(self,d):
|
|
self.validar(d)
|
|
with self.lock:
|
|
key=d.get('chave') or os.environ.get('GEMINI_API_KEY','')
|
|
saved=dict(modelo=d['modelo'])
|
|
if key:saved['chave']=key
|
|
gravar_atomico(self.path,saved,privado=True)
|
|
gemini.MODELO=d['modelo']
|
|
if key:os.environ['GEMINI_API_KEY']=key
|
|
self.ultimo_teste=None
|
|
return self.status()
|
|
def testar(self):
|
|
start=time.monotonic()
|
|
try:
|
|
gemini.gerar_json(dict(contents=[dict(role='user',parts=[dict(text='Responda JSON {"ok":true}. Teste de conexão do Carinha.')])],generationConfig=dict(maxOutputTokens=64,responseMimeType='application/json')))
|
|
result=dict(ok=True,segundos=round(time.monotonic()-start,2),mensagem='Gemini respondeu ao teste.')
|
|
except Exception as e:
|
|
result=dict(ok=False,mensagem=str(e) if isinstance(e,ValueError) else 'Falha de conexão com o Gemini. Confira a rede e tente novamente.')
|
|
with self.lock:self.ultimo_teste=result
|
|
return result
|
|
def rotas(self,h,method):
|
|
route=h.path.split('?',1)[0]
|
|
if not route.startswith('/api/configuracao/'):return False
|
|
try:
|
|
if method=='GET' and route.endswith('/status'):result=self.status()
|
|
elif method=='POST':
|
|
d=ler_json(h,2048)
|
|
if route.endswith('/salvar'):result=self.salvar(d)
|
|
elif route.endswith('/testar'):result=self.testar()
|
|
else:raise ValueError('Rota inválida')
|
|
else:raise ValueError('Método inválido')
|
|
h.enviar(json.dumps(result,ensure_ascii=False))
|
|
except (ValueError,TypeError,OSError):h.enviar(json.dumps(dict(erro='Não foi possível processar a configuração. Confira o modelo e a chave.')),codigo=400)
|
|
return True
|