from __future__ import annotations import json import hashlib import os import re import traceback import smtplib from urllib.parse import quote_plus from email.message import EmailMessage from jinja2 import Template from datetime import datetime, date, time, timedelta from functools import wraps from secrets import token_urlsafe from flask import abort, current_app, request from flask_login import current_user from werkzeug.utils import secure_filename from .models import ( Appointment, Leave, WorkingHour, AuditLog, ProfessionalProfile, AppSetting, ErrorLog, FrontendBlock, ClinicalRecord, Patient, ) from . import db WEEKDAY_LABELS = { 0: 'Lunes', 1: 'Martes', 2: 'Miércoles', 3: 'Jueves', 4: 'Viernes', 5: 'Sábado', 6: 'Domingo', } ACTIVE_APPOINTMENT_STATUSES = {'pending', 'confirmed'} SAFE_TEXT_RE = re.compile(r"[^\w\sÁÉÍÓÚÜÑáéíóúüñ.,;:()/%+\-@#°º\n\r\t]") SAFE_SINGLE_LINE_RE = re.compile(r"[^\w\sÁÉÍÓÚÜÑáéíóúüñ.,;:()/%+\-@#°º]") SAFE_IDENTIFIER_RE = re.compile(r"[^A-Za-z0-9._\-]") UPLOAD_RULES = { 'site': {'extensions': {'.png', '.jpg', '.jpeg', '.webp', '.svg', '.pdf'}, 'max_bytes': 8 * 1024 * 1024}, 'logo': {'extensions': {'.png', '.jpg', '.jpeg', '.webp', '.svg'}, 'max_bytes': 5 * 1024 * 1024}, 'favicon': {'extensions': {'.png', '.jpg', '.jpeg', '.webp', '.svg', '.ico'}, 'max_bytes': 2 * 1024 * 1024}, 'popup': {'extensions': {'.png', '.jpg', '.jpeg', '.webp', '.svg'}, 'max_bytes': 8 * 1024 * 1024}, 'frontend': {'extensions': {'.png', '.jpg', '.jpeg', '.webp', '.svg'}, 'max_bytes': 8 * 1024 * 1024}, 'chat': {'extensions': {'.png', '.jpg', '.jpeg', '.webp', '.pdf', '.txt', '.doc', '.docx', '.xls', '.xlsx', '.csv'}, 'max_bytes': 10 * 1024 * 1024}, 'clinical': {'extensions': {'.png', '.jpg', '.jpeg', '.webp', '.pdf', '.txt', '.doc', '.docx', '.xls', '.xlsx', '.csv'}, 'max_bytes': 12 * 1024 * 1024}, } ALLOWED_CONFIDENTIALITY = {'Paciente', 'Institucional', 'Restringido'} def sanitize_text(value: str, max_length: int = 255, multiline: bool = False) -> str: value = (value or '').strip() if not value: return '' value = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]', '', value) cleaner = SAFE_TEXT_RE if multiline else SAFE_SINGLE_LINE_RE value = cleaner.sub('', value) value = re.sub(r'\s+', ' ', value) if not multiline else re.sub(r'\n{3,}', '\n\n', value) return value[:max_length].strip() def sanitize_multiline_text(value: str, max_length: int = 4000) -> str: return sanitize_text(value, max_length=max_length, multiline=True) def sanitize_identifier(value: str, max_length: int = 120) -> str: value = (value or '').strip() value = SAFE_IDENTIFIER_RE.sub('', value) return value[:max_length] def sanitize_digits(value: str, max_length: int = 32) -> str: return re.sub(r'\D+', '', value or '')[:max_length] def sanitize_code(value: str, max_length: int = 50) -> str: value = (value or '').strip().upper() value = re.sub(r'[^A-Z0-9.\-]', '', value) return value[:max_length] def sanitize_choice(value: str, allowed, default: str = '') -> str: value = sanitize_text(value, max_length=80) return value if value in allowed else default def sanitize_json_text_map(payload, allowed_keys, *, max_value_length: int = 500, multiline: bool = False) -> dict: if isinstance(payload, str): try: payload = json.loads(payload or '{}') except Exception: payload = {} if not isinstance(payload, dict): return {} cleaned = {} for key, value in payload.items(): if key not in allowed_keys: continue if value is None: continue cleaned_value = sanitize_multiline_text(str(value), max_length=max_value_length) if multiline else sanitize_text(str(value), max_length=max_value_length) if cleaned_value: cleaned[key] = cleaned_value return cleaned def _uploaded_file_size(file_storage) -> int: try: pos = file_storage.stream.tell() file_storage.stream.seek(0, os.SEEK_END) size = file_storage.stream.tell() file_storage.stream.seek(pos) return size except Exception: return 0 def validate_uploaded_asset(file_storage, prefix: str = 'site'): if not file_storage or not getattr(file_storage, 'filename', ''): raise ValueError('No se recibió ningún archivo.') rules = UPLOAD_RULES.get(prefix, UPLOAD_RULES['site']) original_name = secure_filename(file_storage.filename or '') ext = os.path.splitext(original_name)[1].lower() if ext not in rules['extensions']: raise ValueError('El tipo de archivo no está permitido para este módulo.') size = _uploaded_file_size(file_storage) if size and size > rules['max_bytes']: raise ValueError('El archivo excede el tamaño máximo permitido.') return original_name, ext JURISDICTION_OPTIONS = [ {'code': '02', 'name': 'Ciudad Autónoma de Buenos Aires'}, {'code': '06', 'name': 'Provincia de Buenos Aires'}, {'code': '10', 'name': 'Provincia de Catamarca'}, {'code': '14', 'name': 'Provincia de Córdoba'}, {'code': '18', 'name': 'Provincia de Corrientes'}, {'code': '22', 'name': 'Provincia del Chaco'}, {'code': '26', 'name': 'Provincia del Chubut'}, {'code': '30', 'name': 'Provincia de Entre Ríos'}, {'code': '34', 'name': 'Provincia de Formosa'}, {'code': '38', 'name': 'Provincia de Jujuy'}, {'code': '42', 'name': 'Provincia de La Pampa'}, {'code': '46', 'name': 'Provincia de La Rioja'}, {'code': '50', 'name': 'Provincia de Mendoza'}, {'code': '54', 'name': 'Provincia de Misiones'}, {'code': '58', 'name': 'Provincia de Neuquén'}, {'code': '62', 'name': 'Provincia de Río Negro'}, {'code': '66', 'name': 'Provincia de Salta'}, {'code': '70', 'name': 'Provincia de San Juan'}, {'code': '74', 'name': 'Provincia de San Luis'}, {'code': '78', 'name': 'Provincia de Santa Cruz'}, {'code': '82', 'name': 'Provincia de Santa Fe'}, {'code': '86', 'name': 'Provincia de Santiago del Estero'}, {'code': '90', 'name': 'Provincia de Tucumán'}, {'code': '94', 'name': 'Provincia de Tierra del Fuego, Antártida e Islas del Atlántico Sur'}, ] PRESCRIPTION_TYPE_OPTIONS = [ {'code': '01', 'name': 'Prescripción de medicamento'}, {'code': '02', 'name': 'Prescripción de dispositivo'}, {'code': '03', 'name': 'Prescripción de estudios / prácticas / procedimientos'}, ] PRESCRIPTION_SUBTYPE_OPTIONS = [ {'code': '01', 'name': 'Expendio libre'}, {'code': '02', 'name': 'Expendio bajo receta'}, {'code': '03', 'name': 'Expendio bajo receta archivada'}, {'code': '04', 'name': 'Expendio legalmente restringido'}, {'code': '00', 'name': 'No aplica'}, ] ORDER_KIND_OPTIONS = [ {'code': 'recipe', 'name': 'Receta', 'description': 'Prescripción farmacológica y dispositivos'}, {'code': 'practice', 'name': 'Prácticas', 'description': 'Órdenes de estudios, prácticas y procedimientos'}, {'code': 'report', 'name': 'Informes', 'description': 'Informes clínicos, aptos y constancias médicas'}, {'code': 'result', 'name': 'Resultados', 'description': 'Resultados e interpretación médica interna'}, ] ORDER_KIND_META = { 'recipe': {'label': 'Receta', 'plural': 'Recetas', 'icon': 'bi-file-earmark-medical', 'public_verify': True, 'portal_visible': True, 'number_label': 'Recetario', 'date_label': 'Fecha Receta'}, 'practice': {'label': 'Práctica', 'plural': 'Prácticas', 'icon': 'bi-clipboard2-pulse', 'public_verify': False, 'portal_visible': False, 'number_label': 'Orden Nro', 'date_label': 'Fecha Orden'}, 'report': {'label': 'Informe', 'plural': 'Informes', 'icon': 'bi-file-earmark-text', 'public_verify': False, 'portal_visible': False, 'number_label': 'Informe Nro', 'date_label': 'Fecha Informe'}, 'result': {'label': 'Resultado', 'plural': 'Resultados', 'icon': 'bi-activity', 'public_verify': False, 'portal_visible': False, 'number_label': 'Resultado Nro', 'date_label': 'Fecha Resultado'}, } DEFAULT_SETTINGS = { 'sisa_enabled': '0', 'sisa_wsdl': 'https://sisa.msal.gov.ar/sisa/services/profesionalService?wsdl', 'sisa_user': '', 'sisa_password': '', 'sisa_operation': 'profesionalNominal', 'site_title': 'Book Appointments Pro', 'site_phone': '', 'site_url': 'http://127.0.0.1:5000', 'site_email': '', 'site_logo_path': '', 'site_favicon_path': '', 'site_country': 'Argentina', 'site_province': '', 'site_municipality': '', 'site_city': '', 'site_platform_number': '0000', 'site_repository_number': '0000', 'site_tagline': 'Psiquiatría clínica · receta electrónica · verificación pública', 'site_seo_title': '', 'site_meta_description': 'Centro médico con reserva online de turnos, atención profesional, historia clínica digital y receta electrónica verificable.', 'site_meta_keywords': 'turnos médicos online, salud, clínica, profesionales, receta electrónica, historia clínica digital', 'site_canonical_url': '', 'site_seo_indexing_enabled': '1', 'site_robots_meta': 'index,follow', 'site_lang': 'es-AR', 'site_locale': 'es_AR', 'site_og_title': '', 'site_og_description': '', 'site_og_image_path': '', 'site_og_image_alt': 'Imagen institucional del sitio', 'site_twitter_card': 'summary_large_image', 'site_twitter_site': '', 'site_twitter_title': '', 'site_twitter_description': '', 'site_schema_type': 'MedicalClinic', 'site_legal_name': '', 'site_price_range': '$$', 'site_google_site_verification': '', 'site_bing_site_verification': '', 'site_contact_address': '', 'site_contact_map_embed': '', 'site_copyright': '© Book Appointments Pro Salud. Todos los derechos reservados.', 'site_nav_cta_label': 'Reservar turno', 'site_social_facebook_enabled': '0', 'site_social_facebook_url': '', 'site_social_instagram_enabled': '0', 'site_social_instagram_url': '', 'site_social_x_enabled': '0', 'site_social_x_url': '', 'site_social_youtube_enabled': '0', 'site_social_youtube_url': '', 'site_social_linkedin_enabled': '0', 'site_social_linkedin_url': '', 'wa_floating_button_enabled': '0', 'wa_public_number': '', 'wa_public_message': 'Hola, quiero realizar una consulta.', 'wa_public_label': 'Escribinos por WhatsApp', 'ai_chatbot_enabled': '0', 'ai_chatbot_floating_enabled': '0', 'ai_chatbot_booking_enabled': '1', 'ai_chatbot_assistant_name': 'Asistente Virtual', 'ai_chatbot_welcome_message': 'Hola, soy el asistente virtual. Puedo orientarte sobre servicios, especialidades, precios y turnos.', 'ai_chatbot_openai_enabled': '0', 'ai_chatbot_openai_model': 'gpt-4.1-mini', 'ai_chatbot_api_key': '', 'ai_chatbot_system_instructions': 'Respondé de forma breve, clara y segura. No brindes diagnóstico médico. Orientá al paciente y sugerí reservar turno cuando corresponda.', 'ai_chatbot_fallback_message': 'Puedo orientarte con especialidades, profesionales, precios y turnos. También puedo ayudarte a iniciar una reserva.', 'ai_chatbot_sync_public_site': '1', 'mercadopago_gateway_enabled': '0', 'mercadopago_access_token': '', 'mercadopago_public_base_url': '', 'mercadopago_currency': 'ARS', 'mercadopago_default_payer_email': '', 'mercadopago_last_test_url': '', 'mercadopago_last_test_preference_id': '', 'smtp_host': '', 'smtp_port': '587', 'smtp_username': '', 'smtp_password': '', 'smtp_use_tls': '1', 'smtp_from_email': '', 'smtp_from_name': 'Book Appointments Pro', 'message_recipe_subject': 'Tu receta electrónica {{ legal_number }}', 'message_recipe_html': '
Hola {{ patient_name }},
Adjuntamos tu receta electrónica emitida por {{ professional_name }}.
Número legal: {{ legal_number }}
CUIR: {{ cuir }}
Podés verificarla aquí: {{ verify_url }}
', 'backup_provider': 'local', 'backup_target_path': '', 'backup_retention_years': '3', 'backup_remote_host': '', 'backup_remote_user': '', 'backup_remote_password': '', 'site_primary_color': '#1977cc', 'site_secondary_color': '#0d6efd', 'site_accent_color': '#0dcaf0', 'site_body_font': 'Roboto', 'site_heading_font': 'Poppins', 'site_base_font_size': '16', 'site_topbar_bg': '#1977cc', 'site_topbar_text': '#ffffff', 'site_header_bg': '#ffffff', 'site_header_text': '#2c4964', 'site_nav_text': '#2c4964', 'site_nav_hover': '#1977cc', 'site_hero_bg': '#f3f9fd', 'site_hero_surface': '#ffffff', 'site_section_bg': '#ffffff', 'site_section_muted_bg': '#f7fbff', 'site_card_bg': '#ffffff', 'site_card_text': '#444444', 'site_title_color': '#2c4964', 'site_footer_bg': '#0f2b45', 'site_footer_text': '#ffffff', 'site_button_text': '#ffffff', 'site_border_radius': '20', 'site_popup_enabled': '0', 'site_popup_title': 'Información importante', 'site_popup_html': 'Bienvenido al sitio.
', 'site_popup_image_path': '', 'site_popup_link_text': '', 'site_popup_link_url': '', 'site_cookie_banner_enabled': '1', 'site_cookie_text': 'Usamos cookies técnicas y de experiencia para mejorar el funcionamiento del sitio.
', 'site_terms_html': 'Completar desde administración.
', 'site_privacy_html': 'Completar desde administración.
', 'admin_sidebar_bg': "#111827", 'admin_sidebar_text': "#cbd5e1", 'admin_sidebar_active_bg': "#243447", 'admin_sidebar_active_text': "#ffffff", 'admin_body_bg': "#f3f6fb", 'admin_surface_bg': "#ffffff", 'admin_text_color': "#1f2937", 'admin_muted_text_color': "#64748b", 'admin_primary_color': "#0d6efd", 'admin_border_radius': "18", 'admin_font_family': "Inter, Roboto, system-ui, sans-serif", 'admin_font_size': "15", 'admin_title_color': "#0f172a", 'admin_footer_bg': "#ffffff", 'admin_modal_bg': "#ffffff", 'admin_section_bg': "#ffffff", 'admin_input_bg': "#ffffff", 'admin_input_border': "#dbe4f0", 'message_practice_subject': "Tu orden de pr\u00E1ctica {{ legal_number }}", 'message_practice_html': "Hola {{ patient_name }},
Adjuntamos tu orden de pr\u00E1ctica emitida por {{ professional_name }}.
N\u00FAmero: {{ legal_number }}
", 'message_report_subject': "Tu informe m\u00E9dico {{ legal_number }}", 'message_report_html': "Hola {{ patient_name }},
Adjuntamos tu informe emitido por {{ professional_name }}.
N\u00FAmero: {{ legal_number }}
", 'message_result_subject': "Tu resultado m\u00E9dico {{ legal_number }}", 'message_result_html': "Hola {{ patient_name }},
Adjuntamos tu resultado emitido por {{ professional_name }}.
N\u00FAmero: {{ legal_number }}
", } def role_required(*roles): def decorator(fn): @wraps(fn) def wrapper(*args, **kwargs): if not current_user.is_authenticated: return abort(401) if current_user.role not in roles: return abort(403) return fn(*args, **kwargs) return wrapper return decorator def log_action(action, entity_type, entity_id=None, details=''): actor = current_user.email if getattr(current_user, 'is_authenticated', False) else 'system/public' item = AuditLog( actor_email=actor, action=action, entity_type=entity_type, entity_id=str(entity_id) if entity_id is not None else None, details=details, ) db.session.add(item) db.session.commit() def system_log_action(action, entity_type, entity_id=None, details=''): item = AuditLog( actor_email='system', action=action, entity_type=entity_type, entity_id=str(entity_id) if entity_id is not None else None, details=details, ) db.session.add(item) db.session.commit() def store_error(function_name: str, exc: Exception, extra: str = ''): actor = current_user.email if getattr(current_user, 'is_authenticated', False) else 'system/public' route = request.path if request else None entry = ErrorLog( actor_email=actor, route=route, function_name=function_name, error_type=exc.__class__.__name__, message=str(exc), stacktrace=traceback.format_exc(), extra=extra, ) db.session.add(entry) db.session.commit() def safe_json_dumps(value) -> str: try: return json.dumps(value, ensure_ascii=False, default=str) except Exception: return str(value) def ensure_default_settings(): changed = False for key, value in DEFAULT_SETTINGS.items(): if not AppSetting.query.filter_by(key=key).first(): db.session.add(AppSetting(key=key, value=value)) changed = True if changed: db.session.commit() def get_setting(key: str, default: str = '') -> str: item = AppSetting.query.filter_by(key=key).first() if item is None: if key in DEFAULT_SETTINGS: item = AppSetting(key=key, value=DEFAULT_SETTINGS[key]) db.session.add(item) db.session.commit() else: return default return item.value if item.value is not None else default def set_setting(key: str, value: str): item = AppSetting.query.filter_by(key=key).first() if item is None: item = AppSetting(key=key, value=value) db.session.add(item) else: item.value = value db.session.commit() def get_bool_setting(key: str, default: bool = False) -> bool: return get_setting(key, '1' if default else '0') == '1' def is_sisa_enabled() -> bool: return get_setting('sisa_enabled', '0') == '1' def get_site_settings() -> dict: whatsapp_public_number = sanitize_digits(get_setting('wa_public_number', ''), max_length=32) whatsapp_public_message = get_setting('wa_public_message', 'Hola, quiero realizar una consulta.') whatsapp_floating_enabled = get_bool_setting('wa_floating_button_enabled') whatsapp_floating_url = '' if whatsapp_floating_enabled and whatsapp_public_number: whatsapp_floating_url = f"https://wa.me/{whatsapp_public_number}?text={quote_plus(whatsapp_public_message)}" ai_chatbot_enabled = get_bool_setting('ai_chatbot_enabled') ai_chatbot_floating_enabled = get_bool_setting('ai_chatbot_floating_enabled') socials = [] raw_socials = {} for network, icon in [ ('facebook', 'bi-facebook'), ('instagram', 'bi-instagram'), ('x', 'bi-twitter-x'), ('youtube', 'bi-youtube'), ('linkedin', 'bi-linkedin'), ]: enabled = get_bool_setting(f'site_social_{network}_enabled') url = get_setting(f'site_social_{network}_url') raw_socials[network] = {'enabled': enabled, 'url': url, 'icon': icon} if enabled and url: socials.append({'network': network, 'url': url, 'icon': icon}) return { 'title': get_setting('site_title', 'Book Appointments Pro'), 'phone': get_setting('site_phone'), 'url': get_setting('site_url', current_app.config.get('BASE_URL', 'http://127.0.0.1:5000')) if current_app else get_setting('site_url'), 'email': get_setting('site_email'), 'logo_path': get_setting('site_logo_path'), 'favicon_path': get_setting('site_favicon_path'), 'country': get_setting('site_country', 'Argentina'), 'province': get_setting('site_province'), 'municipality': get_setting('site_municipality'), 'city': get_setting('site_city'), 'platform_number': get_setting('site_platform_number', '0000'), 'repository_number': get_setting('site_repository_number', '0000'), 'tagline': get_setting('site_tagline', 'Psiquiatría clínica · receta electrónica · verificación pública'), 'seo_title': get_setting('site_seo_title'), 'meta_description': get_setting('site_meta_description', 'Centro médico con reserva online de turnos, atención profesional, historia clínica digital y receta electrónica verificable.'), 'meta_keywords': get_setting('site_meta_keywords'), 'canonical_url': get_setting('site_canonical_url'), 'seo_indexing_enabled': get_bool_setting('site_seo_indexing_enabled', True), 'robots_meta': get_setting('site_robots_meta', 'index,follow') if get_bool_setting('site_seo_indexing_enabled', True) else 'noindex,nofollow', 'lang': get_setting('site_lang', 'es-AR'), 'locale': get_setting('site_locale', 'es_AR'), 'og_title': get_setting('site_og_title'), 'og_description': get_setting('site_og_description'), 'og_image_path': get_setting('site_og_image_path'), 'og_image_alt': get_setting('site_og_image_alt', 'Imagen institucional del sitio'), 'twitter_card': get_setting('site_twitter_card', 'summary_large_image'), 'twitter_site': get_setting('site_twitter_site'), 'twitter_title': get_setting('site_twitter_title'), 'twitter_description': get_setting('site_twitter_description'), 'schema_type': get_setting('site_schema_type', 'MedicalClinic'), 'legal_name': get_setting('site_legal_name'), 'price_range': get_setting('site_price_range', '$$'), 'google_site_verification': get_setting('site_google_site_verification'), 'bing_site_verification': get_setting('site_bing_site_verification'), 'contact_address': get_setting('site_contact_address'), 'contact_map_embed': get_setting('site_contact_map_embed'), 'copyright': get_setting('site_copyright', '© Book Appointments Pro Salud. Todos los derechos reservados.'), 'nav_cta_label': get_setting('site_nav_cta_label', 'Reservar turno'), 'whatsapp_floating_enabled': whatsapp_floating_enabled, 'whatsapp_public_number': whatsapp_public_number, 'whatsapp_public_message': whatsapp_public_message, 'whatsapp_public_label': get_setting('wa_public_label', 'Escribinos por WhatsApp'), 'whatsapp_floating_url': whatsapp_floating_url, 'ai_chatbot_enabled': ai_chatbot_enabled, 'ai_chatbot_floating_enabled': ai_chatbot_floating_enabled, 'ai_chatbot_booking_enabled': get_bool_setting('ai_chatbot_booking_enabled', True), 'ai_chatbot_assistant_name': get_setting('ai_chatbot_assistant_name', 'Asistente Virtual'), 'ai_chatbot_welcome_message': get_setting('ai_chatbot_welcome_message', 'Hola, soy el asistente virtual. Puedo orientarte sobre servicios, especialidades, precios y turnos.'), 'retention_years': get_setting('backup_retention_years', '3'), 'primary_color': get_setting('site_primary_color', '#1977cc'), 'secondary_color': get_setting('site_secondary_color', '#0d6efd'), 'accent_color': get_setting('site_accent_color', '#0dcaf0'), 'body_font': get_setting('site_body_font', 'Roboto'), 'heading_font': get_setting('site_heading_font', 'Poppins'), 'base_font_size': get_setting('site_base_font_size', '16'), 'topbar_bg': get_setting('site_topbar_bg', '#1977cc'), 'topbar_text': get_setting('site_topbar_text', '#ffffff'), 'header_bg': get_setting('site_header_bg', '#ffffff'), 'header_text': get_setting('site_header_text', '#2c4964'), 'nav_text': get_setting('site_nav_text', '#2c4964'), 'nav_hover': get_setting('site_nav_hover', '#1977cc'), 'hero_bg': get_setting('site_hero_bg', '#f3f9fd'), 'hero_surface': get_setting('site_hero_surface', '#ffffff'), 'section_bg': get_setting('site_section_bg', '#ffffff'), 'section_muted_bg': get_setting('site_section_muted_bg', '#f7fbff'), 'card_bg': get_setting('site_card_bg', '#ffffff'), 'card_text': get_setting('site_card_text', '#444444'), 'title_color': get_setting('site_title_color', '#2c4964'), 'footer_bg': get_setting('site_footer_bg', '#0f2b45'), 'footer_text': get_setting('site_footer_text', '#ffffff'), 'button_text': get_setting('site_button_text', '#ffffff'), 'border_radius': get_setting('site_border_radius', '20'), 'popup_enabled': get_bool_setting('site_popup_enabled', False), 'popup_title': get_setting('site_popup_title', 'Información importante'), 'popup_html': get_setting('site_popup_html', 'Bienvenido al sitio.
'), 'popup_image_path': get_setting('site_popup_image_path'), 'popup_link_text': get_setting('site_popup_link_text'), 'popup_link_url': get_setting('site_popup_link_url'), 'cookie_banner_enabled': get_bool_setting('site_cookie_banner_enabled', True), 'cookie_text': get_setting('site_cookie_text', 'Usamos cookies técnicas y de experiencia para mejorar el funcionamiento del sitio.
'), 'terms_html': get_setting('site_terms_html', 'Completar desde administración.
'), 'privacy_html': get_setting('site_privacy_html', 'Completar desde administración.
'), 'admin_style': { 'sidebar_bg': get_setting('admin_sidebar_bg', '#111827'), 'sidebar_text': get_setting('admin_sidebar_text', '#cbd5e1'), 'sidebar_active_bg': get_setting('admin_sidebar_active_bg', '#243447'), 'sidebar_active_text': get_setting('admin_sidebar_active_text', '#ffffff'), 'body_bg': get_setting('admin_body_bg', '#f3f6fb'), 'surface_bg': get_setting('admin_surface_bg', '#ffffff'), 'text_color': get_setting('admin_text_color', '#1f2937'), 'muted_text_color': get_setting('admin_muted_text_color', '#64748b'), 'primary_color': get_setting('admin_primary_color', '#0d6efd'), 'border_radius': get_setting('admin_border_radius', '18'), 'font_family': get_setting('admin_font_family', 'Inter, Roboto, system-ui, sans-serif'), 'font_size': get_setting('admin_font_size', '15'), 'title_color': get_setting('admin_title_color', '#0f172a'), 'footer_bg': get_setting('admin_footer_bg', '#ffffff'), 'modal_bg': get_setting('admin_modal_bg', '#ffffff'), 'section_bg': get_setting('admin_section_bg', '#ffffff'), 'input_bg': get_setting('admin_input_bg', '#ffffff'), 'input_border': get_setting('admin_input_border', '#dbe4f0'), }, 'socials': socials, 'socials_raw': raw_socials, } def generate_public_token(): return token_urlsafe(24) def hash_payload(value) -> str: if not isinstance(value, str): value = safe_json_dumps(value) return hashlib.sha256(value.encode('utf-8')).hexdigest() def overlaps(start_a: datetime, end_a: datetime, start_b: datetime, end_b: datetime): return start_a < end_b and start_b < end_a def professional_is_on_leave(professional_id: int, target_date: date) -> bool: return Leave.query.filter( Leave.professional_id == professional_id, Leave.start_date <= target_date, Leave.end_date >= target_date, ).first() is not None def get_working_blocks(professional_id: int, target_date: date): return WorkingHour.query.filter_by( professional_id=professional_id, weekday=target_date.weekday(), is_active=True, ).order_by(WorkingHour.start_time.asc()).all() def appointment_overlaps_existing(professional_id: int, target_date: date, start_at: datetime, end_at: datetime, ignore_appointment_id: int | None = None): appointments = Appointment.query.filter( Appointment.professional_id == professional_id, Appointment.appointment_date == target_date, Appointment.status.in_(ACTIVE_APPOINTMENT_STATUSES), ).all() for appt in appointments: if ignore_appointment_id and appt.id == ignore_appointment_id: continue if overlaps(start_at, end_at, appt.starts_at, appt.ends_at): return True return False def ensure_clinical_record(patient: Patient, retention_years: int = 10): if not patient: return None existing = ClinicalRecord.query.filter_by(patient_id=patient.id).first() if existing: return existing today = date.today() record = ClinicalRecord( patient_id=patient.id, legajo_number=(patient.documento or f'LEG-{patient.id}').strip(), retention_until=date(today.year + retention_years, today.month, today.day), ) db.session.add(record) db.session.flush() return record def get_frontend_blocks(section: str) -> list[FrontendBlock]: return FrontendBlock.query.filter_by(section=section, is_active=True).order_by(FrontendBlock.sort_order.asc(), FrontendBlock.id.asc()).all() def get_frontend_payload() -> dict: sections = ['navbar_menu', 'about_items', 'stats_items', 'service_cards', 'department_cards', 'faq_items', 'gallery_items', 'contact_items', 'footer_columns'] payload = {section: get_frontend_blocks(section) for section in sections} return payload def get_available_slots(professional, service, target_date: date, step_minutes: int = 15): if not professional or not service or not professional.is_bookable or professional_is_on_leave(professional.id, target_date): return [] if service not in professional.services: return [] now = datetime.now() min_notice_cutoff = now + timedelta(hours=service.min_notice_hours or 0) slot_delta = timedelta(minutes=step_minutes) total_duration = timedelta(minutes=service.duration_minutes) buffer_before = timedelta(minutes=service.buffer_before_minutes or 0) buffer_after = timedelta(minutes=service.buffer_after_minutes or 0) blocks = get_working_blocks(professional.id, target_date) if not blocks: return [] results = [] for block in blocks: block_start = datetime.combine(target_date, block.start_time) block_end = datetime.combine(target_date, block.end_time) cursor = block_start while cursor + total_duration <= block_end: visible_start = cursor visible_end = cursor + total_duration blocked_start = visible_start - buffer_before blocked_end = visible_end + buffer_after if visible_start >= min_notice_cutoff and not appointment_overlaps_existing( professional.id, target_date, blocked_start, blocked_end, ): results.append({ 'start': visible_start, 'end': visible_end, 'label': visible_start.strftime('%H:%M'), }) cursor += slot_delta unique = [] seen = set() for slot in results: if slot['label'] not in seen: unique.append(slot) seen.add(slot['label']) return unique def choose_first_available_professional(service, target_date: date): professionals = [p for p in service.professionals if p.is_bookable] ranked = [] for professional in professionals: slots = get_available_slots(professional, service, target_date) if slots: ranked.append((slots[0]['start'], professional, slots)) ranked.sort(key=lambda item: (item[0], item[1].display_name.lower())) if ranked: _, professional, slots = ranked[0] return professional, slots return None, [] def choose_round_robin_professional(service, target_date: date): professionals = [p for p in service.professionals if p.is_bookable] if not professionals: return None, [] counts = [] for p in professionals: future_count = Appointment.query.filter( Appointment.professional_id == p.id, Appointment.appointment_date >= date.today(), Appointment.status.in_(ACTIVE_APPOINTMENT_STATUSES), ).count() slots = get_available_slots(p, service, target_date) if slots: counts.append((future_count, slots[0]['start'], p, slots)) counts.sort(key=lambda item: (item[0], item[1], item[2].display_name.lower())) if counts: _, _, professional, slots = counts[0] return professional, slots return None, [] def professional_scope_filter(query, professional_profile_id: int | None): if professional_profile_id: return query.filter_by(professional_id=professional_profile_id) return query def parse_date(value: str, default=None): try: return datetime.strptime(value, '%Y-%m-%d').date() except Exception: return default def parse_time(value: str, default=None): try: return datetime.strptime(value, '%H:%M').time() except Exception: return default def today_iso(): return date.today().isoformat() def start_of_week(target: date | None = None): target = target or date.today() return target - timedelta(days=target.weekday()) def build_week_dates(target: date | None = None): monday = start_of_week(target) return [monday + timedelta(days=i) for i in range(7)] def get_due_reminders(window_hours: int = 24, professional_id: int | None = None): now = datetime.now() limit = now + timedelta(hours=window_hours) query = Appointment.query.filter( Appointment.status.in_(ACTIVE_APPOINTMENT_STATUSES), Appointment.reminder_sent.is_(False), Appointment.appointment_date >= now.date(), ) if professional_id: query = query.filter(Appointment.professional_id == professional_id) items = [] for appt in query.order_by(Appointment.appointment_date.asc(), Appointment.start_time.asc()).all(): if now <= appt.starts_at <= limit: items.append(appt) return items def digits_only(value: str) -> str: return re.sub(r'\D+', '', value or '') def pad_numeric(value: str, width: int) -> str: digits = digits_only(value) return digits.zfill(width)[-width:] def mask_last4_dni(value: str) -> str: digits = digits_only(value) if len(digits) <= 4: return digits return f"****{digits[-4:]}" def resolve_jurisdiction_code(value: str) -> str: raw = (value or '').strip() if not raw: return '00' direct = digits_only(raw) if len(direct) >= 2: code = direct[:2] for item in JURISDICTION_OPTIONS: if item['code'] == code: return code raw_lower = raw.lower() for item in JURISDICTION_OPTIONS: if item['name'].lower() == raw_lower: return item['code'] if raw_lower in item['name'].lower() or item['name'].lower() in raw_lower: return item['code'] return '00' def resolve_jurisdiction_name(code_or_value: str) -> str: code = resolve_jurisdiction_code(code_or_value) for item in JURISDICTION_OPTIONS: if item['code'] == code: return item['name'] return (code_or_value or '').strip() def generate_prescription_group() -> str: seed = datetime.utcnow().strftime('%Y%m%d%H%M%S%f') return seed[:25].ljust(25, '0') def build_cuir(platform_number: str, repository_number: str, jurisdiction_code: str, prescription_type: str, prescription_subtype: str, prescription_group: str, item_number: str) -> str: return ''.join([ pad_numeric(platform_number, 4), pad_numeric(repository_number, 4), pad_numeric(jurisdiction_code, 2), pad_numeric(prescription_type, 2) + pad_numeric(prescription_subtype, 2), pad_numeric(prescription_group, 25), pad_numeric(item_number, 2), ]) def generate_legal_number() -> str: return datetime.utcnow().strftime('%Y%m%d%H%M%S%f')[:20] def save_uploaded_asset(file_storage, upload_dir: str, prefix: str = 'site') -> str: if not file_storage or not getattr(file_storage, 'filename', ''): return '' original_name, ext = validate_uploaded_asset(file_storage, prefix=prefix) safe_stem = sanitize_identifier(os.path.splitext(original_name)[0], max_length=60) or prefix safe_name = f"{prefix}_{safe_stem}_{datetime.utcnow().strftime('%Y%m%d%H%M%S%f')}{ext}" os.makedirs(upload_dir, exist_ok=True) full_path = os.path.join(upload_dir, safe_name) file_storage.save(full_path) return f'uploads/{safe_name}' def get_smtp_settings() -> dict: return { 'host': get_setting('smtp_host'), 'port': int(get_setting('smtp_port', '587') or 587), 'username': get_setting('smtp_username'), 'password': get_setting('smtp_password'), 'use_tls': get_bool_setting('smtp_use_tls', True), 'from_email': get_setting('smtp_from_email') or get_setting('site_email'), 'from_name': get_setting('smtp_from_name', 'Book Appointments Pro'), 'recipe_subject': get_setting('message_recipe_subject', 'Tu receta electrónica {{ legal_number }}'), 'recipe_html': get_setting('message_recipe_html', 'Adjuntamos tu receta electrónica.
'), 'practice_subject': get_setting('message_practice_subject', 'Tu orden de práctica {{ legal_number }}'), 'practice_html': get_setting('message_practice_html', 'Adjuntamos tu orden de práctica.
'), 'report_subject': get_setting('message_report_subject', 'Tu informe médico {{ legal_number }}'), 'report_html': get_setting('message_report_html', 'Adjuntamos tu informe médico.
'), 'result_subject': get_setting('message_result_subject', 'Tu resultado médico {{ legal_number }}'), 'result_html': get_setting('message_result_html', 'Adjuntamos tu resultado médico.
'), } def render_message_template(source: str, context: dict) -> str: source = source or '' return Template(source).render(**context) def sha256_file(path: str) -> str: h = hashlib.sha256() with open(path, 'rb') as f: for chunk in iter(lambda: f.read(65536), b''): h.update(chunk) return h.hexdigest() def send_smtp_email(to_email: str, subject: str, html_body: str, attachments: list[dict] | None = None): settings = get_smtp_settings() if not settings['host'] or not settings['from_email']: raise ValueError('SMTP incompleto: faltan host o remitente.') msg = EmailMessage() msg['Subject'] = subject msg['From'] = f"{settings['from_name']} <{settings['from_email']}>" if settings['from_name'] else settings['from_email'] msg['To'] = to_email msg.set_content('Este mensaje contiene una versión HTML.') msg.add_alternative(html_body or 'Sin contenido.
', subtype='html') for item in attachments or []: msg.add_attachment(item['content'], maintype=item.get('maintype', 'application'), subtype=item.get('subtype', 'octet-stream'), filename=item.get('filename', 'archivo.bin')) with smtplib.SMTP(settings['host'], settings['port']) as smtp: if settings['use_tls']: smtp.starttls() if settings['username']: smtp.login(settings['username'], settings['password']) smtp.send_message(msg)