API-Scardua/app/services/holmes.py

import asyncio
import json
import logging
import os
import random
import re
import tempfile
import time

import httpx
from app.config import settings
from app.services.controle_api import registrar_contador
from oracledb import Connection

_TOKEN_FILE = os.path.join(tempfile.gettempdir(), "holmes_token_cache.json")


def _ler_token_arquivo() -> dict | None:
    try:
        with open(_TOKEN_FILE) as f:
            return json.load(f)
    except Exception:
        return None


def _salvar_token_arquivo(token: dict):
    try:
        with open(_TOKEN_FILE, "w") as f:
            json.dump(token, f)
    except Exception:
        pass


def _invalidar_token_usuario():
    try:
        os.remove(_TOKEN_FILE)
    except Exception:
        pass


async def token_usuario():
    cached = _ler_token_arquivo()
    if cached:
        return cached
    await asyncio.sleep(random.uniform(0, 0.5))
    cached = _ler_token_arquivo()
    if cached:
        return cached
    url = "https://app-api.holmesdoc.io/v1/session"
    body = {"email": settings.holmes.usuario , "password": settings.holmes.senha}
    headers = {"Content-Type": "application/json", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36"}
    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=body, headers=headers)
        response.raise_for_status()
        token = {"authorization": response.json()["token"]}
        _salvar_token_arquivo(token)
        return token


async def get_header() -> tuple[dict, str]:
    if random.randint(1, 1) >= 2:
        return {"api_token": settings.holmes.token_api}, "holmes"
    header = await token_usuario()
    return header, "holmes_user"


async def _holmes_request(
    method: str, url: str, **kwargs
) -> tuple[httpx.Response, str]:
    api_nome: str = "holmes"
    for tentativa in range(2):
        header, api_nome = await get_header()
        async with httpx.AsyncClient() as client:
            response = await getattr(client, method)(url, headers=header, **kwargs)
            if (
                response.status_code == 401
                and api_nome == "holmes_user"
                and tentativa == 0
            ):
                _invalidar_token_usuario()
                continue
            return response, api_nome
    raise RuntimeError("Holmes: falha de autenticação após retry")


def extrair_origem(origem: str):
    """Extrair origem da string do Holmes

    Args:
        origem (str): 1548 Entrada de Freio

    Returns:
        str: 1548
    """
    return str(origem[:4])


def extrair_empresa(unidade: str):
    """Extrai a empresa da string inteira do holmes

    Args:
        unidade (str): Ex 10.1 Hyundai Teix. Freitas

    Returns:
        empresa: 10 | None
        revenda: 1  | None
    """
    match = re.search(r"^(\d+)\.(\d+)", unidade)
    if match:
        return match.group(1), match.group(2)
    return None, None


def extrair_transacao(transacao: str) -> str:
    """Extrai apenas a parte transação do texto do Holmes

    Args:
        transacao (str): D15 Entrada de Nota

    Returns:
        str: D15
    """
    return transacao[:3]


async def get_holmes_process(id_processo: str, conn: Connection):
    """
    Busca um processo no Holmes. Centralizado para Peças e Despesas.
    """
    url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}"
    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    try:
        response, api_nome = await _holmes_request("get", url)
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        return response.json()
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
        return None
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao consultar Holmes: {e}")
        return None
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/processes/{id}",
            metodo="GET",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


async def get_holmes_process_details(id_processo: str, conn: Connection):
    """
    Busca os detalhes (properties) de um processo no Holmes.
    """
    url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/details"
    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    try:
        response, api_nome = await _holmes_request("get", url)
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        return response.json()
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro na API Holmes details (ID {id_processo}): {e}")
        return None
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao consultar Holmes details: {e}")
        return None
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/processes/{id}/details",
            metodo="GET",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


async def get_holmes_history(id_processo: str, conn: Connection):
    """
    Busca o historico no holmes (importante para pegar o id da ultima task, para avançar futuramente)
    Args:
        id_processo (str): id do processo no holmes
    """
    url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/history"
    payload = {
        "filters": [],
        "page": 1,
        "per_page": 100,
        "sortBy": ["created_at", "desc"],
    }
    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    try:
        response, api_nome = await _holmes_request("post", url, json=payload)
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        return response.json()
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
        return None
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao consultar Holmes: {e}")
        return None
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/processes/{id}/history",
            metodo="POST",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


