mais types hints...

python/src/sap.py

@@ -1,10 +1,12 @@
 """Módulo para interagir com o SAP GUI."""
+
 import contextlib
 import os
 import subprocess
 import time
 
 import pythoncom
+import pywintypes
 import rich.console
 import win32com.client
 from dotenv import load_dotenv
@@ -13,36 +15,45 @@
 console: rich.console.Console = Console()
 
 
-def reconnect(session):
+def reconnect(session: pywintypes.IID) -> pywintypes.HANDLE:
+    """Reconectar a sessão ao SAP GUI.
+
+    Args:
+    ----
+        session (_type_): _description_
+
+    Returns:
+    -------
+        pywintypes.HANDLE: _description_
+
+    """
     console.print("Tentando obter de volta IID_IDispatch")
-    s_id = pythoncom.CoMarshalInterThreadInterfaceInStream(
-        pythoncom.IID_IDispatch, session)
+    s_id = pythoncom.CoMarshalInterThreadInterfaceInStream(pythoncom.IID_IDispatch, session)
     console.print(f"Obtido com sucesso: {s_id}")
     return s_id
 
 
-def connection_object(n_selected) -> win32com.client.CDispatch:
+def connection_object(n_selected: int) -> win32com.client.CDispatch:
     """Função para listar as conexões ativas."""
     con = get_app()
     connection: win32com.client.CDispatch = con.Item(n_selected)
     return connection
 
 
-def listar_sessoes(n_selected) -> win32com.client.CDispatch:
+def listar_sessoes(n_selected: int) -> win32com.client.CDispatch:
     """Função para listar as sessions ativas."""
     con = get_app()
     con_selected: win32com.client.CDispatch = con.Item(n_selected)
     sessions: win32com.client.CDispatch = con_selected.Sessions
     return sessions
 
 
-def contar_sessoes(n_selected) -> int:
+def contar_sessoes(n_selected: int) -> int:
     """Contar por tamanho de 1 a 6, caso for criar sessão subtrair -1."""
     con = get_app()
     con_selected: win32com.client.CDispatch = con.Item(n_selected)
     sessions: win32com.client.CDispatch = con_selected.Sessions
-    console.print(
-        f"[blue italic]Quantidade de sessões ativas: {sessions.Count}")
+    console.print(f"[blue italic]Quantidade de sessões ativas: {sessions.Count}")
     return sessions.Count
 
 
@@ -55,7 +66,8 @@ def create_session(n_selected: int) -> win32com.client.CDispatch:
     ultimo_indice = len(sessions) - 1
 
     # Criando uma nova sessão com base na última sessão ativa
-    if ultimo_indice < 5:
+    max_sessions = 5
+    if ultimo_indice < max_sessions:
         con_selected.Children(ultimo_indice).CreateSession()
         while ultimo_indice >= len(sessions) - 1:
             sessions = con_selected.Children
@@ -75,7 +87,7 @@ def choose_connection(n_selected: int) -> win32com.client.CDispatch:
     return session
 
 
-def fechar_conexao(n_con) -> None:
+def fechar_conexao(n_con: int) -> None:
     """Função para fechar o SAP."""
     con = get_app()
     connection = con.Item(n_con)
@@ -91,16 +103,27 @@ def encerrar_sap() -> None:
 
 
 def get_connection(token: str) -> str:
+    """Cria nova conexão com o SAPGUI.
+
+    Args:
+    ----
+        token (str): SSO token.
+
+    Returns:
+    -------
+        str: SSO token.
+
+    """
     load_dotenv()
     sap_access = (
         '[System]\n'
         'Name=EP0\n'
         'Client=100\n'
-        fr'GuiParm={os.environ['SERVER']}'
+        rf'GuiParm={os.environ['SERVER']}'
         '\n'
         '[User]\n'
         f'Name={os.environ['USR']}\n'
-        fr'at="MYSAPSSO2={token}"'
+        rf'at="MYSAPSSO2={token}"'
         '\n'
         'Language=PT\n'
         '[Function]\n'
@@ -110,15 +133,15 @@ def get_connection(token: str) -> str:
         'Workplace=false\n'
         'GuiSize=\n'
         '[Options]\n'
-        'Reuse=-1')
+        'Reuse=-1'
+    )
     path_archive = os.getcwd() + "\\shortcut\\repeat\\tx.sap"
     with open(path_archive, "w") as s:
         s.write(sap_access)
 
     # Execute the command
     try:
-        subprocess.run(["powershell", "start", '"' + path_archive + '"'],
-                       shell=True, check=False)
+        subprocess.run(["powershell", "start", '"' + path_archive + '"'], shell=True, check=False)
         time.sleep(5)
         # Verifica se o processo está em execução
         if not is_process_running("powershell.exe"):
