Function builtin

stream/main.py:88–129

(
    queue: Optional[Any] = None, port: Optional[int] = None, proxy: Optional[str] = None
)

async def builtin(
    queue: Optional[Any] = None, port: Optional[int] = None, proxy: Optional[str] = None
) -> None:
    # Set up logging with GridFormatter for consistent output
    set_source("PROXY")

    console_handler = logging.StreamHandler(sys.stderr)
    console_handler.setFormatter(GridFormatter(show_tree=True, colorize=True))
    console_handler.setLevel(logging.INFO)

    # Configure proxy_server logger specifically (not root)
    logger = logging.getLogger("proxy_server")
    logger.handlers.clear()  # Remove any existing handlers
    logger.addHandler(console_handler)
    logger.setLevel(logging.INFO)
    logger.propagate = False  # Prevent double logging

    # Create certs directory
    cert_dir = Path("certs")
    cert_dir.mkdir(exist_ok=True)

    if port is None:
        port = 3120

    # Create and start the proxy server
    proxy_server = ProxyServer(
        host="127.0.0.1",
        port=port,
        intercept_domains=["*.google.com"],
        upstream_proxy=proxy,
        queue=queue,
    )

    try:
        await proxy_server.start()
    except KeyboardInterrupt:
        logger.info("Shutting down proxy server")
    except asyncio.CancelledError:
        raise
    except Exception as e:
        logger.error(f"Error starting proxy server: {e}", exc_info=True)
        sys.exit(1)


if __name__ == "__main__":