chore: python code formatted with ruff, ruff formatter + linter, pyproject.toml now in root
C++ / fmt (pull_request) Successful in 4s
JS/TS / fmt (pull_request) Successful in 15s
JS/TS / lint (pull_request) Successful in 18s
Python / lint (pull_request) Failing after 32s
Python / fmt (pull_request) Failing after 37s
Python / test (pull_request) Successful in 52s
C++ / build (pull_request) Successful in 1m56s
Rust / fmt (pull_request) Successful in 1m22s
Rust / build (pull_request) Successful in 2m5s
Python / buildcheck (pull_request) Successful in 2m39s
Rust / clippy (pull_request) Successful in 2m1s
C++ / clang-tidy (pull_request) Successful in 3m46s

This commit is contained in:
2026-07-08 16:54:41 +02:00
parent 298cbbf424
commit 9300da258e
10 changed files with 1505 additions and 1315 deletions
+51 -33
View File
@@ -1,12 +1,14 @@
from __future__ import annotations from __future__ import annotations
import os import os
import sys import sys
from pathlib import Path from pathlib import Path
import typer import typer
from typer._completion_shared import install, _get_shell_name
from typer._completion_classes import completion_init from typer._completion_classes import completion_init
from zshell.subcommands import shell, scheme, screenshot, wallpaper, record from typer._completion_shared import _get_shell_name, install
from zshell.subcommands import record, scheme, screenshot, shell, wallpaper
app = typer.Typer(name="zshell-cli", add_completion=False) app = typer.Typer(name="zshell-cli", add_completion=False)
@@ -18,40 +20,56 @@ app.add_typer(record.app, name="record")
def _completion_installed() -> bool: def _completion_installed() -> bool:
shell = _get_shell_name() shell = _get_shell_name()
match shell: match shell:
case "zsh": case "zsh":
return (Path.home() / ".zfunc" / "_zshell-cli").exists() return (Path.home() / ".zfunc" / "_zshell-cli").exists()
case "bash": case "bash":
return (Path.home() / ".bash_completions" / "zshell-cli.sh").exists() return (
case "fish": Path.home() / ".bash_completions" / "zshell-cli.sh"
return (Path.home() / ".config" / "fish" / "completions" / "zshell-cli.fish").exists() ).exists()
return False case "fish":
return (
Path.home()
/ ".config"
/ "fish"
/ "completions"
/ "zshell-cli.fish"
).exists()
return False
def _install_completion() -> None: def _install_completion() -> None:
if _completion_installed(): if _completion_installed():
print("zshell-cli: Shell completion already installed.") print("zshell-cli: Shell completion already installed.")
sys.exit(0) sys.exit(0)
shell = _get_shell_name() shell = _get_shell_name()
if shell is None: if shell is None:
print("zshell-cli: Unable to detect shell type.", file=sys.stderr) print("zshell-cli: Unable to detect shell type.", file=sys.stderr)
sys.exit(1) sys.exit(1)
try: try:
_, path = install(prog_name="zshell-cli") _, path = install(prog_name="zshell-cli")
print(f"zshell-cli: Shell completion installed ({shell}: {path})") print(f"zshell-cli: Shell completion installed ({shell}: {path})")
print("zshell-cli: Restart your shell or source the file to enable tab-completion.") print(
except Exception as e: "zshell-cli: Restart your shell or source the file to enable tab-completion."
print(f"zshell-cli: Failed to install shell completion: {e}", file=sys.stderr) )
raise typer.Exit(code=1) except Exception as e:
print(
f"zshell-cli: Failed to install shell completion: {e}",
file=sys.stderr,
)
raise typer.Exit(code=1) from None
def main() -> None: def main() -> None:
if "--install-autocomplete" in sys.argv: if "--install-autocomplete" in sys.argv:
_install_completion() _install_completion()
return return
if "_ZSHELL_CLI_COMPLETE" in os.environ: if "_ZSHELL_CLI_COMPLETE" in os.environ:
completion_init() completion_init()
if sys.stdout.isatty() and not _completion_installed(): if sys.stdout.isatty() and not _completion_installed():
print("zshell-cli: Tip: run with --install-autocomplete for tab completion.", file=sys.stderr) print(
app() "zshell-cli: Tip: run with --install-autocomplete for tab completion.",
file=sys.stderr,
)
app()
+1 -1
View File
@@ -1,4 +1,4 @@
from zshell import main from zshell import main
if __name__ == "__main__": if __name__ == "__main__":
main() main()
+195 -135
View File
@@ -1,9 +1,9 @@
import os import contextlib
import json import json
import os
import subprocess import subprocess
import time import time
from pathlib import Path from pathlib import Path
from typing import Optional
import typer import typer
@@ -18,193 +18,253 @@ TEMP_RECORDING = STATE_DIR / "recording.mp4"
REPLAY_RECORDING = STATE_DIR / "replay.mp4" REPLAY_RECORDING = STATE_DIR / "replay.mp4"
NOTIF_ID_FILE = STATE_DIR / "notifid.txt" NOTIF_ID_FILE = STATE_DIR / "notifid.txt"
RECORDINGS_DIR = os.getenv("ZSHELL_RECORDINGS_DIR", str(Path(HOME) / "Videos/Recordings")) RECORDINGS_DIR = os.getenv(
"ZSHELL_RECORDINGS_DIR", str(Path(HOME) / "Videos/Recordings")
)
def _read_extra_args() -> list[str]: def _read_extra_args() -> list[str]:
try: try:
if CONFIG.is_file(): if CONFIG.is_file():
data = json.loads(CONFIG.read_text()) data = json.loads(CONFIG.read_text())
return data.get("record", {}).get("extraArgs", []) return data.get("record", {}).get("extraArgs", [])
except Exception: except Exception:
pass pass
return [] return []
def _is_recording() -> bool: def _is_recording() -> bool:
return subprocess.run(["pidof", RECORDER], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode == 0 return (
subprocess.run(
["pidof", RECORDER],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
).returncode
== 0
)
def _notify(summary: str, body: str = "", actions: list | None = None, timeout: int = 5000) -> Optional[int]: def _notify(
args = ["notify-send", summary, body, "-t", str(timeout), "-p"] summary: str,
if actions: body: str = "",
for action in actions: actions: list | None = None,
args.extend(["-A", action]) timeout: int = 5000,
try: ) -> int | None:
proc = subprocess.run(args, capture_output=True, text=True) args = ["notify-send", summary, body, "-t", str(timeout), "-p"]
return int(proc.stdout.strip()) if proc.stdout.strip().isdigit() else None if actions:
except Exception: for action in actions:
return None args.extend(["-A", action])
try:
proc = subprocess.run(args, capture_output=True, text=True)
return (
int(proc.stdout.strip()) if proc.stdout.strip().isdigit() else None
)
except Exception:
return None
def _close_notification(notif_id: int): def _close_notification(notif_id: int):
subprocess.run(["notify-send", "--close", str(notif_id)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) subprocess.run(
["notify-send", "--close", str(notif_id)],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
def _get_monitors() -> list[dict]: def _get_monitors() -> list[dict]:
try: try:
res = subprocess.run(["hyprctl", "monitors", "-j"], capture_output=True, text=True) res = subprocess.run(
return json.loads(res.stdout) ["hyprctl", "monitors", "-j"], capture_output=True, text=True
except Exception: )
return [] return json.loads(res.stdout)
except Exception:
return []
def _focused_monitor_name() -> Optional[str]: def _focused_monitor_name() -> str | None:
for m in _get_monitors(): for m in _get_monitors():
if m.get("focused"): if m.get("focused"):
return m["name"] return m["name"]
return None return None
def _monitors_intersecting_region(x: int, y: int, w: int, h: int) -> list[dict]: def _monitors_intersecting_region(x: int, y: int, w: int, h: int) -> list[dict]:
region = (x, y, x + w, y + h) region = (x, y, x + w, y + h)
intersecting = [] intersecting = []
for m in _get_monitors(): for m in _get_monitors():
mx, my, mw, mh = m["x"], m["y"], m["width"], m["height"] mx, my, mw, mh = m["x"], m["y"], m["width"], m["height"]
if not (region[2] <= mx or region[0] >= mx + mw or region[3] <= my or region[1] >= my + mh): if not (
intersecting.append(m) region[2] <= mx
return intersecting or region[0] >= mx + mw
or region[3] <= my
or region[1] >= my + mh
):
intersecting.append(m)
return intersecting
def _highest_refresh(monitors: list[dict]) -> float: def _highest_refresh(monitors: list[dict]) -> float:
return max((m["refreshRate"] for m in monitors), default=60.0) return max((m["refreshRate"] for m in monitors), default=60.0)
def _slurp_region() -> Optional[str]: def _slurp_region() -> str | None:
try: try:
return subprocess.check_output(["slurp", "-f", "%wx%h+%x+%y"], text=True).strip() return subprocess.check_output(
except subprocess.CalledProcessError: ["slurp", "-f", "%wx%h+%x+%y"], text=True
return None ).strip()
except subprocess.CalledProcessError:
return None
def _parse_geometry(geometry: str) -> Optional[tuple[int, int, int, int]]: def _parse_geometry(geometry: str) -> tuple[int, int, int, int] | None:
import re import re
match = re.match(r"(\d+)x(\d+)\+(\d+)\+(\d+)", geometry) match = re.match(r"(\d+)x(\d+)\+(\d+)\+(\d+)", geometry)
if match: if match:
return int(match.group(3)), int(match.group(4)), int(match.group(1)), int(match.group(2)) return (
return None int(match.group(3)),
int(match.group(4)),
int(match.group(1)),
int(match.group(2)),
)
return None
def start_recording(region: Optional[str], sound: bool): def start_recording(region: str | None, sound: bool):
STATE_DIR.mkdir(parents=True, exist_ok=True) STATE_DIR.mkdir(parents=True, exist_ok=True)
cmd = [RECORDER] cmd = [RECORDER]
extra_args = _read_extra_args() extra_args = _read_extra_args()
if region: if region:
if region.lower() == "slurp" or not region: if region.lower() == "slurp" or not region:
geometry = _slurp_region() geometry = _slurp_region()
if not geometry: if not geometry:
typer.echo("Region selection cancelled.") typer.echo("Region selection cancelled.")
raise typer.Abort() raise typer.Abort()
else: else:
geometry = region geometry = region
parsed = _parse_geometry(geometry) parsed = _parse_geometry(geometry)
if not parsed: if not parsed:
typer.echo("Invalid geometry format.") typer.echo("Invalid geometry format.")
raise typer.Abort() raise typer.Abort()
x, y, w, h = parsed x, y, w, h = parsed
monitors = _monitors_intersecting_region(x, y, w, h) monitors = _monitors_intersecting_region(x, y, w, h)
framerate = _highest_refresh(monitors) framerate = _highest_refresh(monitors)
cmd.extend(["-w", "region", "-region", geometry, "-f", str(int(framerate))]) cmd.extend(
["-w", "region", "-region", geometry, "-f", str(int(framerate))]
)
else: else:
monitor_name = _focused_monitor_name() monitor_name = _focused_monitor_name()
if not monitor_name: if not monitor_name:
typer.echo("No focused monitor found.") typer.echo("No focused monitor found.")
raise typer.Abort() raise typer.Abort()
monitors = _get_monitors() monitors = _get_monitors()
mon = next((m for m in monitors if m["name"] == monitor_name), None) mon = next((m for m in monitors if m["name"] == monitor_name), None)
rate = int(mon["refreshRate"]) if mon else 60 rate = int(mon["refreshRate"]) if mon else 60
cmd.extend(["-w", monitor_name, "-f", str(rate)]) cmd.extend(["-w", monitor_name, "-f", str(rate)])
if sound: if sound:
cmd.extend(["-a", "default_output"]) cmd.extend(["-a", "default_output"])
cmd.extend(extra_args) cmd.extend(extra_args)
cmd.extend(["-o", str(TEMP_RECORDING)]) cmd.extend(["-o", str(TEMP_RECORDING)])
subprocess.Popen(cmd, start_new_session=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) subprocess.Popen(
cmd,
start_new_session=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
notif_id = _notify("Recording started", f"Saving to {TEMP_RECORDING}") notif_id = _notify("Recording started", f"Saving to {TEMP_RECORDING}")
if notif_id is not None: if notif_id is not None:
NOTIF_ID_FILE.write_text(str(notif_id)) NOTIF_ID_FILE.write_text(str(notif_id))
time.sleep(1) time.sleep(1)
if not _is_recording(): if not _is_recording():
_notify("Recording failed", "Check gpu-screen-recorder output.", timeout=5000) _notify(
raise typer.Exit(code=1) "Recording failed",
"Check gpu-screen-recorder output.",
timeout=5000,
)
raise typer.Exit(code=1)
def stop_recording(clipboard: bool): def stop_recording(clipboard: bool):
subprocess.run(["pkill", "-f", RECORDER], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) subprocess.run(
["pkill", "-f", RECORDER],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
for _ in range(50): for _ in range(50):
if not _is_recording(): if not _is_recording():
break break
time.sleep(0.1) time.sleep(0.1)
dest_dir = Path(RECORDINGS_DIR) dest_dir = Path(RECORDINGS_DIR)
dest_dir.mkdir(parents=True, exist_ok=True) dest_dir.mkdir(parents=True, exist_ok=True)
timestamp = time.strftime("%Y-%m-%d_%H-%M-%S") timestamp = time.strftime("%Y-%m-%d_%H-%M-%S")
final_path = dest_dir / f"recording_{timestamp}.mp4" final_path = dest_dir / f"recording_{timestamp}.mp4"
if TEMP_RECORDING.exists(): if TEMP_RECORDING.exists():
TEMP_RECORDING.rename(final_path) TEMP_RECORDING.rename(final_path)
if NOTIF_ID_FILE.is_file(): if NOTIF_ID_FILE.is_file():
try: with contextlib.suppress(Exception):
_close_notification(int(NOTIF_ID_FILE.read_text().strip())) _close_notification(int(NOTIF_ID_FILE.read_text().strip()))
except Exception: NOTIF_ID_FILE.unlink()
pass
NOTIF_ID_FILE.unlink()
if clipboard: if clipboard:
subprocess.run( subprocess.run(
["wl-copy", "--type", "text/uri-list", f"file://{final_path}"], ["wl-copy", "--type", "text/uri-list", f"file://{final_path}"],
stdout=subprocess.DEVNULL, stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
) )
_notify("Recording stopped", f"Saved to {final_path}", timeout=5000) _notify("Recording stopped", f"Saved to {final_path}", timeout=5000)
def toggle_pause(): def toggle_pause():
subprocess.run(["pkill", "-USR2", "-f", RECORDER], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) subprocess.run(
typer.echo("Toggled pause.") ["pkill", "-USR2", "-f", RECORDER],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
typer.echo("Toggled pause.")
@app.command() @app.command()
def record( def record(
region: Optional[str] = typer.Option( region: str | None = typer.Option(
None, None,
"--region", "--region",
"-r", "-r",
help="Record a region. Use 'slurp' (or omit value) to select interactively, or give 'WxH+X+Y'.", help="Record a region. Use 'slurp' (or omit value) to select interactively, or give 'WxH+X+Y'.",
), ),
sound: bool = typer.Option(False, "--sound", "-s", help="Record audio from default output."), sound: bool = typer.Option(
pause: bool = typer.Option(False, "--pause", "-p", help="Toggle pause/resume."), False, "--sound", "-s", help="Record audio from default output."
clipboard: bool = typer.Option(False, "--clipboard", "-c", help="Copy the final recording path to clipboard."), ),
pause: bool = typer.Option(
False, "--pause", "-p", help="Toggle pause/resume."
),
clipboard: bool = typer.Option(
False,
"--clipboard",
"-c",
help="Copy the final recording path to clipboard.",
),
): ):
"""Start or stop a screen recording with gpu-screen-recorder.""" """Start or stop a screen recording with gpu-screen-recorder."""
if pause: if pause:
toggle_pause() toggle_pause()
raise typer.Exit() raise typer.Exit()
if _is_recording(): if _is_recording():
stop_recording(clipboard) stop_recording(clipboard)
else: else:
start_recording(region, sound) start_recording(region, sound)
File diff suppressed because it is too large Load Diff
+3 -2
View File
@@ -1,4 +1,5 @@
import subprocess import subprocess
import typer import typer
args = ["qs", "-c", "zshell"] args = ["qs", "-c", "zshell"]
@@ -8,9 +9,9 @@ app = typer.Typer()
@app.command() @app.command()
def start(): def start():
subprocess.run(args + ["ipc"] + ["call"] + ["picker"] + ["open"], check=True) subprocess.run([*args, "ipc", "call", "picker", "open"], check=True)
@app.command() @app.command()
def start_freeze(): def start_freeze():
subprocess.run(args + ["ipc"] + ["call"] + ["picker"] + ["openFreeze"], check=True) subprocess.run([*args, "ipc", "call", "picker", "openFreeze"], check=True)
+55 -47
View File
@@ -11,76 +11,84 @@ app = typer.Typer()
@app.command() @app.command()
def kill(): def kill():
result = subprocess.run(args + ["kill"], capture_output=True) result = subprocess.run([*args, "kill"], capture_output=True)
if result.returncode != 0: if result.returncode != 0:
sys.stderr.write("No running instance to kill.\n") sys.stderr.write("No running instance to kill.\n")
sys.exit(1) sys.exit(1)
sys.stderr.write(result.stderr.decode()) sys.stderr.write(result.stderr.decode())
def start_instance(no_daemon: bool = False) -> None: def start_instance(no_daemon: bool = False) -> None:
result = subprocess.run(args + ["-n"] + ([] if no_daemon else ["-d"]), capture_output=True) result = subprocess.run(
stdout = result.stdout.decode().strip() args + ["-n"] + ([] if no_daemon else ["-d"]), capture_output=True
if stdout: )
if "already running" in stdout.lower(): stdout = result.stdout.decode().strip()
sys.stderr.write(stdout + "\n") if stdout and "already running" in stdout.lower():
sys.exit(1) sys.stderr.write(stdout + "\n")
if result.returncode != 0: sys.exit(1)
stderr = result.stderr.decode().strip() if result.returncode != 0:
sys.stderr.write(stderr + "\n") stderr = result.stderr.decode().strip()
sys.exit(1) sys.stderr.write(stderr + "\n")
sys.exit(1)
@app.command() @app.command()
def start(no_daemon: bool = False): def start(no_daemon: bool = False):
start_instance(no_daemon) start_instance(no_daemon)
@app.command() @app.command()
def restart(no_daemon: bool = False): def restart(no_daemon: bool = False):
subprocess.run(args + ["kill"], capture_output=True) subprocess.run([*args, "kill"], capture_output=True)
deadline = time.monotonic() + 2.5 deadline = time.monotonic() + 2.5
while time.monotonic() < deadline: while time.monotonic() < deadline:
result = subprocess.run(args + ["kill"], capture_output=True) result = subprocess.run([*args, "kill"], capture_output=True)
if result.returncode == 255: if result.returncode == 255:
break break
time.sleep(0.25) time.sleep(0.25)
start_instance(no_daemon=no_daemon) start_instance(no_daemon=no_daemon)
@app.command() @app.command()
def show(): def show():
result = subprocess.run(args + ["ipc"] + ["show"], capture_output=True) result = subprocess.run([*args, "ipc", "show"], capture_output=True)
if result.returncode != 0: if result.returncode != 0:
sys.stderr.write(result.stderr.decode()) sys.stderr.write(result.stderr.decode())
sys.exit(1) sys.exit(1)
sys.stdout.write(result.stdout.decode()) sys.stdout.write(result.stdout.decode())
sys.stderr.write(result.stderr.decode()) sys.stderr.write(result.stderr.decode())
@app.command() @app.command()
def log(): def log():
result = subprocess.run(args + ["log"], capture_output=True) result = subprocess.run([*args, "log"], capture_output=True)
if result.returncode != 0: if result.returncode != 0:
sys.stderr.write(result.stderr.decode()) sys.stderr.write(result.stderr.decode())
sys.exit(1) sys.exit(1)
sys.stdout.write(result.stdout.decode()) sys.stdout.write(result.stdout.decode())
sys.stderr.write(result.stderr.decode()) sys.stderr.write(result.stderr.decode())
@app.command() @app.command()
def lock(): def lock():
result = subprocess.run(args + ["ipc"] + ["call"] + ["lock"] + ["lock"], capture_output=True) result = subprocess.run(
if result.returncode != 0: [*args, "ipc", "call", "lock", "lock"], capture_output=True
sys.stderr.write(result.stderr.decode()) )
sys.exit(1) if result.returncode != 0:
sys.stderr.write(result.stderr.decode()) sys.stderr.write(result.stderr.decode())
sys.exit(1)
sys.stderr.write(result.stderr.decode())
@app.command() @app.command()
def call(target: str, method: str, method_args: list[str] = typer.Argument(None)): def call(
result = subprocess.run(args + ["ipc"] + ["call"] + [target] + [method] + (method_args or []), capture_output=True) target: str, method: str, method_args: list[str] = typer.Argument(None)
if result.returncode != 0: ):
sys.stderr.write(result.stderr.decode()) result = subprocess.run(
sys.exit(1) args + ["ipc"] + ["call"] + [target] + [method] + (method_args or []),
sys.stderr.write(result.stderr.decode()) capture_output=True,
)
if result.returncode != 0:
sys.stderr.write(result.stderr.decode())
sys.exit(1)
sys.stderr.write(result.stderr.decode())
+28 -25
View File
@@ -1,9 +1,9 @@
import subprocess import subprocess
import typer
from typing import Annotated
from PIL import Image, ImageFilter
from pathlib import Path from pathlib import Path
from typing import Annotated
import typer
from PIL import Image, ImageFilter
args = ["qs", "-c", "zshell"] args = ["qs", "-c", "zshell"]
@@ -12,32 +12,35 @@ app = typer.Typer()
@app.command() @app.command()
def set(wallpaper: Path): def set(wallpaper: Path):
subprocess.run(args + ["ipc"] + ["call"] + ["wallpaper"] + ["set"] + [wallpaper], check=True) subprocess.run(
[*args, "ipc", "call", "wallpaper", "set", wallpaper],
check=True,
)
@app.command() @app.command()
def lockscreen( def lockscreen(
input_image: Annotated[ input_image: Annotated[
Path, Path,
typer.Option(), typer.Option(),
], ],
output_path: Annotated[ output_path: Annotated[
Path, Path,
typer.Option(), typer.Option(),
], ],
blur_amount: int = 20, blur_amount: int = 20,
): ):
img = Image.open(input_image) img = Image.open(input_image)
size = img.size size = img.size
if blur_amount == 0: if blur_amount == 0:
img.save(output_path, "PNG") img.save(output_path, "PNG")
return return
if size[0] < 3840 or size[1] < 2160: if size[0] < 3840 or size[1] < 2160:
img = img.resize((size[0] // 2, size[1] // 2), Image.Resampling.NEAREST) img = img.resize((size[0] // 2, size[1] // 2), Image.Resampling.NEAREST)
else: else:
img = img.resize((size[0] // 4, size[1] // 4), Image.Resampling.NEAREST) img = img.resize((size[0] // 4, size[1] // 4), Image.Resampling.NEAREST)
img = img.filter(ImageFilter.GaussianBlur(blur_amount)) img = img.filter(ImageFilter.GaussianBlur(blur_amount))
img.save(output_path, "PNG") img.save(output_path, "PNG")
+98 -95
View File
@@ -10,137 +10,140 @@ ASSETS: Traversable = files("zshell") / "assets" / "schemes"
@dataclass(frozen=True) @dataclass(frozen=True)
class SchemeVariant: class SchemeVariant:
id: str id: str
name: str name: str
modes: frozenset[str] modes: frozenset[str]
accents: tuple[str, ...] = () accents: tuple[str, ...] = ()
@dataclass(frozen=True) @dataclass(frozen=True)
class SchemeMeta: class SchemeMeta:
id: str id: str
name: str name: str
variants: tuple[SchemeVariant, ...] variants: tuple[SchemeVariant, ...]
@dataclass @dataclass
class Palette: class Palette:
colors: dict[str, str] colors: dict[str, str]
mode: str mode: str
scheme: str scheme: str
variant: str variant: str
accent: str | None = None accent: str | None = None
def _parse_txt(path: Traversable) -> dict[str, str]: def _parse_txt(path: Traversable) -> dict[str, str]:
colors: dict[str, str] = {} colors: dict[str, str] = {}
for line in path.read_text().splitlines(): for line in path.read_text().splitlines():
line = line.strip() line = line.strip()
if not line or line.startswith("#"): if not line or line.startswith("#"):
continue continue
parts = line.split(None, 1) parts = line.split(None, 1)
if len(parts) == 2: if len(parts) == 2:
key, val = parts key, val = parts
colors[key] = f"#{val}" if not val.startswith("#") else val colors[key] = f"#{val}" if not val.startswith("#") else val
return colors return colors
def _discover_schemes() -> dict[str, SchemeMeta]: def _discover_schemes() -> dict[str, SchemeMeta]:
schemes: dict[str, SchemeMeta] = {} schemes: dict[str, SchemeMeta] = {}
for scheme_dir in sorted(ASSETS.iterdir(), key=lambda p: p.name): for scheme_dir in sorted(ASSETS.iterdir(), key=lambda p: p.name):
if not scheme_dir.is_dir() or scheme_dir.name.startswith("."): if not scheme_dir.is_dir() or scheme_dir.name.startswith("."):
continue continue
sid = scheme_dir.name sid = scheme_dir.name
display_name = sid.capitalize() display_name = sid.capitalize()
variants: list[SchemeVariant] = [] variants: list[SchemeVariant] = []
for var_dir in sorted(scheme_dir.iterdir(), key=lambda p: p.name): for var_dir in sorted(scheme_dir.iterdir(), key=lambda p: p.name):
if not var_dir.is_dir() or var_dir.name.startswith("."): if not var_dir.is_dir() or var_dir.name.startswith("."):
continue continue
modes: set[str] = set() modes: set[str] = set()
accents: set[str] = set() accents: set[str] = set()
for f in var_dir.iterdir(): for f in var_dir.iterdir():
name = PurePosixPath(f.name) name = PurePosixPath(f.name)
if name.suffix != ".txt": if name.suffix != ".txt":
continue continue
stem = name.stem stem = name.stem
if "-" in stem: if "-" in stem:
maybe_accent, maybe_mode = stem.rsplit("-", 1) maybe_accent, maybe_mode = stem.rsplit("-", 1)
if maybe_mode in ("dark", "light"): if maybe_mode in ("dark", "light"):
modes.add(maybe_mode) modes.add(maybe_mode)
accents.add(maybe_accent) accents.add(maybe_accent)
else: else:
modes.add(stem) modes.add(stem)
else: else:
if stem in ("dark", "light"): if stem in ("dark", "light"):
modes.add(stem) modes.add(stem)
if modes: if modes:
vname = var_dir.name.capitalize() vname = var_dir.name.capitalize()
variants.append( variants.append(
SchemeVariant( SchemeVariant(
id=var_dir.name, id=var_dir.name,
name=vname, name=vname,
modes=frozenset(modes), modes=frozenset(modes),
accents=tuple(sorted(accents)), accents=tuple(sorted(accents)),
) )
) )
schemes[sid] = SchemeMeta( schemes[sid] = SchemeMeta(
id=sid, id=sid,
name=display_name, name=display_name,
variants=tuple(variants), variants=tuple(variants),
) )
return schemes return schemes
SCHEMES: dict[str, SchemeMeta] = _discover_schemes() SCHEMES: dict[str, SchemeMeta] = _discover_schemes()
def get_palette(scheme: str, variant: str, mode: str, accent: str | None = None) -> Palette: def get_palette(
if scheme not in SCHEMES: scheme: str, variant: str, mode: str, accent: str | None = None
raise KeyError( ) -> Palette:
f"Unknown scheme '{scheme}'. Available: {', '.join(SCHEMES)}") if scheme not in SCHEMES:
raise KeyError(
f"Unknown scheme '{scheme}'. Available: {', '.join(SCHEMES)}"
)
meta = SCHEMES[scheme] meta = SCHEMES[scheme]
var_ids = {v.id for v in meta.variants} var_ids = {v.id for v in meta.variants}
if variant not in var_ids: if variant not in var_ids:
raise KeyError( raise KeyError(
f"Unknown variant '{variant}' for scheme '{scheme}'. Available: {', '.join(sorted(var_ids))}") f"Unknown variant '{variant}' for scheme '{scheme}'. Available: {', '.join(sorted(var_ids))}"
)
if accent: filename = f"{accent}-{mode}.txt" if accent else f"{mode}.txt"
filename = f"{accent}-{mode}.txt"
else:
filename = f"{mode}.txt"
txt_path = ASSETS / scheme / variant / filename txt_path = ASSETS / scheme / variant / filename
if not txt_path.is_file(): if not txt_path.is_file():
txt_path = ASSETS / scheme / variant / f"{mode}.txt" txt_path = ASSETS / scheme / variant / f"{mode}.txt"
if not txt_path.is_file(): if not txt_path.is_file():
var_info = next(v for v in meta.variants if v.id == variant) var_info = next(v for v in meta.variants if v.id == variant)
raise FileNotFoundError( raise FileNotFoundError(
f"No {mode} palette for '{scheme}:{variant}'. Available modes: {sorted(var_info.modes)}" f"No {mode} palette for '{scheme}:{variant}'. Available modes: {sorted(var_info.modes)}"
) )
colors = _parse_txt(txt_path) colors = _parse_txt(txt_path)
return Palette(colors=colors, mode=mode, scheme=scheme, variant=variant, accent=accent) return Palette(
colors=colors, mode=mode, scheme=scheme, variant=variant, accent=accent
)
def list_schemes() -> dict[str, SchemeMeta]: def list_schemes() -> dict[str, SchemeMeta]:
return dict(SCHEMES) return dict(SCHEMES)
def resolve_preset(spec: str) -> tuple[str, str]: def resolve_preset(spec: str) -> tuple[str, str]:
parts = spec.split(":") parts = spec.split(":")
if len(parts) == 2: if len(parts) == 2:
return parts[0], parts[1] return parts[0], parts[1]
if len(parts) == 1: if len(parts) == 1:
return parts[0], "default" return parts[0], "default"
raise ValueError(f"Invalid preset spec '{spec}'. Use <scheme>:<variant>") raise ValueError(f"Invalid preset spec '{spec}'. Use <scheme>:<variant>")
+27 -5
View File
@@ -21,17 +21,39 @@ source = "vcs"
[tool.hatch.build] [tool.hatch.build]
include = [ include = [
"src/zshell/assets/**", "cli/src/zshell/assets/**",
] ]
[tool.hatch.build.targets.wheel]
packages = ["cli/src/zshell"]
[tool.hatch.build.targets.sdist] [tool.hatch.build.targets.sdist]
only-include = [ only-include = [
"src", "cli/src",
] ]
[tool.ruff] [tool.ruff]
line-length = 120 line-length = 80
[tool.ruff.format]
quote-style = "double"
indent-style = "tab"
line-ending = "lf"
docstring-code-format = true
docstring-code-line-length = "dynamic"
[tool.ruff.lint]
ignore = ["E501", "B008"]
select = [
"E",
"F",
"I",
"UP",
"B",
"SIM",
"RUF",
]
[tool.pytest.ini_options] [tool.pytest.ini_options]
testpaths = ["tests"] testpaths = ["cli/tests"]
pythonpath = ["src"] pythonpath = ["cli/src"]
+352 -307
View File
@@ -4,22 +4,24 @@ import json
import re import re
import sys import sys
from collections import defaultdict from collections import defaultdict
from functools import lru_cache from functools import cache
from pathlib import Path from pathlib import Path
@lru_cache(maxsize=None) @cache
def read_lines(path: Path) -> tuple[str, ...]: def read_lines(path: Path) -> tuple[str, ...]:
return tuple(path.read_text().splitlines()) return tuple(path.read_text().splitlines())
ROW_RE = re.compile( ROW_RE = re.compile(
r'^\s*(ToggleRow|SliderRow|SelectRow|SpinRow|NavRow|InfoRow|PopupRow|OverlayRow|DefaultRow)\s*\{') r"^\s*(ToggleRow|SliderRow|SelectRow|SpinRow|NavRow|InfoRow|PopupRow|OverlayRow|DefaultRow)\s*\{"
)
LABEL_RE = re.compile(r'^\s*(?:label|text):\s*qsTr\("([^"]+)"\)') LABEL_RE = re.compile(r'^\s*(?:label|text):\s*qsTr\("([^"]+)"\)')
ANCHOR_RE = re.compile(r'^\s*settingAnchor:\s*"([^"]+)"') ANCHOR_RE = re.compile(r'^\s*settingAnchor:\s*"([^"]+)"')
CHECKED_RE = re.compile(r'^\s*checked:\s*(?:Config)\.([\w.]+)\s*$') CHECKED_RE = re.compile(r"^\s*checked:\s*(?:Config)\.([\w.]+)\s*$")
ONTOGGLED_RE = re.compile( ONTOGGLED_RE = re.compile(
r'^\s*onToggled:\s*(?:Config)\.([\w.]+)\s*=\s*checked\s*$') r"^\s*onToggled:\s*(?:Config)\.([\w.]+)\s*=\s*checked\s*$"
)
ICON_RE = re.compile(r'^\s*icon:\s*"([^"]+)"') ICON_RE = re.compile(r'^\s*icon:\s*"([^"]+)"')
SKIP_LABELS = {"Muted", "None"} SKIP_LABELS = {"Muted", "None"}
FIELD_WEIGHT = {"title": 1.0, "keywords": 0.4} FIELD_WEIGHT = {"title": 1.0, "keywords": 0.4}
@@ -27,14 +29,14 @@ STOPWORDS = {"the", "a", "an", "of", "and", "or", "to", "on", "in", "for"}
def find_pages_dir(settings: Path) -> Path: def find_pages_dir(settings: Path) -> Path:
return settings / "Pages" return settings / "Pages"
def discover_files(settings: Path) -> dict[str, Path]: def discover_files(settings: Path) -> dict[str, Path]:
files: dict[str, Path] = {} files: dict[str, Path] = {}
for p in find_pages_dir(settings).rglob("*.qml"): for p in find_pages_dir(settings).rglob("*.qml"):
files[p.stem] = p files[p.stem] = p
return files return files
PAGE_NAME_RE = re.compile(r'^\s*name:\s*qsTr\("([^"]+)"\)') PAGE_NAME_RE = re.compile(r'^\s*name:\s*qsTr\("([^"]+)"\)')
@@ -42,366 +44,409 @@ PAGE_ICON_RE = re.compile(r'^\s*icon:\s*"([^"]+)"')
def parse_page_registry(settings: Path) -> list[tuple[str, str]]: def parse_page_registry(settings: Path) -> list[tuple[str, str]]:
text = (settings / "PageRegistry.qml").read_text().splitlines() text = (settings / "PageRegistry.qml").read_text().splitlines()
start = next( start = next(
i for i, line in enumerate(text) i for i, line in enumerate(text) if re.search(r"\bpages\s*:\s*\[", line)
if re.search(r'\bpages\s*:\s*\[', line) )
)
out: list[tuple[str, str]] = [] out: list[tuple[str, str]] = []
i = start + 1 i = start + 1
while i < len(text): while i < len(text):
line = text[i].strip() line = text[i].strip()
if line.startswith("]"): if line.startswith("]"):
break break
if line.startswith("//") or not line: if line.startswith("//") or not line:
i += 1 i += 1
continue continue
if line.startswith("{"): if line.startswith("{"):
name = None name = None
icon = None icon = None
i += 1 i += 1
while i < len(text): while i < len(text):
s = text[i].strip() s = text[i].strip()
if s.startswith("}"): if s.startswith("}"):
if name is not None: if name is not None:
out.append((icon or "tune", name)) out.append((icon or "tune", name))
break break
if name is None: if name is None:
m = PAGE_NAME_RE.match(text[i]) m = PAGE_NAME_RE.match(text[i])
if m: if m:
name = m.group(1) name = m.group(1)
if icon is None: if icon is None:
mi = PAGE_ICON_RE.match(text[i]) mi = PAGE_ICON_RE.match(text[i])
if mi: if mi:
icon = mi.group(1) icon = mi.group(1)
i += 1 i += 1
i += 1 i += 1
return out return out
BLOCK_RE = re.compile(r'^\s*([A-Za-z_][A-Za-z0-9_]*)\s*\{\s*$') BLOCK_RE = re.compile(r"^\s*([A-Za-z_][A-Za-z0-9_]*)\s*\{\s*$")
def _strip_comment(line: str) -> str: def _strip_comment(line: str) -> str:
return line.split("//", 1)[0].rstrip() return line.split("//", 1)[0].rstrip()
def parse_block(lines: list[str], i: int) -> tuple[str, list[tuple[str, list]], int]: def parse_block(
line = _strip_comment(lines[i]).strip() lines: list[str], i: int
m = BLOCK_RE.match(line) ) -> tuple[str, list[tuple[str, list]], int]:
if not m: line = _strip_comment(lines[i]).strip()
raise ValueError(f"Expected block start at line {i + 1}: {lines[i]!r}") m = BLOCK_RE.match(line)
if not m:
raise ValueError(f"Expected block start at line {i + 1}: {lines[i]!r}")
name = m.group(1) name = m.group(1)
i += 1 i += 1
children: list[tuple[str, list]] = [] children: list[tuple[str, list]] = []
while i < len(lines): while i < len(lines):
s = _strip_comment(lines[i]).strip() s = _strip_comment(lines[i]).strip()
if not s: if not s:
i += 1 i += 1
continue continue
if s.startswith("}"): if s.startswith("}"):
return name, children, i + 1 return name, children, i + 1
if BLOCK_RE.match(s): if BLOCK_RE.match(s):
child_name, child_children, i = parse_block(lines, i) child_name, child_children, i = parse_block(lines, i)
children.append((child_name, child_children)) children.append((child_name, child_children))
continue continue
i += 1 i += 1
raise ValueError(f"Unterminated block: {name}") raise ValueError(f"Unterminated block: {name}")
def collect_page_names(block: tuple[str, list[tuple[str, list]]]) -> list[str]: def collect_page_names(block: tuple[str, list[tuple[str, list]]]) -> list[str]:
name, children = block name, children = block
if name != "Component": if name != "Component":
return [name] return [name]
for child_name, child_children in children: for child_name, child_children in children:
if child_name == "StackPage": if child_name == "StackPage":
out: list[str] = [] out: list[str] = []
for grand_name, grand_children in child_children: for grand_name, grand_children in child_children:
if grand_name == "Component": if grand_name == "Component":
out.extend(collect_page_names((grand_name, grand_children))) out.extend(collect_page_names((grand_name, grand_children)))
return out return out
if child_name != "Component": if child_name != "Component":
return [child_name] return [child_name]
return [] return []
def parse_page_comps(settings: Path) -> list[list[str]]: def parse_page_comps(settings: Path) -> list[list[str]]:
text = (settings / "PageCompRegistry.qml").read_text().splitlines() text = (settings / "PageCompRegistry.qml").read_text().splitlines()
start = next( start = next(
i for i, line in enumerate(text) i
if re.search(r'\bpageComps\s*:\s*\[', _strip_comment(line)) for i, line in enumerate(text)
) if re.search(r"\bpageComps\s*:\s*\[", _strip_comment(line))
)
comps: list[list[str]] = [] comps: list[list[str]] = []
i = start + 1 i = start + 1
while i < len(text): while i < len(text):
s = _strip_comment(text[i]).strip() s = _strip_comment(text[i]).strip()
if not s: if not s:
i += 1 i += 1
continue continue
if s.startswith("]"): if s.startswith("]"):
break break
if BLOCK_RE.match(s) and BLOCK_RE.match(s).group(1) == "Component": if BLOCK_RE.match(s) and BLOCK_RE.match(s).group(1) == "Component":
block = parse_block(text, i) block = parse_block(text, i)
names = collect_page_names((block[0], block[1])) names = collect_page_names((block[0], block[1]))
if names: if names:
comps.append(names) comps.append(names)
i = block[2] i = block[2]
continue continue
i += 1 i += 1
return comps return comps
def dedup_crumbs(labels: list[str], icons: list[str]) -> tuple[list[str], list[str]]: def dedup_crumbs(
out_labels: list[str] = [] labels: list[str], icons: list[str]
out_icons: list[str] = [] ) -> tuple[list[str], list[str]]:
for lbl, ico in zip(labels, icons): out_labels: list[str] = []
if out_labels and out_labels[-1] == lbl: out_icons: list[str] = []
continue for lbl, ico in zip(labels, icons, strict=False):
out_labels.append(lbl) if out_labels and out_labels[-1] == lbl:
out_icons.append(ico) continue
return out_labels, out_icons out_labels.append(lbl)
out_icons.append(ico)
return out_labels, out_icons
def build_nav_map(settings: Path, files: dict[str, Path]) -> dict[str, dict]: def build_nav_map(settings: Path, files: dict[str, Path]) -> dict[str, dict]:
comps = parse_page_comps(settings) comps = parse_page_comps(settings)
registry = parse_page_registry(settings) registry = parse_page_registry(settings)
top_meta: dict[int, tuple[str, str]] = {} top_meta: dict[int, tuple[str, str]] = {}
for i, (icon, label) in enumerate(registry): for i, (icon, label) in enumerate(registry):
top_meta[i] = (icon, label) top_meta[i] = (icon, label)
nav_children: dict[str, dict[int, tuple[str, str, str]]] = {} nav_children: dict[str, dict[int, tuple[str, str, str]]] = {}
for names in comps: for names in comps:
for name in names: for name in names:
pf = files.get(name) pf = files.get(name)
if not pf: if not pf:
continue continue
pending_icon = pending_label = None pending_icon = pending_label = None
section = "" section = ""
expect_section = False expect_section = False
for ln in read_lines(pf): for ln in read_lines(pf):
if SECTION_RE.match(ln): if SECTION_RE.match(ln):
expect_section = True expect_section = True
continue continue
ml = LABEL_RE.match(ln) ml = LABEL_RE.match(ln)
if ml: if ml:
if expect_section: if expect_section:
section = ml.group(1) section = ml.group(1)
expect_section = False expect_section = False
else: else:
pending_label = ml.group(1) pending_label = ml.group(1)
continue continue
mi = ICON_RE.match(ln) mi = ICON_RE.match(ln)
if mi: if mi:
pending_icon = mi.group(1) pending_icon = mi.group(1)
mo = re.search(r"openSubPage\((\d+)\)", ln) mo = re.search(r"openSubPage\((\d+)\)", ln)
if mo: if mo:
pos = int(mo.group(1)) pos = int(mo.group(1))
nav_children.setdefault(name, {})[pos] = ( nav_children.setdefault(name, {})[pos] = (
pending_icon or "tune", pending_label or "", section) pending_icon or "tune",
pending_icon = pending_label = None pending_label or "",
section,
)
pending_icon = pending_label = None
nav: dict[str, dict] = {} nav: dict[str, dict] = {}
for top_idx, names in enumerate(comps): for top_idx, names in enumerate(comps):
if not names: if not names:
continue continue
main = names[0] main = names[0]
main_icon, main_label = top_meta.get(top_idx, ("tune", main)) main_icon, main_label = top_meta.get(top_idx, ("tune", main))
nav[main] = {"pageIdx": top_idx, "subPath": [], nav[main] = {
"crumbIcons": [main_icon], "crumbLabels": [main_label]} "pageIdx": top_idx,
children = dict(nav_children.get(main, {})) "subPath": [],
opened_via_subpage = set() "crumbIcons": [main_icon],
for owner, kids in nav_children.items(): "crumbLabels": [main_label],
owner_group = next((ns for ns in comps if owner in ns), None) }
if not owner_group: children = dict(nav_children.get(main, {}))
continue opened_via_subpage = set()
for kpos in kids: for owner, kids in nav_children.items():
if kpos < len(owner_group): owner_group = next((ns for ns in comps if owner in ns), None)
opened_via_subpage.add(owner_group[kpos]) if not owner_group:
for pos in range(1, len(names)): continue
if pos not in children and names[pos] not in opened_via_subpage: for kpos in kids:
label = re.sub(r"(Detail)?Page$", "", names[pos]) if kpos < len(owner_group):
label = re.sub(r"(?<!^)(?=[A-Z])", " ", label) opened_via_subpage.add(owner_group[kpos])
children[pos] = (main_icon, label, "") for pos in range(1, len(names)):
for pos, (icon, label, section) in children.items(): if pos not in children and names[pos] not in opened_via_subpage:
if pos >= len(names): label = re.sub(r"(Detail)?Page$", "", names[pos])
continue label = re.sub(r"(?<!^)(?=[A-Z])", " ", label)
child = names[pos] children[pos] = (main_icon, label, "")
labels = [main_label] + ([section] if section else []) + [label] for pos, (icon, label, section) in children.items():
icons = [main_icon] + ([icon] if section else []) + [icon] if pos >= len(names):
labels, icons = dedup_crumbs(labels, icons) continue
nav[child] = {"pageIdx": top_idx, "subPath": [pos], child = names[pos]
"crumbIcons": icons, labels = [main_label] + ([section] if section else []) + [label]
"crumbLabels": labels} icons = [main_icon] + ([icon] if section else []) + [icon]
for gpos, (gicon, glabel, gsection) in nav_children.get(child, {}).items(): labels, icons = dedup_crumbs(labels, icons)
if gpos >= len(names): nav[child] = {
continue "pageIdx": top_idx,
glabels = labels + ([gsection] if gsection else []) + [glabel] "subPath": [pos],
gicons = icons + ([gicon] if gsection else []) + [gicon] "crumbIcons": icons,
glabels, gicons = dedup_crumbs(glabels, gicons) "crumbLabels": labels,
nav[names[gpos]] = { }
"pageIdx": top_idx, "subPath": [pos, gpos], for gpos, (gicon, glabel, gsection) in nav_children.get(
"crumbIcons": gicons, child, {}
"crumbLabels": glabels} ).items():
return nav if gpos >= len(names):
continue
glabels = labels + ([gsection] if gsection else []) + [glabel]
gicons = icons + ([gicon] if gsection else []) + [gicon]
glabels, gicons = dedup_crumbs(glabels, gicons)
nav[names[gpos]] = {
"pageIdx": top_idx,
"subPath": [pos, gpos],
"crumbIcons": gicons,
"crumbLabels": glabels,
}
return nav
def tokenize(text: str) -> list[str]: def tokenize(text: str) -> list[str]:
toks: list[str] = [] toks: list[str] = []
for word in text.lower().split(): for word in text.lower().split():
parts = [p for p in re.split(r"[^a-z0-9]+", word) if p] parts = [p for p in re.split(r"[^a-z0-9]+", word) if p]
for p in parts: for p in parts:
if p not in STOPWORDS and p not in toks: if p not in STOPWORDS and p not in toks:
toks.append(p) toks.append(p)
if len(parts) > 1: if len(parts) > 1:
joined = "".join(parts) joined = "".join(parts)
if joined not in toks: if joined not in toks:
toks.append(joined) toks.append(joined)
return toks return toks
SUBTEXT_RE = re.compile(r'^\s*(?:subtext|status):\s*qsTr\("([^"]+)"\)') SUBTEXT_RE = re.compile(r'^\s*(?:subtext|status):\s*qsTr\("([^"]+)"\)')
SECTION_RE = re.compile(r'^\s*SectionHeader\s*\{') SECTION_RE = re.compile(r"^\s*SectionHeader\s*\{")
def extract_settings(files: dict[str, Path], nav: dict[str, dict]) -> list[dict]: def extract_settings(
entries: list[dict] = [] files: dict[str, Path], nav: dict[str, dict]
for comp, meta in nav.items(): ) -> list[dict]:
pf = files.get(comp) entries: list[dict] = []
if not pf: for comp, meta in nav.items():
continue pf = files.get(comp)
lines = read_lines(pf) if not pf:
section = "" continue
i = 0 lines = read_lines(pf)
while i < len(lines): section = ""
if SECTION_RE.match(lines[i]): i = 0
for j in range(i + 1, min(i + 4, len(lines))): while i < len(lines):
m = LABEL_RE.match(lines[j]) if SECTION_RE.match(lines[i]):
if m: for j in range(i + 1, min(i + 4, len(lines))):
section = m.group(1) m = LABEL_RE.match(lines[j])
break if m:
row_match = ROW_RE.match(lines[i]) section = m.group(1)
if row_match: break
row_type = row_match.group(1) row_match = ROW_RE.match(lines[i])
label = anchor = subtext = None if row_match:
checked_path = toggled_path = None row_type = row_match.group(1)
for j in range(i + 1, min(i + 12, len(lines))): label = anchor = subtext = None
if label is None: checked_path = toggled_path = None
m = LABEL_RE.match(lines[j]) for j in range(i + 1, min(i + 12, len(lines))):
if m: if label is None:
label = m.group(1) m = LABEL_RE.match(lines[j])
if anchor is None: if m:
a = ANCHOR_RE.match(lines[j]) label = m.group(1)
if a: if anchor is None:
anchor = a.group(1) a = ANCHOR_RE.match(lines[j])
if subtext is None: if a:
st = SUBTEXT_RE.match(lines[j]) anchor = a.group(1)
if st: if subtext is None:
subtext = st.group(1) st = SUBTEXT_RE.match(lines[j])
if checked_path is None: if st:
ch = CHECKED_RE.match(lines[j]) subtext = st.group(1)
if ch: if checked_path is None:
checked_path = ch.group(1) ch = CHECKED_RE.match(lines[j])
if toggled_path is None: if ch:
tg = ONTOGGLED_RE.match(lines[j]) checked_path = ch.group(1)
if tg: if toggled_path is None:
toggled_path = tg.group(1) tg = ONTOGGLED_RE.match(lines[j])
toggle_path = ( if tg:
checked_path toggled_path = tg.group(1)
if row_type == "ToggleRow" and checked_path and checked_path == toggled_path toggle_path = (
else "" checked_path
) if row_type == "ToggleRow"
if label and label not in SKIP_LABELS and anchor: and checked_path
extra = " ".join(meta["crumbLabels"]) + \ and checked_path == toggled_path
" " + section + " " + (subtext or "") else ""
entries.append({ )
"pageIdx": meta["pageIdx"], "subPath": meta["subPath"], if label and label not in SKIP_LABELS and anchor:
"crumbIcons": meta["crumbIcons"], extra = (
"crumbLabels": meta["crumbLabels"], " ".join(meta["crumbLabels"])
"title": label, "anchor": anchor, + " "
"section": section, + section
"subtext": subtext or "", + " "
"togglePath": toggle_path, + (subtext or "")
"keywords": " ".join(sorted(set(tokenize(label + " " + extra)))), )
}) entries.append(
i += 1 {
return entries "pageIdx": meta["pageIdx"],
"subPath": meta["subPath"],
"crumbIcons": meta["crumbIcons"],
"crumbLabels": meta["crumbLabels"],
"title": label,
"anchor": anchor,
"section": section,
"subtext": subtext or "",
"togglePath": toggle_path,
"keywords": " ".join(
sorted(set(tokenize(label + " " + extra)))
),
}
)
i += 1
return entries
def build_inverted_and_ranking(entries: list[dict]): def build_inverted_and_ranking(entries: list[dict]):
inverted: dict[str, list[int]] = defaultdict(list) inverted: dict[str, list[int]] = defaultdict(list)
ranking: dict[str, dict[int, float]] = defaultdict(dict) ranking: dict[str, dict[int, float]] = defaultdict(dict)
for idx, e in enumerate(entries): for idx, e in enumerate(entries):
fields = {"title": e["title"], "keywords": e["keywords"]} fields = {"title": e["title"], "keywords": e["keywords"]}
seen: set[str] = set() seen: set[str] = set()
for field, text in fields.items(): for field, text in fields.items():
weight = FIELD_WEIGHT.get(field, 0.2) weight = FIELD_WEIGHT.get(field, 0.2)
for tok in tokenize(text): for tok in tokenize(text):
if idx not in inverted[tok]: if idx not in inverted[tok]:
inverted[tok].append(idx) inverted[tok].append(idx)
ranking[tok][idx] = max(ranking[tok].get(idx, 0.0), weight) ranking[tok][idx] = max(ranking[tok].get(idx, 0.0), weight)
seen.add(tok) seen.add(tok)
for tok, ids in inverted.items(): for tok, ids in inverted.items():
ids.sort(key=lambda i: ranking[tok][i], reverse=True) ids.sort(key=lambda i: ranking[tok][i], reverse=True)
return inverted, {t: {str(k): v for k, v in d.items()} for t, d in ranking.items()} return inverted, {
t: {str(k): v for k, v in d.items()} for t, d in ranking.items()
}
def main() -> int: def main() -> int:
if len(sys.argv) != 3: if len(sys.argv) != 3:
print(__doc__) print(__doc__)
return 1 return 1
settings = Path(sys.argv[1]) settings = Path(sys.argv[1])
out = Path(sys.argv[2]) out = Path(sys.argv[2])
files = discover_files(settings) files = discover_files(settings)
nav = build_nav_map(settings, files) nav = build_nav_map(settings, files)
entries = extract_settings(files, nav) entries = extract_settings(files, nav)
inverted, ranking = build_inverted_and_ranking(entries) inverted, ranking = build_inverted_and_ranking(entries)
for e in entries: for e in entries:
e.pop("keywords", None) e.pop("keywords", None)
out.write_text(json.dumps({ out.write_text(
"version": 2, json.dumps(
"entries": entries, {
"inverted": inverted, "version": 2,
"ranking": ranking, "entries": entries,
}, ensure_ascii=False, indent=2)) "inverted": inverted,
print(f"settings index: {len(entries)} entries, " "ranking": ranking,
f"{len(inverted)} tokens -> {out}") },
print("files:", len(files)) ensure_ascii=False,
print("comps:", len(parse_page_comps(settings))) indent=2,
print("registry:", len(parse_page_registry(settings))) )
print("nav:", len(nav)) )
print("entries:", len(entries)) print(
return 0 f"settings index: {len(entries)} entries, "
f"{len(inverted)} tokens -> {out}"
)
print("files:", len(files))
print("comps:", len(parse_page_comps(settings)))
print("registry:", len(parse_page_registry(settings)))
print("nav:", len(nav))
print("entries:", len(entries))
return 0
if __name__ == "__main__": if __name__ == "__main__":
sys.exit(main()) sys.exit(main())