#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# dependencies = [
# "e2b==2.45.1",
# "httpx==0.28.1",
# "python-dotenv==1.2.3",
# ]
# ///
"""E2B Windows Template/Sandbox/Snapshot 完整生命周期 demo（不绑定 region）。
region / endpoint / 镜像 / VPC 等全部由脚本同目录的 .env 指定，切换 region 只改
配置文件即可，无需改动脚本。
运行方式：
uv run windows_snapshot_lifecycle.py
流程：
1. 使用 Windows 镜像构建 Template
2. 使用步骤 1 输出的 template_id 创建源 Sandbox
3. 使用步骤 2 输出的 sandbox_id 创建 Snapshot
4. 使用步骤 3 输出的 snapshot_id 恢复新 Sandbox
5. 输出恢复 Sandbox 的 noVNC 链接
配置全部读取脚本同目录的 .env，并强制使用 override=True，确保 .env 中的值
覆盖当前进程已有的同名环境变量。恢复 Sandbox 默认保留，方便立即访问 noVNC；
可设置 CLEANUP_RESTORED_SANDBOX=true 在脚本结束时自动清理。
"""
from __future__ import annotations
from e2b.exceptions import SandboxException
from e2b.connection_config import ConnectionConfig
from e2b.api.client_sync import get_api_client
from e2b.api.client.api.snapshots import get_snapshots
from e2b import ApiParams, BuildInfo, Sandbox, Template, default_build_logger
import httpx
import base64
import hashlib
import os
import socket
import struct
import time
from pathlib import Path
from urllib.parse import quote
from dotenv import load_dotenv
# =============================================================================
# 配置加载：使用 override=True，确保 .env 覆盖进程中的同名变量
# =============================================================================
ENV_PATH = Path(os.getenv("ENV_FILE", Path(
__file__).with_name(".env"))).resolve()
if not ENV_PATH.is_file():
raise FileNotFoundError(f".env file not found: {ENV_PATH}")
load_dotenv(dotenv_path=ENV_PATH, override=True)
# =============================================================================
# 通用配置与工具
# =============================================================================
SNAPSHOT_DISCOVERY_TIMEOUT_SECONDS = 90
WINDOWS_SERVICE_PORT = int(os.getenv("WINDOWS_SERVICE_PORT", "5000"))
NOVNC_PORT = int(os.getenv("NOVNC_PORT", "6080"))
WINDOWS_READY_COMMAND = "curl -fsS --connect-timeout 1 --max-time 3 http://localhost:5000/health"
def require_env(name: str) -> str:
value = os.getenv(name)
if not value:
raise RuntimeError(
f"missing required env var: {name} (请在 {ENV_PATH} 中配置)")
return value
def env_bool(name: str, default: bool = False) -> bool:
value = os.getenv(name)
return (
default
if value is None
else value.strip().lower() in {"1", "true", "yes", "on"}
)
def milestone(started_at: float, message: str) -> None:
print(message)
def get_api_params() -> ApiParams:
"""构造 Template、Sandbox 和 Snapshot 共用的控制面参数。"""
headers = {
"X-E2B-Template-Build-Mode": os.getenv("E2B_TEMPLATE_BUILD_MODE", "micro"),
"X-E2B-Template-Alpha-OS-Type": os.getenv(
"E2B_TEMPLATE_OS_TYPE", "windows-osworld"
),
"X-E2B-Template-Alpha-Port": str(WINDOWS_SERVICE_PORT),
"X-E2B-Template-Alpha-Disk-Size": os.getenv("E2B_TEMPLATE_DISK_SIZE", "61440"),
"X-E2B-Template-Alpha-Ready-Command": WINDOWS_READY_COMMAND,
}
optional_headers = {
"X-E2B-Template-Source-VPC-ID": os.getenv("E2B_TEMPLATE_SOURCE_VPC_ID"),
"X-E2B-Template-Source-VSwitch-IDs": os.getenv(
"E2B_TEMPLATE_SOURCE_VSWITCH_IDS"
),
"X-E2B-Template-Source-Security-Group-ID": os.getenv(
"E2B_TEMPLATE_SOURCE_SECURITY_GROUP_ID"
),
}
headers.update(
{key: value for key, value in optional_headers.items() if value})
return ApiParams(
request_timeout=int(os.getenv("E2B_REQUEST_TIMEOUT_SECONDS", "600")),
headers=headers,
api_key=require_env("E2B_API_KEY"),
domain=require_env("E2B_DOMAIN"),
api_url=require_env("E2B_API_URL"),
validate_api_key=False,
)
# =============================================================================
# Step 1：使用 Windows 镜像构建 Template
# =============================================================================
def build_windows_template(started_at: float) -> BuildInfo:
rootfs_image = require_env("ROOTFS_IMAGE")
cpu_count = int(os.getenv("E2B_TEMPLATE_CPU_COUNT", "16"))
memory_mb = int(os.getenv("E2B_TEMPLATE_MEMORY_MB", "16448"))
os_type = os.getenv("E2B_TEMPLATE_OS_TYPE", "windows-osworld")
name_prefix = os.getenv("E2B_TEMPLATE_NAME_PREFIX",
"windows-snapshot-lifecycle")
template_name = f"{name_prefix}-{os_type}-{time.time_ns()}"
milestone(
started_at, f"Step 1/5 building Windows Template from {rootfs_image}")
template = Template().build_in_background(
template=Template()
.from_image(image=rootfs_image),
name=template_name,
cpu_count=cpu_count,
memory_mb=memory_mb,
skip_cache=env_bool("E2B_TEMPLATE_SKIP_CACHE", False),
on_build_logs=default_build_logger(),
**get_api_params(),
)
milestone(
started_at,
f"Template build accepted: template_id={template.template_id}, build_id={template.build_id}",
)
deadline = time.monotonic() + int(
os.getenv("TEMPLATE_BUILD_TIMEOUT_SECONDS", "3600")
)
interval = int(os.getenv("TEMPLATE_BUILD_POLL_INTERVAL_SECONDS", "10"))
while time.monotonic() < deadline:
status = Template.get_build_status(
template, logs_offset=0, **get_api_params())
state = status.status.value
if state == "ready":
milestone(
started_at, f"Step 1/5 complete: template_id={template.template_id}"
)
return template
if state == "error":
raise RuntimeError(f"template build failed: {status}")
if state not in {"building", "waiting"}:
raise RuntimeError(f"unexpected template build status: {status}")
print(
f"template_id={template.template_id} status={state}; waiting {interval}s..."
)
time.sleep(interval)
raise TimeoutError(
f"template build timed out: template_id={template.template_id}")
# =============================================================================
# Step 2 / Step 4：使用 Template ID 或 Snapshot ID 创建 Sandbox
# =============================================================================
def find_recent_sandbox(source_id: str) -> Sandbox | None:
"""创建响应断开后，查找由同一 Template/Snapshot 创建的最近实例。"""
try:
paginator = Sandbox.list(limit=50, **get_api_params())
candidates = []
while paginator.has_next:
for info in paginator.next_items():
if info.template_id == source_id and str(info.state) not in {
"killed",
"dead",
"error",
}:
candidates.append(info)
if not candidates:
return None
latest = max(candidates, key=lambda info: info.started_at)
return Sandbox.connect(
latest.sandbox_id,
timeout=int(os.getenv("SANDBOX_TIMEOUT_SECONDS", "900")),
**get_api_params(),
)
except Exception as exc:
print(f"sandbox recovery lookup failed: {type(exc).__name__}: {exc}")
return None
def create_sandbox(source_id: str, started_at: float, step: int) -> Sandbox:
"""从 template_id 或 snapshot_id 创建 Sandbox，并处理 Windows 创建断连。"""
timeout = int(os.getenv("SANDBOX_TIMEOUT_SECONDS", "900"))
retry_waits = (30, 60, 90, 120)
for attempt in range(1, len(retry_waits) + 2):
try:
sandbox = Sandbox.create(
template=source_id,
timeout=timeout,
**get_api_params(),
)
milestone(
started_at,
f"Step {step}/5 complete: source_id={source_id}, sandbox_id={sandbox.sandbox_id}",
)
return sandbox
except httpx.RemoteProtocolError as exc:
if attempt > len(retry_waits):
raise
wait_seconds = retry_waits[attempt - 1]
print(
f"Sandbox.create response disconnected ({attempt}/5): {exc}; "
f"waiting {wait_seconds}s before recovery"
)
time.sleep(wait_seconds)
recovered = find_recent_sandbox(source_id)
if recovered is not None:
milestone(
started_at,
f"Step {step}/5 recovered: source_id={source_id}, "
f"sandbox_id={recovered.sandbox_id}",
)
return recovered
raise RuntimeError(f"failed to create sandbox from source_id={source_id}")
def kill_sandbox(sandbox: Sandbox | None, label: str) -> None:
if sandbox is None:
return
try:
sandbox.kill()
print(f"cleaned {label}: sandbox_id={sandbox.sandbox_id}")
except Exception as exc:
print(
f"warning: failed to clean {label} sandbox_id={sandbox.sandbox_id}: "
f"{type(exc).__name__}: {exc}"
)
# =============================================================================
# Windows 数据面：等待服务并写入/验证快照标记
# =============================================================================
def execute_windows(sandbox: Sandbox, command: list[str]) -> dict:
access_token = sandbox._envd_access_token
if not access_token:
raise RuntimeError(
f"sandbox has no access token: sandbox_id={sandbox.sandbox_id}"
)
response = httpx.post(
f"http://{sandbox.get_host(WINDOWS_SERVICE_PORT)}/execute",
headers={"X-Access-Token": access_token},
json={"command": command},
timeout=30,
)
if response.is_error:
raise RuntimeError(
f"Windows /execute failed: status={response.status_code}, body={response.text}"
)
result = response.json()
if result.get("returncode") != 0 or result.get("status") != "success":
raise RuntimeError(
f"Windows command failed: command={command}, result={result}"
)
return result
def wait_for_windows_service(
sandbox: Sandbox, timeout: int = 180, interval: int = 5
) -> None:
deadline = time.monotonic() + timeout
last_error: Exception | None = None
while time.monotonic() < deadline:
try:
execute_windows(sandbox, ["whoami"])
return
except Exception as exc:
last_error = exc
time.sleep(interval)
raise TimeoutError(
f"Windows service not ready: sandbox_id={sandbox.sandbox_id}, error={last_error}"
)
# =============================================================================
# Step 3：使用源 sandbox_id 创建 Snapshot
# =============================================================================
def find_snapshot_id(snapshots: list, snapshot_name: str) -> str | None:
expected_suffix = f"/{snapshot_name}:default"
for snapshot in snapshots:
if any(
name == snapshot_name or name.endswith(expected_suffix)
for name in snapshot.names
):
return snapshot.snapshot_id
return None
def lookup_snapshot_id(snapshot_name: str, sandbox_id: str) -> str | None:
client = get_api_client(ConnectionConfig(**get_api_params()))
response = get_snapshots.sync_detailed(
client=client,
sandbox_id=sandbox_id,
name=snapshot_name,
limit=100,
)
if response.status_code == 200 and isinstance(response.parsed, list):
return find_snapshot_id(response.parsed, snapshot_name)
return None
def discover_snapshot_after_disconnect(snapshot_name: str, sandbox_id: str) -> str:
"""create_snapshot 响应断开时查询服务端结果，避免重放非幂等请求。"""
deadline = time.monotonic() + SNAPSHOT_DISCOVERY_TIMEOUT_SECONDS
while time.monotonic() < deadline:
snapshot_id = lookup_snapshot_id(snapshot_name, sandbox_id)
if snapshot_id:
return snapshot_id
time.sleep(2)
raise TimeoutError(
f"snapshot not discoverable after disconnected response: "
f"sandbox_id={sandbox_id}, name={snapshot_name}"
)
def create_snapshot(sandbox: Sandbox, started_at: float) -> str:
wait_for_windows_service(sandbox)
probe_novnc_first_frame(
sandbox,
started_at,
timeout=int(os.getenv("NOVNC_READY_TIMEOUT_SECONDS", "600")),
)
snapshot_name = f"windows-snapshot-{time.time_ns()}"
milestone(
started_at,
f"Step 3/5 creating Snapshot from sandbox_id={sandbox.sandbox_id}",
)
max_attempts = int(os.getenv("SNAPSHOT_CREATE_MAX_ATTEMPTS", "12"))
retry_interval = int(
os.getenv("SNAPSHOT_CREATE_RETRY_INTERVAL_SECONDS", "10"))
snapshot_id = ""
for attempt in range(1, max_attempts + 1):
try:
snapshot_info = sandbox.create_snapshot(
name=snapshot_name, **get_api_params()
)
snapshot_id = snapshot_info.snapshot_id
if not snapshot_id:
raise RuntimeError(
f"snapshot creation returned no ID: {snapshot_info}")
break
except httpx.RemoteProtocolError as exc:
print(f"snapshot response disconnected; discovering result: {exc}")
snapshot_id = discover_snapshot_after_disconnect(
snapshot_name, sandbox.sandbox_id
)
break
except SandboxException as exc:
if "503" not in str(exc) or attempt >= max_attempts:
raise
snapshot_id = lookup_snapshot_id(
snapshot_name, sandbox.sandbox_id) or ""
if snapshot_id:
break
print(
f"snapshot is temporarily unavailable ({attempt}/{max_attempts}); "
f"retrying in {retry_interval}s"
)
time.sleep(retry_interval)
milestone(
started_at,
f"Step 3/5 complete: sandbox_id={sandbox.sandbox_id}, snapshot_id={snapshot_id}",
)
return snapshot_id
# =============================================================================
# Step 5：输出由 snapshot_id 恢复的 Sandbox noVNC 链接
# =============================================================================
class WebSocketBinaryStream:
"""用于 noVNC/RFB 就绪探测的最小 WebSocket 二进制流。"""
def __init__(self, sock: socket.socket, initial_data: bytes = b"") -> None:
self.sock = sock
self.raw_buffer = bytearray(initial_data)
self.data_buffer = bytearray()
def _recv_raw_exact(self, size: int) -> bytes:
while len(self.raw_buffer) < size:
chunk = self.sock.recv(65536)
if not chunk:
raise ConnectionError("WebSocket closed by peer")
self.raw_buffer.extend(chunk)
result = bytes(self.raw_buffer[:size])
del self.raw_buffer[:size]
return result
def send_binary(self, payload: bytes, opcode: int = 0x2) -> None:
mask = os.urandom(4)
length = len(payload)
header = bytearray([0x80 | opcode])
if length < 126:
header.append(0x80 | length)
elif length <= 0xFFFF:
header.append(0x80 | 126)
header.extend(struct.pack("!H", length))
else:
header.append(0x80 | 127)
header.extend(struct.pack("!Q", length))
masked = bytes(value ^ mask[index % 4]
for index, value in enumerate(payload))
self.sock.sendall(bytes(header) + mask + masked)
def _recv_data_frame(self) -> bytes:
while True:
first, second = self._recv_raw_exact(2)
opcode = first & 0x0F
masked = bool(second & 0x80)
length = second & 0x7F
if length == 126:
length = struct.unpack("!H", self._recv_raw_exact(2))[0]
elif length == 127:
length = struct.unpack("!Q", self._recv_raw_exact(8))[0]
mask = self._recv_raw_exact(4) if masked else b""
payload = self._recv_raw_exact(length)
if masked:
payload = bytes(
value ^ mask[index % 4] for index, value in enumerate(payload)
)
if opcode == 0x8:
raise ConnectionError("WebSocket close frame received")
if opcode == 0x9:
self.send_binary(payload, opcode=0xA)
continue
if opcode in {0x0, 0x2}:
return payload
def read_exact(self, size: int) -> bytes:
while len(self.data_buffer) < size:
self.data_buffer.extend(self._recv_data_frame())
result = bytes(self.data_buffer[:size])
del self.data_buffer[:size]
return result
def discard_exact(self, size: int) -> None:
remaining = size
while remaining:
if self.data_buffer:
consumed = min(remaining, len(self.data_buffer))
del self.data_buffer[:consumed]
remaining -= consumed
continue
self.data_buffer.extend(self._recv_data_frame())
def connect_websocket(
host: str, access_token: str, timeout: float
) -> WebSocketBinaryStream:
sock = socket.create_connection((host, 80), timeout=timeout)
sock.settimeout(timeout)
key = base64.b64encode(os.urandom(16)).decode("ascii")
request = (
f"GET /websockify?accessToken={access_token} HTTP/1.1\r\n"
f"Host: {host}\r\n"
"Upgrade: websocket\r\n"
"Connection: Upgrade\r\n"
f"Origin: http://{host}\r\n"
f"Sec-WebSocket-Key: {key}\r\n"
"Sec-WebSocket-Version: 13\r\n"
"Sec-WebSocket-Protocol: binary\r\n\r\n"
)
sock.sendall(request.encode("ascii"))
response = bytearray()
while b"\r\n\r\n" not in response:
chunk = sock.recv(4096)
if not chunk:
raise ConnectionError("connection closed during WebSocket upgrade")
response.extend(chunk)
headers, initial_data = bytes(response).split(b"\r\n\r\n", 1)
status_line = headers.split(b"\r\n", 1)[0]
if b" 101 " not in status_line:
sock.close()
raise ConnectionError(
f"WebSocket upgrade failed: {status_line.decode(errors='replace')}"
)
expected_accept = base64.b64encode(
hashlib.sha1(
(key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest()
)
if expected_accept.lower() not in headers.lower():
sock.close()
raise ConnectionError("invalid Sec-WebSocket-Accept response")
return WebSocketBinaryStream(sock, initial_data)
def probe_novnc_first_frame(
sandbox: Sandbox, started_at: float, timeout: int = 600
) -> None:
"""完成 RFB 握手并读取首帧，确认 Windows 桌面已真正启动。"""
host = f"{NOVNC_PORT}-{sandbox.sandbox_id}.{require_env('E2B_DOMAIN')}"
access_token = sandbox._envd_access_token
if not access_token:
raise RuntimeError(
f"sandbox has no access token: sandbox_id={sandbox.sandbox_id}"
)
deadline = time.monotonic() + timeout
last_error: Exception | None = None
while time.monotonic() < deadline:
stream: WebSocketBinaryStream | None = None
try:
stream = connect_websocket(
host, access_token, min(5, deadline - time.monotonic())
)
version = stream.read_exact(12)
if not version.startswith(b"RFB "):
raise RuntimeError(f"unexpected RFB version: {version!r}")
stream.send_binary(version)
security_type_count = stream.read_exact(1)[0]
if security_type_count == 0:
reason_length = struct.unpack("!I", stream.read_exact(4))[0]
reason = stream.read_exact(
reason_length).decode(errors="replace")
raise RuntimeError(f"RFB server rejected connection: {reason}")
security_types = list(stream.read_exact(security_type_count))
if 1 not in security_types:
raise RuntimeError(
f"unsupported RFB security types: {security_types}")
stream.send_binary(b"\x01")
security_result = struct.unpack("!I", stream.read_exact(4))[0]
if security_result != 0:
raise RuntimeError(
f"RFB security negotiation failed: {security_result}"
)
stream.send_binary(b"\x01")
width, height = struct.unpack("!HH", stream.read_exact(4))
pixel_format = stream.read_exact(16)
bytes_per_pixel = pixel_format[0] // 8
name_length = struct.unpack("!I", stream.read_exact(4))[0]
desktop_name = stream.read_exact(
name_length).decode(errors="replace")
stream.send_binary(struct.pack("!BBHi", 2, 0, 1, 0))
stream.send_binary(struct.pack(
"!BBHHHH", 3, 0, 0, 0, width, height))
while True:
message_type = stream.read_exact(1)[0]
if message_type == 0:
header = stream.read_exact(3)
rectangle_count = struct.unpack("!H", header[1:])[0]
if rectangle_count:
for _ in range(rectangle_count):
rectangle = stream.read_exact(12)
_, _, rect_width, rect_height, encoding = struct.unpack(
"!HHHHi", rectangle
)
if encoding != 0:
raise RuntimeError(
f"unexpected framebuffer encoding: {encoding}"
)
stream.discard_exact(
rect_width * rect_height * bytes_per_pixel
)
milestone(
started_at,
f"noVNC first frame ready: {width}x{height} {desktop_name!r}",
)
return
elif message_type == 1:
header = stream.read_exact(5)
color_count = struct.unpack("!H", header[3:])[0]
stream.read_exact(color_count * 6)
elif message_type == 2:
continue
elif message_type == 3:
header = stream.read_exact(5)
text_length = struct.unpack("!I", header[1:])[0]
stream.read_exact(text_length)
else:
raise RuntimeError(
f"unexpected RFB server message type: {message_type}"
)
except (OSError, ConnectionError, TimeoutError) as exc:
last_error = exc
time.sleep(1)
finally:
if stream is not None:
stream.sock.close()
raise TimeoutError(
f"noVNC first frame not ready: sandbox_id={sandbox.sandbox_id}, "
f"error={last_error}"
)
def get_novnc_url(sandbox: Sandbox) -> str:
access_token = sandbox._envd_access_token
if not access_token:
raise RuntimeError(
f"sandbox has no access token: sandbox_id={sandbox.sandbox_id}"
)
websocket_path = quote(f"websockify?accessToken={access_token}", safe="")
scheme = os.getenv("SANDBOX_PUBLIC_SCHEME", "http")
return (
f"{scheme}://{NOVNC_PORT}-{sandbox.sandbox_id}.{require_env('E2B_DOMAIN')}"
f"/vnc.html#autoconnect=true&resize=scale&path={websocket_path}"
)
def wait_for_novnc(sandbox: Sandbox, timeout: int = 600, interval: int = 3) -> str:
novnc_url = get_novnc_url(sandbox)
deadline = time.monotonic() + timeout
last_error: Exception | None = None
while time.monotonic() < deadline:
try:
response = httpx.get(novnc_url, timeout=5)
if response.status_code == 200 and "noVNC" in response.text:
return novnc_url
last_error = RuntimeError(f"HTTP {response.status_code}")
except (httpx.HTTPError, OSError) as exc:
last_error = exc
time.sleep(interval)
raise TimeoutError(
f"noVNC not ready: sandbox_id={sandbox.sandbox_id}, error={last_error}"
)
# =============================================================================
# 主流程：严格串联 5 个步骤，并按配置清理资源
# =============================================================================
def main() -> None:
started_at = time.perf_counter()
source_sandbox: Sandbox | None = None
restored_sandbox: Sandbox | None = None
workflow_succeeded = False
template_id = ""
snapshot_id = ""
novnc_url = ""
try:
# Step 1: Windows image -> Template
template = build_windows_template(started_at)
template_id = template.template_id
# Step 2: template_id -> source Sandbox
source_sandbox = create_sandbox(template_id, started_at, step=2)
# Step 3: source sandbox_id -> Snapshot
snapshot_id = create_snapshot(source_sandbox, started_at)
# Step 4: snapshot_id -> restored Sandbox
restored_sandbox = create_sandbox(snapshot_id, started_at, step=4)
wait_for_windows_service(restored_sandbox)
identity = str(
execute_windows(restored_sandbox, ["whoami"]).get("output", "")
).strip()
milestone(
started_at,
f"Restored Sandbox is ready: snapshot_id={snapshot_id}, "
f"restored_sandbox_id={restored_sandbox.sandbox_id}, whoami={identity!r}",
)
# Step 5: output restored Sandbox noVNC link
probe_novnc_first_frame(
restored_sandbox,
started_at,
timeout=int(os.getenv("NOVNC_READY_TIMEOUT_SECONDS", "600")),
)
novnc_url = wait_for_novnc(restored_sandbox)
milestone(
started_at,
f"Step 5/5 complete: snapshot_id={snapshot_id}, noVNC is ready",
)
workflow_succeeded = True
finally:
if not env_bool("KEEP_SOURCE_SANDBOX", False):
kill_sandbox(source_sandbox, "source")
cleanup_restored = env_bool("CLEANUP_RESTORED_SANDBOX", False)
if restored_sandbox is not None and (
cleanup_restored or not workflow_succeeded
):
kill_sandbox(restored_sandbox, "restored")
if not workflow_succeeded or restored_sandbox is None:
raise RuntimeError(
"Windows Template/Sandbox/Snapshot lifecycle did not complete"
)
print("\nSUCCESS: Windows Template -> Sandbox -> Snapshot -> Restore -> noVNC\n")
print(f"template_id={template_id}")
print(f"snapshot_id={snapshot_id}")
print(f"restored_sandbox_id={restored_sandbox.sandbox_id}")
print(f"restored_sandbox_novnc_url={novnc_url}")
if not env_bool("CLEANUP_RESTORED_SANDBOX", False):
print(
"restored sandbox is intentionally kept alive; "
"set CLEANUP_RESTORED_SANDBOX=true to clean it automatically"
)
if __name__ == "__main__":
main()
