Files
2026-07-22 13:48:46 +08:00

268 lines
8.3 KiB
Python
Executable File

#!/usr/bin/env python3
"""
CADDesigner API server startup script.
"""
import sys
import os
import argparse
import signal
import shutil
from pathlib import Path
from bootstrap_env import load_project_env
load_project_env()
# Add the project root directory to the Python path.
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
default_working_dir = project_root / "workspace"
def resolve_display_host(host: str) -> str:
"""Convert a bind address to a display address better suited for browser access."""
if host in {"0.0.0.0", "::"}:
return "localhost"
return host
def validate_directory(path_str: str) -> Path:
"""Validate and return a valid directory path."""
path = Path(path_str).resolve()
if not path.exists():
try:
path.mkdir(parents=True, exist_ok=True)
print(f"Created working directory: {path}")
except Exception as e:
raise argparse.ArgumentTypeError(f"Unable to create directory {path}: {e}")
elif not path.is_dir():
raise argparse.ArgumentTypeError(f"Path {path} is not a directory")
return path
def prepare_working_directory(working_dir: Path) -> None:
"""Prepare runtime workspace assets inside the active working directory."""
source_skills = project_root / "workspace" / "skills"
target_skills = working_dir / "skills"
if source_skills.exists() and source_skills.resolve() != target_skills.resolve():
shutil.copytree(source_skills, target_skills, dirs_exist_ok=True)
def check_config():
from SimpleLLMFunc.logger import app_log
"""Check the configuration file."""
config_file = project_root / "config" / "provider.json"
template_file = project_root / "config" / "provider_template.json"
if not config_file.exists():
print("Configuration file does not exist")
if template_file.exists():
try:
import shutil
shutil.copy2(template_file, config_file)
app_log("Created configuration file from template")
print(f"Please edit the configuration file: {config_file}")
print("You can modify the configuration later; continuing service startup now...")
except Exception as e:
print(f"Failed to copy configuration template: {e}")
return False
else:
print("Configuration template file does not exist; please check the project structure")
return False
else:
app_log("Configuration file check passed")
return True
def signal_handler(signum, frame):
"""Signal handler for graceful shutdown."""
print(f"\nReceived signal {signum}; shutting down the API server...")
sys.exit(0)
def main():
"""Main function."""
parser = argparse.ArgumentParser(
description="CADDesigner API server launcher",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Usage examples:
# Start with the default configuration
%(prog)s
# Customize host and port
%(prog)s --host 0.0.0.0 --port 8001
# Enable development mode (auto reload)
%(prog)s --reload
# Adjust the number of worker processes
%(prog)s --workers 4
# Start in a specific working directory
%(prog)s --working-dir /path/to/workspace
# Production environment configuration example
%(prog)s --host 0.0.0.0 --port 8000 \\
--workers 4 --log-level info \\
--working-dir /var/lib/caddesigner
API endpoints:
GET / # Server information
GET /health # Health check
GET /v1/models # List available models
POST /v1/chat/completions # Chat completion (OpenAI compatible)
GET /v1/conversations # Conversation management
GET /docs # Swagger API documentation
GET /redoc # ReDoc API documentation
""",
)
# Basic server parameters.
parser.add_argument(
"--host", default="0.0.0.0", help="API server host address (default: 0.0.0.0)"
)
parser.add_argument(
"--port", type=int, default=8000, help="API server port (default: 8000)"
)
parser.add_argument(
"--working-dir",
type=validate_directory,
default=default_working_dir,
help="Working directory path (default: ./workspace)",
)
# Development and debugging parameters.
parser.add_argument(
"--reload", action="store_true", help="Enable development mode (auto reload after file changes)"
)
parser.add_argument(
"--log-level",
choices=["debug", "info", "warning", "error", "critical"],
default="info",
help="Log level (default: info)",
)
parser.add_argument(
"--access-log",
action="store_true",
default=True,
help="Enable access log (default: enabled)",
)
# Performance parameters.
parser.add_argument(
"--workers",
type=int,
default=1,
help="Number of worker processes (default: 1; forced to 1 in reload mode)",
)
parser.add_argument(
"--loop",
choices=["auto", "asyncio", "uvloop"],
default="auto",
help="Event loop type (default: auto)",
)
# Security and limit parameters.
parser.add_argument("--limit-concurrency", type=int, help="Maximum concurrent connection limit")
parser.add_argument("--limit-max-requests", type=int, help="Maximum requests handled per process")
# Debugging parameters.
parser.add_argument("--debug", action="store_true", help="Enable debug mode")
args = parser.parse_args()
# Set signal handlers.
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
# Prepare and switch to the working directory.
original_dir = os.getcwd()
prepare_working_directory(args.working_dir)
os.chdir(args.working_dir)
try:
import uvicorn
from web_interface.server import app
except ImportError as e:
print(f"Import error: {e}")
print(
"Please ensure all dependencies are installed: uv sync (development environment) or pip install -r requirements.txt"
)
sys.exit(1)
try:
display_host = resolve_display_host(args.host)
print("Starting CADDesigner API server...")
print("=" * 60)
print(f"Server address: http://{display_host}:{args.port}")
print(f"API documentation: http://{display_host}:{args.port}/docs")
print(f"ReDoc documentation: http://{display_host}:{args.port}/redoc")
print(f"Health check: http://{display_host}:{args.port}/health")
print(f"Working directory: {args.working_dir}")
if args.reload:
print("Development mode: enabled (auto reload)")
if args.workers > 1 and not args.reload:
print(f"Worker processes: {args.workers}")
if args.debug:
print("Debug mode: enabled")
print("Press Ctrl+C to stop the service")
print("=" * 60)
# Check configuration.
if not check_config():
print("Configuration check failed; please check the configuration file")
sys.exit(1)
# Prepare uvicorn configuration.
uvicorn_config = {
"app": "web_interface.server:app",
"host": args.host,
"port": args.port,
"reload": args.reload,
"log_level": args.log_level,
"access_log": args.access_log,
"workers": 1
if args.reload
else args.workers, # Only one worker can be used in reload mode.
"loop": args.loop,
}
# Add optional parameters.
if args.limit_concurrency:
uvicorn_config["limit_concurrency"] = args.limit_concurrency
if args.limit_max_requests:
uvicorn_config["limit_max_requests"] = args.limit_max_requests
# Start the server.
uvicorn.run(**uvicorn_config)
except KeyboardInterrupt:
print("\nAPI server stopped")
except Exception as e:
print(f"Server startup failed: {e}")
if args.debug:
import traceback
traceback.print_exc()
sys.exit(1)
finally:
# Restore the original working directory.
os.chdir(original_dir)
if __name__ == "__main__":
main()