76 lines
2.0 KiB
Python
76 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
|
|
def e2e_enabled() -> bool:
|
|
return os.getenv("WSCRAPER_E2E", "").strip() == "1"
|
|
|
|
|
|
def base_env() -> dict[str, str]:
|
|
env = os.environ.copy()
|
|
src_path = str(Path.cwd() / "src")
|
|
current_pythonpath = env.get("PYTHONPATH", "").strip()
|
|
env["PYTHONPATH"] = f"{src_path}{os.pathsep}{current_pythonpath}" if current_pythonpath else src_path
|
|
return env
|
|
|
|
|
|
def log(tr, message: str, kind: str = "info") -> None:
|
|
icon = "•"
|
|
style: dict[str, bool] = {}
|
|
if kind == "ok":
|
|
icon = "✅"
|
|
style = {"green": True}
|
|
elif kind == "err":
|
|
icon = "❌"
|
|
style = {"red": True}
|
|
elif kind == "warn":
|
|
icon = "⚠️"
|
|
style = {"yellow": True}
|
|
elif kind == "run":
|
|
icon = "🚀"
|
|
style = {"cyan": True}
|
|
|
|
if tr is not None:
|
|
tr.write_line(f"{icon} {message}", **style)
|
|
else:
|
|
print(f"{icon} {message}")
|
|
|
|
|
|
def run_cli_live(args: list[str], tr, timeout: int = 900) -> tuple[int, str]:
|
|
cmd = [sys.executable, "-m", "wscraper"] + args
|
|
log(tr, f"Running: {' '.join(cmd)}", kind="run")
|
|
|
|
started = time.time()
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
env=base_env(),
|
|
)
|
|
|
|
output_lines: list[str] = []
|
|
assert proc.stdout is not None
|
|
for line in proc.stdout:
|
|
output_lines.append(line)
|
|
clean = line.rstrip("\n")
|
|
if clean:
|
|
if tr is not None:
|
|
tr.write_line(f" {clean}")
|
|
else:
|
|
print(f" {clean}")
|
|
|
|
return_code = proc.wait(timeout=timeout)
|
|
duration = time.time() - started
|
|
if return_code == 0:
|
|
log(tr, f"Command finished successfully in {duration:.2f}s", kind="ok")
|
|
else:
|
|
log(tr, f"Command failed with exit code {return_code} in {duration:.2f}s", kind="err")
|
|
|
|
return return_code, "".join(output_lines)
|