CreatorBox/cli.py at master · xiesx123/CreatorBox · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
import os
import sys

sys.path.insert(0, os.getcwd())

import traceback

import click


@click.group()
def cli():
    """
    CreatorBox CLI

    🚀🎬 Flexible, efficient, and scalable toolbox for editing and dubbing, unleashing creative potential.

    https://github.com/xiesx123/CreatorBox
    """


@cli.command(help="启动服务 (Start service)")
@click.option("--host", "-h", type=str, default="0.0.0.0", show_default=True, required=True, help="host")
@click.option("--port", "-p", type=int, default=8000, show_default=True, required=True, help="port")
@click.option("--debug", is_flag=True, default=False, show_default=True, help="enable debug mode")
@click.option("--browser", is_flag=True, default=False, show_default=True, help="auto open browser")
@click.option("--ngrok", is_flag=True, default=False, show_default=True, help="enable ngrok tunnel")
@click.option("--ngrok_host", "-nh", type=str, default="toucan-real-informally.ngrok-free.app", show_default=True, help="ngrok host")
@click.option("--ngrok_port", "-np", type=int, default=80, show_default=True, help="ngrok port")
def start(host, port, debug, browser, ngrok, ngrok_host, ngrok_port):
    # spawn
    import multiprocessing as mp

    mp.set_start_method("spawn", force=True)

    # args
    os.environ["browser"] = str(browser)
    if "REBOOT_ARGS" not in os.environ:
        import json

        os.environ["REBOOT_ARGS"] = json.dumps(sys.argv)

    # start ngrok
    def start_ngrok(token, hostname, port):
        from pyngrok import ngrok as ng

        ng.set_auth_token(token)
        if hostname:
            public_url = ng.connect(addr=port, hostname=hostname)
        else:
            public_url = ng.connect(addr=port)
        click.echo(f"✅ Ngrok tunnel started: {public_url}")

    # start uvicorn
    def start_uvicorn(host, port, debug):
        import uvicorn

        click.echo(f"🚀 Starting service... http://{host}:{port}")
        uvicorn.run("src.main:asgi", host=host, port=port, reload=debug)

    try:
        if ngrok:
            ngrok_token = os.environ.get("NGROK_AUTH_TOKEN")
            if not ngrok_token:
                click.echo("❌ Ngrok mode requires setting the environment variable NGROK_AUTH_TOKEN")
                return
            start_ngrok(ngrok_token, ngrok_host, ngrok_port or port)

        start_uvicorn(host, port, debug)
    except Exception as e:
        click.echo(f"❌ error: {str(e)}", err=True)
        traceback.print_exc()


@cli.command(help="启动扩展 (Start Extensions)")
@click.option("--name", "-n", default=None, required=True, help="Name of the extension to install")
@click.option("--version", "-v", default="3.11", help="Specify the Python version")
@click.option("--port", "-p", type=int, default=None, show_default=True, required=True, help="The port number to use during the installation process.")
@click.option("--start", is_flag=True, default=False, show_default=True, help="Start the extension immediately after installation")
@click.option("--share", is_flag=True, default=False, show_default=True, help="Share the extension publicly after installation")
def install(name, version, start, port, share):
    try:
        from src.support.launcher.cli import Launcher

        kwargs = {"pyv": version, "share": share}
        if port is not None:
            kwargs["port"] = port

        installer = Launcher.builder(name, **kwargs)
        installer.create()
        click.echo(f"📦 Installing extension '{name}'...")
        installer.install()
        click.echo("✅ Installation complete.")
        if start:
            click.echo(f"🚀 Starting extension '{name}'...")
            installer.start()
    except Exception as e:
        click.echo(f"❌ error: {str(e)}", err=True)
        traceback.print_exc()


if __name__ == "__main__":
    cli()