#!/usr/bin/python3
# Экран установки bj при первой загрузке сервера (bj-firstboot) — в том же оформлении, что экран установки
# системы (bj-install-progress): рамка Bridge&Join, шаги установщика с отметками, полоса хода. Запускает
# установщик, весь его вывод пишет в журнал (--log) и по этому выводу рисует экран; код возврата — как у
# установщика. Последний кадр остаётся на экране (сверху) — под ним bj-firstboot выводит учётные записи.
#
# bj-firstboot-screen --log /var/log/bj/firstboot-install.log -- bj-installer install --bundle …
#
# Установщик с BJ_SCREEN=1 печатает в начале «@@plan шаг|шаг|…», дальше — «[i/n] Шаг готово|ошибка»,
# строки-подробности с отступом, в конце «bj установлен» с адресом и кодом или «Установка не завершена.».
# Знаки ✓ ▶ █ ░ ═ — только со шрифтом консоли установщика (BJ_SCREEN_FONT=1), иначе простые символы.
# Проверка без установки: bj-firstboot-screen --demo <файл вывода установщика> [--html кадр.html]
import os
import re
import select
import shutil
import subprocess
import sys
import time
FANCY = os.environ.get("BJ_SCREEN_FONT") == "1"
G = dict(ok="✓", run="▶", wait="·", err="✗", full="█", empty="░", h="═", v="║", tl="╔", tr="╗", bl="╚", br="╝") if FANCY else \
dict(ok="+", run=">", wait=".", err="x", full="#", empty="-", h="=", v="|", tl="+", tr="+", bl="+", br="+")
RESET, BOLD, DIM = "\033[0m", "\033[1m", "\033[2m"
GREEN, CYAN, RED, WHITE, GREY, YELLOW = "\033[32m", "\033[36m", "\033[31m", "\033[97m", "\033[90m", "\033[33m"
STEP_RE = re.compile(r"\[(\d+)/(\d+)\]\s+(.*?)\s*(готово|ошибка)?\s*$")
class State:
def __init__(self):
self.plan, self.status, self.cur = [], {}, 0 # status: номер шага → ok|err
self.cur_started = time.time()
self.started = time.time()
self.caption = ""
self.done = False
self.error = ""
self.urls, self.code = [], ""
self.buf = ""
def feed(self, text):
self.buf += text
while "\n" in self.buf:
line, self.buf = self.buf.split("\n", 1)
self.line(line)
m = STEP_RE.search(self.buf) # строка шага без конца — шаг идёт
if m:
self.start_step(int(m.group(1)), int(m.group(2)), m.group(3))
def start_step(self, i, n, title):
if not self.plan:
self.plan = [""] * n
if i <= len(self.plan) and not self.plan[i - 1]:
self.plan[i - 1] = title
if i != self.cur:
self.cur, self.cur_started, self.caption = i, time.time(), ""
def line(self, line):
s = line.strip()
if s.startswith("@@plan "):
self.plan = s[7:].split("|")
return
m = STEP_RE.search(line)
if m:
i = int(m.group(1))
self.start_step(i, int(m.group(2)), m.group(3))
if m.group(4) == "готово":
self.status[i] = "ok"
elif m.group(4) == "ошибка":
self.status[i] = "err"
return
if s in ("готово", "ошибка") and self.cur: # отметка шага пришла отдельной строкой
self.status[self.cur] = "ok" if s == "готово" else "err"
return
if s.startswith("Установка не завершена."):
self.error = s[len("Установка не завершена."):].strip()
elif s == "bj установлен":
self.done = True
elif "/first-run" in s and s.startswith("http"):
self.urls.append(s)
elif s.startswith("Одноразовый код первичной настройки:"):
self.code = s.split(":", 1)[1].strip()
elif line.startswith(" ") and s and self.cur and not self.done:
self.caption = s # подробность последнего шага
def percent(self):
n = max(1, len(self.plan))
k = sum(1 for v in self.status.values() if v == "ok")
if self.cur and self.cur not in self.status: # текущий шаг — доля по времени, не больше 90 %
k += min(0.9, (time.time() - self.cur_started) / 60.0)
return min(100, int(100 * k / n))
def mmss(sec):
sec = int(max(0, sec))
return "%d:%02d" % (sec // 60, sec % 60)
def frame(st, cols, rows, final=False):
W = 70
inner = W - 2
lines = []
def row(parts=()):
parts, text = list(parts), ""
for i, (t, c) in enumerate(parts):
if len(text) + len(t) > inner:
parts = parts[:i] + [(t[: inner - len(text)], c)]
text += parts[-1][0]
break
text += t
lines.append([(G["v"], GREY)] + parts + [(" " * (inner - len(text)), ""), (G["v"], GREY)])
lines.append([(G["tl"] + G["h"] * inner + G["tr"], GREY)])
row()
row([(" Bridge&Join", BOLD + WHITE)])
row([(" шлюз, объединяющий системы финансовой организации", DIM)])
row()
row([(" Установка bj на сервер", BOLD)])
row()
for i, title in enumerate(st.plan, 1):
title = title or "…"
stv = st.status.get(i)
if stv == "ok":
row([(" " + G["ok"] + " ", GREEN), (title, "")])
elif stv == "err":
row([(" " + G["err"] + " ", RED), (title, BOLD + RED)])
elif i == st.cur and not final:
t = mmss(time.time() - st.cur_started)
row([(" " + G["run"] + " ", CYAN), (title, BOLD + CYAN), (" " * max(1, inner - 9 - len(title) - len(t)), ""), (t, DIM)])
else:
row([(" " + G["wait"] + " ", GREY), (title, GREY)])
row()
if st.error or (final and not st.done):
row([(" " + G["err"] + " Установка не завершена.", BOLD + RED)])
for chunk in wrap(st.error or "подробности — в журнале установки", inner - 6):
row([(" " + chunk, "")])
elif st.done:
row([(" " + G["ok"] + " bj установлен", BOLD + GREEN)])
row()
row([(" Откройте в браузере на своём компьютере:", "")])
for u in st.urls[:3]:
row([(" " + u, BOLD + WHITE)])
if st.code:
row([(" Одноразовый код первичной настройки: ", ""), (st.code, BOLD + YELLOW)])
row()
row([(" Браузер предупредит о сертификате — это сертификат самого", DIM)])
row([(" сервера bj: подтвердите исключение.", DIM)])
else:
pct = st.percent()
bar_w = 48
full = int(bar_w * pct / 100)
row([(" ", ""), (G["full"] * full, CYAN), (G["empty"] * (bar_w - full), GREY), (" %3d %%" % pct, BOLD)])
row([(" " + st.caption, DIM)] if st.caption else [])
row()
row([(" Не выключайте сервер: обычно это занимает несколько минут.", "")])
row()
lines.append([(G["bl"] + G["h"] * inner + G["br"], GREY)])
foot = "прошло " + mmss(time.time() - st.started)
lines.append([(" " * max(0, (W - len(foot)) // 2) + foot, GREY)])
left = max(0, (cols - W) // 2)
top = 1 if final else max(0, (rows - len(lines)) // 2) # итог — сверху: ниже bj-firstboot выведет учётные записи
return [[("", "")]] * top + [[(" " * left, "")] + ln for ln in lines]
def wrap(text, width):
out, cur = [], ""
for w in text.split():
if cur and len(cur) + 1 + len(w) > width:
out.append(cur)
cur = w
else:
cur = (cur + " " + w).strip()
if cur:
out.append(cur)
return out[:4] or [""]
def to_ansi(fr):
return "\033[H\033[2J" + "".join("".join((c + t + RESET) if c else t for t, c in ln) + "\n" for ln in fr)
def to_html(fr):
css = {GREEN: "#4e9a06", CYAN: "#06989a", RED: "#cc0000", GREY: "#888", DIM: "#9a9a9a", WHITE: "#fff", YELLOW: "#c4a000"}
rows = []
for ln in fr:
h = ""
for t, c in ln:
t = t.replace("&", "&").replace("<", "<")
if not c:
h += t
continue
color = next((css[k] for k in (GREEN, CYAN, RED, GREY, WHITE, YELLOW, DIM) if k in c), "#ddd")
h += '%s' % (color, "700" if BOLD in c else "400", t)
rows.append(h)
return ('' + "\n".join(rows) + "
")
def main():
a = sys.argv[1:]
if len(a) >= 2 and a[0] == "--demo":
st = State()
st.feed(open(a[1], encoding="utf-8").read())
st.started -= 95
st.cur_started -= 24
fr = frame(st, 100, 40, final=st.done or bool(st.error))
if len(a) >= 4 and a[2] == "--html":
open(a[3], "w", encoding="utf-8").write(to_html(fr))
else:
sys.stdout.write(to_ansi(fr))
return 0
if len(a) < 4 or a[0] != "--log" or a[2] != "--":
sys.stderr.write("bj-firstboot-screen --log ФАЙЛ -- команда …\n")
return 2
log_path, cmd = a[1], a[3:]
os.makedirs(os.path.dirname(log_path), exist_ok=True)
log = open(log_path, "ab", buffering=0)
log.write(("\n==== %s: %s\n" % (time.strftime("%Y-%m-%d %H:%M:%S"), " ".join(cmd))).encode())
env = dict(os.environ, BJ_SCREEN="1", NO_COLOR="1")
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, env=env, bufsize=0)
st = State()
sys.stdout.write("\033[?25l")
try:
while True:
r, _, _ = select.select([p.stdout], [], [], 1.0)
if r:
chunk = os.read(p.stdout.fileno(), 65536)
if not chunk:
break
log.write(chunk)
st.feed(chunk.decode("utf-8", "replace"))
cols, rows = shutil.get_terminal_size((100, 40))
sys.stdout.write(to_ansi(frame(st, cols, rows)))
sys.stdout.flush()
code = p.wait()
if code != 0 and not st.error:
st.error = "установщик завершился с кодом %d" % code
cols, rows = shutil.get_terminal_size((100, 40))
sys.stdout.write(to_ansi(frame(st, cols, rows, final=True)))
return code
finally:
sys.stdout.write("\033[?25h")
sys.stdout.flush()
if __name__ == "__main__":
sys.exit(main())