LQAdminPlatform/update_app_callback.py

#!/usr/bin/env python3
"""
更新应用回调URL
"""
import pymysql
from urllib.parse import urlparse
import os
import json
from dotenv import load_dotenv

load_dotenv()

def update_app_callback():
    """更新应用回调URL"""
    database_url = os.getenv('DATABASE_URL', '')
    parsed = urlparse(database_url)
    config = {
        'host': parsed.hostname or 'localhost',
        'port': parsed.port or 3306,
        'user': parsed.username or 'root',
        'password': parsed.password or '',
        'database': parsed.path[1:] if parsed.path else 'sso_db',
        'charset': 'utf8mb4'
    }

    try:
        conn = pymysql.connect(**config)
        cursor = conn.cursor()
        
        # 更新回调URL
        new_redirect_uris = [
            "http://localhost:8001/auth/callback",  # 子系统后端回调
            "http://localhost:3001/auth/callback",  # 子系统前端回调
            "http://localhost:3000/callback",       # 原有的回调URL
            "http://localhost:8080/callback"        # 原有的回调URL
        ]
        
        cursor.execute("""
            UPDATE apps 
            SET redirect_uris = %s, updated_at = NOW()
            WHERE app_key = %s
        """, (json.dumps(new_redirect_uris), 'eqhoIdAyAWbA8MsYHsNqQqNLJbCayTjY'))
        
        conn.commit()
        
        # 验证更新
        cursor.execute("""
            SELECT name, redirect_uris 
            FROM apps 
            WHERE app_key = %s
        """, ('eqhoIdAyAWbA8MsYHsNqQqNLJbCayTjY',))
        
        result = cursor.fetchone()
        if result:
            print(f"✅ 应用回调URL更新成功")
            print(f"应用名称: {result[0]}")
            print(f"回调URL列表:")
            urls = json.loads(result[1])
            for i, url in enumerate(urls, 1):
                print(f"  {i}. {url}")
        
        cursor.close()
        conn.close()
        
    except Exception as e:
        print(f'❌ 更新失败: {e}')

if __name__ == "__main__":
    update_app_callback()