import json import re import time from datetime import datetime, timezone from pathlib import Path from app.alunos_client import buscar_aluno_por_id, buscar_aluno_por_ra from app.services import ( caminho_leituras, carregar_json, carregar_leituras_lote, resolver_chave_leitura, salvar_leituras_lote, ) from app.lotes_store import obter_estado_lote, salvar_estado_lote _VALIDACAO_CACHE_TTL_SEGUNDOS = 300 _validacoes_em_memoria: dict[str, dict] = {} def adicionar_rotulos_exibicao(validacao: dict) -> dict: """Adds coordinator-friendly labels without changing workflow status codes.""" origem = str(validacao.get("origem_id") or "").strip() status = str(validacao.get("status_validacao") or "").strip() motivo = str(validacao.get("motivo") or "").strip() validacao["rotulo_origem_id"] = { "CODIGO_BARRAS_CARTAO": "Código de barras do cartão", "API_RA": "Localizado pelo RA", "VALIDACAO_MANUAL": "Corrigido manualmente", }.get(origem, origem or "-") validacao["rotulo_status_validacao"] = { "VALIDADO_CODIGO_BARRAS": "Identificação confirmada", "VALIDADO_API_RA": "Identificação confirmada", "VALIDADO_MANUAL": "Corrigido manualmente", "PENDENTE_SEM_ID_PROVA": "Revisão necessária", "PENDENTE_RA_NAO_ENCONTRADO": "RA não localizado", "PENDENTE_SEM_IDENTIFICACAO": "Identificação não encontrada", "PENDENTE_ALUNO_FORA_AVALIACAO": "Aluno fora da avaliação", }.get(status, status or "-") validacao["motivo_exibicao"] = motivo.replace( "pela API", "automaticamente pelo RA" ) return validacao def _assinatura_validacao(nome_avaliacao: str) -> tuple[int, ...]: from app.lotes_store import caminho_lote caminhos = ( caminho_lote(nome_avaliacao), caminho_leituras(nome_avaliacao), caminho_validacao_manual(nome_avaliacao), caminho_pendencias_importacao(nome_avaliacao), caminho_envios_keepedu(nome_avaliacao), ) assinatura: list[int] = [] for caminho in caminhos: try: assinatura.append(Path(caminho).stat().st_mtime_ns) except OSError: assinatura.append(0) return tuple(assinatura) def obter_validacao_cadastral_em_cache(nome_avaliacao: str) -> dict | None: entrada = _validacoes_em_memoria.get(str(nome_avaliacao or "")) if not entrada or time.monotonic() >= entrada["expira_em"]: return None if entrada["assinatura"] != _assinatura_validacao(nome_avaliacao): return None return entrada["dados"] def invalidar_validacao_cadastral_em_cache(nome_avaliacao: str) -> None: _validacoes_em_memoria.pop(str(nome_avaliacao or ""), None) def leitura_bloqueada_id_avaliacao(dados_cartao: dict) -> bool: """Cartão com avaliação diferente da configurada no lote — bloqueia envio KeepEdu.""" if not isinstance(dados_cartao, dict): return False for erro in dados_cartao.get("erros") or []: texto = str(erro or "").lower() if "diferente da avaliação selecionada" in texto and "processamento bloqueado" in texto: return True return False def caminho_validacao_manual(nome_avaliacao: str): caminho_leitura = caminho_leituras(nome_avaliacao) pasta_processamento = caminho_leitura.parent return pasta_processamento / "validacao_manual.json" def carregar_validacao_manual(nome_avaliacao: str): estado = obter_estado_lote(nome_avaliacao) if isinstance(estado.get("validacao_manual"), dict): return dict(estado["validacao_manual"]) caminho = caminho_validacao_manual(nome_avaliacao) if not caminho.exists(): return {} with open(caminho, "r", encoding="utf-8") as f: return json.load(f) def salvar_validacao_manual(nome_avaliacao: str, dados: dict) -> None: estado = obter_estado_lote(nome_avaliacao) estado["validacao_manual"] = dict(dados or {}) estado["validacao_cadastral_desatualizada"] = True salvar_estado_lote(nome_avaliacao, estado) def caminho_pendencias_importacao(nome_avaliacao: str): caminho_leitura = caminho_leituras(nome_avaliacao) pasta_processamento = caminho_leitura.parent return pasta_processamento / "pendencias_importacao.json" def carregar_pendencias_importacao(nome_avaliacao: str): estado = obter_estado_lote(nome_avaliacao) if isinstance(estado.get("pendencias_importacao"), dict): return dict(estado["pendencias_importacao"]) caminho = caminho_pendencias_importacao(nome_avaliacao) if not caminho.exists(): return {} with open(caminho, "r", encoding="utf-8") as f: return json.load(f) def salvar_pendencias_importacao(nome_avaliacao: str, dados: dict): estado = obter_estado_lote(nome_avaliacao) estado["pendencias_importacao"] = dict(dados or {}) estado["validacao_cadastral_desatualizada"] = True salvar_estado_lote(nome_avaliacao, estado) return None def registrar_pendencia_importacao( nome_avaliacao: str, nome_imagem: str, status_validacao: str, motivo: str, detalhes: dict | None = None, ): dados = carregar_pendencias_importacao(nome_avaliacao) dados[nome_imagem] = { "status_validacao": status_validacao, "motivo": str(motivo or "").strip(), "detalhes": detalhes or {}, } salvar_pendencias_importacao(nome_avaliacao, dados) return dados[nome_imagem] def limpar_pendencia_importacao(nome_avaliacao: str, nome_imagem: str): dados = carregar_pendencias_importacao(nome_avaliacao) if nome_imagem in dados: dados.pop(nome_imagem, None) salvar_pendencias_importacao(nome_avaliacao, dados) return True return False def caminho_envios_keepedu(nome_avaliacao: str): caminho_leitura = caminho_leituras(nome_avaliacao) pasta_processamento = caminho_leitura.parent return pasta_processamento / "envios_keepedu.json" def _migrar_envios_do_relatorio_importacao(nome_avaliacao: str) -> dict: from app.keepedu_importacao import caminho_relatorio_importacao caminho_relatorio = caminho_relatorio_importacao(nome_avaliacao, modo="importacao") if not caminho_relatorio.exists(): return {} with open(caminho_relatorio, "r", encoding="utf-8") as arquivo: relatorio = json.load(arquivo) envios: dict = {} for item in relatorio.get("detalhes") or []: if not isinstance(item, dict) or not item.get("sucesso"): continue nome_imagem = str(item.get("imagem") or "").strip() if not nome_imagem: continue envios[nome_imagem] = { "vetores": True, "imagens": False, "origem": "importacao_legada", "id_aval": str(relatorio.get("idAval") or "").strip(), "enviado_em": "", } return envios def _mesclar_registros_envio_keepedu(*fontes: dict) -> dict: """Combina fontes persistidas sem perder um envio parcial já confirmado. Os formatos antigos podiam salvar vetores e imagens em chaves equivalentes diferentes. Aqui ainda preservamos as chaves originais; a consolidação por nome equivalente acontece em ``resolver_envio_keepedu``. """ mesclados: dict = {} for fonte in fontes: if not isinstance(fonte, dict): continue for chave, registro in fonte.items(): chave_limpa = str(chave or "").strip() if not chave_limpa or not isinstance(registro, dict): continue anterior = dict(mesclados.get(chave_limpa) or {}) combinado = {**anterior, **registro} combinado["vetores"] = bool(anterior.get("vetores") or registro.get("vetores")) combinado["imagens"] = bool(anterior.get("imagens") or registro.get("imagens")) mesclados[chave_limpa] = combinado return mesclados def carregar_envios_keepedu(nome_avaliacao: str, estado_lote: dict | None = None) -> dict: estado = ( dict(estado_lote) if isinstance(estado_lote, dict) else obter_estado_lote(nome_avaliacao) ) envios_no_estado = estado.get("envios_keepedu") caminho = caminho_envios_keepedu(nome_avaliacao) envios_legados = {} if caminho.exists(): try: with open(caminho, "r", encoding="utf-8") as arquivo: dados = json.load(arquivo) if isinstance(dados, dict): envios_legados = dados except (OSError, json.JSONDecodeError): envios_legados = {} try: envios_migrados = _migrar_envios_do_relatorio_importacao(nome_avaliacao) except (OSError, json.JSONDecodeError): envios_migrados = {} # O estado consolidado é a fonte mais nova, mas não pode apagar confirmações # existentes nas fontes antigas. Os booleanos são combinados por OR. dados_unificados = _mesclar_registros_envio_keepedu( envios_legados, envios_migrados, envios_no_estado if isinstance(envios_no_estado, dict) else {}, ) if dados_unificados != (envios_no_estado if isinstance(envios_no_estado, dict) else {}): salvar_envios_keepedu(nome_avaliacao, dados_unificados) return dados_unificados def salvar_envios_keepedu(nome_avaliacao: str, dados: dict): estado = obter_estado_lote(nome_avaliacao) estado["envios_keepedu"] = dict(dados or {}) salvar_estado_lote(nome_avaliacao, estado) return None def registrar_envio_keepedu( nome_avaliacao: str, nome_imagem: str, *, vetores: bool = False, imagens: bool = False, origem: str = "validacao", id_aval=None, ) -> dict: nome_imagem = str(nome_imagem or "").strip() if not nome_imagem: return {} dados = carregar_envios_keepedu(nome_avaliacao) atual = dict(dados.get(nome_imagem) or {}) if vetores: atual["vetores"] = True if imagens: atual["imagens"] = True if origem: atual["origem"] = origem if id_aval: atual["id_aval"] = str(id_aval).strip() atual["enviado_em"] = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") dados[nome_imagem] = atual salvar_envios_keepedu(nome_avaliacao, dados) return atual def resolver_envio_keepedu(nome_imagem: str, arquivo_original: str, envios: dict) -> dict | None: """Localiza o envio mesmo quando ele foi salvo com outro nome de arquivo. Processamentos antigos podem registrar ``template_arquivo.jpg`` enquanto a leitura usa ``arquivo.png`` ou a chave interna. O nome-base, sem o prefixo ``template_``, identifica o mesmo cartão. """ if not isinstance(envios, dict) or not envios: return None candidatos = {_stem_arquivo(nome_imagem), _stem_arquivo(arquivo_original)} candidatos.discard("") encontrados: list[dict] = [] for chave, envio in envios.items(): chave_limpa = str(chave or "").strip() if not isinstance(envio, dict): continue if chave_limpa in {str(nome_imagem or "").strip(), str(arquivo_original or "").strip()}: encontrados.append(envio) elif _stem_arquivo(chave_limpa) in candidatos: encontrados.append(envio) if not encontrados: return None combinado = _mesclar_registros_envio_keepedu( {str(indice): registro for indice, registro in enumerate(encontrados)} ) registros = list(combinado.values()) resultado: dict = {} for registro in registros: anterior = dict(resultado) resultado.update(registro) resultado["vetores"] = bool(anterior.get("vetores") or registro.get("vetores")) resultado["imagens"] = bool(anterior.get("imagens") or registro.get("imagens")) return resultado def rotulo_envio_keepedu(envio: dict | None) -> str: if not envio: return "" vetores = bool(envio.get("vetores")) imagens = bool(envio.get("imagens")) if vetores and imagens: return "Enviado ao KeepEdu" if vetores: return "Vetores no KeepEdu" if imagens: return "Imagem no KeepEdu" return "" def envio_keepedu_completo(envio: dict | None) -> bool: return bool(envio and envio.get("vetores") and envio.get("imagens")) def cartao_enviado_keepedu_completo( nome_avaliacao: str, nome_imagem: str, dados_cartao: dict | None = None, ) -> bool: leituras = carregar_leituras(nome_avaliacao) chave = resolver_chave_leitura(leituras, nome_imagem) dados = dados_cartao if isinstance(dados_cartao, dict) else leituras.get(chave, {}) or {} arquivo_original = str(dados.get("arquivo_original") or "").strip() envio = resolver_envio_keepedu(chave or nome_imagem, arquivo_original, carregar_envios_keepedu(nome_avaliacao)) return envio_keepedu_completo(envio) def resumo_status_lote_keepedu(nome_avaliacao: str) -> dict: """ Status do lote na página de processamento. Só considera finalizado quando todos os cartões tiverem vetores e imagem na KeepEdu. """ leituras = carregar_leituras(nome_avaliacao) if not isinstance(leituras, dict): leituras = {} envios = carregar_envios_keepedu(nome_avaliacao) total = 0 completos = 0 parciais = 0 enviados_vetores = 0 enviados_imagens = 0 cartoes_vistos: set[str] = set() for chave, dados_cartao in leituras.items(): if not isinstance(dados_cartao, dict): continue arquivo = str(dados_cartao.get("arquivo_original") or chave or "").strip() identidade = _stem_arquivo(arquivo) or _stem_arquivo(chave) if identidade and identidade in cartoes_vistos: continue if identidade: cartoes_vistos.add(identidade) total += 1 envio = resolver_envio_keepedu(chave, arquivo, envios) if envio and envio.get("vetores"): enviados_vetores += 1 if envio and envio.get("imagens"): enviados_imagens += 1 if envio_keepedu_completo(envio): completos += 1 elif envio and (envio.get("vetores") or envio.get("imagens")): parciais += 1 base = { "total_cartoes": total, "enviados_completos": completos, "enviados_parciais": parciais, "enviados_vetores": enviados_vetores, "enviados_imagens": enviados_imagens, "faltam_vetores": max(total - enviados_vetores, 0), "faltam_imagens": max(total - enviados_imagens, 0), "finalizado_keepedu": False, } if total == 0: return { **base, "titulo": "Processado", "descricao": "Nenhum cartão registrado neste lote.", } if completos >= total: return { **base, "titulo": "Finalizado", "descricao": "Lote enviado ao KeepEdu.", "finalizado_keepedu": True, } if completos > 0 or parciais > 0: return { **base, "titulo": "Em envio", "descricao": f"{completos} de {total} cartões enviados à KeepEdu.", } return { **base, "titulo": "Processado", "descricao": "Pronto para revisão e validação.", } def _aplicar_envio_keepedu(resultado: dict, envio: dict | None): if not envio: return resultado resultado["envio_keepedu"] = envio resultado["rotulo_envio_keepedu"] = rotulo_envio_keepedu(envio) resultado["enviado_keepedu_completo"] = envio_keepedu_completo(envio) return resultado def _stem_arquivo(nome: str) -> str: from pathlib import Path texto = str(nome or "").strip() if not texto: return "" if texto.lower().startswith("template_"): texto = texto[9:] return Path(texto).stem.lower() def buscar_registro_manual( validacoes_manuais: dict, nome_imagem: str, dados_cartao: dict | None = None, ) -> dict | None: if not isinstance(validacoes_manuais, dict) or not validacoes_manuais: return None if nome_imagem in validacoes_manuais: return validacoes_manuais[nome_imagem] dados_cartao = dados_cartao if isinstance(dados_cartao, dict) else {} arquivo_original = str(dados_cartao.get("arquivo_original") or "").strip() if arquivo_original and arquivo_original in validacoes_manuais: return validacoes_manuais[arquivo_original] stems_alvo = {_stem_arquivo(nome_imagem)} if arquivo_original: stems_alvo.add(_stem_arquivo(arquivo_original)) for chave, registro in validacoes_manuais.items(): if _stem_arquivo(chave) in stems_alvo: return registro return None def resolver_validacao_cartao( validacoes: dict, validacoes_manuais: dict, nome_imagem: str, dados_cartao: dict, ) -> dict: validacao = dict(validacoes.get(nome_imagem) or {}) stem_alvo = _stem_arquivo(nome_imagem) if not str(validacao.get("id_final") or "").strip() and not str(validacao.get("codigo_barras_final") or "").strip(): for chave, entrada in validacoes.items(): if _stem_arquivo(chave) == stem_alvo: validacao = dict(entrada) break manual = buscar_registro_manual(validacoes_manuais, nome_imagem, dados_cartao) if manual: validacao["id_final"] = manual.get("id_final", "") validacao["codigo_barras_final"] = manual.get("codigo_barras_final", "") validacao["origem_id"] = manual.get("origem_id", "VALIDACAO_MANUAL") validacao["status_validacao"] = manual.get("status_validacao", "VALIDADO_MANUAL") validacao["motivo"] = manual.get("motivo", "ID informado manualmente.") validacao["precisa_validacao_manual"] = False return validacao def carregar_leituras(nome_avaliacao: str): return carregar_leituras_lote(nome_avaliacao) def salvar_validacao(nome_avaliacao: str, dados_validacao: dict): estado = obter_estado_lote(nome_avaliacao) estado["validacao_cadastral"] = dict(dados_validacao or {}) estado["validacao_cadastral_desatualizada"] = False salvar_estado_lote(nome_avaliacao, estado) return None def _recalcular_resumo_validacao(dados_validacao: dict) -> None: validacoes = dados_validacao.get("validacoes") or {} resumo = dados_validacao.get("resumo") if not isinstance(resumo, dict): return resumo["total"] = len(validacoes) resumo["validados"] = sum(not bool(item.get("precisa_validacao_manual")) for item in validacoes.values() if isinstance(item, dict)) resumo["pendentes"] = resumo["total"] - resumo["validados"] for campo, status in ( ("com_codigo_barras", "VALIDADO_CODIGO_BARRAS"), ("com_id_api", "VALIDADO_API_RA"), ("sem_identificacao", "PENDENTE_SEM_IDENTIFICACAO"), ("sem_id_prova", "PENDENTE_SEM_ID_PROVA"), ("fora_avaliacao", "PENDENTE_ALUNO_FORA_AVALIACAO"), ): resumo[campo] = sum(item.get("status_validacao") == status for item in validacoes.values() if isinstance(item, dict)) def reconciliar_validacao_com_leituras(dados_validacao: dict, leituras: dict) -> bool: """Remove de um snapshot cartões que não existem mais no lote atual.""" if not isinstance(dados_validacao, dict) or not isinstance(dados_validacao.get("validacoes"), dict): return False stems_leituras = { _stem_arquivo(chave) or _stem_arquivo((dados or {}).get("arquivo_original", "")) for chave, dados in (leituras or {}).items() if isinstance(dados, dict) } stems_leituras.discard("") validacoes = dados_validacao["validacoes"] removidas = False for chave, registro in list(validacoes.items()): stem = _stem_arquivo(chave) if not stem and isinstance(registro, dict): stem = _stem_arquivo(registro.get("arquivo_original", "")) if stem not in stems_leituras: validacoes.pop(chave, None) removidas = True if not removidas: return False _recalcular_resumo_validacao(dados_validacao) return True def validacao_cadastral_corresponde_as_leituras( dados_validacao: dict | None, leituras: dict, ) -> bool: """Confere se o snapshot representa exatamente os cartoes atuais do lote.""" if not isinstance(dados_validacao, dict): return False validacoes = dados_validacao.get("validacoes") if not isinstance(validacoes, dict): return False def stems(registros: dict) -> set[str]: resultado: set[str] = set() for chave, registro in registros.items(): stem = _stem_arquivo(chave) if not stem and isinstance(registro, dict): stem = _stem_arquivo(registro.get("arquivo_original", "")) if stem: resultado.add(stem) return resultado return stems(validacoes) == stems(leituras or {}) def carregar_validacao_cadastral_para_exibicao( nome_avaliacao: str, leituras: dict | None = None, ) -> dict: """Reutiliza o snapshot valido e o reconstroi apenas apos alteracoes.""" estado = obter_estado_lote(nome_avaliacao) dados_salvos = carregar_validacao_cadastral_salva( nome_avaliacao, estado_lote=estado, ) desatualizada = bool(estado.get("validacao_cadastral_desatualizada")) if ( not desatualizada and isinstance(dados_salvos, dict) and isinstance(dados_salvos.get("validacoes"), dict) and ( not isinstance(leituras, dict) or validacao_cadastral_corresponde_as_leituras(dados_salvos, leituras) ) ): return dados_salvos leituras_atuais = leituras if isinstance(leituras, dict) else carregar_leituras(nome_avaliacao) if not leituras_atuais: return {} return gerar_validacao_cadastral( nome_avaliacao, persistir=True, leituras=leituras_atuais, consultar_nomes_opcionais=False, ) def remover_registros_cartao_excluido(nome_avaliacao: str, nomes_arquivo: list[str]) -> bool: """Remove do estado de validação os dados de uma leitura já excluída. A tela de validação abre o snapshot persistido em ``validacao_cadastral``. Portanto, apenas invalidar o cache não basta: o cartão removido voltaria a aparecer no próximo carregamento da página. """ stems_alvo = {_stem_arquivo(nome) for nome in nomes_arquivo if str(nome or "").strip()} if not stems_alvo: return False def pertence_ao_cartao(chave: str, registro: dict | None = None) -> bool: if _stem_arquivo(chave) in stems_alvo: return True if isinstance(registro, dict): return _stem_arquivo(registro.get("arquivo_original", "")) in stems_alvo return False estado = obter_estado_lote(nome_avaliacao) alterado = False snapshot = estado.get("validacao_cadastral") if not isinstance(snapshot, dict): caminho_snapshot = caminho_validacao_cadastral(nome_avaliacao) if caminho_snapshot.exists(): dados_legados = carregar_json(caminho_snapshot, {}) if isinstance(dados_legados, dict): snapshot = dados_legados estado["validacao_cadastral"] = snapshot if isinstance(snapshot, dict) and isinstance(snapshot.get("validacoes"), dict): validacoes = snapshot["validacoes"] for chave, registro in list(validacoes.items()): if pertence_ao_cartao(chave, registro): validacoes.pop(chave, None) alterado = True if alterado: _recalcular_resumo_validacao(snapshot) for campo in ("validacao_manual", "pendencias_importacao"): registros = estado.get(campo) if not isinstance(registros, dict): continue for chave, registro in list(registros.items()): if pertence_ao_cartao(chave, registro): registros.pop(chave, None) alterado = True if alterado: salvar_estado_lote(nome_avaliacao, estado) invalidar_validacao_cadastral_em_cache(nome_avaliacao) return alterado def caminho_validacao_cadastral(nome_avaliacao: str): caminho_leitura = caminho_leituras(nome_avaliacao) pasta_processamento = caminho_leitura.parent return pasta_processamento / "validacao_cadastral.json" def _hidratar_validacao_cadastral(snapshot: dict, envios: dict) -> dict: """Reconstroi campos de exibicao que nao precisam ocupar espaco no banco.""" dados = { chave: dict(valor) if isinstance(valor, dict) else valor for chave, valor in snapshot.items() if chave != "validacoes" } validacoes_hidratadas = {} for nome_imagem, registro in (snapshot.get("validacoes") or {}).items(): if not isinstance(registro, dict): continue validacao = dict(registro) validacao.setdefault("imagem", nome_imagem) # Snapshots antigos/compactados podem omitir a chave; o template acessa # aluno_api.nome e o Jinja quebra se a chave não existir. validacao.setdefault("aluno_api", None) for campo_derivado in ( "envio_keepedu", "rotulo_envio_keepedu", "enviado_keepedu_completo", ): validacao.pop(campo_derivado, None) adicionar_rotulos_exibicao(validacao) envio = resolver_envio_keepedu( nome_imagem, str(validacao.get("arquivo_original") or ""), envios, ) validacoes_hidratadas[nome_imagem] = _aplicar_envio_keepedu( validacao, dict(envio) if envio else None, ) dados["validacoes"] = validacoes_hidratadas return dados def carregar_validacao_cadastral_salva(nome_avaliacao: str, estado_lote: dict | None = None): estado = ( dict(estado_lote) if isinstance(estado_lote, dict) else obter_estado_lote(nome_avaliacao) ) if isinstance(estado.get("validacao_cadastral"), dict): dados = estado["validacao_cadastral"] if not isinstance(dados.get("validacoes"), dict): return dict(dados) envios = carregar_envios_keepedu(nome_avaliacao, estado_lote=estado) return _hidratar_validacao_cadastral(dados, envios) caminho = caminho_validacao_cadastral(nome_avaliacao) if not caminho.exists(): return None dados = carregar_json(caminho, {}) if isinstance(dados, dict) and isinstance(dados.get("validacoes"), dict): envios = carregar_envios_keepedu(nome_avaliacao) return _hidratar_validacao_cadastral(dados, envios) return dados def obter_validacao_cadastral_atual(nome_avaliacao: str) -> dict: """ Fonte primária da validação passa a ser o cálculo atual do lote. O arquivo salvo fica apenas como compatibilidade/fallback implícito. """ em_cache = obter_validacao_cadastral_em_cache(nome_avaliacao) if em_cache is not None: return em_cache return gerar_validacao_cadastral(nome_avaliacao, persistir=False) def obter_validacao_cadastral_para_envio(nome_avaliacao: str) -> dict: """Validação cadastral atualizada, incluindo correções manuais recentes.""" return obter_validacao_cadastral_atual(nome_avaliacao) def salvar_correcao_manual( nome_avaliacao: str = "", nome_imagem: str = "", valor_manual: str = "", nome_processamento: str | None = None, ): if nome_processamento and not nome_avaliacao: nome_avaliacao = nome_processamento valor_manual = str(valor_manual or "").strip() if not valor_manual: raise ValueError("Informe o ID do aluno ou o código completo.") leituras = carregar_leituras(nome_avaliacao) chave_leitura = resolver_chave_leitura(leituras, nome_imagem) dados_cartao = leituras.get(chave_leitura, {}) if cartao_enviado_keepedu_completo(nome_avaliacao, chave_leitura or nome_imagem, dados_cartao): raise ValueError( "Este cartão já foi enviado ao KeepEdu. Para alterá-lo, faça um novo processamento." ) id_prova_processamento = resolver_id_prova_processamento(nome_avaliacao, leituras) if "A" in valor_manual: id_prova, id_aluno = quebrar_codigo_barras(valor_manual) if not id_prova or not id_aluno: raise ValueError( "Código de barras inválido. Use o formato IDPROVAAIDALUNO. Exemplo: 12347A38121" ) codigo_barras_final = valor_manual id_final = id_aluno else: id_final = valor_manual if not id_prova_processamento: raise ValueError( "Não foi possível identificar o ID da prova na avaliação. " "Informe o código completo no formato 12347A38121." ) codigo_barras_final = f"{id_prova_processamento}A{id_final}" aluno_api = buscar_aluno_por_id( id_final, extrair_ra_possivel(dados_cartao), ) dados = carregar_validacao_manual(nome_avaliacao) registro_manual = { "id_final": id_final, "codigo_barras_final": codigo_barras_final, "origem_id": "VALIDACAO_MANUAL", "status_validacao": "VALIDADO_MANUAL", "motivo": "ID informado manualmente na Validação Cadastral.", "aluno_api": aluno_api if aluno_api.get("encontrado") else None, } dados[chave_leitura] = registro_manual if chave_leitura != nome_imagem: dados[nome_imagem] = registro_manual arquivo_original = str(dados_cartao.get("arquivo_original") or "").strip() if arquivo_original and arquivo_original not in {chave_leitura, nome_imagem}: dados[arquivo_original] = registro_manual salvar_validacao_manual(nome_avaliacao, dados) limpar_pendencia_importacao(nome_avaliacao, chave_leitura) if chave_leitura != nome_imagem: limpar_pendencia_importacao(nome_avaliacao, nome_imagem) # Atualiza também o snapshot persistido consumido ao trocar de página. # Sem isso, o ID manual ficava somente no cache da página atual. gerar_validacao_cadastral(nome_avaliacao, persistir=True) return registro_manual def remover_correcao_manual( nome_avaliacao: str = "", nome_imagem: str = "", nome_processamento: str | None = None, ): if nome_processamento and not nome_avaliacao: nome_avaliacao = nome_processamento leituras = carregar_leituras(nome_avaliacao) chave_leitura = resolver_chave_leitura(leituras, nome_imagem) dados_cartao = leituras.get(chave_leitura, {}) if cartao_enviado_keepedu_completo(nome_avaliacao, chave_leitura or nome_imagem, dados_cartao): raise ValueError( "Este cartão já foi enviado ao KeepEdu. Para alterá-lo, faça um novo processamento." ) dados = carregar_validacao_manual(nome_avaliacao) stems_alvo = {_stem_arquivo(chave_leitura), _stem_arquivo(nome_imagem)} arquivo_original = str(dados_cartao.get("arquivo_original") or "").strip() if arquivo_original: stems_alvo.add(_stem_arquivo(arquivo_original)) chaves_remover = [ chave for chave in list(dados.keys()) if _stem_arquivo(chave) in stems_alvo ] if not chaves_remover: raise ValueError("Não há correção manual para desfazer neste cartão.") for chave in chaves_remover: dados.pop(chave, None) salvar_validacao_manual(nome_avaliacao, dados) limpar_pendencia_importacao(nome_avaliacao, chave_leitura) if chave_leitura != nome_imagem: limpar_pendencia_importacao(nome_avaliacao, nome_imagem) dados_validacao = gerar_validacao_cadastral(nome_avaliacao, persistir=True) validacao_atual = dados_validacao.get("validacoes", {}).get(chave_leitura, {}) return { "removido": True, "validacao": validacao_atual, } def extrair_codigo_barras_possivel(dados_cartao: dict): possiveis_campos = [ "codigo_barras", "codigo_de_barras", "barcode", "qrcode", "codigo_barras_qrcode", "Código de Barras/QRCode001", "Codigo de Barras/QRCode001", ] for campo in possiveis_campos: valor = dados_cartao.get(campo) if valor: return str(valor).strip() respostas = dados_cartao.get("respostas", {}) if isinstance(respostas, dict): for chave, valor in respostas.items(): chave_normalizada = str(chave or "").lower() if "codigo" in chave_normalizada and "barras" in chave_normalizada and valor: return str(valor).strip() return "" def extrair_id_prova_do_codigo(codigo_barras: str) -> str: """ Extrai o ID da avaliação (parte antes do 'A') do código de barras. Ex.: 12345A38121 -> 12345 | 12480 -> 12480 """ codigo = str(codigo_barras or "").strip().upper() if not codigo: return "" if "A" in codigo: return codigo.split("A", 1)[0].strip() if codigo.isdigit(): return codigo return "" def quebrar_codigo_barras(codigo_barras: str): """ Formato esperado: ID_PROVA + A + ID_ALUNO Exemplo: 12345A38121 """ codigo_barras = str(codigo_barras or "").strip().upper() if "A" not in codigo_barras: id_prova = extrair_id_prova_do_codigo(codigo_barras) return id_prova, "" partes = codigo_barras.split("A", 1) id_prova = partes[0].strip() id_aluno = partes[1].strip() return id_prova, id_aluno def extrair_ra_possivel(dados_cartao: dict): """ Tenta recuperar o RA lido no cartão. Primeiro procura campos diretos. Se não encontrar, monta o RA pelas Perguntas001 a 006, que correspondem aos dígitos do Registro Acadêmico. """ possiveis_campos = [ "registro_academico", "ra", "RA", "codigo_ra", "id_aluno_ra" ] for campo in possiveis_campos: valor = dados_cartao.get(campo) if valor: ra = "".join(ch for ch in str(valor).strip() if ch.isdigit()) if len(ra) >= 5 and ra != "000000": return ra respostas = dados_cartao.get("respostas", {}) if isinstance(respostas, dict): digitos = [] for numero in range(1, 7): chave = f"Pergunta{numero:03d}" valor = respostas.get(chave, "") valor = str(valor or "").strip() if valor.isdigit(): digitos.append(valor) else: digitos.append("") ra_montado = "".join(digitos).strip() if len(ra_montado) >= 5 and ra_montado != "000000": return ra_montado return "" def extrair_nome_possivel(dados_cartao: dict): possiveis_campos = [ "nome", "nome_aluno", "aluno", "nome_completo" ] for campo in possiveis_campos: valor = dados_cartao.get(campo) if valor: return str(valor).strip() return "" def detectar_id_prova_do_log(nome_processamento: str) -> str: from app.services import caminho_log import csv caminho = caminho_log(nome_processamento) if not caminho.exists(): return "" with open(caminho, "r", encoding="utf-8-sig", newline="") as f: leitor = csv.DictReader(f, delimiter=";") for linha in leitor: id_prova = extrair_id_prova_do_codigo(linha.get("codigo_barras", "")) if id_prova: return id_prova return "" def detectar_id_prova_do_processamento(leituras: dict, nome_processamento: str = ""): """ Procura em qualquer cartão processado um código de barras válido para descobrir o ID da prova daquela avaliação (parte antes do 'A'). """ for dados_cartao in leituras.values(): codigo_barras = extrair_codigo_barras_possivel(dados_cartao) # Valores numericos soltos podem ser falsos positivos do decodificador # de barras. Eles nao sao suficientes para definir a avaliacao do # lote: o cartao nominal sempre usa ID_AVALIACAOAID_ALUNO. if not re.fullmatch(r"\d+A\d+", str(codigo_barras or "").strip().upper()): continue id_prova = extrair_id_prova_do_codigo(codigo_barras) if id_prova: return id_prova id_prova, _ = quebrar_codigo_barras(codigo_barras) if id_prova: return id_prova # Cartoes sem codigo de barras legivel ainda carregam o ID da avaliacao # no prefixo do arquivo original (ex.: 12932_ARS_...). Esse ID permite # vincular o aluno localizado pelo RA sem exigir preenchimento manual. for chave, dados_cartao in leituras.items(): if not isinstance(dados_cartao, dict): continue nome_arquivo = str(dados_cartao.get("arquivo_original") or chave or "").strip() stem = Path(nome_arquivo).stem if stem.lower().startswith("template_"): stem = stem[len("template_"):] match = re.match(r"^(\d+)_", stem) if match: return match.group(1) if nome_processamento: return detectar_id_prova_do_log(nome_processamento) return "" def resolver_id_prova_processamento(nome_avaliacao: str, leituras: dict | None = None) -> str: """ ID da prova usado na validação cadastral. Prioriza o ID configurado no painel/meta do lote sobre o impresso no cartão. """ from app.avaliacao_contexto import resolver_id_avaliacao_esperado_lote leituras_atual = leituras if isinstance(leituras, dict) else carregar_leituras(nome_avaliacao) id_configurado = resolver_id_avaliacao_esperado_lote(nome_avaliacao, leituras_atual) if id_configurado: return id_configurado return detectar_id_prova_do_processamento(leituras_atual, nome_avaliacao) def _aplicar_pendencia_importacao(resultado: dict, pendencia_importacao: dict | None): if not pendencia_importacao: return resultado status_validacao = str(pendencia_importacao.get("status_validacao") or "").strip() motivo = str(pendencia_importacao.get("motivo") or "").strip() if not status_validacao and not motivo: return resultado if status_validacao: resultado["status_validacao"] = status_validacao if motivo: resultado["motivo"] = motivo resultado["pendencia_importacao"] = pendencia_importacao resultado["precisa_validacao_manual"] = True return resultado def validar_cartao( nome_imagem: str, dados_cartao: dict, id_prova_avaliacao: str, validacoes_manuais: dict, pendencias_importacao: dict, *, consultar_nome_opcional: bool = False, ): codigo_barras_detectado = extrair_codigo_barras_possivel(dados_cartao) id_prova_detectado, id_aluno_detectado = quebrar_codigo_barras(codigo_barras_detectado) ra_detectado = extrair_ra_possivel(dados_cartao) nome_detectado = extrair_nome_possivel(dados_cartao) resultado = { "imagem": nome_imagem, "arquivo_original": dados_cartao.get("arquivo_original", nome_imagem), "codigo_barras_detectado": codigo_barras_detectado, "id_prova_detectado": id_prova_detectado, "id_aluno_detectado": id_aluno_detectado, "ra_detectado": ra_detectado, "nome_detectado": nome_detectado, "id_final": "", "codigo_barras_final": "", "origem_id": "", "status_validacao": "", "motivo": "", "aluno_api": None, "pendencia_importacao": None, "precisa_validacao_manual": False } manual = buscar_registro_manual(validacoes_manuais, nome_imagem, dados_cartao) pendencia_importacao = pendencias_importacao.get(nome_imagem) if not pendencia_importacao: arquivo_original = dados_cartao.get("arquivo_original", "") pendencia_importacao = pendencias_importacao.get(arquivo_original) if not pendencia_importacao: for chave, valor in pendencias_importacao.items(): if _stem_arquivo(chave) == _stem_arquivo(nome_imagem): pendencia_importacao = valor break if manual: resultado["id_final"] = manual.get("id_final", "") resultado["codigo_barras_final"] = manual.get("codigo_barras_final", "") resultado["origem_id"] = "VALIDACAO_MANUAL" resultado["status_validacao"] = "VALIDADO_MANUAL" resultado["motivo"] = manual.get("motivo", "ID informado manualmente.") aluno_manual = manual.get("aluno_api") if isinstance(aluno_manual, dict) and aluno_manual.get("encontrado"): resultado["aluno_api"] = aluno_manual elif consultar_nome_opcional and ra_detectado: aluno = buscar_aluno_por_ra(ra_detectado) if aluno.get("encontrado"): resultado["aluno_api"] = aluno resultado["precisa_validacao_manual"] = False return _aplicar_pendencia_importacao(resultado, pendencia_importacao) # 1. Se o cartão já tem código de barras válido, ele é a fonte principal. if codigo_barras_detectado and id_prova_detectado and id_aluno_detectado: resultado["id_final"] = id_aluno_detectado # O codigo de barras ja valida o ID. A consulta pelo RA, usada somente # para exibir o nome, nao deve bloquear a geracao das paginas. if consultar_nome_opcional and ra_detectado: aluno = buscar_aluno_por_ra(ra_detectado) if aluno.get("encontrado"): resultado["aluno_api"] = aluno if ( id_prova_avaliacao and str(id_prova_detectado).strip() != str(id_prova_avaliacao).strip() ): resultado["codigo_barras_final"] = f"{id_prova_avaliacao}A{id_aluno_detectado}" resultado["origem_id"] = "CODIGO_BARRAS_CARTAO" resultado["status_validacao"] = "VALIDADO_CODIGO_BARRAS" resultado["motivo"] = ( "Código de barras lido no cartão e remapeado para o ID da avaliação configurada no lote." ) else: resultado["codigo_barras_final"] = codigo_barras_detectado resultado["origem_id"] = "CODIGO_BARRAS_CARTAO" resultado["status_validacao"] = "VALIDADO_CODIGO_BARRAS" resultado["motivo"] = "Código de barras lido diretamente do cartão." resultado["precisa_validacao_manual"] = False return _aplicar_pendencia_importacao(resultado, pendencia_importacao) # 2. Se não tem código de barras, mas tem RA, busca o ID do aluno pela API. if ra_detectado: aluno = buscar_aluno_por_ra(ra_detectado) if aluno.get("encontrado") and aluno.get("id"): id_final = str(aluno.get("id")) if not id_prova_avaliacao: resultado["id_final"] = id_final resultado["origem_id"] = "API_RA" resultado["status_validacao"] = "PENDENTE_SEM_ID_PROVA" resultado["motivo"] = ( "ID do aluno localizado pela API, mas não foi possível identificar " "o ID da prova em nenhum código de barras do processamento." ) resultado["aluno_api"] = aluno resultado["precisa_validacao_manual"] = True return _aplicar_pendencia_importacao(resultado, pendencia_importacao) resultado["id_final"] = id_final resultado["codigo_barras_final"] = f"{id_prova_avaliacao}A{id_final}" resultado["origem_id"] = "API_RA" resultado["status_validacao"] = "VALIDADO_API_RA" resultado["motivo"] = "ID do aluno localizado pela API a partir do RA e código final montado com o ID da avaliação configurada no lote." resultado["aluno_api"] = aluno resultado["precisa_validacao_manual"] = False return _aplicar_pendencia_importacao(resultado, pendencia_importacao) resultado["status_validacao"] = "PENDENTE_RA_NAO_ENCONTRADO" resultado["motivo"] = aluno.get("motivo", "RA informado, mas ID não foi localizado pela API.") resultado["aluno_api"] = aluno resultado["precisa_validacao_manual"] = True return _aplicar_pendencia_importacao(resultado, pendencia_importacao) # 3. Sem código de barras e sem RA. resultado["status_validacao"] = "PENDENTE_SEM_IDENTIFICACAO" resultado["motivo"] = "Cartão sem código de barras válido e sem RA preenchido." resultado["precisa_validacao_manual"] = True return _aplicar_pendencia_importacao(resultado, pendencia_importacao) def gerar_validacao_cadastral( nome_avaliacao: str, *, persistir: bool = False, leituras: dict | None = None, consultar_nomes_opcionais: bool = False, ): leituras = leituras if isinstance(leituras, dict) else carregar_leituras(nome_avaliacao) id_prova_avaliacao = resolver_id_prova_processamento(nome_avaliacao, leituras) validacoes_manuais = carregar_validacao_manual(nome_avaliacao) pendencias_importacao = carregar_pendencias_importacao(nome_avaliacao) envios_keepedu = carregar_envios_keepedu(nome_avaliacao) validacoes = {} resumo = { "total": 0, "validados": 0, "pendentes": 0, "com_codigo_barras": 0, "com_id_api": 0, "sem_identificacao": 0, "sem_id_prova": 0, "fora_avaliacao": 0, "id_prova_processamento": id_prova_avaliacao } for nome_imagem, dados_cartao in leituras.items(): validacao = validar_cartao( nome_imagem, dados_cartao, id_prova_avaliacao, validacoes_manuais, pendencias_importacao, consultar_nome_opcional=consultar_nomes_opcionais, ) adicionar_rotulos_exibicao(validacao) envio = resolver_envio_keepedu( nome_imagem, validacao.get("arquivo_original", ""), envios_keepedu, ) validacoes[nome_imagem] = _aplicar_envio_keepedu(validacao, envio) resumo["total"] += 1 if validacao["precisa_validacao_manual"]: resumo["pendentes"] += 1 else: resumo["validados"] += 1 if validacao["status_validacao"] == "VALIDADO_CODIGO_BARRAS": resumo["com_codigo_barras"] += 1 if validacao["status_validacao"] == "VALIDADO_API_RA": resumo["com_id_api"] += 1 if validacao["status_validacao"] == "PENDENTE_SEM_IDENTIFICACAO": resumo["sem_identificacao"] += 1 if validacao["status_validacao"] == "PENDENTE_SEM_ID_PROVA": resumo["sem_id_prova"] += 1 if validacao["status_validacao"] == "PENDENTE_ALUNO_FORA_AVALIACAO": resumo["fora_avaliacao"] += 1 dados_saida = { "nome_avaliacao": nome_avaliacao, "resumo": resumo, "validacoes": validacoes } if persistir: salvar_validacao(nome_avaliacao, dados_saida) _validacoes_em_memoria[str(nome_avaliacao or "")] = { "assinatura": _assinatura_validacao(nome_avaliacao), "dados": dados_saida, "expira_em": time.monotonic() + _VALIDACAO_CACHE_TTL_SEGUNDOS, } return dados_saida _CAMPOS_CODIGO_BARRAS_CARTAO = [ "codigo_barras", "codigo_de_barras", "barcode", "qrcode", "codigo_barras_qrcode", "Código de Barras/QRCode001", "Codigo de Barras/QRCode001", ] def substituir_id_prova_em_codigo_barras( codigo_barras: str, id_antigo: str, id_novo: str, ) -> str: codigo = str(codigo_barras or "").strip() id_antigo = str(id_antigo or "").strip() id_novo = str(id_novo or "").strip() if not codigo or not id_antigo or not id_novo or id_antigo == id_novo: return codigo id_prova, id_aluno = quebrar_codigo_barras(codigo) if id_prova == id_antigo: if id_aluno: return f"{id_novo}A{id_aluno}" return id_novo return codigo def _atualizar_codigo_barras_em_cartao( dados_cartao: dict, id_antigo: str, id_novo: str, ) -> bool: if not isinstance(dados_cartao, dict): return False alterou = False for campo in _CAMPOS_CODIGO_BARRAS_CARTAO: valor = dados_cartao.get(campo) if not valor: continue novo_valor = substituir_id_prova_em_codigo_barras(str(valor), id_antigo, id_novo) if novo_valor != str(valor).strip(): dados_cartao[campo] = novo_valor alterou = True respostas = dados_cartao.get("respostas", {}) if isinstance(respostas, dict): for chave, valor in list(respostas.items()): chave_normalizada = str(chave or "").lower() if "codigo" not in chave_normalizada or "barras" not in chave_normalizada or not valor: continue novo_valor = substituir_id_prova_em_codigo_barras(str(valor), id_antigo, id_novo) if novo_valor != str(valor).strip(): respostas[chave] = novo_valor alterou = True return alterou def _atualizar_log_id_prova(nome_processamento: str, id_antigo: str, id_novo: str) -> int: from app.services import caminho_log import csv caminho = caminho_log(nome_processamento) if not caminho.exists(): return 0 linhas: list[dict] = [] alteradas = 0 with open(caminho, "r", encoding="utf-8-sig", newline="") as arquivo: leitor = csv.DictReader(arquivo, delimiter=";") campos = list(leitor.fieldnames or []) for linha in leitor: codigo = str(linha.get("codigo_barras") or "").strip() if codigo: novo_codigo = substituir_id_prova_em_codigo_barras(codigo, id_antigo, id_novo) if novo_codigo != codigo: linha["codigo_barras"] = novo_codigo alteradas += 1 linhas.append(linha) if not campos: return 0 with open(caminho, "w", encoding="utf-8-sig", newline="") as arquivo: escritor = csv.DictWriter(arquivo, fieldnames=campos, delimiter=";", lineterminator="\n") escritor.writeheader() escritor.writerows(linhas) return alteradas def _atualizar_importacao_keepedu_id_prova( nome_processamento: str, id_antigo: str, id_novo: str, ) -> bool: from app.services import caminho_leituras caminho = caminho_leituras(nome_processamento).parent / "importacao_keepedu.json" if not caminho.exists(): return False with open(caminho, "r", encoding="utf-8") as arquivo: dados = json.load(arquivo) if not isinstance(dados, dict): return False alterou = False id_atual = str(dados.get("idAval") or "").strip() if id_atual == id_antigo: dados["idAval"] = id_novo alterou = True for item in dados.get("detalhes") or []: if not isinstance(item, dict): continue barcode = str(item.get("barcode") or "").strip() if barcode: novo_barcode = substituir_id_prova_em_codigo_barras(barcode, id_antigo, id_novo) if novo_barcode != barcode: item["barcode"] = novo_barcode alterou = True payload = item.get("payload") if isinstance(payload, dict): id_payload = str(payload.get("idAval") or "").strip() if id_payload == id_antigo: try: payload["idAval"] = int(id_novo) except ValueError: payload["idAval"] = id_novo alterou = True if alterou: with open(caminho, "w", encoding="utf-8") as arquivo: json.dump(dados, arquivo, ensure_ascii=False, indent=4) return alterou def alterar_id_prova_do_lote( nome_processamento: str, id_prova_novo: str, id_prova_antigo: str = "", ) -> dict: """ Remapeia o ID da avaliação em todo o lote (meta, leituras, log, validação e envios). Útil quando os cartões foram impressos com ID de outra prova. """ from app.avaliacao_contexto import ( carregar_meta_avaliacao_do_lote, salvar_meta_avaliacao_no_lote, ) from app.services import caminho_leituras, salvar_json nome_limpo = str(nome_processamento or "").strip() id_novo = str(id_prova_novo or "").strip() if not nome_limpo: raise ValueError("Processamento inválido.") if not id_novo or not id_novo.isdigit(): raise ValueError("Informe o novo ID da avaliação (apenas números).") caminho_leitura = caminho_leituras(nome_limpo) if not caminho_leitura.exists(): raise ValueError("Lote não encontrado ou sem leituras.") leituras = carregar_leituras(nome_limpo) if not leituras: raise ValueError("Este lote não possui cartões processados.") id_antigo = str(id_prova_antigo or "").strip() if not id_antigo: meta = carregar_meta_avaliacao_do_lote(nome_limpo) id_antigo = str(meta.get("id_avaliacao") or meta.get("id") or "").strip() if not id_antigo: id_antigo = str(detectar_id_prova_do_processamento(leituras, nome_limpo) or "").strip() if id_antigo and id_antigo == id_novo: raise ValueError("O novo ID é igual ao ID atual da avaliação.") cartoes_alterados = 0 for dados_cartao in leituras.values(): if _atualizar_codigo_barras_em_cartao(dados_cartao, id_antigo, id_novo): cartoes_alterados += 1 salvar_leituras_lote(nome_limpo, leituras) linhas_log = _atualizar_log_id_prova(nome_limpo, id_antigo, id_novo) validacao_manual = carregar_validacao_manual(nome_limpo) manuais_alterados = 0 for chave, entrada in list(validacao_manual.items()): if not isinstance(entrada, dict): continue codigo_final = str(entrada.get("codigo_barras_final") or "").strip() if not codigo_final: continue novo_codigo = substituir_id_prova_em_codigo_barras(codigo_final, id_antigo, id_novo) if novo_codigo != codigo_final: entrada["codigo_barras_final"] = novo_codigo manuais_alterados += 1 if validacao_manual: salvar_validacao_manual(nome_limpo, validacao_manual) envios = carregar_envios_keepedu(nome_limpo) envios_alterados = 0 ja_enviados_keepedu = False for entrada in envios.values(): if not isinstance(entrada, dict): continue if entrada.get("vetores") or entrada.get("imagens"): ja_enviados_keepedu = True if str(entrada.get("id_aval") or "").strip() == id_antigo: entrada["id_aval"] = id_novo envios_alterados += 1 if envios: salvar_envios_keepedu(nome_limpo, envios) _atualizar_importacao_keepedu_id_prova(nome_limpo, id_antigo, id_novo) meta_anterior = carregar_meta_avaliacao_do_lote(nome_limpo) meta_extra: dict[str, str] = {"id_configurado_no_painel": "true"} if id_antigo and id_antigo != id_novo: meta_extra["id_avaliacao_impresso"] = id_antigo if meta_anterior.get("nome_simulado"): meta_extra.setdefault("nome_simulado", str(meta_anterior.get("nome_simulado") or "")) salvar_meta_avaliacao_no_lote(nome_limpo, id_novo, meta_extra=meta_extra) validacao = gerar_validacao_cadastral(nome_limpo) avisos: list[str] = [] if ja_enviados_keepedu: avisos.append( "Alguns cartões já foram enviados ao KeepEdu com o ID anterior. " "Reenvie vetores e imagens se precisar vinculá-los à nova avaliação." ) return { "status": "ok", "nome_processamento": nome_limpo, "id_prova_anterior": id_antigo, "id_prova_novo": id_novo, "total_cartoes": len(leituras), "cartoes_com_codigo_atualizado": cartoes_alterados, "linhas_log_atualizadas": linhas_log, "validacoes_manuais_atualizadas": manuais_alterados, "envios_keepedu_atualizados": envios_alterados, "resumo_validacao": validacao.get("resumo", {}), "avisos": avisos, }