"""API HTTP do AIDA. Mudanças em relação à versão anterior: * o pipeline (§3) é carregado **uma vez**, na subida do processo. Antes, cada requisição recarregava modelo e imputer; agora o backbone CLIP — que é o componente caro — fica residente; * a resposta expõe o terceiro estado ``INCONCLUSIVO`` (§6.4) e o campo booleano ``inconclusivo``, além de ``scores_modulos``, ``qualidade``, ``limitacoes`` e as evidências visuais do §8.3; * ``GET /evidencia//`` devolve os PNGs gerados para a análise; * ``GET /analise/`` passa a responder do histórico local em vez de devolver um texto de "a implementar"; * ``GET /saude`` informa quais módulos estão carregados — útil para diagnosticar uma implantação sem modelos treinados; * as exceções deixam de ser engolidas em um 500 genérico: erros de validação de imagem viram 400 com mensagem útil, e o traceback vai para o log do servidor. O campo ``resultado`` agora tem **quatro** valores possíveis: ``"REAL"``, ``"REAL_EDITADA"``, ``"IA/MANIPULADA"`` e ``"INCONCLUSIVO"``. Clientes que faziam ``if resultado == "REAL" ... else ...`` precisam tratar os outros casos, sob pena de rotular como IA algo que o sistema declarou indeterminado ou uma foto real que só passou por um editor de imagem. """ from __future__ import annotations import logging import os import sys import tempfile import uuid from pathlib import Path from flask import Flask, jsonify, render_template, request, send_file from flask_cors import CORS from PIL import Image BASE_DIR = Path(__file__).resolve().parents[1] sys.path.append(str(BASE_DIR)) from aida.configuracao import ( # noqa: E402 EXPLICABILIDADE_DIR, MODELOS_DIR, TAMANHO_MAX_BYTES, criar_estrutura, ) from aida.inferencia import cache_analise # noqa: E402 from aida.inferencia.forense import montar_dossie # noqa: E402 from aida.inferencia.pipeline import PipelineAIDA # noqa: E402 from aida.proveniencia.extracao import C2PA_DISPONIVEL # noqa: E402 from aida.proveniencia.fusao import ARQUIVO_LLR # noqa: E402 from aida.utilitarios.imagem import ImagemInvalida # noqa: E402 from analisar_imagem import analisar_imagem # noqa: E402 from historico import buscar_analise_por_id # noqa: E402 from site_api.limite import LimitePorCliente, chamada_interna, identificar_cliente # noqa: E402 from site_api.origem import instalar as instalar_checagem_origem # noqa: E402 logging.basicConfig(level=logging.INFO) logger = logging.getLogger("aida.api") app = Flask(__name__) app.config["MAX_CONTENT_LENGTH"] = TAMANHO_MAX_BYTES def _origens_permitidas(): """Origens autorizadas a chamar a API, de ``ALLOWED_ORIGINS``. O front-end roda em outro domínio, então sem CORS correto o navegador bloqueia toda chamada — e o erro aparece só no console, o que faz parecer que a API está fora do ar. ``*`` libera geral (padrão, conveniente para demonstração); em produção, liste os domínios separados por vírgula. """ valor = os.environ.get("ALLOWED_ORIGINS", "*").strip() if valor == "*": return "*" origens = [o.strip() for o in valor.split(",") if o.strip()] return origens or "*" CORS(app, resources={r"/*": {"origins": _origens_permitidas()}}) # O proxy do *.hf.space reflete qualquer origem por cima do CORS acima: a recusa # de verdade (403) é feita no servidor, com a mesma lista. instalar_checagem_origem(app, "ALLOWED_ORIGINS", isenta=chamada_interna) EXTENSOES_PERMITIDAS = {".jpg", ".jpeg", ".png", ".webp", ".heic", ".heif", ".avif", ".bmp"} criar_estrutura() # Carregado na subida: recarregar o CLIP por requisição custaria segundos. # AIDA_MODO_RAPIDO=1 sobe só o ramo CLIP (~0,2 s/imagem em vez de ~2 s), # ao custo do módulo forense — útil para demonstração ao vivo. MODO_RAPIDO = os.environ.get("AIDA_MODO_RAPIDO", "").strip().lower() in ("1", "true", "sim") # AIDA_SEM_CLIP=1 sobe apenas o ramo forense. Existe para máquinas sem memória # para o backbone CLIP: os pesos pedem ~2 GB de commit no pico e, sem eles, o # processo morre no carregamento — a API nem chega a abrir a porta. Sem o CLIP # o sistema perde o módulo mais forte e a fusão cai para a média dos módulos # presentes, o que `GET /saude` e o campo `fusao.metodo` declaram. SEM_CLIP = os.environ.get("AIDA_SEM_CLIP", "").strip().lower() in ("1", "true", "sim") if SEM_CLIP and MODO_RAPIDO: # Modo rápido é *só* o ramo CLIP: pedir os dois é contraditório, e o # pipeline resolveria a favor do CLIP — justamente o que não cabe aqui. logger.warning("AIDA_SEM_CLIP=1 tem precedência sobre AIDA_MODO_RAPIDO.") MODO_RAPIDO = False PIPELINE = PipelineAIDA(modo_rapido=MODO_RAPIDO, usar_clip=not SEM_CLIP, verbose=True) @app.route("/", methods=["GET"]) def home(): return jsonify({ "status": "online", "mensagem": "API AIDA funcionando", "modulos": list(PIPELINE.modulos.disponiveis), "estados_possiveis": ["REAL", "REAL_EDITADA", "IA/MANIPULADA", "INCONCLUSIVO"], "contrato": "/contrato", "saude": "/saude", "forense": "/forense/", "pagina_de_teste": "/teste", }) @app.route("/teste", methods=["GET"]) def pagina_de_teste(): """Página técnica para verificar a API sem o front-end (que fica noutro repo).""" return render_template("index.html") @app.route("/saude", methods=["GET"]) def saude(): """Diagnóstico da implantação: o que está carregado e o que falta.""" return jsonify({ "modulos_carregados": list(PIPELINE.modulos.disponiveis), "ensemble_b4": PIPELINE.ensemble["nome"] if PIPELINE.ensemble else None, "probabilidade_calibrada": PIPELINE.ensemble is not None, "guarda_fora_de_dominio": PIPELINE.dominio is not None, "modo_rapido": PIPELINE.modo_rapido, "politica": PIPELINE.politica.para_dict(), # Sem a biblioteca c2pa a implantação continua funcionando, mas deixa de # ler manifestos — e isso precisa aparecer no diagnóstico, senão uma # queda silenciosa de cobertura seria indistinguível de "nenhuma imagem # tinha credencial". "proveniencia": { "habilitada": PIPELINE.usar_proveniencia, "c2pa_biblioteca_disponivel": C2PA_DISPONIVEL, "llr_calibrado": (MODELOS_DIR / ARQUIVO_LLR).is_file(), "detector_synthid": False, }, # Sem cache não há AIDA Forensics: o dossiê é montado a partir dele. # Uma queda aqui apareceria como "análise não disponível" na tela do # Forensics, e o diagnóstico precisa distinguir isso de um id errado. "cache_analises": { "entradas": len(list(cache_analise.listar())), "ttl_horas": round(cache_analise.ttl_segundos() / 3600, 2), }, "avisos": PIPELINE.avisos, "pronto": bool(PIPELINE.modulos.disponiveis), }) @app.route("/contrato", methods=["GET"]) def contrato(): """Contrato de ``POST /analisar``, para o front-end integrar sem adivinhar. Servido pela própria API para que a documentação não descole do código: quem integra lê daqui, não de um arquivo que pode estar desatualizado. """ return jsonify({ "endpoint": "POST /analisar", "requisicao": { "content_type": "multipart/form-data", "campos": { "imagem": "arquivo (obrigatório). Máx. 15 MB.", "historico_habilitado": "'true' | 'false' (opcional, padrão false)", "evidencias": "'true' | 'false' (opcional, padrão true). " "false não gera os mapas e responde mais rápido.", }, "formatos_aceitos": sorted(EXTENSOES_PERMITIDAS), }, "resposta_200": { "resultado": "'REAL' | 'REAL_EDITADA' | 'IA/MANIPULADA' | 'INCONCLUSIVO' — " "QUATRO estados. 'IA/MANIPULADA' = gerada por IA (inteira ou em parte); " "'REAL_EDITADA' = fotografia real com edição registrada nos metadados", "editada": "bool — o arquivo registra edição em editor de imagem (vale também " "quando o resultado é IA ou INCONCLUSIVO)", "editores": "list[str] — editores encontrados, ex.: ['Adobe Lightroom']", "inconclusivo": "bool — atalho para resultado == 'INCONCLUSIVO'", "fora_de_dominio": "bool — a imagem não é fotográfica (texto, desenho, cor sólida). " "Quando true, mostre o campo 'titulo', não as probabilidades", "probabilidade_ia": "float 0..1 — probabilidade CALIBRADA. Use em métricas, " "não na interface: o ponto de corte não é 0,50", "probabilidade_real": "float 0..1 (= 1 - probabilidade_ia)", "probabilidade_ia_exibicao": "float 0..1 — MOSTRE ESTE. Mesma decisão, " "reescalada para que o corte caia em 0,50", "probabilidade_real_exibicao": "float 0..1 (= 1 - probabilidade_ia_exibicao)", "limiar_exibicao": "0.5 — constante, por construção da escala de exibição", "banda_inconclusiva_exibicao": "[inferior, superior] na escala de exibição", "confianca": "'baixa' | 'média' | 'alta'", "limiar": "float — ponto de corte em vigor, na escala calibrada", "banda_inconclusiva": "[inferior, superior] — faixa de abstenção (escala calibrada)", "motivos": "list[str] — por que foi inconclusivo (vazio se decidiu)", "titulo": "str — título pronto para exibição", "explicacao": "str — texto pronto, já com a ressalva do §8.2", "ressalva": "str — aviso obrigatório; NÃO omitir na interface", "limitacoes": "list[str] — limitações do arquivo analisado", "scores_modulos": "{modulo: float} — score por módulo", "fusao": "{metodo, calibrado: bool, desacordo: float}", "probabilidade_ia_conteudo": "float 0..1 — probabilidade ANTES da fusão com " "a proveniência. Preservada para auditoria e ablação", "proveniencia": "{status, score, confianca, llr_total, c2pa_presente, c2pa_valido, " "c2pa_cadeia_confiavel, declaracao_ia, tipo_fonte_digital, " "softwares, assertions, emissor_certificado, atribuicao, " "evidencias, parcelas_fusao, divergencia, avisos}. " "status: AI_PROVENANCE_CONFIRMED | AI_EDIT_PROVENANCE | " "AI_PROVENANCE_LIKELY | CAPTURE_PROVENANCE_CONFIRMED | " "C2PA_PRESENT | C2PA_INVALID | METADATA_AI_SIGNAL | " "NO_PROVENANCE_FOUND | UNKNOWN. " "NO_PROVENANCE_FOUND NÃO significa imagem real", "atribuicao_gerador": "{gerador, confianca, base, alternativas} — 'desconhecido' " "quando não há evidência suficiente. Não force atribuição", "divergencia": "objeto|null — presente quando proveniência e conteúdo discordam", "qualidade": "{qualidade_score, ...} — indicadores em 0..1", "evidencias": "{urls: {nome: '/evidencia//'}, patches_suspeitos: [...]}", "modo_rapido": "bool — se true, o módulo forense não participou", "duracao_s": "float", "id_analise": "str — use em GET /analise/", "forense_url": "str — '/forense/'. Dossiê do AIDA Forensics montado " "a partir DESTA análise, sem recalcular nada", }, "endpoints_relacionados": { "GET /forense/": { "descricao": "Dossiê forense da análise: arquivo, resultado, métodos " "executados, mapas de evidência, proveniência, qualidade " "e limitações. Reaproveita o resultado em cache.", "secoes": ["arquivo", "resultado", "metodos", "fusao", "evidencias", "proveniencia", "atribuicao_gerador", "divergencia", "qualidade", "limitacoes", "ressalva"], "erros": {"404": "análise ausente do cache, expirada ou de outra " "versão do modelo — refaça a análise"}, }, "DELETE /forense/": { "descricao": "Remove a análise do cache. Chame ao excluir um item do " "histórico ou a conta do usuário.", }, "GET /evidencia//": { "descricao": "PNG de um mapa de explicabilidade da análise.", }, }, "erros": { "400": "imagem ausente, formato não aceito ou arquivo inválido", "413": "arquivo acima de 15 MB", "429": "limite de análises por cliente atingido (AIDA_LIMITE_ANALISES; desligado " "por padrão); respeite o cabeçalho Retry-After", "503": "nenhum modelo treinado — consulte GET /saude", "500": "erro interno (detalhe vai para o log do servidor, não para a resposta)", }, "avisos_de_integracao": [ "resultado tem QUATRO valores. Código com if/else binário rotulará como IA " "aquilo que o sistema declarou indeterminado ou uma foto real editada.", "REAL_EDITADA depende dos metadados do arquivo: uma foto editada que perdeu " "os metadados (reencaminhada, print) sai como REAL.", "'ressalva' é obrigatória na interface: o resultado é indício, não prova.", "quando fora_de_dominio for true, as probabilidades não têm significado.", "exiba probabilidade_ia_EXIBICAO, não probabilidade_ia. O limiar operacional " "vem do índice de Youden e não cai em 0,50; mostrar a probabilidade calibrada " "produz telas como '28% de IA' ao lado do veredito 'IA/MANIPULADA'. A escala " "de exibição é uma transformação monótona: mesma decisão, mesmo ordenamento, " "corte em 50%.", ], "versao_modelo": PIPELINE.politica.versao_modelo, }) def _validar_imagem(caminho: Path) -> bool: try: with Image.open(caminho) as img: img.verify() return True except Exception: return False LIMITE_ANALISES = LimitePorCliente() @app.route("/analisar", methods=["POST"]) def analisar(): permitido, espera = (True, 0) if chamada_interna(request) else LIMITE_ANALISES.permitir(identificar_cliente(request)) if not permitido: resposta = jsonify({ "erro": "Muitas análises em pouco tempo. Tente novamente em alguns minutos.", "tentar_novamente_em_s": espera, }) resposta.headers["Retry-After"] = str(espera) return resposta, 429 if "imagem" not in request.files: return jsonify({"erro": "Nenhuma imagem enviada."}), 400 arquivo = request.files["imagem"] if not arquivo.filename: return jsonify({"erro": "Nome de arquivo inválido."}), 400 extensao = Path(arquivo.filename).suffix.lower() if extensao not in EXTENSOES_PERMITIDAS: return jsonify({ "erro": "Formato de imagem não permitido.", "formatos_aceitos": sorted(EXTENSOES_PERMITIDAS), }), 400 # `historico` é o nome usado pelo front-end atual (herdado do wrapper # apps/metodo/app.py); `historico_habilitado` é o nome canônico. Aceitar os # dois evita que a troca de backend derrube o histórico em silêncio. bruto = request.form.get("historico_habilitado", request.form.get("historico", "false")) historico_habilitado = str(bruto).strip().lower() in ("true", "1", "sim", "on", "yes") gerar_mapas = request.form.get("evidencias", "true").lower() != "false" id_analise = str(uuid.uuid4()) try: with tempfile.TemporaryDirectory() as pasta_temp: caminho_imagem = Path(pasta_temp) / f"{id_analise}{extensao}" arquivo.save(caminho_imagem) if not _validar_imagem(caminho_imagem): return jsonify({"erro": "O arquivo enviado não é uma imagem válida."}), 400 resultado = analisar_imagem( caminho_imagem=str(caminho_imagem), id_analise=id_analise, historico_habilitado=historico_habilitado, imagem_mantida=False, gerar_mapas=gerar_mapas, pipeline=PIPELINE, ) except ImagemInvalida as exc: return jsonify({"erro": str(exc)}), 400 except RuntimeError as exc: # Tipicamente: nenhum modelo treinado disponível. logger.exception("Falha de configuração ao analisar imagem") return jsonify({"erro": str(exc), "dica": "Consulte GET /saude."}), 503 except Exception: logger.exception("Erro interno ao analisar a imagem") return jsonify({"erro": "Erro interno ao analisar a imagem."}), 500 # O arquivo temporário recebe o id como nome, então `nome_imagem` vira algo # como "8f3a…-… .jpg". Para o dossiê forense isso não serve: quem investiga # precisa reconhecer o arquivo que enviou. O nome original entra num campo # próprio, saneado — ele veio do cliente e será exibido em tela. nome_enviado = Path(arquivo.filename).name.strip() resultado["nome_imagem_original"] = "".join( c for c in nome_enviado if c.isprintable() and c not in '\\/:*?"<>|' )[:120] # Guarda ANTES de podar: o dossiê forense é montado a partir deste payload, # e o cache aplica a sua própria poda (o vetor de características não entra # nem aqui nem lá). Uma falha de gravação não invalida a resposta — o # usuário já tem o resultado; o que se perde é a reabertura no Forensics. cache_analise.guardar(resultado, id_analise=id_analise) # O vetor completo de features (centenas de colunas) fica fora da resposta HTTP. resultado.pop("principais_metricas", None) # Compatibilidade: o front-end atual só aceita a resposta se `sucesso` for # verdadeiro (contrato do wrapper antigo). Sem este campo, toda análise # bem-sucedida seria tratada como erro. resultado["sucesso"] = True evidencias = resultado.get("evidencias") or {} if evidencias.get("arquivos"): evidencias["urls"] = { nome: f"/evidencia/{id_analise}/{nome}" for nome in evidencias["arquivos"] } # Ponte declarada para o AIDA Forensics: o front-end não precisa montar a # URL nem saber que existe um cache por trás dela. resultado["forense_url"] = f"/forense/{id_analise}" return jsonify(resultado), 200 @app.route("/forense/", methods=["GET"]) def forense(id_analise: str): """Dossiê do AIDA Forensics, montado a partir da análise já calculada. Não recalcula nada. Quando o cache não tem a análise — expirou, foi invalidada ou nasceu em outra versão do modelo — a resposta é 404 com a orientação de refazer a análise, e não uma reexecução silenciosa do pipeline: repetir a análise aqui poderia devolver números diferentes dos que o usuário viu no AIDA Image. """ analise = cache_analise.recuperar_analise(id_analise) if analise is None: return jsonify({ "erro": "Análise não disponível para investigação forense.", "id_analise": id_analise, "motivo": "A análise não está em cache, expirou ou foi feita por outra " "versão do modelo.", "acao": "Refaça a análise da imagem no AIDA Image.", }), 404 return jsonify(montar_dossie(analise, id_analise)), 200 @app.route("/forense/", methods=["DELETE"]) def apagar_forense(id_analise: str): """Remove a análise do cache — usado por exclusão de conta e do histórico.""" removida = cache_analise.invalidar(id_analise) return jsonify({"id_analise": id_analise, "removida": removida}), 200 @app.route("/evidencia//", methods=["GET"]) def evidencia(id_analise: str, mapa: str): """Serve um mapa de explicabilidade (§8.3) gerado para a análise.""" # Nomes vêm da URL: sanitiza para não escapar da pasta de evidências. if not id_analise.replace("-", "").isalnum() or not mapa.replace("_", "").isalnum(): return jsonify({"erro": "Identificador inválido."}), 400 for extensao in (".png", ".jpg", ".webp"): caminho = (EXPLICABILIDADE_DIR / f"{id_analise}_{mapa}{extensao}").resolve() if caminho.is_file() and EXPLICABILIDADE_DIR.resolve() in caminho.parents: return send_file(caminho) return jsonify({"erro": "Evidência não encontrada."}), 404 @app.route("/analise/", methods=["GET"]) def buscar_analise(id_analise: str): registro = buscar_analise_por_id(id_analise) if registro is None: return jsonify({"erro": "Análise não encontrada.", "id": id_analise}), 404 return jsonify(registro), 200 @app.errorhandler(413) def arquivo_grande(_erro): return jsonify({ "erro": f"Arquivo acima do limite de {TAMANHO_MAX_BYTES // (1024 * 1024)} MB." }), 413 if __name__ == "__main__": # PaaS (Render, Railway, Fly) injetam a porta por variável de ambiente e # derrubam o serviço se ele não escutar exatamente nela. porta = int(os.environ.get("PORT", "5000")) app.run(host="0.0.0.0", port=porta, debug=False)