""" Montagem do pós-processamento a partir dos arquivos persistidos no lote. """ from __future__ import annotations import csv 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 carregar_meta_avaliacao_do_lote, obter_simulado_por_id 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_pasta_processamento, carregar_json, listar_imagens_debug, listar_processamentos_recentes, localizar_imagem_original, 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, quebrar_codigo_barras, resolver_envio_keepedu, rotulo_envio_keepedu, detectar_id_prova_do_processamento, ) def _stem_imagem(nome: str) -> str: texto = normalizar_nome_arquivo_debug(str(nome or "").strip()) return Path(texto).stem.lower() def caminho_resumo_processamento_txt(nome_processamento: str) -> Path: return caminho_pasta_processamento(nome_processamento) / "resumo_processamento.txt" def lote_possui_arquivos_historico(nome_processamento: str) -> bool: 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 = carregar_meta_avaliacao_do_lote(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_json(caminho_leituras(nome_processamento), {}) dados_val = carregar_validacao_cadastral_salva(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) if not caminho.exists(): return {} with open(caminho, "r", encoding="utf-8-sig", newline="") as arquivo: leitor = csv.DictReader(arquivo, delimiter=";") indice: dict[str, dict] = {} for linha in leitor: 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) -> list[dict]: leituras = carregar_json(caminho_leituras(nome_processamento), {}) if not leituras: return [] dados_validacao = carregar_validacao_cadastral_salva(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) registros: list[dict] = [] for chave, dados_cartao in leituras.items(): if not isinstance(dados_cartao, dict): continue 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 "—" ), } ) registros.sort(key=lambda item: item["arquivo"].lower()) return registros def carregar_resumo_processamento_txt(nome_processamento: str) -> str: caminho = caminho_resumo_processamento_txt(nome_processamento) if not caminho.exists(): return "" try: return caminho.read_text(encoding="utf-8").strip() except Exception: return "" 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) -> 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) dados_validacao = carregar_validacao_cadastral_salva(nome_processamento) or {} resumo_val = dados_validacao.get("resumo") or {} relatorio_importacao = carregar_relatorio_importacao(nome_processamento) registros = montar_registros_pos_processamento(nome_processamento) rotulo_data = "" for meta in listar_processamentos_recentes(): if meta.get("nome") == nome_processamento: rotulo_data = str(meta.get("rotulo") or "") break arquivos_presentes = [] for nome_arquivo in ( "envios_keepedu.json", "importacao_keepedu.json", "leituras_omr.json", "log_leitura_omr.csv", "resumo_processamento.txt", "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_processamento_txt": carregar_resumo_processamento_txt(nome_processamento), "resumo_proc": resumo_proc, "resumo_validacao": resumo_val, "relatorio_importacao": relatorio_importacao, "registros": registros, "arquivos_presentes": arquivos_presentes, "total_registros": len(registros), } def listar_lotes_historico_processos() -> list[dict]: """ Metadados dos lotes para a página de histórico (rótulo por prova + ID). """ itens = [] for meta in listar_processamentos_recentes(): nome = meta["nome"] if not lote_possui_arquivos_historico(nome): continue identificacao = obter_identificacao_prova_lote(nome) resumo_proc = resumo_processamento(nome) dados_salvos = carregar_validacao_cadastral_salva(nome) resumo_val = (dados_salvos or {}).get("resumo") or {} arquivado = lote_esta_arquivado(nome) slug = gerar_slug_historico_lote(nome) itens.append( { **meta, "titulo_prova": identificacao["titulo"], "nome_prova": identificacao["nome_prova"], "id_prova": identificacao["id_avaliacao"], "slug_historico": slug, "rotulo_data": meta.get("rotulo") or "", "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": int((carregar_relatorio_importacao(nome) or {}).get("total") or 0), "importacao_ok": int((carregar_relatorio_importacao(nome) or {}).get("importadas") or 0), } ) return itens HISTORICO_LOTES_POR_PAGINA = 12 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), "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: total = int(lote.get("importacao_total") or 0) ok = int(lote.get("importacao_ok") or 0) return total > 0 and ok >= total 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)) 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_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 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. """ lotes = listar_lotes_historico_processos() avaliacoes = agrupar_avaliacoes_historico(lotes) cartoes_processados = sum(int(lote.get("total_imagens") or 0) for lote in lotes) return { "avaliacoes_processadas": len(avaliacoes), "cartoes_processados": cartoes_processados, } _VIS_TEMP_MAX_IDADE_SEG = 1800 _VIS_LEGENDA_ALTURA = 156 _VIS_PADDING = 48 _VIS_LARGURA_MAXIMA = 1280 def _pasta_visualizacao_temp(nome_processamento: str) -> Path: pasta = caminho_pasta_processamento(nome_processamento) / "_vis_temp" pasta.mkdir(parents=True, exist_ok=True) return pasta def _limpar_visualizacoes_antigas(pasta: Path, max_idade_seg: int = _VIS_TEMP_MAX_IDADE_SEG) -> None: agora = time.time() for arquivo in pasta.glob("*.jpg"): try: if agora - arquivo.stat().st_mtime > max_idade_seg: arquivo.unlink(missing_ok=True) except OSError: pass def _remover_visualizacoes_do_cartao(pasta: Path, stem: str) -> None: for arquivo in pasta.glob(f"{stem}_*.jpg"): try: arquivo.unlink(missing_ok=True) except OSError: pass 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 _localizar_imagem_base(nome_processamento: str, dados_cartao: dict, chave: str) -> Path | None: caminho_debug = str(dados_cartao.get("debug_bolhas") or "").strip() if caminho_debug: caminho = Path(caminho_debug) if caminho.exists() and caminho.is_file(): return caminho arquivo_original = str( dados_cartao.get("arquivo_original") or normalizar_nome_arquivo_debug(chave) ).strip() pasta_processamento = caminho_pasta_processamento(nome_processamento) nomes_busca: list[str] = [] if arquivo_original: nomes_busca.append(arquivo_original) nomes_busca.append(f"template_{arquivo_original}") nomes_busca.append(normalizar_nome_arquivo_debug(chave)) nomes_busca.append(f"template_{normalizar_nome_arquivo_debug(chave)}") vistos: set[str] = set() for subpasta in ("debug_omr", "originais", "pendencias"): pasta = pasta_processamento / subpasta if not pasta.exists(): continue for nome_arquivo in nomes_busca: if not nome_arquivo or nome_arquivo in vistos: continue vistos.add(nome_arquivo) caminho = pasta / nome_arquivo if caminho.exists() and caminho.is_file(): return caminho leituras = carregar_json(caminho_leituras(nome_processamento), {}) return localizar_imagem_original(nome_processamento, chave, leituras) 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() dados_validacao = carregar_validacao_cadastral_salva(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 {} aluno_id = str(validacao.get("id_final") or manual.get("id_final") or "").strip() if not aluno_id: _, id_aluno = quebrar_codigo_barras(dados_cartao.get("codigo_barras") or "") aluno_id = str(id_aluno or "").strip() simulado_id = str(obter_identificacao_prova_lote(nome_processamento).get("id_avaliacao") or "").strip() if not simulado_id: id_prova, _ = quebrar_codigo_barras(dados_cartao.get("codigo_barras") or "") simulado_id = str(id_prova or "").strip() return aluno_id, simulado_id 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) dados_validacao = carregar_validacao_cadastral_salva(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) mapear, _ = _criar_mapeador_cartao(dados_cartao, largura_corpo, altura_corpo, offset_y=0) _desenhar_cantos_cartao(corpo, dados_cartao, mapear) _desenhar_marcacoes_cartao(corpo, dados_cartao, mapear) legenda = _renderizar_faixa_legenda(largura_corpo, meta) return np.vstack([legenda, corpo]) 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]) def gerar_visualizacao_cartao_temporaria( nome_processamento: str, chave_imagem: str, ) -> Path | None: """ Gera JPG temporário (curta duração) com marcações e indicações de correção manual. """ leituras = carregar_json(caminho_leituras(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 pasta_temp = _pasta_visualizacao_temp(nome_processamento) _limpar_visualizacoes_antigas(pasta_temp) stem = _stem_imagem(chave) _remover_visualizacoes_do_cartao(pasta_temp, stem) caminho_saida = pasta_temp / f"{stem}_{int(time.time() * 1000)}.jpg" imagem_base_path = _localizar_imagem_base(nome_processamento, dados_cartao, chave) reconstruida = imagem_base_path is None meta = _montar_meta_visualizacao( nome_processamento, chave, dados_cartao, reconstruida=reconstruida, ) if imagem_base_path: imagem = cv2.imread(str(imagem_base_path)) if imagem is not None: caminho_texto = str(imagem_base_path).lower().replace("\\", "/") usa_debug_pronto = "/debug_omr/" in caminho_texto or caminho_texto.endswith( Path(str(dados_cartao.get("debug_bolhas") or "")).name.lower() ) corpo = imagem if usa_debug_pronto else _desenhar_marcacoes_sobre_imagem(imagem, dados_cartao) saida = _empilhar_legenda_sobre_imagem(corpo, meta) cv2.imwrite(str(caminho_saida), saida, [int(cv2.IMWRITE_JPEG_QUALITY), 88]) return caminho_saida aluno_id, simulado_id = _resolver_ids_folha_keepedu(nome_processamento, chave, dados_cartao) if aluno_id and simulado_id: caminho_keepedu = gerar_visualizacao_folha_keepedu_temporaria( nome_processamento, chave, aluno_id, simulado_id, ) if caminho_keepedu and caminho_keepedu.exists(): return caminho_keepedu saida = _criar_visualizacao_reconstruida(dados_cartao, meta) cv2.imwrite(str(caminho_saida), saida, [int(cv2.IMWRITE_JPEG_QUALITY), 88]) return caminho_saida def gerar_visualizacao_folha_keepedu_temporaria( nome_processamento: str, chave_imagem: str, aluno_id: Any, simulado_id: Any, ) -> Path | 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_json(caminho_leituras(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 pasta_temp = _pasta_visualizacao_temp(nome_processamento) _limpar_visualizacoes_antigas(pasta_temp) stem = _stem_imagem(chave) _remover_visualizacoes_do_cartao(pasta_temp, f"{stem}_keepedu") caminho_saida = pasta_temp / f"{stem}_keepedu_{int(time.time() * 1000)}.jpg" 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) cv2.imwrite(str(caminho_saida), saida, [int(cv2.IMWRITE_JPEG_QUALITY), 88]) return caminho_saida