import os import shutil import decimal import datetime import re import sqlite3 from flask import Flask, jsonify, render_template, request, send_file, session, redirect, url_for from werkzeug.security import check_password_hash import pyodbc import dbf app = Flask(__name__) # Clave secreta para firmar cookies de sesión de forma segura app.secret_key = os.environ.get("SECRET_KEY", "7d3a2e5c9b1f8a4d2c6e0b7f5a3c1e9d") # Credenciales del administrador. La contraseña está encriptada usando scrypt. # Contraseña sugerida: "AlohaSecure2026!" ADMIN_USER = "admin" ADMIN_PASSWORD_HASH = os.environ.get( "ADMIN_PASSWORD_HASH", "scrypt:32768:8:1$ntS6bBIgEoRaTWrZ$d47ac59c00c0fbd47a61bdb466694e1de640f047d9aa7d76f67e850f23e452d148df0d84297df06eb06f1e744d8408d48801cf8bab7d11e75fd26c1239d91b0b" ) @app.before_request def require_login(): """Protege todas las rutas de la aplicación requiriendo inicio de sesión.""" # Permitir libre acceso a la página de login y a recursos estáticos allowed_endpoints = ['login', 'static'] if request.endpoint not in allowed_endpoints and not session.get('logged_in'): # Redireccionar de forma relativa para soportar publicación en subcarpetas de Apache return redirect('login') # ------------------------------------------------------------- # Protocolos de Seguridad y Validación de Entradas # ------------------------------------------------------------- def validate_inputs(store_id, date_str): """Valida de forma estricta los tipos de datos y formatos para evitar SQL Injection.""" if not store_id or not date_str: return False, "Faltan parámetros requeridos: 'store_id' o 'date'." # Validar que el ID de tienda contenga únicamente dígitos numéricos if not store_id.isdigit(): return False, "El ID de la tienda (PDV) debe ser un valor numérico entero." # Validar formato estricto de fecha YYYY-MM-DD usando expresiones regulares if not re.match(r'^\d{4}-\d{2}-\d{2}$', date_str): return False, "El formato de la fecha es incorrecto (debe ser YYYY-MM-DD)." return True, "" @app.after_request def add_security_headers(response): """Añade cabeceras HTTP estándar para mitigar vulnerabilidades web (OWASP top 10).""" response.headers['X-Frame-Options'] = 'DENY' response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['X-XSS-Protection'] = '1; mode=block' response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' # Habilitar política de seguridad de contenido (CSP) restrictiva response.headers['Content-Security-Policy'] = ( "default-src 'self' https://fonts.googleapis.com https://fonts.gstatic.com https://cdn.jsdelivr.net; " "style-src 'self' 'unsafe-inline' https://fonts.googleapis.com https://cdn.jsdelivr.net; " "font-src 'self' https://fonts.gstatic.com; " "img-src 'self' data:;" ) return response # Detectar los controladores de SQL Server instalados en la máquina para garantizar portabilidad def get_best_sql_driver(): drivers = pyodbc.drivers() for d in ['ODBC Driver 18 for SQL Server', 'ODBC Driver 17 for SQL Server', 'SQL Server Native Client 11.0', 'SQL Server']: if d in drivers: return d return 'SQL Server' SQL_DRIVER = get_best_sql_driver() # Si es un driver moderno (ODBC Driver 17/18), añadimos TrustServerCertificate=yes para redes corporativas trust = ";TrustServerCertificate=yes" if "ODBC Driver" in SQL_DRIVER else "" # Conexión directa DSN-less por IP, puerto, base de datos y credenciales ODBC_CONN_STR = os.environ.get( "ODBC_CONN_STR", f"Driver={{{SQL_DRIVER}}};Server=10.74.28.28,2544;Database=RPJ;Uid=rpjaloha;Pwd=KkeL7by%&!!OO;{trust}" ) # Directorio temporal para las cargas y archivos ajustados BASE_DIR = os.path.dirname(os.path.abspath(__file__)) TEMP_DIR = os.path.join(BASE_DIR, "temp_uploads") os.makedirs(TEMP_DIR, exist_ok=True) def get_db_connection(): """Retorna una conexión ODBC activa conectándose directamente por IP y puerto sin depender de DSN configurados en Windows.""" return pyodbc.connect(ODBC_CONN_STR) def serialize_row_value(val): """Convierte Decimales y Fechas a formatos compatibles con JSON.""" if isinstance(val, decimal.Decimal): return float(val) elif isinstance(val, (datetime.date, datetime.datetime)): return val.isoformat() return val def row_to_dict(cursor, row): """Convierte una fila en un diccionario.""" return {col[0]: serialize_row_value(val) for col, val in zip(cursor.description, row)} @app.route('/') def index(): """Carga la interfaz gráfica del usuario.""" return render_template('index.html') @app.route('/login', methods=['GET', 'POST']) def login(): """Muestra el formulario de inicio de sesión y valida las credenciales.""" if session.get('logged_in'): return redirect('./') error = None if request.method == 'POST': username = request.form.get('username') password = request.form.get('password') if username == ADMIN_USER and check_password_hash(ADMIN_PASSWORD_HASH, password): session['logged_in'] = True session['username'] = username # Redireccionar de forma relativa para volver al inicio del subdirectorio actual return redirect('./') else: error = "Usuario o contraseña incorrectos." return render_template('login.html', error=error) @app.route('/logout') def logout(): """Cierra la sesión actual del usuario.""" session.clear() # Redireccionar de forma relativa al login return redirect('login') @app.route('/api/stores', methods=['GET']) def get_stores(): """API para obtener la lista de puntos de venta activos (estado = 1).""" conn = None cursor = None try: conn = get_db_connection() cursor = conn.cursor() query = """ SELECT ID_ALOHA, NOMBREPDV FROM dbo.tiendas_aco WHERE estado = 1 ORDER BY NOMBREPDV ASC; """ cursor.execute(query) rows = cursor.fetchall() stores = [row_to_dict(cursor, r) for r in rows] return jsonify({ "success": True, "stores": stores }) except Exception as e: return jsonify({ "success": False, "error": f"Error de conexión ODBC: {str(e)}" }), 500 finally: if cursor: cursor.close() if conn: conn.close() @app.route('/api/query', methods=['GET']) def run_query(): """API para consultar diferencias en base de datos (con empleado '9800').""" store_id = request.args.get('store_id') date_str = request.args.get('date') is_valid, err_msg = validate_inputs(store_id, date_str) if not is_valid: return jsonify({ "success": False, "error": err_msg }), 400 conn = None cursor = None try: conn = get_db_connection() cursor = conn.cursor() query = """ SELECT t.NOMBREPDV, c.DateOfBusiness AS dateofbussiness, c.FKStoreId AS fkstoreid, c.FKEmployeeNumber, c.CheckID, c.ItemSales, c.Payment FROM [RPJ].[dbo].[DpvHstCheckSummary] c INNER JOIN dbo.tiendas_aco t ON c.FKStoreId = t.ID_ALOHA WHERE t.ID_ALOHA = ? AND CAST(c.DateOfBusiness AS DATE) = ? AND c.FKEmployeeNumber = '9800'; """ cursor.execute(query, (store_id, date_str)) rows = cursor.fetchall() results = [row_to_dict(cursor, r) for r in rows] return jsonify({ "success": True, "results": results }) except Exception as e: return jsonify({ "success": False, "error": f"Error de consulta en base de datos: {str(e)}" }), 500 finally: if cursor: cursor.close() if conn: conn.close() @app.route('/api/compare-dbf', methods=['POST']) def compare_dbf(): """Carga un archivo DBF, lo compara con SQL y devuelve la lista de pre-visualización.""" store_id = request.form.get('store_id') date_str = request.form.get('date') use_local = request.form.get('use_local') == 'true' is_valid, err_msg = validate_inputs(store_id, date_str) if not is_valid: return jsonify({"success": False, "error": err_msg}), 400 temp_path = os.path.join(TEMP_DIR, "temp_uploaded.dbf") # Manejar el archivo: subido por el cliente o buscar archivo local en la raíz if use_local: local_files = ["GNDTNDR.dbf", "GNDTNDR.DBF", "GNDTNDR (1).dbf"] found = False for f_name in local_files: local_path = os.path.join(BASE_DIR, f_name) if os.path.exists(local_path): shutil.copy(local_path, temp_path) found = True break if not found: return jsonify({"success": False, "error": "No se encontró ningún archivo GNDTNDR.DBF o GNDTNDR (1).dbf en la raíz del proyecto."}), 404 else: if 'file' not in request.files: return jsonify({"success": False, "error": "No se ha subido ningún archivo y no se solicitó usar el local."}), 400 file = request.files['file'] if file.filename == '': return jsonify({"success": False, "error": "Nombre de archivo vacío."}), 400 # Validar la extensión del archivo para evitar la carga de scripts o payloads maliciosos if not file.filename.lower().endswith('.dbf'): return jsonify({"success": False, "error": "Tipo de archivo no permitido. Solo se permiten archivos con extensión .dbf"}), 400 file.save(temp_path) conn = None cursor = None try: # 1. Leer los registros del DBF temporal dbf_records = [] table = dbf.Table(temp_path) table.open(mode=dbf.READ_ONLY) for r in table: # Extraer campos clave emp = str(r['EMPLOYEE']).strip() # Filtrar estrictamente por empleado 9800 if emp != '9800': continue chk = str(r['CHECK']).strip() dbf_date = r['DATE'] dbf_date_str = dbf_date.isoformat() if dbf_date else "" dbf_amt = float(r['AMOUNT']) typeid = int(r['TYPEID']) dbf_records.append({ "employee": emp, "check": chk, "dbf_date": dbf_date_str, "dbf_amount": dbf_amt, "typeid": typeid }) table.close() # 2. Consultar la base de datos SQL para ese PDV y Fecha (obtenemos todas las ventas de la fecha para cruzar) conn = get_db_connection() cursor = conn.cursor() query = """ SELECT FKEmployeeNumber, CheckID, ItemSales FROM [RPJ].[dbo].[DpvHstCheckSummary] WHERE FKStoreId = ? AND CAST(DateOfBusiness AS DATE) = ? AND FKEmployeeNumber = '9800'; """ cursor.execute(query, (store_id, date_str)) sql_rows = cursor.fetchall() # Map de base de datos {(employee, check): itemsales} sql_map = {} for r in sql_rows: emp = str(r.FKEmployeeNumber).strip() chk = str(r.CheckID).strip() sql_map[(emp, chk)] = float(r.ItemSales) # 3. Cruzar información preview_records = [] differences_count = 0 matches_count = 0 for r in dbf_records: # Validar si coincide la fecha if r['dbf_date'] != date_str: status = "Fecha distinta" sql_sales = None adjusted = r['dbf_amount'] else: key = (r['employee'], r['check']) sql_sales = sql_map.get(key, None) if sql_sales is not None: matches_count += 1 adjusted = sql_sales if r['dbf_amount'] != sql_sales: status = "Ajustar" differences_count += 1 else: status = "Correcto" else: status = "No encontrado en SQL" adjusted = r['dbf_amount'] preview_records.append({ "check": r['check'], "employee": r['employee'], "typeid": r['typeid'], "dbf_amount": r['dbf_amount'], "sql_sales": sql_sales, "adjusted_amount": adjusted, "status": status }) return jsonify({ "success": True, "total_records": len(dbf_records), "matches": matches_count, "differences": differences_count, "records": preview_records }) except Exception as e: return jsonify({ "success": False, "error": f"Error al procesar y comparar el DBF: {str(e)}" }), 500 finally: if cursor: cursor.close() if conn: conn.close() @app.route('/api/download-adjusted-dbf', methods=['POST']) def download_adjusted_dbf(): """Modifica el DBF temporal aplicando el valor de ItemSales en AMOUNT y lo sirve para descarga.""" store_id = request.form.get('store_id') date_str = request.form.get('date') is_valid, err_msg = validate_inputs(store_id, date_str) if not is_valid: return jsonify({"success": False, "error": err_msg}), 400 src_temp = os.path.join(TEMP_DIR, "temp_uploaded.dbf") dest_adjusted = os.path.join(TEMP_DIR, "adjusted_GNDTNDR.dbf") if not os.path.exists(src_temp): return jsonify({"success": False, "error": "No hay ningún archivo DBF procesado previamente."}), 400 # Crear una copia limpia sobre la cual aplicar los cambios shutil.copy(src_temp, dest_adjusted) conn = None cursor = None try: # 1. Obtener los datos de SQL para mapear los ajustes conn = get_db_connection() cursor = conn.cursor() query = """ SELECT FKEmployeeNumber, CheckID, ItemSales FROM [RPJ].[dbo].[DpvHstCheckSummary] WHERE FKStoreId = ? AND CAST(DateOfBusiness AS DATE) = ? AND FKEmployeeNumber = '9800'; """ cursor.execute(query, (store_id, date_str)) sql_rows = cursor.fetchall() sql_map = {} for r in sql_rows: emp = str(r.FKEmployeeNumber).strip() chk = str(r.CheckID).strip() sql_map[(emp, chk)] = float(r.ItemSales) cursor.close() conn.close() # 2. Modificar el DBF en modo Lectura/Escritura table = dbf.Table(dest_adjusted) table.open(mode=dbf.READ_WRITE) adjusted_rows = 0 for r in table: emp = str(r['EMPLOYEE']).strip() # Filtrar estrictamente por empleado 9800 if emp == '9800': chk = str(r['CHECK']).strip() dbf_date = r['DATE'] dbf_date_str = dbf_date.isoformat() if dbf_date else "" # Solo modificamos los registros que correspondan a la fecha seleccionada if dbf_date_str == date_str: key = (emp, chk) if key in sql_map: sql_sales = sql_map[key] if float(r['AMOUNT']) != sql_sales: dbf.write(r, AMOUNT=sql_sales) adjusted_rows += 1 table.close() print(f"Modificación del DBF finalizada. Se ajustaron {adjusted_rows} registros.") # Servir el archivo modificado para descarga return send_file( dest_adjusted, as_attachment=True, download_name="GNDTNDR.DBF", mimetype="application/x-dbf" ) except Exception as e: return jsonify({ "success": False, "error": f"Error al generar el archivo DBF ajustado: {str(e)}" }), 500 # ------------------------------------------------------------- # Funciones Auxiliares para Cruce de Cuentas por Cobrar (CXC) # ------------------------------------------------------------- def find_cxc_negative_records_dbf(dbf_path): """Abre el archivo DBF de CXC y extrae registros de notas de crédito (monto negativo).""" table = dbf.Table(dbf_path) table.open() try: columns = table.field_names # Buscar nombres de columnas aproximados col_check = next((c for c in columns if any(k in c.lower() for k in ['check', 'factura', 'doc', 'id_factura', 'checknumber', 'checkid'])), None) col_date = next((c for c in columns if any(k in c.lower() for k in ['date', 'fecha', 'dia'])), None) col_store = next((c for c in columns if any(k in c.lower() for k in ['store', 'tienda', 'sucursal', 'pdv', 'storeid', 'store_id'])), None) col_amount = next((c for c in columns if any(k in c.lower() for k in ['amount', 'monto', 'valor', 'payment', 'total', 'neto'])), None) records = [] if col_amount and col_check: for row in table: amount_val = row[col_amount] try: amount = float(amount_val) except: continue if amount < 0: chk = str(row[col_check]).strip() if col_check else "" dt_val = row[col_date] if col_date else None dt_str = "" if dt_val: if isinstance(dt_val, (datetime.date, datetime.datetime)): dt_str = dt_val.isoformat() else: dt_str = str(dt_val).strip() if len(dt_str) >= 10: dt_str = dt_str[:10] store = str(row[col_store]).strip() if col_store else "" records.append({ "check": chk, "date": dt_str, "store_id": store, "amount": amount }) return records except Exception as e: print(f"Error al procesar archivo DBF: {str(e)}") return [] finally: table.close() def get_check_items(cursor, store_id, date_str, check_id): """Obtiene los productos, nombres, precios y cantidades asociados a un check.""" try: query = """ SELECT i.FKItemId, it.ShortName, i.Price, i.Quantity FROM dpvHstGndItem i LEFT JOIN Item it ON i.FKItemId = it.ItemId WHERE i.CheckNumber = ? AND i.FKStoreId = ? AND CAST(i.DateOfBusiness AS DATE) = ? """ cursor.execute(query, (int(check_id), int(store_id), date_str)) items = [] for r in cursor.fetchall(): items.append({ "item_id": int(r.FKItemId), "name": str(r.ShortName).strip() if r.ShortName else f"Producto {r.FKItemId}", "price": float(r.Price) if r.Price is not None else 0.0, "quantity": float(r.Quantity) if r.Quantity is not None else 0.0 }) return items except Exception as e: print(f"Error al obtener items para check {check_id}: {str(e)}") return [] def find_matching_positive_check(cursor, store_id, date_str, neg_check_id, neg_amount_sql): """Busca en SQL Server una factura con valor positivo que tenga exactamente los mismos productos y monto.""" try: # 1. Obtener la lista de productos y cantidades (en valor absoluto) del check negativo en dpvHstGndItem query_items = """ SELECT FKItemId, abs(Quantity) as Qty FROM dpvHstGndItem WHERE CheckNumber = ? AND FKStoreId = ? AND CAST(DateOfBusiness AS DATE) = ? """ cursor.execute(query_items, (int(neg_check_id), int(store_id), date_str)) neg_items = {} for r in cursor.fetchall(): item_id = int(r.FKItemId) qty = float(r.Qty) neg_items[item_id] = neg_items.get(item_id, 0.0) + qty if not neg_items: return None # 2. Buscar checks positivos en el mismo día y tienda que tengan el mismo monto total (en valor absoluto) query_pos_checks = """ SELECT c.CheckID, c.Payment, c.DateOfBusiness FROM DpvHstCheckSummary c WHERE c.FKStoreID = ? AND CAST(c.DateOfBusiness AS DATE) = ? AND c.Payment > 0 AND abs(c.Payment - abs(?)) < 1.0 """ cursor.execute(query_pos_checks, (int(store_id), date_str, float(neg_amount_sql))) candidates = cursor.fetchall() for cand in candidates: cand_check_id = cand.CheckID # Obtener los ítems de este check candidato query_cand_items = """ SELECT FKItemId, Quantity FROM dpvHstGndItem WHERE CheckNumber = ? AND FKStoreId = ? AND CAST(DateOfBusiness AS DATE) = ? """ cursor.execute(query_cand_items, (int(cand_check_id), int(store_id), date_str)) cand_items = {} for r in cursor.fetchall(): item_id = int(r.FKItemId) qty = float(r.Quantity) cand_items[item_id] = cand_items.get(item_id, 0.0) + qty # Comparar si las listas de ítems coinciden if cand_items == neg_items: # Obtener el medio de pago del check candidato query_tender = """ SELECT g.TypeId, t.Name as TenderName FROM dpvHstGndTender g JOIN Tender t ON g.TypeId = t.TenderId WHERE g.CheckNumber = ? AND g.FKStoreId = ? AND CAST(g.DateOfBusiness AS DATE) = ? """ cursor.execute(query_tender, (int(cand_check_id), int(store_id), date_str)) t_row = cursor.fetchone() if t_row: return { "check_id": int(cand_check_id), "payment": float(cand.Payment), "tender_id": int(t_row.TypeId), "tender_name": str(t_row.TenderName) } except Exception as e: print(f"Error al buscar check positivo coincidente: {str(e)}") return None # ------------------------------------------------------------- # Endpoints de la Pestaña "Diferencia CXC" # ------------------------------------------------------------- @app.route('/api/tenders', methods=['GET']) def get_tenders(): """Retorna la lista de medios de pago activos de la base de datos SQL Server.""" conn = None cursor = None try: conn = get_db_connection() cursor = conn.cursor() query = "SELECT TenderId, Name FROM Tender WHERE Active = 'Y' ORDER BY Name;" cursor.execute(query) rows = cursor.fetchall() tenders = [{"id": int(r.TenderId), "name": str(r.Name)} for r in rows] return jsonify({"success": True, "tenders": tenders}) except Exception as e: return jsonify({"success": False, "error": f"Error al consultar Tender en SQL Server: {str(e)}"}), 500 finally: if cursor: cursor.close() if conn: conn.close() @app.route('/api/cxc-compare', methods=['POST']) def cxc_compare(): """Recibe un archivo .dbf, lee notas de crédito y las cruza por productos con facturas positivas en SQL Server.""" if 'file' not in request.files: return jsonify({"success": False, "error": "No se subió ningún archivo."}), 400 file = request.files['file'] if file.filename == '': return jsonify({"success": False, "error": "Archivo vacío."}), 400 if not file.filename.lower().endswith('.dbf'): return jsonify({"success": False, "error": "Formato de archivo no permitido. Solo se aceptan archivos .dbf"}), 400 req_store_id = request.form.get('store_id') req_date_str = request.form.get('date') dbf_path = os.path.join(TEMP_DIR, "uploaded_cxc.dbf") file.save(dbf_path) # Guardar el nombre original del archivo para servirlo con el mismo nombre en la descarga filename_path = os.path.join(TEMP_DIR, "uploaded_cxc_filename.txt") with open(filename_path, "w", encoding="utf-8") as fn_file: fn_file.write(file.filename) dbf_records = find_cxc_negative_records_dbf(dbf_path) if not dbf_records: return jsonify({ "success": True, "total_records": 0, "records": [], "message": "No se encontraron registros con importes negativos en la base de datos subida." }) conn = None cursor = None results = [] try: conn = get_db_connection() cursor = conn.cursor() for rec in dbf_records: chk_num = rec['check'] # Priorizar los parámetros seleccionados en la UI store_id = req_store_id if req_store_id else rec['store_id'] date_str = req_date_str if req_date_str else rec['date'] neg_amount = rec['amount'] if not chk_num or not date_str or not store_id: continue if not store_id.isdigit() or not chk_num.isdigit(): continue if not re.match(r'^\d{4}-\d{2}-\d{2}$', date_str): continue # 1. Consultar el check negativo en SQL Server query_neg_sql = """ SELECT c.CheckID, c.Payment, g.TypeId, t.Name as TenderName FROM DpvHstCheckSummary c LEFT JOIN dpvHstGndTender g ON c.CheckID = g.CheckNumber AND c.FKStoreID = g.FKStoreId AND CAST(c.DateOfBusiness AS DATE) = CAST(g.DateOfBusiness AS DATE) LEFT JOIN Tender t ON g.TypeId = t.TenderId WHERE c.CheckID = ? AND c.FKStoreID = ? AND CAST(c.DateOfBusiness AS DATE) = ? AND c.Payment < 0 """ cursor.execute(query_neg_sql, (int(chk_num), int(store_id), date_str)) neg_row = cursor.fetchone() # Obtener los items del check negativo neg_items = get_check_items(cursor, store_id, date_str, chk_num) neg_check_data = { "check_id": int(chk_num), "date": date_str, "store_id": int(store_id), "amount_db": neg_amount, "amount_sql": float(neg_row.Payment) if neg_row and neg_row.Payment else None, "tender_id": int(neg_row.TypeId) if neg_row and neg_row.TypeId else None, "tender_name": str(neg_row.TenderName) if neg_row and neg_row.TenderName else "N/A", "items": neg_items } # 2. Buscar la factura positiva coincidente por productos y monto total absoluto target_amount = neg_check_data['amount_sql'] if neg_check_data['amount_sql'] is not None else neg_amount pos_check_data = find_matching_positive_check(cursor, store_id, date_str, chk_num, target_amount) if pos_check_data: # Obtener los items del check positivo coincidente pos_items = get_check_items(cursor, store_id, date_str, pos_check_data['check_id']) pos_check_data['items'] = pos_items has_diff = False if pos_check_data and neg_check_data['tender_id'] is not None: if pos_check_data['tender_id'] != neg_check_data['tender_id']: has_diff = True results.append({ "neg_check": neg_check_data, "pos_check": pos_check_data, "has_difference": has_diff }) return jsonify({ "success": True, "total_records": len(dbf_records), "records": results }) except Exception as e: return jsonify({"success": False, "error": f"Error al procesar cruce en SQL Server: {str(e)}"}), 500 finally: if cursor: cursor.close() if conn: conn.close() @app.route('/api/cxc-update-tender', methods=['POST']) def cxc_update_tender(): """Actualiza el medio de pago (Tender) de un check específico en SQL Server.""" store_id = request.form.get('store_id') date_str = request.form.get('date') check_number = request.form.get('check_number') old_tender_id = request.form.get('old_tender_id') new_tender_id = request.form.get('new_tender_id') if not store_id or not date_str or not check_number or not old_tender_id or not new_tender_id: return jsonify({"success": False, "error": "Faltan parámetros requeridos."}), 400 if not store_id.isdigit() or not check_number.isdigit() or not old_tender_id.isdigit() or not new_tender_id.isdigit(): return jsonify({"success": False, "error": "Los IDs y números de check deben ser numéricos."}), 400 if not re.match(r'^\d{4}-\d{2}-\d{2}$', date_str): return jsonify({"success": False, "error": "Formato de fecha inválido."}), 400 conn = None cursor = None try: conn = get_db_connection() cursor = conn.cursor() # 1. Obtener el nombre del tender de la base de datos (operación de solo lectura para mapear nombres de texto si el DBF lo requiere) cursor.execute("SELECT Name FROM Tender WHERE TenderId = ?", (int(new_tender_id),)) t_name_row = cursor.fetchone() new_tender_name = str(t_name_row.Name).strip() if t_name_row else f"Tender {new_tender_id}" # 2. Actualizar el archivo local DBF sin cambiar su estructura dbf_path = os.path.join(TEMP_DIR, "uploaded_cxc.dbf") if os.path.exists(dbf_path): try: table = dbf.Table(dbf_path) table.open(mode=dbf.READ_WRITE) columns = table.field_names col_check = next((c for c in columns if any(k in c.lower() for k in ['check', 'factura', 'doc', 'id_factura', 'checknumber', 'checkid'])), None) col_store = next((c for c in columns if any(k in c.lower() for k in ['store', 'tienda', 'sucursal', 'pdv', 'storeid', 'store_id'])), None) col_tender = None if 'typeid' in [c.lower() for c in columns]: col_tender = next(c for c in columns if c.lower() == 'typeid') elif 'tenderid' in [c.lower() for c in columns]: col_tender = next(c for c in columns if c.lower() == 'tenderid') else: col_tender = next((c for c in columns if any(k in c.lower() for k in ['tender', 'type', 'payment', 'pago', 'medio'])), None) if col_check and col_tender: updated_count = 0 for row in table: chk_val = str(row[col_check]).strip() store_val = str(row[col_store]).strip() if col_store else "" # Comparación robusta del número de check check_matches = False try: if int(float(chk_val)) == int(float(check_number)): check_matches = True except: check_matches = (chk_val == check_number) # Comparación robusta del store si existe la columna store_matches = True if col_store and store_id: try: if int(float(store_val)) != int(float(store_id)): store_matches = False except: store_matches = False if check_matches and store_matches: current_val = row[col_tender] # Validar que el medio de pago anterior coincida (evitando modificar otras transacciones del mismo check si las hubiera) tender_matches = True if old_tender_id: try: if int(float(str(current_val).strip())) != int(float(old_tender_id)): tender_matches = False except: pass if tender_matches: if isinstance(current_val, (int, float)): dbf.write(row, **{col_tender: int(new_tender_id)}) else: dbf.write(row, **{col_tender: str(new_tender_id)}) updated_count += 1 table.close() print(f"Archivo DBF local corregido exitosamente (SQL intacto). Registros corregidos en DBF: {updated_count}") except Exception as dbf_err: print(f"Error al escribir en archivo local DBF: {str(dbf_err)}") return jsonify({"success": False, "error": f"Error al escribir en el archivo DBF: {str(dbf_err)}"}), 500 else: return jsonify({"success": False, "error": "El archivo DBF subido no se encuentra en el servidor."}), 400 return jsonify({ "success": True, "message": "Medio de pago actualizado correctamente en el archivo DBF." }) except Exception as e: return jsonify({"success": False, "error": f"Error al actualizar medio de pago: {str(e)}"}), 500 finally: if cursor: cursor.close() if conn: conn.close() @app.route('/api/cxc-download-dbf', methods=['GET']) def cxc_download_dbf(): """Sirve el archivo DBF de CXC con las correcciones aplicadas para su descarga.""" dest_path = os.path.join(TEMP_DIR, "uploaded_cxc.dbf") if not os.path.exists(dest_path): return jsonify({"success": False, "error": "No hay ningún archivo DBF procesado previamente."}), 400 # Leer el nombre de archivo original si existe filename_path = os.path.join(TEMP_DIR, "uploaded_cxc_filename.txt") download_name = "CXC_Corregido.DBF" if os.path.exists(filename_path): try: with open(filename_path, "r", encoding="utf-8") as fn_file: download_name = fn_file.read().strip() except: pass return send_file( dest_path, as_attachment=True, download_name=download_name, mimetype="application/x-dbf" ) if __name__ == '__main__': print("Iniciando aplicación web local en http://127.0.0.1:5000 ...") app.run(host='0.0.0.0', port=5000, debug=True)