async def get_holmes_rateio(id_processo: str, conn: Connection):
    url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/tables/e124b2d0-ee14-11ef-95b4-25dee32fe73f/table_items?page=1&per_page=800"
    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    try:
        response, api_nome = await _holmes_request("get", url)
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        return response.json()
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
        return None
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao consultar Holmes: {e}")
        return None
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/processes/{id}/table/(rateio)",
            metodo="GET",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


async def task_id_recente(id_processo: str, conn: Connection):
    dados_tasks = await get_holmes_history(id_processo, conn)

    if (
        not dados_tasks
        or "histories" not in dados_tasks
        or not dados_tasks["histories"]
    ):
        return None

    mais_recente = max(dados_tasks["histories"], key=lambda x: x["created_at"])

    return mais_recente["properties"]["task_id"]


async def task_mais_recente(id_processo: str, conn: Connection):
    dados_tasks = await get_holmes_history(id_processo, conn)

    if not dados_tasks or "histories" not in dados_tasks:
        return None  # Tratamento se a API falhar

    # print(dados_tasks)

    mais_recente = max(dados_tasks["histories"], key=lambda x: x["created_at"])

    return mais_recente


async def historicos_task(id_processo: str, conn: Connection) -> dict | None:
    """_summary_

    Args:
        id_processo (str): Id do processo no holmes

    Returns:
        dict | None : dicionario do historico | None
    """
    dados_tasks = await get_holmes_history(id_processo, conn)
    if not dados_tasks or "histories" not in dados_tasks:
        return None  # Tratamento se a API falhar

    return dados_tasks


async def buscar_processo(
    conn: Connection,
    chave: str | None = None,
    fluxos: list[str] | bool = False,
    ativos: bool = True,
    payload: dict | bool = False,
) -> dict:
    """Obtem os processos que existem com a sua chave

    Args:
        chave (str): Chave principal a ser procurada, preferencialmente unica pfvr, ajuda ae po.
        ativos (bool) Defaults to True
        fluxos (list[str] | bool, optional): _description_. Defaults to False. se quer pegar de um fluxo específico ou geral. Padrão: Geral

    Returns:
        dict: _description_
    """
    if not payload and chave is None:
        raise ValueError(
            "Chave é obrigatória caso o payload não seja enviado chefia, fica alerta ae rapa"
        )

    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    url = "https://app-api.holmesdoc.io/v2/search"

    if not payload:
        payload = {
            "query": {
                "from": 0,
                "size": 200,
                "context": "process",
                "sort": "updated_at",
                "order": "desc",
                "groups": [
                    {
                        "match_all": True,
                        "terms": [
                            {
                                "value": f"{chave}",
                                "type": "match_phrase",
                                "field": "_content",
                            }
                        ],
                    }
                ],
            },
            "trash": False,
            "deleted_by_me": False,
        }
    try:
        response, api_nome = await _holmes_request("post", url, json=payload)
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        dados = response.json()
        docs = dados.get("docs", [])
        if ativos:
            docs = [d for d in docs if d.get("status") != "canceled"]
        if fluxos:
            docs = [d for d in docs if d.get("name") in fluxos]
        return {"status": True, "dados": {**dados, "docs": docs, "total": len(docs)}}
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro na API Holmes (ID {chave}): {e}")
        return {"status": False, "error": e}
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao consultar Holmes: {e}")
        return {"status": False, "error": e}
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/processes/{id}/search/por-chave",
            metodo="POST",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


async def buscar_processo_por_chaves(
    conn: Connection,
    combinacoes: list[list[str]],
    fluxos: list[str] | bool = False,
    ativos: bool = True,
) -> dict:
    """Busca processos no Holmes usando combinações de termos.

    Cada item de combinacoes é uma lista de valores que juntos identificam
    um processo único (ex: [cnpj, numero_nf]). Cada combinação vira um group
    separado na query.

    Args:
        combinacoes: Ex: [["03657256000164", "19"], ["698cd1c570fd0f8f5f8436a4"]]
        ativos: Ignora processos cancelados. Padrão: True.
        fluxos: Filtra por nome de fluxo. Padrão: False (todos).
    """
    url = "https://app-api.holmesdoc.io/v2/search"
    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    payload = {
        "query": {
            "from": 0,
            "size": 200,
            "context": "process",
            "sort": "updated_at",
            "order": "desc",
            "groups": [
                {
                    "match_all": True,
                    "terms": [
                        {"value": termo, "type": "match_phrase", "field": "_content"}
                        for termo in combinacao
                    ],
                }
                for combinacao in combinacoes
            ],
        },
        "trash": False,
        "deleted_by_me": False,
    }

    try:
        response, api_nome = await _holmes_request("post", url, json=payload)
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        dados = response.json()
        docs = dados.get("docs", [])
        if ativos:
            docs = [d for d in docs if d.get("status") != "canceled"]
        if fluxos:
            docs = [d for d in docs if d.get("name") in fluxos]
        return {"status": True, "dados": {**dados, "docs": docs, "total": len(docs)}}
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro na API Holmes (combinacoes {combinacoes}): {e}")
        return {"status": False, "error": e}
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao consultar Holmes: {e}")
        return {"status": False, "error": e}
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/processes/{id}/search/por-chaves",
            metodo="POST",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


