546 lines
21 KiB
Python
546 lines
21 KiB
Python
"""Synthesize one storyboard's narration with the Chatterbox TTS server.
|
|
|
|
Reads ``output/<storyboard>/narration-script.json`` (emitted by
|
|
``dist/preflight.js``) and POSTs one cue at a time to the Chatterbox server's
|
|
native ``/tts`` endpoint. Per-cue WAVs and an index manifest go to
|
|
``output/<storyboard>/audio/`` for the recording step (which reads measured
|
|
cue durations) and the mux step (which drops each WAV at its videoTime).
|
|
|
|
Voice, language and sampling come from the storyboard via the ``voice`` block
|
|
of the narration script.
|
|
|
|
Chatterbox ships two checkpoints and the server holds exactly ONE at a time:
|
|
|
|
* ``chatterbox`` (reported by the server as type ``original``) is English-only.
|
|
* ``multilingual`` covers 23 languages, German and Chinese among them.
|
|
|
|
We pick the checkpoint from ``voice.language`` and hot-swap the server when it
|
|
is on the wrong one. The swap is load-bearing, not cosmetic: the original
|
|
checkpoint happily ACCEPTS ``language: "de"`` and returns HTTP 200, but reads
|
|
the text with an English frontend, so a missed swap yields confident-sounding
|
|
rubbish instead of an error. Nothing downstream would catch that, hence
|
|
``ensure_model``.
|
|
|
|
Unlike the Qwen3-TTS pipeline this replaces, there is no voice-design or
|
|
reference-cloning stage: a Chatterbox predefined voice IS the reference, it
|
|
lives on the server, and it is byte-identical for every cue and every
|
|
storyboard. That deletes the whole minted-reference/referenceHash machinery
|
|
along with the "two voices in one video" class of bug it existed to prevent.
|
|
|
|
Run from the ``video/`` directory (the server must already be reachable;
|
|
render.sh health-checks it first):
|
|
|
|
uv run --project tts python tts/synth.py --storyboard recording
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import io
|
|
import json
|
|
import os
|
|
import statistics
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
import wave
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
DEFAULT_SERVER = "http://host.docker.internal:8004"
|
|
|
|
# repo_id we POST to /save_settings -> the `type` /api/model-info reports back.
|
|
# The two vocabularies differ, so we cannot compare them directly.
|
|
ENGLISH_REPO_ID = "chatterbox"
|
|
MULTILINGUAL_REPO_ID = "multilingual"
|
|
MODEL_TYPE_BY_REPO_ID = {ENGLISH_REPO_ID: "original", MULTILINGUAL_REPO_ID: "multilingual"}
|
|
|
|
# Cues are one or two sentences (the longest across every storyboard is ~105
|
|
# chars), so we synthesize each in a single pass: server-side splitting would
|
|
# only add chunk seams inside a cue for no benefit. If narration ever grows
|
|
# past this, revisit rather than silently degrade.
|
|
SPLIT_TEXT_WARN_CHARS = 300
|
|
|
|
# A hot-swap unloads the current checkpoint, clears VRAM and loads the other
|
|
# one; it took ~22s on the dev box. Generation itself is ~3s per cue.
|
|
SWAP_TIMEOUT_S = 300
|
|
GENERATE_TIMEOUT_S = 300
|
|
INFO_TIMEOUT_S = 30
|
|
|
|
GENERATE_ATTEMPTS = 3
|
|
RETRY_BACKOFF_S = 2.0
|
|
|
|
# Chatterbox sometimes keeps going after it has finished the line: the cue's
|
|
# words all arrive, then several seconds of invented babble follow. It is a
|
|
# property of the seed, not the text (the Chinese outro rambled for 12.9s on
|
|
# seed 49 and read cleanly in 3.1-5.5s on all ten neighbouring seeds), so a
|
|
# re-roll fixes it. Nothing downstream would notice: the runner sizes the
|
|
# cue's wall-clock to the measured audio, so the babble would stretch the cut
|
|
# and play out over the video.
|
|
#
|
|
# Detection is relative, never absolute: speech rate is language-dependent
|
|
# (~13 chars/s of English, ~6 of Chinese), so each storyboard is scored
|
|
# against its OWN median seconds-per-character. A ramble sits ~5x over that
|
|
# median while honest variation stays under ~1.4x, which is a wide gap to put
|
|
# a threshold in. Needs enough cues for the median to mean anything; the
|
|
# median of 4+ is unmoved by a single outlier, and every storyboard has 4+.
|
|
RAMBLE_TOLERANCE = 1.8
|
|
RAMBLE_MIN_CUES = 4
|
|
# Re-roll with CONSECUTIVE seeds (+1, +2, …), never with a round stride.
|
|
# Rambling turned out to track the seed modulo a power of ten rather than the
|
|
# seed itself: on the Chinese outro, seeds ≡ 49 (mod 1000) rambled 7 times out
|
|
# of 8 (49, 1049, 2049, 3049, 4049, 6049, 7049) while the four seeds either
|
|
# side of 49 were clean 4 times out of 4. A +1000 stride therefore re-rolls
|
|
# straight back into the same failure; +1 walks out of it. Colliding with a
|
|
# neighbouring cue's seed is harmless: different text, different audio, and
|
|
# voice identity comes from the predefined voice rather than the seed.
|
|
RAMBLE_RETRY_SEEDS = 4
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class VoiceSettings:
|
|
"""The storyboard's ``voice`` block, resolved against defaults."""
|
|
|
|
voice: str
|
|
language: str
|
|
temperature: float
|
|
exaggeration: float
|
|
cfg_weight: float
|
|
speed_factor: float
|
|
seed: int
|
|
|
|
@property
|
|
def repo_id(self) -> str:
|
|
"""Checkpoint that can actually speak this language."""
|
|
return ENGLISH_REPO_ID if self.language == "en" else MULTILINGUAL_REPO_ID
|
|
|
|
def fingerprint(self) -> dict:
|
|
"""Settings that invalidate cached cue WAVs when any of them changes.
|
|
|
|
Stored at the top of index.json and compared wholesale, so adding a
|
|
field here is enough to make it cache-invalidating.
|
|
"""
|
|
return {
|
|
"voice": self.voice,
|
|
"language": self.language,
|
|
"model": self.repo_id,
|
|
"temperature": self.temperature,
|
|
"exaggeration": self.exaggeration,
|
|
"cfgWeight": self.cfg_weight,
|
|
"speedFactor": self.speed_factor,
|
|
"seed": self.seed,
|
|
}
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--storyboard",
|
|
required=True,
|
|
help="Storyboard slug (matches Storyboard.name in src/storyboard.ts).",
|
|
)
|
|
parser.add_argument(
|
|
"--output-dir",
|
|
type=Path,
|
|
default=Path("output"),
|
|
help="Root output directory; per-storyboard files live in <root>/<storyboard>/.",
|
|
)
|
|
parser.add_argument(
|
|
"--server",
|
|
default=None,
|
|
help=f"Chatterbox TTS server base URL (default {DEFAULT_SERVER}, or $CHATTERBOX_URL).",
|
|
)
|
|
return parser.parse_args()
|
|
|
|
|
|
def _post(url: str, payload: dict, timeout: float) -> bytes:
|
|
body = json.dumps(payload).encode()
|
|
request = urllib.request.Request(
|
|
url, data=body, headers={"Content-Type": "application/json"}
|
|
)
|
|
with urllib.request.urlopen(request, timeout=timeout) as response:
|
|
return response.read()
|
|
|
|
|
|
def _get_json(url: str, timeout: float) -> dict:
|
|
with urllib.request.urlopen(url, timeout=timeout) as response:
|
|
return json.loads(response.read())
|
|
|
|
|
|
def _describe_http_error(error: urllib.error.HTTPError) -> str:
|
|
"""Surface the server's `detail` message instead of a bare status code."""
|
|
try:
|
|
return str(json.loads(error.read()).get("detail", error.reason))
|
|
except (json.JSONDecodeError, OSError, AttributeError):
|
|
return str(error.reason)
|
|
|
|
|
|
def model_type(server: str) -> str:
|
|
info = _get_json(f"{server}/api/model-info", INFO_TIMEOUT_S)
|
|
if not info.get("loaded"):
|
|
raise SystemExit(f"[synth] {server} reports no model loaded")
|
|
return str(info.get("type", ""))
|
|
|
|
|
|
def ensure_model(server: str, repo_id: str) -> None:
|
|
"""Make the server serve `repo_id`, hot-swapping only if it is on the other.
|
|
|
|
Checking first matters: synth.py runs once per storyboard, so a set that
|
|
mixes English and localized cuts would otherwise pay the swap on every
|
|
single one.
|
|
"""
|
|
want_type = MODEL_TYPE_BY_REPO_ID[repo_id]
|
|
current = model_type(server)
|
|
if current == want_type:
|
|
print(f"[synth] server already on the {current} checkpoint", flush=True)
|
|
return
|
|
|
|
print(
|
|
f"[synth] hot-swapping checkpoint {current} -> {want_type} (~20s)",
|
|
flush=True,
|
|
)
|
|
try:
|
|
_post(f"{server}/save_settings", {"model": {"repo_id": repo_id}}, INFO_TIMEOUT_S)
|
|
_post(f"{server}/restart_server", {}, SWAP_TIMEOUT_S)
|
|
except urllib.error.HTTPError as error:
|
|
raise SystemExit(f"[synth] checkpoint swap failed: {_describe_http_error(error)}") from error
|
|
except urllib.error.URLError as error:
|
|
raise SystemExit(f"[synth] checkpoint swap failed: {error.reason}") from error
|
|
|
|
# Trust nothing: a swap that reports success but leaves the old checkpoint
|
|
# resident would synthesize German with the English frontend and no error.
|
|
settled = model_type(server)
|
|
if settled != want_type:
|
|
raise SystemExit(
|
|
f"[synth] server is on the {settled} checkpoint after requesting {want_type}"
|
|
)
|
|
print(f"[synth] checkpoint now {settled}", flush=True)
|
|
|
|
|
|
def generate_cue(server: str, text: str, voice: VoiceSettings, seed: int) -> bytes:
|
|
"""POST one cue to /tts and return the WAV bytes.
|
|
|
|
WAV, never MP3: the server hardcodes a bitrate-less MP3 export that lands
|
|
around 32 kbps, which is audible as phone-quality speech in the final mux.
|
|
"""
|
|
payload = {
|
|
"text": text,
|
|
"voice_mode": "predefined",
|
|
"predefined_voice_id": voice.voice,
|
|
"output_format": "wav",
|
|
"language": voice.language,
|
|
"temperature": voice.temperature,
|
|
"exaggeration": voice.exaggeration,
|
|
"cfg_weight": voice.cfg_weight,
|
|
"speed_factor": voice.speed_factor,
|
|
"seed": seed,
|
|
"split_text": False,
|
|
}
|
|
last_error: Exception | None = None
|
|
for attempt in range(1, GENERATE_ATTEMPTS + 1):
|
|
try:
|
|
return _post(f"{server}/tts", payload, GENERATE_TIMEOUT_S)
|
|
except urllib.error.HTTPError as error:
|
|
# 4xx is our payload being wrong (bad voice name, bad language);
|
|
# retrying cannot fix it, so fail loudly on the first one.
|
|
detail = _describe_http_error(error)
|
|
if error.code < 500:
|
|
raise SystemExit(f"[synth] /tts rejected cue: {detail}") from error
|
|
last_error = error
|
|
except (urllib.error.URLError, TimeoutError, OSError) as error:
|
|
last_error = error
|
|
if attempt < GENERATE_ATTEMPTS:
|
|
print(
|
|
f"[synth] /tts attempt {attempt}/{GENERATE_ATTEMPTS} failed ({last_error}); retrying",
|
|
flush=True,
|
|
)
|
|
time.sleep(RETRY_BACKOFF_S * attempt)
|
|
raise SystemExit(f"[synth] /tts failed after {GENERATE_ATTEMPTS} attempts: {last_error}")
|
|
|
|
|
|
def wav_stats(data: bytes) -> tuple[int, int]:
|
|
"""(sampleRate, durationMs) of a WAV, read straight from its header."""
|
|
with wave.open(io.BytesIO(data), "rb") as handle:
|
|
frame_rate = handle.getframerate()
|
|
frames = handle.getnframes()
|
|
if not frame_rate or not frames:
|
|
raise SystemExit("[synth] server returned a WAV with no audio")
|
|
return frame_rate, int(round(frames * 1000 / frame_rate))
|
|
|
|
|
|
def secs_per_char(text: str, duration_ms: int) -> float:
|
|
"""Speech rate, inverted so a rambling cue scores HIGH. See RAMBLE_TOLERANCE."""
|
|
return (duration_ms / 1000) / max(len(text.strip()), 1)
|
|
|
|
|
|
def ramble_baseline(items: list[dict]) -> float | None:
|
|
"""The storyboard's own median seconds-per-char, or None if uncalibratable."""
|
|
if len(items) < RAMBLE_MIN_CUES:
|
|
return None
|
|
baseline = statistics.median(secs_per_char(it["text"], it["durationMs"]) for it in items)
|
|
return baseline if baseline > 0 else None
|
|
|
|
|
|
def rambling_cues(items: list[dict]) -> list[dict]:
|
|
"""Cues running far enough past the storyboard's pace to look hallucinated."""
|
|
baseline = ramble_baseline(items)
|
|
if baseline is None:
|
|
return []
|
|
return [
|
|
it
|
|
for it in items
|
|
if secs_per_char(it["text"], it["durationMs"]) > baseline * RAMBLE_TOLERANCE
|
|
]
|
|
|
|
|
|
def repair_rambling_cues(
|
|
server: str,
|
|
audio_dir: Path,
|
|
voice: VoiceSettings,
|
|
items: list[dict],
|
|
) -> int:
|
|
"""Re-roll the seed on any cue that runs far past the storyboard's pace.
|
|
|
|
Mutates `items` in place and returns how many cues were replaced. Applies
|
|
to reused cues as well as fresh ones, so a storyboard whose cache predates
|
|
this guard heals itself instead of keeping the bad take forever.
|
|
"""
|
|
baseline = ramble_baseline(items)
|
|
if baseline is None:
|
|
return 0
|
|
|
|
repaired = 0
|
|
for item in rambling_cues(items):
|
|
rate = secs_per_char(item["text"], item["durationMs"])
|
|
print(
|
|
f"[synth] cue {item['cueIndex']} runs {rate / baseline:.1f}x the storyboard's pace "
|
|
f"({item['durationMs']}ms); re-rolling the seed «{item['text']}»",
|
|
flush=True,
|
|
)
|
|
|
|
# Score candidates by distance from the storyboard's own pace: that
|
|
# rejects a truncated take as firmly as a rambling one, where
|
|
# "shortest wins" would happily keep a cue with its ending cut off.
|
|
best = None
|
|
best_error = abs(rate - baseline)
|
|
for attempt in range(1, RAMBLE_RETRY_SEEDS + 1):
|
|
seed = voice.seed + int(item["cueIndex"]) + attempt
|
|
data = generate_cue(server, item["text"].strip(), voice, seed)
|
|
sample_rate, duration_ms = wav_stats(data)
|
|
candidate_rate = secs_per_char(item["text"], duration_ms)
|
|
error = abs(candidate_rate - baseline)
|
|
if error < best_error:
|
|
best, best_error = (data, sample_rate, duration_ms, seed), error
|
|
if candidate_rate <= baseline * RAMBLE_TOLERANCE:
|
|
break
|
|
|
|
if best is None:
|
|
print(
|
|
f"[synth] cue {item['cueIndex']}: no re-roll beat the original; keeping it",
|
|
flush=True,
|
|
)
|
|
continue
|
|
data, sample_rate, duration_ms, seed = best
|
|
(audio_dir / item["wav"]).write_bytes(data)
|
|
item["sampleRate"] = sample_rate
|
|
item["durationMs"] = duration_ms
|
|
item["seed"] = seed
|
|
repaired += 1
|
|
settled = secs_per_char(item["text"], duration_ms) / baseline
|
|
note = "" if settled <= RAMBLE_TOLERANCE else " STILL LONG: listen before publishing"
|
|
print(
|
|
f"[synth] cue {item['cueIndex']}: {duration_ms}ms on seed {seed} "
|
|
f"({settled:.1f}x pace){note}",
|
|
flush=True,
|
|
)
|
|
return repaired
|
|
|
|
|
|
def load_reusable_items(index_path: Path, cues: list[dict], voice: VoiceSettings) -> dict[int, dict]:
|
|
"""Cue-indexed cached items that still match the script and the settings.
|
|
|
|
Partial by design, so an interrupted run resumes cue-by-cue instead of
|
|
re-synthesizing the whole storyboard.
|
|
"""
|
|
if not index_path.exists():
|
|
return {}
|
|
try:
|
|
cached = json.loads(index_path.read_text())
|
|
except json.JSONDecodeError:
|
|
return {}
|
|
if {k: cached.get(k) for k in voice.fingerprint()} != voice.fingerprint():
|
|
return {}
|
|
|
|
cue_by_index = {int(c["cueIndex"]): c for c in cues}
|
|
reusable: dict[int, dict] = {}
|
|
for item in cached.get("items", []):
|
|
cue_index = int(item.get("cueIndex", -1))
|
|
cue = cue_by_index.get(cue_index)
|
|
wav = item.get("wav")
|
|
if cue is None or not wav or not (index_path.parent / wav).exists():
|
|
continue
|
|
if cue["text"].strip() != str(item.get("text", "")).strip():
|
|
continue
|
|
if int(cue.get("gapBeforeMs", 0)) != int(item.get("gapBeforeMs", -1)):
|
|
continue
|
|
reusable[cue_index] = item
|
|
return reusable
|
|
|
|
|
|
def resolve_voice(script_path: Path, block: dict) -> VoiceSettings:
|
|
voice = str(block.get("voice", "")).strip()
|
|
language = str(block.get("language", "")).strip()
|
|
if not voice or not language:
|
|
raise SystemExit(
|
|
f"[synth] {script_path} voice block needs `voice` and `language`. Re-run preflight."
|
|
)
|
|
return VoiceSettings(
|
|
voice=voice,
|
|
language=language,
|
|
temperature=float(block.get("temperature", 0.8)),
|
|
exaggeration=float(block.get("exaggeration", 0.5)),
|
|
cfg_weight=float(block.get("cfgWeight", 0.5)),
|
|
speed_factor=float(block.get("speedFactor", 1.0)),
|
|
seed=int(block.get("seed", 42)),
|
|
)
|
|
|
|
|
|
def check_voice_exists(server: str, voice: str) -> None:
|
|
"""Fail before generating rather than after a storyboard of wrong audio.
|
|
|
|
/tts does reject an unknown predefined voice, but this names the mistake
|
|
and lists the alternatives instead of surfacing a server stack trace.
|
|
"""
|
|
try:
|
|
voices = _get_json(f"{server}/v1/audio/voices", INFO_TIMEOUT_S).get("voices", [])
|
|
except (urllib.error.URLError, OSError, json.JSONDecodeError):
|
|
return # Non-fatal: /tts still validates the name.
|
|
if voices and voice not in voices:
|
|
raise SystemExit(
|
|
f"[synth] unknown predefined voice {voice!r}. Available: {', '.join(sorted(voices))}"
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
args = parse_args()
|
|
server = (args.server or os.environ.get("CHATTERBOX_URL") or DEFAULT_SERVER).rstrip("/")
|
|
|
|
storyboard_dir = args.output_dir / args.storyboard
|
|
script_path = storyboard_dir / "narration-script.json"
|
|
audio_dir = storyboard_dir / "audio"
|
|
|
|
if not script_path.exists():
|
|
print(f"[synth] script not found: {script_path}", file=sys.stderr)
|
|
return 1
|
|
|
|
script = json.loads(script_path.read_text())
|
|
cues = [c for c in script.get("items", []) if c.get("text", "").strip()]
|
|
if not cues:
|
|
print("[synth] script has no cues; nothing to generate.", file=sys.stderr)
|
|
return 1
|
|
|
|
block = script.get("voice")
|
|
if not block:
|
|
print(f"[synth] {script_path} has no `voice` block. Re-run preflight.", file=sys.stderr)
|
|
return 1
|
|
voice = resolve_voice(script_path, block)
|
|
|
|
audio_dir.mkdir(parents=True, exist_ok=True)
|
|
index_path = audio_dir / "index.json"
|
|
|
|
print(f"[synth] [{args.storyboard}] server: {server}", flush=True)
|
|
print(
|
|
f"[synth] [{args.storyboard}] voice={voice.voice} language={voice.language} "
|
|
f"model={voice.repo_id}",
|
|
flush=True,
|
|
)
|
|
print(
|
|
f"[synth] [{args.storyboard}] temperature={voice.temperature} "
|
|
f"exaggeration={voice.exaggeration} cfg_weight={voice.cfg_weight} "
|
|
f"speed_factor={voice.speed_factor} seed={voice.seed}",
|
|
flush=True,
|
|
)
|
|
|
|
reusable = load_reusable_items(index_path, cues, voice)
|
|
# A fully cached storyboard still has to be re-checked for a rambling cue:
|
|
# durations live in the index, so spotting one costs no server call, and
|
|
# falling through to the generation path is what re-rolls its seed.
|
|
stale = rambling_cues(list(reusable.values()))
|
|
if len(reusable) == len(cues) and not stale:
|
|
print(
|
|
f"[synth] [{args.storyboard}] cached audio matches the current script: "
|
|
"skipping generation",
|
|
flush=True,
|
|
)
|
|
return 0
|
|
|
|
# Only touch the server once we know there is something to synthesize:
|
|
# a fully cached storyboard must not trigger a 20s checkpoint swap.
|
|
check_voice_exists(server, voice.voice)
|
|
ensure_model(server, voice.repo_id)
|
|
|
|
out_index_base = {"storyboard": args.storyboard, **voice.fingerprint()}
|
|
|
|
def write_index(items: list[dict]) -> None:
|
|
index_path.write_text(json.dumps({**out_index_base, "items": items}, indent=2))
|
|
|
|
items: list[dict] = []
|
|
for cue in cues:
|
|
cue_index = int(cue["cueIndex"])
|
|
cached_item = reusable.get(cue_index)
|
|
if cached_item:
|
|
items.append(cached_item)
|
|
write_index(items)
|
|
print(
|
|
f"[synth] reusing {cached_item['wav']} "
|
|
f"{int(cached_item['durationMs']):>5d}ms «{cue['text']}»",
|
|
flush=True,
|
|
)
|
|
continue
|
|
|
|
text = cue["text"].strip()
|
|
if len(text) > SPLIT_TEXT_WARN_CHARS:
|
|
print(
|
|
f"[synth] cue {cue_index} is {len(text)} chars; Chatterbox quality drops on "
|
|
"long single-pass text. Split the cue in the storyboard.",
|
|
flush=True,
|
|
)
|
|
|
|
# Vary the seed per cue: one seed for every cue in a storyboard makes
|
|
# consecutive lines land on near-identical intonation contours, which
|
|
# reads as robotic across a whole video. The voice identity is pinned
|
|
# by the predefined voice, not the seed, so this costs no consistency.
|
|
seed = voice.seed + cue_index
|
|
wav_bytes = generate_cue(server, text, voice, seed)
|
|
sample_rate, duration_ms = wav_stats(wav_bytes)
|
|
wav_name = f"cue_{cue_index:03d}.wav"
|
|
(audio_dir / wav_name).write_bytes(wav_bytes)
|
|
items.append(
|
|
{
|
|
"cueIndex": cue_index,
|
|
"text": cue["text"],
|
|
"gapBeforeMs": int(cue.get("gapBeforeMs", 0)),
|
|
"wav": wav_name,
|
|
"sampleRate": sample_rate,
|
|
"durationMs": duration_ms,
|
|
"seed": seed,
|
|
}
|
|
)
|
|
write_index(items)
|
|
print(f"[synth] wrote {wav_name} {duration_ms:>5d}ms «{cue['text']}»", flush=True)
|
|
|
|
repair_rambling_cues(server, audio_dir, voice, items)
|
|
|
|
write_index(items)
|
|
total_ms = sum(it["gapBeforeMs"] + it["durationMs"] for it in items)
|
|
print(
|
|
f"[synth] [{args.storyboard}] {len(items)} cues, {total_ms}ms of audio "
|
|
f"(incl. gaps) -> {audio_dir}",
|
|
flush=True,
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|