mirror of
https://github.com/gradio-app/gradio.git
synced 2025-01-06 10:25:17 +08:00
38b3682c3a
* wip * add changeset * sha256 * add changeset * hash * tunneling * add changeset * validate ip address * format * vendor in hash func --------- Co-authored-by: gradio-pr-bot <gradio-pr-bot@users.noreply.github.com>
68 lines
2.1 KiB
Python
68 lines
2.1 KiB
Python
"""
|
|
Defines helper methods useful for setting up ports, launching servers, and
|
|
creating tunnels.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ipaddress
|
|
import os
|
|
import time
|
|
import warnings
|
|
|
|
import httpx
|
|
|
|
from gradio.routes import App # HACK: to avoid circular import # noqa: F401
|
|
from gradio.tunneling import Tunnel
|
|
|
|
GRADIO_API_SERVER = "https://api.gradio.app/v2/tunnel-request"
|
|
GRADIO_SHARE_SERVER_ADDRESS = os.getenv("GRADIO_SHARE_SERVER_ADDRESS")
|
|
|
|
|
|
def setup_tunnel(
|
|
local_host: str, local_port: int, share_token: str, share_server_address: str | None
|
|
) -> str:
|
|
share_server_address = (
|
|
GRADIO_SHARE_SERVER_ADDRESS
|
|
if share_server_address is None
|
|
else share_server_address
|
|
)
|
|
if share_server_address is None:
|
|
try:
|
|
response = httpx.get(GRADIO_API_SERVER, timeout=30)
|
|
payload = response.json()[0]
|
|
remote_host, remote_port = payload["host"], int(payload["port"])
|
|
except Exception as e:
|
|
raise RuntimeError(
|
|
"Could not get share link from Gradio API Server."
|
|
) from e
|
|
try:
|
|
ipaddress.ip_address(remote_host)
|
|
except ValueError as e:
|
|
raise ValueError(
|
|
f"Invalid IP address received from Gradio API Server: {remote_host}"
|
|
) from e
|
|
else:
|
|
remote_host, remote_port = share_server_address.split(":")
|
|
remote_port = int(remote_port)
|
|
try:
|
|
tunnel = Tunnel(remote_host, remote_port, local_host, local_port, share_token)
|
|
address = tunnel.start_tunnel()
|
|
return address
|
|
except Exception as e:
|
|
raise RuntimeError(str(e)) from e
|
|
|
|
|
|
def url_ok(url: str) -> bool:
|
|
try:
|
|
for _ in range(5):
|
|
with warnings.catch_warnings():
|
|
warnings.filterwarnings("ignore")
|
|
r = httpx.head(url, timeout=3, verify=False)
|
|
if r.status_code in (200, 401, 302): # 401 or 302 if auth is set
|
|
return True
|
|
time.sleep(0.500)
|
|
except (ConnectionError, httpx.ConnectError, httpx.TimeoutException):
|
|
return False
|
|
return False
|