utils.py

# Standard library imports
import asyncio
import atexit
import functools
import hashlib
import html
import os
import random
import re
import time
import uuid
from contextvars import ContextVar
from typing import Optional

# Third-party imports
import requests

# Frappe framework imports
import frappe

# ─────────────────────────────────────────────────────────────────────────────
# Telegram API Session with Connection Pooling
# ─────────────────────────────────────────────────────────────────────────────

# Reusable session with connection pooling for Telegram API calls
_telegram_session: Optional[requests.Session] = None

def get_telegram_session() -> requests.Session:
    """
    Get or create a requests.Session with connection pooling.
    Uses a persistent session to reuse TCP connections across API calls.
    """
    global _telegram_session
    if _telegram_session is None:
        _telegram_session = requests.Session()
        adapter = requests.adapters.HTTPAdapter(
            pool_connections=10,
            pool_maxsize=20,
            max_retries=3,
            pool_block=False
        )
        _telegram_session.mount("https://", adapter)
        _telegram_session.mount("http://", adapter)
    return _telegram_session

def close_telegram_session():
    """Close the Telegram session (called on app shutdown via atexit)."""
    global _telegram_session
    if _telegram_session:
        _telegram_session.close()
        _telegram_session = None

# PERF-13 FIX: Register atexit handler so the session is closed when the process exits
atexit.register(close_telegram_session)

# ─────────────────────────────────────────────────────────────────────────────
# Structured Logging with Correlation IDs
# ─────────────────────────────────────────────────────────────────────────────

# Context variable to store correlation ID across the request/job lifecycle
_correlation_id_var: ContextVar[str] = ContextVar('correlation_id', default='')

def get_correlation_id() -> str:
    """Get the current correlation ID or generate a new one. Thread-safe and async-safe."""
    corr_id = _correlation_id_var.get()
    if not corr_id:
        corr_id = uuid.uuid4().hex[:12].upper()
        _correlation_id_var.set(corr_id)
    return corr_id

def set_correlation_id(corr_id: str) -> None:
    """Set a specific correlation ID for the current context."""
    _correlation_id_var.set(corr_id)

def clear_correlation_id() -> None:
    """Clear the correlation ID for the current context."""
    _correlation_id_var.set('')

def structured_log(level: str, event: str, **context) -> None:
    """Structured logging with correlation ID and arbitrary context."""
    corr_id = get_correlation_id()
    log_data = {
        'correlation_id': corr_id,
        'event': event,
        **context
    }
    logger = frappe.logger("bespo_notifications")
    log_func = getattr(logger, level.lower(), logger.info)
    log_func(f"[{corr_id}] {event}: {log_data}")

def get_log_context(**defaults):
    """Get a context dict with correlation_id pre-populated."""
    return {
        'correlation_id': get_correlation_id(),
        **defaults
    }

# ─────────────────────────────────────────────────────────────────────────────
# Centralized URL Normalization
# ─────────────────────────────────────────────────────────────────────────────

def normalize_telegram_api_url(api_url=None):
    """
    Normalize Telegram API URL by stripping trailing slashes.
    Centralized utility to replace duplicate patterns across codebase.
    """
    return (api_url or "https://api.telegram.org/bot").rstrip('/')

# ─────────────────────────────────────────────────────────────────────────────
# Async PDF Generation
# ─────────────────────────────────────────────────────────────────────────────

def generate_pdf_safely_async(doctype, docname, site_name=None):
    """
    Async wrapper for PDF generation to prevent blocking in async contexts.
    Offloads PDF generation to a thread pool executor.
    
    Returns:
        coroutine that resolves to PDF bytes or None on failure
    """
    import concurrent.futures
    
    def _generate():
        return generate_pdf_safely(doctype, docname, site_name)
    
    loop = asyncio.new_event_loop()
    executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)
    try:
        future = executor.submit(_generate)
        return loop.run_in_executor(executor, _generate)
    finally:
        executor.shutdown(wait=False)

