Fix pale Wi-Fi/Bluetooth tray icons, das-panel-icon help, publisher and deployer (0.9.5)
- Light theme uses Papirus-Light: the panel icons of Papirus are #dfdfdf, made for dark panels (xsettings, theme index, login screen). - das-panel-icon help/-h/--help print usage instead of setting an icon; unknown options are rejected. - New packages publisher (publish) and deployer (deploy) on the desktop image: config in ~/.config/<name>/ (neutral example copied on first run), history in ~/.local/state/<name>/. - das-theme 0.3.0, das-settings 0.6.0. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Sonnet 5.5
@@ -0,0 +1,2 @@
|
||||
"""Publish apk, AppImage and binaries to the apps page of the home Tomcat over ssh + rsync."""
|
||||
__version__ = "0.3.0"
|
||||
@@ -0,0 +1,173 @@
|
||||
"""Command-line mode: the same pipeline as the TUI, for scripts and quick publishing."""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
from . import __version__, config, detect, packages, pipeline, remote
|
||||
from .pipeline import Options, Plan
|
||||
|
||||
SIGN = {"new": ("+", "32"), "replace": ("~", "33"), "meta": ("~", "36"), "notes": ("~", "36"),
|
||||
"delete": ("-", "31"), "same": ("=", "2"), "skip": ("!", "33")}
|
||||
COLOR = True
|
||||
|
||||
|
||||
def paint(text: str, code: str) -> str:
|
||||
return f"\033[{code}m{text}\033[0m" if COLOR else text
|
||||
|
||||
|
||||
def print_plan(plan: Plan) -> None:
|
||||
a = plan.app
|
||||
print(f"\n{a.name} {a.src} -> {plan.server.target}:{plan.server.apps_dir}"
|
||||
f" {('rev ' + plan.git_rev) if plan.git_rev else ''}")
|
||||
for i in plan.items:
|
||||
sign, code = SIGN[i.action]
|
||||
print(paint(f" {sign} {i.describe(a)}", code))
|
||||
for n in plan.notes:
|
||||
print(f" ! {n}")
|
||||
|
||||
|
||||
def confirm(question: str, yes: bool) -> bool:
|
||||
if yes:
|
||||
return True
|
||||
if not sys.stdin.isatty():
|
||||
print("not a terminal: pass --yes to confirm", file=sys.stderr)
|
||||
return False
|
||||
return input(f"{question} [y/N] ").strip().lower() in ("y", "yes", "т", "так")
|
||||
|
||||
|
||||
def notes_text(args) -> str:
|
||||
if args.notes_file:
|
||||
return Path(args.notes_file).expanduser().read_text()
|
||||
return (args.notes or "").replace("\\n", "\n")
|
||||
|
||||
|
||||
def options(args) -> Options:
|
||||
return Options(build=not args.no_build, force=args.force, prune=not args.no_prune, meta=not args.no_meta,
|
||||
notes=notes_text(args))
|
||||
|
||||
|
||||
def cmd_list(cfg, args):
|
||||
st, others = pipeline.status_all(cfg)
|
||||
style = {"newer": "1;32", "unpublished": "33", "older": "33", "error": "31", "no build": "2"}
|
||||
for a in cfg.apps:
|
||||
s = st[a.name]
|
||||
print(f"{a.name:22} " + paint(f"{s.state:12}", style.get(s.state, "0"))
|
||||
+ f" local: {s.local or '-':32} site: {s.remote or '-'}")
|
||||
if others:
|
||||
print(f"\nnot in {cfg.path.name}: {', '.join(others)}")
|
||||
|
||||
|
||||
def cmd_publish(cfg, args):
|
||||
names = [a.name for a in cfg.apps] if args.all else args.apps
|
||||
if not names:
|
||||
print("name apps or pass --all", file=sys.stderr)
|
||||
return 1
|
||||
if args.all:
|
||||
st, _ = pipeline.status_all(cfg)
|
||||
names = [n for n in names if st[n].state in ("newer", "unpublished")]
|
||||
print(f"newer locally: {', '.join(names) or 'none'}")
|
||||
rc = 0
|
||||
for name in names:
|
||||
plan = pipeline.preview(cfg, cfg.app(name), options(args), log=print)
|
||||
print_plan(plan)
|
||||
if args.dry_run or not plan.changes:
|
||||
continue
|
||||
if not confirm(f"publish {name}?", args.yes):
|
||||
rc = rc or 1
|
||||
continue
|
||||
res = pipeline.execute(plan, log=print)
|
||||
rc = rc if res.ok else 2
|
||||
return rc
|
||||
|
||||
|
||||
def cmd_versions(cfg, args):
|
||||
for p in pipeline.versions(cfg, cfg.app(args.app)):
|
||||
print(f"{p.version:14} {p.kind.label:9} {p.size / 1048576:8.1f} MB {p.file}")
|
||||
return 0
|
||||
|
||||
|
||||
def cmd_remove(cfg, args):
|
||||
app = cfg.app(args.app)
|
||||
if not confirm(f"delete {app.name} v{args.version} from the site?", args.yes):
|
||||
return 1
|
||||
res = pipeline.remove(cfg, app, args.version, log=print)
|
||||
print(res.message)
|
||||
return 0 if res.ok else 2
|
||||
|
||||
|
||||
def cmd_add(cfg, args):
|
||||
app = detect.new_app(Path(args.path), args.name)
|
||||
print(config.app_to_toml(app))
|
||||
try:
|
||||
builds, notes = packages.resolve(app)
|
||||
for b in builds:
|
||||
print(f"# found {b.path.name} -> {b.site_name(app.name)}")
|
||||
for n in notes:
|
||||
print(f"# {n}")
|
||||
except packages.PackageError as e:
|
||||
print(f"# {e}")
|
||||
if confirm(f"append to {cfg.path}?", args.yes):
|
||||
config.append_apps(cfg.path, [app])
|
||||
return 0
|
||||
|
||||
|
||||
def main(argv=None) -> int:
|
||||
p = argparse.ArgumentParser(prog="publish", description="Publish apk, AppImage and binaries to the apps page "
|
||||
"of the home Tomcat. No command: start the TUI.")
|
||||
p.add_argument("--config", default=None, help="publish.toml (default: next to publish.py)")
|
||||
p.add_argument("--no-color", action="store_true")
|
||||
p.add_argument("-V", "--version", action="version", version=__version__)
|
||||
sub = p.add_subparsers(dest="cmd")
|
||||
|
||||
sub.add_parser("list", help="local and published versions of the apps")
|
||||
for name, hlp in (("publish", "preview and publish apps"), ("preview", "show what a publish would change")):
|
||||
s = sub.add_parser(name, help=hlp)
|
||||
s.add_argument("apps", nargs="*")
|
||||
s.add_argument("-a", "--all", action="store_true", help="every app with a newer local build")
|
||||
s.add_argument("-y", "--yes", action="store_true", help="do not ask for confirmation")
|
||||
s.add_argument("-n", "--dry-run", action="store_true", default=name == "preview")
|
||||
s.add_argument("-m", "--notes", help="what's new in this version (\\n for new lines)")
|
||||
s.add_argument("--notes-file", help="what's new, from a file")
|
||||
s.add_argument("--force", action="store_true", help="publish an older build, overwrite changed files")
|
||||
s.add_argument("--no-build", action="store_true")
|
||||
s.add_argument("--no-prune", action="store_true", help="keep versions beyond `keep`")
|
||||
s.add_argument("--no-meta", action="store_true", help="do not upload the description and the icon")
|
||||
s = sub.add_parser("versions", help="published packages of an app")
|
||||
s.add_argument("app")
|
||||
s = sub.add_parser("remove", help="unpublish a version (all its packages and notes)")
|
||||
s.add_argument("app")
|
||||
s.add_argument("version")
|
||||
s.add_argument("-y", "--yes", action="store_true")
|
||||
s = sub.add_parser("add", help="add a local project to publish.toml")
|
||||
s.add_argument("path")
|
||||
s.add_argument("--name")
|
||||
s.add_argument("-y", "--yes", action="store_true")
|
||||
sub.add_parser("check", help="validate publish.toml")
|
||||
sub.add_parser("tui", help="start the TUI (default)")
|
||||
|
||||
args = p.parse_args(argv)
|
||||
global COLOR
|
||||
COLOR = sys.stdout.isatty() and not args.no_color
|
||||
cfg_path = Path(args.config) if args.config else Path(__file__).resolve().parent.parent / "publish.toml"
|
||||
try:
|
||||
cfg = config.load(cfg_path)
|
||||
except (OSError, config.ConfigError) as e:
|
||||
print(f"config: {e}", file=sys.stderr)
|
||||
return 1
|
||||
if args.cmd in (None, "tui"):
|
||||
from .tui import run
|
||||
return run(cfg)
|
||||
if args.cmd == "check":
|
||||
print(f"{cfg_path}: {len(cfg.apps)} apps, OK")
|
||||
return 0
|
||||
handlers = {"list": cmd_list, "publish": cmd_publish, "preview": cmd_publish, "versions": cmd_versions,
|
||||
"remove": cmd_remove, "add": cmd_add}
|
||||
try:
|
||||
return handlers[args.cmd](cfg, args) or 0
|
||||
except (config.ConfigError, pipeline.PublishError, remote.RemoteError, OSError) as e:
|
||||
print(f"error: {e}", file=sys.stderr)
|
||||
return 1
|
||||
except KeyboardInterrupt:
|
||||
return 130
|
||||
@@ -0,0 +1,187 @@
|
||||
"""publish.toml: server settings and the registry of published apps."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
import tomllib
|
||||
from dataclasses import dataclass, field, fields
|
||||
from pathlib import Path
|
||||
|
||||
# <AppName> as apps.jsp groups it: no "_<digit>", "-v<digit>", ... inside (that starts the version)
|
||||
NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.\-]{0,63}$")
|
||||
# Version lines in build files: versionName = "1.2.0" (gradle), Version = "0.10.0" (Go), version = "0.1.0"
|
||||
DEFAULT_VERSION_RE = r"""(?:versionName|Version|version)\s*[=:]?\s*["']([0-9][^"']*)["']"""
|
||||
|
||||
|
||||
class ConfigError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass
|
||||
class Server:
|
||||
host: str = "192.168.88.60"
|
||||
user: str = "das"
|
||||
port: int = 22
|
||||
# Package storage of apps.jsp: only download.jsp serves it
|
||||
apps_dir: str = "/home/das/app/docker/tomcat/data/webapps/ttt/WEB-INF/apps"
|
||||
# Requested on the server after an upload; the file name is appended (URL-encoded)
|
||||
check_url: str = "http://localhost:8010/download.jsp?f="
|
||||
state_dir: str = "/home/das/app_deploy"
|
||||
keep: int = 3 # versions of each package type per app; 0 = keep all
|
||||
local_roots: list[str] = field(default_factory=lambda: ["~/Dev/Projects"])
|
||||
|
||||
@property
|
||||
def target(self) -> str:
|
||||
return f"{self.user}@{self.host}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class App:
|
||||
name: str
|
||||
src: str
|
||||
# globs relative to src; each one gives one package (the newest match)
|
||||
packages: list[str] = field(default_factory=list)
|
||||
# version for files without _v<version> in their name, and a staleness check for the others
|
||||
version_file: str = ""
|
||||
version_regex: str = ""
|
||||
build: str = ""
|
||||
build_env: dict[str, str] = field(default_factory=dict)
|
||||
# local files uploaded as <name>.txt (title + description) and <name>.png|.webp (icon)
|
||||
info: str = ""
|
||||
icon: str = ""
|
||||
keep: int = -1 # -1 = server keep
|
||||
|
||||
@property
|
||||
def src_path(self) -> Path:
|
||||
return Path(os.path.expanduser(self.src))
|
||||
|
||||
def local(self, rel: str) -> Path:
|
||||
p = Path(os.path.expanduser(rel))
|
||||
return p if p.is_absolute() else self.src_path / p
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config:
|
||||
path: Path
|
||||
server: Server
|
||||
apps: list[App]
|
||||
|
||||
def app(self, name: str) -> App:
|
||||
for a in self.apps:
|
||||
if a.name.lower() == name.lower():
|
||||
return a
|
||||
raise ConfigError(f"unknown app: {name}")
|
||||
|
||||
def keep(self, app: App) -> int:
|
||||
return self.server.keep if app.keep < 0 else app.keep
|
||||
|
||||
|
||||
def _check_types(obj, data: dict, where: str):
|
||||
known = {f.name: f for f in fields(obj)}
|
||||
for k, v in data.items():
|
||||
if k not in known:
|
||||
raise ConfigError(f"{where}: unknown key '{k}'")
|
||||
default = getattr(obj, k)
|
||||
if isinstance(default, bool):
|
||||
ok = isinstance(v, bool)
|
||||
elif isinstance(default, int):
|
||||
ok = isinstance(v, int) and not isinstance(v, bool)
|
||||
elif isinstance(default, list):
|
||||
ok = isinstance(v, list) and all(isinstance(x, str) for x in v)
|
||||
elif isinstance(default, dict):
|
||||
ok = isinstance(v, dict) and all(isinstance(x, str) for x in v.values())
|
||||
else:
|
||||
ok = isinstance(v, str)
|
||||
if not ok:
|
||||
raise ConfigError(f"{where}: '{k}' has wrong type")
|
||||
setattr(obj, k, v)
|
||||
|
||||
|
||||
def validate_server(s: Server) -> None:
|
||||
d = s.apps_dir.rstrip("/")
|
||||
if not d.startswith("/") or d.count("/") < 3 or ".." in Path(d).parts:
|
||||
raise ConfigError(f"[server]: unsafe apps_dir '{s.apps_dir}'")
|
||||
if s.keep < 0:
|
||||
raise ConfigError("[server]: keep must be >= 0")
|
||||
|
||||
|
||||
def validate_app(app: App) -> None:
|
||||
w = f"app '{app.name}'"
|
||||
if not NAME_RE.match(app.name) or re.search(r"[_-]v?\d", app.name):
|
||||
raise ConfigError(f"{w}: bad name (letters, digits, '_.-'; no '_<digit>' or '_v<digit>')")
|
||||
if not app.src:
|
||||
raise ConfigError(f"{w}: src is required")
|
||||
if not app.packages:
|
||||
raise ConfigError(f"{w}: packages (globs relative to src) are required")
|
||||
for g in app.packages:
|
||||
if not g.strip() or ".." in Path(g).parts:
|
||||
raise ConfigError(f"{w}: bad package glob '{g}'")
|
||||
if app.version_regex:
|
||||
try:
|
||||
if re.compile(app.version_regex).groups < 1:
|
||||
raise ConfigError(f"{w}: version_regex needs a (group)")
|
||||
except re.error as e:
|
||||
raise ConfigError(f"{w}: version_regex: {e}") from None
|
||||
if app.icon and not app.icon.lower().endswith((".png", ".webp")):
|
||||
raise ConfigError(f"{w}: icon must be .png or .webp")
|
||||
if app.keep < -1:
|
||||
raise ConfigError(f"{w}: keep must be >= 0")
|
||||
|
||||
|
||||
def load(path: str | Path) -> Config:
|
||||
path = Path(path)
|
||||
with open(path, "rb") as f:
|
||||
try:
|
||||
data = tomllib.load(f)
|
||||
except tomllib.TOMLDecodeError as e:
|
||||
raise ConfigError(f"{path}: {e}") from None
|
||||
server = Server()
|
||||
_check_types(server, data.get("server", {}), "[server]")
|
||||
validate_server(server)
|
||||
apps, seen = [], set()
|
||||
for i, raw in enumerate(data.get("app", [])):
|
||||
raw = dict(raw)
|
||||
try:
|
||||
app = App(name=raw.pop("name"), src=raw.pop("src"))
|
||||
except KeyError as e:
|
||||
raise ConfigError(f"app #{i + 1}: missing {e}") from None
|
||||
_check_types(app, raw, f"app '{app.name}'")
|
||||
validate_app(app)
|
||||
if app.name.lower() in seen:
|
||||
raise ConfigError(f"duplicate app '{app.name}'")
|
||||
seen.add(app.name.lower())
|
||||
apps.append(app)
|
||||
return Config(path, server, apps)
|
||||
|
||||
|
||||
def _toml_value(v) -> str:
|
||||
if isinstance(v, bool):
|
||||
return "true" if v else "false"
|
||||
if isinstance(v, int):
|
||||
return str(v)
|
||||
if isinstance(v, list):
|
||||
return "[" + ", ".join(_toml_value(x) for x in v) + "]"
|
||||
if isinstance(v, dict):
|
||||
return "{ " + ", ".join(f"{k} = {_toml_value(x)}" for k, x in v.items()) + " }"
|
||||
s = str(v).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n")
|
||||
return f'"{s}"'
|
||||
|
||||
|
||||
def app_to_toml(app: App) -> str:
|
||||
"""[[app]] block with the non-default keys only."""
|
||||
blank = App(name="", src="")
|
||||
lines = ["[[app]]"]
|
||||
for f in fields(app):
|
||||
v = getattr(app, f.name)
|
||||
if f.name in ("name", "src") or v != getattr(blank, f.name):
|
||||
lines.append(f"{f.name} = {_toml_value(v)}")
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def append_apps(path: Path, apps: list[App]) -> None:
|
||||
"""Append blocks to the file: hand-written comments and order are kept."""
|
||||
text = path.read_text() if path.exists() else ""
|
||||
for a in apps:
|
||||
validate_app(a)
|
||||
text = text.rstrip("\n") + "\n\n" + app_to_toml(a)
|
||||
path.write_text(text)
|
||||
@@ -0,0 +1,81 @@
|
||||
"""Guess an [[app]] block for a local project: Android (+ electron desktop), Go, or any dir with builds."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
from . import packages
|
||||
from .config import App
|
||||
|
||||
_ROOT_NAME = re.compile(r"""rootProject\.name\s*=\s*["']([^"']+)["']""")
|
||||
_GO_VERSION = re.compile(r'Version\s*=\s*"')
|
||||
|
||||
|
||||
def _rel(p: Path, base: Path) -> str:
|
||||
return str(p.relative_to(base))
|
||||
|
||||
|
||||
def _home(p: Path) -> str:
|
||||
home = Path.home()
|
||||
return "~/" + str(p.relative_to(home)) if p.is_relative_to(home) else str(p)
|
||||
|
||||
|
||||
def _site_name(raw: str) -> str:
|
||||
"""A name apps.jsp groups correctly: no spaces, no _<digit>."""
|
||||
name = re.sub(r"[^A-Za-z0-9_.\-]+", "", raw)
|
||||
return re.sub(r"[_-]v?(?=\d)", "", name) or "App"
|
||||
|
||||
|
||||
def _name_from_builds(files: list[Path]) -> str:
|
||||
for f in files:
|
||||
m = packages.LOCAL_VERSION.search(f.name)
|
||||
if m and m.start() > 0:
|
||||
return f.name[:m.start()]
|
||||
return ""
|
||||
|
||||
|
||||
def _first(base: Path, *names: str) -> Path | None:
|
||||
for n in names:
|
||||
if (base / n).is_file():
|
||||
return base / n
|
||||
return None
|
||||
|
||||
|
||||
def new_app(path: Path, name: str | None = None) -> App:
|
||||
path = path.expanduser().resolve()
|
||||
# no build command is guessed: Android Studio signs into app/release, gradle into app/build/outputs
|
||||
globs, builds, version_file = [], [], ""
|
||||
|
||||
gradle = _first(path, "app/build.gradle.kts", "app/build.gradle")
|
||||
if gradle:
|
||||
version_file = _rel(gradle, path)
|
||||
# both places of a signed release APK: the best build of the two is published
|
||||
for d in ("app/release", "app/build/outputs/apk/release"):
|
||||
globs.append(f"{d}/*.apk")
|
||||
builds += sorted((path / d).glob("*.apk"))
|
||||
dist = path / "desktop/dist"
|
||||
if dist.is_dir():
|
||||
for ext in (".AppImage", ".deb", ".rpm", ".exe", ".msi", ".dmg"):
|
||||
if any(dist.glob(f"*{ext}")):
|
||||
globs.append(f"desktop/dist/*{ext}")
|
||||
else:
|
||||
go_version = next((p for p in path.glob("internal/version/*.go")
|
||||
if _GO_VERSION.search(p.read_text(errors="replace"))), None)
|
||||
if go_version:
|
||||
version_file = _rel(go_version, path)
|
||||
elif (path / "pyproject.toml").is_file():
|
||||
version_file = "pyproject.toml"
|
||||
for d in ("build", "dist", "bin", "release"):
|
||||
for f in sorted((path / d).glob("*")) if (path / d).is_dir() else []:
|
||||
if f.is_file() and packages.kind_of(f.name, os.access(f, os.X_OK)):
|
||||
builds.append(f)
|
||||
k = packages.kind_of(f.name)
|
||||
globs.append(f"{d}/*{k.ext}" if k and k.ext else f"{d}/{f.name}")
|
||||
globs = list(dict.fromkeys(globs))
|
||||
|
||||
if not name:
|
||||
settings = _first(path, "settings.gradle.kts", "settings.gradle")
|
||||
m = _ROOT_NAME.search(settings.read_text(errors="replace")) if settings else None
|
||||
name = _name_from_builds(builds) or (m.group(1) if m else "") or path.name
|
||||
return App(name=_site_name(name), src=_home(path), packages=globs, version_file=version_file)
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Package types and names, the same rules as WEB-INF/incl/packages.jspf and apps.jsp on the server."""
|
||||
from __future__ import annotations
|
||||
|
||||
import glob
|
||||
import os
|
||||
import re
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from .config import DEFAULT_VERSION_RE, App
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Kind:
|
||||
ext: str # as written in published names; "" = executable binary
|
||||
label: str
|
||||
|
||||
@property
|
||||
def executable(self) -> bool:
|
||||
return self.ext in ("", ".AppImage")
|
||||
|
||||
|
||||
# Order of apps.jsp download buttons. ".tar.gz" is a compound extension, so it must stay last:
|
||||
# kind_of() returns the first endswith match and nothing else ends in ".tar.gz".
|
||||
KINDS = (Kind(".apk", "APK"), Kind(".AppImage", "AppImage"), Kind(".deb", "DEB"), Kind(".rpm", "RPM"),
|
||||
Kind(".exe", "EXE"), Kind(".msi", "MSI"), Kind(".dmg", "DMG"), Kind(".tar.gz", "tar.gz"))
|
||||
BINARY = Kind("", "binary")
|
||||
NOT_PACKAGE = re.compile(r"(?i).*\.(txt|md|png|webp|jpe?g|svg|sh|jsp|html?|json|ya?ml|zip|log)$")
|
||||
|
||||
# Server side: apps.jsp PKG_NAME
|
||||
SITE_NAME = re.compile(r"(?i)^(.+?)[_-]v?(\d+(?:\.\d+)*[\w.\-]*)$")
|
||||
# Local build names: Foo_v1.2.0.apk, Foo_v1.2.0-release.apk, Foo-1.5-release.apk, Foo_v0.7.0_linux_x86_64.AppImage
|
||||
LOCAL_VERSION = re.compile(r"[_-]v(\d+(?:\.\d+)*)|[_-](\d+(?:\.\d+)+)")
|
||||
|
||||
|
||||
def kind_of(name: str, executable: bool = True) -> Kind | None:
|
||||
if name.startswith("."): # also rsync temp files of an upload in progress
|
||||
return None
|
||||
lower = name.lower()
|
||||
for k in KINDS:
|
||||
if lower.endswith(k.ext.lower()):
|
||||
return k
|
||||
return BINARY if executable and not NOT_PACKAGE.match(name) else None
|
||||
|
||||
|
||||
def version_key(v: str) -> tuple[int, ...]:
|
||||
"""Numeric-aware, like compareVersions() in apps.jsp: 1.10.0 > 1.9.3, 1.0 == 1.0.0."""
|
||||
parts = [int(x) for x in re.split(r"\D+", v) if x]
|
||||
while parts and parts[-1] == 0:
|
||||
parts.pop()
|
||||
return tuple(parts)
|
||||
|
||||
|
||||
def site_name(app: str, version: str, kind: Kind) -> str:
|
||||
return f"{app}_v{version}{kind.ext}"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Published:
|
||||
"""A package file in apps_dir."""
|
||||
file: str
|
||||
app: str
|
||||
version: str
|
||||
kind: Kind
|
||||
size: int = 0
|
||||
mtime: float = 0.0
|
||||
|
||||
|
||||
def parse_published(file: str, executable: bool, size: int = 0, mtime: float = 0.0) -> Published | None:
|
||||
kind = kind_of(file, executable)
|
||||
if kind is None:
|
||||
return None
|
||||
base = file[:len(file) - len(kind.ext)] if kind.ext else file
|
||||
m = SITE_NAME.match(base)
|
||||
return Published(file, m.group(1) if m else base, m.group(2) if m else "", kind, size, mtime)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Local:
|
||||
"""A local build chosen for publishing."""
|
||||
path: Path
|
||||
kind: Kind
|
||||
version: str
|
||||
glob: str
|
||||
|
||||
def site_name(self, app: str) -> str:
|
||||
return site_name(app, self.version, self.kind)
|
||||
|
||||
|
||||
class PackageError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def file_version(path: Path) -> str:
|
||||
stem = path.name
|
||||
kind = kind_of(path.name)
|
||||
if kind and kind.ext:
|
||||
stem = stem[:len(stem) - len(kind.ext)]
|
||||
m = LOCAL_VERSION.search(stem)
|
||||
return (m.group(1) or m.group(2)) if m else ""
|
||||
|
||||
|
||||
def source_version(app: App) -> str:
|
||||
"""Version from app.version_file, '' when not configured."""
|
||||
if not app.version_file:
|
||||
return ""
|
||||
path = app.local(app.version_file)
|
||||
try:
|
||||
text = path.read_text(errors="replace")
|
||||
except OSError as e:
|
||||
raise PackageError(f"version_file: {e}") from None
|
||||
m = re.search(app.version_regex or DEFAULT_VERSION_RE, text)
|
||||
if not m:
|
||||
raise PackageError(f"no version found in {path}")
|
||||
return m.group(1).strip()
|
||||
|
||||
|
||||
def resolve(app: App) -> tuple[list[Local], list[str]]:
|
||||
"""The best build of each package type over all globs (highest version, then the newest file),
|
||||
and notes about missing or suspicious ones. Several globs of one type are alternative places
|
||||
of the same build, e.g. app/release (Android Studio) and app/build/outputs/apk/release (gradle)."""
|
||||
notes = []
|
||||
src_version = source_version(app)
|
||||
by_kind: dict[Kind, list[Local]] = {}
|
||||
for pattern in app.packages:
|
||||
for p in sorted(glob.glob(str(app.local(pattern)))):
|
||||
path = Path(p)
|
||||
if not path.is_file():
|
||||
continue
|
||||
kind = kind_of(path.name, os.access(path, os.X_OK))
|
||||
if kind is not None:
|
||||
by_kind.setdefault(kind, []).append(Local(path, kind, file_version(path) or src_version, pattern))
|
||||
for pattern in app.packages:
|
||||
expected = kind_of(Path(pattern).name)
|
||||
if expected not in by_kind and not any(b.glob == pattern for bs in by_kind.values() for b in bs):
|
||||
notes.append(f"no build matches {pattern}")
|
||||
|
||||
found = []
|
||||
for kind in [k for k in (*KINDS, BINARY) if k in by_kind]:
|
||||
builds = {b.path.resolve(): b for b in by_kind[kind]}.values() # globs may overlap
|
||||
best = max(builds, key=lambda b: (version_key(b.version), b.path.stat().st_mtime))
|
||||
if not best.version:
|
||||
raise PackageError(f"{best.path.name}: no version in the name and no version_file")
|
||||
twins = [b for b in builds if b is not best and version_key(b.version) == version_key(best.version)
|
||||
and b.path.stat().st_size != best.path.stat().st_size]
|
||||
if twins:
|
||||
notes.append(f"{len(twins) + 1} different builds of {kind.label} v{best.version}: the newest is used, "
|
||||
f"{best.path.relative_to(app.src_path) if best.path.is_relative_to(app.src_path) else best.path}")
|
||||
if src_version and version_key(best.version) < version_key(src_version):
|
||||
notes.append(f"{best.path.name} is v{best.version}, the source is v{src_version}: rebuild?")
|
||||
found.append(best)
|
||||
return found, notes
|
||||
@@ -0,0 +1,407 @@
|
||||
"""Publish steps: preview() builds a Plan (nothing changes on the server), execute() applies it."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import hashlib
|
||||
import subprocess
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
|
||||
from . import packages, remote
|
||||
from .config import App, Config
|
||||
from .packages import Kind, Local, Published, version_key
|
||||
|
||||
# Das OS package: /usr/lib is read-only, so the history goes to the user's state dir
|
||||
HISTORY = Path(os.environ.get("XDG_STATE_HOME") or Path.home() / ".local/state") / "publisher" / "history.log"
|
||||
|
||||
# Plan item actions; the first five change the server
|
||||
NEW, REPLACE, META, NOTES, DELETE, SAME, SKIP = "new", "replace", "meta", "notes", "delete", "same", "skip"
|
||||
CHANGING = (NEW, REPLACE, META, NOTES, DELETE)
|
||||
|
||||
|
||||
class PublishError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass
|
||||
class Options:
|
||||
build: bool = True
|
||||
force: bool = False # publish an older version, overwrite a published file with other content
|
||||
prune: bool = True # delete versions beyond keep
|
||||
meta: bool = True # upload changed <App>.txt / icon
|
||||
notes: str = "" # "what's new" of the published version
|
||||
|
||||
|
||||
@dataclass
|
||||
class Item:
|
||||
action: str
|
||||
target: str # file name in apps_dir
|
||||
local: Path | None = None
|
||||
kind: Kind | None = None
|
||||
version: str = ""
|
||||
reason: str = ""
|
||||
|
||||
@property
|
||||
def is_package(self) -> bool:
|
||||
return self.kind is not None
|
||||
|
||||
def describe(self, app: App) -> str:
|
||||
"""'<local file, relative to the project> -> <target>' and size/reason, for plan listings."""
|
||||
src = ""
|
||||
if self.local:
|
||||
rel = self.local.relative_to(app.src_path) if self.local.is_relative_to(app.src_path) else self.local
|
||||
src = f"{rel} -> " if str(rel) != self.target else ""
|
||||
size = f" ({self.local.stat().st_size / 1048576:.1f} MB)" if self.local and self.is_package else ""
|
||||
return f"{src}{self.target}{size}" + (f" [{self.reason}]" if self.reason else "")
|
||||
|
||||
|
||||
@dataclass
|
||||
class Plan:
|
||||
cfg: Config
|
||||
app: App
|
||||
opts: Options
|
||||
items: list[Item] = field(default_factory=list)
|
||||
notes: list[str] = field(default_factory=list)
|
||||
git_rev: str = ""
|
||||
|
||||
@property
|
||||
def server(self):
|
||||
return self.cfg.server
|
||||
|
||||
@property
|
||||
def changes(self) -> list[Item]:
|
||||
return [i for i in self.items if i.action in CHANGING]
|
||||
|
||||
|
||||
@dataclass
|
||||
class Result:
|
||||
ok: bool
|
||||
message: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class RemoteState:
|
||||
# name -> (size, mtime, executable)
|
||||
files: dict[str, tuple[int, float, bool]] = field(default_factory=dict)
|
||||
hashes: dict[str, str] = field(default_factory=dict)
|
||||
|
||||
def published(self, app: str | None = None) -> list[Published]:
|
||||
"""Packages of an app (all packages for None), grouped case-insensitively like apps.jsp."""
|
||||
out = []
|
||||
for name, (size, mtime, ex) in self.files.items():
|
||||
p = packages.parse_published(name, ex, size, mtime)
|
||||
if p and (app is None or p.app.lower() == app.lower()):
|
||||
out.append(p)
|
||||
return out
|
||||
|
||||
|
||||
_STATE = r'''
|
||||
[ -d "$DIR" ] || { echo "no such dir: $DIR" >&2; exit 3; }
|
||||
find "$DIR" -maxdepth 1 -type f -printf 'F %s %T@ %m %f\n'
|
||||
cd "$DIR" && for n in "${HASH[@]}"; do [ -f "$n" ] && sha256sum -- "$n" | sed 's/^/H /'; done
|
||||
true
|
||||
'''
|
||||
|
||||
|
||||
def remote_state(cfg: Config, hash_names: list[str] = ()) -> RemoteState:
|
||||
out = remote.run(cfg.server, _STATE, DIR=cfg.server.apps_dir, HASH=list(hash_names))
|
||||
st = RemoteState()
|
||||
for line in out.splitlines():
|
||||
if line.startswith("F "):
|
||||
_, size, mtime, mode, name = line.split(" ", 4)
|
||||
st.files[name] = (int(size), float(mtime), int(mode, 8) & 0o111 != 0)
|
||||
elif line.startswith("H "):
|
||||
_, sha, name = line.split(None, 2)
|
||||
st.hashes[name] = sha
|
||||
return st
|
||||
|
||||
|
||||
def sha256(path: Path) -> str:
|
||||
h = hashlib.sha256()
|
||||
with open(path, "rb") as f:
|
||||
for chunk in iter(lambda: f.read(1 << 20), b""):
|
||||
h.update(chunk)
|
||||
return h.hexdigest()
|
||||
|
||||
|
||||
def git_rev(src: Path) -> str:
|
||||
try:
|
||||
rev = subprocess.run(["git", "-C", str(src), "rev-parse", "--short", "HEAD"], capture_output=True,
|
||||
text=True, timeout=10).stdout.strip()
|
||||
if not rev:
|
||||
return ""
|
||||
dirty = subprocess.run(["git", "-C", str(src), "status", "--porcelain"], capture_output=True, text=True,
|
||||
timeout=10).stdout.strip()
|
||||
return rev + ("-dirty" if dirty else "")
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return ""
|
||||
|
||||
|
||||
def meta_files(app: App) -> list[tuple[Path, str]]:
|
||||
"""(local file, name in apps_dir) of the description and the icon."""
|
||||
out = []
|
||||
if app.info:
|
||||
out.append((app.local(app.info), f"{app.name}.txt"))
|
||||
if app.icon:
|
||||
icon = app.local(app.icon)
|
||||
out.append((icon, f"{app.name}{icon.suffix.lower()}"))
|
||||
return out
|
||||
|
||||
|
||||
def notes_name(app: str, version: str) -> str:
|
||||
return f"{app}_v{version}.txt"
|
||||
|
||||
|
||||
def prune(published: list[Published], keep: int) -> list[Published]:
|
||||
"""Packages beyond the newest `keep` versions of each type."""
|
||||
if keep <= 0:
|
||||
return []
|
||||
out = []
|
||||
for kind in {p.kind for p in published}:
|
||||
same = sorted((p for p in published if p.kind == kind), key=lambda p: (version_key(p.version), p.mtime),
|
||||
reverse=True)
|
||||
out += same[keep:]
|
||||
return out
|
||||
|
||||
|
||||
def preview(cfg: Config, app: App, opts: Options, log=print) -> Plan:
|
||||
src = app.src_path
|
||||
if not src.exists():
|
||||
raise PublishError(f"local source not found: {src}")
|
||||
if opts.build and app.build:
|
||||
log(f"== build: {app.build}")
|
||||
remote.run_local(app.build, src if src.is_dir() else src.parent, log, app.build_env)
|
||||
try:
|
||||
builds, notes = packages.resolve(app)
|
||||
except packages.PackageError as e:
|
||||
raise PublishError(str(e)) from None
|
||||
if not builds:
|
||||
raise PublishError("nothing to publish: " + "; ".join(notes))
|
||||
meta = [(p, n) for p, n in meta_files(app) if opts.meta]
|
||||
for p, _ in meta:
|
||||
if not p.is_file():
|
||||
notes.append(f"no local {p}")
|
||||
meta = [(p, n) for p, n in meta if p.is_file()]
|
||||
|
||||
log(f"== probe {cfg.server.target}:{cfg.server.apps_dir}")
|
||||
st = remote_state(cfg, [b.site_name(app.name) for b in builds] + [n for _, n in meta])
|
||||
published = st.published(app.name)
|
||||
plan = Plan(cfg, app, opts, notes=notes, git_rev=git_rev(src if src.is_dir() else src.parent))
|
||||
|
||||
for b in builds:
|
||||
plan.items.append(_package_item(app, b, st, published, opts))
|
||||
added = [i for i in plan.items if i.action in (NEW, REPLACE)]
|
||||
|
||||
for path, name in meta:
|
||||
if st.hashes.get(name) != sha256(path):
|
||||
plan.items.append(Item(META, name, path, reason="update" if name in st.files else "new"))
|
||||
|
||||
if opts.notes.strip():
|
||||
for v in sorted({b.version for b in builds}, key=version_key):
|
||||
name = notes_name(app.name, v)
|
||||
plan.items.append(Item(NOTES, name, version=v, reason="update" if name in st.files else "new"))
|
||||
|
||||
if opts.prune:
|
||||
after = [p for p in published if p.file not in {i.target for i in added}]
|
||||
after += [Published(i.target, app.name, i.version, i.kind, mtime=float("inf")) for i in added]
|
||||
gone = prune(after, cfg.keep(app))
|
||||
for p in gone:
|
||||
plan.items.append(Item(DELETE, p.file, kind=p.kind, version=p.version, reason=f"keep {cfg.keep(app)}"))
|
||||
left = {version_key(p.version) for p in after if p not in gone}
|
||||
for v in sorted({p.version for p in gone}, key=version_key):
|
||||
n = notes_name(app.name, v)
|
||||
if version_key(v) not in left and n in st.files:
|
||||
plan.items.append(Item(DELETE, n, version=v, reason="notes of a deleted version"))
|
||||
|
||||
if not added:
|
||||
plan.notes.append("no new packages" + ("" if opts.force else " (--force publishes older or changed builds)"))
|
||||
if plan.git_rev.endswith("-dirty"):
|
||||
plan.notes.append("working tree has uncommitted changes")
|
||||
return plan
|
||||
|
||||
|
||||
def _package_item(app: App, b: Local, st: RemoteState, published: list[Published], opts: Options) -> Item:
|
||||
target = b.site_name(app.name)
|
||||
item = Item(NEW, target, b.path, b.kind, b.version)
|
||||
if target in st.files:
|
||||
if st.hashes.get(target) == sha256(b.path):
|
||||
item.action, item.reason = SAME, "already published"
|
||||
elif opts.force:
|
||||
item.action, item.reason = REPLACE, "same name, other content"
|
||||
else:
|
||||
item.action, item.reason = SKIP, "published with other content: --force overwrites"
|
||||
return item
|
||||
newest = max((p for p in published if p.kind == b.kind), key=lambda p: version_key(p.version), default=None)
|
||||
if newest and version_key(b.version) <= version_key(newest.version):
|
||||
what = "older than" if version_key(b.version) < version_key(newest.version) else "same version as"
|
||||
if opts.force:
|
||||
item.reason = f"{what} {newest.file}"
|
||||
else:
|
||||
item.action, item.reason = SKIP, f"{what} {newest.file}"
|
||||
return item
|
||||
|
||||
|
||||
_WRITE = r'''
|
||||
umask 022
|
||||
printf '%s\n' "$TEXT" > "$DIR/.$NAME.tmp" && mv -f "$DIR/.$NAME.tmp" "$DIR/$NAME" && echo "wrote $NAME"
|
||||
'''
|
||||
|
||||
_DELETE = r'''
|
||||
cd "$DIR" || exit 1
|
||||
for n in "${NAMES[@]}"; do rm -f -- "$n" && echo "deleted $n"; done
|
||||
'''
|
||||
|
||||
_CHECK = r'''
|
||||
rc=0
|
||||
for i in "${!URLS[@]}"; do
|
||||
h=$(curl -sI --max-time 15 "${URLS[$i]}" | tr -d '\r')
|
||||
code=$(printf '%s\n' "$h" | head -1 | cut -d' ' -f2)
|
||||
len=$(printf '%s\n' "$h" | grep -i '^content-length:' | tail -1 | awk '{print $2}')
|
||||
if [ "$code" = 200 ] && [ "$len" = "${SIZES[$i]}" ]; then
|
||||
echo "${NAMES[$i]}: HTTP 200, $len bytes"
|
||||
else
|
||||
echo "ERROR: ${NAMES[$i]}: HTTP ${code:-none}, ${len:-?} bytes (expected ${SIZES[$i]})"; rc=1
|
||||
fi
|
||||
done
|
||||
exit $rc
|
||||
'''
|
||||
|
||||
|
||||
def _delete(cfg: Config, names: list[str], log) -> None:
|
||||
if any("/" in n or n.startswith(".") or not n for n in names):
|
||||
raise PublishError(f"refusing to delete {names}")
|
||||
remote.run(cfg.server, _DELETE, log, DIR=cfg.server.apps_dir, NAMES=names)
|
||||
|
||||
|
||||
def check(cfg: Config, files: list[tuple[str, int]], log) -> bool:
|
||||
"""The site serves each file with the right size (asked on the server)."""
|
||||
if not files or not cfg.server.check_url:
|
||||
return True
|
||||
try:
|
||||
remote.run(cfg.server, _CHECK, log, URLS=[cfg.server.check_url + quote(n) for n, _ in files],
|
||||
NAMES=[n for n, _ in files], SIZES=[s for _, s in files])
|
||||
except remote.RemoteError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _record(cfg: Config, app: App, action: str, result: str, files: list[str], rev: str, log) -> None:
|
||||
ts = datetime.now().isoformat(timespec="seconds")
|
||||
line = f"{ts} {app.name} {action} {result} rev={rev or '-'} {','.join(files) or '-'}"
|
||||
try:
|
||||
remote.run(cfg.server, 'mkdir -p "$STATE" && printf "%s\\n" "$LINE" >> "$STATE/publish.log"',
|
||||
STATE=cfg.server.state_dir, LINE=line)
|
||||
except remote.RemoteError as e:
|
||||
log(f"WARN: publish.log not written: {e}")
|
||||
HISTORY.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(HISTORY, "a") as f:
|
||||
f.write(line + "\n")
|
||||
|
||||
|
||||
def execute(plan: Plan, log=print) -> Result:
|
||||
cfg, app, d = plan.cfg, plan.app, plan.server.apps_dir
|
||||
done: list[str] = []
|
||||
try:
|
||||
# notes first: the page shows them together with the new package
|
||||
for i in plan.items:
|
||||
if i.action == NOTES:
|
||||
remote.run(cfg.server, _WRITE, log, DIR=d, NAME=i.target, TEXT=plan.opts.notes.strip())
|
||||
done.append(i.target)
|
||||
for i in plan.items:
|
||||
if i.action in (NEW, REPLACE, META):
|
||||
size = i.local.stat().st_size
|
||||
log(f"== upload {i.local.name} -> {i.target} ({size / 1048576:.1f} MB)")
|
||||
mode = "755" if i.kind and i.kind.executable else "644"
|
||||
remote.upload(cfg.server, str(i.local), f"{d}/{i.target}", mode, log)
|
||||
done.append(i.target)
|
||||
gone = [i.target for i in plan.items if i.action == DELETE]
|
||||
if gone:
|
||||
log("== prune")
|
||||
_delete(cfg, gone, log)
|
||||
done += [f"-{n}" for n in gone]
|
||||
uploaded = [(i.target, i.local.stat().st_size) for i in plan.items if i.action in (NEW, REPLACE)]
|
||||
ok = True
|
||||
if uploaded and cfg.server.check_url:
|
||||
log("== check")
|
||||
ok = check(cfg, uploaded, log)
|
||||
except (remote.RemoteError, PublishError, OSError) as e:
|
||||
log(f"ERROR: {e}")
|
||||
_record(cfg, app, "publish", "FAILED", done, plan.git_rev, log)
|
||||
return Result(False, str(e))
|
||||
_record(cfg, app, "publish", "ok" if ok else "UNCHECKED", done, plan.git_rev, log)
|
||||
msg = "published" if ok else "published, but the site does not serve it correctly"
|
||||
log(f"== {msg}")
|
||||
return Result(ok, msg)
|
||||
|
||||
|
||||
def versions(cfg: Config, app: App) -> list[Published]:
|
||||
"""Published packages of the app, newest first."""
|
||||
pub = remote_state(cfg).published(app.name)
|
||||
return sorted(pub, key=lambda p: (version_key(p.version), -packages.KINDS.index(p.kind)
|
||||
if p.kind in packages.KINDS else 0), reverse=True)
|
||||
|
||||
|
||||
def remove(cfg: Config, app: App, version: str, log=print) -> Result:
|
||||
"""Unpublish one version: all its packages and its notes."""
|
||||
st = remote_state(cfg)
|
||||
names = [p.file for p in st.published(app.name) if version_key(p.version) == version_key(version)]
|
||||
if not names:
|
||||
return Result(False, f"{app.name} v{version} is not published")
|
||||
n = notes_name(app.name, version)
|
||||
if n in st.files:
|
||||
names.append(n)
|
||||
try:
|
||||
_delete(cfg, names, log)
|
||||
except (remote.RemoteError, PublishError) as e:
|
||||
log(f"ERROR: {e}")
|
||||
return Result(False, str(e))
|
||||
_record(cfg, app, "remove", "ok", [f"-{x}" for x in names], "", log)
|
||||
return Result(True, f"removed {', '.join(names)}")
|
||||
|
||||
|
||||
@dataclass
|
||||
class Status:
|
||||
state: str # current | newer | unpublished | older | no build | error
|
||||
local: str = "" # "apk 1.2.0, AppImage 1.2.0"
|
||||
remote: str = ""
|
||||
|
||||
|
||||
def _summary(pairs: list[tuple[Kind, str]]) -> str:
|
||||
return ", ".join(f"{k.label} {v}" for k, v in sorted(pairs, key=lambda x: version_key(x[1]), reverse=True))
|
||||
|
||||
|
||||
def status_all(cfg: Config) -> tuple[dict[str, Status], list[str]]:
|
||||
"""name -> Status (versions only, contents are compared by preview), and published apps
|
||||
that are not in publish.toml."""
|
||||
st = remote_state(cfg)
|
||||
pub = st.published()
|
||||
out = {}
|
||||
for a in cfg.apps:
|
||||
mine = [p for p in pub if p.app.lower() == a.name.lower()]
|
||||
latest: dict[Kind, Published] = {}
|
||||
for p in mine:
|
||||
if p.kind not in latest or version_key(p.version) > version_key(latest[p.kind].version):
|
||||
latest[p.kind] = p
|
||||
rem = _summary([(k, p.version) for k, p in latest.items()])
|
||||
try:
|
||||
builds, _ = packages.resolve(a)
|
||||
except packages.PackageError as e:
|
||||
out[a.name] = Status("error", str(e), rem)
|
||||
continue
|
||||
loc = _summary([(b.kind, b.version) for b in builds])
|
||||
if not builds:
|
||||
state = "no build"
|
||||
elif not latest:
|
||||
state = "unpublished"
|
||||
else:
|
||||
cmp = [version_key(b.version) > version_key(latest[b.kind].version) if b.kind in latest else True
|
||||
for b in builds]
|
||||
older = [b.kind in latest and version_key(b.version) < version_key(latest[b.kind].version)
|
||||
for b in builds]
|
||||
state = "newer" if any(cmp) else ("older" if any(older) else "current")
|
||||
out[a.name] = Status(state, loc, rem)
|
||||
known = {a.name.lower() for a in cfg.apps}
|
||||
others = sorted({p.app for p in pub if p.app.lower() not in known}, key=str.lower)
|
||||
return out, others
|
||||
@@ -0,0 +1,74 @@
|
||||
"""ssh / rsync wrappers. One multiplexed ssh connection is reused by all calls of a run."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shlex
|
||||
import subprocess
|
||||
from typing import Callable
|
||||
|
||||
from .config import Server
|
||||
|
||||
Log = Callable[[str], None]
|
||||
|
||||
|
||||
class RemoteError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
def ssh_opts(server: Server) -> list[str]:
|
||||
return ["-o", "BatchMode=yes", "-o", "ConnectTimeout=10", "-o", "ControlMaster=auto",
|
||||
"-o", f"ControlPath={os.path.expanduser('~/.ssh')}/cm-%r@%h:%p", "-o", "ControlPersist=120",
|
||||
"-p", str(server.port)]
|
||||
|
||||
|
||||
def _stream(cmd: list[str], log: Log | None, stdin: str | None = None, cwd=None, env=None) -> tuple[int, str]:
|
||||
proc = subprocess.Popen(cmd, stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL,
|
||||
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=cwd, env=env,
|
||||
errors="replace")
|
||||
if stdin is not None:
|
||||
proc.stdin.write(stdin)
|
||||
proc.stdin.close()
|
||||
out = []
|
||||
for line in proc.stdout:
|
||||
line = line.rstrip("\n")
|
||||
out.append(line)
|
||||
if log:
|
||||
log(line)
|
||||
return proc.wait(), "\n".join(out)
|
||||
|
||||
|
||||
def shell_vars(**values) -> str:
|
||||
"""Bash assignments for passing values into a remote script."""
|
||||
lines = []
|
||||
for k, v in values.items():
|
||||
if isinstance(v, (list, tuple)):
|
||||
lines.append(f"{k}=({' '.join(shlex.quote(str(x)) for x in v)})")
|
||||
else:
|
||||
lines.append(f"{k}={shlex.quote(str(v))}")
|
||||
return "\n".join(lines) + "\n"
|
||||
|
||||
|
||||
def run(server: Server, script: str, log: Log | None = None, check: bool = True, **values) -> str:
|
||||
"""Run a bash script on the server; values become shell variables in front of it."""
|
||||
body = "set -uo pipefail\n" + shell_vars(**values) + script
|
||||
rc, out = _stream(["ssh", *ssh_opts(server), server.target, "bash", "-s"], log, stdin=body)
|
||||
if check and rc != 0:
|
||||
raise RemoteError(f"remote command failed (exit {rc})" + (f": {out.splitlines()[-1]}" if out else ""))
|
||||
return out
|
||||
|
||||
|
||||
def run_local(cmd: str, cwd, log: Log | None, env: dict[str, str] | None = None) -> None:
|
||||
full_env = {**os.environ, **{k: os.path.expandvars(os.path.expanduser(v)) for k, v in (env or {}).items()}}
|
||||
rc, _ = _stream(["bash", "-c", cmd], log, cwd=cwd, env=full_env)
|
||||
if rc != 0:
|
||||
raise RemoteError(f"local command failed (exit {rc}): {cmd}")
|
||||
|
||||
|
||||
def upload(server: Server, src: str, dest: str, mode: str, log: Log | None = None) -> None:
|
||||
"""Copy one file. rsync writes a hidden temp file and renames it, so the site never lists a partial
|
||||
package; the server time becomes the publish time (the "new" badge of apps.jsp)."""
|
||||
cmd = ["rsync", "--checksum", "--perms", f"--chmod=F{mode}", "--mkpath", "--out-format=sent %n %b bytes",
|
||||
"-e", shlex.join(["ssh", *ssh_opts(server)]), src, f"{server.target}:{dest}"]
|
||||
rc, out = _stream(cmd, log)
|
||||
if rc != 0:
|
||||
raise RemoteError(f"rsync failed (exit {rc})" + (f": {out.splitlines()[-1]}" if out else ""))
|
||||
@@ -0,0 +1,466 @@
|
||||
"""Textual UI: app list with local and published versions, preview → confirm → publish, remove versions, add."""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import shlex
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
from rich.markup import escape
|
||||
from rich.text import Text
|
||||
from textual import on, work
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.binding import Binding
|
||||
from textual.containers import Horizontal, Vertical, VerticalScroll
|
||||
from textual.coordinate import Coordinate
|
||||
from textual.screen import ModalScreen
|
||||
from textual.theme import Theme
|
||||
from textual.widgets import Button, Checkbox, DataTable, Footer, Input, Label, OptionList, RichLog, Static, TextArea
|
||||
|
||||
from . import __version__, config, detect, pipeline, remote
|
||||
from .config import Config
|
||||
from .packages import Published
|
||||
from .pipeline import Options, Plan
|
||||
|
||||
# Classic CGA palette of Norton / Midnight Commander (same as Deployer)
|
||||
BLACK, BLUE, CYAN, GRAY, WHITE, YELLOW = "#000000", "#0000AA", "#00AAAA", "#AAAAAA", "#FFFFFF", "#FFFF55"
|
||||
|
||||
MIDNIGHT = Theme(
|
||||
name="midnight", primary=CYAN, secondary=BLUE, accent=YELLOW, warning=YELLOW, error="#FF5555",
|
||||
success="#55FF55", foreground=GRAY, background=BLUE, surface=BLUE, panel=CYAN, dark=True,
|
||||
variables={
|
||||
"border": GRAY, "border-blurred": GRAY,
|
||||
"block-cursor-foreground": BLACK, "block-cursor-background": CYAN, "block-cursor-text-style": "none",
|
||||
"block-cursor-blurred-foreground": BLACK, "block-cursor-blurred-background": CYAN,
|
||||
"block-cursor-blurred-text-style": "none", "block-hover-background": BLUE,
|
||||
"footer-background": BLACK, "footer-foreground": WHITE, "footer-key-foreground": WHITE,
|
||||
"footer-key-background": BLACK, "footer-description-foreground": BLACK,
|
||||
"footer-description-background": CYAN, "footer-item-background": BLACK,
|
||||
"input-cursor-background": BLACK, "input-cursor-foreground": CYAN,
|
||||
"scrollbar": CYAN, "scrollbar-background": BLUE, "scrollbar-corner-color": BLUE,
|
||||
},
|
||||
)
|
||||
|
||||
STATE_STYLE = {"newer": "bold #55FF55", "current": "#55FF55", "unpublished": "#FFFF55", "older": "#FFFF55",
|
||||
"error": "bold #FF5555", "no build": GRAY}
|
||||
ITEM_STYLE = {"new": ("+", "#55FF55"), "replace": ("~", "#FFFF55"), "meta": ("~", "#55FFFF"),
|
||||
"notes": ("~", "#55FFFF"), "delete": ("-", "bold #FF5555"), "same": ("=", GRAY),
|
||||
"skip": ("!", "#FFFF55")}
|
||||
|
||||
|
||||
def line_text(line: str) -> Text:
|
||||
style = ""
|
||||
if line.startswith("== "):
|
||||
style = "bold #55FFFF"
|
||||
elif line.startswith(("ERROR", "error:")) or "failed" in line.lower():
|
||||
style = "bold #FF5555"
|
||||
elif line.startswith("WARN"):
|
||||
style = "#FFFF55"
|
||||
return Text(line, style=style)
|
||||
|
||||
|
||||
def item_line(i: pipeline.Item, app: config.App) -> tuple[str, str]:
|
||||
sign, style = ITEM_STYLE[i.action]
|
||||
return f"{sign} {i.describe(app).replace(' -> ', ' → ', 1)}", style
|
||||
|
||||
|
||||
class ConfirmPlan(ModalScreen[bool]):
|
||||
BINDINGS = [Binding("escape,n", "dismiss(False)", "Cancel"), Binding("y", "dismiss(True)", "Publish")]
|
||||
|
||||
def __init__(self, plan: Plan):
|
||||
super().__init__()
|
||||
self.plan = plan
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
p = self.plan
|
||||
body = Text()
|
||||
body.append(f"{p.app.src}\n→ {p.server.target}:{p.server.apps_dir}\n", style="bold")
|
||||
if p.git_rev:
|
||||
body.append(f"git {p.git_rev}\n")
|
||||
body.append("\n")
|
||||
for i in p.items:
|
||||
text, style = item_line(i, p.app)
|
||||
body.append(text + "\n", style=style)
|
||||
if p.opts.notes.strip():
|
||||
body.append("\nWhat's new:\n", style="bold")
|
||||
body.append(p.opts.notes.strip() + "\n")
|
||||
for n in p.notes:
|
||||
body.append(f"! {n}\n", style="bold #FFFF55")
|
||||
with Vertical(id="dialog"):
|
||||
yield Label(Text(f" Publish {p.app.name} "), id="dialog-title")
|
||||
with VerticalScroll():
|
||||
yield Static(body)
|
||||
with Horizontal(id="buttons"):
|
||||
yield Button("Publish (y)", id="yes", variant="success")
|
||||
yield Button("Cancel (n)", id="no")
|
||||
|
||||
@on(Button.Pressed)
|
||||
def pressed(self, e: Button.Pressed) -> None:
|
||||
self.dismiss(e.button.id == "yes")
|
||||
|
||||
|
||||
class Ask(ModalScreen[bool]):
|
||||
BINDINGS = [Binding("escape,n", "dismiss(False)", "No"), Binding("y", "dismiss(True)", "Yes")]
|
||||
|
||||
def __init__(self, title: str, text: str):
|
||||
super().__init__()
|
||||
self.title_, self.text = title, text
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Vertical(id="dialog", classes="small"):
|
||||
yield Label(Text(f" {self.title_} "), id="dialog-title")
|
||||
with VerticalScroll():
|
||||
yield Static(self.text)
|
||||
with Horizontal(id="buttons"):
|
||||
yield Button("Yes (y)", id="yes", variant="warning")
|
||||
yield Button("No (n)", id="no")
|
||||
|
||||
@on(Button.Pressed)
|
||||
def pressed(self, e: Button.Pressed) -> None:
|
||||
self.dismiss(e.button.id == "yes")
|
||||
|
||||
|
||||
class PickVersion(ModalScreen[str | None]):
|
||||
BINDINGS = [Binding("escape", "dismiss(None)", "Cancel")]
|
||||
|
||||
def __init__(self, app_name: str, pub: list[Published]):
|
||||
super().__init__()
|
||||
self.app_name = app_name
|
||||
# one line per version: all its packages are removed together
|
||||
self.versions = list(dict.fromkeys(p.version for p in pub))
|
||||
self.lines = [f"v{v:12} " + ", ".join(f"{p.kind.label} {p.size / 1048576:.1f} MB"
|
||||
for p in pub if p.version == v) for v in self.versions]
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Vertical(id="dialog", classes="small"):
|
||||
yield Label(Text(f" {self.app_name}: published versions (Enter removes) "), id="dialog-title")
|
||||
yield OptionList(*self.lines)
|
||||
|
||||
@on(OptionList.OptionSelected)
|
||||
def picked(self, e: OptionList.OptionSelected) -> None:
|
||||
self.dismiss(self.versions[e.option_index])
|
||||
|
||||
|
||||
class AddApp(ModalScreen[config.App | None]):
|
||||
BINDINGS = [Binding("escape", "dismiss(None)", "Cancel")]
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
with Vertical(id="dialog"):
|
||||
yield Label(" New app ", id="dialog-title")
|
||||
yield Label("Local project dir:")
|
||||
yield Input(placeholder="~/Dev/Projects/android/MyApp", id="path")
|
||||
yield Label("Name on the site (empty = detect):")
|
||||
yield Input(id="name")
|
||||
yield Static("", id="toml")
|
||||
with Horizontal(id="buttons"):
|
||||
yield Button("Detect", id="detect")
|
||||
yield Button("Add", id="add", variant="success")
|
||||
yield Button("Cancel", id="cancel")
|
||||
|
||||
def build(self) -> config.App | None:
|
||||
raw = self.query_one("#path", Input).value.strip()
|
||||
path = Path(os.path.expanduser(raw)) if raw else None
|
||||
if not path or not path.is_dir():
|
||||
self.query_one("#toml", Static).update(Text("dir not found", style="bold #FF5555"))
|
||||
return None
|
||||
app = detect.new_app(path, self.query_one("#name", Input).value.strip() or None)
|
||||
self.query_one("#toml", Static).update(Text(config.app_to_toml(app)))
|
||||
return app
|
||||
|
||||
@on(Button.Pressed)
|
||||
def pressed(self, e: Button.Pressed) -> None:
|
||||
if e.button.id == "cancel":
|
||||
self.dismiss(None)
|
||||
elif e.button.id == "detect":
|
||||
self.build()
|
||||
else:
|
||||
app = self.build()
|
||||
if app:
|
||||
try:
|
||||
config.validate_app(app)
|
||||
except config.ConfigError as err:
|
||||
self.query_one("#toml", Static).update(Text(str(err), style="bold #FF5555"))
|
||||
return
|
||||
self.dismiss(app)
|
||||
|
||||
|
||||
class PublisherApp(App):
|
||||
TITLE = f"Publisher {__version__}"
|
||||
CSS = """
|
||||
Screen { background: $background; }
|
||||
#top { height: 1fr; }
|
||||
#apps { width: 3fr; border: double $border; border-title-color: $accent; }
|
||||
#apps > .datatable--header { background: $panel; color: #000000; text-style: bold; }
|
||||
#apps > .datatable--header-hover { background: $panel; color: #000000; }
|
||||
#apps > .datatable--header-cursor { background: $panel; color: #000000; }
|
||||
#side { width: 2fr; border: double $border; border-title-color: $accent; padding: 0 1; }
|
||||
#info { height: auto; margin-bottom: 1; }
|
||||
#side Checkbox { background: $background; border: none; padding: 0; height: 1; }
|
||||
#side Label { margin-top: 1; }
|
||||
#notes { height: 6; background: $background; border: round $border; }
|
||||
#log { height: 14; border: double $border; border-title-color: $accent; background: $background; }
|
||||
ModalScreen { align: center middle; background: $background 60%; }
|
||||
#dialog { width: 100; max-width: 95%; height: 80%; background: $panel; color: #000000;
|
||||
border: double #FFFFFF; padding: 0 1; }
|
||||
#dialog.small { height: auto; max-height: 80%; }
|
||||
#dialog VerticalScroll { height: 1fr; background: $background; color: $foreground; }
|
||||
#dialog.small VerticalScroll { height: auto; max-height: 20; }
|
||||
#dialog-title { width: 100%; content-align: center middle; color: #0000AA; text-style: bold; }
|
||||
#dialog Input { margin-bottom: 1; }
|
||||
#dialog OptionList { height: auto; max-height: 20; }
|
||||
#toml { height: auto; background: $background; color: $foreground; }
|
||||
#buttons { height: 3; align: center middle; }
|
||||
#buttons Button { margin: 0 2; }
|
||||
Toast { background: #000000; color: #FFFFFF; }
|
||||
"""
|
||||
BINDINGS = [
|
||||
Binding("f2", "edit", "Edit"),
|
||||
Binding("f3", "preview", "Preview"),
|
||||
Binding("f5", "publish", "Publish"),
|
||||
Binding("f6", "refresh", "Refresh"),
|
||||
Binding("f7", "add", "Add"),
|
||||
Binding("f8", "versions", "Versions"),
|
||||
Binding("f10", "quit", "Quit"),
|
||||
]
|
||||
|
||||
def __init__(self, cfg: Config):
|
||||
super().__init__()
|
||||
self.cfg = cfg
|
||||
self.status: dict[str, pipeline.Status] = {}
|
||||
self.busy = False
|
||||
|
||||
# ---------- layout ----------
|
||||
def compose(self) -> ComposeResult:
|
||||
with Horizontal(id="top"):
|
||||
yield DataTable(id="apps", cursor_type="row", zebra_stripes=False)
|
||||
with VerticalScroll(id="side"):
|
||||
yield Static(id="info")
|
||||
yield Checkbox("Build locally", True, id="o_build")
|
||||
yield Checkbox("Upload description / icon", True, id="o_meta")
|
||||
yield Checkbox("Delete versions beyond keep", True, id="o_prune")
|
||||
yield Checkbox("Force: older build, overwrite", False, id="o_force")
|
||||
yield Label("What's new in this version:")
|
||||
yield TextArea(id="notes", soft_wrap=True)
|
||||
yield RichLog(id="log", wrap=True, markup=False, max_lines=5000)
|
||||
yield Footer()
|
||||
|
||||
def on_mount(self) -> None:
|
||||
self.register_theme(MIDNIGHT)
|
||||
self.theme = "midnight"
|
||||
t = self.query_one("#apps", DataTable)
|
||||
t.border_title = f"Apps → {self.cfg.server.target}"
|
||||
t.add_columns("Name", "State", "Local", "Site")
|
||||
self.query_one("#side").border_title = "Details / options"
|
||||
self.query_one("#log").border_title = "Output"
|
||||
self.fill_table()
|
||||
self.action_refresh()
|
||||
|
||||
def fill_table(self) -> None:
|
||||
t = self.query_one("#apps", DataTable)
|
||||
row = t.cursor_row
|
||||
t.clear()
|
||||
for a in self.cfg.apps:
|
||||
s = self.status.get(a.name, pipeline.Status("…"))
|
||||
t.add_row(a.name, Text(s.state, style=STATE_STYLE.get(s.state, "")), s.local, s.remote, key=a.name)
|
||||
if self.cfg.apps:
|
||||
t.move_cursor(row=min(row, len(self.cfg.apps) - 1))
|
||||
self.show_details()
|
||||
|
||||
def current(self) -> config.App | None:
|
||||
t = self.query_one("#apps", DataTable)
|
||||
if not self.cfg.apps or t.row_count == 0:
|
||||
return None
|
||||
key = t.coordinate_to_cell_key(Coordinate(t.cursor_row, 0)).row_key.value
|
||||
return self.cfg.app(key)
|
||||
|
||||
@on(DataTable.RowHighlighted)
|
||||
def show_details(self) -> None:
|
||||
a = self.current()
|
||||
info = self.query_one("#info", Static)
|
||||
if not a:
|
||||
info.update("No apps: F7 add one")
|
||||
return
|
||||
lines = [f"[b]{escape(a.name)}[/b]", f"src: {escape(a.src)}"]
|
||||
lines += [f" {escape(g)}" for g in a.packages]
|
||||
if a.version_file:
|
||||
lines.append(f"version: {escape(a.version_file)}")
|
||||
if a.build:
|
||||
lines.append(f"build: {escape(a.build)}")
|
||||
meta = ", ".join(x for x in (a.info, a.icon) if x)
|
||||
if meta:
|
||||
lines.append(f"meta: {escape(meta)}")
|
||||
lines.append(f"keep: {self.cfg.keep(a) or 'all'}")
|
||||
info.update("\n".join(lines))
|
||||
self.query_one("#o_build", Checkbox).disabled = not a.build
|
||||
|
||||
def options(self) -> Options:
|
||||
v = lambda i: self.query_one(f"#{i}", Checkbox).value # noqa: E731
|
||||
return Options(build=v("o_build"), meta=v("o_meta"), prune=v("o_prune"), force=v("o_force"),
|
||||
notes=self.query_one("#notes", TextArea).text)
|
||||
|
||||
# ---------- output ----------
|
||||
def log_line(self, line: str) -> None:
|
||||
self.query_one("#log", RichLog).write(line_text(line))
|
||||
|
||||
def tlog(self, line: str) -> None:
|
||||
"""Log from a worker thread."""
|
||||
self.call_from_thread(self.log_line, line)
|
||||
|
||||
def start(self) -> bool:
|
||||
if self.busy:
|
||||
self.notify("Another operation is running", severity="warning")
|
||||
return False
|
||||
self.busy = True
|
||||
return True
|
||||
|
||||
def done(self) -> None:
|
||||
self.busy = False
|
||||
|
||||
# ---------- actions ----------
|
||||
def action_refresh(self) -> None:
|
||||
self.refresh_status()
|
||||
|
||||
@work(thread=True, group="status", exclusive=True)
|
||||
def refresh_status(self) -> None:
|
||||
try:
|
||||
st, others = pipeline.status_all(self.cfg)
|
||||
except remote.RemoteError as e:
|
||||
self.tlog(f"ERROR: status: {e}")
|
||||
return
|
||||
if others:
|
||||
self.tlog(f"published, not in {self.cfg.path.name}: {', '.join(others)}")
|
||||
self.call_from_thread(self._set_status, st)
|
||||
|
||||
def _set_status(self, st) -> None:
|
||||
self.status = st
|
||||
self.fill_table()
|
||||
|
||||
def action_preview(self) -> None:
|
||||
a = self.current()
|
||||
if a and self.start():
|
||||
self.do_preview(a, self.options(), then_publish=False)
|
||||
|
||||
def action_publish(self) -> None:
|
||||
a = self.current()
|
||||
if a and self.start():
|
||||
self.do_preview(a, self.options(), then_publish=True)
|
||||
|
||||
@work(thread=True, group="op")
|
||||
def do_preview(self, a: config.App, opts: Options, then_publish: bool) -> None:
|
||||
try:
|
||||
self.tlog(f"== preview {a.name}")
|
||||
plan = pipeline.preview(self.cfg, a, opts, log=self.tlog)
|
||||
except (pipeline.PublishError, remote.RemoteError, OSError) as e:
|
||||
self.tlog(f"ERROR: {e}")
|
||||
self.call_from_thread(self.done)
|
||||
return
|
||||
for i in plan.items:
|
||||
self.tlog(" " + item_line(i, a)[0])
|
||||
for n in plan.notes:
|
||||
self.tlog(f"WARN: {n}")
|
||||
if not then_publish or not plan.changes:
|
||||
if then_publish:
|
||||
self.tlog("nothing to publish")
|
||||
self.call_from_thread(self.done)
|
||||
return
|
||||
|
||||
def answered(ok: bool | None) -> None:
|
||||
if ok:
|
||||
self.do_execute(plan)
|
||||
else:
|
||||
self.log_line("cancelled")
|
||||
self.done()
|
||||
|
||||
self.call_from_thread(self.push_screen, ConfirmPlan(plan), answered)
|
||||
|
||||
@work(thread=True, group="op")
|
||||
def do_execute(self, plan: Plan) -> None:
|
||||
res = pipeline.execute(plan, log=self.tlog)
|
||||
self.call_from_thread(self.after_execute, plan, res)
|
||||
|
||||
def after_execute(self, plan: Plan, res: pipeline.Result) -> None:
|
||||
self.done()
|
||||
self.notify(f"{plan.app.name}: {res.message}", severity="information" if res.ok else "error", timeout=8)
|
||||
if res.ok:
|
||||
self.query_one("#notes", TextArea).text = ""
|
||||
self.action_refresh()
|
||||
|
||||
def action_versions(self) -> None:
|
||||
a = self.current()
|
||||
if a and self.start():
|
||||
self.load_versions(a)
|
||||
|
||||
@work(thread=True, group="op")
|
||||
def load_versions(self, a: config.App) -> None:
|
||||
try:
|
||||
pub = pipeline.versions(self.cfg, a)
|
||||
except remote.RemoteError as e:
|
||||
pub = []
|
||||
self.tlog(f"ERROR: {e}")
|
||||
self.call_from_thread(self._pick_version, a, pub)
|
||||
|
||||
def _pick_version(self, a: config.App, pub: list[Published]) -> None:
|
||||
if not pub:
|
||||
self.done()
|
||||
self.notify(f"{a.name}: nothing published", severity="warning")
|
||||
return
|
||||
|
||||
def picked(version: str | None) -> None:
|
||||
if not version:
|
||||
self.done()
|
||||
return
|
||||
|
||||
def confirmed(ok: bool | None) -> None:
|
||||
if ok:
|
||||
self.do_remove(a, version)
|
||||
else:
|
||||
self.done()
|
||||
self.push_screen(Ask("Remove", f"Delete {a.name} v{version} (all its packages and notes) "
|
||||
"from the site?"), confirmed)
|
||||
|
||||
self.push_screen(PickVersion(a.name, pub), picked)
|
||||
|
||||
@work(thread=True, group="op")
|
||||
def do_remove(self, a: config.App, version: str) -> None:
|
||||
res = pipeline.remove(self.cfg, a, version, log=self.tlog)
|
||||
self.tlog(f"== {res.message}")
|
||||
self.call_from_thread(self._removed, a, res)
|
||||
|
||||
def _removed(self, a: config.App, res: pipeline.Result) -> None:
|
||||
self.done()
|
||||
self.notify(f"{a.name}: {res.message}", severity="information" if res.ok else "error", timeout=8)
|
||||
self.action_refresh()
|
||||
|
||||
def action_edit(self) -> None:
|
||||
if self.busy:
|
||||
return
|
||||
editor = os.environ.get("VISUAL") or os.environ.get("EDITOR") or "nano"
|
||||
with self.suspend():
|
||||
subprocess.call([*shlex.split(editor), str(self.cfg.path)])
|
||||
self.reload()
|
||||
|
||||
def reload(self) -> None:
|
||||
try:
|
||||
self.cfg = config.load(self.cfg.path)
|
||||
except (OSError, config.ConfigError) as e:
|
||||
self.log_line(f"ERROR: config: {e} (old config kept)")
|
||||
return
|
||||
self.log_line(f"== config reloaded: {len(self.cfg.apps)} apps")
|
||||
self.fill_table()
|
||||
self.action_refresh()
|
||||
|
||||
def action_add(self) -> None:
|
||||
def added(app: config.App | None) -> None:
|
||||
if app:
|
||||
if any(x.name.lower() == app.name.lower() for x in self.cfg.apps):
|
||||
self.notify(f"{app.name} already exists", severity="error")
|
||||
return
|
||||
config.append_apps(self.cfg.path, [app])
|
||||
self.log_line(f"== added {app.name}: review it with F2, then F3 preview")
|
||||
self.reload()
|
||||
self.push_screen(AddApp(), added)
|
||||
|
||||
|
||||
def run(cfg: Config) -> int:
|
||||
PublisherApp(cfg).run()
|
||||
return 0
|
||||
Reference in New Issue
Block a user