Aviso legal

regalias_vzla es una herramienta informática de carácter educativo y de referencia técnica. No constituye asesoría legal, fiscal ni financiera. No está afiliada, avalada ni patrocinada por PDVSA, el Ministerio del Poder Popular de Petróleo, el SENIAT ni ningún otro ente del Estado venezolano. Los resultados son estimaciones referenciales sin validez oficial para declaraciones fiscales, liquidaciones ante PDVSA, fiscalizaciones, procedimientos administrativos ni procesos judiciales. Verifique siempre el instrumento vigente en la Gaceta Oficial de la República Bolivariana de Venezuela.

ingesta

regalias_vzla.ingesta

Capa de adaptadores: ingesta segura de datos externos hacia el dominio.

parsear_registros

parsear_registros(registros: Iterable[Mapping[str, Any]]) -> list[FluidoCrudo]

Valida una secuencia de diccionarios crudos contra el dominio.

Acepta claves canónicas (volumen_bruto) o alias habituales (bbls, BSW, api). Los registros inválidos lanzan pydantic.ValidationError.

Source code in src/regalias_vzla/ingesta.py

def parsear_registros(registros: Iterable[Mapping[str, Any]]) -> list[FluidoCrudo]:
    """Valida una secuencia de diccionarios crudos contra el dominio.

    Acepta claves canónicas (``volumen_bruto``) o alias habituales
    (``bbls``, ``BSW``, ``api``). Los registros inválidos lanzan
    ``pydantic.ValidationError``.
    """
    return [
        FluidoCrudo.model_validate(_mapear_registro(registro, f"registro {numero}"))
        for numero, registro in enumerate(registros, start=1)
    ]

parsear_csv

parsear_csv(contenido: str | bytes) -> list[FluidoCrudo]

Parsea el contenido textual de un CSV hacia modelos del dominio.

Source code in src/regalias_vzla/ingesta.py

def parsear_csv(contenido: str | bytes) -> list[FluidoCrudo]:
    """Parsea el contenido textual de un CSV hacia modelos del dominio."""
    if isinstance(contenido, bytes):
        contenido = contenido.decode("utf-8-sig")
    lector = csv.DictReader(io.StringIO(contenido))
    if lector.fieldnames is None:
        raise ValueError("CSV vacío: se requiere al menos la fila de encabezados.")
    filas = list(lector)
    return [
        FluidoCrudo.model_validate(_mapear_registro(fila, f"fila {numero}"))
        for numero, fila in enumerate(filas, start=2)
    ]

cargar_csv

cargar_csv(ruta: str | Path) -> list[FluidoCrudo]

Lee un archivo CSV desde disco y lo parsea de forma segura.

Source code in src/regalias_vzla/ingesta.py

def cargar_csv(ruta: str | Path) -> list[FluidoCrudo]:
    """Lee un archivo CSV desde disco y lo parsea de forma segura."""
    try:
        contenido = Path(ruta).read_bytes().decode("utf-8-sig")
    except UnicodeDecodeError as error:
        raise ValueError(f"No se pudo decodificar '{ruta}' como UTF-8: {error}") from error
    return parsear_csv(contenido)

parsear_json

parsear_json(contenido: str | bytes) -> list[FluidoCrudo]

Parsea JSON (lista de registros u objeto con clave 'registros').

Source code in src/regalias_vzla/ingesta.py

def parsear_json(contenido: str | bytes) -> list[FluidoCrudo]:
    """Parsea JSON (lista de registros u objeto con clave 'registros')."""
    datos = json.loads(contenido)
    if isinstance(datos, Mapping) and "registros" in datos:
        datos = datos["registros"]
    if not isinstance(datos, list):
        raise ValueError(
            "El JSON debe ser una lista de registros o un objeto con clave 'registros'."
        )
    return parsear_registros(datos)

cargar_json

cargar_json(ruta: str | Path) -> list[FluidoCrudo]

Lee un archivo JSON desde disco y lo parsea de forma segura.

Source code in src/regalias_vzla/ingesta.py

def cargar_json(ruta: str | Path) -> list[FluidoCrudo]:
    """Lee un archivo JSON desde disco y lo parsea de forma segura."""
    try:
        contenido = Path(ruta).read_bytes()
    except OSError as error:
        raise ValueError(f"No se pudo leer el archivo '{ruta}': {error}") from error
    try:
        return parsear_json(contenido)
    except json.JSONDecodeError as error:
        raise ValueError(f"JSON inválido en '{ruta}': {error}") from error