CloudSearch/cloudsearch_admin/server.py

#!/usr/bin/env python3
"""
CloudSearch 管理后台 v2.2.0
功能开关一键管理 — 支持本地 SQLite + MySQL 双向同步
"""

import os
import json
import sqlite3
import logging
from datetime import datetime
from typing import Dict, Optional

from flask import Flask, render_template, request, jsonify

# ── 日志 ──────────────────────────────────────────────────
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("admin")

app = Flask(__name__)

# ── 配置 ──────────────────────────────────────────────────
ADMIN_PORT = int(os.getenv("ADMIN_PORT", "9531"))
ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD", "admin123")
DB_PATH = os.getenv("ADMIN_DB_PATH", "/data/admin_flags.sqlite")

# MySQL（主应用 system_configs 表，可选）
MYSQL_HOST = os.getenv("MYSQL_HOST", "")
MYSQL_PORT = int(os.getenv("MYSQL_PORT", "3306"))
MYSQL_USER = os.getenv("MYSQL_USER", "")
MYSQL_PASS = os.getenv("MYSQL_PASS", "")
MYSQL_DB   = os.getenv("MYSQL_DB", "cloudsearch")

# ── 功能开关定义 ──────────────────────────────────────────
FEATURES: Dict[str, dict] = {
    "feature_quark_pid":      {"name": "夸克推广PID",     "group": "核心", "default": True},
    "feature_seo":            {"name": "SEO / Sitemap",   "group": "核心", "default": True},
    "feature_link_monitor":   {"name": "失效链接监控",     "group": "核心", "default": True},
    "feature_tmdb":           {"name": "TMDB影视刮削",    "group": "增强", "default": True},
    "feature_telegram_bot":   {"name": "Telegram Bot",    "group": "增强", "default": False},
    "feature_subscription":   {"name": "关键词订阅通知",   "group": "增强", "default": False},
    "feature_alist":          {"name": "AList打通",        "group": "增强", "default": False},
    "feature_transfer_quark":   {"name": "夸克转存",    "group": "转存", "default": True},
    "feature_transfer_baidu":   {"name": "百度转存",    "group": "转存", "default": False},
    "feature_transfer_aliyun":  {"name": "阿里转存",    "group": "转存", "default": False},
    "feature_transfer_uc":      {"name": "UC转存",      "group": "转存", "default": False},
    "feature_transfer_xunlei":  {"name": "迅雷转存",    "group": "转存", "default": False},
    "feature_transfer_115":     {"name": "115转存",     "group": "转存", "default": False},
    "feature_transfer_123":     {"name": "123转存",     "group": "转存", "default": False},
    "feature_transfer_cloud189":{"name": "天翼转存",    "group": "转存", "default": False},
}

