add client implementation

consulta_pj/client.py

@@ -0,0 +1,142 @@
+from urllib.parse import urlencode
+
+import aiohttp
+import orjson
+
+from .exceptions import ProcesosJudicialesClientException
+from .schemas import (
+    ActuacionesJudicialesRequest,
+    CausaActor,
+    CausasRequest,
+    CausasResponse,
+    CausasSchema,
+    ContarCausasRequest,
+    GetActuacionesJudicialesResponse,
+    GetExisteIngresoDirectoRequest,
+    GetExisteIngresoDirectoResponse,
+    GetIncidenteJudicaturaResponse,
+    GetInformacionJuicioResponse,
+)
+
+
+class WebClient:
+    API_URL: str = ""
+    BASE_HEADERS: dict[str, str] = dict()
+
+    def __init__(self):
+        self.session: aiohttp.ClientSession | None = None
+
+    def get_headers(self) -> dict[str, str]:
+        return self.BASE_HEADERS
+
+    async def __aenter__(self):
+        self.session = await aiohttp.ClientSession(
+            headers=self.get_headers(),
+            raise_for_status=self.check_status,
+        ).__aenter__()
+        return self
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.session.__aexit__(exc_type, exc, tb)
+
+    async def check_status(self, response: aiohttp.ClientResponse) -> None:
+        pass
+
+
+class ProcesosJudicialesClient(WebClient):
+    API_URL = "https://api.funcionjudicial.gob.ec/EXPEL-CONSULTA-CAUSAS-SERVICE/api/consulta-causas/informacion/"
+    BASE_HEADERS = {
+        "User-Agent": "Mozilla/5.0 (X11; Ubuntu; Linux x86_64; rv:127.0) Gecko/20100101 Firefox/127.0",
+        "Accept": "application/json, text/plain, */*",
+        "Accept-Language": "en-US,en;q=0.5",
+        "Accept-Encoding": "gzip, deflate, br, zstd",
+        "Content-Type": "application/json",
+        "Connection": "keep-alive",
+        "Sec-Fetch-Dest": "empty",
+        "Sec-Fetch-Mode": "cors",
+        "Sec-Fetch-Site": "same-site",
+        "Pragma": "no-cache",
+        "Cache-Control": "no-cache",
+        "Origin": "https://procesosjudiciales.funcionjudicial.gob.ec",
+        "Referer": "https://procesosjudiciales.funcionjudicial.gob.ec/",
+    }
+
+    async def check_status(self, response: aiohttp.ClientResponse) -> None:
+        if response.status >= 400:
+            raise ProcesosJudicialesClientException(f"Error en la petición: {response.status} - {response.reason}")
+
+    async def get_contar_causas(self, request: ContarCausasRequest) -> int:
+        endpoint = "contarCausas"
+        url = f"{self.API_URL}{endpoint}"
+        async with self.session.post(url, data=request.model_dump_json()) as response:
+            total_causas = int(await response.text())
+            return total_causas
+
+    async def get_causas(self, request: CausasSchema) -> list[CausasResponse]:
+        endpoint = "buscarCausas"
+        contar_causas_request = ContarCausasRequest(**request.model_dump())
+        total_causas = await self.get_contar_causas(contar_causas_request)
+        pagination = {"page": 1, "size": total_causas}
+        url = f"{self.API_URL}{endpoint}?{urlencode(pagination)}"
+        data = CausasRequest(**request.model_dump(), **pagination)
+
+        async with self.session.post(url, data=data.model_dump_json()) as response:
+            raw_response_data = await response.text()
+            response_data = orjson.loads(raw_response_data)
+            causas_actor = [CausasResponse(**causa) for causa in response_data]
+            return causas_actor
+
+    async def get_causas_actor(self, cedulaActor: str) -> list[CausasResponse]:
+        endpoint = "buscarCausas"
+        total_causas = await self.get_contar_causas(cedulaActor)
+        pagination = {"page": 1, "size": total_causas}
+        url = f"{self.API_URL}{endpoint}?{urlencode(pagination)}"
+        data = CausasRequest(actor=CausaActor(cedulaActor=cedulaActor), **pagination)
+
+        async with self.session.post(url, data=data.model_dump_json()) as response:
+            raw_response_data = await response.text()
+            response_data = orjson.loads(raw_response_data)
+            causas_actor = [CausasResponse(**causa) for causa in response_data]
+            return causas_actor
+
+    async def get_incidente_judicatura(self, proceso: str) -> GetIncidenteJudicaturaResponse:
+        endpoint = f"getIncidenteJudicatura/{proceso}"
+        api_url = "https://api.funcionjudicial.gob.ec/EXPEL-CONSULTA-CAUSAS-CLEX-SERVICE/api/consulta-causas-clex/informacion/"
+        url = f"{api_url}{endpoint}"
+
+        async with self.session.get(url) as response:
+            raw_response_data = await response.text()
+            response_data = orjson.loads(raw_response_data)
+            return GetIncidenteJudicaturaResponse(incidentesJudicaturas=response_data)
+
+    async def get_informacion_juicio(self, proceso: str) -> GetInformacionJuicioResponse:
+        endpoint = f"getInformacionJuicio/{proceso}"
+        url = f"{self.API_URL}{endpoint}"
+
+        async with self.session.get(url) as response:
+            raw_response_data = await response.text()
+            response_data = orjson.loads(raw_response_data)
+            return GetInformacionJuicioResponse(causas=response_data)
+
+    async def get_existe_ingreso_directo(
+        self, request: GetExisteIngresoDirectoRequest
+    ) -> GetExisteIngresoDirectoResponse:
+        endpoint = "existeIngresoDirecto"
+        api_url = "https://api.funcionjudicial.gob.ec/EXPEL-CONSULTA-CAUSAS-CLEX-SERVICE/api/consulta-causas-clex/informacion/"
+        url = f"{api_url}{endpoint}"
+
+        async with self.session.post(url, data=request.model_dump_json()) as response:
+            raw_response_data = await response.text()
+            response_data = GetExisteIngresoDirectoResponse(**orjson.loads(raw_response_data))
+            return response_data
+
+    async def get_actuaciones_judiciales(
+        self, request: ActuacionesJudicialesRequest
+    ) -> GetActuacionesJudicialesResponse:
+        endpoint = "actuacionesJudiciales"
+        url = f"{self.API_URL}{endpoint}"
+
+        async with self.session.post(url, data=request.model_dump_json()) as response:
+            raw_response_data = await response.text()
+            response_data = orjson.loads(raw_response_data)
+            return GetActuacionesJudicialesResponse(actuaciones_judiciales=response_data)