def generate_pdf_safely(doctype, docname, site_name=None):
    """
    Generate PDF for a document safely. Synchronous version.
    
    Returns:
        PDF bytes or None on failure
    """
    if not doctype or not docname:
        return None
        
    try:
        if site_name:
            frappe.init(site_name)
            frappe.connect()
        
        base_url = frappe.utils.get_url()
        pdf_url = (
            f"{base_url}/api/method/frappe.utils.print_format.download_pdf"
            f"?doctype={doctype}&name={docname}&format=Standard&no_letterhead=0"
        )
        
        frappe.local.session_obj = frappe.session
        sid = frappe.session.get("sid") or frappe.db.get_value(
            "Session Default", {"default": "Website"}, "user"
        )
        
        headers = {}
        if frappe.session:
            headers["Cookie"] = f"sid={frappe.session.sid}" if hasattr(frappe.session, "sid") else ""
        
        # Use pooled session for PDF generation
        session = get_telegram_session()
        response = session.get(pdf_url, headers=headers, timeout=30)
        
        if response.ok and response.content.startswith(b"%PDF"):
            return response.content
        else:
            frappe.logger("bespo_notifications").warning(
                f"PDF generation failed for {doctype}/{docname}: status={response.status_code}"
            )
            return None
            
    except Exception as e:
        frappe.logger("bespo_notifications").error(
            f"Error generating PDF for {doctype}/{docname}: {e}"
        )
        return None

def run_async_safely(coro):
    """
    Safely executes an asynchronous coroutine within Frappe's synchronous 
    RQ worker threads, preventing 'Event loop is closed' errors.
    """
    try:
        loop = asyncio.get_event_loop()
        if loop.is_closed():
            raise RuntimeError
    except RuntimeError:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
    
    try:
        return loop.run_until_complete(coro)
    finally:
        pending = asyncio.all_tasks(loop)
        for task in pending:
            task.cancel()

def sanitize_telegram_html(text):
    """
    Strips unsupported HTML tags from a string before sending to Telegram
    (Parse Mode: HTML). Converts block tags to newlines and removes unsupported tags.
    """
    if not text:
        return ""
    
    # 1. Replace block-level tags with newlines
    text = re.sub(r'</?(p|div|br\s*/?)>', '\n', text, flags=re.IGNORECASE)
    
    # 2. Strip unsupported tags but keep valid ones
    valid_tags = ['b', 'strong', 'i', 'em', 'u', 'ins', 's', 'strike', 'del', 'a', 'code', 'pre', 'blockquote']
    
    def replace_tag(match):
        full_tag = match.group(0)
        tag_name = match.group(1).lower()
        if tag_name in valid_tags:
            return full_tag
        return ""
    
    text = re.sub(r'</?([a-zA-Z0-9]+)[^>]*>', replace_tag, text)
    
    # 3. Clean up excessive newlines
    text = re.sub(r'\n{3,}', '\n\n', text).strip()
    return text

def clean_jinja_template(template_html):
    """Removes HTML tags accidentally inserted inside Jinja tags."""
    if not template_html:
        return ""
    
    def replacer(match):
        content = match.group(0)
        cleaned = re.sub(r'<[^>]+>', '', content)
        return html.unescape(cleaned)
    
    template_html = re.sub(r'\{\{.*?(?:\}\}|$)', replacer, template_html, flags=re.DOTALL)
    template_html = re.sub(r'\{%.*?(?:%\}|$)', replacer, template_html, flags=re.DOTALL)
    return template_html

def _debug_log(msg):
    frappe.logger("bespo_notifications").debug(msg)

def _get_proxies(bot_name):
    if not bot_name: return None
    try:
        data = frappe.db.get_value(
            "Telegram Bot", bot_name,
            ["use_proxy", "http_proxy", "https_proxy"],
            as_dict=True, ignore_permissions=True
        )
        if not data or not data.use_proxy: return None
        proxies = {}
        if data.http_proxy: proxies["http"] = data.http_proxy
        if data.https_proxy: proxies["https"] = data.https_proxy
        return proxies if proxies else None
    except Exception:
        return None

def _tg_reply(bot_token, chat_id, text, parse_mode="HTML", bot_api_url=None, bot_name=None):
    """Send a reply message to Telegram. Best-effort - failures are logged as warnings."""
    base_url = (bot_api_url or "https://api.telegram.org/bot").rstrip('/')
    proxies = _get_proxies(bot_name)
    try: 
        # Use pooled session
        session = get_telegram_session()
        session.post(
            f"{base_url}{bot_token}/sendMessage", 
            json={"chat_id": chat_id, "text": text, "parse_mode": parse_mode}, 
            timeout=5, 
            proxies=proxies
        )
    except Exception as e: 
        frappe.logger("bespo_notifications").warning(f"Failed to send Telegram reply to {chat_id}: {e}")

