kg-agent/scripts/start_service.py

"""Start the frontend and QA API without fetching or updating any data."""
import argparse
import ipaddress
import socket
from pathlib import Path
from step2_graph_building.runtime import release_snapshot
from step4_web.api import run
from step2_graph_building.config import get_service_address


def access_urls(host: str, port: int) -> list[str]:
    """Return browser URLs; a wildcard bind address is never a client URL."""
    if host != '0.0.0.0':
        return [f'http://{host}:{port}/']
    urls = [f'http://127.0.0.1:{port}/']
    try:
        addresses = socket.gethostbyname_ex(socket.gethostname())[2]
    except OSError:
        addresses = []
    for address in sorted(set(addresses)):
        try:
            parsed = ipaddress.ip_address(address)
        except ValueError:
            continue
        if parsed.version == 4 and not (parsed.is_loopback or parsed.is_link_local or parsed.is_unspecified):
            urls.append(f'http://{address}:{port}/')
    return urls


def print_access_urls(host: str, port: int) -> None:
    print(f'服务监听地址: {host}:{port}')
    for index, url in enumerate(access_urls(host, port)):
        label = '本机访问' if index == 0 else '局域网访问'
        print(f'{label}: {url}')
    if host == '0.0.0.0':
        print('提示: 0.0.0.0 仅用于服务监听，不能作为浏览器访问地址。')


def main():
    ap=argparse.ArgumentParser(description=__doc__)
    ap.add_argument('--host', help='监听地址，覆盖 SERVICE_HOST')
    ap.add_argument('--port', type=int, help='监听端口，覆盖 SERVICE_PORT')
    ap.add_argument('--check-config', action='store_true', help='仅检查监听配置，不读取发布数据或启动服务')
    ap.add_argument('--restart', action='store_true', help='优雅停止本项目受管服务后启动')
    args=ap.parse_args()
    try:
        host, port = get_service_address(args.host, args.port)
    except (RuntimeError, ValueError) as exc:
        ap.error(str(exc))
    if args.check_config:
        print(f'监听配置有效: {host}:{port}')
        print_access_urls(host, port)
        return
    current=release_snapshot()
    for field in ('schema_file','display_schema_file','assessment_file'):
        if not Path(current[field]).is_file():
            raise SystemExit('发布文件不完整，请先运行 scripts/update_data.py --mode templates')
    if args.restart:
        from step4_web.service_control import stop_for_restart
        stop_for_restart()
    print_access_urls(host, port)
    run(host=host,port=port)


if __name__=='__main__':
    main()