feat(sync): add academic data reset tool, configurable sheets URL and fix sync timeout

This commit is contained in:
2026-09-19 11:13:24 -03:00
parent 42b31ec44e
commit 5087e22988
10 changed files with 835 additions and 101 deletions
+3 -1
View File
@@ -10,6 +10,7 @@ from .role import Role, Permission, SYSTEM_MODULES
from .milestone import MilestoneType, AcademicMilestone, BASE_MILESTONE_TYPES
from .audit_log import AuditLog
from .enrollment import StudentEnrollment
from .setting import SystemSetting
__all__ = [
'User',
@@ -31,5 +32,6 @@ __all__ = [
'AcademicMilestone',
'BASE_MILESTONE_TYPES',
'AuditLog',
'StudentEnrollment'
'StudentEnrollment',
'SystemSetting'
]
+50
View File
@@ -0,0 +1,50 @@
from app import db
from datetime import datetime
class SystemSetting(db.Model):
"""
Parámetros de configuración del sistema persistentes en base de datos.
Permite almacenar enlaces de integración (e.g. Google Sheets), flags globales y metadatos.
"""
__tablename__ = 'system_settings'
id = db.Column(db.Integer, primary_key=True)
key = db.Column(db.String(100), unique=True, nullable=False, index=True)
value = db.Column(db.Text, nullable=True)
description = db.Column(db.String(255), nullable=True)
updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
@classmethod
def get_value(cls, key, default=None):
try:
record = cls.query.filter_by(key=key).first()
if record and record.value is not None and record.value.strip() != '':
return record.value
except Exception:
pass
return default
@classmethod
def set_value(cls, key, value, description=None):
record = cls.query.filter_by(key=key).first()
if not record:
record = cls(key=key, value=value, description=description)
db.session.add(record)
else:
record.value = value
if description:
record.description = description
record.updated_at = datetime.utcnow()
db.session.commit()
return record
def to_dict(self):
return {
'key': self.key,
'value': self.value,
'description': self.description,
'updated_at': self.updated_at.isoformat() if self.updated_at else None
}
def __repr__(self):
return f'<SystemSetting {self.key}={self.value}>'
+115
View File
@@ -981,3 +981,118 @@ def delete_milestone_type(id):
'message': 'Tipificación eliminada permanentemente.'
}), 200
# ---------------------------------------------------------
# GOOGLE SHEETS CONFIGURATION & ACADEMIC DATA RESET TOOL
# ---------------------------------------------------------
@api_admin_bp.route('/sheets-config', methods=['GET'])
@jwt_required
def get_sheets_config():
"""Obtiene la configuración actual del Google Sheet para sincronización."""
from app.models.setting import SystemSetting
from app.services.sheets_importer import BASE_CSV_URL, normalize_sheets_url
import os
db_url = SystemSetting.get_value('google_sheets_url')
env_url = os.getenv('GOOGLE_SHEETS_URL')
active_url = db_url or env_url or BASE_CSV_URL
source = 'database' if db_url else ('environment' if env_url else 'default')
return jsonify({
'status': 'success',
'config': {
'active_url': active_url,
'source': source,
'db_url': db_url,
'default_url': BASE_CSV_URL,
'normalized_url': normalize_sheets_url(active_url)
}
}), 200
@api_admin_bp.route('/sheets-config', methods=['POST'])
@jwt_required
def update_sheets_config():
"""Actualiza la URL del Google Sheet en SystemSetting."""
from app.models.setting import SystemSetting
from app.services.sheets_importer import normalize_sheets_url
data = request.get_json(silent=True) or request.form.to_dict() or {}
url = (data.get('url') or data.get('google_sheets_url') or '').strip()
if not url:
return jsonify({
'error': 'ValidationError',
'message': 'La URL de Google Sheets no puede estar vacía.'
}), 400
normalized = normalize_sheets_url(url)
SystemSetting.set_value('google_sheets_url', url, 'URL del Google Sheet académico configurada desde panel')
return jsonify({
'status': 'success',
'message': 'Configuración de Google Sheets guardada correctamente.',
'config': {
'active_url': url,
'normalized_url': normalized,
'source': 'database'
}
}), 200
@api_admin_bp.route('/reset-academic-data', methods=['POST'])
@jwt_required
def reset_academic_data_api():
"""
Herramienta de limpieza/purga de datos académicos para pruebas del importador.
Permite eliminar reservas, comisiones, asignaturas, carreras y aulas.
"""
from scripts.reset_academic_data import execute_academic_reset
data = request.get_json(silent=True) or request.form.to_dict() or {}
reset_all = data.get('all', False) in [True, 'true', '1', 'on']
reset_reservations = reset_all or data.get('reservations', True) in [True, 'true', '1', 'on']
reset_commissions = reset_all or data.get('commissions', False) in [True, 'true', '1', 'on']
reset_subjects = reset_all or data.get('subjects', False) in [True, 'true', '1', 'on']
reset_careers = reset_all or data.get('careers', False) in [True, 'true', '1', 'on']
reset_classrooms = reset_all or data.get('classrooms', False) in [True, 'true', '1', 'on']
if reset_all:
reset_reservations = True
reset_commissions = True
reset_subjects = True
reset_careers = True
reset_classrooms = True
result = execute_academic_reset(
reset_reservations=reset_reservations,
reset_commissions=reset_commissions,
reset_subjects=reset_subjects,
reset_careers=reset_careers,
reset_classrooms=reset_classrooms
)
if result.get('success'):
try:
from app.models.audit_log import AuditLog
audit = AuditLog(
user_id=getattr(request, 'current_user_id', None),
action_type='ACADEMIC_DATA_PURGE',
details=f"Purga académica ejecutada: {result.get('deleted')}",
ip_address=request.remote_addr
)
db.session.add(audit)
db.session.commit()
except Exception:
pass
return jsonify({
'status': 'success',
'message': result.get('message', 'Datos reiniciados con éxito.'),
'deleted': result.get('deleted', {})
}), 200
else:
return jsonify({
'status': 'error',
'message': result.get('error', 'Error durante la purga de datos.')
}), 500
+5 -2
View File
@@ -169,11 +169,14 @@ def sync_google_sheets():
"""Ejecuta la sincronización real con las hojas de cálculo de Google Sheets."""
from app.services.sheets_importer import GoogleSheetsImporter
try:
importer = GoogleSheetsImporter()
data = request.get_json(silent=True) or {}
custom_url = data.get('sheets_url') or data.get('base_url')
importer = GoogleSheetsImporter(base_url=custom_url)
stats = importer.sync()
return jsonify({
'status': 'success',
'stats': stats
'stats': stats,
'source_url': importer.base_url
}), 200
except Exception as e:
return jsonify({
+44 -3
View File
@@ -2,6 +2,7 @@ import urllib.request
import csv
import io
import re
import os
from datetime import datetime, date, time, timedelta
from app import db
from app.models.classroom import Classroom
@@ -12,6 +13,34 @@ from app.models.user import User
BASE_CSV_URL = 'https://docs.google.com/spreadsheets/d/e/2PACX-1vSc_T_BQjbn3uPelioCgx52UM5Py-qNhJN0TYPd1kmsN5jdb3Q8rAaIvNMF_2ZTzQt6bH--yIWQKrKR/pub?single=true&output=csv'
def normalize_sheets_url(url):
"""
Normaliza enlaces de Google Sheets (compartidos, de edición o publicados)
para obtener la URL base que acepta parámetro gid=<gid>&output=csv o format=csv.
"""
if not url:
return BASE_CSV_URL
url = url.strip()
# 1. Enlace publicado web (/d/e/2PACX-.../pub...)
if '/d/e/' in url:
base_match = re.match(r'^(https://docs\.google\.com/spreadsheets/d/e/[^/?#]+)/pub', url)
if base_match:
return f"{base_match.group(1)}/pub?single=true&output=csv"
clean = url.split('&gid=')[0].split('?gid=')[0]
sep = '&' if '?' in clean else '?'
if 'output=csv' not in clean:
clean = f"{clean}{sep}single=true&output=csv"
return clean
# 2. Enlace de edición o visualización (/d/<SPREADSHEET_ID>/edit...)
sheet_id_match = re.search(r'/spreadsheets/d/([a-zA-Z0-9-_]+)', url)
if sheet_id_match:
sheet_id = sheet_id_match.group(1)
return f"https://docs.google.com/spreadsheets/d/{sheet_id}/export?format=csv"
return url
SHEETS_CONFIG = [
{ 'name': 'Lunes', 'gid': '1915752353', 'day_idx': 0 },
{ 'name': 'Martes', 'gid': '466778263', 'day_idx': 1 },
@@ -30,16 +59,28 @@ class GoogleSheetsImporter:
"""
def __init__(self, base_url=None):
self.base_url = base_url or BASE_CSV_URL
if base_url:
self.base_url = normalize_sheets_url(base_url)
else:
configured_url = None
try:
from app.models.setting import SystemSetting
configured_url = SystemSetting.get_value('google_sheets_url')
except Exception:
pass
if not configured_url:
configured_url = os.getenv('GOOGLE_SHEETS_URL')
self.base_url = normalize_sheets_url(configured_url) if configured_url else BASE_CSV_URL
def fetch_sheet_csv(self, gid):
"""Fetch CSV string for a specific sheet gid"""
url = f"{self.base_url}&gid={gid}"
separator = '&' if '?' in self.base_url else '?'
url = f"{self.base_url}{separator}gid={gid}"
req = urllib.request.Request(
url,
headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) EduSpace/2.0'}
)
with urllib.request.urlopen(req, timeout=15) as resp:
with urllib.request.urlopen(req, timeout=25) as resp:
return resp.read().decode('utf-8', errors='replace')
def parse_sheet_rows(self, csv_content, sheet_name):