ssh_panel/appstore-server/app.py

"""
应用商店 HTTP API 服务
从 MySQL 读取应用模板数据，返回 JSON 给 Electron 客户端
所有数据库连接参数通过环境变量配置
"""

import os
import json
from flask import Flask, jsonify, request
from flask_cors import CORS
import mysql.connector

app = Flask(__name__)

# 允许跨域（Electron 渲染进程直接 fetch 时需要）
CORS(app)

# ─── 数据库配置（环境变量） ───
DB_HOST = os.environ.get('DB_HOST', '127.0.0.1')
DB_PORT = int(os.environ.get('DB_PORT', '3306'))
DB_USER = os.environ.get('DB_USER', 'root')
DB_PASSWORD = os.environ.get('DB_PASSWORD', '123456')
DB_NAME = os.environ.get('DB_NAME', 'appstore')

# API 监听端口
API_PORT = int(os.environ.get('API_PORT', '3001'))

# API 鉴权 token（可选，客户端通过 Authorization header 传递）
API_TOKEN = os.environ.get('API_TOKEN', '')


def get_db():
    """获取数据库连接"""
    return mysql.connector.connect(
        host=DB_HOST,
        port=DB_PORT,
        user=DB_USER,
        password=DB_PASSWORD,
        database=DB_NAME,
        charset='utf8mb4'
    )


def check_token():
    """校验 API token，未配置 token 时跳过"""
    if not API_TOKEN:
        return True
    auth = request.headers.get('Authorization', '')
    return auth == f'Bearer {API_TOKEN}'


def fetch_template_list():
    """查询所有模板（列表摘要）"""
    conn = get_db()
    cursor = conn.cursor(dictionary=True)
    cursor.execute(
        'SELECT id, app_key, name, description, icon, category, '
        'image, container_name, sort_order, is_active '
        'FROM app_templates WHERE is_active = 1 ORDER BY sort_order, id'
    )
    rows = cursor.fetchall()
    cursor.close()
    conn.close()
    return rows


def fetch_template_detail(app_key):
    """查询单个模板的完整详情（含子表）"""
    conn = get_db()
    cursor = conn.cursor(dictionary=True)

    # 主表
    cursor.execute('SELECT * FROM app_templates WHERE app_key = %s AND is_active = 1', (app_key,))
    tpl = cursor.fetchone()
    if not tpl:
        cursor.close()
        conn.close()
        return None

    # 子表批量查询
    sub_tables = ['app_ports', 'app_volumes', 'app_envs', 'app_mkdirs', 'app_write_files', 'app_params']
    for table in sub_tables:
        cursor.execute(f'SELECT * FROM {table} WHERE template_id = %s ORDER BY sort_order', (tpl['id'],))
        rows = cursor.fetchall()
        # 序列化：去掉 template_id 和 id，JSON 友好
        cleaned = []
        for r in rows:
            item = {k: v for k, v in r.items() if k not in ('id', 'template_id')}
            cleaned.append(item)
        # 子表名映射到 deploy 对应字段
        field_map = {
            'app_ports': 'ports',
            'app_volumes': 'volumes',
            'app_envs': 'env',
            'app_mkdirs': 'mkdirs',
            'app_write_files': 'writeFiles',
            'app_params': 'params'
        }
        tpl[field_map[table]] = cleaned

    # 删除不需要的主表字段
    tpl.pop('id', None)
    tpl.pop('is_active', None)
    tpl.pop('sort_order', None)
    tpl.pop('created_at', None)
    tpl.pop('updated_at', None)

    cursor.close()
    conn.close()
    return tpl


def fetch_all_templates_full():
    """一次性查询所有模板的完整详情"""
    conn = get_db()
    cursor = conn.cursor(dictionary=True)

    # 所有主表记录
    cursor.execute(
        'SELECT * FROM app_templates WHERE is_active = 1 ORDER BY sort_order, id'
    )
    templates = cursor.fetchall()

    sub_tables = ['app_ports', 'app_volumes', 'app_envs', 'app_mkdirs', 'app_write_files', 'app_params']
    field_map = {
        'app_ports': 'ports',
        'app_volumes': 'volumes',
        'app_envs': 'env',
        'app_mkdirs': 'mkdirs',
        'app_write_files': 'writeFiles',
        'app_params': 'params'
    }

    for tpl in templates:
        tpl_id = tpl['id']
        for table in sub_tables:
            cursor.execute(
                f'SELECT * FROM {table} WHERE template_id = %s ORDER BY sort_order',
                (tpl_id,)
            )
            rows = cursor.fetchall()
            cleaned = []
            for r in rows:
                item = {k: v for k, v in r.items() if k not in ('id', 'template_id')}
                cleaned.append(item)
            tpl[field_map[table]] = cleaned

        # 清理主表多余字段
        tpl.pop('id', None)
        tpl.pop('is_active', None)
        tpl.pop('sort_order', None)
        tpl.pop('created_at', None)
        tpl.pop('updated_at', None)

    cursor.close()
    conn.close()
    return templates


# ─── API 路由 ───

@app.route('/api/apps', methods=['GET'])
def list_apps():
    """获取所有应用模板列表（完整详情）"""
    if not check_token():
        return jsonify({'ok': False, 'error': '未授权'}), 401

    try:
        templates = fetch_all_templates_full()
        return jsonify({'ok': True, 'data': templates, 'count': len(templates)})
    except Exception as e:
        return jsonify({'ok': False, 'error': str(e)}), 500


@app.route('/api/apps/<app_key>', methods=['GET'])
def get_app(app_key):
    """获取单个应用模板详情"""
    if not check_token():
        return jsonify({'ok': False, 'error': '未授权'}), 401

    try:
        tpl = fetch_template_detail(app_key)
        if not tpl:
            return jsonify({'ok': False, 'error': '模板不存在'}), 404
        return jsonify({'ok': True, 'data': tpl})
    except Exception as e:
        return jsonify({'ok': False, 'error': str(e)}), 500


@app.route('/api/health', methods=['GET'])
def health():
    """健康检查"""
    return jsonify({'ok': True, 'status': 'running'})


@app.route('/api/version', methods=['GET'])
def get_version():
    """获取客户端最新版本信息"""
    if not check_token():
        return jsonify({'ok': False, 'error': '未授权'}), 401

    try:
        conn = get_db()
        cursor = conn.cursor(dictionary=True)
        cursor.execute(
            'SELECT version, download_url, release_notes, is_latest, '
            'created_at FROM app_version WHERE is_latest = 1 ORDER BY id DESC LIMIT 1'
        )
        row = cursor.fetchone()
        cursor.close()
        conn.close()

        if not row:
            return jsonify({'ok': False, 'error': '暂无版本信息'}), 404

        return jsonify({
            'ok': True,
            'data': {
                'version': row['version'],
                'downloadUrl': row['download_url'],
                'releaseNotes': row['release_notes'] or '',
                'isLatest': bool(row['is_latest']),
                'createdAt': row['created_at'].isoformat() if row['created_at'] else None,
            }
        })
    except Exception as e:
        return jsonify({'ok': False, 'error': str(e)}), 500


if __name__ == '__main__':
    print(f'[AppStore API] DB: {DB_HOST}:{DB_PORT}/{DB_NAME}')
    print(f'[AppStore API] Listen on :{API_PORT}')
    print(f'[AppStore API] Token auth: {"on" if API_TOKEN else "off"}')
    app.run(host='0.0.0.0', port=API_PORT, debug=False)