Desafio

import subprocess
import time
import os

print("Instalando Ollama...")
subprocess.run("curl -fsSL https://ollama.com/install.sh | sh", shell=True, check=True)
print("Iniciando servidor Ollama...")
subprocess.Popen("nohup ollama serve > ollama.log 2>&1 &", shell=True)
for _ in range(20):
try:
r = subprocess.run("curl -s http://localhost:11434/api/tags",
shell=True, capture_output=True, text=True, timeout=2)
if r.returncode == 0:
print("Servidor Ollama pronto!\n")
break
except:
pass
time.sleep(1)
print("Baixando modelo llama3.2:1b...")
subprocess.run("ollama pull llama3.2:1b", shell=True, check=True)
print("Modelo pronto!\n")
import requests
import json
import re
import ollama
import threading
from concurrent.futures import ThreadPoolExecutor
from typing import Optional
from tqdm.notebook import tqdm
url = "https://cdn3.gnarususercontent.com.br/4790-python/Resenhas_App_ChatGPT.txt"
response = requests.get(url, timeout=30)
response.encoding = "utf-8"
resenhas = [linha.strip() for linha in response.text.split("\n") if linha.strip()]
print(f"{len(resenhas)} resenhas carregadas\n")
def extrair_json(texto: str) -> Optional[dict]:
# 1) JSON puro
try:
return json.loads(texto)
except json.JSONDecodeError:
pass

# 2) ```json ... ```
match = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", texto, re.DOTALL)
if match:
    try:
        return json.loads(match.group(1))
    except json.JSONDecodeError:
        pass

# 3) Bloco {...} genérico
match = re.search(r"\{.*\}", texto, re.DOTALL)
if match:
    try:
        return json.loads(match.group(0))
    except json.JSONDecodeError as e:
        print(f"JSON inválido: {e}\n   Trecho: {texto[:200]}...")

return None

CAMPOS_OBRIGATORIOS = {"resenha_original", "sentimento"}
SENTIMENTOS_VALIDOS = {"positivo", "negativo", "neutro"}

def validar_json(dados: dict) -> bool:
if not isinstance(dados, dict):
return False
if not CAMPOS_OBRIGATORIOS.issubset(dados.keys()):
return False
if str(dados.get("sentimento", "")).lower() not in SENTIMENTOS_VALIDOS:
return False
if not isinstance(dados.get("resenha_original"), str):
return False
return True

def analisar_resenha(texto: str, tentativas: int = 3) -> Optional[dict]:
prompt = f"""Você é um sistema que retorna APENAS JSON válido.
NÃO escreva nada fora do JSON. Sem markdown, sem explicações.

Formato obrigatório:
{{
"usuario": "<apelido fictício coerente com a resenha>",
"resenha_original": "<a resenha fornecida, sem alterações>",
"sentimento": "positivo | negativo | neutro"
}}

Resenha:
"""{texto}"""
"""

for i in range(tentativas):
    try:
        resposta = ollama.chat(
            model="llama3.2:1b",
            messages=[{"role": "user", "content": prompt}],
            options={"temperature": 0.1},
        )
        conteudo = resposta["message"]["content"]
        dados = extrair_json(conteudo)

        if dados and validar_json(dados):
            # normaliza sentimento
            dados["sentimento"] = dados["sentimento"].lower()
            return dados

    except Exception as e:
        print(f"Tentativa {i+1} falhou: {e}")

return None

ARQUIVO_SAIDA = "resultados.json"
ARQUIVO_CHECKPOINT = "checkpoint.json"
TRAVA = threading.Lock()
if os.path.exists(ARQUIVO_CHECKPOINT):
with open(ARQUIVO_CHECKPOINT, "r", encoding="utf-8") as f:
resultados = json.load(f)
ja_feitos = {r["__indice"] for r in resultados if "__indice" in r}
print(f"♻ Retomando: {len(ja_feitos)} já processados\n")
else:
resultados = []
ja_feitos = set()

def processar(item):
i, texto = item
if i in ja_feitos:
return i, None

dados = analisar_resenha(texto)
if dados:
    dados["__indice"] = i
return i, dados

def salvar(dados):
with TRAVA:
resultados.append(dados)
with open(ARQUIVO_CHECKPOINT, "w", encoding="utf-8") as f:
json.dump(resultados, f, ensure_ascii=False, indent=2)