TappoCloud/store/scripts/publish.py

#!/usr/bin/env python3
"""
TappoCloud Store 应用发布脚本

用法:
    python publish.py <zip文件路径> [--api URL]
    python publish.py cloudfs-v0.1.0.zip
    python publish.py *.zip  # 批量上传
"""

import argparse
import getpass
import json
import sys
from pathlib import Path

try:
    import requests
except ImportError:
    print("请先安装 requests: pip install requests")
    sys.exit(1)

TOKEN_FILE = Path.home() / ".tappocloud_token"


def login(api_base: str) -> str:
    username = input("用户名: ")
    password = getpass.getpass("密码: ")

    resp = requests.post(
        f"{api_base}/api/v1/admin/login",
        json={"username": username, "password": password},
    )

    if resp.status_code != 200:
        print(f"登录失败: {resp.text}")
        sys.exit(1)

    token = resp.json()["token"]
    TOKEN_FILE.write_text(token)
    print("登录成功，token 已保存")
    return token


def get_token(api_base: str) -> str:
    if TOKEN_FILE.exists():
        token = TOKEN_FILE.read_text().strip()
        # 验证 token
        resp = requests.get(
            f"{api_base}/api/v1/apps",
            headers={"Authorization": f"Bearer {token}"},
        )
        if resp.status_code == 200:
            return token
        TOKEN_FILE.unlink()

    return login(api_base)


def upload(api_base: str, token: str, zip_path: Path) -> dict:
    print(f"正在上传: {zip_path.name}")

    with open(zip_path, "rb") as f:
        resp = requests.post(
            f"{api_base}/api/v1/admin/apps/upload",
            headers={"Authorization": f"Bearer {token}"},
            files={"file": (zip_path.name, f, "application/zip")},
        )

    if resp.status_code != 200:
        print(f"上传失败: {resp.text}")
        return None

    return resp.json()


def main():
    parser = argparse.ArgumentParser(description="TappoCloud 应用发布工具")
    parser.add_argument("files", nargs="+", help="要上传的 zip 文件")
    parser.add_argument("--api", default="https://app.tappo.dev", help="API 地址")
    args = parser.parse_args()

    token = get_token(args.api)

    for file_path in args.files:
        path = Path(file_path)
        if not path.exists():
            print(f"文件不存在: {path}")
            continue
        if not path.suffix == ".zip":
            print(f"跳过非 zip 文件: {path}")
            continue

        result = upload(args.api, token, path)
        if result:
            print(f"✓ {result['app_uid']} v{result['version']} 上传成功")
            print(f"  存储地址: {result['storage_url']}")
        print()


if __name__ == "__main__":
    main()