consulta_pj/exceptions.py

@@ -0,0 +1,2 @@
+class ProcesosJudicialesClientException(Exception):
+    pass

consulta_pj/schemas.py

@@ -0,0 +1,136 @@
+from datetime import datetime
+
+from pydantic import BaseModel
+
+
+class CausaActor(BaseModel):
+    cedulaActor: str = ""
+    nombreActor: str = ""
+
+
+class CausaDemandado(BaseModel):
+    cedulaDemandado: str = ""
+    nombreDemandado: str = ""
+
+
+class PaginatedRequest(BaseModel):
+    page: int = 1
+    size: int = 10
+
+
+class CausasSchema(BaseModel):
+    numeroCausa: str = ""
+    actor: CausaActor = CausaActor()
+    demandado: CausaDemandado = CausaDemandado()
+    provincia: str = ""
+    numeroFiscalia: str = ""
+    recaptcha: str = "verdad"
+
+
+class CausasRequest(CausasSchema, PaginatedRequest):
+    pass
+
+
+class ContarCausasRequest(CausasSchema):
+    pass
+
+
+class CausasResponse(BaseModel):
+    id: int
+    idJuicio: str
+    estadoActual: str | None = None
+    idMateria: int | None = None
+    idProvincia: int | None = None
+    idCanton: int | None = None
+    idJudicatura: int | None = None
+    nombreDelito: str
+    fechaIngreso: datetime
+    nombre: str | None = None
+    cedula: str | None = None
+    idEstadoJuicio: int | None = None
+    nombreMateria: str | None = None
+    nombreEstadoJuicio: str | None = None
+    nombreJudicatura: str | None = None
+    nombreTipoResolucion: str | None = None
+    nombreTipoAccion: str | None = None
+    fechaProvidencia: datetime | None = None
+    nombreProvidencia: str | None = None
+    nombreProvincia: str | None = None
+    iedocumentoAdjunto: str | None = None
+
+
+class LitiganteSchema(BaseModel):
+    tipoLitigante: str
+    nombresLitigante: str
+    representadoPor: str
+    idLitigante: int
+
+
+class IncidenteJudicaturaSchema(BaseModel):
+    idIncidenteJudicatura: int
+    idMovimientoJuicioIncidente: int
+    idJudicaturaDestino: str
+    fechaCrea: datetime
+    incidente: int
+    lstLitiganteActor: list[LitiganteSchema]
+    lstLitiganteDemandado: list[LitiganteSchema]
+    litiganteActor: str | None = None
+    litiganteDemandado: str | None = None
+
+
+class JudicaturaSchema(BaseModel):
+    idJudicatura: str
+    nombreJudicatura: str
+    ciudad: str
+    lstIncidenteJudicatura: list[IncidenteJudicaturaSchema]
+
+
+class GetIncidenteJudicaturaResponse(BaseModel):
+    incidentesJudicaturas: list[JudicaturaSchema]
+
+
+class GetInformacionJuicioResponse(BaseModel):
+    causas: list[CausasResponse]
+
+
+class GetExisteIngresoDirectoRequest(BaseModel):
+    idJuicio: str
+    idMovimientoJuicioIncidente: int
+
+
+class GetExisteIngresoDirectoResponse(BaseModel):
+    ingresado: str
+
+
+class ActuacionesJudicialesRequest(BaseModel):
+    aplicativo: str = "web"
+    idIncidenteJudicatura: int
+    idJudicatura: str
+    idJuicio: str
+    idMovimientoJuicioIncidente: int
+    incidente: int
+    nombreJudicatura: str
+
+
+class ActuacionJudicial(BaseModel):
+    codigo: int
+    idJudicatura: str
+    idJuicio: str
+    fecha: str
+    tipo: str
+    actividad: str
+    visible: str
+    origen: str
+    idMovimientoJuicioIncidente: int
+    ieTablaReferencia: str
+    ieDocumentoAdjunto: str
+    escapeOut: str
+    uuid: str
+    alias: str
+    nombreArchivo: str
+    tipoIngreso: str
+    idTablaReferencia: str
+
+
+class GetActuacionesJudicialesResponse(BaseModel):
+    actuaciones_judiciales: list[ActuacionJudicial]

pyproject.toml

@@ -4,15 +4,16 @@ version = "0.1.0"
 description = "Data Extraction Script"
 requires-python = ">=3.12.0"
 dependencies = [
-    "tortoise-orm[asyncpg]",
+    "aiohttp[speedups]",
+    "fastapi[all]",
+    "orjson",
     "pydantic",
     "pydantic_settings",
-    "beautifulsoup4",
-    "requests",
+    "tortoise-orm[asyncpg]",
 ]
 
 [project.optional-dependencies]
-dev = ["ruff", "mypy", "pip-tools", "pytest"]
+dev = ["ruff", "mypy", "pip-tools", "pytest", "pytest-asyncio"]
 
 [project.urls]
 Repository = "https://github.com/lpardey/prueba-tusdatos"
@@ -62,3 +63,7 @@ no_implicit_reexport = false
 strict_equality = true
 
 exclude = ['^build/.*', '^tests/.*']
+
+[tool.pytest.ini_options]
+python_files = ["tests.py", "test_*.py", "*_tests.py"]
+asyncio_mode = "auto"