@@ -134,8 +157,7 @@ def get_connection(token: str) -> str:
 def is_process_running(process_name: str) -> bool | None:
     """Verifica se o processo está em execução."""
     try:
-        subprocess.check_output(
-            f'tasklist /FI "IMAGENAME eq {process_name}"', shell=True)
+        subprocess.check_output(f'tasklist /FI "IMAGENAME eq {process_name}"', shell=True)
         return True
     except subprocess.CalledProcessError:
         return False
@@ -150,6 +172,5 @@ def get_app() -> win32com.client.CDispatch:
 
     """
     pythoncom.CoInitialize()
-    app: win32com.client.CDispatch = win32com.client.GetObject(
-        "SAPGUI").GetScriptingEngine
+    app: win32com.client.CDispatch = win32com.client.GetObject("SAPGUI").GetScriptingEngine
     return app.Connections

python/src/servicos_executados.py

@@ -1,12 +1,18 @@
 # ServicosExecutados.py
 """Módulo de TSE."""
 
+from __future__ import annotations
+
+import typing
+
 import numpy as np
 
 from python.src.excel_tbs import load_worksheets
 from python.src.lista_reposicao import dict_reposicao
 from python.src.tsepai import pai_dicionario
 
+if typing.TYPE_CHECKING:
+    from win32com.client import CDispatch
 (
     lista,
     materiais,
@@ -32,7 +38,22 @@
 ) = load_worksheets()
 
 
-def verifica_tse(servico, contrato, session):
+def verifica_tse(
+    servico: CDispatch,
+    contrato: str,
+    session: CDispatch,
+) -> tuple[
+    str,
+    int,
+    str | None,
+    list[str],
+    bool,
+    list[tuple[str, str, str, str, str]],
+    list[tuple[str, str, str, str, str]],
+    tuple[str, str, str, str, str] | None,
+    tuple[str, str, str, str, str] | None,
+    np.ndarray | None,
+]:
     """Agrupador de serviço e indexador de classes."""
     empresa, *_ = contrato
     sondagem = ["591000", "567000", "321000", "321500", "283000", "283500"]
@@ -136,11 +157,12 @@ def verifica_tse(servico, contrato, session):
             tse_proibida = "Iara não quer."
             break
 
-        # ---------- REVOGADO --------------------------------
+        """---------- REVOGADO --------------------------------
         # TROCA PÉ DE CV PREVENTIVO - Solicitado por Ivan/Estevan
         # if sap_tse in ('153000', '153500'):
         #     tse_proibida = 'Ivan não quer.'
         #     break