def _send_telegram_document(bot_token, chat_id, file_content, filename, caption=None, parse_mode="HTML", bot_api_url=None, bot_name=None, message_thread_id=None):
    base_url = (bot_api_url or "https://api.telegram.org/bot").rstrip('/')
    proxies = _get_proxies(bot_name)
    try:
        url = f"{base_url}{bot_token}/sendDocument"
        files = {'document': (filename, file_content, 'application/pdf')}
        data = {'chat_id': chat_id, 'caption': caption, 'parse_mode': parse_mode}
        if message_thread_id:
            try:
                data['message_thread_id'] = int(message_thread_id)
            except ValueError:
                frappe.logger("bespo_notifications").warning(f"Invalid message_thread_id: {message_thread_id}")
        # Use pooled session
        session = get_telegram_session()
        resp = session.post(url, data=data, files=files, timeout=20, proxies=proxies)
        return resp.json()
    except Exception as e:
        frappe.logger("bespo_notifications").error(f"Error sending Telegram document: {e}")
        return {"ok": False, "description": str(e)}

def _process_telegram_update(bot_name, bot_token, update):
    try:
        cq = update.get("callback_query")
        if cq:
            bot_api_url = (frappe.db.get_value("Telegram Bot", bot_name, "bot_api_url") or "https://api.telegram.org/bot").rstrip('/')
            _handle_workflow_callback(bot_name, bot_token, cq, bot_api_url=bot_api_url)
            return True

        my_chat_member = update.get("my_chat_member")
        if my_chat_member:
            chat_info = my_chat_member.get("chat", {})
            new_status = my_chat_member.get("new_chat_member", {}).get("status")
            chat_id = str(chat_info.get("id"))
            chat_title = chat_info.get("title") or chat_info.get("first_name", "Unknown Chat")
            chat_type = chat_info.get("type")
            erp_chat_type = {"private": "Private User", "group": "Group", "supergroup": "Supergroup", "channel": "Channel"}.get(chat_type, "Group")

            if new_status in ["member", "administrator"]:
                existing_group = frappe.db.get_value("Telegram Chat Profile", {"telegram_chat_id": chat_id, "linked_bot": bot_name}, "name")
                if not existing_group:
                    doc = frappe.new_doc("Telegram Chat Profile")
                    doc.chat_name = f"{chat_title} ({chat_id})"
                    doc.telegram_chat_id = chat_id
                    doc.chat_type = erp_chat_type
                    doc.linked_bot = bot_name
                    if erp_chat_type == "Supergroup": doc.topics_enabled = 1
                    try:
                        doc.insert(ignore_permissions=True)
                        bot_api_url = (frappe.db.get_value("Telegram Bot", bot_name, "bot_api_url") or "https://api.telegram.org/bot").rstrip('/')
                        _tg_reply(bot_token, chat_id, f"✅ Bespo QMS Integration Active (Bot: {bot_name}).\nThis chat has been registered as: *{doc.name}*.", parse_mode="Markdown", bot_api_url=bot_api_url)
                    except Exception as e: 
                        frappe.logger("bespo_notifications").error(f"Error registering new chat profile {chat_id}: {e}")
            return True

        message = update.get("message") or update.get("channel_post") or {}
        text = (message.get("text") or "").strip()
        chat_info = message.get("chat", {})
        chat_id = str(chat_info.get("id", ""))
        chat_type = chat_info.get("type", "private")
        
        if not chat_id: return False

        if not text:
            if chat_type == "private" and any(k in message for k in ["photo", "document", "video", "voice", "audio"]):
                bot_api_url = (frappe.db.get_value("Telegram Bot", bot_name, "bot_api_url") or "https://api.telegram.org/bot").rstrip('/')
                _tg_reply(bot_token, chat_id, "I am a notification bot and cannot process media attachments. Please log into the ERP to upload files.", bot_api_url=bot_api_url, bot_name=bot_name)
            return False

        if text.lower() in ["/start", "/menu", "menu"]:
            bot_api_url = (frappe.db.get_value("Telegram Bot", bot_name, "bot_api_url") or "https://api.telegram.org/bot").rstrip('/')
            _handle_telegram_menu(bot_token, chat_id, message, bot_api_url=bot_api_url, bot_name=bot_name)
            return True

        if text.lower().startswith("/verify"):
            parts = text.split()
            if len(parts) < 2:
                bot_api_url = (frappe.db.get_value("Telegram Bot", bot_name, "bot_api_url") or "https://api.telegram.org/bot").rstrip('/')