import json import re import time import uuid from datetime import datetime from pathlib import Path from threading import RLock from app.avaliacao_contexto import obter_simulado_por_id from app.db import get_db_session, storage_usa_banco from app.settings import CAMINHO_KANBAN from app.time_utils import agora_local KANBAN_LOCK = RLock() _INDICE_HISTORICO_LOTES: dict[str, list[dict]] | None = None _INDICE_HISTORICO_LOTES_TS: float = 0.0 _INDICE_HISTORICO_LOTES_TTL = 30.0 ETAPAS_PADRAO = [ {"id": "lote_recebido", "titulo": "Lote Recebido", "cor": "#788394", "eh_inicial": True, "eh_final": False}, {"id": "scan", "titulo": "Scan", "cor": "#2a8cdb", "eh_inicial": False, "eh_final": False}, {"id": "correcao_log_erros", "titulo": "Correção de Log de Erros", "cor": "#d97706", "eh_inicial": False, "eh_final": False}, {"id": "importacao_vetores_keepedu", "titulo": "Importação de Vetores - KeepEdu", "cor": "#9b6b43", "eh_inicial": False, "eh_final": False}, {"id": "padronizacao_bernoulli", "titulo": "Padronização Bernoulli", "cor": "#7c5cbf", "eh_inicial": False, "eh_final": False}, {"id": "importacao_vetores_bernoulli", "titulo": "Importação de Vetores - Bernoulli", "cor": "#c9a227", "eh_inicial": False, "eh_final": False}, {"id": "concluido", "titulo": "Concluído", "cor": "#2a7e44", "eh_inicial": False, "eh_final": True}, ] ETAPA_LEGACY_BOLETIM = "processamento_boletim" def _etapa_para_dict( *, slug: str, titulo: str, cor: str, ordem: int = 0, etapa_id: int | None = None, eh_inicial: bool = False, eh_final: bool = False, ativo: bool = True, ) -> dict: slug_limpo = str(slug or "").strip() return { "id": slug_limpo, "etapa_id": etapa_id, "slug": slug_limpo, "titulo": titulo, "cor": cor, "ordem": ordem, "eh_inicial": bool(eh_inicial), "eh_final": bool(eh_final), "ativo": bool(ativo), } def _coluna_model_para_dict(modelo) -> dict: return _etapa_para_dict( etapa_id=modelo.etapa_id, slug=modelo.slug, titulo=modelo.titulo, cor=modelo.cor, ordem=modelo.ordem or 0, eh_inicial=bool(modelo.eh_inicial), eh_final=bool(modelo.eh_final), ativo=bool(modelo.ativo), ) def _ordenar_etapas(etapas: list[dict]) -> list[dict]: return sorted( etapas, key=lambda item: ( int(item.get("ordem") or 0), int(item.get("etapa_id") or 0), item.get("slug") or item.get("id") or "", ), ) def _etapas_ativas(etapas: list[dict], *, incluir_inativas: bool = False) -> list[dict]: if incluir_inativas: return _ordenar_etapas(etapas) return _ordenar_etapas([etapa for etapa in etapas if etapa.get("ativo", True)]) def _ids_etapas_validas() -> set[str]: return {etapa["id"] for etapa in listar_etapas()} def _meta_etapa(slug: str) -> dict | None: slug_limpo = _normalizar_etapa(slug) for etapa in listar_etapas(incluir_inativas=True): if etapa.get("slug") == slug_limpo or etapa.get("id") == slug_limpo: return etapa return None def _etapa_inicial_id() -> str: etapas = listar_etapas() for etapa in etapas: if etapa.get("eh_inicial"): return etapa["id"] return etapas[0]["id"] if etapas else "lote_recebido" def _progresso_da_etapa(etapa_id: str) -> int: ids = [etapa["id"] for etapa in listar_etapas()] if etapa_id not in ids: return 0 indice = ids.index(etapa_id) total = max(len(ids) - 1, 1) return round((indice / total) * 100) def _status_da_etapa(etapa_id: str) -> str: meta = _meta_etapa(etapa_id) if not meta: return "em_andamento" if meta.get("eh_inicial"): return "nao_iniciado" if meta.get("eh_final"): return "concluido" return "em_andamento" def _etapa_eh_final(etapa_id: str) -> bool: meta = _meta_etapa(etapa_id) return bool(meta and meta.get("eh_final")) def _normalizar_serie(valor: str) -> str: partes: list[str] = [] vistos: set[str] = set() for parte in re.split(r"[,;|]", str(valor or "")): limpa = parte.strip() if not limpa: continue chave = limpa.casefold() if chave in vistos: continue vistos.add(chave) partes.append(limpa) return ", ".join(partes) def _normalizar_unidade(valor: str) -> str: return str(valor or "").strip() def _unidades_equivalentes(a: str, b: str) -> bool: unidade_a = _normalizar_unidade(a) unidade_b = _normalizar_unidade(b) if not unidade_a or not unidade_b: return False return unidade_a.casefold() == unidade_b.casefold() def _unidade_lote(nome_processamento: str) -> str: from app.avaliacao_contexto import carregar_meta_avaliacao_do_lote meta = carregar_meta_avaliacao_do_lote(nome_processamento) return _normalizar_unidade(meta.get("unidade") or "") def _lote_pertence_ao_card(nome_processamento: str, card: dict) -> bool: nome = str(nome_processamento or "").strip() if not nome or not isinstance(card, dict): return False if str(card.get("nome_processamento") or "").strip() == nome: return True for item in _normalizar_historico_processamentos(card.get("historico_processamentos")): if str(item.get("nome_processamento") or "").strip() == nome: return True unidade_card = _normalizar_unidade(card.get("unidade") or "") unidade_lote = _unidade_lote(nome) if unidade_card and unidade_lote: return _unidades_equivalentes(unidade_card, unidade_lote) if unidade_card and not unidade_lote: return False return True def _normalizar_etapa(etapa_id: str) -> str: etapa_limpa = str(etapa_id or "").strip() if etapa_limpa == ETAPA_LEGACY_BOLETIM: return "importacao_vetores_bernoulli" return etapa_limpa def _eh_simulado_bernoulli(card: dict) -> bool: campos = [ card.get("tipo_prova", ""), card.get("titulo", ""), card.get("serie", ""), ] return any("bernoulli" in str(valor or "").lower() for valor in campos) def _resolver_tipo_prova(id_avaliacao: str = "", titulo: str = "") -> str: id_limpo = str(id_avaliacao or "").strip() if id_limpo: simulado = obter_simulado_por_id(id_limpo) if simulado: return str(simulado.get("tipo_prova") or simulado.get("nome") or "").strip() titulo_limpo = str(titulo or "").strip() if "bernoulli" in titulo_limpo.lower(): return "Simulado Bernoulli" return titulo_limpo def _indice_etapa(etapa_id: str) -> int: ids = [etapa["id"] for etapa in listar_etapas()] etapa_limpa = _normalizar_etapa(etapa_id) return ids.index(etapa_limpa) if etapa_limpa in ids else -1 def _etapa_atual_card(card: dict) -> str: return _normalizar_etapa(card.get("etapa", "")) def _processado_em_do_lote(nome_processamento: str) -> str: from app.services import _obter_data_hora_processamento, caminho_pasta_processamento nome = str(nome_processamento or "").strip() if not nome: return _agora_iso() pasta = caminho_pasta_processamento(nome) return _obter_data_hora_processamento(nome, pasta).isoformat() def _timestamp_nome_lote_kanban(nome: str) -> int: match = re.match(r"^processamento_(\d{8})_(\d{6})$", str(nome or "").strip()) if not match: return 0 try: return int(datetime.strptime(f"{match.group(1)}_{match.group(2)}", "%Y%m%d_%H%M%S").timestamp()) except ValueError: return 0 def invalidar_cache_historico_kanban() -> None: global _INDICE_HISTORICO_LOTES, _INDICE_HISTORICO_LOTES_TS _INDICE_HISTORICO_LOTES = None _INDICE_HISTORICO_LOTES_TS = 0.0 def _construir_indice_historico_lotes(*, forcar: bool = False) -> dict[str, list[dict]]: global _INDICE_HISTORICO_LOTES, _INDICE_HISTORICO_LOTES_TS agora = time.time() if ( not forcar and _INDICE_HISTORICO_LOTES is not None and (agora - _INDICE_HISTORICO_LOTES_TS) < _INDICE_HISTORICO_LOTES_TTL ): return _INDICE_HISTORICO_LOTES from app.historico_pos_processamento import lote_possui_arquivos_historico, obter_identificacao_prova_lote from app.services import listar_processamentos_recentes, resumo_processamento indice: dict[str, list[dict]] = {} for meta in listar_processamentos_recentes(): nome = str(meta.get("nome") or "").strip() if not nome or not lote_possui_arquivos_historico(nome): continue identificacao = obter_identificacao_prova_lote(nome) id_limpo = str(identificacao.get("id_avaliacao") or "").strip() if not id_limpo: continue resumo = resumo_processamento(nome) from app.avaliacao_contexto import carregar_meta_avaliacao_do_lote meta = carregar_meta_avaliacao_do_lote(nome) indice.setdefault(id_limpo, []).append( { "nome_processamento": nome, "processado_em": _processado_em_do_lote(nome), "usuario": "", "total_cartoes": int(resumo.get("total_imagens") or 0), "unidade": _normalizar_unidade(meta.get("unidade") or ""), } ) for entradas in indice.values(): entradas.sort( key=lambda item: _timestamp_nome_lote_kanban(str(item.get("nome_processamento") or "")), reverse=True, ) _INDICE_HISTORICO_LOTES = indice _INDICE_HISTORICO_LOTES_TS = agora return indice def _listar_historico_lotes_por_id_avaliacao( id_avaliacao: str, *, indice_historico: dict[str, list[dict]] | None = None, ) -> list[dict]: id_limpo = str(id_avaliacao or "").strip() if not id_limpo: return [] if indice_historico is None: indice_historico = _construir_indice_historico_lotes() return list(indice_historico.get(id_limpo, [])) def _sincronizar_historico_processamentos_card( card: dict, *, indice_historico: dict[str, list[dict]] | None = None, ) -> dict: id_avaliacao = str(card.get("id_avaliacao") or "").strip() salvo_por_nome = { item["nome_processamento"]: item for item in _normalizar_historico_processamentos(card.get("historico_processamentos")) } disco = _listar_historico_lotes_por_id_avaliacao(id_avaliacao, indice_historico=indice_historico) if disco: historico: list[dict] = [] for entrada in disco: nome = entrada["nome_processamento"] if not _lote_pertence_ao_card(nome, card): continue existente = salvo_por_nome.get(nome, {}) historico.append( { "nome_processamento": nome, "processado_em": entrada["processado_em"], "usuario": str(existente.get("usuario") or "").strip() or str(card.get("responsavel_correcao") or "").strip() or "Sistema", "total_cartoes": int(entrada.get("total_cartoes") or existente.get("total_cartoes") or 0), "unidade": _normalizar_unidade( existente.get("unidade") or entrada.get("unidade") or card.get("unidade") or "" ), } ) for nome, existente in salvo_por_nome.items(): if any(item["nome_processamento"] == nome for item in historico): continue historico.append(existente) historico.sort( key=lambda item: _timestamp_nome_lote_kanban(str(item.get("nome_processamento") or "")), reverse=True, ) card["historico_processamentos"] = historico[:20] return card card["historico_processamentos"] = list(salvo_por_nome.values())[:20] return card def _aplicar_estado_etapa_avancar(card: dict, etapa_id: str) -> dict: etapa_atual = _normalizar_etapa(card.get("etapa", "")) etapa_nova = _normalizar_etapa(etapa_id) if _indice_etapa(etapa_nova) > _indice_etapa(etapa_atual): return _aplicar_estado_etapa(card, etapa_nova) return card def _normalizar_historico_processamentos(valor) -> list[dict]: if not isinstance(valor, list): return [] historico: list[dict] = [] for item in valor: if not isinstance(item, dict): continue nome = str(item.get("nome_processamento") or "").strip() if not nome: continue historico.append( { "nome_processamento": nome, "processado_em": str(item.get("processado_em") or "").strip(), "usuario": str(item.get("usuario") or "").strip() or "Sistema", "total_cartoes": int(item.get("total_cartoes") or 0), "unidade": _normalizar_unidade(item.get("unidade") or ""), } ) return historico def _registrar_historico_processamento( card: dict, *, nome_processamento: str, usuario: str = "", total_cartoes: int | None = None, ) -> dict: nome = str(nome_processamento or "").strip() if not nome: return card historico = _normalizar_historico_processamentos(card.get("historico_processamentos")) if any(item.get("nome_processamento") == nome for item in historico): return card total = int(total_cartoes or 0) if total <= 0: try: from app.services import resumo_processamento total = int(resumo_processamento(nome).get("total_imagens") or 0) except Exception: total = 0 historico.insert( 0, { "nome_processamento": nome, "processado_em": _processado_em_do_lote(nome), "usuario": str(usuario or "").strip() or "Sistema", "total_cartoes": total, "unidade": _normalizar_unidade(card.get("unidade") or ""), }, ) card["historico_processamentos"] = historico[:20] invalidar_cache_historico_kanban() return card def _aplicar_estado_etapa(card: dict, etapa_id: str) -> dict: etapa = _normalizar_etapa(etapa_id) if etapa not in _ids_etapas_validas(): raise ValueError("Etapa inválida.") card["etapa"] = etapa card["progresso"] = _progresso_da_etapa(etapa) card["status_progresso"] = _status_da_etapa(etapa) if _etapa_eh_final(etapa): if not card.get("data_conclusao"): card["data_conclusao"] = _agora_iso()[:10] else: card["data_conclusao"] = "" return card def _agora_iso() -> str: return agora_local().isoformat() def _card_padrao(**kwargs) -> dict: agora = _agora_iso() card = { "card_id": kwargs.get("card_id") or str(uuid.uuid4()), "titulo": str(kwargs.get("titulo") or "").strip(), "id_avaliacao": str(kwargs.get("id_avaliacao") or "").strip(), "responsavel_entrega": str(kwargs.get("responsavel_entrega") or "").strip(), "responsavel_correcao": str(kwargs.get("responsavel_correcao") or "").strip(), "etapa": kwargs.get("etapa") or _etapa_inicial_id(), "nome_processamento": str(kwargs.get("nome_processamento") or "").strip(), "total_cartoes": int(kwargs.get("total_cartoes") or 0), "progresso": int(kwargs.get("progresso") or 0), "status_prazo": str(kwargs.get("status_prazo") or "no_prazo"), "status_progresso": str(kwargs.get("status_progresso") or "nao_iniciado"), "data_entrega": kwargs.get("data_entrega") or agora[:10], "data_previsao_entrega": str(kwargs.get("data_previsao_entrega") or "").strip()[:10], "unidade": str(kwargs.get("unidade") or "").strip(), "serie": _normalizar_serie(kwargs.get("serie") or ""), "turma": str(kwargs.get("turma") or "").strip(), "tipo_prova": str(kwargs.get("tipo_prova") or "").strip(), "data_conclusao": str(kwargs.get("data_conclusao") or "").strip(), "posicao": int(kwargs.get("posicao") or 0), "historico_processamentos": _normalizar_historico_processamentos( kwargs.get("historico_processamentos") ), "etapa_manual": bool(kwargs.get("etapa_manual")), "criado_em": kwargs.get("criado_em") or agora, "atualizado_em": kwargs.get("atualizado_em") or agora, } return card def _carregar_arquivo() -> dict: caminho = Path(CAMINHO_KANBAN) if not caminho.exists(): return {"cards": [], "colunas": []} try: with open(caminho, "r", encoding="utf-8") as arquivo: dados = json.load(arquivo) except Exception: return {"cards": [], "colunas": []} if not isinstance(dados, dict): return {"cards": [], "colunas": []} cards = dados.get("cards") if not isinstance(cards, list): cards = [] colunas = dados.get("colunas") if not isinstance(colunas, list): colunas = [] return {"cards": cards, "colunas": colunas} def _salvar_arquivo(dados: dict) -> None: caminho = Path(CAMINHO_KANBAN) caminho.parent.mkdir(parents=True, exist_ok=True) caminho_tmp = caminho.with_suffix(".tmp") with open(caminho_tmp, "w", encoding="utf-8") as arquivo: json.dump(dados, arquivo, ensure_ascii=False, indent=2) caminho_tmp.replace(caminho) def _model_para_dict(modelo) -> dict: return { "card_id": modelo.card_id, "titulo": modelo.titulo, "id_avaliacao": modelo.id_avaliacao, "responsavel_entrega": modelo.responsavel_entrega, "responsavel_correcao": getattr(modelo, "responsavel_correcao", "") or "", "etapa": _normalizar_etapa(modelo.etapa), "nome_processamento": modelo.nome_processamento or "", "total_cartoes": modelo.total_cartoes or 0, "progresso": modelo.progresso or 0, "status_prazo": modelo.status_prazo or "no_prazo", "status_progresso": modelo.status_progresso or "iniciado", "data_entrega": ( modelo.data_entrega.isoformat()[:10] if isinstance(modelo.data_entrega, datetime) else str(modelo.data_entrega or "")[:10] ), "data_previsao_entrega": ( modelo.data_previsao_entrega.isoformat()[:10] if getattr(modelo, "data_previsao_entrega", None) and isinstance(modelo.data_previsao_entrega, datetime) else str(getattr(modelo, "data_previsao_entrega", "") or "")[:10] ), "unidade": modelo.unidade or "", "serie": modelo.serie or "", "turma": getattr(modelo, "turma", "") or "", "tipo_prova": getattr(modelo, "tipo_prova", "") or "", "data_conclusao": ( modelo.data_conclusao.isoformat()[:10] if getattr(modelo, "data_conclusao", None) and isinstance(modelo.data_conclusao, datetime) else str(getattr(modelo, "data_conclusao", "") or "")[:10] ), "posicao": modelo.posicao or 0, "historico_processamentos": _normalizar_historico_processamentos( getattr(modelo, "historico_processamentos", None) ), "etapa_manual": bool(getattr(modelo, "etapa_manual", False)), "criado_em": modelo.criado_em.isoformat() if modelo.criado_em else _agora_iso(), "atualizado_em": modelo.atualizado_em.isoformat() if modelo.atualizado_em else _agora_iso(), } def _normalizar_data_iso(valor: str) -> str: texto = str(valor or "").strip() if not texto: return "" if re.match(r"^\d{4}-\d{2}-\d{2}", texto): return texto[:10] br = re.match(r"^(\d{2})/(\d{2})/(\d{4})", texto) if br: return f"{br.group(3)}-{br.group(2)}-{br.group(1)}" br_traco = re.match(r"^(\d{2})-(\d{2})-(\d{4})", texto) if br_traco: return f"{br_traco.group(3)}-{br_traco.group(2)}-{br_traco.group(1)}" return "" def _previsao_entrega_do_simulado(id_avaliacao: str) -> str: id_limpo = str(id_avaliacao or "").strip() if not id_limpo: return "" simulado = obter_simulado_por_id(id_limpo) or {} return _normalizar_data_iso(str(simulado.get("termino") or "")) def _ordenar_cards(cards: list[dict]) -> list[dict]: ids_ordenados = [etapa["id"] for etapa in listar_etapas(incluir_inativas=True)] etapa_inicial = _etapa_inicial_id() def indice_etapa(etapa_id: str) -> int: if etapa_id in ids_ordenados: return ids_ordenados.index(etapa_id) return 99 return sorted( cards, key=lambda item: ( item.get("etapa", etapa_inicial) in ids_ordenados, indice_etapa(item.get("etapa", etapa_inicial)), int(item.get("posicao") or 0), item.get("criado_em") or "", ), ) def _normalizar_card( card: dict, *, indice_historico: dict[str, list[dict]] | None = None, incluir_pendencias: bool = False, ) -> dict: card = dict(card) card.setdefault("responsavel_correcao", "") card.setdefault("tipo_prova", "") card.setdefault("data_conclusao", "") card.setdefault("data_previsao_entrega", "") card.setdefault("turma", "") card.setdefault("etapa_manual", False) card["historico_processamentos"] = _normalizar_historico_processamentos( card.get("historico_processamentos") ) card = _sincronizar_historico_processamentos_card(card, indice_historico=indice_historico) if not card.get("tipo_prova"): card["tipo_prova"] = _resolver_tipo_prova(card.get("id_avaliacao", ""), card.get("titulo", "")) etapa = _normalizar_etapa(card.get("etapa", _etapa_inicial_id())) if etapa in _ids_etapas_validas(): card["etapa"] = etapa card["progresso"] = _progresso_da_etapa(etapa) card["status_progresso"] = _status_da_etapa(etapa) card["serie"] = _normalizar_serie(card.get("serie") or "") previsao_salva = _normalizar_data_iso(str(card.get("data_previsao_entrega") or "")) card["data_previsao_entrega"] = previsao_salva or _previsao_entrega_do_simulado( str(card.get("id_avaliacao") or "") ) card = _sincronizar_etapa_card_pos_processamento(card) card = _sincronizar_etapa_card_por_envio_keepedu(card) if incluir_pendencias: card["pendencias"] = listar_pendencias_card(card) return card def _etapas_padrao_serializadas() -> list[dict]: return [ _etapa_para_dict( slug=etapa["id"], titulo=etapa["titulo"], cor=etapa["cor"], ordem=indice, eh_inicial=bool(etapa.get("eh_inicial")), eh_final=bool(etapa.get("eh_final")), ativo=True, ) for indice, etapa in enumerate(ETAPAS_PADRAO) ] def _listar_etapas_db(*, incluir_inativas: bool = False) -> list[dict]: from app.db_models import KanbanColunaModel with KANBAN_LOCK, get_db_session() as session: modelos = session.query(KanbanColunaModel).order_by( KanbanColunaModel.ordem, KanbanColunaModel.etapa_id, ).all() etapas = [_coluna_model_para_dict(modelo) for modelo in modelos] return _etapas_ativas(etapas, incluir_inativas=incluir_inativas) def _listar_etapas_arquivo(*, incluir_inativas: bool = False) -> list[dict]: with KANBAN_LOCK: dados = _carregar_arquivo() colunas = dados.get("colunas") or [] if not colunas: colunas = _etapas_padrao_serializadas() etapas = [] for indice, coluna in enumerate(colunas): if not isinstance(coluna, dict): continue slug = str(coluna.get("slug") or coluna.get("id") or "").strip() if not slug: continue etapa_id_valor = coluna.get("etapa_id") etapa_id = int(etapa_id_valor) if etapa_id_valor is not None and str(etapa_id_valor).isdigit() else None etapas.append( _etapa_para_dict( slug=slug, etapa_id=etapa_id, titulo=str(coluna.get("titulo") or slug).strip(), cor=str(coluna.get("cor") or "#788394").strip(), ordem=int(coluna.get("ordem", indice) or indice), eh_inicial=bool(coluna.get("eh_inicial")), eh_final=bool(coluna.get("eh_final")), ativo=coluna.get("ativo", True) is not False, ) ) if not etapas: etapas = _etapas_padrao_serializadas() return _etapas_ativas(etapas, incluir_inativas=incluir_inativas) def listar_etapas(*, incluir_inativas: bool = False) -> list[dict]: if storage_usa_banco(): return _listar_etapas_db(incluir_inativas=incluir_inativas) return _listar_etapas_arquivo(incluir_inativas=incluir_inativas) def garantir_etapas_padrao() -> None: if storage_usa_banco(): from app.db_models import KanbanColunaModel with KANBAN_LOCK, get_db_session() as session: if session.query(KanbanColunaModel).count() > 0: return for indice, etapa in enumerate(ETAPAS_PADRAO): session.add( KanbanColunaModel( slug=etapa["id"], titulo=etapa["titulo"], cor=etapa["cor"], ordem=indice, eh_inicial=bool(etapa.get("eh_inicial")), eh_final=bool(etapa.get("eh_final")), ativo=True, ) ) return with KANBAN_LOCK: dados = _carregar_arquivo() if dados.get("colunas"): return dados["colunas"] = _etapas_padrao_serializadas() _salvar_arquivo(dados) def _salvar_colunas_arquivo(colunas: list[dict]) -> None: with KANBAN_LOCK: dados = _carregar_arquivo() dados["colunas"] = colunas _salvar_arquivo(dados) def _validar_slug_etapa(slug: str) -> str: slug_limpo = str(slug or "").strip().lower() if not slug_limpo: raise ValueError("Informe o slug da coluna.") if not all(caractere.isalnum() or caractere == "_" for caractere in slug_limpo): raise ValueError("Use apenas letras, números e underscore no slug da coluna.") return slug_limpo def criar_coluna( *, slug: str, titulo: str, cor: str = "#788394", ordem: int | None = None, eh_inicial: bool = False, eh_final: bool = False, ) -> dict: etapa_limpa = _validar_slug_etapa(slug) titulo_limpo = str(titulo or "").strip() if not titulo_limpo: raise ValueError("Informe o título da coluna.") etapas_existentes = listar_etapas(incluir_inativas=True) if any(etapa["id"] == etapa_limpa for etapa in etapas_existentes): raise ValueError("Já existe uma coluna com esse identificador.") ordem_final = max(0, int(ordem)) if ordem is not None else len(etapas_existentes) if storage_usa_banco(): from app.db_models import KanbanColunaModel with KANBAN_LOCK, get_db_session() as session: if eh_inicial: session.query(KanbanColunaModel).update({KanbanColunaModel.eh_inicial: False}) if eh_final: session.query(KanbanColunaModel).update({KanbanColunaModel.eh_final: False}) modelo = KanbanColunaModel( slug=etapa_limpa, titulo=titulo_limpo, cor=str(cor or "#788394").strip() or "#788394", ordem=ordem_final, eh_inicial=bool(eh_inicial), eh_final=bool(eh_final), ativo=True, ) session.add(modelo) return _meta_etapa(etapa_limpa) or {} colunas = listar_etapas(incluir_inativas=True) if eh_inicial: for coluna in colunas: coluna["eh_inicial"] = False if eh_final: for coluna in colunas: coluna["eh_final"] = False colunas.append( _etapa_para_dict( slug=etapa_limpa, titulo=titulo_limpo, cor=str(cor or "#788394").strip() or "#788394", ordem=ordem_final, eh_inicial=bool(eh_inicial), eh_final=bool(eh_final), ativo=True, ) ) _salvar_colunas_arquivo(_ordenar_etapas(colunas)) return _meta_etapa(etapa_limpa) or {} def atualizar_coluna( slug: str, *, titulo: str | None = None, cor: str | None = None, ordem: int | None = None, eh_inicial: bool | None = None, eh_final: bool | None = None, ativo: bool | None = None, ) -> dict | None: etapa_limpa = _normalizar_etapa(slug) if not _meta_etapa(etapa_limpa): return None if storage_usa_banco(): from app.db_models import KanbanColunaModel with KANBAN_LOCK, get_db_session() as session: modelo = session.query(KanbanColunaModel).filter_by(slug=etapa_limpa).first() if not modelo: return None if titulo is not None: titulo_limpo = str(titulo).strip() if not titulo_limpo: raise ValueError("Informe o título da coluna.") modelo.titulo = titulo_limpo if cor is not None: modelo.cor = str(cor).strip() or modelo.cor if ordem is not None: modelo.ordem = max(0, int(ordem)) if eh_inicial is not None: if eh_inicial: session.query(KanbanColunaModel).update({KanbanColunaModel.eh_inicial: False}) modelo.eh_inicial = bool(eh_inicial) if eh_final is not None: if eh_final: session.query(KanbanColunaModel).update({KanbanColunaModel.eh_final: False}) modelo.eh_final = bool(eh_final) if ativo is not None: if not ativo and any(card.get("etapa") == etapa_limpa for card in listar_cards()): raise ValueError("Não é possível desativar uma coluna que ainda possui cards.") modelo.ativo = bool(ativo) modelo.atualizado_em = agora_local() return _meta_etapa(etapa_limpa) colunas = listar_etapas(incluir_inativas=True) coluna = next((item for item in colunas if item["id"] == etapa_limpa), None) if not coluna: return None if titulo is not None: titulo_limpo = str(titulo).strip() if not titulo_limpo: raise ValueError("Informe o título da coluna.") coluna["titulo"] = titulo_limpo if cor is not None: coluna["cor"] = str(cor).strip() or coluna.get("cor", "#788394") if ordem is not None: coluna["ordem"] = max(0, int(ordem)) if eh_inicial is not None: if eh_inicial: for item in colunas: item["eh_inicial"] = False coluna["eh_inicial"] = bool(eh_inicial) if eh_final is not None: if eh_final: for item in colunas: item["eh_final"] = False coluna["eh_final"] = bool(eh_final) if ativo is not None: if not ativo and any(card.get("etapa") == etapa_limpa for card in listar_cards()): raise ValueError("Não é possível desativar uma coluna que ainda possui cards.") coluna["ativo"] = bool(ativo) _salvar_colunas_arquivo(_ordenar_etapas(colunas)) return _meta_etapa(etapa_limpa) def excluir_coluna(slug: str) -> bool: etapa_limpa = _normalizar_etapa(slug) if not _meta_etapa(etapa_limpa): return False if any(card.get("etapa") == etapa_limpa for card in listar_cards()): raise ValueError("Não é possível excluir uma coluna que ainda possui cards.") if storage_usa_banco(): from app.db_models import KanbanColunaModel with KANBAN_LOCK, get_db_session() as session: modelo = session.query(KanbanColunaModel).filter_by(slug=etapa_limpa).first() if not modelo: return False session.delete(modelo) return True colunas = [item for item in listar_etapas(incluir_inativas=True) if item["id"] != etapa_limpa] if len(colunas) == len(listar_etapas(incluir_inativas=True)): return False _salvar_colunas_arquivo(_ordenar_etapas(colunas)) return True def _listar_cards_db() -> list[dict]: from app.db_models import KanbanCardModel indice_historico = _construir_indice_historico_lotes() with KANBAN_LOCK, get_db_session() as session: modelos = session.query(KanbanCardModel).order_by( KanbanCardModel.etapa, KanbanCardModel.posicao, KanbanCardModel.criado_em, ).all() return [ _normalizar_card(_model_para_dict(modelo), indice_historico=indice_historico) for modelo in modelos ] def listar_cards() -> list[dict]: if storage_usa_banco(): try: return _listar_cards_db() except Exception as erro: if "historico_processamentos" not in str(erro): raise from app.db import _garantir_colunas_kanban _garantir_colunas_kanban() return _listar_cards_db() with KANBAN_LOCK: indice_historico = _construir_indice_historico_lotes() cards = [ _normalizar_card(card, indice_historico=indice_historico) for card in _carregar_arquivo().get("cards", []) ] return _ordenar_cards(cards) def _buscar_card_bruto( *, card_id: str = "", id_avaliacao: str = "", ) -> dict | None: card_id_limpo = str(card_id or "").strip() id_limpo = str(id_avaliacao or "").strip() if storage_usa_banco(): from app.db_models import KanbanCardModel with KANBAN_LOCK, get_db_session() as session: modelo = None if card_id_limpo: modelo = session.query(KanbanCardModel).filter_by(card_id=card_id_limpo).first() if not modelo and id_limpo: modelo = session.query(KanbanCardModel).filter_by(id_avaliacao=id_limpo).first() if modelo: return _model_para_dict(modelo) return None with KANBAN_LOCK: for card in _carregar_arquivo().get("cards", []): if card_id_limpo and card.get("card_id") == card_id_limpo: return dict(card) if id_limpo and str(card.get("id_avaliacao") or "").strip() == id_limpo: return dict(card) return None def _contar_cards_etapa(etapa_id: str) -> int: etapa = _normalizar_etapa(etapa_id) if storage_usa_banco(): from app.db_models import KanbanCardModel with KANBAN_LOCK, get_db_session() as session: return int(session.query(KanbanCardModel).filter_by(etapa=etapa).count() or 0) with KANBAN_LOCK: return sum( 1 for card in _carregar_arquivo().get("cards", []) if _normalizar_etapa(card.get("etapa", "")) == etapa ) def _montar_historico_processamentos_disco( id_avaliacao: str, *, salvo_por_nome: dict[str, dict] | None = None, usuario_padrao: str = "", indice_historico: dict[str, list[dict]] | None = None, ) -> list[dict]: salvo_por_nome = salvo_por_nome or {} disco = _listar_historico_lotes_por_id_avaliacao(id_avaliacao, indice_historico=indice_historico) historico: list[dict] = [] for entrada in disco: nome = entrada["nome_processamento"] existente = salvo_por_nome.get(nome, {}) historico.append( { "nome_processamento": nome, "processado_em": entrada["processado_em"], "usuario": str(existente.get("usuario") or "").strip() or str(usuario_padrao).strip() or "Sistema", "total_cartoes": int(entrada.get("total_cartoes") or existente.get("total_cartoes") or 0), } ) for nome, existente in salvo_por_nome.items(): if any(item["nome_processamento"] == nome for item in historico): continue historico.append(existente) historico.sort( key=lambda item: _timestamp_nome_lote_kanban(str(item.get("nome_processamento") or "")), reverse=True, ) return historico[:20] def historico_processamentos_por_referencia( *, card_id: str = "", id_avaliacao: str = "", ) -> list[dict] | None: card_id_limpo = str(card_id or "").strip() id_limpo = str(id_avaliacao or "").strip() indice_historico = _construir_indice_historico_lotes() card_ref = _buscar_card_bruto(card_id=card_id_limpo, id_avaliacao=id_limpo) if card_ref: if not id_limpo: id_limpo = str(card_ref.get("id_avaliacao") or "").strip() card_sincronizado = _sincronizar_historico_processamentos_card( dict(card_ref), indice_historico=indice_historico, ) historico_salvo = card_sincronizado.get("historico_processamentos") or [] if historico_salvo: return historico_salvo if not id_limpo: return None salvo_por_nome = { item["nome_processamento"]: item for item in _normalizar_historico_processamentos((card_ref or {}).get("historico_processamentos")) } usuario = str((card_ref or {}).get("responsavel_correcao") or "").strip() return _montar_historico_processamentos_disco( id_limpo, salvo_por_nome=salvo_por_nome, usuario_padrao=usuario, indice_historico=indice_historico, ) def historico_processamentos_do_card(card_id: str, id_avaliacao: str = "") -> list[dict] | None: return historico_processamentos_por_referencia(card_id=card_id, id_avaliacao=id_avaliacao) def _nome_processamento_ativo_card(card: dict) -> str: nome = str(card.get("nome_processamento") or "").strip() if nome: return nome historico = _normalizar_historico_processamentos(card.get("historico_processamentos")) if historico: return str(historico[0].get("nome_processamento") or "").strip() return "" def _processado_em_pendencia(card: dict | None, nome_processamento: str) -> str: nome = str(nome_processamento or "").strip() if not nome: return "" if isinstance(card, dict): for item in _normalizar_historico_processamentos(card.get("historico_processamentos")): if str(item.get("nome_processamento") or "").strip() == nome: processado_em = str(item.get("processado_em") or "").strip() if processado_em: return processado_em return _processado_em_do_lote(nome) def _adicionar_pendencia( pendencias: list[dict], *, titulo: str, quantidade: int = 0, url: str = "", nome_processamento: str = "", processado_em: str = "", card: dict | None = None, ) -> None: titulo_limpo = str(titulo or "").strip() if not titulo_limpo: return quantidade_limpa = max(0, int(quantidade or 0)) nome_lote = str(nome_processamento or "").strip() processado_em_lote = str(processado_em or "").strip() or _processado_em_pendencia(card, nome_lote) for item in pendencias: mesmo_titulo = item.get("titulo") == titulo_limpo mesmo_lote = str(item.get("nome_processamento") or "").strip() == nome_lote if mesmo_titulo and mesmo_lote: item["quantidade"] = max(int(item.get("quantidade") or 0), quantidade_limpa) if url and not item.get("url"): item["url"] = url if nome_lote and not item.get("nome_processamento"): item["nome_processamento"] = nome_lote if processado_em_lote and not item.get("processado_em"): item["processado_em"] = processado_em_lote return entrada = { "titulo": titulo_limpo, "quantidade": quantidade_limpa, "url": str(url or "").strip(), } if nome_lote: entrada["nome_processamento"] = nome_lote entrada["processado_em"] = processado_em_lote pendencias.append(entrada) _STATUS_VALIDADO_SEM_PENDENCIA = frozenset({ "VALIDADO_API_RA", "VALIDADO_CODIGO_BARRAS", "VALIDADO_MANUAL", }) def _obter_validacao_cartao(chave: str, arquivo: str, validacoes: dict) -> dict: if chave in validacoes and isinstance(validacoes[chave], dict): return validacoes[chave] if arquivo in validacoes and isinstance(validacoes[arquivo], dict): return validacoes[arquivo] for dados in validacoes.values(): if not isinstance(dados, dict): continue if str(dados.get("arquivo_original") or "").strip() == arquivo: return dados return {} def _cartao_validado_sem_pendencia(validacao: dict) -> bool: return str(validacao.get("status_validacao") or "") in _STATUS_VALIDADO_SEM_PENDENCIA def _cartao_enviado_imagens_keepedu(envio: dict | None) -> bool: return bool(envio and envio.get("imagens")) def _cartao_ignora_pendencias_operacionais(validacao: dict, envio: dict | None) -> bool: return _cartao_validado_sem_pendencia(validacao) or _cartao_enviado_imagens_keepedu(envio) def _lote_imagens_enviadas_keepedu(nome_processamento: str) -> bool: from app.services import caminho_leituras, carregar_json from app.validacao_cadastral import carregar_envios_keepedu, resolver_envio_keepedu nome = str(nome_processamento or "").strip() if not nome: return False leituras = carregar_json(caminho_leituras(nome), {}) if not isinstance(leituras, dict) or not leituras: return False envios = carregar_envios_keepedu(nome) total = 0 com_imagem = 0 for chave, dados in leituras.items(): if not isinstance(dados, dict): continue total += 1 arquivo = str(dados.get("arquivo_original") or chave or "").strip() envio = resolver_envio_keepedu(chave, arquivo, envios) if _cartao_enviado_imagens_keepedu(envio): com_imagem += 1 return com_imagem >= total def _contar_pendencias_processamento(nome_processamento: str) -> tuple[int, int]: """ Retorna (correcao_leitura, validacao_cadastro), desconsiderando cartões validados pela API ou já enviados ao KeepEdu. """ from app.log_omr_csv import erros_exigem_correcao_manual from app.services import caminho_leituras, carregar_json from app.validacao_cadastral import ( carregar_envios_keepedu, carregar_validacao_cadastral_salva, resolver_envio_keepedu, ) nome = str(nome_processamento or "").strip() if not nome: return 0, 0 leituras = carregar_json(caminho_leituras(nome), {}) if not isinstance(leituras, dict) or not leituras: return 0, 0 dados_val = carregar_validacao_cadastral_salva(nome) or {} validacoes = dados_val.get("validacoes") or {} if not isinstance(validacoes, dict): validacoes = {} envios = carregar_envios_keepedu(nome) correcao = 0 validacao = 0 vistos: set[str] = set() for chave, dados in leituras.items(): if not isinstance(dados, dict): continue arquivo = str(dados.get("arquivo_original") or chave or "").strip() if not arquivo or arquivo in vistos: continue vistos.add(arquivo) validacao_cartao = _obter_validacao_cartao(chave, arquivo, validacoes) envio = resolver_envio_keepedu(chave, arquivo, envios) if _cartao_ignora_pendencias_operacionais(validacao_cartao, envio): continue if bool(validacao_cartao.get("precisa_validacao_manual")): validacao += 1 tem_confianca = int(dados.get("total_pendencias_confianca") or 0) > 0 if tem_confianca or erros_exigem_correcao_manual(dados.get("erros")): correcao += 1 return correcao, validacao _ETAPAS_ANTES_IMPORTACAO_VETORES_KEEPEDU = frozenset({ "lote_recebido", "scan", "correcao_log_erros", }) def _lote_processado(nome_processamento: str) -> bool: nome = str(nome_processamento or "").strip() if not nome: return False try: from app.services import resumo_processamento return int(resumo_processamento(nome).get("total_imagens") or 0) > 0 except Exception: return False def _sincronizar_etapa_card_pos_processamento(card: dict) -> dict: """Move cards processados que ainda estão em Lote Recebido para Scan.""" if card.get("etapa_manual") or _etapa_eh_final(_etapa_atual_card(card)): return card nome_processamento = _nome_processamento_ativo_card(card) if not nome_processamento or not _lote_pertence_ao_card(nome_processamento, card): return card if _etapa_atual_card(card) != _etapa_inicial_id(): return card if not _lote_processado(nome_processamento): return card _aplicar_estado_etapa(card, "scan") card["atualizado_em"] = _agora_iso() return _persistir_card(card) or card def _sincronizar_etapa_card_por_envio_keepedu(card: dict) -> dict: """Sincroniza a etapa do card com o progresso de envio ao KeepEdu.""" if card.get("etapa_manual"): return card if _etapa_eh_final(_etapa_atual_card(card)): return card nome_processamento = _nome_processamento_ativo_card(card) if not nome_processamento or not _lote_pertence_ao_card(nome_processamento, card): return card try: faltam_vetores, faltam_imagens = _contar_envios_pendentes_keepedu(nome_processamento) except Exception: return card if faltam_vetores > 0: return card etapa_atual = _etapa_atual_card(card) indice_atual = _indice_etapa(etapa_atual) indice_importacao = _indice_etapa("importacao_vetores_keepedu") if faltam_imagens > 0: if indice_atual >= 0 and indice_importacao >= 0 and indice_atual < indice_importacao: _aplicar_estado_etapa(card, "importacao_vetores_keepedu") card["atualizado_em"] = _agora_iso() return _persistir_card(card) or card return card if _eh_simulado_bernoulli(card): return card if etapa_atual == "concluido": return card _aplicar_estado_etapa(card, "concluido") card["etapa_manual"] = False card["atualizado_em"] = _agora_iso() return _persistir_card(card) or card def _contar_envios_pendentes_keepedu(nome_processamento: str) -> tuple[int, int]: from app.services import caminho_leituras, carregar_json from app.validacao_cadastral import carregar_envios_keepedu, resolver_envio_keepedu leituras = carregar_json(caminho_leituras(nome_processamento), {}) if not isinstance(leituras, dict) or not leituras: return 0, 0 envios = carregar_envios_keepedu(nome_processamento) faltam_vetores = 0 faltam_imagens = 0 for chave, dados in leituras.items(): if not isinstance(dados, dict): continue arquivo = str(dados.get("arquivo_original") or chave or "").strip() envio = resolver_envio_keepedu(chave, arquivo, envios) if not envio or not envio.get("vetores"): faltam_vetores += 1 if not envio or not envio.get("imagens"): faltam_imagens += 1 return faltam_vetores, faltam_imagens def _pendencias_envio_keepedu(nome_processamento: str, *, card: dict | None = None) -> list[dict]: pendencias: list[dict] = [] nome = str(nome_processamento or "").strip() if not nome: return pendencias processado_em = _processado_em_pendencia(card, nome) try: faltam_vetores, faltam_imagens = _contar_envios_pendentes_keepedu(nome) except Exception: return pendencias if faltam_vetores > 0: _adicionar_pendencia( pendencias, titulo="Enviar vetores", quantidade=faltam_vetores, url=f"/validacao/{nome}", nome_processamento=nome, processado_em=processado_em, card=card, ) if faltam_imagens > 0: _adicionar_pendencia( pendencias, titulo="Enviar imagens", quantidade=faltam_imagens, url=f"/validacao/{nome}", nome_processamento=nome, processado_em=processado_em, card=card, ) return pendencias def listar_pendencias_card(card: dict) -> list[dict]: if not isinstance(card, dict): return [] etapa = _etapa_atual_card(card) nome_processamento = _nome_processamento_ativo_card(card) pendencias: list[dict] = [] if not nome_processamento: if etapa == _etapa_inicial_id(): _adicionar_pendencia(pendencias, titulo="Processar cartões", url="/painel#novo-processamento") elif etapa == "scan": _adicionar_pendencia(pendencias, titulo="Enviar imagens para leitura", url="/painel#novo-processamento") return pendencias try: from app.services import resumo_processamento resumo = resumo_processamento(nome_processamento) total_correcao, pendentes_val = _contar_pendencias_processamento(nome_processamento) except Exception: return pendencias if int(resumo.get("total_imagens") or 0) <= 0: _adicionar_pendencia( pendencias, titulo="Enviar imagens para leitura", url=f"/avaliacao/{nome_processamento}", nome_processamento=nome_processamento, card=card, ) if total_correcao > 0: _adicionar_pendencia( pendencias, titulo="Corrigir erros de leitura", quantidade=total_correcao, url=f"/correcao/{nome_processamento}", nome_processamento=nome_processamento, card=card, ) if pendentes_val > 0: _adicionar_pendencia( pendencias, titulo="Validar cadastro", quantidade=pendentes_val, url=f"/validacao/{nome_processamento}", nome_processamento=nome_processamento, card=card, ) for item in _pendencias_envio_keepedu(nome_processamento, card=card): _adicionar_pendencia( pendencias, titulo=item.get("titulo", ""), quantidade=int(item.get("quantidade") or 0), url=item.get("url", ""), nome_processamento=item.get("nome_processamento", nome_processamento), processado_em=item.get("processado_em", ""), card=card, ) if _eh_simulado_bernoulli(card): try: faltam_vetores, faltam_imagens = _contar_envios_pendentes_keepedu(nome_processamento) except Exception: faltam_vetores, faltam_imagens = 1, 1 if faltam_vetores == 0 and faltam_imagens == 0: if etapa == "padronizacao_bernoulli": _adicionar_pendencia(pendencias, titulo="Padronizar vetores Bernoulli", url="/bernoulli") elif etapa == "importacao_vetores_bernoulli": _adicionar_pendencia(pendencias, titulo="Importar vetores Bernoulli", url="/bernoulli") return pendencias def pendencias_do_card(card_id: str, id_avaliacao: str = "") -> list[dict] | None: card_id_limpo = str(card_id or "").strip() id_limpo = str(id_avaliacao or "").strip() card = None if card_id_limpo: card = next((item for item in listar_cards() if item.get("card_id") == card_id_limpo), None) if not card and id_limpo: card = next( (item for item in listar_cards() if str(item.get("id_avaliacao") or "").strip() == id_limpo), None, ) if not card: return None return listar_pendencias_card(card) def _salvar_card_db(card: dict): from app.db_models import KanbanCardModel with KANBAN_LOCK, get_db_session() as session: modelo = session.query(KanbanCardModel).filter_by(card_id=card["card_id"]).first() if not modelo: modelo = KanbanCardModel(card_id=card["card_id"]) session.add(modelo) modelo.titulo = card["titulo"] modelo.id_avaliacao = card["id_avaliacao"] modelo.responsavel_entrega = card["responsavel_entrega"] if hasattr(modelo, "responsavel_correcao"): modelo.responsavel_correcao = card.get("responsavel_correcao") or "" modelo.etapa = card["etapa"] modelo.nome_processamento = card.get("nome_processamento") or "" modelo.total_cartoes = card.get("total_cartoes") or 0 modelo.progresso = card.get("progresso") or 0 modelo.status_prazo = card.get("status_prazo") or "no_prazo" modelo.status_progresso = card.get("status_progresso") or "iniciado" modelo.unidade = card.get("unidade") or "" modelo.serie = card.get("serie") or "" if hasattr(modelo, "turma"): modelo.turma = card.get("turma") or "" if hasattr(modelo, "tipo_prova"): modelo.tipo_prova = card.get("tipo_prova") or "" modelo.posicao = card.get("posicao") or 0 if hasattr(modelo, "historico_processamentos"): modelo.historico_processamentos = _normalizar_historico_processamentos( card.get("historico_processamentos") ) if hasattr(modelo, "etapa_manual"): modelo.etapa_manual = bool(card.get("etapa_manual")) data_entrega = card.get("data_entrega") if data_entrega: try: modelo.data_entrega = datetime.fromisoformat(str(data_entrega)[:10]) except ValueError: modelo.data_entrega = agora_local() if hasattr(modelo, "data_previsao_entrega"): data_previsao = card.get("data_previsao_entrega") if data_previsao: try: modelo.data_previsao_entrega = datetime.fromisoformat(str(data_previsao)[:10]) except ValueError: modelo.data_previsao_entrega = None else: modelo.data_previsao_entrega = None if hasattr(modelo, "data_conclusao"): data_conclusao = card.get("data_conclusao") if data_conclusao: try: modelo.data_conclusao = datetime.fromisoformat(str(data_conclusao)[:10]) except ValueError: modelo.data_conclusao = None else: modelo.data_conclusao = None modelo.atualizado_em = agora_local() def _salvar_card_arquivo(card: dict) -> dict: with KANBAN_LOCK: dados = _carregar_arquivo() cards = dados.get("cards", []) encontrado = False for indice, item in enumerate(cards): if item.get("card_id") == card["card_id"]: card["atualizado_em"] = _agora_iso() cards[indice] = card encontrado = True break if not encontrado: cards.append(card) dados["cards"] = cards _salvar_arquivo(dados) return card def _persistir_card(card: dict) -> dict: if storage_usa_banco(): _salvar_card_db(card) return card return _salvar_card_arquivo(card) def registrar_entrega( *, titulo: str, id_avaliacao: str, responsavel_entrega: str, responsavel_correcao: str = "", total_cartoes: int = 0, data_entrega: str = "", data_previsao_entrega: str = "", unidade: str = "", serie: str = "", turma: str = "", ) -> dict: titulo_limpo = str(titulo or "").strip() id_limpo = str(id_avaliacao or "").strip() responsavel = str(responsavel_entrega or "").strip() unidade_limpa = _normalizar_unidade(unidade) if not titulo_limpo: raise ValueError("Informe o título da avaliação.") if not id_limpo: raise ValueError("Informe o ID da prova.") if not responsavel: raise ValueError("Informe quem foi o responsável pela entrega.") existente = _buscar_por_id_avaliacao_e_unidade(id_limpo, unidade_limpa) if existente: card = dict(existente) card["titulo"] = titulo_limpo card["responsavel_entrega"] = responsavel card["responsavel_correcao"] = str(responsavel_correcao or "").strip() card["total_cartoes"] = max(0, int(total_cartoes or 0)) card["data_entrega"] = data_entrega or _agora_iso()[:10] card["data_previsao_entrega"] = str(data_previsao_entrega or "").strip()[:10] card["unidade"] = unidade_limpa card["serie"] = _normalizar_serie(serie) card["turma"] = str(turma or "").strip() card["tipo_prova"] = _resolver_tipo_prova(id_limpo, titulo_limpo) card["nome_processamento"] = "" card["etapa_manual"] = False _aplicar_estado_etapa(card, _etapa_inicial_id()) card["atualizado_em"] = _agora_iso() return _persistir_card(card) etapa_inicial = _etapa_inicial_id() card = _card_padrao( titulo=titulo_limpo, id_avaliacao=id_limpo, responsavel_entrega=responsavel, responsavel_correcao=str(responsavel_correcao or "").strip(), tipo_prova=_resolver_tipo_prova(id_limpo, titulo_limpo), etapa=etapa_inicial, total_cartoes=total_cartoes, data_entrega=data_entrega or _agora_iso()[:10], data_previsao_entrega=data_previsao_entrega, unidade=unidade_limpa, serie=serie, turma=turma, posicao=_contar_cards_etapa(etapa_inicial), progresso=0, status_progresso="nao_iniciado", status_prazo="no_prazo", etapa_manual=False, ) return _persistir_card(card) def atualizar_etapa(card_id: str, etapa: str, posicao: int | None = None) -> dict | None: return atualizar_card(card_id, etapa=etapa, posicao=posicao) def atualizar_card( card_id: str, *, etapa: str | None = None, posicao: int | None = None, titulo: str | None = None, id_avaliacao: str | None = None, responsavel_entrega: str | None = None, responsavel_correcao: str | None = None, total_cartoes: int | None = None, data_entrega: str | None = None, data_previsao_entrega: str | None = None, unidade: str | None = None, serie: str | None = None, turma: str | None = None, status_prazo: str | None = None, ) -> dict | None: cards = listar_cards() card = next((item for item in cards if item.get("card_id") == card_id), None) if not card: return None if titulo is not None: titulo_limpo = str(titulo).strip() if not titulo_limpo: raise ValueError("Informe o título da avaliação.") card["titulo"] = titulo_limpo if id_avaliacao is not None: id_limpo = str(id_avaliacao).strip() if not id_limpo: raise ValueError("Informe o ID da prova.") card["id_avaliacao"] = id_limpo card["tipo_prova"] = _resolver_tipo_prova(id_limpo, card.get("titulo", "")) if responsavel_entrega is not None: card["responsavel_entrega"] = str(responsavel_entrega).strip() if responsavel_correcao is not None: card["responsavel_correcao"] = str(responsavel_correcao).strip() if total_cartoes is not None: card["total_cartoes"] = max(0, int(total_cartoes)) if data_entrega is not None: card["data_entrega"] = str(data_entrega).strip()[:10] or card.get("data_entrega") if data_previsao_entrega is not None: card["data_previsao_entrega"] = str(data_previsao_entrega).strip()[:10] if unidade is not None: card["unidade"] = str(unidade).strip() if serie is not None: card["serie"] = _normalizar_serie(serie) if turma is not None: card["turma"] = str(turma).strip() if status_prazo is not None: prazo = str(status_prazo).strip() if prazo not in {"no_prazo", "atrasado"}: raise ValueError("Status do prazo inválido.") card["status_prazo"] = prazo if etapa is not None: card["etapa_manual"] = True _aplicar_estado_etapa(card, etapa) if posicao is not None: card["posicao"] = max(0, int(posicao)) card["atualizado_em"] = _agora_iso() card_normalizado = _normalizar_card(card, incluir_pendencias=False) card_normalizado["etapa_manual"] = bool(card.get("etapa_manual")) if etapa is not None: card_normalizado["etapa"] = card["etapa"] card_normalizado["progresso"] = card.get("progresso") or 0 card_normalizado["status_progresso"] = card.get("status_progresso") or "em_andamento" card_normalizado["data_conclusao"] = card.get("data_conclusao") or "" _persistir_card(card_normalizado) for item in listar_cards(): if item.get("card_id") == card_id: return item return card_normalizado def excluir_card(card_id: str) -> bool: card_id_limpo = str(card_id or "").strip() if not card_id_limpo: return False if storage_usa_banco(): from app.db_models import KanbanCardModel with KANBAN_LOCK, get_db_session() as session: modelo = session.query(KanbanCardModel).filter_by(card_id=card_id_limpo).first() if not modelo: return False session.delete(modelo) return True with KANBAN_LOCK: dados = _carregar_arquivo() cards = dados.get("cards", []) filtrados = [item for item in cards if item.get("card_id") != card_id_limpo] if len(filtrados) == len(cards): return False dados["cards"] = filtrados _salvar_arquivo(dados) return True def _buscar_por_id_avaliacao(id_avaliacao: str, etapa: str | None = None) -> dict | None: id_limpo = str(id_avaliacao or "").strip() if not id_limpo: return None for card in listar_cards(): if str(card.get("id_avaliacao") or "").strip() != id_limpo: continue if etapa and card.get("etapa") != etapa: continue return card return None def _buscar_por_id_avaliacao_e_unidade( id_avaliacao: str, unidade: str = "", *, etapa: str | None = None, ) -> dict | None: id_limpo = str(id_avaliacao or "").strip() unidade_limpa = _normalizar_unidade(unidade) if not id_limpo: return None matches: list[dict] = [] for card in listar_cards(): if str(card.get("id_avaliacao") or "").strip() != id_limpo: continue if etapa and card.get("etapa") != etapa: continue if unidade_limpa and not _unidades_equivalentes(card.get("unidade") or "", unidade_limpa): continue matches.append(card) if not matches: return None if len(matches) == 1: return matches[0] matches.sort(key=lambda item: item.get("atualizado_em") or "", reverse=True) return matches[0] def _buscar_card_para_processamento(id_avaliacao: str, unidade: str = "") -> dict | None: id_limpo = str(id_avaliacao or "").strip() if not id_limpo: return None unidade_limpa = _normalizar_unidade(unidade) if unidade_limpa: card = _buscar_por_id_avaliacao_e_unidade(id_limpo, unidade_limpa) if card: return card card = _buscar_por_id_avaliacao(id_limpo, etapa=_etapa_inicial_id()) if card: return card candidatos = [ item for item in listar_cards() if str(item.get("id_avaliacao") or "").strip() == id_limpo ] if not candidatos: return None if len(candidatos) == 1: return candidatos[0] abertos = [ item for item in candidatos if not _etapa_eh_final(_etapa_atual_card(item)) ] if len(abertos) == 1: return abertos[0] abertos.sort(key=lambda item: item.get("atualizado_em") or "", reverse=True) if abertos: return abertos[0] candidatos.sort(key=lambda item: item.get("atualizado_em") or "", reverse=True) return candidatos[0] def _persistir_unidade_no_lote(nome_processamento: str, card: dict, id_avaliacao: str = "") -> None: from app.avaliacao_contexto import salvar_meta_avaliacao_no_lote nome_limpo = str(nome_processamento or "").strip() if not nome_limpo: return unidade_card = _normalizar_unidade(card.get("unidade") or "") meta_extra = {"unidade": unidade_card} if unidade_card else None salvar_meta_avaliacao_no_lote( nome_limpo, str(card.get("id_avaliacao") or id_avaliacao or "").strip(), str(card.get("titulo") or card.get("tipo_prova") or ""), meta_extra=meta_extra, ) def _buscar_por_processamento(nome_processamento: str) -> dict | None: nome_limpo = str(nome_processamento or "").strip() if not nome_limpo: return None for card in listar_cards(): if str(card.get("nome_processamento") or "").strip() == nome_limpo: return card for item in _normalizar_historico_processamentos(card.get("historico_processamentos")): if str(item.get("nome_processamento") or "").strip() == nome_limpo: return card return None def automacao_processamento_iniciado( id_avaliacao: str, nome_processamento: str = "", usuario: str = "", total_cartoes: int | None = None, unidade: str = "", ) -> dict | None: nome_limpo = str(nome_processamento or "").strip() card = _buscar_por_processamento(nome_limpo) if nome_limpo else None if not card: card = _buscar_card_para_processamento(id_avaliacao, unidade=unidade) if not card: return None if nome_limpo: card["nome_processamento"] = nome_limpo _registrar_historico_processamento( card, nome_processamento=nome_limpo, usuario=usuario, total_cartoes=total_cartoes, ) _persistir_unidade_no_lote(nome_limpo, card, id_avaliacao) if not card.get("tipo_prova"): card["tipo_prova"] = _resolver_tipo_prova(card.get("id_avaliacao", ""), card.get("titulo", "")) if not card.get("etapa_manual"): etapa_atual = _etapa_atual_card(card) if not nome_limpo: # Upload iniciado: Lote Recebido -> Scan. if etapa_atual == _etapa_inicial_id(): card["etapa_manual"] = False _aplicar_estado_etapa(card, "scan") elif etapa_atual == _etapa_inicial_id() and _lote_processado(nome_limpo): # Processamento concluído sem passar pela correção: permanece em Scan. _aplicar_estado_etapa(card, "scan") card["atualizado_em"] = _agora_iso() return _persistir_card(card) def automacao_lote_aberto(nome_processamento: str, id_avaliacao: str = "") -> dict | None: card = _buscar_por_processamento(nome_processamento) if not card and id_avaliacao: card = _buscar_card_para_processamento(id_avaliacao) if not card: return None card["nome_processamento"] = str(nome_processamento).strip() _persistir_unidade_no_lote(nome_processamento, card, id_avaliacao) if not card.get("etapa_manual"): etapa_atual = _etapa_atual_card(card) if etapa_atual in _ETAPAS_ANTES_IMPORTACAO_VETORES_KEEPEDU: _aplicar_estado_etapa_avancar(card, "correcao_log_erros") card["atualizado_em"] = _agora_iso() return _persistir_card(card) def _preparar_card_automacao_keepedu( nome_processamento: str, id_avaliacao: str = "", ) -> dict | None: card = _buscar_por_processamento(nome_processamento) if not card and id_avaliacao: card = _buscar_card_para_processamento(id_avaliacao) if not card: return None card["nome_processamento"] = str(nome_processamento).strip() if id_avaliacao: card["id_avaliacao"] = str(id_avaliacao).strip() if not card.get("tipo_prova"): card["tipo_prova"] = _resolver_tipo_prova(card.get("id_avaliacao", ""), card.get("titulo", "")) _persistir_unidade_no_lote(nome_processamento, card, id_avaliacao) return card def automacao_inicio_vetores_keepedu(nome_processamento: str, id_avaliacao: str = "") -> dict | None: card = _preparar_card_automacao_keepedu(nome_processamento, id_avaliacao) if not card: return None etapa_atual = _etapa_atual_card(card) if not card.get("etapa_manual") and etapa_atual != "concluido": _aplicar_estado_etapa(card, "importacao_vetores_keepedu") card["atualizado_em"] = _agora_iso() return _persistir_card(card) def automacao_envio_keepedu(nome_processamento: str, id_avaliacao: str = "") -> dict | None: """ Chamado após o envio das imagens à KeepEdu. """ card = _preparar_card_automacao_keepedu(nome_processamento, id_avaliacao) if not card: return None nome_limpo = str(nome_processamento or "").strip() etapa_atual = _etapa_atual_card(card) imagens_enviadas = ( _lote_imagens_enviadas_keepedu(nome_limpo) and _lote_pertence_ao_card(nome_limpo, card) ) if _eh_simulado_bernoulli(card): if etapa_atual == "importacao_vetores_keepedu" and imagens_enviadas and not card.get("etapa_manual"): _aplicar_estado_etapa_avancar(card, "padronizacao_bernoulli") elif etapa_atual == "padronizacao_bernoulli" and not card.get("etapa_manual"): _aplicar_estado_etapa_avancar(card, "importacao_vetores_bernoulli") elif etapa_atual == "importacao_vetores_bernoulli" and not card.get("etapa_manual"): _aplicar_estado_etapa_avancar(card, "concluido") elif imagens_enviadas and etapa_atual != "concluido" and not card.get("etapa_manual"): _aplicar_estado_etapa(card, "concluido") card["etapa_manual"] = False card["atualizado_em"] = _agora_iso() return _persistir_card(card) def montar_board() -> dict: garantir_etapas_padrao() etapas = listar_etapas() cards = listar_cards() colunas = [] for etapa in etapas: itens = [card for card in cards if card.get("etapa") == etapa["id"]] itens.sort(key=lambda item: (int(item.get("posicao") or 0), item.get("criado_em") or "")) colunas.append({**etapa, "cards": itens, "total": len(itens)}) return {"etapas": etapas, "colunas": colunas, "cards": cards}