import requests import random import os import hashlib import mysql.connector from PIL import Image from io import BytesIO from deep_translator import GoogleTranslator import cairosvg import time import re # Adicionado para validação de tokens # Configuração do banco de dados DB_CONFIG = { "host": "localhost", "user": "victor", "password": "toninho13", "database": "qualifica_brasil" } # Configuração das APIs de imagens API_KEYS = { "pexels": ["s0n5sD62LD1ET94Wxt2uyySPoaQfnmUeaMUuU1YY33sg1Snvi2evRnbV"], "unsplash": ["SEU_TOKEN_UNSPLASH"], "pixabay": ["SEU_TOKEN_PIXABAY"], "openverse": [] # Openverse não exige token } API_URLS = { "pexels": "https://api.pexels.com/v1/search", "unsplash": "https://api.unsplash.com/search/photos", "pixabay": "https://pixabay.com/api/", "openverse": "https://api.openverse.engineering/v1/images/" } # pexels s0n5sD62LD1ET94Wxt2uyySPoaQfnmUeaMUuU1YY33sg1Snvi2evRnbV # # Unsplash Access Key - 5VERLgULUrKSOqQY2YnJ88Ineo8RnBvY54Yr_SznR5A # Unsplash Secret Key - IFD_LptdbaTpJE4TGGLS-k8-guy3wIc4pYTTFXXDgp4 os.makedirs("imagens_cursos", exist_ok=True) def log(msg): """Imprime mensagens de log com timestamp""" print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}") def traduzir_texto(texto, destino="en"): """Traduz um texto para o inglês.""" log(f"Traduzindo: {texto} -> {destino}") try: traduzido = GoogleTranslator(source="auto", target=destino).translate(texto) log(f"Tradução concluída: {traduzido}") return traduzido except Exception as e: log(f"⚠️ Erro ao traduzir '{texto}': {e}") return texto def escolher_token(api): """Escolhe um token aleatório para a API.""" return random.choice(API_KEYS[api]) if API_KEYS[api] else None #def gerar_hash(nome): # """Gera um hash SHA1 para nomes de arquivos únicos.""" # hash_obj = hashlib.sha1(nome.encode()) # return hash_obj.hexdigest()[:10] def gerar_hash(nome): if nome is None or not isinstance(nome, str) or nome.strip() == "": mensagem = f"❌ Erro: token inválido ao gerar hash. Valor recebido: {repr(nome)}" print(mensagem) # ou use log(mensagem) se estiver com sistema de log raise ValueError(mensagem) hash_obj = hashlib.sha1(nome.encode()) return hash_obj.hexdigest() def validar_tokens(tokens): """Remove tokens inválidos (números, telefones ou caracteres especiais isolados).""" if not tokens or re.fullmatch(r"[\s\d\-\(\)\.]+", tokens): return False return True def buscar_imagem(tokens, id_token): """Busca imagens nas APIs e retorna a melhor opção.""" # tempo = random.uniform(10,30) # intervalo de 20 segundos em média, para evitar bloqueio, considerando que PEXELS aceita no máximo 200 requisições por hora, segundo o GPT tempo = random.normalvariate(20, 5) # média 20s, desvio padrão 5s print(f"Aguardando {tempo:.2f} segundos...") time.sleep(tempo) # Evita bloqueio por excesso de requisições if not tokens: log("⚠️ Nenhum token de busca disponível.") return None melhores_tokens = " ".join(tokens.split("|")[:5]) if not validar_tokens(melhores_tokens): log(f"⚠️ Tokens inválidos ignorados: {melhores_tokens}") return None tokens_traduzidos = traduzir_texto(melhores_tokens, "en") log(f"📌 Tokens originais: {melhores_tokens}") log(f"📌 Tokens traduzidos: {tokens_traduzidos}") for api in ["pexels", "unsplash", "pixabay", "openverse"]: token = escolher_token(api) params = {"query": tokens_traduzidos, "per_page": 3} headers = {"Authorization": token} if token else {} if api == "pixabay": params["key"] = token log(f"🔍 Buscando imagens na API: {api}") try: response = requests.get(API_URLS[api], headers=headers, params=params, timeout=5) if response.status_code == 401: log(f"⚠️ Erro de autenticação na API {api}. Verifique o token.") continue if response.status_code != 200: log(f"⚠️ Erro ao buscar na API {api}: {response.status_code}") continue data = response.json() img_urls = [] if api == "pexels": img_urls = [photo["src"]["large"] for photo in data.get("photos", [])] elif api == "unsplash": img_urls = [result["urls"]["regular"] for result in data.get("results", [])] elif api == "pixabay": img_urls = [hit["largeImageURL"] for hit in data.get("hits", [])] elif api == "openverse": img_urls = [result["url"] for result in data.get("results", [])] if not img_urls: log(f"⚠️ Nenhuma imagem encontrada na API {api}.") continue for img_url in img_urls: if "wikimedia.org" not in img_url: return salvar_imagem(img_url, tokens_traduzidos, melhores_tokens) log("⚠️ Todas as imagens eram do Wikimedia. Tentando outra API...") except requests.exceptions.RequestException as e: log(f"⚠️ Erro ao conectar à API {api}: {e}") log("⚠️ Nenhuma imagem válida encontrada.") return None def salvar_imagem(img_url, token, token_vernaculo): """Baixa e salva a imagem no diretório.""" log(f"🔹 Baixando imagem de: {img_url}") response = requests.get(img_url, timeout=5) content_type = response.headers.get("Content-Type", "") hash_nome = gerar_hash(token) if "svg" in content_type: log(f"🔄 Convertendo SVG para PNG: {img_url}") nome_arquivo = f"imagens_tokens/{token.replace(' ', '_')}_{token_vernaculo.replace(' ','_')}_{hash_nome}.png" cairosvg.svg2png(bytestring=response.content, write_to=nome_arquivo) return nome_arquivo, content_type nome_arquivo = f"imagens_tokens/{token.replace(' ', '_')}_{token_vernaculo.replace(' ','_')}_{hash_nome}.jpg" with open(nome_arquivo, "wb") as file: file.write(response.content) log(f"✅ Imagem salva como: {nome_arquivo}") return nome_arquivo, content_type def obter_tokens(): """Executa a query no MySQL para obter os tokens de busca e IDs dos títulos.""" log("Conectando ao banco de dados...") conn = mysql.connector.connect(**DB_CONFIG) cursor = conn.cursor(dictionary=True) log("Executando consulta SQL...") cursor.execute(""" select group_concat(id_ocupacao) , group_concat((SELECT nome_ocupacao FROM ocupacoes WHERE id_chave_ocupacao = id_ocupacao)), nome_token, acentuada tokens_de_busca, id_tipo_token, id_tipo_token_spaCy, id_token, count(id_ocupacao) conta, "ocupacoes" from tokens_ocupacoes left join tokens on id_token = id_chave_token where id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "sigla") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "pontuacao") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "preposicao") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "numeral") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "codigo") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "indeterminado") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "nao_atribuido") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "hora") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "prefixo") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "preposicao") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "artigo") AND id_token not in (select id_token from tokens_imagens) # para nao pegar de novo no repositorio imagens que ja foram baixadas - importante se o processo teve que ser interrompido group by id_token, id_tipo_token_spaCy having conta < 2 UNION select group_concat(id_titulo) , group_concat((SELECT titulo FROM frases_opt WHERE id_titulo = fr.id_titulo)), nome_token, acentuada tokens_de_busca, id_tipo_token, id_tipo_token_spaCy, id_token, count(id_titulo) conta, "titulos_do_curso" from frases as fr left join tokens on id_token = id_chave_token where id_titulo in (select id_titulo from campos_cursos where id_tipo_classificacao = (select id_chave_tipo_classificacao from tipos_classificacoes where nome_tipo_classificacao = "titulo_do_curso")) and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "sigla") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "pontuacao") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "preposicao") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "numeral") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "codigo") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "indeterminado") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "nao_atribuido") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "hora") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "prefixo") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "preposicao") and id_tipo_token_spaCy <> (select id_chave_tipo_token from tipos_tokens where nome_tipo_token = "artigo") AND id_token not in (select id_token from tokens_imagens) # para nao pegar de novo no repositorio imagens que ja foram baixadas - importante se o processo teve que ser interrompido group by id_token, id_tipo_token_spaCy having conta < 2 order by conta; """) resultados = cursor.fetchall() conn.close() log(f"Consulta finalizada. {len(resultados)} registros encontrados.") tokens_list = [] for row in resultados: tokens_de_busca = row["tokens_de_busca"] if row["tokens_de_busca"] is not None else "" id_token = row["id_token"] if row["id_token"] else [] id_tipo_token_spaCy = row["id_tipo_token_spaCy"] if row["id_tipo_token_spaCy"] else "" tokens_list.append({"tokens_de_busca": tokens_de_busca, "id_token": id_token, "id_tipo_token_spaCy": id_tipo_token_spaCy}) return tokens_list def gerar_inserts(): """Obtém tokens do banco e busca imagens.""" log("🔎 Obtendo tokens do banco de dados...") dados = obter_tokens() with open("insert_tokens_imagens.sql", "w", encoding="utf-8") as f: for dado in dados: tokens_de_busca = dado["tokens_de_busca"] id_tokens = dado["id_token"] # lista id_tipo_token_spaCy = dado["id_tipo_token_spaCy"] # Crasha se houver mais de um id_token #if len(id_tokens) != 1: # raise ValueError(f"Erro: esperado apenas 1 id_token, mas encontrei {len(id_tokens)}: {id_tokens}") id_token = id_tokens nome_arquivo, content_type = buscar_imagem(tokens_de_busca, id_token) f.write(f"INSERT IGNORE INTO imagens (nome_imagem, id_tipo_arquivo) VALUES ('{nome_arquivo}', (SELECT id_chave_tipo_arquivo FROM tipos_arquivos WHERE descricao = \"{content_type}\"));\n") f.write(f"INSERT IGNORE INTO tokens_imagens (nome_token_imagem, id_token, id_imagem) VALUES (CONCAT(\"{tokens_de_busca}\",\"#\",\"{nome_arquivo}\"),(SELECT id_chave_token FROM tokens WHERE acentuada = \"{tokens_de_busca}\" AND id_tipo_token_spaCy=\"{id_tipo_token_spaCy}\"), (SELECT id_chave_imagem FROM imagens WHERE nome_imagem = \"{nome_arquivo}\"));\n") f.flush() # 👈 força a gravação imediata no disco #def gerar_inserts(): # """Obtém tokens do banco e busca imagens.""" # log("🔎 Obtendo tokens do banco de dados...") # dados = obter_tokens() # # for dado in dados: # tokens_de_busca = dado["tokens_de_busca"] # id_token = dado["id_token"] # nome_arquivo = buscar_imagem(tokens_de_busca, id_token) # with open("insert_tokens_imagens.sql", "w", encoding="utf-8") as f: # f.write("INSERT INTO imagens (nome_imagem, id_tipo_arquivo) VALUES ('" + nome_arquivo + "', 1);\nINSERT INTO tokens_imagens (id_token, id_imagem) VALUES (" + str(id_token) + ", (SELECT id_chave_imagem FROM imagens WHERE nome_imagem = \""+ nome_arquivo +"\" ));\n") # Executar busca gerar_inserts()