Files
docker-images/linkki-tiedotus/linkki_poster/progress.py
T
ajp_anton 8ec3cde4f2
Build custom container images / build (map[base_image:php:8-fpm-alpine build_args:PHP_VERSION=8 context:php8-pgsql fingerprint_command:{ apk info -v | LC_ALL=C sort; find /usr/local/lib/php/extensions /usr/local/etc/php/conf.d -type f -exec sha256sum {} + | LC_ALL=C sort; } name:ph… (push) Successful in 48s
Build custom container images / build (map[base_image:postgres:18 build_args:PG_VERSION=18 POSTGIS_VERSION=3 VCHORD_VERSION=0.5.3 context:postgres fingerprint_command:{ dpkg-query -W -f='${binary:Package}=${Version}\n' | LC_ALL=C sort; find /usr/lib/postgresql -type f -exec sha256su… (push) Successful in 55s
Build custom container images / build (map[base_image:python:3 build_args:PYTHON_VERSION=3 context:python-tools fingerprint_command:{ dpkg-query -W -f='${binary:Package}=${Version}\n' | LC_ALL=C sort; pip freeze | LC_ALL=C sort; } name:python-tools oci_labels:org.opencontainers.ima… (push) Successful in 54s
Build custom container images / build (map[base_image:python:3.12-slim build_args:PYTHON_VERSION=3.12-slim context:linkki-tiedotus fingerprint_command:{ dpkg-query -W -f='${binary:Package}=${Version}\n' | LC_ALL=C sort; pip freeze | LC_ALL=C sort; } name:linkki-tiedotus oci_labels:… (push) Successful in 39s
Add Linkki broadcast image
2026-08-19 20:51:26 +00:00

88 lines
2.8 KiB
Python

import sys
from dataclasses import dataclass
@dataclass
class ProgressStep:
label: str
status: str = "pending"
detail: str = ""
class ProgressDisplay:
def __init__(self, stream=None):
self.stream = stream or sys.stdout
self.steps: list[ProgressStep] = []
self._rendered_line_count = 0
self._supports_rewrite = hasattr(self.stream, "isatty") and self.stream.isatty()
def add_step(self, label: str, status: str = "pending", detail: str = "") -> int:
self.steps.append(ProgressStep(label=label, status=status, detail=detail))
self._render()
return len(self.steps) - 1
def update(self, step_index: int, status: str, detail: str = "") -> None:
step = self.steps[step_index]
step.status = status
step.detail = detail
self._render()
def insert_step(self, index: int, label: str, status: str = "pending", detail: str = "") -> int:
self.steps.insert(index, ProgressStep(label=label, status=status, detail=detail))
self._render()
return index
def _render(self) -> None:
lines = [format_step(step) for step in self.steps]
output = "\n".join(lines)
if output:
output += "\n"
if self._supports_rewrite and self._rendered_line_count:
self.stream.write(f"\x1b[{self._rendered_line_count}A")
for _ in range(self._rendered_line_count):
self.stream.write("\x1b[2K\x1b[1B")
self.stream.write(f"\x1b[{self._rendered_line_count}A")
self.stream.write(output)
self.stream.flush()
self._rendered_line_count = len(lines)
class LogProgressDisplay:
def __init__(self, on_change=None):
self.steps: list[ProgressStep] = []
self.on_change = on_change
def add_step(self, label: str, status: str = "pending", detail: str = "") -> int:
self.steps.append(ProgressStep(label=label, status=status, detail=detail))
self._notify()
return len(self.steps) - 1
def update(self, step_index: int, status: str, detail: str = "") -> None:
step = self.steps[step_index]
step.status = status
step.detail = detail
self._notify()
def get_output(self) -> str:
return "\n".join(format_step(step) for step in self.steps)
def _notify(self) -> None:
if self.on_change is not None:
self.on_change(self.get_output())
def format_step(step: ProgressStep) -> str:
prefix_by_status = {
"pending": "[ ]",
"in_progress": "[>]",
"done": "[x]",
"failed": "[!]",
"skipped": "[-]",
}
prefix = prefix_by_status.get(step.status, "[?]")
if step.detail:
return f"{prefix} {step.label} - {step.detail}"
return f"{prefix} {step.label}"