main.py

import zipfile
import yaml
from pywebio.input import *
from pywebio.output import *
from pywebio.platform import config
from pywebio.platform.tornado import start_server
from pathlib import Path
import shutil
import logging
import os
import io
import re

# 环境变量
APPS_DIR = Path("apps")
DEFAULT_LOGO = Path("default_logo.png")

# 初始化logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 校验key是否为英文字符串
def is_valid_key(key):
    return bool(re.match(r'^[a-zA-Z]+$', key))

# 校验基本信息
def check_base_info(data):
    required_fields = [
        "name", "key", "tags", "shortDescZh", "shortDescEn",
        "type", "crossVersionUpdate", "website", "github", "document"
    ]
    for field in required_fields:
        if not data[field]:
            return (field, f"{field} 不能为空")
    if len(data["shortDescZh"]) > 30:
        return ("shortDescZh", "中文描述不能超过30个字")
    if not is_valid_key(data["key"]):
        return ("key", "key 必须是纯英文字符串")
    return None

# 保存文件
def save_file(path, content, mode='w', encoding=None):
    try:
        if 'b' in mode:  # 二进制模式
            with open(path, mode) as f:
                f.write(content)
        else:  # 文本模式
            with open(path, mode, encoding=encoding or 'utf-8') as f:
                f.write(content)
        logging.info(f"File saved successfully: {path}")
    except IOError as e:
        logging.error(f"Error saving file {path}: {e}")
        raise

# 复制文件
def copy_file(src, dst):
    try:
        shutil.copy(src, dst)
        logging.info(f"File copied successfully from {src} to {dst}")
    except IOError as e:
        logging.error(f"Error copying file from {src} to {dst}: {e}")
        raise

# 创建目录
def create_directory(path):
    try:
        path.mkdir(parents=True, exist_ok=True)
        logging.info(f"Directory created: {path}")
    except OSError as e:
        logging.error(f"Error creating directory {path}: {e}")
        raise

# 创建版本
def create_version(app_dir, existing_versions):
    while True:
        version = input("请输入应用的版本 （不要以v开头）")
        if version in existing_versions:
            put_error(f"版本 {version} 已存在，请输入一个新的版本号")
        else:
            break

    version_dir = app_dir / version
    create_directory(version_dir)

    version_info = input_group("版本信息", [
        textarea("请编写docker-compose.yml", name="docker_compose", code={"mode": "yaml", "theme": ""}),
        textarea("请编写data.yml", name="data", code={"mode": "yaml", "theme": ""}),
    ])



    save_file(version_dir / "data.yml", version_info["data"])
    save_file(version_dir / "docker-compose.yml", version_info["docker_compose"])
    put_success(f"已成功创建版本 {version}")

    return version

# 压缩文件夹
def zip_folder(folder_path, output_path):
    with zipfile.ZipFile(output_path, 'w', zipfile.ZIP_DEFLATED) as zipf:
        for root, _, files in os.walk(folder_path):
            for file in files:
                file_path = os.path.join(root, file)
                arcname = os.path.relpath(file_path, folder_path)
                zipf.write(file_path, arcname)

# 主函数
def main():
    base_info = input_group(
        "自助创建 1Panel 应用",
        [
            input("1. 请输入应用名称* ", name="name", type=TEXT),
            input("2. 请输入应用的key* （仅限英文，用于创建文件夹）", name="key", type=TEXT),
            checkbox("3. 选择应用标签*(可以有多个)", inline=True, options=[
                {"label": "建站", "value": "WebSite"},
                {"label": "Web 服务器", "value": "Server"},
                {"label": "运行环境", "value": "Runtime"},
                {"label": "数据库", "value": "Database"},
                {"label": "工具", "value": "Tool"},
                {"label": "CI/CD", "value": "CI/CD"},
                {"label": "本地", "value": "Local"},
            ], name="tags"),
            input("4. 请输入应用中文描述*（不要超过30个字）", name="shortDescZh", type=TEXT),
            input("5. 请输入应用英文描述*", name="shortDescEn", type=TEXT),
            select("6. 选择应用类型*", options=[
                {"label": "工具类应用，如 phpMyAdmin redis-commander jenkins", "value": "tool"},
                {"label": "支持一键部署的站点类应用类型，如 wordpress halo", "value": "website"},
                {"label": "服务类型的运行时应用，如	mysql openresty redis", "value": "runtime"},
            ], name="type"),
            select("7. 是否可跨大版本升级*", options=[
                {"label": "是", "value": True},
                {"label": "否", "value": False},
            ], name="crossVersionUpdate"),
            slider("8. 应用安装数量限制，(0 代表无限制)*", name="limit", min=0, max=100, step=1, value=0),
            input("9. 官网地址*", name="website", type=URL),
            input("10. Github 地址*", name="github", type=URL),
            input("11. 文档地址*", name="document", type=URL),
            file_upload("上传应用Logo图片(最好是 180 * 180 px)(可选): ", name="logo", accept=[".png", ".jpg", ".jpeg"], max_size="5M"),
        ],
        validate=check_base_info,
    )

    app_dir = APPS_DIR / base_info["key"]
    create_directory(app_dir)

    app_info = {
        "additionalProperties": {
            "key": base_info["key"],
            "name": base_info["name"],
            "tags": base_info["tags"],
            "shortDescZh": base_info["shortDescZh"],
            "shortDescEn": base_info["shortDescEn"],
            "type": base_info["type"],
            "crossVersionUpdate": base_info["crossVersionUpdate"],
            "limit": base_info["limit"],
            "website": base_info["website"],
            "github": base_info["github"],
            "document": base_info["document"],
        }
    }

    save_file(app_dir / "data.yml", yaml.dump(app_info, allow_unicode=True))

    if base_info["logo"]:
        _, file_extension = os.path.splitext(base_info["logo"]["filename"])
        logo_filename = f"logo{file_extension.lower()}"
        save_file(app_dir / logo_filename, base_info["logo"]["content"], mode='wb')
    else:
        copy_file(DEFAULT_LOGO, app_dir / "logo.png")

    put_success("已成功创建基本信息")

    readme = textarea("请编写README", code={"mode": "markdown", "theme": ""})
    save_file(app_dir / "README.md", readme)
    put_success("已成功创建README")

    versions = []
    while True:
        version = create_version(app_dir, versions)
        versions.append(version)
        if not actions("是否继续创建新版本？", [
            {"label": "是", "value": "yes"},
            {"label": "否", "value": "no"},
        ]) == "yes":
            break

    # 压缩应用文件夹
    zip_buffer = io.BytesIO()
    zip_folder(app_dir, zip_buffer)
    zip_buffer.seek(0)

    # 美化下载按钮
    put_button(
        f"下载 {base_info['name']} 应用文件",
        onclick=lambda: put_file(f"{base_info['key']}.zip", zip_buffer.getvalue()),
        color="success",
        outline=True
    )

if __name__ == "__main__":
    config(title="自助创建 1Panel 应用")
    start_server(main, debug=False, port=8080, cdn=False)