""" Montagem do pós-processamento a partir dos arquivos persistidos no lote. """ from __future__ import annotations import csv from datetime import datetime, timezone from functools import lru_cache from typing import Any import math import re import shutil import time import unicodedata from pathlib import Path import cv2 import numpy as np from app.avaliacao_contexto import obter_simulado_por_id, resolver_meta_avaliacao_lote from app.gerador_csv import TOTAL_PERGUNTAS, resposta from app.keepedu_importacao import caminho_relatorio_importacao from app.services import ( caminho_leituras, caminho_log, caminho_originais, caminho_pasta_processamento, carregar_json, carregar_leituras_lote, gerar_linhas_log_virtuais, listar_imagens_debug, localizar_imagem_original, listar_processamentos_recentes, normalizar_nome_arquivo_debug, pontos_cantos_para_dict, resumo_processamento, ) from app.validacao_cadastral import ( carregar_envios_keepedu, carregar_validacao_cadastral_salva, carregar_validacao_manual, extrair_codigo_barras_possivel, quebrar_codigo_barras, obter_validacao_cadastral_em_cache, obter_validacao_cadastral_atual, resolver_envio_keepedu, rotulo_envio_keepedu, resumo_status_lote_keepedu, detectar_id_prova_do_processamento, ) from app.lotes_store import cache_lotes_durante_operacao from app.time_utils import agora_local, obter_timezone_app def _stem_imagem(nome: str) -> str: texto = normalizar_nome_arquivo_debug(str(nome or "").strip()) return Path(texto).stem.lower() def lote_possui_arquivos_historico(nome_processamento: str) -> bool: from app.lotes_store import lote_existe if lote_existe(nome_processamento): return True pasta = caminho_pasta_processamento(nome_processamento) if not pasta.exists(): return False return caminho_leituras(nome_processamento).exists() or caminho_log(nome_processamento).exists() def lote_esta_arquivado(nome_processamento: str) -> bool: """True quando imagens/debug foram removidos após envio à KeepEdu.""" if not lote_possui_arquivos_historico(nome_processamento): return False return len(listar_imagens_debug(nome_processamento)) == 0 def obter_identificacao_prova_lote(nome_processamento: str) -> dict[str, str]: meta = resolver_meta_avaliacao_lote(nome_processamento=nome_processamento) id_av = str(meta.get("id_avaliacao") or meta.get("id") or "").strip() nome_prova = str(meta.get("tipo_prova") or meta.get("nome_simulado") or "").strip() leituras = carregar_leituras_lote(nome_processamento) # The history index must not trigger a full remote validation for each lot. dados_val = obter_validacao_cadastral_em_cache(nome_processamento) or {} resumo_val = dados_val.get("resumo") or {} if not id_av: id_av = str( resumo_val.get("id_prova_processamento") or detectar_id_prova_do_processamento(leituras, nome_processamento) or "" ).strip() if not nome_prova and id_av: item = obter_simulado_por_id(id_av) if item: nome_prova = str(item.get("tipo_prova") or item.get("nome_simulado") or "").strip() if id_av and nome_prova: titulo = f"(ID {id_av}) · {nome_prova}" elif id_av: titulo = f"(ID {id_av})" elif nome_prova: titulo = nome_prova else: titulo = nome_processamento.replace("processamento_", "").replace("_", " ") return { "id_avaliacao": id_av, "nome_prova": nome_prova, "titulo": titulo, } def _slugificar_texto_historico(texto: str) -> str: normalizado = unicodedata.normalize("NFKD", str(texto or "")) sem_acento = "".join(ch for ch in normalizado if not unicodedata.combining(ch)) minusculo = sem_acento.lower() slug = re.sub(r"[^a-z0-9]+", "-", minusculo) return re.sub(r"-+", "-", slug).strip("-") def gerar_slug_avaliacao_historico(id_avaliacao: str = "", nome_prova: str = "") -> str: id_av = str(id_avaliacao or "").strip() nome = str(nome_prova or "").strip() if id_av and nome: return f"{id_av}-{_slugificar_texto_historico(nome)}" if id_av: return id_av if nome: return _slugificar_texto_historico(nome) return "" def gerar_slug_historico_lote(nome_processamento: str) -> str: identificacao = obter_identificacao_prova_lote(nome_processamento) slug = gerar_slug_avaliacao_historico( str(identificacao.get("id_avaliacao") or ""), str(identificacao.get("nome_prova") or ""), ) return slug or str(nome_processamento or "").strip() def resolver_nome_lote_historico(identificador: str) -> str | None: texto = str(identificador or "").strip() if not texto: return None if texto.startswith("processamento_") and lote_possui_arquivos_historico(texto): return texto candidatos: list[str] = [] for meta in listar_processamentos_recentes(): nome = meta["nome"] if not lote_possui_arquivos_historico(nome): continue if gerar_slug_historico_lote(nome) == texto: candidatos.append(nome) if not candidatos: return None if len(candidatos) == 1: return candidatos[0] candidatos.sort( key=lambda nome: _timestamp_nome_lote_historico(nome), reverse=True, ) return candidatos[0] def _carregar_linhas_log(nome_processamento: str) -> dict[str, dict]: caminho = caminho_log(nome_processamento) linhas: list[dict] = [] if caminho.exists(): with open(caminho, "r", encoding="utf-8-sig", newline="") as arquivo: leitor = csv.DictReader(arquivo, delimiter=";") linhas = list(leitor) else: linhas = gerar_linhas_log_virtuais(nome_processamento) indice: dict[str, dict] = {} for linha in linhas: arquivo_nome = str(linha.get("arquivo") or "").strip() if arquivo_nome: indice[_stem_imagem(arquivo_nome)] = linha return indice def _carregar_importacao_por_imagem(nome_processamento: str) -> dict[str, dict]: caminho = caminho_relatorio_importacao(nome_processamento, modo="importacao") if not caminho.exists(): return {} relatorio = carregar_json(caminho, {}) indice: dict[str, dict] = {} for item in relatorio.get("detalhes") or []: if not isinstance(item, dict): continue chave = str(item.get("imagem") or "").strip() if chave: indice[_stem_imagem(chave)] = item nome_arquivo = str(item.get("nomeArquivo") or "").strip() if nome_arquivo: indice.setdefault(_stem_imagem(nome_arquivo), item) return indice def _extrair_aluno_id_para_keepedu( validacao: dict, manual: dict | None, dados_cartao: dict, log_linha: dict, ) -> str: candidatos: list[Any] = [] if manual: candidatos.append(manual.get("id_final")) candidatos.append(validacao.get("id_final")) for valor in candidatos: texto = str(valor or "").strip() if texto and texto != "—" and texto.isdigit(): return texto for origem in (dados_cartao, log_linha): codigo = str(origem.get("codigo_barras") or "").strip() if not codigo: continue _, id_aluno = quebrar_codigo_barras(codigo) id_texto = str(id_aluno or "").strip() if id_texto.isdigit(): return id_texto return "" def _resumo_respostas_marcadas(respostas: dict) -> dict: if not isinstance(respostas, dict): respostas = {} itens: list[dict] = [] for numero in range(9, TOTAL_PERGUNTAS + 1): valor = resposta(respostas, numero) if valor: itens.append({"questao": numero, "marcacao": valor}) preview = "; ".join(f"Q{item['questao']}:{item['marcacao']}" for item in itens[:12]) if len(itens) > 12: preview = f"{preview}; …" return { "total_marcadas": len(itens), "itens": itens, "preview": preview or "—", } def montar_registros_pos_processamento( nome_processamento: str, *, leituras: dict | None = None, dados_validacao: dict | None = None, pagina: int = 1, por_pagina: int = 0, ) -> list[dict]: leituras = leituras if isinstance(leituras, dict) else carregar_leituras_lote(nome_processamento) if not leituras: return [] dados_validacao = dados_validacao if isinstance(dados_validacao, dict) else (obter_validacao_cadastral_atual(nome_processamento) or {}) validacoes = dados_validacao.get("validacoes") or {} validacao_manual = carregar_validacao_manual(nome_processamento) linhas_log = _carregar_linhas_log(nome_processamento) importacao = _carregar_importacao_por_imagem(nome_processamento) envios_keepedu = carregar_envios_keepedu(nome_processamento) leituras_ordenadas = sorted( ( (chave, dados_cartao) for chave, dados_cartao in leituras.items() if isinstance(dados_cartao, dict) ), key=lambda item: str( item[1].get("arquivo_original") or normalizar_nome_arquivo_debug(item[0]) ).lower(), ) if por_pagina > 0: pagina = max(1, int(pagina or 1)) inicio = (pagina - 1) * por_pagina leituras_ordenadas = leituras_ordenadas[inicio : inicio + por_pagina] registros: list[dict] = [] for chave, dados_cartao in leituras_ordenadas: arquivo_original = str(dados_cartao.get("arquivo_original") or normalizar_nome_arquivo_debug(chave)).strip() stem = _stem_imagem(arquivo_original or chave) validacao = validacoes.get(chave) or validacoes.get(arquivo_original) or {} manual = validacao_manual.get(chave) or validacao_manual.get(arquivo_original) log_linha = linhas_log.get(stem, {}) importacao_item = importacao.get(stem, {}) envio_keepedu = resolver_envio_keepedu(chave, arquivo_original, envios_keepedu) rotulo_envio = rotulo_envio_keepedu(envio_keepedu) if not rotulo_envio and importacao_item.get("sucesso"): rotulo_envio = "Vetores no KeepEdu" respostas_info = _resumo_respostas_marcadas(dados_cartao.get("respostas") or {}) erros = dados_cartao.get("erros") or [] origem_cantos = str(dados_cartao.get("origem_cantos") or "").strip() registros.append( { "chave": chave, "arquivo": arquivo_original, "status_log": str(log_linha.get("status") or "—").strip() or "—", "precisa_correcao_manual": str(log_linha.get("precisa_correcao_manual") or "").strip(), "detalhe_log": str(log_linha.get("detalhe") or "").strip(), "ra": str(dados_cartao.get("registro_academico") or log_linha.get("registro_academico") or "—").strip() or "—", "codigo_barras": str(dados_cartao.get("codigo_barras") or log_linha.get("codigo_barras") or "—").strip() or "—", "id_final": str( validacao.get("id_final") or (manual.get("id_final") if manual else "") or "" ).strip() or "—", "aluno_id": _extrair_aluno_id_para_keepedu( validacao, manual, dados_cartao, log_linha, ), "status_validacao": str(validacao.get("status_validacao") or "—").strip() or "—", "motivo_validacao": str(validacao.get("motivo") or "").strip(), "validacao_manual": bool(manual), "correcao_cantos_manual": origem_cantos.upper() == "MANUAL", "total_respostas_lidas": int(log_linha.get("total_respostas_lidas") or respostas_info["total_marcadas"] or 0), "respostas_preview": respostas_info["preview"], "respostas_itens": respostas_info["itens"], "erros": [str(e).strip() for e in erros if str(e).strip()], "importacao_sucesso": bool(importacao_item.get("sucesso")) if importacao_item else None, "envio_keepedu": envio_keepedu, "importacao_status": rotulo_envio or ( "Falhou" if importacao_item and not importacao_item.get("sucesso") else "—" ), } ) return registros def carregar_relatorio_importacao(nome_processamento: str) -> dict: caminho = caminho_relatorio_importacao(nome_processamento, modo="importacao") if not caminho.exists(): return {} dados = carregar_json(caminho, {}) return dados if isinstance(dados, dict) else {} def montar_dados_pos_processamento( nome_processamento: str, *, pagina: int = 1, por_pagina: int = 0, ) -> dict | None: if not lote_possui_arquivos_historico(nome_processamento): return None identificacao = obter_identificacao_prova_lote(nome_processamento) resumo_proc = resumo_processamento(nome_processamento) # O histórico é uma consulta de arquivos persistidos. Revalidar o lote aqui # pode disparar dezenas de consultas remotas e atrasar a abertura da página. dados_validacao = ( carregar_validacao_cadastral_salva(nome_processamento) or obter_validacao_cadastral_em_cache(nome_processamento) or {} ) resumo_val = dados_validacao.get("resumo") or {} relatorio_importacao = carregar_relatorio_importacao(nome_processamento) leituras = carregar_leituras_lote(nome_processamento) total_registros = sum(1 for leitura in leituras.values() if isinstance(leitura, dict)) total_paginas = max(1, (total_registros + por_pagina - 1) // por_pagina) if por_pagina > 0 else 1 pagina_final = max(1, min(int(pagina or 1), total_paginas)) registros = montar_registros_pos_processamento( nome_processamento, leituras=leituras, dados_validacao=dados_validacao, pagina=pagina_final, por_pagina=por_pagina, ) # O rótulo de data é opcional no detalhe. Evitamos listar todos os lotes # aqui, pois essa varredura bloqueava a abertura de um único histórico. rotulo_data = "" arquivos_presentes = [] for nome_arquivo in ( "envios_keepedu.json", "leituras_omr.json", "log_leitura_omr.csv", "validacao_cadastral.json", "validacao_manual.json", "meta_avaliacao.json", ): if (caminho_pasta_processamento(nome_processamento) / nome_arquivo).exists(): arquivos_presentes.append(nome_arquivo) return { "identificacao": identificacao, "nome_pasta": nome_processamento, "rotulo_data": rotulo_data, "arquivado": lote_esta_arquivado(nome_processamento), "resumo_proc": resumo_proc, "resumo_validacao": resumo_val, "relatorio_importacao": relatorio_importacao, "registros": registros, "arquivos_presentes": arquivos_presentes, "total_registros": total_registros, "page": pagina_final, "total_pages": total_paginas, } def _data_hora_lote_resumido(nome: str, lote: dict) -> datetime: metadados = lote.get("metadados") or {} candidatos = (metadados.get("criado_em"), lote.get("criado_em")) for indice, valor in enumerate(candidatos): data_hora = valor if isinstance(valor, datetime) else None if data_hora is None and valor: try: data_hora = datetime.fromisoformat(str(valor).strip().replace("Z", "+00:00")) except ValueError: data_hora = None if data_hora is None: continue if data_hora.tzinfo is not None: return data_hora.astimezone(obter_timezone_app()) # ``criado_em`` do banco e gravado em UTC; o ISO dos metadados novos # ja possui fuso e passa pelo ramo acima. if indice == 1: return data_hora.replace(tzinfo=timezone.utc).astimezone(obter_timezone_app()) return data_hora match = re.match(r"^processamento_(\d{8})_(\d{6})$", str(nome or "").strip()) if match: try: return datetime.strptime( f"{match.group(1)}_{match.group(2)}", "%Y%m%d_%H%M%S" ) except ValueError: pass return agora_local() def _rotulo_data_lote_resumido(data_hora: datetime) -> str: hoje = agora_local().date() if data_hora.date() == hoje: return "Hoje" if data_hora.date().toordinal() == hoje.toordinal() - 1: return "Ontem" return { 0: "Segunda", 1: "Terça", 2: "Quarta", 3: "Quinta", 4: "Sexta", 5: "Sábado", 6: "Domingo", }.get(data_hora.weekday(), "Lote") def _identificacao_lote_resumido(nome: str, metadados: dict, estado: dict) -> dict[str, str]: snapshot = estado.get("validacao_cadastral") or {} resumo_validacao = snapshot.get("resumo") if isinstance(snapshot, dict) else {} resumo_validacao = resumo_validacao if isinstance(resumo_validacao, dict) else {} id_avaliacao = str( metadados.get("id_avaliacao") or metadados.get("id") or resumo_validacao.get("id_prova_processamento") or "" ).strip() nome_prova = str( metadados.get("tipo_prova") or metadados.get("nome_simulado") or metadados.get("nome") or "" ).strip() if id_avaliacao and nome_prova: titulo = f"(ID {id_avaliacao}) · {nome_prova}" elif id_avaliacao: titulo = f"(ID {id_avaliacao})" elif nome_prova: titulo = nome_prova else: titulo = nome.replace("processamento_", "").replace("_", " ") return { "id_avaliacao": id_avaliacao, "nome_prova": nome_prova, "titulo": titulo, } def _resumo_keepedu_do_estado(estado: dict, total_cartoes: int) -> dict: envios = estado.get("envios_keepedu") or {} agrupados: dict[str, dict[str, bool]] = {} if isinstance(envios, dict): for chave, registro in envios.items(): if not isinstance(registro, dict): continue identidade = _stem_imagem(str(chave or "")) if not identidade: continue atual = agrupados.setdefault(identidade, {"vetores": False, "imagens": False}) atual["vetores"] = bool(atual["vetores"] or registro.get("vetores")) atual["imagens"] = bool(atual["imagens"] or registro.get("imagens")) completos = sum(item["vetores"] and item["imagens"] for item in agrupados.values()) parciais = sum(item["vetores"] != item["imagens"] for item in agrupados.values()) vetores = sum(item["vetores"] for item in agrupados.values()) completos = min(completos, total_cartoes) parciais = min(parciais, max(total_cartoes - completos, 0)) vetores = min(vetores, total_cartoes) finalizado = total_cartoes > 0 and completos >= total_cartoes titulo = "Finalizado" if finalizado else "Em envio" if completos or parciais else "Processado" return { "total_cartoes": total_cartoes, "enviados_completos": completos, "enviados_parciais": parciais, "enviados_vetores": vetores, "finalizado_keepedu": finalizado, "titulo": titulo, } def listar_lotes_historico_processos() -> list[dict]: """ Metadados dos lotes para a página de histórico (rótulo por prova + ID). """ from app.lotes_store import listar_lotes_resumidos from app.tenant import get_school global _HISTORICO_LISTAGEM_CACHE, _HISTORICO_LISTAGEM_TS, _HISTORICO_LISTAGEM_SCHOOL school = str(get_school() or "").strip().lower() agora = time.monotonic() if ( _HISTORICO_LISTAGEM_CACHE is not None and _HISTORICO_LISTAGEM_SCHOOL == school and (agora - _HISTORICO_LISTAGEM_TS) < _HISTORICO_LISTAGEM_TTL ): return list(_HISTORICO_LISTAGEM_CACHE) itens = [] for nome, lote in listar_lotes_resumidos(): metadados = dict(lote.get("metadados") or {}) estado = dict(lote.get("estado") or {}) total_cartoes = int(lote.get("total_leituras") or 0) identificacao = _identificacao_lote_resumido(nome, metadados, estado) snapshot = estado.get("validacao_cadastral") or {} resumo_val = snapshot.get("resumo") if isinstance(snapshot, dict) else {} resumo_val = resumo_val if isinstance(resumo_val, dict) else {} resumo_proc = estado.get("resumo_processamento") or {} if not isinstance(resumo_proc, dict) or not resumo_proc: # Compatibilidade com lotes criados antes do resumo persistido. A # contagem vem do SQL e evita baixar cada JSON de leitura. pendentes = min(int(resumo_val.get("pendentes") or 0), total_cartoes) resumo_proc = { "total_imagens": total_cartoes, "total_ok": max(total_cartoes - pendentes, 0), "total_erro": pendentes, "total_manual": pendentes, "total_pendencias": pendentes, } resumo_keepedu = _resumo_keepedu_do_estado(estado, total_cartoes) arquivado = bool(estado.get("arquivado")) data_hora = _data_hora_lote_resumido(nome, lote) meta_exibicao = { "nome": nome, "titulo": nome.replace("_", " "), "rotulo": _rotulo_data_lote_resumido(data_hora), "data": data_hora.strftime("%d/%m/%Y"), "hora": data_hora.strftime("%H:%M:%S"), } slug = gerar_slug_avaliacao_historico( identificacao["id_avaliacao"], identificacao["nome_prova"] ) or nome itens.append( { **meta_exibicao, "titulo_prova": identificacao["titulo"], "nome_prova": identificacao["nome_prova"], "id_prova": identificacao["id_avaliacao"], "slug_historico": slug, "rotulo_data": meta_exibicao["rotulo"], "url_detalhe": f"/historico/{nome}", "url_processamento": f"/historico/{nome}", "url_correcao": f"/correcao/{nome}" if not arquivado else "", "url_validacao": f"/validacao/{nome}", "arquivado": arquivado, "total_imagens": int(resumo_proc.get("total_imagens") or 0), "total_ok": int(resumo_proc.get("total_ok") or 0), "total_erro": int(resumo_proc.get("total_erro") or 0), "total_manual": int(resumo_proc.get("total_manual") or 0), "total_pendencias": int(resumo_proc.get("total_pendencias") or 0), "validados": int(resumo_val.get("validados") or 0), "pendentes_validacao": int(resumo_val.get("pendentes") or 0), "importacao_total": total_cartoes, "importacao_ok": int(resumo_keepedu.get("enviados_vetores") or 0), "envio_total_cartoes": int(resumo_keepedu.get("total_cartoes") or 0), "envio_cartoes_completos": int(resumo_keepedu.get("enviados_completos") or 0), "envio_cartoes_parciais": int(resumo_keepedu.get("enviados_parciais") or 0), "envio_keepedu_finalizado": bool(resumo_keepedu.get("finalizado_keepedu")), "envio_keepedu_titulo": str(resumo_keepedu.get("titulo") or "Processado"), } ) _HISTORICO_LISTAGEM_CACHE = itens _HISTORICO_LISTAGEM_TS = agora _HISTORICO_LISTAGEM_SCHOOL = school return itens HISTORICO_LOTES_POR_PAGINA = 12 _HISTORICO_LISTAGEM_CACHE: list[dict] | None = None _HISTORICO_LISTAGEM_TS: float = 0.0 _HISTORICO_LISTAGEM_SCHOOL: str = "" _HISTORICO_LISTAGEM_TTL = 5.0 _RESUMO_STATS_CACHE: dict[str, Any] | None = None _RESUMO_STATS_TS: float = 0.0 _RESUMO_STATS_SCHOOL: str = "" _RESUMO_STATS_TTL = 5.0 def _chave_dedup_lote_historico(lote: dict) -> str | None: id_prova = str(lote.get("id_prova") or "").strip() if id_prova: return f"id:{id_prova}" nome = str(lote.get("nome") or "").strip() if nome: return f"nome:{nome}" return None def _timestamp_nome_lote_historico(nome: str) -> int: match = re.match(r"^processamento_(\d{8})_(\d{6})$", str(nome or "").strip()) if not match: return 0 return int(match.group(1) + match.group(2)) def _rotulo_data_hora_processamento(lote: dict) -> str: data = str(lote.get("data") or "").strip() hora = str(lote.get("hora") or "").strip() if data and hora: return f"{data} às {hora}" if data: return data return str(lote.get("rotulo_data") or lote.get("nome") or "—") def _montar_item_processamento_modal(lote: dict) -> dict: nome = str(lote.get("nome") or "").strip() arquivado = bool(lote.get("arquivado")) return { "nome": nome, "data": str(lote.get("data") or ""), "hora": str(lote.get("hora") or ""), "rotulo_data": str(lote.get("rotulo_data") or ""), "data_hora": _rotulo_data_hora_processamento(lote), "total_imagens": int(lote.get("total_imagens") or 0), "total_ok": int(lote.get("total_ok") or 0), "total_erro": int(lote.get("total_erro") or 0), "importacao_ok": int(lote.get("importacao_ok") or 0), "importacao_total": int(lote.get("importacao_total") or 0), "envio_total_cartoes": int(lote.get("envio_total_cartoes") or 0), "envio_cartoes_completos": int(lote.get("envio_cartoes_completos") or 0), "envio_cartoes_parciais": int(lote.get("envio_cartoes_parciais") or 0), "envio_keepedu_finalizado": bool(lote.get("envio_keepedu_finalizado")), "envio_keepedu_titulo": str(lote.get("envio_keepedu_titulo") or "Processado"), "pendentes_validacao": int(lote.get("pendentes_validacao") or 0), "arquivado": arquivado, "url_detalhe": f"/historico/{nome}", "url_correcao": f"/correcao/{nome}" if not arquivado else "", "url_validacao": f"/validacao/{nome}", } def _lote_foi_enviado(lote: dict) -> bool: return bool(lote.get("envio_keepedu_finalizado")) def _lote_tem_envio_keepedu(lote: dict) -> bool: return bool( int(lote.get("envio_cartoes_completos") or 0) or int(lote.get("envio_cartoes_parciais") or 0) ) def agrupar_avaliacoes_historico(lotes: list[dict]) -> list[dict]: grupos: dict[str, dict] = {} ordem_insercao: list[str] = [] for lote in lotes: chave = _chave_dedup_lote_historico(lote) if not chave: continue if chave not in grupos: id_prova = str(lote.get("id_prova") or "").strip() nome_prova = str(lote.get("nome_prova") or "").strip() slug = gerar_slug_avaliacao_historico(id_prova, nome_prova) or str(lote.get("nome") or "") grupos[chave] = { "id_prova": id_prova, "titulo_prova": lote.get("titulo_prova") or nome_prova or lote.get("nome"), "nome_prova": nome_prova, "slug_avaliacao": slug, "processamentos": [], } ordem_insercao.append(chave) grupos[chave]["processamentos"].append(lote) avaliacoes: list[dict] = [] for chave in ordem_insercao: grupo = grupos[chave] processamentos_ordenados = sorted( grupo["processamentos"], key=lambda item: _timestamp_nome_lote_historico(str(item.get("nome") or "")), reverse=True, ) itens_modal = [_montar_item_processamento_modal(item) for item in processamentos_ordenados] recente = processamentos_ordenados[0] tem_alerta = any( int(item.get("total_erro") or 0) > 0 or int(item.get("total_manual") or 0) > 0 for item in processamentos_ordenados ) tem_arquivado = bool(processamentos_ordenados) and all( bool(item.get("arquivado")) for item in processamentos_ordenados ) pendentes_total = sum(int(item.get("pendentes_validacao") or 0) for item in processamentos_ordenados) lotes_total = len(processamentos_ordenados) lotes_enviados = sum(1 for item in processamentos_ordenados if _lote_foi_enviado(item)) lotes_com_envio = sum(1 for item in processamentos_ordenados if _lote_tem_envio_keepedu(item)) lotes_parciais = sum( 1 for item in processamentos_ordenados if _lote_tem_envio_keepedu(item) and not _lote_foi_enviado(item) ) avaliacoes.append( { "titulo_prova": grupo["titulo_prova"], "id_prova": grupo["id_prova"], "nome_prova": grupo["nome_prova"], "slug_avaliacao": grupo["slug_avaliacao"], "total_processamentos": lotes_total, "rotulo_data": str(recente.get("rotulo_data") or ""), "data_hora_ultimo": _rotulo_data_hora_processamento(recente), "total_imagens": sum(int(item.get("total_imagens") or 0) for item in processamentos_ordenados), "total_ok": sum(int(item.get("total_ok") or 0) for item in processamentos_ordenados), "total_erro": sum(int(item.get("total_erro") or 0) for item in processamentos_ordenados), "lotes_enviados": lotes_enviados, "lotes_com_envio": lotes_com_envio, "lotes_parciais": lotes_parciais, "lotes_total": lotes_total, "importacao_ok": int(recente.get("importacao_ok") or 0), "importacao_total": int(recente.get("importacao_total") or 0), "pendentes_validacao": pendentes_total, "arquivado": tem_arquivado, "tem_alerta": tem_alerta, "processamentos": itens_modal, } ) avaliacoes.sort( key=lambda item: _timestamp_nome_lote_historico( str((item.get("processamentos") or [{}])[0].get("nome") or "") ), reverse=True, ) return avaliacoes @cache_lotes_durante_operacao() def filtrar_e_paginar_lotes_historico( query: str = "", page: int = 1, por_pagina: int = HISTORICO_LOTES_POR_PAGINA, ) -> dict: avaliacoes = agrupar_avaliacoes_historico(listar_lotes_historico_processos()) query_norm = str(query or "").strip().lower() if query_norm: filtrados = [] for avaliacao in avaliacoes: texto_busca = " ".join( [ str(avaliacao.get("titulo_prova") or ""), str(avaliacao.get("nome_prova") or ""), str(avaliacao.get("id_prova") or ""), str(avaliacao.get("slug_avaliacao") or ""), str(avaliacao.get("rotulo_data") or ""), str(avaliacao.get("data_hora_ultimo") or ""), *[ " ".join( [ str(processamento.get("nome") or ""), str(processamento.get("data") or ""), str(processamento.get("hora") or ""), str(processamento.get("rotulo_data") or ""), str(processamento.get("data_hora") or ""), ] ) for processamento in (avaliacao.get("processamentos") or []) ], ] ).lower() if query_norm in texto_busca: filtrados.append(avaliacao) avaliacoes = filtrados total = len(avaliacoes) total_pages = max(1, math.ceil(total / por_pagina)) if total else 1 pagina = max(1, min(int(page or 1), total_pages)) inicio = (pagina - 1) * por_pagina pagina_avaliacoes = avaliacoes[inicio : inicio + por_pagina] return { "avaliacoes": pagina_avaliacoes, "lotes": pagina_avaliacoes, "total": total, "page": pagina, "total_pages": total_pages, "por_pagina": por_pagina, "query": str(query or "").strip(), } def resumo_estatisticas_globais_processamento() -> dict: """ Totais agregados para o painel: avaliações distintas com leitura e cartões lidos. """ from app.tenant import get_school global _RESUMO_STATS_CACHE, _RESUMO_STATS_TS, _RESUMO_STATS_SCHOOL school = str(get_school() or "").strip().lower() agora = time.monotonic() if ( _RESUMO_STATS_CACHE is not None and _RESUMO_STATS_SCHOOL == school and (agora - _RESUMO_STATS_TS) < _RESUMO_STATS_TTL ): return dict(_RESUMO_STATS_CACHE) lotes = listar_lotes_historico_processos() avaliacoes = agrupar_avaliacoes_historico(lotes) cartoes_processados = sum(int(lote.get("total_imagens") or 0) for lote in lotes) resultado = { "avaliacoes_processadas": len(avaliacoes), "cartoes_processados": cartoes_processados, } _RESUMO_STATS_CACHE = resultado _RESUMO_STATS_TS = agora _RESUMO_STATS_SCHOOL = school return resultado _VIS_LEGENDA_ALTURA = 156 _VIS_PADDING = 48 _VIS_LARGURA_MAXIMA = 1280 def resolver_chave_leitura(leituras: dict, chave_param: str) -> str | None: chave_texto = str(chave_param or "").strip() if not chave_texto or not isinstance(leituras, dict): return None if chave_texto in leituras: return chave_texto stem_alvo = _stem_imagem(chave_texto) for chave, dados in leituras.items(): if not isinstance(dados, dict): continue if _stem_imagem(chave) == stem_alvo: return chave arquivo = str(dados.get("arquivo_original") or "").strip() if arquivo and _stem_imagem(arquivo) == stem_alvo: return chave return None def _resolver_ids_folha_keepedu( nome_processamento: str, chave: str, dados_cartao: dict, ) -> tuple[str, str]: arquivo_original = str( dados_cartao.get("arquivo_original") or normalizar_nome_arquivo_debug(chave) ).strip() # A visualização pode ser solicitada dezenas de vezes pela galeria. Use o # estado persistido/cache e nunca revalide o lote (a revalidação consulta a # API por cartão e bloqueia o carregamento das miniaturas). dados_validacao = ( carregar_validacao_cadastral_salva(nome_processamento) or obter_validacao_cadastral_em_cache(nome_processamento) or {} ) validacoes = dados_validacao.get("validacoes") or {} validacao = validacoes.get(chave) or validacoes.get(arquivo_original) or {} if not validacao: stem_alvo = _stem_imagem(chave or arquivo_original) for chave_validacao, candidato in validacoes.items(): if _stem_imagem(chave_validacao) == stem_alvo: validacao = candidato or {} break manual_val = carregar_validacao_manual(nome_processamento) manual = manual_val.get(chave) or manual_val.get(arquivo_original) or {} if not manual: stem_alvo = _stem_imagem(chave or arquivo_original) for chave_manual, candidato in manual_val.items(): if _stem_imagem(chave_manual) == stem_alvo: manual = candidato or {} break aluno_id = str(validacao.get("id_final") or manual.get("id_final") or "").strip() if not aluno_id: _, id_aluno = quebrar_codigo_barras(extrair_codigo_barras_possivel(dados_cartao)) aluno_id = str(id_aluno or "").strip() # Lotes já enviados podem não ter a validação persistida. Para recuperar o # espelho, resolve apenas este aluno pelo RA (com cache), sem revalidar todo # o lote nem bloquear a página pela consulta dos demais cartões. if not aluno_id: ra = str(dados_cartao.get("registro_academico") or "").strip() if ra: aluno_id = _buscar_id_keepedu_por_ra_em_cache(ra) simulado_id = str(obter_identificacao_prova_lote(nome_processamento).get("id_avaliacao") or "").strip() if not simulado_id: id_prova, _ = quebrar_codigo_barras(extrair_codigo_barras_possivel(dados_cartao)) simulado_id = str(id_prova or "").strip() return aluno_id, simulado_id @lru_cache(maxsize=512) def _buscar_id_keepedu_por_ra_em_cache(ra: str) -> str: from app.alunos_client import buscar_aluno_por_ra try: aluno = buscar_aluno_por_ra(ra) except Exception: return "" return str(aluno.get("id") or "").strip() if aluno.get("encontrado") else "" def _extrair_xy_ponto(ponto) -> tuple[float, float] | None: if isinstance(ponto, dict): try: return float(ponto.get("x", 0)), float(ponto.get("y", 0)) except (TypeError, ValueError): return None if isinstance(ponto, (list, tuple)) and len(ponto) >= 2: try: return float(ponto[0]), float(ponto[1]) except (TypeError, ValueError): return None return None def _coletar_limites_cartao(dados_cartao: dict) -> tuple[float, float, float, float]: min_x = float("inf") min_y = float("inf") max_x = float("-inf") max_y = float("-inf") def registrar(ponto) -> None: nonlocal min_x, min_y, max_x, max_y coords = _extrair_xy_ponto(ponto) if not coords: return x, y = coords min_x = min(min_x, x) min_y = min(min_y, y) max_x = max(max_x, x) max_y = max(max_y, y) cantos = pontos_cantos_para_dict(dados_cartao.get("pontos_cantos")) for ponto in cantos.values(): registrar(ponto) pontos_mapeados = dados_cartao.get("pontos_mapeados") or {} if isinstance(pontos_mapeados, dict): for alternativas in pontos_mapeados.values(): if not isinstance(alternativas, dict): continue for ponto in alternativas.values(): registrar(ponto) if min_x == float("inf"): return 0.0, 0.0, 900.0, 1300.0 margem = 80.0 return min_x - margem, min_y - margem, max_x + margem, max_y + margem def _montar_meta_visualizacao( nome_processamento: str, chave: str, dados_cartao: dict, *, reconstruida: bool, ) -> dict: arquivo_original = str( dados_cartao.get("arquivo_original") or normalizar_nome_arquivo_debug(chave) ).strip() stem = _stem_imagem(arquivo_original or chave) # Esta função é chamada para cada miniatura da galeria. Revalidar o lote # aqui dispara consultas remotas para todos os cartões. dados_validacao = ( carregar_validacao_cadastral_salva(nome_processamento) or obter_validacao_cadastral_em_cache(nome_processamento) or {} ) validacoes = dados_validacao.get("validacoes") or {} validacao = validacoes.get(chave) or validacoes.get(arquivo_original) or {} manual_val = carregar_validacao_manual(nome_processamento) manual = manual_val.get(chave) or manual_val.get(arquivo_original) or {} log_linha = _carregar_linhas_log(nome_processamento).get(stem, {}) envio = resolver_envio_keepedu(chave, arquivo_original, carregar_envios_keepedu(nome_processamento)) origem_id = str(validacao.get("origem_id") or "").strip() origem_cantos = str(dados_cartao.get("origem_cantos") or "").strip().upper() precisa_correcao = str(log_linha.get("precisa_correcao_manual") or "").strip().upper() return { "arquivo": arquivo_original or chave, "ra": str(dados_cartao.get("registro_academico") or log_linha.get("registro_academico") or "—").strip() or "—", "codigo_barras": str(dados_cartao.get("codigo_barras") or log_linha.get("codigo_barras") or "—").strip() or "—", "id_final": str(validacao.get("id_final") or manual.get("id_final") or "—").strip() or "—", "origem_id": origem_id or "—", "id_manual": bool(manual) or origem_id == "VALIDACAO_MANUAL", "cantos_manual": origem_cantos == "MANUAL", "marcacoes_manual": precisa_correcao == "SIM", "reconstruida": reconstruida, "envio_keepedu": rotulo_envio_keepedu(envio) or "—", "erros": [str(item).strip() for item in (dados_cartao.get("erros") or []) if str(item).strip()], } def _renderizar_faixa_legenda(largura: int, meta: dict) -> np.ndarray: faixa = np.full((_VIS_LEGENDA_ALTURA, largura, 3), 236, dtype=np.uint8) cv2.rectangle(faixa, (0, 0), (largura - 1, _VIS_LEGENDA_ALTURA - 1), (205, 210, 218), 1) linhas = [ f"Arquivo: {meta.get('arquivo', '—')}", f"RA: {meta.get('ra', '—')} | Cod. barras: {meta.get('codigo_barras', '—')} | ID final: {meta.get('id_final', '—')}", ( f"ID manual: {'Sim' if meta.get('id_manual') else 'Nao'} | " f"Cantos manual: {'Sim' if meta.get('cantos_manual') else 'Nao'} | " f"Marcacoes manual: {'Sim' if meta.get('marcacoes_manual') else 'Nao'}" ), f"KeepEdu: {meta.get('envio_keepedu', '—')}", ] if meta.get("fonte_imagem"): linhas.append(str(meta["fonte_imagem"])) if meta.get("reconstruida"): linhas.append("Visualizacao reconstruida dos dados salvos (imagem original removida apos envio).") if meta.get("erros"): linhas.append(f"Alertas: {' | '.join(meta['erros'][:2])}") y = 24 for texto in linhas[:6]: cv2.putText( faixa, texto[:140], (14, y), cv2.FONT_HERSHEY_SIMPLEX, 0.48, (28, 36, 48), 1, cv2.LINE_AA, ) y += 22 return faixa def _criar_mapeador_cartao( dados_cartao: dict, largura: int, altura: int, offset_y: int = 0, ): min_x, min_y, max_x, max_y = _coletar_limites_cartao(dados_cartao) src_w = max(max_x - min_x, 1.0) src_h = max(max_y - min_y, 1.0) area_util_w = max(largura - (_VIS_PADDING * 2), 1) area_util_h = max(altura - (_VIS_PADDING * 2), 1) escala = min(area_util_w / src_w, area_util_h / src_h, 1.0) def mapear(x: float, y: float) -> tuple[int, int]: px = int(round(_VIS_PADDING + (x - min_x) * escala)) py = int(round(offset_y + _VIS_PADDING + (y - min_y) * escala)) return px, py return mapear, escala def _numero_pergunta_omr(pergunta) -> int | None: if isinstance(pergunta, int): return pergunta texto = str(pergunta or "").strip() if texto.isdigit(): return int(texto) match = re.match(r"^Pergunta0*(\d+)$", texto, flags=re.IGNORECASE) if match: return int(match.group(1)) return None def _texto_rotulo_marcacao(pergunta, resposta_sel: str) -> str: """ RA (Pergunta001–006): somente o dígito marcado. Questões (Pergunta009+): somente a letra marcada. """ valor = str(resposta_sel or "").strip() if not valor: return "" numero = _numero_pergunta_omr(pergunta) if numero is not None: if 1 <= numero <= 6: return valor if numero >= 9: return valor return valor def _desenhar_cantos_cartao( imagem: np.ndarray, dados_cartao: dict, mapear, ) -> None: cantos = pontos_cantos_para_dict(dados_cartao.get("pontos_cantos")) if len(cantos) < 4: return ordem = ["TOP_LEFT", "TOP_RIGHT", "BOTTOM_RIGHT", "BOTTOM_LEFT"] pontos = [] for chave in ordem: coords = _extrair_xy_ponto(cantos.get(chave)) if not coords: return pontos.append(mapear(coords[0], coords[1])) manual = str(dados_cartao.get("origem_cantos") or "").strip().upper() == "MANUAL" cor_borda = (0, 140, 255) if manual else (255, 120, 0) cv2.polylines(imagem, [np.array(pontos, dtype=np.int32)], True, cor_borda, 2, cv2.LINE_AA) rotulos = {"TOP_LEFT": "TL", "TOP_RIGHT": "TR", "BOTTOM_RIGHT": "BR", "BOTTOM_LEFT": "BL"} for chave in ordem: coords = _extrair_xy_ponto(cantos.get(chave)) if not coords: continue px, py = mapear(coords[0], coords[1]) cv2.circle(imagem, (px, py), 8, cor_borda, 2, cv2.LINE_AA) cv2.circle(imagem, (px, py), 3, (20, 20, 20), -1, cv2.LINE_AA) cv2.putText( imagem, rotulos[chave], (px + 10, py - 8), cv2.FONT_HERSHEY_SIMPLEX, 0.45, cor_borda, 1, cv2.LINE_AA, ) def _desenhar_marcacoes_cartao( imagem: np.ndarray, dados_cartao: dict, mapear, ) -> None: respostas = dados_cartao.get("respostas") or {} pontos_mapeados = dados_cartao.get("pontos_mapeados") or {} if not isinstance(pontos_mapeados, dict): return for pergunta, alternativas in pontos_mapeados.items(): if not isinstance(alternativas, dict): continue resposta_sel = str(respostas.get(pergunta) or "").strip() if not resposta_sel and str(pergunta).isdigit(): numero = int(pergunta) resposta_sel = str(respostas.get(numero) or resposta(respostas, numero) or "").strip() for alt, ponto in alternativas.items(): coords = _extrair_xy_ponto(ponto) if not coords: continue px, py = mapear(coords[0], coords[1]) alt_str = str(alt).strip() selecionada = bool(resposta_sel and alt_str == resposta_sel) if selecionada: cv2.circle(imagem, (px, py), 11, (0, 190, 70), 2, cv2.LINE_AA) cv2.circle(imagem, (px, py), 5, (0, 210, 80), -1, cv2.LINE_AA) else: cv2.circle(imagem, (px, py), 9, (40, 40, 220), 2, cv2.LINE_AA) rotulo = _texto_rotulo_marcacao(pergunta, resposta_sel) if rotulo: coords_ref = None for alt, ponto in alternativas.items(): if str(alt).strip() == resposta_sel: coords_ref = _extrair_xy_ponto(ponto) break if not coords_ref: primeiro = next((p for p in alternativas.values() if _extrair_xy_ponto(p)), None) coords_ref = _extrair_xy_ponto(primeiro) if coords_ref: px_ref, py_ref = mapear(coords_ref[0], coords_ref[1]) numero = _numero_pergunta_omr(pergunta) eh_ra = numero is not None and 1 <= numero <= 6 offset_x = -8 if eh_ra else -6 offset_y = -14 if eh_ra else -12 cv2.putText( imagem, rotulo, (max(4, px_ref + offset_x), max(16, py_ref + offset_y)), cv2.FONT_HERSHEY_SIMPLEX, 0.45 if eh_ra else 0.4, (0, 170, 60), 1, cv2.LINE_AA, ) def _desenhar_marcacoes_sobre_imagem(imagem: np.ndarray, dados_cartao: dict) -> np.ndarray: saida = imagem.copy() altura, largura = saida.shape[:2] def mapear(x: float, y: float) -> tuple[int, int]: return int(round(x)), int(round(y)) _desenhar_cantos_cartao(saida, dados_cartao, mapear) _desenhar_marcacoes_cartao(saida, dados_cartao, mapear) return saida def _criar_visualizacao_reconstruida(dados_cartao: dict, meta: dict) -> np.ndarray: min_x, min_y, max_x, max_y = _coletar_limites_cartao(dados_cartao) src_w = max(max_x - min_x, 320.0) src_h = max(max_y - min_y, 420.0) escala = min((_VIS_LARGURA_MAXIMA - (_VIS_PADDING * 2)) / src_w, 1.0) largura_corpo = int(min(_VIS_LARGURA_MAXIMA, src_w * escala + (_VIS_PADDING * 2))) altura_corpo = int(src_h * escala + (_VIS_PADDING * 2)) corpo = np.full((altura_corpo, largura_corpo, 3), 252, dtype=np.uint8) cv2.rectangle(corpo, (8, 8), (largura_corpo - 9, altura_corpo - 9), (214, 220, 228), 1) # A reconstrução é usada somente quando nem o original local nem o espelho # KeepEdu estão disponíveis. Os cantos e as respostas são desenhados pelo # overlay JSON no navegador; não os incorpore na imagem-base para evitar # linhas azuis ou marcações duplicadas. legenda = _renderizar_faixa_legenda(largura_corpo, meta) return np.vstack([legenda, corpo]) def _criar_base_reconstruida_no_plano_omr(dados_cartao: dict) -> np.ndarray: """Cria somente o plano neutro das marcações, sem uma segunda imagem.""" _, _, max_x, max_y = _coletar_limites_cartao(dados_cartao) largura = max(320, int(np.ceil(max_x)) + 16) altura = max(420, int(np.ceil(max_y)) + 16) return np.full((altura, largura, 3), (250, 248, 246), dtype=np.uint8) def _empilhar_legenda_sobre_imagem(imagem: np.ndarray, meta: dict) -> np.ndarray: largura = imagem.shape[1] legenda = _renderizar_faixa_legenda(largura, meta) return np.vstack([legenda, imagem]) @lru_cache(maxsize=256) def _baixar_espelho_keepedu_em_cache(url_imagem: str) -> bytes | None: """Mantém em memória apenas espelhos válidos para evitar downloads repetidos.""" from app.keepedu_folha_resposta import baixar_imagem_folha_resposta try: conteudo, _ = baixar_imagem_folha_resposta(url_imagem) return conteudo or None except ValueError: return None def _alinhar_espelho_ao_sistema_omr(conteudo: bytes, dados_cartao: dict) -> bytes | None: """Projeta o espelho KeepEdu no mesmo plano dos pontos OMR salvos.""" imagem = cv2.imdecode(np.frombuffer(conteudo, dtype=np.uint8), cv2.IMREAD_COLOR) cantos = pontos_cantos_para_dict(dados_cartao.get("pontos_cantos")) ordem = ["TOP_LEFT", "TOP_RIGHT", "BOTTOM_RIGHT", "BOTTOM_LEFT"] if imagem is None or any(chave not in cantos for chave in ordem): return conteudo if imagem is not None else None destino = np.float32([[cantos[chave]["x"], cantos[chave]["y"]] for chave in ordem]) largura = max(1, int(np.ceil(max(ponto[0] for ponto in destino))) + 1) altura = max(1, int(np.ceil(max(ponto[1] for ponto in destino))) + 1) origem = np.float32([ [0, 0], [imagem.shape[1] - 1, 0], [imagem.shape[1] - 1, imagem.shape[0] - 1], [0, imagem.shape[0] - 1], ]) transformacao = cv2.getPerspectiveTransform(origem, destino) alinhada = cv2.warpPerspective(imagem, transformacao, (largura, altura), borderValue=(246, 248, 252)) sucesso, jpg = cv2.imencode(".jpg", alinhada, [int(cv2.IMWRITE_JPEG_QUALITY), 88]) return jpg.tobytes() if sucesso else conteudo def _localizar_imagem_base_original( nome_processamento: str, chave: str, dados_cartao: dict, ) -> Path | None: """Localiza o scan para exibição: originais primeiro, debug_omr como fallback.""" candidatos = [dados_cartao.get("caminho_original_processamento")] pasta_originais = caminho_originais(nome_processamento) pasta_debug = caminho_pasta_processamento(nome_processamento) / "debug_omr" nomes = { Path(str(chave or "")).name, Path(str(dados_cartao.get("arquivo_original") or "")).name, } candidatos.extend(pasta_originais / nome for nome in nomes if nome) for candidato in candidatos: if not candidato: continue caminho = Path(str(candidato)) if caminho.exists() and caminho.is_file(): partes = {parte.lower() for parte in caminho.parts} if "pendencias" not in partes and "debug_omr" not in partes: return caminho if pasta_originais.is_dir(): stems = {Path(nome).stem.lower() for nome in nomes if nome} encontrado = next( ( arquivo for arquivo in pasta_originais.iterdir() if arquivo.is_file() and arquivo.stem.lower() in stems ), None, ) if encontrado: return encontrado # Sem original em runtime (retenção), a galeria ainda tem o artefato em debug_omr. if pasta_debug.is_dir(): stems = {Path(nome).stem.lower() for nome in nomes if nome} for arquivo in pasta_debug.iterdir(): if arquivo.is_file() and arquivo.stem.lower() in stems: return arquivo for nome in nomes: if not nome: continue caminho = pasta_debug / Path(nome).name if caminho.exists() and caminho.is_file(): return caminho return None def _imagem_foi_enviada_ao_keepedu( nome_processamento: str, chave: str, dados_cartao: dict, ) -> bool: """Indica se este scan já tem a imagem persistida no KeepEdu.""" arquivo_original = str(dados_cartao.get("arquivo_original") or "").strip() envio = resolver_envio_keepedu( chave, arquivo_original, carregar_envios_keepedu(nome_processamento), ) return bool(envio and envio.get("imagens")) def _codificar_imagem_base_cartao(imagem: np.ndarray, largura_maxima: int | None = None) -> bytes | None: """Codifica a imagem e, quando necessario, gera uma miniatura leve.""" if largura_maxima and imagem.shape[1] > largura_maxima: escala = largura_maxima / imagem.shape[1] imagem = cv2.resize( imagem, (largura_maxima, max(1, round(imagem.shape[0] * escala))), interpolation=cv2.INTER_AREA, ) sucesso, imagem_jpg = cv2.imencode(".jpg", imagem, [int(cv2.IMWRITE_JPEG_QUALITY), 82]) return imagem_jpg.tobytes() if sucesso else None def gerar_imagem_base_cartao( nome_processamento: str, chave_imagem: str, *, permitir_espelho_keepedu: bool = True, largura_maxima: int | None = None, ) -> bytes | None: """ Retorna a imagem base limpa para o painel de correção. """ leituras = carregar_leituras_lote(nome_processamento) chave = resolver_chave_leitura(leituras, chave_imagem) if not chave: return None dados_cartao = leituras.get(chave) if not isinstance(dados_cartao, dict): return None # Antes do envio, o scan local é a fonte de verdade. O espelho só existe # depois que a imagem foi persistida no KeepEdu. imagem_enviada = _imagem_foi_enviada_ao_keepedu( nome_processamento, chave, dados_cartao ) if not imagem_enviada: caminho_local = _localizar_imagem_base_original( nome_processamento, chave, dados_cartao ) if caminho_local: imagem_local = cv2.imread(str(caminho_local), cv2.IMREAD_COLOR) if imagem_local is not None: return _codificar_imagem_base_cartao(imagem_local, largura_maxima) # Depois do envio, a imagem exibida é o espelho do KeepEdu. Não # consultamos esse espelho para imagens que ainda não foram enviadas. aluno_id, simulado_id = _resolver_ids_folha_keepedu(nome_processamento, chave, dados_cartao) if imagem_enviada and permitir_espelho_keepedu and aluno_id and simulado_id: from app.keepedu_folha_resposta import ver_folha_resposta_keepedu resultado_keepedu = ver_folha_resposta_keepedu(aluno_id, simulado_id) url_espelho = str(resultado_keepedu.get("imagem_url") or "") imagem_espelho = _baixar_espelho_keepedu_em_cache(url_espelho) if url_espelho else None if imagem_espelho: return _alinhar_espelho_ao_sistema_omr(imagem_espelho, dados_cartao) # Fallback: a KeepEdu ainda não publicou o espelho ou devolveu URL indisponível. # O overlay do navegador usa as coordenadas originais do OMR. O fallback # precisa manter esse mesmo plano para evitar uma segunda visualização # desalinhada das marcações. imagem_reconstruida = _criar_base_reconstruida_no_plano_omr(dados_cartao) return _codificar_imagem_base_cartao(imagem_reconstruida, largura_maxima) def gerar_visualizacao_cartao_keepedu( nome_processamento: str, chave_imagem: str, *, priorizar_keepedu: bool = False, ) -> bytes | None: """ Gera JPG com marcações sobre o scan local ou espelho do KeepEdu. """ leituras = carregar_leituras_lote(nome_processamento) chave = resolver_chave_leitura(leituras, chave_imagem) if not chave: return None dados_cartao = leituras.get(chave) if not isinstance(dados_cartao, dict): return None imagem_enviada = _imagem_foi_enviada_ao_keepedu( nome_processamento, chave, dados_cartao ) def _via_local() -> bytes | None: caminho_local = _localizar_imagem_base_original( nome_processamento, chave, dados_cartao ) if not caminho_local: return None imagem_local = cv2.imread(str(caminho_local), cv2.IMREAD_COLOR) if imagem_local is None: return None meta = _montar_meta_visualizacao( nome_processamento, chave, dados_cartao, reconstruida=False ) meta["fonte_imagem"] = "Imagem local + marcações OMR" corpo = _desenhar_marcacoes_sobre_imagem(imagem_local, dados_cartao) saida = _empilhar_legenda_sobre_imagem(corpo, meta) sucesso, jpg = cv2.imencode( ".jpg", saida, [int(cv2.IMWRITE_JPEG_QUALITY), 88] ) return jpg.tobytes() if sucesso else None if not imagem_enviada: imagem_local = _via_local() if imagem_local: return imagem_local reconstruida = _criar_visualizacao_reconstruida( dados_cartao, _montar_meta_visualizacao( nome_processamento, chave, dados_cartao, reconstruida=True ), ) sucesso, jpg = cv2.imencode( ".jpg", reconstruida, [int(cv2.IMWRITE_JPEG_QUALITY), 88] ) return jpg.tobytes() if sucesso else None aluno_id, simulado_id = _resolver_ids_folha_keepedu(nome_processamento, chave, dados_cartao) def _via_keepedu() -> bytes | None: if not (aluno_id and simulado_id): return None return gerar_visualizacao_folha_keepedu_bytes( nome_processamento, chave, aluno_id, simulado_id, ) imagem = _via_keepedu() if imagem: return imagem # Sem a imagem real, mantenha apenas o plano neutro que recebe o overlay # interativo da tela de correção. imagem_reconstruida = _criar_base_reconstruida_no_plano_omr(dados_cartao) sucesso, imagem_jpg = cv2.imencode(".jpg", imagem_reconstruida, [int(cv2.IMWRITE_JPEG_QUALITY), 88]) return imagem_jpg.tobytes() if sucesso else None def gerar_visualizacao_folha_keepedu_bytes( nome_processamento: str, chave_imagem: str, aluno_id: Any, simulado_id: Any, ) -> bytes | None: """ Baixa a folha do KeepEdu e sobrepõe cantos/marcações detectadas na leitura OMR. """ from app.keepedu_folha_resposta import baixar_imagem_folha_resposta, ver_folha_resposta_keepedu leituras = carregar_leituras_lote(nome_processamento) chave = resolver_chave_leitura(leituras, chave_imagem) if not chave: return None dados_cartao = leituras.get(chave) if not isinstance(dados_cartao, dict): return None resultado = ver_folha_resposta_keepedu(aluno_id, simulado_id) if not resultado.get("ok") or not resultado.get("success") or not resultado.get("imagem_url"): return None try: conteudo, _ = baixar_imagem_folha_resposta(resultado["imagem_url"]) except ValueError: return None buffer = np.frombuffer(conteudo, dtype=np.uint8) imagem = cv2.imdecode(buffer, cv2.IMREAD_COLOR) if imagem is None: return None meta = _montar_meta_visualizacao( nome_processamento, chave, dados_cartao, reconstruida=False, ) meta["fonte_imagem"] = "KeepEdu + marcações OMR" corpo = _desenhar_marcacoes_sobre_imagem(imagem, dados_cartao) saida = _empilhar_legenda_sobre_imagem(corpo, meta) sucesso, imagem_jpg = cv2.imencode(".jpg", saida, [int(cv2.IMWRITE_JPEG_QUALITY), 88]) return imagem_jpg.tobytes() if sucesso else None