1583 lines
52 KiB
Python
Executable File
1583 lines
52 KiB
Python
Executable File
import sqlite3
|
|
import os
|
|
import io
|
|
from flask import Flask, render_template, session, request, redirect, url_for, g, flash, send_file, jsonify
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
from functools import wraps
|
|
from datetime import datetime, timedelta
|
|
import unicodedata
|
|
|
|
from werkzeug.utils import secure_filename
|
|
from werkzeug.exceptions import RequestEntityTooLarge
|
|
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = 'clave-secreta-navbar-app'
|
|
# Límite máximo de subida: 5 MiB
|
|
# app.config['MAX_CONTENT_LENGTH'] = 5 * 1024 * 1024
|
|
|
|
DATABASE = 'database/pyfleet.db'
|
|
# DATABASE = 'database/auto.db'
|
|
|
|
PASSWORD_HASH = generate_password_hash('123')
|
|
|
|
# Duración de inactividad máxima
|
|
SESSION_TIMEOUT = timedelta(hours=4)
|
|
|
|
UPLOAD_FOLDER = 'uploads/documentos'
|
|
ALLOWED_EXTENSIONS = {'pdf'}
|
|
|
|
|
|
# ── Auth ─────────────────────────────────────────────────────
|
|
|
|
def login_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if not session.get('autenticado'):
|
|
return redirect(url_for('login'))
|
|
|
|
# Comprobar timeout de inactividad
|
|
# ultima_actividad = session.get('ultima_actividad')
|
|
# if ultima_actividad:
|
|
# ultima = datetime.fromisoformat(ultima_actividad)
|
|
# if datetime.now() - ultima > SESSION_TIMEOUT:
|
|
# session.clear()
|
|
# flash('Tu sesión ha expirado por inactividad. Vuelve a iniciar sesión.', 'error')
|
|
# return redirect(url_for('login'))
|
|
|
|
# # Actualizar timestamp de última actividad
|
|
# session['ultima_actividad'] = datetime.now().isoformat()
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
@app.route('/api/session')
|
|
def check_session():
|
|
from flask import jsonify
|
|
if session.get('autenticado'):
|
|
# Actualizar timestamp
|
|
session['ultima_actividad'] = datetime.now().isoformat()
|
|
return jsonify({'activa': True})
|
|
return jsonify({'activa': False}), 401
|
|
|
|
|
|
@app.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
session["animate_navbar"] = True
|
|
error = None
|
|
if request.method == 'POST':
|
|
pwd = request.form.get('password', '')
|
|
if check_password_hash(PASSWORD_HASH, pwd):
|
|
session['autenticado'] = True
|
|
return redirect(url_for('vehiculos_list'))
|
|
error = 'Contraseña incorrecta. Inténtalo de nuevo.'
|
|
return render_template('login.html', error=error)
|
|
|
|
@app.route('/logout')
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for('login'))
|
|
|
|
|
|
# ── Funcion normalizar ───────────────────────────────────────────
|
|
|
|
def normalizar(texto):
|
|
# if texto is None:
|
|
# return ''
|
|
texto = str(texto).strip().lower()
|
|
|
|
# Quitar tildes
|
|
texto = unicodedata.normalize('NFD', texto)
|
|
texto = ''.join(
|
|
c for c in texto
|
|
if unicodedata.category(c) != 'Mn'
|
|
)
|
|
|
|
texto = ' '.join(texto.split()) # Convertir varios espacios seguidos en uno
|
|
|
|
return texto
|
|
|
|
# ── DB helpers ────────────────────────────────────────────────
|
|
def get_db():
|
|
if 'db' not in g:
|
|
g.db = sqlite3.connect(DATABASE)
|
|
g.db.create_function("NORM", 1, normalizar)
|
|
g.db.row_factory = sqlite3.Row
|
|
return g.db
|
|
|
|
@app.teardown_appcontext
|
|
def close_db(e=None):
|
|
db = g.pop('db', None)
|
|
if db: db.close()
|
|
|
|
def init_db():
|
|
db = sqlite3.connect(DATABASE)
|
|
db.create_function("NORM", 1, normalizar)
|
|
# db.commit()
|
|
# db.close()
|
|
|
|
# ... tablas existentes ...
|
|
db.execute('''
|
|
CREATE TABLE IF NOT EXISTS documentos (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
vehiculo_id INTEGER NOT NULL REFERENCES vehiculos(id) ON DELETE CASCADE,
|
|
nombre TEXT,
|
|
filename TEXT NOT NULL,
|
|
fecha DATETIME DEFAULT (datetime('now','localtime'))
|
|
)
|
|
''')
|
|
db.commit()
|
|
db.close()
|
|
|
|
|
|
# ── Rutas generales ───────────────────────────────────────────
|
|
@app.route('/', methods=['GET', 'POST'])
|
|
@login_required
|
|
def index():
|
|
if request.method == 'POST':
|
|
texto = request.form.get('mensaje', '').strip()
|
|
if texto:
|
|
get_db().execute('INSERT INTO mensajes (texto) VALUES (?)', (texto,))
|
|
get_db().commit()
|
|
return redirect(url_for('index'))
|
|
return render_template('index.html', page='inicio')
|
|
|
|
@app.route('/explorar', methods=['GET', 'POST'])
|
|
@login_required
|
|
def explorar():
|
|
if request.method == 'POST':
|
|
texto = request.form.get('mensaje', '').strip()
|
|
if texto:
|
|
get_db().execute('INSERT INTO mensajes (texto) VALUES (?)', (texto,))
|
|
get_db().commit()
|
|
return redirect(url_for('explorar'))
|
|
return render_template('explorar.html', page='explorar')
|
|
|
|
@app.route('/contacto', methods=['GET', 'POST'])
|
|
@login_required
|
|
def contacto():
|
|
if request.method == 'POST':
|
|
texto = request.form.get('mensaje', '').strip()
|
|
if texto:
|
|
get_db().execute('INSERT INTO mensajes (texto) VALUES (?)', (texto,))
|
|
get_db().commit()
|
|
return redirect(url_for('contacto'))
|
|
return render_template('contacto.html', page='contacto')
|
|
|
|
@app.route('/limpiar')
|
|
@login_required
|
|
def limpiar():
|
|
ref = request.referrer or url_for('index')
|
|
return redirect(ref)
|
|
|
|
|
|
@app.context_processor
|
|
def inject_globals():
|
|
from datetime import date
|
|
mensajes = get_db().execute(
|
|
'SELECT * FROM mensajes ORDER BY fecha DESC'
|
|
).fetchall()
|
|
return {
|
|
'now': date.today().isoformat(),
|
|
'mensajes': mensajes
|
|
}
|
|
|
|
@app.route('/mensajes/<int:id>/eliminar', methods=['POST'])
|
|
@login_required
|
|
def mensaje_eliminar(id):
|
|
get_db().execute('DELETE FROM mensajes WHERE id=?', (id,))
|
|
get_db().commit()
|
|
return redirect(request.referrer or url_for('index'))
|
|
|
|
@app.route('/mensajes/eliminar_todos', methods=['POST'])
|
|
@login_required
|
|
def mensajes_eliminar_todos():
|
|
get_db().execute('DELETE FROM mensajes')
|
|
get_db().commit()
|
|
return redirect(request.referrer or url_for('index'))
|
|
|
|
# ── CRUD Vehículos ────────────────────────────────────────────
|
|
|
|
@app.route('/vehiculos')
|
|
@login_required
|
|
def vehiculos_list():
|
|
db = get_db()
|
|
q = request.args.get('q', '').strip()
|
|
solo_filas = request.args.get('solo_filas') == '1'
|
|
incl_bajas = request.args.get('incl_bajas', '0') == '1'
|
|
|
|
# Si viene incl_bajas en la URL, actualizamos la sesión
|
|
if 'incl_bajas' in request.args:
|
|
incl_bajas = request.args.get('incl_bajas') == '1'
|
|
session['incl_bajas'] = incl_bajas
|
|
else: # Si no viene en la URL, recuperamos el último estado
|
|
incl_bajas = session.get('incl_bajas', False)
|
|
|
|
terminos = [t.strip() for t in q.split(',') if t.strip()] if q else []
|
|
|
|
# Filtro base de bajas
|
|
filtro_bajas = '' if incl_bajas else "AND (situacion IS NULL OR UPPER(situacion) != 'BAJA')"
|
|
|
|
if terminos:
|
|
condiciones = []
|
|
params = []
|
|
|
|
for t in terminos:
|
|
t_norm = normalizar(t)
|
|
|
|
if t_norm == 'lanzadestellos':
|
|
condiciones.append('lanzadestellos = 1')
|
|
continue
|
|
|
|
like = f'%{t_norm}%'
|
|
condiciones.append('''(
|
|
NORM(m_oficial) LIKE ? OR
|
|
NORM(m_reservada) LIKE ? OR
|
|
NORM(VIN) LIKE ? OR
|
|
NORM(marca) LIKE ? OR
|
|
NORM(modelo) LIKE ? OR
|
|
NORM(poblacion) LIKE ? OR
|
|
NORM(unidad) LIKE ? OR
|
|
NORM(tipo) LIKE ? OR
|
|
NORM(situacion) LIKE ?
|
|
)''')
|
|
params.extend([like] * 9)
|
|
|
|
where = ' AND '.join(condiciones)
|
|
|
|
rows = db.execute(
|
|
f'''SELECT id, m_oficial, m_reservada, marca, modelo, poblacion, unidad,
|
|
tipo, situacion, lanzadestellos
|
|
FROM vehiculos
|
|
WHERE {where} {filtro_bajas}
|
|
ORDER BY id''',
|
|
params
|
|
).fetchall()
|
|
|
|
else: # PARA QUE LA LISTA SALGA EN BLANCO INICIALMENTE
|
|
rows = []
|
|
# else:
|
|
# rows = db.execute(
|
|
# '''SELECT id, m_oficial, m_reservada, marca, modelo, poblacion, unidad, tipo, tipo, situacion
|
|
# FROM vehiculos
|
|
# WHERE situacion IS NULL OR UPPER(situacion) != 'BAJA'
|
|
# ORDER BY id'''
|
|
# ).fetchall()
|
|
|
|
|
|
if solo_filas:
|
|
return render_template('vehiculos/rows.html', rows=rows, q=q, incl_bajas=incl_bajas)
|
|
return render_template('vehiculos/list.html', page='vehiculos', rows=rows, q=q, incl_bajas=incl_bajas)
|
|
|
|
|
|
|
|
|
|
@app.route('/vehiculos/nuevo', methods=['GET', 'POST'])
|
|
@login_required
|
|
def vehiculos_nuevo():
|
|
db = get_db()
|
|
tipo = db.execute('SELECT * FROM tipo ORDER BY tipo').fetchall()
|
|
unidades = db.execute('SELECT * FROM unidades ORDER BY unidad').fetchall()
|
|
poblaciones = db.execute('SELECT * FROM poblaciones ORDER BY poblacion').fetchall()
|
|
|
|
|
|
vehiculo = None
|
|
if request.method == 'POST':
|
|
f = request.form
|
|
lanzadestellos = 1 if f.get('lanzadestellos') else 0
|
|
|
|
cursor = db.execute('''
|
|
INSERT INTO vehiculos (
|
|
m_oficial, m_reservada, m_propia, VIN, poliza, viat, legajo, marca, modelo,
|
|
tipo, color, combustible, plazas, lanzadestellos
|
|
) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
|
''', (
|
|
f.get('m_oficial'), f.get('m_reservada'), f.get('m_propia'), f.get('VIN'),
|
|
f.get('poliza'), f.get('viat'), f.get('legajo') or None,
|
|
f.get('marca'), f.get('modelo'), f.get('tipo'), f.get('color'),
|
|
f.get('combustible'), f.get('plazas'), lanzadestellos
|
|
))
|
|
# nuevo_id = cursor.lastrowid
|
|
db.commit()
|
|
|
|
flash('Vehículo registrado correctamente.', 'ok')
|
|
return redirect(url_for('vehiculos_list'))
|
|
return render_template('vehiculos/form.html', page='vehiculos', vehiculo=vehiculo, tipo=tipo, unidades=unidades, poblaciones=poblaciones)
|
|
|
|
|
|
|
|
@app.route('/vehiculos/<int:id>/editar', methods=['GET', 'POST'])
|
|
@login_required
|
|
def vehiculos_editar(id):
|
|
db = get_db()
|
|
vehiculo = db.execute('SELECT * FROM vehiculos WHERE id=?', (id,)).fetchone()
|
|
tipo = db.execute('SELECT * FROM tipo ORDER BY tipo').fetchall()
|
|
poblaciones = db.execute('SELECT * FROM poblaciones ORDER BY poblacion').fetchall()
|
|
unidades = db.execute('SELECT * FROM unidades ORDER BY unidad').fetchall()
|
|
|
|
historial = db.execute(
|
|
'''
|
|
SELECT h.*, v.m_oficial
|
|
FROM historial h
|
|
JOIN vehiculos v ON h.vehiculo_id = v.id
|
|
WHERE h.vehiculo_id = ?
|
|
ORDER BY h.id DESC
|
|
''',
|
|
(vehiculo['id'],)
|
|
).fetchall() if vehiculo else []
|
|
|
|
# Recoger parámetros GET
|
|
q = request.args.get('q', '')
|
|
desde = request.args.get('desde', '')
|
|
hasta = request.args.get('hasta', '')
|
|
q_itv = request.args.get('q_itv', '')
|
|
origen = request.args.get('origen', '')
|
|
q_viat = request.args.get('q_viat', '')
|
|
|
|
if not vehiculo:
|
|
flash('Vehículo no encontrado.', 'error')
|
|
return redirect(url_for('vehiculos_list'))
|
|
|
|
if request.method == 'POST':
|
|
# Recoger parámetros del POST (campos ocultos)
|
|
q = request.form.get('_q', '')
|
|
desde = request.form.get('_desde', '')
|
|
hasta = request.form.get('_hasta', '')
|
|
q_itv = request.args.get('q_itv', '')
|
|
f = request.form
|
|
origen = request.form.get('_origen', '')
|
|
q_viat = request.form.get('_q_viat', '')
|
|
|
|
lanzadestellos = 1 if f.get('lanzadestellos') else 0
|
|
|
|
db.execute('''
|
|
UPDATE vehiculos SET
|
|
m_oficial=?, m_reservada=?, m_propia=?, VIN=?, poliza=?, viat=?, legajo=?,
|
|
marca=?, modelo=?, tipo=?, color=?, combustible=?, plazas=?, lanzadestellos=?,
|
|
situacion=?, poblacion=?, unidad=?,
|
|
int_operacion=?, int_grupo=?, int_juzgado=?, int_procedimiento=?,
|
|
km=?, km_1=?, itv_ultima=?, itv=?, notas=?
|
|
WHERE id=?
|
|
''', (
|
|
f.get('m_oficial'), f.get('m_reservada'), f.get('m_propia'), f.get('VIN'), f.get('poliza'), f.get('viat') or '', f.get('legajo') or None,
|
|
f.get('marca'), f.get('modelo'), f.get('tipo'), f.get('color'), f.get('combustible'), f.get('plazas'), lanzadestellos,
|
|
f.get('situacion'), f.get('poblacion'), f.get('unidad'),
|
|
f.get('int_operacion'), f.get('int_grupo'), f.get('int_juzgado'), f.get('int_procedimiento'),
|
|
f.get('km') or None, f.get('km_1') or None, f.get('itv_ultima') or None, f.get('itv') or None,
|
|
f.get('notas'),
|
|
id
|
|
))
|
|
db.commit()
|
|
flash('Vehículo actualizado.', 'ok')
|
|
|
|
# Volver a ITV si venía de ahí
|
|
if desde or hasta or q_itv:
|
|
return redirect(f'/itv?desde={desde}&hasta={hasta}&q_itv={q_itv}')
|
|
elif origen == 'viat':
|
|
return redirect(f'/viat?q={q_viat}')
|
|
return redirect(url_for('vehiculos_list', q=q) if q else url_for('vehiculos_list'))
|
|
|
|
documentos = db.execute(
|
|
'SELECT * FROM documentos WHERE vehiculo_id=? ORDER BY fecha DESC',
|
|
(id,)
|
|
).fetchall()
|
|
|
|
matricula = vehiculo['m_oficial'] or vehiculo['m_reservada'] or ''
|
|
return render_template('vehiculos/form.html',
|
|
page='vehiculos',
|
|
vehiculo=vehiculo,
|
|
tipo=tipo,
|
|
poblaciones=poblaciones,
|
|
unidades=unidades,
|
|
historial=historial,
|
|
matricula=matricula,
|
|
q=q,
|
|
desde=desde,
|
|
hasta=hasta,
|
|
q_itv=q_itv,
|
|
origen=origen,
|
|
q_viat=q_viat,
|
|
documentos=documentos)
|
|
|
|
|
|
@app.route('/vehiculos/<int:id>/eliminar', methods=['POST'])
|
|
@login_required
|
|
def vehiculos_eliminar(id):
|
|
db = get_db()
|
|
db.execute('DELETE FROM vehiculos WHERE id=?', (id,))
|
|
db.commit()
|
|
flash('Vehículo eliminado.', 'ok')
|
|
q = request.form.get('_q', '')
|
|
return redirect(url_for('vehiculos_list', q=q) if q else url_for('vehiculos_list'))
|
|
|
|
|
|
@app.route('/vehiculos/<int:vid>/historial/nuevo', methods=['POST'])
|
|
@login_required
|
|
def historial_nuevo(vid):
|
|
db = get_db()
|
|
|
|
# Recoger TODOS los parámetros de navegación del POST
|
|
q = request.form.get('_q', '')
|
|
desde = request.form.get('_desde', '')
|
|
hasta = request.form.get('_hasta', '')
|
|
q_itv = request.form.get('_q_itv', '')
|
|
origen = request.form.get('_origen', '')
|
|
q_viat = request.form.get('_q_viat', '') # ← POST no GET
|
|
|
|
# Datos del historial
|
|
h_fecha = request.form.get('h_fecha') or None
|
|
h_vehiculo_id = request.form.get('h_vehiculo_id', '').strip()
|
|
h_matricula = request.form.get('h_matricula', '').strip()
|
|
h_situacion = request.form.get('h_situacion', '').strip()
|
|
h_poblacion_id = request.form.get('h_poblacion', '').strip()
|
|
h_unidad = request.form.get('h_unidad', '').strip()
|
|
h_grupo = request.form.get('h_grupo', '').strip()
|
|
h_observaciones = request.form.get('h_observaciones', '').strip()
|
|
|
|
# Obtener nombre de población
|
|
poblacion_nombre = ''
|
|
if h_poblacion_id:
|
|
row = db.execute('SELECT poblacion FROM poblaciones WHERE id=?', (h_poblacion_id,)).fetchone()
|
|
if row:
|
|
poblacion_nombre = row['poblacion']
|
|
|
|
db.execute('''
|
|
INSERT INTO historial (fecha, vehiculo_id, matricula, situacion, poblacion, unidad, grupo, observaciones)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
''', (h_fecha, h_vehiculo_id, h_matricula, h_situacion, poblacion_nombre, h_unidad, h_grupo, h_observaciones))
|
|
|
|
# Actualizar situacion del vehiculo
|
|
if h_situacion.upper() == 'BAJA':
|
|
db.execute("UPDATE vehiculos SET situacion='BAJA', poblacion='', unidad='' WHERE id=?", (vid,))
|
|
else:
|
|
db.execute(
|
|
"UPDATE vehiculos SET situacion=?, poblacion=?, unidad=?, grupo=? WHERE id=?",
|
|
(h_situacion, poblacion_nombre, h_unidad, h_grupo, vid)
|
|
)
|
|
|
|
db.commit()
|
|
flash('Entrada añadida al historial.', 'ok')
|
|
|
|
# Redirect según origen
|
|
if origen == 'viat':
|
|
return redirect(f'/vehiculos/{vid}/editar?origen=viat&q_viat={q_viat}')
|
|
if desde or hasta or q_itv:
|
|
return redirect(f'/vehiculos/{vid}/editar?desde={desde}&hasta={hasta}&q_itv={q_itv}')
|
|
return redirect(url_for('vehiculos_editar', id=vid, q=q))
|
|
|
|
|
|
|
|
|
|
|
|
@app.route('/vehiculos/<int:vid>/historial/<int:hid>/eliminar', methods=['POST'])
|
|
@login_required
|
|
def historial_eliminar(vid, hid):
|
|
q = request.form.get('_q', '')
|
|
desde = request.form.get('_desde', '')
|
|
hasta = request.form.get('_hasta', '')
|
|
|
|
db = get_db()
|
|
|
|
# Recoger parámetros de navegación
|
|
q = request.form.get('_q', '')
|
|
desde = request.form.get('_desde', '')
|
|
hasta = request.form.get('_hasta', '')
|
|
q_itv = request.form.get('_q_itv', '')
|
|
origen = request.form.get('_origen', '')
|
|
q_viat = request.form.get('_q_viat', '') # ← POST no GET
|
|
|
|
|
|
# 1. Comprobar que el vehículo existe
|
|
vehiculo = db.execute('''
|
|
SELECT id, m_oficial
|
|
FROM vehiculos
|
|
WHERE id = ?
|
|
''', (vid,)).fetchone()
|
|
|
|
if not vehiculo:
|
|
flash('El vehículo no existe.', 'error')
|
|
return redirect(url_for('vehiculos'))
|
|
|
|
# 2. Comprobar que el registro de historial pertenece realmente a este vehículo
|
|
asiento = db.execute('''
|
|
SELECT id, vehiculo_id, situacion
|
|
FROM historial
|
|
WHERE id = ?
|
|
AND vehiculo_id = ?
|
|
''', (hid, vid)).fetchone()
|
|
|
|
if not asiento:
|
|
flash('El registro de historial no existe o no pertenece a este vehículo.', 'error')
|
|
return redirect(url_for('vehiculo_historial', vid=vid))
|
|
|
|
# 3. Eliminar el registro
|
|
db.execute('''
|
|
DELETE FROM historial
|
|
WHERE id = ?
|
|
AND vehiculo_id = ?
|
|
''', (hid, vid))
|
|
|
|
# 4. Buscar el ÚLTIMO registro que queda para este vehículo
|
|
ultimo = db.execute('''
|
|
SELECT
|
|
id,
|
|
situacion,
|
|
poblacion,
|
|
unidad
|
|
FROM historial
|
|
WHERE vehiculo_id = ?
|
|
ORDER BY id DESC
|
|
LIMIT 1
|
|
''', (vid,)).fetchone()
|
|
|
|
# 5. Si queda historial, actualizar vehiculos con los datos del último registro
|
|
if ultimo:
|
|
|
|
situacion = (ultimo['situacion'] or '').strip().upper()
|
|
poblacion = (ultimo['poblacion'] or '').strip()
|
|
unidad = (ultimo['unidad'] or '').strip()
|
|
|
|
db.execute('''
|
|
UPDATE vehiculos
|
|
SET
|
|
situacion = ?,
|
|
poblacion = ?,
|
|
unidad = ?
|
|
WHERE id = ?
|
|
''', (
|
|
situacion,
|
|
poblacion,
|
|
unidad,
|
|
vid
|
|
))
|
|
|
|
# 6. Si no queda ningún historial, limpiar el vehículo
|
|
else:
|
|
db.execute('''
|
|
UPDATE vehiculos
|
|
SET
|
|
situacion = NULL,
|
|
poblacion = '',
|
|
unidad = ''
|
|
WHERE id = ?
|
|
''', (vid,))
|
|
|
|
db.commit()
|
|
|
|
flash('Entrada eliminada del historial.', 'ok')
|
|
|
|
# Redirect según origen
|
|
if origen == 'viat':
|
|
return redirect(f'/vehiculos/{vid}/editar?origen=viat&q_viat={q_viat}')
|
|
if desde or hasta or q_itv:
|
|
return redirect(f'/vehiculos/{vid}/editar?desde={desde}&hasta={hasta}&q_itv={q_itv}')
|
|
|
|
url = f'/vehiculos/{vid}/editar'
|
|
if q:
|
|
url += f'?q={q}'
|
|
return redirect(url)
|
|
|
|
|
|
@app.route('/vehiculos/exportar')
|
|
@login_required
|
|
def vehiculos_exportar():
|
|
import io, openpyxl
|
|
from openpyxl.styles import Font, PatternFill, Alignment
|
|
from flask import send_file
|
|
|
|
db = get_db()
|
|
q = request.args.get('q', '').strip()
|
|
|
|
incl_bajas = request.args.get('incl_bajas', '0') == '1'
|
|
filtro_bajas = '' if incl_bajas else "AND (situacion IS NULL OR UPPER(situacion) != 'BAJA')"
|
|
terminos = [t.strip() for t in q.split(',') if t.strip()] if q else []
|
|
|
|
|
|
# terminos = [t.strip() for t in q.split(',') if t.strip()] if q else []
|
|
|
|
# Filtro base de bajas
|
|
filtro_bajas = '' if incl_bajas else "AND (situacion IS NULL OR UPPER(situacion) != 'BAJA')"
|
|
|
|
if terminos:
|
|
condiciones = []
|
|
params = []
|
|
|
|
for t in terminos:
|
|
t_norm = normalizar(t)
|
|
|
|
if t_norm == 'lanzadestellos':
|
|
condiciones.append('lanzadestellos = 1')
|
|
continue
|
|
|
|
like = f'%{t_norm}%'
|
|
condiciones.append('''(
|
|
NORM(m_oficial) LIKE ? OR
|
|
NORM(m_reservada) LIKE ? OR
|
|
NORM(VIN) LIKE ? OR
|
|
NORM(marca) LIKE ? OR
|
|
NORM(modelo) LIKE ? OR
|
|
NORM(poblacion) LIKE ? OR
|
|
NORM(unidad) LIKE ? OR
|
|
NORM(denominacion) LIKE ? OR
|
|
NORM(situacion) LIKE ?
|
|
)''')
|
|
params.extend([like] * 9)
|
|
|
|
where = ' AND '.join(condiciones)
|
|
|
|
rows = db.execute(
|
|
f'''SELECT id, m_oficial, m_reservada, marca, modelo, poblacion, unidad,
|
|
denominacion, tipo, situacion, lanzadestellos
|
|
FROM vehiculos
|
|
WHERE {where} {filtro_bajas}
|
|
ORDER BY id''',
|
|
params
|
|
).fetchall()
|
|
else:
|
|
rows = db.execute(
|
|
'''SELECT m_oficial, m_reservada, marca, modelo, poblacion, unidad, grupo, tipo, situacion
|
|
FROM vehiculos
|
|
WHERE situacion IS NULL OR UPPER(situacion) != 'BAJA'
|
|
ORDER BY id'''
|
|
).fetchall()
|
|
|
|
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
ws.title = 'Vehículos'
|
|
|
|
headers = ['M. Oficial', 'M. Reservada', 'Marca', 'Modelo',
|
|
'Poblacion', 'Unidad', 'Grupo', 'Denominación', 'Situación' ]
|
|
header_fill = PatternFill('solid', fgColor='5C6EF8')
|
|
header_font = Font(bold=True, color='FFFFFF', size=11)
|
|
|
|
for col, h in enumerate(headers, 1):
|
|
cell = ws.cell(row=1, column=col, value=h)
|
|
cell.fill = header_fill
|
|
cell.font = header_font
|
|
cell.alignment = Alignment(horizontal='center', vertical='center')
|
|
ws.row_dimensions[1].height = 22
|
|
|
|
for row in rows:
|
|
ws.append(list(row))
|
|
|
|
col_widths = [8, 14, 14, 16, 16, 24, 16, 16, 18, 10, 14]
|
|
for i, w in enumerate(col_widths, 1):
|
|
ws.column_dimensions[ws.cell(1, i).column_letter].width = w
|
|
|
|
zebra = PatternFill('solid', fgColor='EEF0F8')
|
|
for row_idx in range(2, ws.max_row + 1):
|
|
if row_idx % 2 == 0:
|
|
for col_idx in range(1, len(headers) + 1):
|
|
ws.cell(row_idx, col_idx).fill = zebra
|
|
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
buf.seek(0)
|
|
|
|
nombre_q = '_'.join(terminos).replace(' ', '-') if terminos else ''
|
|
filename = f'vehiculos{"_" + nombre_q if nombre_q else ""}.xlsx'
|
|
return send_file(buf, as_attachment=True,
|
|
download_name=filename,
|
|
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
|
|
|
|
|
|
|
|
|
|
# UPLOADS FICHAS TECNICAS PDF
|
|
def allowed_file(filename):
|
|
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
|
|
|
|
|
|
# @app.errorhandler(RequestEntityTooLarge)
|
|
# def handle_file_too_large(e):
|
|
# flash('Error: El archivo pdf no puede superar los 5 MB.', 'error')
|
|
# return redirect(request.referrer or '/')
|
|
|
|
|
|
@app.route('/vehiculos/<int:id>/documentos/upload', methods=['POST'])
|
|
@login_required
|
|
def documentos_upload(id):
|
|
q = request.form.get('_q', '')
|
|
incl_bajas = request.form.get('_incl_bajas', '0')
|
|
desde = request.form.get('_desde', '')
|
|
hasta = request.form.get('_hasta', '')
|
|
q_itv = request.form.get('_q_itv', '')
|
|
origen = request.form.get('_origen', '')
|
|
q_viat = request.form.get('_q_viat', '')
|
|
nombre = request.form.get('nombre_doc', '').strip() + '.pdf'
|
|
|
|
|
|
if 'pdf' not in request.files:
|
|
flash('No se seleccionó ningún archivo.', 'error')
|
|
return redirect(f'/vehiculos/{id}/editar')
|
|
|
|
file = request.files['pdf']
|
|
if file.filename == '' or not allowed_file(file.filename):
|
|
flash('Solo se permiten archivos PDF.', 'error')
|
|
return redirect(f'/vehiculos/{id}/editar')
|
|
|
|
# Comprobar tamaño máximo: 5 MiB
|
|
MAX_FILE_SIZE = 5 * 1024 * 1024
|
|
|
|
file.seek(0, os.SEEK_END)
|
|
file_size = file.tell()
|
|
file.seek(0)
|
|
|
|
if file_size > MAX_FILE_SIZE:
|
|
flash('El archivo PDF no puede superar los 5 MB.', 'error')
|
|
return redirect(f'/vehiculos/{id}/editar?q={q}')
|
|
|
|
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
|
|
|
|
# Nombre único: vehiculo_id_timestamp.pdf
|
|
import time
|
|
filename = f'v{id}_{int(time.time())}.pdf'
|
|
file.save(os.path.join(UPLOAD_FOLDER, filename))
|
|
|
|
# Si no puso nombre, usar el nombre original del archivo
|
|
if not nombre:
|
|
nombre = secure_filename(file.filename)
|
|
|
|
db = get_db()
|
|
db.execute(
|
|
'INSERT INTO documentos (vehiculo_id, nombre, filename) VALUES (?,?,?)',
|
|
(id, nombre, filename)
|
|
)
|
|
db.commit()
|
|
flash(f'Documento «{nombre}» subido correctamente.', 'ok')
|
|
|
|
url = f'/vehiculos/{id}/editar?q={q}'
|
|
if origen == 'viat':
|
|
url = f'/vehiculos/{id}/editar?origen=viat&q_viat={q_viat}'
|
|
elif desde or hasta or q_itv:
|
|
url = f'/vehiculos/{id}/editar?desde={desde}&hasta={hasta}&q_itv={q_itv}'
|
|
return redirect(url)
|
|
|
|
|
|
@app.route('/vehiculos/documentos/<int:doc_id>/ver')
|
|
@login_required
|
|
def documentos_ver(doc_id):
|
|
from flask import send_from_directory
|
|
db = get_db()
|
|
doc = db.execute('SELECT * FROM documentos WHERE id=?', (doc_id,)).fetchone()
|
|
if not doc:
|
|
flash('Documento no encontrado.', 'error')
|
|
return redirect(url_for('vehiculos_list'))
|
|
return send_from_directory(UPLOAD_FOLDER, doc['filename'], as_attachment=False)
|
|
|
|
|
|
@app.route('/vehiculos/documentos/<int:doc_id>/descargar')
|
|
@login_required
|
|
def documentos_descargar(doc_id):
|
|
from flask import send_from_directory
|
|
db = get_db()
|
|
doc = db.execute('SELECT * FROM documentos WHERE id=?', (doc_id,)).fetchone()
|
|
if not doc:
|
|
flash('Documento no encontrado.', 'error')
|
|
return redirect(url_for('vehiculos_list'))
|
|
return send_from_directory(UPLOAD_FOLDER, doc['filename'],
|
|
as_attachment=True,
|
|
download_name=doc['nombre'])
|
|
# download_name=doc['nombre'] + '.pdf'
|
|
|
|
@app.route('/vehiculos/documentos/<int:doc_id>/eliminar', methods=['POST'])
|
|
@login_required
|
|
def documentos_eliminar(doc_id):
|
|
db = get_db()
|
|
vid = request.form.get('vid', '')
|
|
q = request.form.get('_q', '')
|
|
incl_bajas = request.form.get('_incl_bajas', '0')
|
|
|
|
doc = db.execute('SELECT * FROM documentos WHERE id=?', (doc_id,)).fetchone()
|
|
if doc:
|
|
filepath = os.path.join(UPLOAD_FOLDER, doc['filename'])
|
|
if os.path.exists(filepath):
|
|
os.remove(filepath)
|
|
db.execute('DELETE FROM documentos WHERE id=?', (doc_id,))
|
|
db.commit()
|
|
flash('Documento eliminado.', 'ok')
|
|
|
|
return redirect(f'/vehiculos/{vid}/editar?q={q}&incl_bajas={incl_bajas}')
|
|
|
|
|
|
|
|
# Ruta API que devuelve unidades según población
|
|
@app.route('/api/unidades/<int:poblacion_id>')
|
|
@login_required
|
|
def api_unidades(poblacion_id):
|
|
from flask import jsonify
|
|
db = get_db()
|
|
rows = db.execute(
|
|
'SELECT id, unidad FROM unidades WHERE poblacion_id=? ORDER BY unidad',
|
|
(poblacion_id,)
|
|
).fetchall()
|
|
return jsonify([{'id': r['id'], 'unidad': r['unidad']} for r in rows])
|
|
|
|
|
|
|
|
|
|
@app.route('/itv')
|
|
@login_required
|
|
def itv_list():
|
|
db = get_db()
|
|
desde = request.args.get('desde', '').strip()
|
|
hasta = request.args.get('hasta', '').strip()
|
|
q_itv = request.args.get('q_itv', '').strip()
|
|
rows = []
|
|
buscado = False
|
|
|
|
if desde or hasta or q_itv:
|
|
buscado = True
|
|
desde_q = desde or '1900-01-01'
|
|
hasta_q = hasta or '2999-12-31'
|
|
terminos = [t.strip() for t in q_itv.split(',') if t.strip()] if q_itv else []
|
|
|
|
if terminos:
|
|
condiciones = []
|
|
params = [desde_q, hasta_q]
|
|
for t in terminos:
|
|
like = f'%{t}%'
|
|
condiciones.append('''(
|
|
poblacion LIKE ? OR
|
|
unidad LIKE ? OR
|
|
tipo LIKE ?
|
|
)''')
|
|
params.extend([like] * 3)
|
|
where_q = ' AND '.join(condiciones)
|
|
rows = db.execute(f'''
|
|
SELECT id, m_oficial, m_reservada, marca, modelo, tipo,
|
|
tipo, unidad, poblacion, itv
|
|
FROM vehiculos
|
|
WHERE situacion !='BAJA'
|
|
AND itv IS NOT NULL AND itv != ''
|
|
AND itv BETWEEN ? AND ?
|
|
AND {where_q}
|
|
ORDER BY itv ASC
|
|
''', params).fetchall()
|
|
else:
|
|
rows = db.execute('''
|
|
SELECT id, m_oficial, m_reservada, marca, modelo, tipo,
|
|
tipo, unidad, poblacion, itv
|
|
FROM vehiculos
|
|
WHERE situacion !='BAJA'
|
|
AND itv IS NOT NULL AND itv != ''
|
|
AND itv BETWEEN ? AND ?
|
|
ORDER BY itv ASC
|
|
''', (desde_q, hasta_q)).fetchall()
|
|
|
|
return render_template('itv/list.html', page='itv',
|
|
rows=rows, desde=desde, hasta=hasta,
|
|
q_itv=q_itv, buscado=buscado)
|
|
|
|
|
|
@app.route('/itv/exportar')
|
|
@login_required
|
|
def itv_exportar():
|
|
import io, openpyxl
|
|
from openpyxl.styles import Font, PatternFill, Alignment
|
|
from flask import send_file
|
|
|
|
db = get_db()
|
|
desde = request.args.get('desde', '').strip()
|
|
hasta = request.args.get('hasta', '').strip()
|
|
q_itv = request.args.get('q_itv', '').strip()
|
|
|
|
desde_q = desde or '1900-01-01'
|
|
hasta_q = hasta or '2999-12-31'
|
|
terminos = [t.strip() for t in q_itv.split(',') if t.strip()] if q_itv else []
|
|
|
|
if terminos:
|
|
condiciones = []
|
|
params = [desde_q, hasta_q]
|
|
for t in terminos:
|
|
like = f'%{t}%'
|
|
condiciones.append('''(
|
|
poblacion LIKE ? OR
|
|
unidad LIKE ? OR
|
|
tipo LIKE ?
|
|
)''')
|
|
params.extend([like] * 3)
|
|
where_q = ' AND '.join(condiciones)
|
|
rows = db.execute(f'''
|
|
SELECT m_oficial, m_reservada, marca, modelo, tipo,
|
|
tipo, unidad, poblacion, itv
|
|
FROM vehiculos
|
|
WHERE situacion !='BAJA'
|
|
AND itv IS NOT NULL AND itv != ''
|
|
AND itv BETWEEN ? AND ?
|
|
AND {where_q}
|
|
ORDER BY itv ASC
|
|
''', params).fetchall()
|
|
else:
|
|
rows = db.execute('''
|
|
SELECT m_oficial, m_reservada, marca, modelo, tipo,
|
|
tipo, unidad, poblacion, itv
|
|
FROM vehiculos
|
|
WHERE situacion !='BAJA'
|
|
AND itv IS NOT NULL AND itv != ''
|
|
AND itv BETWEEN ? AND ?
|
|
ORDER BY itv ASC
|
|
''', (desde_q, hasta_q)).fetchall()
|
|
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
ws.title = 'ITV'
|
|
|
|
headers = ['M. Oficial', 'M. Reservada', 'Marca', 'Modelo',
|
|
'Denominación', 'Tipo', 'Unidad', 'Población', 'Fecha ITV']
|
|
header_fill = PatternFill('solid', fgColor='5C6EF8')
|
|
header_font = Font(bold=True, color='FFFFFF', size=11)
|
|
|
|
for col, h in enumerate(headers, 1):
|
|
cell = ws.cell(row=1, column=col, value=h)
|
|
cell.fill = header_fill
|
|
cell.font = header_font
|
|
cell.alignment = Alignment(horizontal='center', vertical='center')
|
|
ws.row_dimensions[1].height = 22
|
|
|
|
for row in rows:
|
|
ws.append(list(row))
|
|
|
|
col_widths = [8, 14, 14, 16, 16, 24, 14, 18, 18, 14]
|
|
for i, w in enumerate(col_widths, 1):
|
|
ws.column_dimensions[ws.cell(1, i).column_letter].width = w
|
|
|
|
zebra = PatternFill('solid', fgColor='EEF0F8')
|
|
for row_idx in range(2, ws.max_row + 1):
|
|
if row_idx % 2 == 0:
|
|
for col_idx in range(1, len(headers) + 1):
|
|
ws.cell(row_idx, col_idx).fill = zebra
|
|
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
buf.seek(0)
|
|
nombre_q = '_'.join(terminos).replace(' ', '-') if terminos else ''
|
|
filename = f'itv_{desde}__{hasta}{"_" + nombre_q if nombre_q else ""}.xlsx'
|
|
return send_file(buf, as_attachment=True,
|
|
download_name=filename,
|
|
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
|
|
|
|
|
|
|
|
# ── CRUD Poblaciones ──────────────────────────────────────────
|
|
@app.route('/poblaciones')
|
|
@login_required
|
|
def poblaciones_list():
|
|
db = get_db()
|
|
rows = db.execute('SELECT * FROM poblaciones ORDER BY poblacion').fetchall()
|
|
return render_template('tablas/poblaciones.html', page='poblaciones', rows=rows)
|
|
|
|
@app.route('/poblaciones/nueva', methods=['POST'])
|
|
@login_required
|
|
def poblaciones_nueva():
|
|
nombre = request.form.get('poblacion', '').strip()
|
|
if nombre:
|
|
try:
|
|
get_db().execute('INSERT INTO poblaciones (poblacion) VALUES (?)', (nombre,))
|
|
get_db().commit()
|
|
flash(f'Población «{nombre}» añadida.', 'ok')
|
|
except:
|
|
flash('Esa población ya existe.', 'error')
|
|
return redirect(url_for('poblaciones_list'))
|
|
|
|
|
|
@app.route("/editar_poblaciones", methods=["POST"])
|
|
def editar_poblaciones():
|
|
print(request.form)
|
|
|
|
accion = request.form.get("action")
|
|
db = get_db()
|
|
cur = db.cursor()
|
|
|
|
if accion == "edit":
|
|
|
|
id = request.form["id"]
|
|
poblacion = request.form.get("poblacion")
|
|
|
|
cur.execute("""UPDATE poblaciones SET poblacion=? WHERE id=?""",(poblacion, id))
|
|
db.commit()
|
|
flash(f'Poblacion «{poblacion}» modificada.', 'ok')
|
|
|
|
elif accion == "delete":
|
|
id = request.form["id"]
|
|
cur.execute("DELETE FROM poblaciones WHERE id=?",(id,))
|
|
db.commit()
|
|
flash(f'Poblacion eliminada.', 'ok')
|
|
|
|
db.close()
|
|
return jsonify({
|
|
"action": accion,
|
|
"id": request.form["id"]
|
|
})
|
|
|
|
|
|
# @app.route('/poblaciones/<int:id>/eliminar', methods=['POST'])
|
|
# @login_required
|
|
# def poblaciones_eliminar(id):
|
|
# db = get_db()
|
|
# db.execute('DELETE FROM poblaciones WHERE id=?', (id,))
|
|
# db.commit()
|
|
# flash('Población eliminada.', 'ok')
|
|
|
|
|
|
# ── CRUD Unidades ─────────────────────────────────────────────
|
|
@app.route('/unidades')
|
|
@login_required
|
|
def unidades_list():
|
|
db = get_db()
|
|
rows = db.execute('''
|
|
SELECT u.*, p.poblacion
|
|
FROM unidades u
|
|
LEFT JOIN poblaciones p ON u.poblacion_id = p.id
|
|
ORDER BY p.poblacion, u.unidad
|
|
''').fetchall()
|
|
poblaciones = db.execute('SELECT * FROM poblaciones ORDER BY poblacion').fetchall()
|
|
return render_template('tablas/unidades.html', page='unidades', rows=rows, poblaciones=poblaciones)
|
|
|
|
@app.route('/unidades/nueva', methods=['POST'])
|
|
@login_required
|
|
def unidades_nueva():
|
|
unidad = request.form.get('unidad', '').strip()
|
|
poblacion_id = request.form.get('poblacion_id', '').strip()
|
|
if unidad and poblacion_id:
|
|
get_db().execute(
|
|
'INSERT INTO unidades (unidad, poblacion_id) VALUES (?,?)',
|
|
(unidad, poblacion_id)
|
|
)
|
|
get_db().commit()
|
|
flash(f'Unidad «{unidad}» añadida.', 'ok')
|
|
return redirect(url_for('unidades_list'))
|
|
|
|
|
|
@app.route("/editar_unidades", methods=["POST"])
|
|
def editar_unidades():
|
|
print(request.form)
|
|
|
|
accion = request.form.get("action")
|
|
db = get_db()
|
|
cur = db.cursor()
|
|
|
|
if accion == "edit":
|
|
|
|
id = request.form["id"]
|
|
poblacion = request.form.get("poblacion")
|
|
unidad = request.form.get("unidad")
|
|
|
|
fila = db.execute(
|
|
'SELECT id FROM poblaciones WHERE poblacion=?',
|
|
(poblacion,)
|
|
).fetchone()
|
|
|
|
if fila is None:
|
|
return jsonify({"error": "La población no existe"}), 400
|
|
|
|
poblacion_id = fila["id"]
|
|
|
|
cur.execute(
|
|
"""UPDATE unidades
|
|
SET poblacion_id=?, unidad=?
|
|
WHERE id=?""",
|
|
(poblacion_id, unidad, id)
|
|
)
|
|
|
|
db.commit()
|
|
flash(f'Unidad «{unidad}» modificada.', 'ok')
|
|
|
|
elif accion == "delete":
|
|
id = request.form["id"]
|
|
cur.execute("DELETE FROM unidades WHERE id=?",(id,))
|
|
db.commit()
|
|
flash(f'Unidad eliminada.', 'ok')
|
|
|
|
db.close()
|
|
|
|
return jsonify({
|
|
"action": accion,
|
|
"id": request.form["id"]
|
|
})
|
|
|
|
|
|
# @app.route('/unidades/<int:id>/eliminar', methods=['POST'])
|
|
# @login_required
|
|
# def unidades_eliminar(id):
|
|
# db = get_db()
|
|
# db.execute('DELETE FROM unidades WHERE id=?', (id,))
|
|
# db.commit()
|
|
# flash('Unidad eliminada.', 'ok')
|
|
# return redirect(url_for('unidades_list'))
|
|
|
|
|
|
############### API PARA USAR TABULATOR ######################################
|
|
|
|
# @app.route("/api/unidades/<int:unidad_id>", methods=["PATCH"])
|
|
# @login_required
|
|
# def actualizar_unidad(unidad_id):
|
|
|
|
# datos = request.get_json(silent=True) or {}
|
|
|
|
# campo = datos.get("campo")
|
|
# valor = datos.get("valor")
|
|
|
|
# # Solo permitimos estos campos
|
|
# if campo not in ("poblacion_id", "unidad"):
|
|
# return jsonify({
|
|
# "error": "Campo no permitido"
|
|
# }), 400
|
|
|
|
# db = get_db()
|
|
|
|
# # Comprobar que existe la unidad
|
|
# unidad_actual = db.execute(
|
|
# "SELECT id FROM unidades WHERE id=?",
|
|
# (unidad_id,)
|
|
# ).fetchone()
|
|
|
|
# if unidad_actual is None:
|
|
|
|
# return jsonify({
|
|
# "error": "La unidad no existe"
|
|
# }), 404
|
|
|
|
# # CAMBIO DE POBLACIÓN
|
|
# if campo == "poblacion_id":
|
|
|
|
# try:
|
|
# poblacion_id = int(valor)
|
|
# except (ValueError, TypeError):
|
|
|
|
# return jsonify({
|
|
# "error": "Población no válida"
|
|
# }), 400
|
|
|
|
|
|
# poblacion = db.execute(
|
|
# """ SELECT id, poblacion FROM poblaciones WHERE id=? """,
|
|
# (poblacion_id,)
|
|
# ).fetchone()
|
|
|
|
# if poblacion is None:
|
|
|
|
# return jsonify({
|
|
# "error": "La población no existe"
|
|
# }), 400
|
|
|
|
# valor = poblacion_id
|
|
|
|
# # CAMBIO DE UNIDAD
|
|
# elif campo == "unidad":
|
|
|
|
# valor = str(valor or "").strip()
|
|
|
|
# if not valor:
|
|
|
|
# return jsonify({
|
|
# "error": "El nombre de la unidad no puede estar vacío"
|
|
# }), 400
|
|
|
|
# # UPDATE
|
|
# db.execute(f""" UPDATE unidades SET {campo}=? WHERE id=?""",
|
|
# (valor, unidad_id)
|
|
# )
|
|
|
|
# db.commit()
|
|
|
|
# # DEVOLVER EL REGISTRO ACTUALIZADO
|
|
# fila = db.execute(
|
|
# """
|
|
# SELECT
|
|
# u.id,
|
|
# u.poblacion_id,
|
|
# u.unidad,
|
|
# p.poblacion
|
|
# FROM unidades u
|
|
# LEFT JOIN poblaciones p
|
|
# ON u.poblacion_id = p.id
|
|
# WHERE u.id=? """,
|
|
# (unidad_id,)
|
|
# ).fetchone()
|
|
|
|
|
|
# return jsonify({
|
|
# "id": fila["id"],
|
|
# "poblacion_id": fila["poblacion_id"],
|
|
# "poblacion": fila["poblacion"],
|
|
# "unidad": fila["unidad"]
|
|
# })
|
|
|
|
|
|
# ── CRUD tipo ─────────────────────────────────────────
|
|
@app.route('/tipo')
|
|
@login_required
|
|
def tipo_list():
|
|
db = get_db()
|
|
rows = db.execute('SELECT * FROM tipo ORDER BY tipo').fetchall()
|
|
return render_template('tablas/tipo.html', page='tipo', rows=rows)
|
|
|
|
@app.route('/tipo/nueva', methods=['POST'])
|
|
@login_required
|
|
def tipo_nueva():
|
|
tipo = request.form.get('tipo', '').strip()
|
|
literal = request.form.get('literal', '').strip()
|
|
if tipo:
|
|
try:
|
|
get_db().execute(
|
|
'INSERT INTO tipo (tipo) VALUES (?)',
|
|
(tipo)
|
|
)
|
|
get_db().commit()
|
|
flash(f'Tipo vehículo «{tipo}» añadido.', 'ok')
|
|
except:
|
|
flash('Ese tipo ya existe.', 'error')
|
|
return redirect(url_for('tipo_list'))
|
|
|
|
|
|
|
|
@app.route("/editar_tipo", methods=["POST"])
|
|
def editar_tipo():
|
|
print(request.form)
|
|
|
|
accion = request.form.get("action")
|
|
db = get_db()
|
|
cur = db.cursor()
|
|
|
|
if accion == "edit":
|
|
|
|
id = request.form["id"]
|
|
tipo = request.form.get("tipo")
|
|
|
|
cur.execute("UPDATE tipo SET tipo=? WHERE id=?",
|
|
(tipo, id)
|
|
)
|
|
|
|
db.commit()
|
|
flash(f'Tipo vehículo «{tipo}» modificado.', 'ok')
|
|
|
|
elif accion == "delete":
|
|
id = request.form["id"]
|
|
cur.execute("DELETE FROM tipo WHERE id=?",(id,))
|
|
db.commit()
|
|
flash(f'Tipo vehículo eliminado.', 'ok')
|
|
|
|
db.close()
|
|
|
|
return jsonify({
|
|
"action": accion,
|
|
"id": request.form["id"]
|
|
})
|
|
|
|
|
|
# @app.route('/tipo/<int:id>/eliminar', methods=['POST'])
|
|
# @login_required
|
|
# def tipo_eliminar(id):
|
|
# db = get_db()
|
|
# db.execute('DELETE FROM tipo WHERE id=?', (id,))
|
|
# db.commit()
|
|
# flash('Denominación eliminada.', 'ok')
|
|
# return redirect(url_for('tipo_list'))
|
|
|
|
|
|
# ── TEMPLATE VIA-T ──────────────────────────────────────────────
|
|
|
|
@app.route('/viat')
|
|
@login_required
|
|
def viat_list():
|
|
db = get_db()
|
|
q = request.args.get('q', '').strip()
|
|
solo_filas = request.args.get('solo_filas') == '1'
|
|
terminos = [t.strip() for t in q.split(',') if t.strip()] if q else []
|
|
|
|
sql = '''
|
|
SELECT id, viat, m_oficial, m_reservada, marca, modelo,
|
|
poblacion, unidad, tipo
|
|
FROM vehiculos
|
|
WHERE viat IS NOT NULL
|
|
AND viat != ''
|
|
'''
|
|
|
|
params = []
|
|
|
|
if terminos:
|
|
condiciones = []
|
|
|
|
for termino in terminos:
|
|
condiciones.append('''
|
|
(
|
|
viat LIKE ?
|
|
OR m_oficial LIKE ?
|
|
OR m_reservada LIKE ?
|
|
OR marca LIKE ?
|
|
OR modelo LIKE ?
|
|
OR poblacion LIKE ?
|
|
OR unidad LIKE ?
|
|
OR tipo LIKE ?
|
|
)
|
|
''')
|
|
|
|
like = f'%{termino}%'
|
|
params.extend([like] * 8)
|
|
|
|
sql += ' AND ' + ' AND '.join(condiciones)
|
|
sql += ' ORDER BY id'
|
|
rows = db.execute(sql, params).fetchall()
|
|
|
|
|
|
if solo_filas:
|
|
return render_template('viat/rows.html', rows=rows, q=q)
|
|
return render_template('viat/list.html', page='viat', rows=rows, q=q)
|
|
|
|
|
|
|
|
@app.route('/viat/exportar')
|
|
@login_required
|
|
def viat_exportar():
|
|
import io, openpyxl
|
|
from openpyxl.styles import Font, PatternFill, Alignment
|
|
from flask import send_file
|
|
|
|
db = get_db()
|
|
q = request.args.get('q', '').strip()
|
|
|
|
terminos = [t.strip() for t in q.split(',') if t.strip()] if q else []
|
|
|
|
sql = '''
|
|
SELECT id, viat, m_oficial, m_reservada, marca, modelo,
|
|
poblacion, unidad, tipo
|
|
FROM vehiculos
|
|
WHERE viat IS NOT NULL
|
|
AND viat != ''
|
|
'''
|
|
|
|
params = []
|
|
|
|
if terminos:
|
|
condiciones = []
|
|
|
|
for termino in terminos:
|
|
condiciones.append('''
|
|
(
|
|
viat LIKE ?
|
|
OR m_oficial LIKE ?
|
|
OR m_reservada LIKE ?
|
|
OR marca LIKE ?
|
|
OR modelo LIKE ?
|
|
OR poblacion LIKE ?
|
|
OR unidad LIKE ?
|
|
OR tipo LIKE ?
|
|
)
|
|
''')
|
|
|
|
like = f'%{termino}%'
|
|
params.extend([like] * 8)
|
|
|
|
sql += ' AND ' + ' AND '.join(condiciones)
|
|
sql += ' ORDER BY id'
|
|
rows = db.execute(sql, params).fetchall()
|
|
|
|
wb = openpyxl.Workbook()
|
|
ws = wb.active
|
|
ws.title = 'Via-T'
|
|
|
|
headers = ['Via-T', 'M. Oficial', 'M. Reservada', 'Marca',
|
|
'Modelo', 'Población', 'Unidad', 'Tipo']
|
|
header_fill = PatternFill('solid', fgColor='5C6EF8')
|
|
header_font = Font(bold=True, color='FFFFFF', size=11)
|
|
|
|
for col, h in enumerate(headers, 1):
|
|
cell = ws.cell(row=1, column=col, value=h)
|
|
cell.fill = header_fill
|
|
cell.font = header_font
|
|
cell.alignment = Alignment(horizontal='center', vertical='center')
|
|
ws.row_dimensions[1].height = 22
|
|
|
|
for row in rows:
|
|
ws.append([row['viat'], row['m_oficial'], row['m_reservada'],
|
|
row['marca'], row['modelo'], row['poblacion'],
|
|
row['unidad'], row['tipo']])
|
|
|
|
col_widths = [16, 14, 14, 16, 16, 18, 18, 14]
|
|
for i, w in enumerate(col_widths, 1):
|
|
ws.column_dimensions[ws.cell(1, i).column_letter].width = w
|
|
|
|
zebra = PatternFill('solid', fgColor='EEF0F8')
|
|
for row_idx in range(2, ws.max_row + 1):
|
|
if row_idx % 2 == 0:
|
|
for col_idx in range(1, len(headers) + 1):
|
|
ws.cell(row_idx, col_idx).fill = zebra
|
|
|
|
buf = io.BytesIO()
|
|
wb.save(buf)
|
|
buf.seek(0)
|
|
nombre_q = '_'.join(terminos).replace(' ', '-') if terminos else ''
|
|
filename = f'viat{"_" + nombre_q if nombre_q else ""}.xlsx'
|
|
return send_file(buf, as_attachment=True,
|
|
download_name=filename,
|
|
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')
|
|
|
|
|
|
# ── CRUD Campus ──────────────────────────────────────────────
|
|
@app.route('/campus')
|
|
@login_required
|
|
def campus_list():
|
|
db = get_db()
|
|
q = request.args.get('q', '').strip()
|
|
solo_filas = request.args.get('solo_filas') == '1'
|
|
base_sql = '''
|
|
SELECT c.*, cat.tipo as categoria
|
|
FROM campus c
|
|
LEFT JOIN categorias cat ON c.categoria_id = cat.id
|
|
'''
|
|
if q:
|
|
like = f'%{q}%'
|
|
rows = db.execute(
|
|
base_sql + ' WHERE c.nombre LIKE ? OR c.telefono LIKE ? OR c.email LIKE ? OR CAST(c.id AS TEXT) LIKE ? OR cat.tipo LIKE ? ORDER BY c.id',
|
|
(like, like, like, like, like)
|
|
).fetchall()
|
|
else:
|
|
rows = db.execute(base_sql + ' ORDER BY c.id').fetchall()
|
|
if solo_filas:
|
|
return render_template('campus/rows.html', rows=rows, q=q)
|
|
return render_template('campus/list.html', page='campus', rows=rows, q=q)
|
|
|
|
@app.route('/campus/nuevo', methods=['GET', 'POST'])
|
|
@login_required
|
|
def campus_nuevo():
|
|
db = get_db()
|
|
categorias = db.execute('SELECT * FROM categorias ORDER BY tipo').fetchall()
|
|
if request.method == 'POST':
|
|
nombre = request.form['nombre'].strip()
|
|
telefono = request.form['telefono'].strip()
|
|
email = request.form['email'].strip()
|
|
categoria_id = request.form.get('categoria_id') or None
|
|
error = None
|
|
if not nombre: error = 'El nombre es obligatorio.'
|
|
elif not telefono: error = 'El teléfono es obligatorio.'
|
|
elif not email: error = 'El email es obligatorio.'
|
|
if error:
|
|
flash(error, 'error')
|
|
else:
|
|
try:
|
|
db.execute(
|
|
'INSERT INTO campus (nombre, telefono, email, categoria_id) VALUES (?, ?, ?, ?)',
|
|
(nombre, telefono, email, categoria_id)
|
|
)
|
|
db.commit()
|
|
flash('Campus creado correctamente.', 'ok')
|
|
return redirect(url_for('campus_list'))
|
|
except sqlite3.IntegrityError:
|
|
flash('Ese email ya existe en la base de datos.', 'error')
|
|
return render_template('campus/form.html', page='campus', campus=None, categorias=categorias)
|
|
|
|
|
|
|
|
@app.route('/campus/<int:id>/editar', methods=['GET', 'POST'])
|
|
@login_required
|
|
def campus_editar(id):
|
|
db = get_db()
|
|
campus = db.execute('SELECT * FROM campus WHERE id=?', (id,)).fetchone()
|
|
categorias = db.execute('SELECT * FROM categorias ORDER BY tipo').fetchall()
|
|
if not campus:
|
|
flash('Campus no encontrado.', 'error')
|
|
return redirect(url_for('campus_list'))
|
|
if request.method == 'POST':
|
|
nombre = request.form['nombre'].strip()
|
|
telefono = request.form['telefono'].strip()
|
|
email = request.form['email'].strip()
|
|
categoria_id = request.form.get('categoria_id') or None
|
|
error = None
|
|
if not nombre: error = 'El nombre es obligatorio.'
|
|
elif not telefono: error = 'El teléfono es obligatorio.'
|
|
elif not email: error = 'El email es obligatorio.'
|
|
if error:
|
|
flash(error, 'error')
|
|
else:
|
|
try:
|
|
db.execute(
|
|
'UPDATE campus SET nombre=?, telefono=?, email=?, categoria_id=? WHERE id=?',
|
|
(nombre, telefono, email, categoria_id, id)
|
|
)
|
|
db.commit()
|
|
flash('Campus actualizado.', 'ok')
|
|
return redirect(url_for('campus_list'))
|
|
except sqlite3.IntegrityError:
|
|
flash('Ese email ya está en uso por otro registro.', 'error')
|
|
return render_template('campus/form.html', page='campus', campus=campus, categorias=categorias)
|
|
|
|
|
|
@app.route('/campus/<int:id>/eliminar', methods=['POST'])
|
|
@login_required
|
|
def campus_eliminar(id):
|
|
db = get_db()
|
|
db.execute('DELETE FROM campus WHERE id=?', (id,))
|
|
db.commit()
|
|
flash('Campus eliminado.', 'ok')
|
|
return redirect(url_for('campus_list'))
|
|
|
|
|
|
|
|
@app.context_processor
|
|
def inject_globals():
|
|
from datetime import date
|
|
mensajes = get_db().execute(
|
|
'SELECT * FROM mensajes ORDER BY fecha DESC'
|
|
).fetchall()
|
|
return {
|
|
'now': date.today().isoformat(),
|
|
'mensajes': mensajes
|
|
}
|
|
|
|
|
|
@app.template_filter('fecha_es')
|
|
def fecha_es(valor):
|
|
if not valor:
|
|
return '—'
|
|
try:
|
|
from datetime import datetime
|
|
return datetime.strptime(valor[:10], '%Y-%m-%d').strftime('%d-%m-%Y')
|
|
except:
|
|
return valor
|
|
|
|
@app.template_filter('fecha_es_hora')
|
|
def fecha_es(valor):
|
|
if not valor:
|
|
return '—'
|
|
try:
|
|
from datetime import datetime
|
|
dt = datetime.strptime(str(valor), '%Y-%m-%d %H:%M:%S')
|
|
return dt.strftime('%d-%m-%Y %H:%M')
|
|
except:
|
|
return valor
|
|
|
|
|
|
|
|
|
|
# ------------------------------------------------------------
|
|
if __name__ == '__main__':
|
|
init_db()
|
|
app.run(host="0.0.0.0", port=5000, debug=True)
|
|
# app.run(debug=True)
|
|
|
|
|
|
|
|
|
|
|
|
# CODIGO PREVIO A INCLUIR NORMALIZACION
|
|
# if terminos:
|
|
# condiciones = []
|
|
# params = []
|
|
# for t in terminos:
|
|
# like = f'%{t}%'
|
|
# condiciones.append('''(
|
|
# m_oficial LIKE ? OR m_reservada LIKE ? OR VIN LIKE ?
|
|
# OR marca LIKE ? OR modelo LIKE ? OR poblacion LIKE ?
|
|
# OR unidad LIKE ? OR grupo LIKE ? OR tipo LIKE ?
|
|
# OR situacion LIKE ? OR int_operacion LIKE ?
|
|
# )''')
|
|
|
|
# params.extend([like] * 11)
|
|
# where = ' AND '.join(condiciones)
|
|
|
|
# rows = db.execute(
|
|
# f'SELECT id, m_oficial, m_reservada, marca, modelo, poblacion, unidad, grupo, tipo, situacion FROM vehiculos WHERE {where} {filtro_bajas} ORDER BY id',
|
|
# params
|
|
# ).fetchall()
|
|
|