+        """
 
         # INSTALDO CAIXA UMA (PARTE CIVIL)
         if sap_tse == "136000":
@@ -163,9 +185,7 @@ def verifica_tse(servico, contrato, session):
             # Coloca a tse existente na lista temporária
             tse_temp.append(sap_tse)
             # pylint: disable=E1121
-            (reposicao, tse_proibida, identificador, etapa_reposicao) = (
-                pai_dicionario.pai_servico_unitario(sap_tse, session)
-            )
+            (reposicao, tse_proibida, identificador, etapa_reposicao) = pai_dicionario.pai_servico_unitario(sap_tse, session)
             identificador_list.append(identificador)
             chave_unitario = (
                 sap_tse,
@@ -186,9 +206,7 @@ def verifica_tse(servico, contrato, session):
             servico.modifyCell(n_tse, "CODIGO", "5")  # Despesa
             # Coloca a tse existente na lista temporária
             tse_temp.append(sap_tse)
-            (reposicao, tse_proibida, identificador, etapa_reposicao) = (
-                pai_dicionario.pai_servico_cesta(sap_tse, session)
-            )
+            (reposicao, tse_proibida, identificador, etapa_reposicao) = pai_dicionario.pai_servico_cesta(sap_tse, session)
             rem_base_reposicao.append(reposicao)
             identificador_list.append(identificador)
             chave_rb_despesa = (
@@ -209,9 +227,7 @@ def verifica_tse(servico, contrato, session):
             servico.modifyCell(n_tse, "CODIGO", "6")  # Investimento
             # Coloca a tse existente na lista temporária
             tse_temp.append(sap_tse)
-            (reposicao, tse_proibida, identificador, etapa_reposicao) = (
-                pai_dicionario.pai_servico_cesta(sap_tse, session)
-            )
+            (reposicao, tse_proibida, identificador, etapa_reposicao) = pai_dicionario.pai_servico_cesta(sap_tse, session)
             rem_base_reposicao.append(reposicao)
             identificador_list.append(identificador)
             chave_rb_investimento = (
@@ -233,8 +249,9 @@ def verifica_tse(servico, contrato, session):
             # Coloca a tse existente na lista temporária
             tse_temp.append(sap_tse)
             # pylint: disable=E1121
-            (reposicao, tse_proibida, identificador, etapa_reposicao) = (
-                pai_dicionario.pai_servico_desobstrucao(sap_tse, session)
+            (reposicao, tse_proibida, identificador, etapa_reposicao) = pai_dicionario.pai_servico_desobstrucao(
+                sap_tse,
+                session,
             )
             identificador_list.append(identificador)
             chave_rb_despesa = (
@@ -343,10 +360,7 @@ def verifica_tse(servico, contrato, session):
     if (
         chave_rb_despesa is not None
         and pai_tse == 1
-        or (
-            chave_rb_despesa is not None
-            and all(tse in sondagem for tse in chave_rb_despesa[0])
-        )
+        or (chave_rb_despesa is not None and all(tse in sondagem for tse in chave_rb_despesa[0]))
     ) and chave_rb_despesa[0] in sondagem:
         for n_tse, sap_tse in enumerate(range(num_tse_linhas)):
             sap_tse = servico.GetCellValue(n_tse, "TSE")
@@ -366,7 +380,6 @@ def verifica_tse(servico, contrato, session):
                 servico.modifyCell(n_tse, "PAGAR", "n")  # Cesta
                 servico.modifyCell(n_tse, "CODIGO", "6")  # Investimento
     # Fim da condicional.
-    # sys.exit()
     servico.pressEnter()
 
     return (

python/src/tsepai/pai_dicionario.py

@@ -2,17 +2,23 @@
 """Módulo Dicionário Pai."""
 
 # Biblotecas
+from __future__ import annotations
+
 import sys
+import typing
 
 from python.src.tsepai import pais
 
+if typing.TYPE_CHECKING:
+    from win32com.client import CDispatch
+
 
-def oh_pai(session):
+def oh_pai(session: CDispatch) -> pais.Pai:
     """Aleluia Irmãos."""
     return pais.Pai(session)
 
 
-def preservacao_interferencia():
+def preservacao_interferencia() -> tuple[list[str], None | str, str, list[str]]:
     """Captador da tse preservação."""
     tse_temp_reposicao = [""]
     tse_proibida = None
@@ -21,7 +27,7 @@ def preservacao_interferencia():
     return tse_temp_reposicao, tse_proibida, identificador, etapa_reposicao
 
 
-def transformacao_lig():
+def transformacao_lig() -> tuple[list[str], None | str, str, list[str]]:
     """Captador da tse Transformação."""
     tse_temp_reposicao = []
     tse_proibida = "Ramo Transformação"
@@ -30,7 +36,7 @@ def transformacao_lig():
     return tse_temp_reposicao, tse_proibida, identificador, etapa_reposicao
 
 
-def troca_de_ramal_agua_un():
+def troca_de_ramal_agua_un() -> tuple[list[str], None | str, str, list[str]]:
     """Captador da tse TRA."""
     tse_temp_reposicao = []
     tse_proibida = "TRA"
@@ -39,7 +45,7 @@ def troca_de_ramal_agua_un():
     return tse_temp_reposicao, tse_proibida, identificador, etapa_reposicao
 
 
-def pai_servico_unitario(servico_temp, session):
+def pai_servico_unitario(servico_temp: str, session: CDispatch) -> tuple[list[str], None | str, str, list[str]]:
     """Função condicional das chaves do dicionário unitário."""
     pai_unitario = pais.Unitario(session)
 
@@ -121,7 +127,7 @@ def pai_servico_unitario(servico_temp, session):
     return reposicao, tse_proibida, identificador, etapa_reposicao
 
 
-def pai_servico_cesta(servico_temp, session):
+def pai_servico_cesta(servico_temp: str, session: CDispatch) -> tuple[list[str], None | str, str, list[str]]:
     """Função condicional das chaves do dicionário Remuneração Base."""
     pai_cesta = pais.Cesta(session)
     pai_sondagem = pais.Sondagem(session)
@@ -166,8 +172,9 @@ def pai_servico_cesta(servico_temp, session):
     return reposicao, tse_proibida, identificador, etapa_reposicao
 
 
-def pai_servico_desobstrucao(servico_temp, session):
-    """Agregador de TSE de contrato NORTE SUL
+def pai_servico_desobstrucao(servico_temp: str, session: CDispatch) -> tuple[list[str], None | str, str, list[str]]:
+    """Agregador de TSE de contrato NORTE SUL.
+
     para serviços de DD e DC.
     """
     pai_desobstrucao = oh_pai(session)