import requests
from bs4 import BeautifulSoup
import re
import time
from fake_useragent import UserAgent
import logging
class ReniecAPI:
def __init__(self):
[Link] = UserAgent()
[Link] = [Link]()
[Link]({
'User-Agent': [Link],
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'es-PE,es;q=0.9,en;q=0.8',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
})
[Link] = {'requests': 0, 'last_reset': [Link]()}
def consultar_dni(self, dni: str) -> dict:
"""
Consulta RENIEC por DNI
Retorna: dict con datos o error
"""
if not self._validar_dni(dni):
return {"error": "DNI inválido (8 dígitos numéricos)"}
try:
# Rate limit: max 1 req/segundo
self._respetar_limite()
data = self._scraping_reniec(dni)
return data
except Exception as e:
[Link](f"Error RENIEC {dni}: {str(e)}")
return {"error": f"Error consultando RENIEC: {str(e)}"}
def _validar_dni(self, dni: str) -> bool:
return [Link](r'^\d{8}$', dni) is not None
def _respetar_limite(self):
now = [Link]()
if now - [Link]['last_reset'] > 60:
[Link] = {'requests': 0, 'last_reset': now}
if [Link]['requests'] >= 50: # 50/minuto máx
[Link](60)
[Link] = {'requests': 0, 'last_reset': [Link]()}
[Link]['requests'] += 1
[Link](1.2) # 1 req/segundo
def _scraping_reniec(self, dni: str) -> dict:
# Paso 1: Obtener token CSRF
url_token = "[Link]
resp = [Link](url_token, timeout=10)
soup = BeautifulSoup([Link], 'lxml')
token = [Link]('input', {'name': 'token'})['value']
cod_hash = [Link]('input', {'name': 'cod_hash'})['value']
# Paso 2: Submit consulta
url_consulta = "[Link]
data = {
'numdoc': dni,
'token': token,
'cod_hash': cod_hash,
'action': 'ConsulteRuc'
resp = [Link](url_consulta, data=data, timeout=15)
soup = BeautifulSoup([Link], 'lxml')
# Extraer datos
datos = self._parsear_resultado(soup, dni)
return datos
def _parsear_resultado(self, soup: BeautifulSoup, dni: str) -> dict:
# Buscar tabla de resultados
tabla = [Link]('table', class_='listadocont')
if not tabla:
# Verificar si es CAPTCHA o error
if 'captcha' in [Link]():
return {"error": "CAPTCHA detectado, intenta más tarde"}
return {"error": "DNI no encontrado"}
filas = tabla.find_all('tr')
resultado = {
"dni": dni,
"estado": "ENCONTRADO",
"datos": {}
# Parsear filas
for fila in filas[1:]: # Skip header
celdas = fila.find_all('td')
if len(celdas) >= 2:
clave = celdas[0].[Link]().replace(':', '')
valor = celdas[1].[Link]()
resultado["datos"][clave] = valor
# Campos esperados
campos_esperados = {
'Apellidos y nombres': 'nombre_completo',
'Fecha de nacimiento': 'fecha_nacimiento',
'Estado civil': 'estado_civil',
'Distrito': 'distrito',
'Provincia': 'provincia',
'Departamento': 'departamento'
# Normalizar
datos_normalizados = {}
for reniec_key, crm_key in campos_esperados.items():
datos_normalizados[crm_key] = resultado["datos"].get(reniec_key, "")
return {
**resultado,
"datos_normalizados": datos_normalizados
}
# USO en tu CRM
reniec = ReniecAPI()
# Ejemplo 1: Consulta simple
resultado = reniec.consultar_dni("70364031")
print(resultado)
# Ejemplo 2: Integración CRM
def actualizar_cliente_crm(cliente_id: int, dni: str):
datos = reniec.consultar_dni(dni)
if "error" not in datos:
# Actualizar en tu DB
sql = """
UPDATE clientes
SET nombre_reniec = %s,
fecha_nac_reniec = %s,
distrito_reniec = %s
WHERE id = %s AND dni = %s
"""
# [Link](sql, (datos['datos_normalizados']['nombre_completo'], ...))
print(f"✅ Cliente {cliente_id} actualizado con RENIEC")
else:
print(f"❌ Error RENIEC {dni}: {datos['error']}")