import json
import re
import time
from openai import OpenAI
from google.api_core.exceptions import TooManyRequests
# Configuração do cliente local
client = OpenAI(
base_url="http://127.0.0.1:1234/v1",
api_key="sk-proj-1234567890"
)
# ==========================================================
# 1. Carregar arquivo TXT em uma lista Python
# ==========================================================
linhas_resenhas = []
with open("./Resenhas.txt", "r", encoding="utf-8") as arquivo:
for linha in arquivo:
linha_limpa = linha.strip()
if linha_limpa:
linhas_resenhas.append(linha_limpa)
print("Resenhas carregadas:")
for linha in linhas_resenhas:
print(linha)
# ==========================================================
# Converter as linhas para uma lista de dicionários
# ==========================================================
dados_resenhas = []
for resenha_str in linhas_resenhas:
try:
partes = resenha_str.split("$", 2)
if len(partes) == 3:
reviewer_id = partes[0]
reviewer_name = partes[1]
review_text = partes[2]
dados_resenhas.append({
"reviewerID": reviewer_id,
"reviewerName": reviewer_name,
"reviewText": review_text
})
else:
print(f"Linha ignorada: {resenha_str}")
except Exception as erro:
print(f"Erro ao processar linha: {erro}")
# ==========================================================
# 2. Enviar ao modelo local
# ==========================================================
resenhas_json = json.dumps(
dados_resenhas,
ensure_ascii=False
)
prompt = f"""
Analise as resenhas abaixo.
Para cada resenha retorne um objeto JSON contendo:
- usuario
- resenha_original
- resenha_pt
- avaliacao
A avaliação deve ser em português:
- positiva
- negativa
- neutra
Retorne SOMENTE uma lista JSON válida.
Resenhas:
{resenhas_json}
"""
max_retries = 5
initial_delay = 2
lista_resenhas_processadas = None
for tentativa in range(max_retries):
try:
resposta = client.chat.completions.create(
model="google/gemma-3-1b",
messages=[
{
"role": "user",
"content": prompt
}
],
temperature=0.3,
max_completion_tokens=4000,
stream=False
)
conteudo = resposta.choices[0].message.content.strip()
# Tenta converter diretamente
try:
lista_resenhas_processadas = json.loads(conteudo)
except json.JSONDecodeError:
# Tenta extrair JSON caso venha misturado com texto
json_match = re.search(
r'\[\s*\{.*\}\s*\]',
conteudo,
re.DOTALL
)
if json_match:
lista_resenhas_processadas = json.loads(
json_match.group(0)
)
if lista_resenhas_processadas:
break
except TooManyRequests:
tempo_espera = initial_delay * (2 ** tentativa)
print(
f"Limite excedido. "
f"Tentando novamente em {tempo_espera}s..."
)
time.sleep(tempo_espera)
except Exception as erro:
print(f"Erro inesperado: {erro}")
break
# ==========================================================
# 3. Transformar resposta em lista de dicionários
# ==========================================================
if lista_resenhas_processadas:
print("\nLista de dicionários criada com sucesso:\n")
print(
json.dumps(
lista_resenhas_processadas,
indent=2,
ensure_ascii=False
)
)
else:
print("Não foi possível obter um JSON válido.")
Continua no coment
