import sqlite3 import os import uuid import json from datetime import datetime BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DB_PATH = os.path.join(BASE_DIR, "db", "prestamos.db") SCHEMA = """ CREATE TABLE IF NOT EXISTS usuarios ( id INTEGER PRIMARY KEY AUTOINCREMENT, nombre TEXT NOT NULL, email TEXT NOT NULL UNIQUE, password_hash TEXT NOT NULL, telefono TEXT, tipo_documento TEXT, numero_documento TEXT, direccion TEXT, acepto_terminos INTEGER NOT NULL DEFAULT 0, acepto_terminos_fecha TEXT, creado_en TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS prestamos ( id INTEGER PRIMARY KEY AUTOINCREMENT, prestamista_id INTEGER NOT NULL, ciudad TEXT, prestatario_nombre TEXT NOT NULL, prestatario_telefono TEXT, prestatario_email TEXT, prestatario_tipo_documento TEXT, prestatario_numero_documento TEXT, prestatario_direccion TEXT, monto REAL NOT NULL, moneda TEXT NOT NULL DEFAULT 'COP', tasa_interes REAL NOT NULL DEFAULT 0, tipo_tasa TEXT NOT NULL DEFAULT 'mes_vencido', tasa_interes_mora REAL NOT NULL DEFAULT 0, fecha_inicio TEXT NOT NULL, fecha_vencimiento TEXT NOT NULL, numero_cuotas INTEGER NOT NULL DEFAULT 1, tipo_periodo TEXT NOT NULL DEFAULT 'mes', estado TEXT NOT NULL DEFAULT 'activo' CHECK(estado IN ('activo', 'pagado', 'mora', 'cancelado')), notas TEXT, token_publico TEXT UNIQUE, tasa_excede_usura INTEGER NOT NULL DEFAULT 0, clausulas TEXT, codeudor_nombre TEXT, codeudor_telefono TEXT, codeudor_email TEXT, codeudor_tipo_documento TEXT, codeudor_numero_documento TEXT, codeudor_direccion TEXT, servicio_seguro INTEGER NOT NULL DEFAULT 0, costo_seguro REAL NOT NULL DEFAULT 0, servicio_whatsapp INTEGER NOT NULL DEFAULT 0, costo_whatsapp REAL NOT NULL DEFAULT 0, prestatario_ingreso_mensual REAL, prestatario_gastos_mensuales REAL, creado_en TEXT NOT NULL, FOREIGN KEY(prestamista_id) REFERENCES usuarios(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS firmas ( id INTEGER PRIMARY KEY AUTOINCREMENT, prestamo_id INTEGER NOT NULL, rol TEXT NOT NULL CHECK(rol IN ('prestamista', 'prestatario', 'codeudor')), nombre_firmante TEXT NOT NULL, ip TEXT, hash_acuerdo TEXT NOT NULL, documento_identidad TEXT, firma_dibujada TEXT, fecha_firma TEXT NOT NULL, FOREIGN KEY(prestamo_id) REFERENCES prestamos(id) ON DELETE CASCADE, UNIQUE(prestamo_id, rol) ); CREATE TABLE IF NOT EXISTS cuotas ( id INTEGER PRIMARY KEY AUTOINCREMENT, prestamo_id INTEGER NOT NULL, numero INTEGER NOT NULL, fecha_vencimiento TEXT NOT NULL, monto REAL NOT NULL, pagada INTEGER NOT NULL DEFAULT 0, fecha_pago TEXT, interes_mora_pagado REAL NOT NULL DEFAULT 0, FOREIGN KEY(prestamo_id) REFERENCES prestamos(id) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS idx_prestamos_prestamista ON prestamos(prestamista_id); CREATE INDEX IF NOT EXISTS idx_cuotas_prestamo ON cuotas(prestamo_id); CREATE INDEX IF NOT EXISTS idx_prestamos_prestatario_documento ON prestamos(prestatario_numero_documento); """ def get_connection(): os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON") return conn MIGRACIONES = [ "ALTER TABLE prestamos ADD COLUMN tasa_interes_mora REAL NOT NULL DEFAULT 0", "ALTER TABLE cuotas ADD COLUMN interes_mora_pagado REAL NOT NULL DEFAULT 0", "ALTER TABLE prestamos ADD COLUMN token_publico TEXT", "ALTER TABLE usuarios ADD COLUMN acepto_terminos INTEGER NOT NULL DEFAULT 0", "ALTER TABLE usuarios ADD COLUMN acepto_terminos_fecha TEXT", "ALTER TABLE prestamos ADD COLUMN tasa_excede_usura INTEGER NOT NULL DEFAULT 0", "ALTER TABLE prestamos ADD COLUMN clausulas TEXT", "ALTER TABLE prestamos ADD COLUMN codeudor_nombre TEXT", "ALTER TABLE prestamos ADD COLUMN codeudor_telefono TEXT", "ALTER TABLE prestamos ADD COLUMN codeudor_email TEXT", "ALTER TABLE prestamos ADD COLUMN moneda TEXT NOT NULL DEFAULT 'COP'", "ALTER TABLE prestamos ADD COLUMN tipo_tasa TEXT NOT NULL DEFAULT 'mes_vencido'", "ALTER TABLE prestamos ADD COLUMN tipo_periodo TEXT NOT NULL DEFAULT 'mes'", "ALTER TABLE usuarios ADD COLUMN tipo_documento TEXT", "ALTER TABLE usuarios ADD COLUMN numero_documento TEXT", "ALTER TABLE usuarios ADD COLUMN direccion TEXT", "ALTER TABLE prestamos ADD COLUMN ciudad TEXT", "ALTER TABLE prestamos ADD COLUMN prestatario_tipo_documento TEXT", "ALTER TABLE prestamos ADD COLUMN prestatario_numero_documento TEXT", "ALTER TABLE prestamos ADD COLUMN prestatario_direccion TEXT", "ALTER TABLE prestamos ADD COLUMN codeudor_tipo_documento TEXT", "ALTER TABLE prestamos ADD COLUMN codeudor_numero_documento TEXT", "ALTER TABLE prestamos ADD COLUMN codeudor_direccion TEXT", "ALTER TABLE firmas ADD COLUMN documento_identidad TEXT", "ALTER TABLE prestamos ADD COLUMN servicio_seguro INTEGER NOT NULL DEFAULT 0", "ALTER TABLE prestamos ADD COLUMN costo_seguro REAL NOT NULL DEFAULT 0", "ALTER TABLE prestamos ADD COLUMN servicio_whatsapp INTEGER NOT NULL DEFAULT 0", "ALTER TABLE prestamos ADD COLUMN costo_whatsapp REAL NOT NULL DEFAULT 0", "ALTER TABLE prestamos ADD COLUMN prestatario_ingreso_mensual REAL", "ALTER TABLE prestamos ADD COLUMN prestatario_gastos_mensuales REAL", "ALTER TABLE firmas ADD COLUMN firma_dibujada TEXT", ] def _migrar_firmas_agregar_codeudor(conn): """El CHECK de la columna 'rol' no se puede alterar con ALTER TABLE en SQLite, asi que si la tabla existente no admite 'codeudor' se reconstruye preservando los datos.""" row = conn.execute( "SELECT sql FROM sqlite_master WHERE type='table' AND name='firmas'" ).fetchone() if not row or "codeudor" in row["sql"]: return conn.execute("ALTER TABLE firmas RENAME TO firmas_old") conn.execute( """CREATE TABLE firmas ( id INTEGER PRIMARY KEY AUTOINCREMENT, prestamo_id INTEGER NOT NULL, rol TEXT NOT NULL CHECK(rol IN ('prestamista', 'prestatario', 'codeudor')), nombre_firmante TEXT NOT NULL, ip TEXT, hash_acuerdo TEXT NOT NULL, documento_identidad TEXT, fecha_firma TEXT NOT NULL, FOREIGN KEY(prestamo_id) REFERENCES prestamos(id) ON DELETE CASCADE, UNIQUE(prestamo_id, rol) )""" ) conn.execute( "INSERT INTO firmas (id, prestamo_id, rol, nombre_firmante, ip, hash_acuerdo, fecha_firma) " "SELECT id, prestamo_id, rol, nombre_firmante, ip, hash_acuerdo, fecha_firma FROM firmas_old" ) conn.execute("DROP TABLE firmas_old") def init_db(): conn = get_connection() try: conn.executescript(SCHEMA) conn.commit() for migracion in MIGRACIONES: try: conn.execute(migracion) conn.commit() except sqlite3.OperationalError: pass # la columna ya existe (base de datos creada con un esquema previo) _migrar_firmas_agregar_codeudor(conn) conn.commit() # Rellena token_publico para prestamos creados antes de que existiera esta columna. sin_token = conn.execute("SELECT id FROM prestamos WHERE token_publico IS NULL").fetchall() for row in sin_token: conn.execute( "UPDATE prestamos SET token_publico = ? WHERE id = ?", (uuid.uuid4().hex, row["id"]) ) if sin_token: conn.commit() finally: conn.close() # ---------- Usuarios ---------- def crear_usuario( nombre, email, password_hash, telefono=None, acepto_terminos=False, tipo_documento=None, numero_documento=None, direccion=None, ): conn = get_connection() try: ahora = datetime.now().isoformat() cur = conn.execute( """INSERT INTO usuarios (nombre, email, password_hash, telefono, tipo_documento, numero_documento, direccion, acepto_terminos, acepto_terminos_fecha, creado_en) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( nombre, email, password_hash, telefono, tipo_documento, numero_documento, direccion, 1 if acepto_terminos else 0, ahora if acepto_terminos else None, ahora, ), ) conn.commit() return cur.lastrowid finally: conn.close() def obtener_usuario_por_email(email): conn = get_connection() try: row = conn.execute("SELECT * FROM usuarios WHERE email = ?", (email,)).fetchone() return dict(row) if row else None finally: conn.close() def obtener_usuario(usuario_id): conn = get_connection() try: row = conn.execute("SELECT * FROM usuarios WHERE id = ?", (usuario_id,)).fetchone() return dict(row) if row else None finally: conn.close() def actualizar_perfil_usuario(usuario_id, nombre, email, telefono, direccion): """Actualiza solo los datos de contacto del prestamista. Deliberadamente NO recibe tipo_documento ni numero_documento: esos campos no son editables desde este flujo.""" conn = get_connection() try: conn.execute( "UPDATE usuarios SET nombre = ?, email = ?, telefono = ?, direccion = ? WHERE id = ?", (nombre, email, telefono, direccion, usuario_id), ) conn.commit() finally: conn.close() # ---------- Prestamos ---------- def crear_prestamo(data, cuotas): conn = get_connection() try: cur = conn.execute( """INSERT INTO prestamos (prestamista_id, ciudad, prestatario_nombre, prestatario_telefono, prestatario_email, prestatario_tipo_documento, prestatario_numero_documento, prestatario_direccion, monto, moneda, tasa_interes, tipo_tasa, tasa_interes_mora, fecha_inicio, fecha_vencimiento, numero_cuotas, tipo_periodo, estado, notas, token_publico, tasa_excede_usura, clausulas, codeudor_nombre, codeudor_telefono, codeudor_email, codeudor_tipo_documento, codeudor_numero_documento, codeudor_direccion, servicio_seguro, costo_seguro, servicio_whatsapp, costo_whatsapp, creado_en) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( data["prestamista_id"], data.get("ciudad"), data["prestatario_nombre"], data.get("prestatario_telefono"), data.get("prestatario_email"), data.get("prestatario_tipo_documento"), data.get("prestatario_numero_documento"), data.get("prestatario_direccion"), data["monto"], data.get("moneda", "COP"), data.get("tasa_interes", 0), data.get("tipo_tasa", "mes_vencido"), data.get("tasa_interes_mora", 0), data["fecha_inicio"], data["fecha_vencimiento"], data.get("numero_cuotas", 1), data.get("tipo_periodo", "mes"), data.get("estado", "activo"), data.get("notas"), uuid.uuid4().hex, 1 if data.get("tasa_excede_usura") else 0, json.dumps(data.get("clausulas") or {}), data.get("codeudor_nombre"), data.get("codeudor_telefono"), data.get("codeudor_email"), data.get("codeudor_tipo_documento"), data.get("codeudor_numero_documento"), data.get("codeudor_direccion"), 1 if data.get("servicio_seguro") else 0, data.get("costo_seguro", 0), 1 if data.get("servicio_whatsapp") else 0, data.get("costo_whatsapp", 0), datetime.now().isoformat(), ), ) prestamo_id = cur.lastrowid for cuota in cuotas: conn.execute( "INSERT INTO cuotas (prestamo_id, numero, fecha_vencimiento, monto, pagada) VALUES (?, ?, ?, ?, 0)", (prestamo_id, cuota["numero"], cuota["fecha_vencimiento"], cuota["monto"]), ) conn.commit() return prestamo_id finally: conn.close() def listar_prestamos(prestamista_id): conn = get_connection() try: rows = conn.execute( "SELECT * FROM prestamos WHERE prestamista_id = ? ORDER BY creado_en DESC", (prestamista_id,), ).fetchall() return [dict(r) for r in rows] finally: conn.close() def obtener_prestamo(prestamo_id, prestamista_id): conn = get_connection() try: prestamo = conn.execute( "SELECT * FROM prestamos WHERE id = ? AND prestamista_id = ?", (prestamo_id, prestamista_id), ).fetchone() if not prestamo: return None return _armar_prestamo(conn, dict(prestamo)) finally: conn.close() def obtener_prestamo_por_token(token): conn = get_connection() try: prestamo = conn.execute( "SELECT * FROM prestamos WHERE token_publico = ?", (token,) ).fetchone() if not prestamo: return None return _armar_prestamo(conn, dict(prestamo)) finally: conn.close() def _armar_prestamo(conn, prestamo): cuotas = conn.execute( "SELECT * FROM cuotas WHERE prestamo_id = ? ORDER BY numero", (prestamo["id"],) ).fetchall() firmas = conn.execute( "SELECT * FROM firmas WHERE prestamo_id = ?", (prestamo["id"],) ).fetchall() prestamo["cuotas"] = [dict(c) for c in cuotas] prestamo["firmas"] = [dict(f) for f in firmas] try: prestamo["clausulas"] = json.loads(prestamo["clausulas"]) if prestamo["clausulas"] else {} except (TypeError, ValueError): prestamo["clausulas"] = {} return prestamo def actualizar_estado_prestamo(prestamo_id, estado): conn = get_connection() try: conn.execute("UPDATE prestamos SET estado = ? WHERE id = ?", (estado, prestamo_id)) conn.commit() finally: conn.close() def marcar_cuota_pagada(cuota_id, pagada=True, interes_mora=0): conn = get_connection() try: fecha_pago = datetime.now().isoformat() if pagada else None conn.execute( "UPDATE cuotas SET pagada = ?, fecha_pago = ?, interes_mora_pagado = ? WHERE id = ?", (1 if pagada else 0, fecha_pago, interes_mora if pagada else 0, cuota_id), ) conn.commit() row = conn.execute("SELECT prestamo_id FROM cuotas WHERE id = ?", (cuota_id,)).fetchone() return row["prestamo_id"] if row else None finally: conn.close() def contar_cuotas_pendientes(prestamo_id): conn = get_connection() try: row = conn.execute( "SELECT COUNT(*) AS n FROM cuotas WHERE prestamo_id = ? AND pagada = 0", (prestamo_id,) ).fetchone() return row["n"] finally: conn.close() # ---------- Firmas ---------- def crear_firma(prestamo_id, rol, nombre_firmante, ip, hash_acuerdo, documento_identidad=None, firma_dibujada=None): conn = get_connection() try: existente = conn.execute( "SELECT id FROM firmas WHERE prestamo_id = ? AND rol = ?", (prestamo_id, rol) ).fetchone() if existente: return None # ese rol ya firmo, no se puede firmar dos veces cur = conn.execute( """INSERT INTO firmas (prestamo_id, rol, nombre_firmante, ip, hash_acuerdo, documento_identidad, firma_dibujada, fecha_firma) VALUES (?, ?, ?, ?, ?, ?, ?, ?)""", (prestamo_id, rol, nombre_firmante, ip, hash_acuerdo, documento_identidad, firma_dibujada, datetime.now().isoformat()), ) conn.commit() return cur.lastrowid finally: conn.close() def completar_datos_persona( prestamo_id, rol, telefono=None, direccion=None, email=None, ingreso_mensual=None, gastos_mensuales=None, ): """Al firmar por el enlace publico, el deudor o codeudor puede completar sus propios datos de contacto (telefono/direccion/email): solo se actualizan si estaban vacios, para que este enlace publico no pueda usarse para sobreescribir datos ya provistos por el prestamista. El ingreso y gastos mensuales (solo se piden al deudor, para alimentar el futuro scoring de credito) SI se actualizan cada vez que se reportan, porque pueden cambiar de un prestamo a otro en el tiempo.""" prefijo = "prestatario" if rol == "prestatario" else "codeudor" conn = get_connection() try: row = conn.execute( f"SELECT {prefijo}_telefono, {prefijo}_direccion, {prefijo}_email FROM prestamos WHERE id = ?", (prestamo_id,), ).fetchone() if not row: return campos = {} if telefono and not row[f"{prefijo}_telefono"]: campos[f"{prefijo}_telefono"] = telefono if direccion and not row[f"{prefijo}_direccion"]: campos[f"{prefijo}_direccion"] = direccion if email and not row[f"{prefijo}_email"]: campos[f"{prefijo}_email"] = email if rol == "prestatario": if ingreso_mensual is not None: campos["prestatario_ingreso_mensual"] = ingreso_mensual if gastos_mensuales is not None: campos["prestatario_gastos_mensuales"] = gastos_mensuales if campos: set_clause = ", ".join(f"{k} = ?" for k in campos) conn.execute(f"UPDATE prestamos SET {set_clause} WHERE id = ?", (*campos.values(), prestamo_id)) conn.commit() finally: conn.close() # ---------- Estadisticas de administrador ---------- def contar_usuarios_totales(): conn = get_connection() try: return conn.execute("SELECT COUNT(*) AS n FROM usuarios").fetchone()["n"] finally: conn.close() def contar_usuarios_nuevos_desde(fecha_iso): conn = get_connection() try: row = conn.execute( "SELECT COUNT(*) AS n FROM usuarios WHERE creado_en >= ?", (fecha_iso,) ).fetchone() return row["n"] finally: conn.close() def listar_todos_los_prestamos_completos(): """Todos los prestamos de todos los prestamistas (con sus cuotas), solo para uso interno del dashboard de administrador: nunca se expone a un prestamista normal.""" conn = get_connection() try: rows = conn.execute("SELECT * FROM prestamos").fetchall() return [_armar_prestamo(conn, dict(r)) for r in rows] finally: conn.close() def listar_prestamistas_admin(): """Todos los prestamistas registrados, cada uno con el numero de prestamos que ha creado, para el listado del dashboard de administrador.""" conn = get_connection() try: rows = conn.execute( """SELECT u.*, COUNT(p.id) AS numero_prestamos FROM usuarios u LEFT JOIN prestamos p ON p.prestamista_id = u.id GROUP BY u.id ORDER BY u.creado_en DESC""" ).fetchall() return [dict(r) for r in rows] finally: conn.close() def listar_deudores_unicos(): """Personas distintas (por numero de documento) que aparecen como prestatario en al menos un prestamo, sin importar cual prestamista se lo otorgo. Es la vista de 'deudores' del dashboard de administrador, base del futuro scoring de credito propio.""" conn = get_connection() try: rows = conn.execute( """SELECT prestatario_numero_documento AS numero_documento, MAX(prestatario_nombre) AS nombre, COUNT(*) AS numero_prestamos FROM prestamos WHERE prestatario_numero_documento IS NOT NULL AND prestatario_numero_documento != '' GROUP BY prestatario_numero_documento ORDER BY numero_prestamos DESC""" ).fetchall() return [dict(r) for r in rows] finally: conn.close() def historial_prestamos_por_documento(numero_documento): """Historial de prestamos donde la persona identificada por ese numero de documento fue el prestatario, sin importar cual prestamista se lo otorgo. Es la base para el futuro scoring de credito propio: usa el numero de documento como identidad estable, ya que el prestatario no tiene cuenta propia en la plataforma.""" conn = get_connection() try: rows = conn.execute( "SELECT * FROM prestamos WHERE prestatario_numero_documento = ? ORDER BY creado_en", (numero_documento,), ).fetchall() return [_armar_prestamo(conn, dict(r)) for r in rows] finally: conn.close() def eliminar_prestamo(prestamo_id, prestamista_id): """'Eliminar' es un archivado, no un borrado: el préstamo se marca como 'cancelado' (se oculta de la vista normal del prestamista) pero se conserva junto con sus cuotas y firmas para que el histórico y las estadísticas de administrador no pierdan datos.""" conn = get_connection() try: conn.execute( "UPDATE prestamos SET estado = 'cancelado' WHERE id = ? AND prestamista_id = ?", (prestamo_id, prestamista_id), ) conn.commit() finally: conn.close()