# ── SQLite ────────────────────────────────────────────────
def get_db():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS flags (
            key   TEXT PRIMARY KEY,
            value INTEGER NOT NULL DEFAULT 0,
            updated_at TEXT NOT NULL DEFAULT (datetime('now'))
        )
    """)
    # 初始化默认值
    for key in FEATURES:
        conn.execute(
            "INSERT OR IGNORE INTO flags(key, value) VALUES(?, ?)",
            (key, int(FEATURES[key]["default"]))
        )
    conn.commit()
    return conn

def read_flags_sqlite() -> Dict[str, bool]:
    conn = get_db()
    rows = conn.execute("SELECT key, value, updated_at FROM flags ORDER BY key").fetchall()
    conn.close()
    return {r["key"]: bool(r["value"]) for r in rows}, {r["key"]: r["updated_at"] for r in rows}

def write_flag_sqlite(key: str, value: bool):
    conn = get_db()
    conn.execute(
        "INSERT INTO flags(key, value, updated_at) VALUES(?, ?, datetime('now')) "
        "ON CONFLICT(key) DO UPDATE SET value=excluded.value, updated_at=datetime('now')",
        (key, int(value))
    )
    conn.commit()
    conn.close()

# ── MySQL 同步 ────────────────────────────────────────────
def sync_to_mysql(key: str, value: bool):
    """将开关状态同步到主应用的 system_configs 表"""
    if not MYSQL_HOST:
        return  # MySQL 未配置，跳过
    try:
        import pymysql
        conn = pymysql.connect(
            host=MYSQL_HOST, port=MYSQL_PORT,
            user=MYSQL_USER, password=MYSQL_PASS,
            database=MYSQL_DB, charset="utf8mb4",
            connect_timeout=5
        )
        with conn.cursor() as cur:
            cur.execute("""
                INSERT INTO system_configs (config_key, config_value, updated_at)
                VALUES (%s, %s, NOW())
                ON DUPLICATE KEY UPDATE config_value=VALUES(config_value), updated_at=NOW()
            """, (key, "true" if value else "false"))
        conn.commit()
        conn.close()
        log.info(f"MySQL 同步成功: {key} = {value}")
    except Exception as e:
        log.warning(f"MySQL 同步失败 ({key}): {e}")

def read_flags_mysql() -> Optional[Dict[str, bool]]:
    if not MYSQL_HOST:
        return None
    try:
        import pymysql
        conn = pymysql.connect(
            host=MYSQL_HOST, port=MYSQL_PORT,
            user=MYSQL_USER, password=MYSQL_PASS,
            database=MYSQL_DB, charset="utf8mb4",
            connect_timeout=5
        )
        with conn.cursor(pymysql.cursors.DictCursor) as cur:
            cur.execute("SELECT config_key, config_value FROM system_configs WHERE config_key LIKE 'feature_%'")
            rows = cur.fetchall()
        conn.close()
        return {r["config_key"]: r["config_value"].lower() == "true" for r in rows}
    except Exception as e:
        log.warning(f"MySQL 读取失败: {e}")
        return None

# ── 路由 ──────────────────────────────────────────────────
@app.route("/")
def index():
    """管理后台首页"""
    return render_template("admin.html", features=FEATURES)

@app.route("/health")
def health():
    return jsonify({"status": "ok", "service": "cloudsearch-admin"})

@app.route("/api/flags", methods=["GET"])
def api_list_flags():
    """列出所有开关"""
    flags_sqlite, updated = read_flags_sqlite()
    flags_mysql = read_flags_mysql()

    result = {}
    for key, meta in FEATURES.items():
        result[key] = {
            "name": meta["name"],
            "group": meta["group"],
            "value": flags_sqlite.get(key, meta["default"]),
            "mysql_value": flags_mysql.get(key) if flags_mysql else None,
            "synced": (flags_mysql is None) or (flags_sqlite.get(key) == flags_mysql.get(key)),
            "updated_at": updated.get(key, ""),
        }
    return jsonify(result)

@app.route("/api/flags/<key>", methods=["PUT"])
def api_set_flag(key):
    """设置单个开关"""
    if key not in FEATURES:
        return jsonify({"error": f"未知开关: {key}"}), 404

    data = request.get_json(force=True)
    value = bool(data.get("value", False))

    # 写本地 SQLite
    write_flag_sqlite(key, value)

    # 同步到 MySQL
    sync_to_mysql(key, value)

    log.info(f"开关切换: {key} = {value}")
    return jsonify({"ok": True, "key": key, "value": value})

@app.route("/api/flags/batch", methods=["PUT"])
def api_batch_set_flags():
    """批量设置开关"""
    data = request.get_json(force=True)
    if not isinstance(data, dict):
        return jsonify({"error": "请求体需为 {key: value} 字典"}), 400

    results = {}
    for key, value in data.items():
        if key not in FEATURES:
            results[key] = {"error": "未知"}
            continue
        val = bool(value)
        write_flag_sqlite(key, val)
        sync_to_mysql(key, val)
        results[key] = val
        log.info(f"批量切换: {key} = {val}")

    return jsonify({"ok": True, "results": results})


if __name__ == "__main__":
    from waitress import serve
    log.info(f"管理后台启动: http://0.0.0.0:{ADMIN_PORT}")
    log.info(f"MySQL 同步: {'已配置' if MYSQL_HOST else '未配置 (仅使用本地 SQLite)'}")
    serve(app, host="0.0.0.0", port=ADMIN_PORT, threads=4)