import os import csv import io import json import re import shutil from datetime import datetime from pathlib import Path from threading import Lock from app.settings import ( PASTA_PROCESSAMENTOS, PASTA_RUNTIME, PASTA_UPLOADS_TEMP, nome_pasta_processamentos_para_escola, ) from app.tenant import get_school from app.time_utils import agora_local, fromtimestamp_local from omr_reader import ( processar_imagens_omr, detectar_respostas_por_template, caminho_modelo_omr_padrao ) from app.progresso import atualizar_job, registrar_evento_job EXTENSOES_IMAGEM = [".jpg", ".jpeg", ".png"] REPROCESSAMENTO_OMR_LOCK = Lock() def _calcular_resumo_leituras(leituras: dict | None) -> dict[str, int]: """Calcula o resumo pequeno que pode ser consultado sem reler os cartoes.""" from app.log_omr_csv import erros_exigem_correcao_manual resumo = { "total_imagens": 0, "total_ok": 0, "total_erro": 0, "total_manual": 0, "total_pendencias": 0, } if not isinstance(leituras, dict) or not leituras: return resumo registros = [dados for dados in leituras.values() if isinstance(dados, dict)] resumo["total_imagens"] = len(registros) resumo["total_erro"] = sum( 1 for dados in registros if erros_exigem_correcao_manual(dados.get("erros") or []) ) resumo["total_pendencias"] = sum( 1 for dados in registros if dados.get("erros") or [] ) resumo["total_manual"] = sum( 1 for dados in registros if str(dados.get("origem_cantos") or "").strip().upper() == "MANUAL" or erros_exigem_correcao_manual(dados.get("erros") or []) ) resumo["total_ok"] = max(resumo["total_imagens"] - resumo["total_erro"], 0) return resumo def pasta_processamentos_escola() -> Path: """Raiz de arquivos OMR da school ativa dentro do repositório do servidor.""" # Com MySQL, a pasta é apenas uma área temporária de execução; o lote # consolidado é salvo no banco e nenhuma pasta `processamentos/` é criada. from app.db import storage_usa_banco pasta = PASTA_PROCESSAMENTOS / nome_pasta_processamentos_para_escola(get_school()) pasta.mkdir(parents=True, exist_ok=True) return pasta def carregar_lote_consolidado(nome_processamento: str) -> dict: """Lê um lote do arquivo único da escola; retorna vazio quando ainda é legado.""" from app.lotes_store import obter_lote return obter_lote(nome_processamento) def carregar_leituras_lote(nome_processamento: str) -> dict: lote = carregar_lote_consolidado(nome_processamento) if lote: return dict(lote.get("leituras") or {}) return carregar_json(caminho_leituras(nome_processamento), {}) def salvar_leituras_lote(nome_processamento: str, leituras: dict, metadados: dict | None = None) -> None: from app.lotes_store import obter_lote, salvar_lote lote = obter_lote(nome_processamento) if lote: estado = dict(lote.get("estado") or {}) # O snapshot da Validacao pode ser reutilizado entre paginas, mas uma # alteracao nas leituras exige uma unica reconstrucao. estado["validacao_cadastral_desatualizada"] = True estado["resumo_processamento"] = _calcular_resumo_leituras(leituras) salvar_lote( nome_processamento, metadados or lote.get("metadados") or {}, leituras, estado=estado, ) else: salvar_json(caminho_leituras(nome_processamento), leituras) # A validacao cadastral e derivada das leituras. Em MySQL os artefatos de # arquivo nao mudam de mtime, portanto a assinatura do cache nao percebe # uma releitura. Invalide explicitamente para que RA/barcode novos sejam # validados no proximo carregamento da tela. from app.validacao_cadastral import invalidar_validacao_cadastral_em_cache invalidar_validacao_cadastral_em_cache(nome_processamento) def carregar_json_bruto(caminho) -> dict: caminho = Path(caminho) if not caminho.exists(): return {} try: with open(caminho, "r", encoding="utf-8") as arquivo: dados = json.load(arquivo) except (OSError, json.JSONDecodeError): return {} return dados if isinstance(dados, dict) else {} def salvar_resultado_omr_em_lote(resultado: dict, meta_extra: dict | None = None) -> str: """Move a leitura temporária do OMR para o JSON plano definitivo do lote.""" pasta_temporaria = Path(str(resultado.get("pasta_processamento") or "")) caminho_leituras_temp = Path(resultado.get("leituras_omr") or pasta_temporaria / "leituras_omr.json") dados = carregar_json_bruto(caminho_leituras_temp) if not isinstance(dados, dict): raise ValueError("A leitura OMR temporária é inválida.") metadados = dict(dados.get("metadados") or {}) nome_processamento = str(metadados.get("id_processamento") or pasta_temporaria.name).strip() metadados.update(dict(meta_extra or {})) metadados.setdefault("criado_em", agora_local().isoformat()) metadados["id_processamento"] = nome_processamento metadados["versao_formato"] = 3 leituras = dados.get("leituras") if isinstance(dados.get("leituras"), dict) else dados # Mantemos os cartões originais para que as prévias e a correção continuem # exibindo o espelho real mesmo após o envio à KeepEdu. pasta_originais = caminho_originais(nome_processamento) for leitura in leituras.values() if isinstance(leituras, dict) else []: if not isinstance(leitura, dict): continue origem = Path(str(leitura.get("caminho_original_processamento") or "")) if not origem.is_file(): continue pasta_originais.mkdir(parents=True, exist_ok=True) destino = pasta_originais / origem.name if origem.resolve() != destino.resolve(): shutil.copy2(origem, destino) leitura["caminho_original_processamento"] = str(destino) from app.lotes_store import salvar_lote salvar_lote( nome_processamento, metadados, leituras, estado={"resumo_processamento": _calcular_resumo_leituras(leituras)}, ) # ``leituras_omr.json`` é apenas a ponte entre o leitor OMR e o JSON # consolidado. Pendências e correções manuais, quando existirem, ficam no # mesmo diretório e não podem ser removidas aqui. pasta_suporte = caminho_pasta_processamento(nome_processamento) if pasta_temporaria == pasta_suporte: caminho_leituras_temp.unlink(missing_ok=True) try: pasta_suporte.rmdir() except OSError: # Há pendências/manual_omr ou outro artefato que ainda é útil. pass # A execução OMR pode gerar JSONs e depuração em disco; no modo MySQL # esses artefatos são temporários e o lote já foi consolidado no banco. from app.db import storage_usa_banco if storage_usa_banco() and pasta_temporaria.is_dir(): shutil.rmtree(pasta_temporaria, ignore_errors=True) return nome_processamento def _obter_data_hora_processamento(nome_pasta: str, pasta: Path) -> datetime: candidatos = [ pasta / "log_leitura_omr.csv", pasta / "leituras_omr.json", pasta ] for candidato in candidatos: if candidato.exists(): try: return fromtimestamp_local(candidato.stat().st_mtime) except OSError: pass match = re.match(r"^processamento_(\d{8})_(\d{6})$", str(nome_pasta 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 _formatar_data_hora_processamento(nome_pasta: str, pasta: Path) -> tuple[str, str]: data_hora = _obter_data_hora_processamento(nome_pasta, pasta) return ( data_hora.strftime("%d/%m/%Y"), data_hora.strftime("%H:%M:%S") ) def _rotulo_data_processamento(nome_pasta: str, pasta: Path) -> str: data_hora = _obter_data_hora_processamento(nome_pasta, pasta) hoje = agora_local().date() data_referencia = data_hora.date() if data_referencia == hoje: return "Hoje" if data_referencia == hoje.fromordinal(hoje.toordinal() - 1): return "Ontem" dias_semana = { 0: "Segunda", 1: "Terça", 2: "Quarta", 3: "Quinta", 4: "Sexta", 5: "Sábado", 6: "Domingo", } return dias_semana.get(data_referencia.weekday(), "Lote") from pathlib import Path import os def listar_processamentos(): """ Lista todos os lotes de cartões-resposta escaneados a partir do diretório configurado pelo aplicativo. """ nomes_lotes = [] pasta_raiz = pasta_processamentos_escola() if pasta_raiz.exists() and pasta_raiz.is_dir(): for subpasta in pasta_raiz.iterdir(): if subpasta.is_dir() and not subpasta.name.startswith("."): nomes_lotes.append(subpasta.name) from app.lotes_store import listar_lotes nomes_lotes.extend(listar_lotes()) return sorted(set(nomes_lotes), reverse=True) def listar_processamentos_recentes(): """ Monta os metadados exibidos na home para os lotes recentes. """ lotes = [] pasta_raiz = pasta_processamentos_escola() if not pasta_raiz.exists() or not pasta_raiz.is_dir(): return lotes for subpasta in pasta_raiz.iterdir(): if not subpasta.is_dir() or subpasta.name.startswith("."): continue data_texto, hora_texto = _formatar_data_hora_processamento(subpasta.name, subpasta) lotes.append( { "nome": subpasta.name, "titulo": subpasta.name.replace("_", " "), "rotulo": _rotulo_data_processamento(subpasta.name, subpasta), "data": data_texto, "hora": hora_texto, "timestamp": _obter_data_hora_processamento(subpasta.name, subpasta), } ) from app.lotes_store import listar_lotes_resumidos for nome, lote in listar_lotes_resumidos(): if any(item["nome"] == nome for item in lotes): continue metadados = lote.get("metadados") or {} data_hora = agora_local() lotes.append({ "nome": nome, "titulo": nome.replace("_", " "), "rotulo": _rotulo_data_processamento(nome, pasta_raiz), "data": str(metadados.get("criado_em") or ""), "hora": "", "timestamp": data_hora, }) lotes.sort(key=lambda item: item["timestamp"], reverse=True) for item in lotes: item.pop("timestamp", None) return lotes def resumo_processamento(nome_processamento: str): # Lotes novos vivem no arquivo consolidado; o caminho legado fica apenas # como compatibilidade dentro de carregar_leituras_lote. leituras = carregar_leituras_lote(nome_processamento) return _calcular_resumo_leituras(leituras) def listar_imagens_debug(nome_processamento: str): pasta_debug = caminho_pasta_processamento(nome_processamento) / "debug_omr" imagens = set() leituras = carregar_leituras_lote(nome_processamento) # A pasta de debug preserva arquivos usados durante a leitura. Ao excluir # um cartao, a leitura deixa de existir, mas esse artefato pode permanecer # na pasta. A galeria deve exibir somente cartoes ainda ativos no lote. stems_leituras = { Path(normalizar_nome_arquivo_debug(nome_imagem)).stem.lower() for nome_imagem in leituras } if pasta_debug.exists(): for arquivo in pasta_debug.iterdir(): if ( arquivo.suffix.lower() in EXTENSOES_IMAGEM and arquivo.name.startswith("template_") and arquivo.stem.lower() in stems_leituras ): imagens.add(arquivo.name) for nome_imagem in leituras.keys(): if str(nome_imagem).startswith("template_"): imagens.add(str(nome_imagem)) return sorted(imagens) def processar_pasta_local(pasta_origem: str): pasta_origem = pasta_origem.strip().strip('"') if not os.path.exists(pasta_origem): raise FileNotFoundError(f"Pasta não encontrada: {pasta_origem}") if not os.path.isdir(pasta_origem): raise NotADirectoryError(f"O caminho informado não é uma pasta: {pasta_origem}") resultado = processar_imagens_omr( pasta_origem, pasta_saida=str(pasta_processamentos_escola()) ) return salvar_resultado_omr_em_lote(resultado) async def processar_uploads(arquivos): timestamp = agora_local().strftime("%Y%m%d_%H%M%S") pasta_upload = PASTA_UPLOADS_TEMP / f"upload_{timestamp}" pasta_upload.mkdir(parents=True, exist_ok=True) try: for arquivo in arquivos: # Quando o navegador envia uma pasta, o filename pode vir com subpastas. # Aqui pegamos apenas o nome final do arquivo para evitar erro de diretório. nome_arquivo = os.path.basename(arquivo.filename) if not nome_arquivo: continue extensao = os.path.splitext(nome_arquivo)[1].lower() if extensao not in EXTENSOES_IMAGEM: continue destino = pasta_upload / nome_arquivo with open(destino, "wb") as f: conteudo = await arquivo.read() f.write(conteudo) resultado = processar_imagens_omr( str(pasta_upload), pasta_saida=str(pasta_processamentos_escola()) ) return salvar_resultado_omr_em_lote(resultado) finally: shutil.rmtree(pasta_upload, ignore_errors=True) def caminho_imagem_debug(nome_processamento: str, nome_imagem: str): return caminho_pasta_processamento(nome_processamento) / "debug_omr" / nome_imagem def carregar_leitura(nome_avaliacao: str, em_json: bool = True): """ Resolve o arquivo principal do lote a partir do diretório configurado. """ caminho_principal = caminho_leituras(nome_avaliacao) if caminho_principal.exists(): return caminho_principal nome_puro = str(nome_avaliacao or "").replace("processamento_", "") candidatos = [ PASTA_PROCESSAMENTOS / nome_avaliacao / "leituras_omr.json", PASTA_PROCESSAMENTOS / nome_puro / "leituras_omr.json", PASTA_PROCESSAMENTOS / f"{nome_avaliacao}.json", PASTA_PROCESSAMENTOS / f"{nome_puro}.json", ] for candidato in candidatos: if candidato.exists() and candidato.is_file(): return candidato return caminho_principal def caminho_leituras(nome_processamento: str): return pasta_processamentos_escola() / nome_processamento / "leituras_omr.json" def caminho_log(nome_processamento: str): return caminho_pasta_processamento(nome_processamento) / "log_leitura_omr.csv" def gerar_linhas_log_virtuais(nome_processamento: str, leituras: dict | None = None) -> list[dict]: from app.log_omr_csv import montar_linha_log if leituras is None: leituras = carregar_json(caminho_leituras(nome_processamento), {}) if not isinstance(leituras, dict): return [] linhas: list[dict] = [] for chave, leitura in leituras.items(): if not isinstance(leitura, dict): continue arquivo_original = str( leitura.get("arquivo_original") or normalizar_nome_arquivo_debug(str(chave or "")) ).strip() erros = leitura.get("erros") or [] total_perguntas = 0 respostas = leitura.get("respostas") if isinstance(respostas, dict): total_perguntas = sum( 1 for campo in respostas.keys() if str(campo).startswith("Questao") ) total_respostas = total_perguntas origem_cantos = str(leitura.get("origem_cantos") or "").strip().upper() precisa_manual = ( bool(erros) or origem_cantos == "MANUAL" ) linhas.append( montar_linha_log( arquivo=arquivo_original, status="ERRO" if erros else "OK", registro_academico=leitura.get("registro_academico", ""), codigo_barras=leitura.get("codigo_barras", ""), total_perguntas_modelo=total_perguntas, total_respostas_lidas=total_respostas, precisa_correcao_manual="SIM" if precisa_manual else "NAO", detalhe=" | ".join(str(item) for item in erros if item), ) ) return linhas def gerar_csv_log_virtual(nome_processamento: str) -> bytes: from app.log_omr_csv import LOG_OMR_COLUNAS linhas = gerar_linhas_log_virtuais(nome_processamento) buffer = io.StringIO() escritor = csv.DictWriter(buffer, fieldnames=LOG_OMR_COLUNAS, delimiter=";") escritor.writeheader() escritor.writerows(linhas) return buffer.getvalue().encode("utf-8-sig") def caminho_correcoes_web(nome_processamento: str): pasta_manual = caminho_pasta_processamento(nome_processamento) / "manual_omr" return pasta_manual / "correcoes_web.json" def aplicar_correcoes_web_nas_leituras(nome_processamento: str, correcoes: dict) -> int: """Converte cliques manuais em respostas persistidas do cartao. O editor guarda os cliques para auditoria, mas as demais telas leem ``respostas``. Por isso cada clique e associado ao ponto OMR mais proximo. """ leituras = carregar_leituras_lote(nome_processamento) if not isinstance(leituras, dict): return 0 correcoes = correcoes if isinstance(correcoes, dict) else {} alteracoes = 0 for nome_imagem, marcacoes in correcoes.items(): if not isinstance(marcacoes, list): continue chave = resolver_chave_leitura(leituras, str(nome_imagem or "")) cartao = leituras.get(chave) if not isinstance(cartao, dict): continue respostas_atuais = cartao.get("respostas") or {} if not isinstance(respostas_atuais, dict): respostas_atuais = {} # Reconstroi a partir da leitura OMR original a cada salvamento, para # que remover um clique manual tambem remova sua alteracao. if "respostas_omr_original" not in cartao: cartao["respostas_omr_original"] = dict(respostas_atuais) cartao["precisa_correcao_manual_omr"] = cartao.get("precisa_correcao_manual") respostas = dict(cartao.get("respostas_omr_original") or respostas_atuais) for marcacao in marcacoes: if not isinstance(marcacao, dict): continue try: x = float(marcacao.get("x")) y = float(marcacao.get("y")) except (TypeError, ValueError): continue candidato = None distancia_minima = float("inf") for pergunta, alternativas in (cartao.get("pontos_mapeados") or {}).items(): if not isinstance(alternativas, dict): continue for alternativa, ponto in alternativas.items(): if not isinstance(ponto, dict): continue try: dx = float(ponto.get("x")) - x dy = float(ponto.get("y")) - y except (TypeError, ValueError): continue distancia = dx * dx + dy * dy if distancia < distancia_minima: distancia_minima = distancia candidato = (str(pergunta), str(alternativa)) # Os pontos OMR ficam separados por cerca de 55 px nas imagens # atuais; 45 px absorve a imprecisao do clique sem trocar a bolha. if not candidato or distancia_minima > 45 * 45: continue pergunta, alternativa = candidato if str(marcacao.get("cor") or "verde").lower() == "vermelho": if str(respostas.get(pergunta) or "") == alternativa: respostas[pergunta] = "" else: respostas[pergunta] = alternativa cartao["respostas"] = respostas cartao["marcacoes_corrigidas_manualmente"] = bool(marcacoes) if marcacoes: cartao["precisa_correcao_manual"] = "SIM" else: cartao["precisa_correcao_manual"] = cartao.get("precisa_correcao_manual_omr") or "NAO" alteracoes += 1 if alteracoes: salvar_leituras_lote(nome_processamento, leituras) return alteracoes def caminho_pasta_processamento(nome_processamento: str): return pasta_processamentos_escola() / nome_processamento def caminho_originais(nome_processamento: str): return PASTA_RUNTIME / "omr_originais" / str(nome_processamento or "").strip() def caminho_manual_arquivo(nome_processamento: str, nome_arquivo: str): pasta_manual = caminho_pasta_processamento(nome_processamento) / "manual_omr" nome_json = f"{os.path.splitext(nome_arquivo)[0]}.json" return pasta_manual / nome_json def normalizar_nome_arquivo_debug(nome_imagem: str): return nome_imagem[9:] if nome_imagem.startswith("template_") else nome_imagem def carregar_json(caminho, padrao): caminho = Path(caminho) if not caminho.exists(): return padrao with open(caminho, "r", encoding="utf-8") as f: dados = json.load(f) if isinstance(dados, dict) and isinstance(dados.get("leituras"), dict) and "metadados" in dados: return dados["leituras"] return dados def salvar_json(caminho, dados): caminho = Path(caminho) dados_para_salvar = dados if caminho.name == "leituras_omr.json" and caminho.exists(): try: with open(caminho, "r", encoding="utf-8") as arquivo: existente = json.load(arquivo) if isinstance(existente, dict) and isinstance(existente.get("leituras"), dict): dados_para_salvar = {**existente, "leituras": dados} except (OSError, json.JSONDecodeError): pass with open(caminho, "w", encoding="utf-8") as f: json.dump(dados_para_salvar, f, ensure_ascii=False, indent=4) def pontos_cantos_para_lista(pontos_cantos): if not pontos_cantos: return None chaves = ["TOP_LEFT", "TOP_RIGHT", "BOTTOM_LEFT", "BOTTOM_RIGHT"] pontos = [] for chave in chaves: ponto = pontos_cantos.get(chave) if isinstance(pontos_cantos, dict) else None if not ponto: return None pontos.append([float(ponto["x"]), float(ponto["y"])]) return pontos def pontos_cantos_para_dict(pontos_cantos): chaves = ["TOP_LEFT", "TOP_RIGHT", "BOTTOM_LEFT", "BOTTOM_RIGHT"] def extrair_xy(ponto): if isinstance(ponto, dict): return float(ponto["x"]), float(ponto["y"]) if isinstance(ponto, (list, tuple)) and len(ponto) >= 2: return float(ponto[0]), float(ponto[1]) raise TypeError("Formato de ponto inválido") if isinstance(pontos_cantos, dict): saida = {} for chave in chaves: ponto = pontos_cantos.get(chave) if not ponto: continue try: x, y = extrair_xy(ponto) saida[chave] = {"x": x, "y": y} except (TypeError, ValueError, KeyError): continue return saida if isinstance(pontos_cantos, list) and len(pontos_cantos) == 4: saida = {} for chave, ponto in zip(chaves, pontos_cantos): try: x, y = extrair_xy(ponto) saida[chave] = {"x": x, "y": y} except (TypeError, ValueError, KeyError): continue return saida return {} def resolver_chave_leitura(leituras: dict, nome_imagem: str) -> str: if not isinstance(leituras, dict) or not nome_imagem: return nome_imagem if nome_imagem in leituras: return nome_imagem stem_alvo = Path(normalizar_nome_arquivo_debug(nome_imagem)).stem.lower() for chave in leituras: if Path(normalizar_nome_arquivo_debug(chave)).stem.lower() == stem_alvo: return chave return nome_imagem def _coletar_nomes_arquivo_imagem(nome_imagem: str, leitura_atual: dict | None = None) -> list[str]: leitura_atual = leitura_atual if isinstance(leitura_atual, dict) else {} nomes: list[str] = [] def adicionar(valor): if not valor: return texto = os.path.basename(str(valor).strip()) if texto and texto not in nomes: nomes.append(texto) adicionar(nome_imagem) adicionar(normalizar_nome_arquivo_debug(nome_imagem)) adicionar(leitura_atual.get("arquivo_original")) adicionar(leitura_atual.get("nome_arquivo")) for nome in list(nomes): if nome and not nome.lower().startswith("template_"): adicionar(f"template_{nome}") return nomes def _buscar_imagem_por_stem(pasta: Path, nome_arquivo: str): if not pasta.exists() or not pasta.is_dir(): return None stem_alvo = Path(nome_arquivo).stem.lower() for arquivo in pasta.iterdir(): if not arquivo.is_file(): continue if arquivo.suffix.lower() not in EXTENSOES_IMAGEM: continue if arquivo.stem.lower() == stem_alvo: return arquivo return None def localizar_imagem_original(nome_processamento: str, nome_imagem: str, leituras: dict): leitura_atual = leituras.get(nome_imagem, {}) if isinstance(leituras, dict) else {} nomes_arquivo = _coletar_nomes_arquivo_imagem(nome_imagem, leitura_atual) pasta_processamento = caminho_pasta_processamento(nome_processamento) pasta_originais = caminho_originais(nome_processamento) pasta_pendencias = pasta_processamento / "pendencias" pasta_debug = pasta_processamento / "debug_omr" candidatos = [] caminho_salvo = leitura_atual.get("caminho_original_processamento", "") if caminho_salvo: candidatos.append(caminho_salvo) for nome_ref in nomes_arquivo: candidatos.extend([ pasta_originais / nome_ref, pasta_pendencias / nome_ref, pasta_debug / nome_ref, ]) caminho_manual = caminho_manual_arquivo(nome_processamento, nome_ref) if caminho_manual.exists(): dados_manual = carregar_json(caminho_manual, {}) candidatos.extend([ dados_manual.get("caminho_imagem_pendencia", ""), dados_manual.get("caminho_imagem", ""), ]) for candidato in candidatos: if not candidato: continue caminho = candidato if hasattr(candidato, "exists") else Path(str(candidato)) if caminho.exists(): return caminho for nome_ref in nomes_arquivo: for pasta_busca in (pasta_originais, pasta_pendencias, pasta_debug): encontrado = _buscar_imagem_por_stem(pasta_busca, nome_ref) if encontrado: return encontrado return None def atualizar_log_processamento(nome_processamento: str, nome_arquivo: str, resultado: dict, detalhe: str, caminho_manual: str): from app.log_omr_csv import LOG_OMR_COLUNAS, montar_linha_log, normalizar_linha_log_bruta 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=";") linhas = [normalizar_linha_log_bruta(linha) for linha in leitor] from app.log_omr_csv import calcular_precisa_correcao_manual erros = resultado.get("erros", []) or [] atualizacao = montar_linha_log( arquivo=nome_arquivo, status="OK", registro_academico=resultado.get("registro_academico", ""), codigo_barras=resultado.get("codigo_barras", ""), total_perguntas_modelo=resultado.get("total_perguntas_modelo", 0), total_respostas_lidas=resultado.get("total_respostas_lidas", 0), precisa_correcao_manual=calcular_precisa_correcao_manual( "OK", resultado, erros, tem_json_manual=bool(caminho_manual), ), detalhe=detalhe, ) linha_encontrada = False for linha in linhas: if linha.get("arquivo") == nome_arquivo: linha.update(atualizacao) linha_encontrada = True break if not linha_encontrada: linhas.append(atualizacao) with open(caminho, "w", encoding="utf-8-sig", newline="") as f: escritor = csv.DictWriter(f, fieldnames=LOG_OMR_COLUNAS, delimiter=";") escritor.writeheader() escritor.writerows(linhas) def reprocessar_imagem_processamento( nome_processamento: str, nome_imagem: str, pontos_cantos: dict | None = None, usar_cantos_salvos: bool = True, ): """Executa uma releitura OMR por vez, inclusive entre abas e lotes.""" with REPROCESSAMENTO_OMR_LOCK: return _reprocessar_imagem_processamento_serial( nome_processamento=nome_processamento, nome_imagem=nome_imagem, pontos_cantos=pontos_cantos, usar_cantos_salvos=usar_cantos_salvos, ) def _reprocessar_imagem_processamento_serial( nome_processamento: str, nome_imagem: str, pontos_cantos: dict | None = None, usar_cantos_salvos: bool = True, ): leituras = carregar_leituras_lote(nome_processamento) chave_leitura = resolver_chave_leitura(leituras, nome_imagem) from app.validacao_cadastral import cartao_enviado_keepedu_completo if cartao_enviado_keepedu_completo(nome_processamento, chave_leitura or nome_imagem, leituras.get(chave_leitura, {})): raise ValueError("Este cartão já foi enviado ao KeepEdu. Para alterá-lo, faça um novo processamento.") caminho_imagem = localizar_imagem_original(nome_processamento, chave_leitura, leituras) if caminho_imagem is None: raise FileNotFoundError( "Nao foi possivel localizar a imagem original deste processamento. " "Gere um novo processamento para habilitar o reprocessamento por cantos." ) caminho_modelo = caminho_modelo_omr_padrao() if not os.path.exists(caminho_modelo): raise FileNotFoundError(f"Modelo .xtmpl nao encontrado: {caminho_modelo}") leitura_atual = leituras.get(chave_leitura, {}) cantos_enviados = pontos_cantos_para_dict(pontos_cantos or {}) cantos_informados = pontos_cantos is not None and len(cantos_enviados) > 0 if usar_cantos_salvos and len(cantos_enviados) < 4: cantos_salvos = pontos_cantos_para_dict(leitura_atual.get("pontos_cantos") or {}) if len(cantos_salvos) == 4: cantos_enviados = cantos_salvos if len(cantos_enviados) == 4: pontos_override = pontos_cantos_para_lista(cantos_enviados) elif cantos_informados: raise ValueError("Defina os 4 cantos do cartão antes de reprocessar.") else: pontos_override = None resultado, detalhe = detectar_respostas_por_template( str(caminho_imagem), caminho_modelo, None, pontos_imagem_override=pontos_override ) if not resultado: raise ValueError(detalhe) from app.avaliacao_contexto import ( resolver_id_avaliacao_esperado_lote, validar_cartao_para_avaliacao_lote, ) from app.validacao_cadastral import extrair_id_prova_do_codigo from omr_reader import codigo_barras_completo nome_arquivo = normalizar_nome_arquivo_debug(chave_leitura) cantos_finais = cantos_enviados or pontos_cantos_para_dict(resultado.get("pontos_cantos", {})) origem_cantos = "MANUAL" if pontos_override is not None else str(resultado.get("origem_cantos") or "AUTO") id_esperado = resolver_id_avaliacao_esperado_lote(nome_processamento, leituras) codigo_barras_lido = str(resultado.get("codigo_barras") or "") erros_leitura = list(resultado.get("erros") or []) # Sem barcode legivel, a leitura OMR continua valida: RA e respostas # serao vinculados posteriormente pela validacao cadastral via API. if not codigo_barras_completo(codigo_barras_lido): cartao_ok, motivo_bloqueio = True, "" else: cartao_ok, motivo_bloqueio = validar_cartao_para_avaliacao_lote( nome_arquivo, codigo_barras_lido, id_esperado, # Um ID escolhido para o lote define o destino do envio. Nao # remapeamos silenciosamente cartoes impressos para outra prova. permitir_id_impresso_divergente=False, ) if not cartao_ok: erros_leitura = [motivo_bloqueio] detalhe = motivo_bloqueio leitura_atual.update({ "arquivo_original": nome_arquivo, "caminho_original_processamento": str(caminho_imagem), "respostas": {} if not cartao_ok else resultado.get("respostas", {}), "pontos_mapeados": resultado.get("pontos_mapeados", {}), "pontos_cantos": cantos_finais, "origem_cantos": origem_cantos, "registro_academico": resultado.get("registro_academico", ""), "codigo_barras": resultado.get("codigo_barras", ""), "debug_bolhas": resultado.get("debug_bolhas", ""), "erros": erros_leitura, }) # Uma releitura por cantos gera uma nova fonte de verdade para as respostas. # Nao podemos manter o snapshot/correcoes manuais da leitura anterior: a rota # /leituras reaplicaria esses dados antigos e sobrescreveria o resultado que # acabou de ser calculado. leitura_atual.pop("respostas_omr_original", None) leitura_atual.pop("precisa_correcao_manual_omr", None) leitura_atual.pop("marcacoes_corrigidas_manualmente", None) if chave_leitura != nome_imagem and nome_imagem in leituras: leituras.pop(nome_imagem, None) leituras[chave_leitura] = leitura_atual leitura_atual["status_revisao"] = "PENDENTE" salvar_leituras_lote(nome_processamento, leituras) # Remove os cliques manuais que pertenciam ao mapeamento anterior. Eles # seriam desenhados sobre a nova leitura e, pior, alterariam novamente as # respostas ao carregar a tela de correcao. from app.lotes_store import obter_estado_lote, salvar_estado_lote estado = obter_estado_lote(nome_processamento) correcoes_web = estado.get("correcoes_web") or {} if isinstance(correcoes_web, dict): stem_atual = Path(normalizar_nome_arquivo_debug(chave_leitura)).stem.lower() alterou_correcoes = False for chave in list(correcoes_web): if Path(normalizar_nome_arquivo_debug(chave)).stem.lower() == stem_atual: correcoes_web.pop(chave, None) alterou_correcoes = True if alterou_correcoes: estado["correcoes_web"] = correcoes_web salvar_estado_lote(nome_processamento, estado) erros = erros_leitura or [] resultado_log = dict(resultado) resultado_log["erros"] = erros # O estado necessario para revisao ja esta na leitura consolidada acima. # Nao criamos mais arquivos auxiliares dentro de `processamentos`. caminho_manual_str = "" atualizar_log_processamento( nome_processamento, nome_arquivo, resultado_log, detalhe, caminho_manual_str ) return { "status": "ok", "nome_imagem": chave_leitura, "arquivo_original": nome_arquivo, "detalhe": detalhe, "pontos_cantos": cantos_finais, "origem_cantos": origem_cantos, "erros": erros, "bloqueado": not cartao_ok, "resultado": resultado_log } def remover_linha_log_processamento(nome_processamento: str, nomes_arquivo: list[str]): caminho = caminho_log(nome_processamento) if not caminho.exists(): return stems_alvo = {Path(nome).stem.lower() for nome in nomes_arquivo if nome} with open(caminho, "r", encoding="utf-8-sig", newline="") as f: leitor = csv.DictReader(f, delimiter=";") linhas = list(leitor) cabecalho = leitor.fieldnames or [] if not cabecalho: return linhas_filtradas = [ linha for linha in linhas if Path(str(linha.get("arquivo", "")).strip()).stem.lower() not in stems_alvo ] with open(caminho, "w", encoding="utf-8-sig", newline="") as f: escritor = csv.DictWriter(f, fieldnames=cabecalho, delimiter=";") escritor.writeheader() escritor.writerows(linhas_filtradas) def deletar_leitura_processamento(nome_processamento: str, nome_imagem: str): leituras = carregar_leituras_lote(nome_processamento) chave_encontrada = None if nome_imagem in leituras: chave_encontrada = nome_imagem else: stem_alvo = Path(normalizar_nome_arquivo_debug(nome_imagem)).stem.lower() for chave in list(leituras.keys()): if Path(normalizar_nome_arquivo_debug(chave)).stem.lower() == stem_alvo: chave_encontrada = chave break if not chave_encontrada: raise FileNotFoundError("Leitura não encontrada neste lote.") from app.validacao_cadastral import cartao_enviado_keepedu_completo if cartao_enviado_keepedu_completo(nome_processamento, chave_encontrada, leituras.get(chave_encontrada, {})): raise ValueError("Este cartão já foi enviado ao KeepEdu. Para alterá-lo, faça um novo processamento.") leitura_atual = leituras.pop(chave_encontrada) salvar_leituras_lote(nome_processamento, leituras) nomes_arquivo = _coletar_nomes_arquivo_imagem(chave_encontrada, leitura_atual) # A tela de validação usa um snapshot persistido; remova dele o cartão e # quaisquer correções manuais/pendências vinculadas à mesma imagem. from app.validacao_cadastral import remover_registros_cartao_excluido remover_registros_cartao_excluido(nome_processamento, nomes_arquivo) from app.lotes_store import obter_estado_lote, salvar_estado_lote estado = obter_estado_lote(nome_processamento) correcoes = estado.get("correcoes_web") or {} if isinstance(correcoes, dict): for chave in list(correcoes.keys()): if Path(normalizar_nome_arquivo_debug(chave)).stem.lower() == Path(normalizar_nome_arquivo_debug(chave_encontrada)).stem.lower(): correcoes.pop(chave, None) estado["correcoes_web"] = correcoes salvar_estado_lote(nome_processamento, estado) remover_linha_log_processamento(nome_processamento, nomes_arquivo) return { "status": "ok", "nome_imagem": chave_encontrada, "total_restante": len(leituras), } def deletar_processamento(nome_processamento: str): nome = str(nome_processamento or "").strip() if not nome or nome.startswith("."): raise ValueError("Nome de lote inválido.") from app.lotes_store import remover_lote if not remover_lote(nome): raise FileNotFoundError("Lote não encontrado.") return { "status": "ok", "nome_processamento": nome, } def reprocessar_lote_com_progresso(job_id: str, nome_processamento: str): from app.avaliacao_contexto import ( extrair_id_avaliacao_do_nome_arquivo, resolver_id_avaliacao_esperado_lote, ) from app.validacao_cadastral import leitura_bloqueada_id_avaliacao leituras = carregar_leituras_lote(nome_processamento) id_esperado = resolver_id_avaliacao_esperado_lote(nome_processamento, leituras) imagens = list(leituras.keys()) if id_esperado: for nome_imagem in imagens: id_nome = extrair_id_avaliacao_do_nome_arquivo(nome_imagem) if id_nome and id_nome != id_esperado: registrar_evento_job( job_id, tipo="warning", titulo="ID impresso divergente no arquivo", descricao=( f"Arquivo com prefixo {id_nome}; o lote usa o ID configurado ({id_esperado}). " "A leitura seguirá com o ID do painel." ), arquivo=normalizar_nome_arquivo_debug(nome_imagem), ) total = len(imagens) if total == 0: raise ValueError("Nenhuma imagem restante para reprocessar neste lote.") # Um novo processamento parte apenas das imagens originais: descarta os # cliques manuais e redetecta os quatro cantos de todos os cartoes. from app.lotes_store import obter_estado_lote, salvar_estado_lote estado = obter_estado_lote(nome_processamento) if estado.get("correcoes_web"): estado["correcoes_web"] = {} salvar_estado_lote(nome_processamento, estado) registrar_evento_job( job_id, tipo="info", titulo="Reprocessamento iniciado", descricao=f"{total} cartão(ões) em fila sequencial: uma releitura por vez.", ) atualizar_job( job_id, status="processando", percentual=5, mensagem="Preparando reprocessamento do lote...", nome_processamento=nome_processamento, ) erros_total = 0 for indice, nome_imagem in enumerate(imagens, start=1): percentual = 5 + int((indice / total) * 90) atualizar_job( job_id, status="processando", percentual=percentual, mensagem=f"Reprocessando cartão {indice} de {total}...", arquivo_atual=normalizar_nome_arquivo_debug(nome_imagem), nome_processamento=nome_processamento, ) try: resultado_reprocess = reprocessar_imagem_processamento( nome_processamento=nome_processamento, nome_imagem=nome_imagem, pontos_cantos=None, usar_cantos_salvos=False, ) erros_cartao = resultado_reprocess.get("erros") or [] if resultado_reprocess.get("bloqueado") or leitura_bloqueada_id_avaliacao({"erros": erros_cartao}): erros_total += 1 registrar_evento_job( job_id, tipo="error", titulo="Cartão bloqueado", descricao=erros_cartao[0] if erros_cartao else "Cartão incompatível com a avaliação do lote.", arquivo=normalizar_nome_arquivo_debug(nome_imagem), ) else: registrar_evento_job( job_id, tipo="success", titulo="Cartão reprocessado", arquivo=normalizar_nome_arquivo_debug(nome_imagem), ) except Exception as erro: erros_total += 1 registrar_evento_job( job_id, tipo="error", titulo="Falha ao reprocessar cartão", descricao=str(erro), arquivo=normalizar_nome_arquivo_debug(nome_imagem), ) resumo = resumo_processamento(nome_processamento) mensagem_final = "Reprocessamento do lote concluído." if erros_total: mensagem_final = f"Reprocessamento concluído com {erros_total} falha(s)." atualizar_job( job_id, status="concluido", percentual=100, mensagem=mensagem_final, nome_processamento=nome_processamento, resumo=resumo, ) registrar_evento_job( job_id, tipo="success" if erros_total == 0 else "warning", titulo="Reprocessamento finalizado", descricao=mensagem_final, ) return { "status": "ok", "nome_processamento": nome_processamento, "total": total, "erros": erros_total, "resumo": resumo, } def processar_pasta_temporaria_com_progresso( job_id, pasta_upload, id_avaliacao_esperado=None, nome_simulado=None, meta_avaliacao=None, usuario_nome: str = "", ): nome_processamento = None try: arquivos = [ arquivo for arquivo in pasta_upload.iterdir() if arquivo.suffix.lower() in EXTENSOES_IMAGEM ] total = len(arquivos) registrar_evento_job( job_id, tipo="info", titulo="Processamento iniciado", descricao="Arquivos recebidos e fila de leitura preparada." ) atualizar_job( job_id, status="processando", percentual=10, mensagem="Arquivos recebidos. Iniciando leitura OMR...", resumo={ "info": 0, "warning": 0, "error": 0, "success": 0, "total_eventos": 0, "total_imagens": total, "total_ok": 0, "total_erro": 0, "total_manual": 0 } ) if total == 0: raise ValueError("Nenhuma imagem válida encontrada para processamento.") def progresso_callback(atual, total_arquivos, nome_arquivo=""): percentual = 10 + int((atual / total_arquivos) * 85) atualizar_job( job_id, status="processando", percentual=percentual, mensagem=f"Lendo cartão {atual} de {total_arquivos}...", arquivo_atual=nome_arquivo ) def evento_callback(tipo, titulo, descricao="", arquivo=""): registrar_evento_job( job_id, tipo=tipo, titulo=titulo, descricao=descricao, arquivo=arquivo ) kwargs_omr = { "pasta_saida": str(pasta_processamentos_escola()), "progresso_callback": progresso_callback, "evento_callback": evento_callback, } if id_avaliacao_esperado: kwargs_omr["id_avaliacao_esperado"] = id_avaliacao_esperado try: resultado = processar_imagens_omr(str(pasta_upload), **kwargs_omr) except TypeError: atualizar_job( job_id, status="processando", percentual=35, mensagem="Lendo cartões OMR..." ) kwargs_fallback = {"pasta_saida": str(pasta_processamentos_escola())} if id_avaliacao_esperado: kwargs_fallback["id_avaliacao_esperado"] = id_avaliacao_esperado resultado = processar_imagens_omr(str(pasta_upload), **kwargs_fallback) metadados_lote = dict(meta_avaliacao or {}) if isinstance(meta_avaliacao, dict) else {} if id_avaliacao_esperado: metadados_lote["id_avaliacao"] = str(id_avaliacao_esperado) metadados_lote["id"] = str(id_avaliacao_esperado) if nome_simulado: metadados_lote["nome_simulado"] = str(nome_simulado) metadados_lote.setdefault("tipo_prova", str(nome_simulado)) nome_processamento = salvar_resultado_omr_em_lote(resultado, metadados_lote) from app.validacao_cadastral import extrair_id_prova_do_codigo leituras_processadas = carregar_leituras_lote(nome_processamento) cartoes_sem_id_avaliacao = [ nome_imagem for nome_imagem, dados_cartao in leituras_processadas.items() if not extrair_id_prova_do_codigo(str((dados_cartao or {}).get("codigo_barras") or "")) ] resumo_atual = { "info": 0, "warning": 0, "error": 0, "success": 0, "total_eventos": 0, "total_imagens": int(resultado.get("total_imagens", 0)), "total_ok": int(resultado.get("total_ok", 0)), "total_erro": int(resultado.get("total_erro", 0)), "total_manual": int(resultado.get("total_manual", 0)) } # Deixe o snapshot cadastral pronto antes de liberar o lote. Cartoes # com codigo de barras sao validados localmente, sem esperar uma # consulta externa apenas para preencher o nome de exibicao. try: atualizar_job( job_id, status="processando", percentual=97, mensagem="Preparando painel de validação...", ) from app.validacao_cadastral import gerar_validacao_cadastral gerar_validacao_cadastral( nome_processamento, persistir=True, leituras=leituras_processadas, consultar_nomes_opcionais=False, ) except Exception as erro_validacao: registrar_evento_job( job_id, tipo="warning", titulo="Validação será preparada ao abrir o painel", descricao=str(erro_validacao), ) try: from app.progresso import obter_job resumo_atual.update(obter_job(job_id).get("resumo", {})) resumo_atual.update({ "total_imagens": int(resultado.get("total_imagens", 0)), "total_ok": int(resultado.get("total_ok", 0)), "total_erro": int(resultado.get("total_erro", 0)), "total_manual": int(resultado.get("total_manual", 0)) }) except Exception: pass atualizar_job( job_id, status="concluido", percentual=100, mensagem="Processamento concluído.", nome_processamento=nome_processamento, resultado={ "solicitar_id_avaliacao": bool(cartoes_sem_id_avaliacao and not id_avaliacao_esperado), "cartoes_sem_id_avaliacao": len(cartoes_sem_id_avaliacao), }, resumo=resumo_atual ) registrar_evento_job( job_id, tipo="success", titulo="Processamento concluído", descricao="Os arquivos já podem ser revisados no painel." ) if id_avaliacao_esperado and nome_processamento: try: from app.kanban import automacao_processamento_iniciado automacao_processamento_iniciado( str(id_avaliacao_esperado), nome_processamento=nome_processamento, usuario=str(usuario_nome or "").strip(), total_cartoes=int(resultado.get("total_imagens", 0)), unidade=str(metadados_lote.get("unidade") or ""), ) except Exception: pass except Exception as e: registrar_evento_job( job_id, tipo="error", titulo="Erro durante o processamento", descricao=str(e) ) atualizar_job( job_id, status="erro", percentual=100, mensagem="Erro durante o processamento.", erro=str(e) ) finally: shutil.rmtree(pasta_upload, ignore_errors=True) def limpar_pos_importacao_keepedu(nome_processamento: str, imagens_enviadas: list[str] | None = None): try: print(f"[LIMPEZA KEEPEDU] Iniciando limpeza para imagens enviadas em: {nome_processamento}") if not imagens_enviadas: print("[LIMPEZA KEEPEDU] Nenhuma imagem para limpar.") return pasta_processamento = caminho_pasta_processamento(nome_processamento) leituras = carregar_leituras_lote(nome_processamento) print(f"[DEBUG LIMPEZA] Pasta encontrada: {pasta_processamento}") print(f"[DEBUG LIMPEZA] Existe? {pasta_processamento.exists()}") for nome_imagem_original in imagens_enviadas: nomes_para_limpar: set[str] = set() nome_limpo = str(nome_imagem_original or "").strip() if nome_limpo: nomes_para_limpar.add(nome_limpo) stem = Path(nome_limpo).stem nomes_para_limpar.add(f"template_{stem}.jpg") nomes_para_limpar.add(f"template_{stem}.jpeg") if not nome_limpo.lower().startswith("template_"): nomes_para_limpar.add(f"template_{nome_limpo}") for chave, dados in (leituras or {}).items(): if not isinstance(dados, dict): continue arquivo = str(dados.get("arquivo_original") or "").strip() stem_chave = Path(str(chave)).stem stem_alvo = Path(nome_limpo).stem if nome_limpo else "" if ( arquivo == nome_limpo or stem_chave == stem_alvo or str(chave).strip() == nome_limpo ): nomes_para_limpar.add(str(chave).strip()) if arquivo: nomes_para_limpar.add(arquivo) debug_bolhas = str(dados.get("debug_bolhas") or "").strip() if debug_bolhas: Path(debug_bolhas).unlink(missing_ok=True) for nome_arquivo in nomes_para_limpar: if not nome_arquivo: continue caminho_pendencia = pasta_processamento / "pendencias" / nome_arquivo if caminho_pendencia.exists(): caminho_pendencia.unlink(missing_ok=True) nome_base_sem_ext = Path(nome_arquivo).stem caminho_manual = pasta_processamento / "manual_omr" / f"{nome_base_sem_ext}.json" if caminho_manual.exists(): caminho_manual.unlink(missing_ok=True) caminho_original = caminho_originais(nome_processamento) / Path(nome_arquivo).name if caminho_original.exists(): caminho_original.unlink(missing_ok=True) pasta_originais = caminho_originais(nome_processamento) try: pasta_originais.rmdir() pasta_originais.parent.rmdir() except OSError: pass # Quando todos os cartões foram enviados, os diretórios de apoio podem # ter ficado vazios após a exclusão das pendências/correções. Relatórios # e snapshots ainda existentes impedem a remoção, como esperado. for pasta in ( pasta_processamento / "pendencias", pasta_processamento / "manual_omr", pasta_processamento, ): try: pasta.rmdir() except OSError: pass print(f"[LIMPEZA KEEPEDU] Limpeza de {len(imagens_enviadas)} imagem(ns) concluída.") except Exception: import traceback print("[ERRO LIMPEZA KEEPEDU]") traceback.print_exc()