62 lines
3.6 KiB
Python
62 lines
3.6 KiB
Python
"""Configuração privada do Gemini; respostas nunca incluem a chave."""
|
|
import json,os,re,threading,time
|
|
from pathlib import Path
|
|
import gemini
|
|
|
|
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
|
|
tmp=self.path.with_suffix('.tmp');fd=os.open(str(tmp),os.O_WRONLY|os.O_CREAT|os.O_TRUNC,0o600)
|
|
with os.fdopen(fd,'w',encoding='utf-8') as f:json.dump(saved,f)
|
|
os.chmod(tmp,0o600);tmp.replace(self.path);os.chmod(self.path,0o600)
|
|
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':
|
|
origin=h.headers.get('Origin')
|
|
if origin and origin not in ('http://'+h.headers.get('Host',''),'https://'+h.headers.get('Host','')):raise ValueError('Origem inválida')
|
|
size=int(h.headers.get('Content-Length',0))
|
|
if not 0<size<=2048 or h.headers.get('Content-Type','').split(';')[0]!='application/json':raise ValueError('Use JSON até 2048 bytes')
|
|
d=json.loads(h.rfile.read(size))
|
|
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
|