orchestrator/alert.py

import sqlite3
import os
from datetime import datetime

DB_PATH = os.path.join(os.path.dirname(__file__), 'data', 'mvp.db')

def ensure_system_error_log_table():
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS system_error_log (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            module TEXT,
            error_type TEXT,
            error_msg TEXT,
            task_id TEXT,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    conn.close()

def send_feishu_alert(error_type, error_msg, module):
    """
    Mock sending a webhook to Feishu and inserting the error into `system_error_log` table in `mvp.db`.
    """
    ensure_system_error_log_table()
    
    # Mock sending to Feishu webhook
    print(f"🚨 [Feishu Alert] 发送飞书报警:")
    print(f"   模块: {module}")
    print(f"   类型: {error_type}")
    print(f"   信息: {error_msg}")
    
    # Insert error into system_error_log
    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
        cursor.execute("""
            INSERT INTO system_error_log (module, error_type, error_msg, created_at)
            VALUES (?, ?, ?, ?)
        """, (module, error_type, error_msg, datetime.now().isoformat()))
        conn.commit()
        print("✅ [Feishu Alert] 报警日志已记录至 system_error_log。")
    except Exception as e:
        print(f"❌ [Feishu Alert] 无法写入系统错误日志: {e}")
    finally:
        if 'conn' in locals():
            conn.close()

if __name__ == '__main__':
    # Test the function
    send_feishu_alert(
        error_type="DatabaseConnectionError",
        error_msg="Failed to connect to primary replica.",
        module="data_sync"
    )