João Henrique
b541f502ba
- Adicionado estrutura completa do projeto - Configurado MCP server para Premiere Pro - Adicionado documentação e skills - Configurado Gitignore para o projeto
67 lines
3.1 KiB
Python
67 lines
3.1 KiB
Python
import json
|
|
import mimetypes
|
|
import ssl
|
|
import uuid
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
from urllib import request
|
|
|
|
from ...scanner.modelos import SegmentoDeTranscricao
|
|
|
|
|
|
class ProviderDeTranscricaoGroq:
|
|
"""Provider Groq compatível com o contrato de transcrição do scanner."""
|
|
|
|
endpoint = "https://api.groq.com/openai/v1/audio/transcriptions"
|
|
|
|
def __init__(self, api_key: str, modelo: str = "whisper-large-v3-turbo",
|
|
idioma: str = "pt", tempo_limite: float = 120.0,
|
|
requisicao: Callable[..., Any] | None = None) -> None:
|
|
if not api_key.strip():
|
|
raise ValueError("A chave da API Groq é obrigatória.")
|
|
self.api_key = api_key.strip()
|
|
self.modelo = modelo
|
|
self.idioma = idioma
|
|
self.tempo_limite = tempo_limite
|
|
self._requisicao = requisicao or request.urlopen
|
|
try:
|
|
import certifi
|
|
self._contexto_ssl = ssl.create_default_context(cafile=certifi.where())
|
|
except ImportError:
|
|
self._contexto_ssl = ssl.create_default_context()
|
|
|
|
def transcrever(self, clipe: Any) -> list[SegmentoDeTranscricao]:
|
|
arquivo = getattr(clipe, "arquivo", None)
|
|
if not arquivo:
|
|
raise ValueError("O clipe não possui arquivo de origem para transcrição.")
|
|
caminho = Path(arquivo)
|
|
if not caminho.is_file():
|
|
raise FileNotFoundError(f"Arquivo do clipe não encontrado: {caminho}")
|
|
corpo, tipo = self._multipart(caminho)
|
|
req = request.Request(self.endpoint, data=corpo, method="POST", headers={
|
|
"Authorization": f"Bearer {self.api_key}",
|
|
"Content-Type": tipo,
|
|
})
|
|
argumentos = {"timeout": self.tempo_limite}
|
|
if self._requisicao is request.urlopen:
|
|
argumentos["context"] = self._contexto_ssl
|
|
with self._requisicao(req, **argumentos) as resposta:
|
|
dados = json.loads(resposta.read().decode("utf-8"))
|
|
return [SegmentoDeTranscricao(float(item["start"]), float(item["end"]),
|
|
str(item.get("text", "")).strip(),
|
|
None)
|
|
for item in dados.get("segments", [])]
|
|
|
|
def _multipart(self, caminho: Path) -> tuple[bytes, str]:
|
|
limite = "----engine-groq-" + uuid.uuid4().hex
|
|
mime = mimetypes.guess_type(caminho.name)[0] or "application/octet-stream"
|
|
partes: list[bytes] = []
|
|
campos = {"model": self.modelo, "language": self.idioma,
|
|
"response_format": "verbose_json", "timestamp_granularities[]": "segment"}
|
|
for nome, valor in campos.items():
|
|
partes.append(f"--{limite}\r\nContent-Disposition: form-data; name=\"{nome}\"\r\n\r\n{valor}\r\n".encode())
|
|
partes.append(f"--{limite}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{caminho.name}\"\r\nContent-Type: {mime}\r\n\r\n".encode())
|
|
partes.append(caminho.read_bytes())
|
|
partes.append(f"\r\n--{limite}--\r\n".encode())
|
|
return b"".join(partes), f"multipart/form-data; boundary={limite}"
|