async def action(payload: dict, id_task: str, id_processo: str, conn: Connection):
    url = f"https://app-api.holmesdoc.io/v1/tasks/{id_task}/action"
    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                url, headers={"api_token": settings.holmes.token_api}, json=payload
            )
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        return True, response.json()
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro na API Holmes (ID {id_processo}): {e}")
        return False, e
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao consultar Holmes: {e}")
        return False, e
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/processes/{id}/action",
            metodo="POST",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


async def cria_processo(
    id_start: str,
    payload: dict,
    conn: Connection
) -> tuple[bool, dict | str]:
    url = f"https://app-api.holmesdoc.io/v1/workflows/{id_start}/start"
    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    try:
        response, api_nome = await _holmes_request("post", url, json=payload)
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        return True, response.json()
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro na API Holmes - Criar Processo ({payload}): {e}")
        return False, str(e)
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao criar processo no Holmes: {e}")
        return False, str(e)
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/workflows/{id}/start",
            metodo="POST",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


async def enviar_documento(
    id_processo: str,
    arquivo: bytes,
    nome_arquivo: str,
    id_documento: str,
    conn: Connection,
) -> tuple[bool, str | dict]:
    task_id = await task_id_recente(id_processo, conn)

    if not task_id:
        return False, "Não foi possível obter a task mais recente do Holmes"

    url = f"https://app-api.holmesdoc.io/v1/tasks/{task_id}/documents/{id_documento}"
    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    try:
        files = {"file": (nome_arquivo, arquivo, "application/pdf")}
        async with httpx.AsyncClient() as client:
            response = await client.post(
                url, headers={"api_token": settings.holmes.token_api}, files=files
            )
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        return True, {"task_id": task_id}
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro ao enviar documento Holmes (ID {id_processo}): {e}")
        return False, str(e)
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao enviar documento Holmes: {e}")
        return False, str(e)
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/tasks/{id}/documents/{id_documento}",
            metodo="POST",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


# Para testar as funcoes

# async def main():
#     from app.infra.database import db_instance
#     db_instance.create_pool()
#     conn = db_instance.pool.acquire()
#     try:
#         print(await get_holmes_history('69e65ea83fad950fad5715ed', conn))
#     finally:
#         db_instance.pool.release(conn)

# if __name__ == '__main__':
#     import asyncio
#     asyncio.run(main())


async def cancela_processo(
    id_processo: str, conn: Connection
) -> tuple[bool, str | dict]:
    url = f"https://app-api.holmesdoc.io/v1/processes/{id_processo}/cancel"
    payload = {"reason": "Erro na emissão, data de vencimento. Problema na Disal."}
    inicio = time.perf_counter()
    status = None
    sucesso = False
    erro = None
    api_nome = "holmes"

    try:
        response, api_nome = await _holmes_request("put", url, json=payload)
        response.raise_for_status()
        status = response.status_code
        sucesso = True
        return True, response.json() if response.content else {
            "mensagem": "processo cancelado"
        }
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        erro = str(e)
        logging.error(f"Erro ao cancelar processo Holmes (ID {id_processo}): {e}")
        return False, str(e)
    except Exception as e:
        erro = str(e)
        logging.error(f"Erro inesperado ao cancelar processo Holmes: {e}")
        return False, str(e)
    finally:
        duracao = (time.perf_counter() - inicio) * 1000
        registrar_contador(
            conn=conn,
            api=api_nome,
            endpoint="/v1/processes/{id}/cancel",
            metodo="PUT",
            status_code=status,
            sucesso=sucesso,
            duracao_ms=duracao,
            erro=erro,
        )


async def main():
    # print(aaaa())
    print(await token_usuario())


if __name__ == "__main__":
    import asyncio

    asyncio.run(main())