feat: streamline onboarding and add --version

This commit is contained in:
banteg
2025-12-29 19:33:27 +04:00
parent d151788edb
commit e1fcc681eb
11 changed files with 53 additions and 91 deletions
+2
View File
@@ -1 +1,3 @@
"""Takopi — Telegram Codex bridge package."""
__version__ = "0.1.0"
+5 -33
View File
@@ -3,42 +3,14 @@ from __future__ import annotations
import tomllib
from pathlib import Path
from .constants import HOME_CONFIG_PATH, LOCAL_CONFIG_NAME
LOCAL_CONFIG_NAME = Path(".codex") / "takopi.toml"
HOME_CONFIG_PATH = Path.home() / ".codex" / "takopi.toml"
class ConfigError(RuntimeError):
pass
_EXAMPLE_CONFIG = (
'bot_token = "123456789:ABCdefGHIjklMNOpqrsTUVwxyz"\nchat_id = 123456789\n'
)
def _display_path(path: Path) -> str:
cwd = Path.cwd()
if path.is_relative_to(cwd):
return f"./{path.relative_to(cwd).as_posix()}"
home = Path.home()
if path.is_relative_to(home):
return f"~/{path.relative_to(home).as_posix()}"
return str(path)
def _missing_config_message(primary: Path, alternate: Path | None = None) -> str:
example = "Example config:\n```\n" + _EXAMPLE_CONFIG + "```\n"
if alternate is None:
return f"Missing config file `{_display_path(primary)}`.\n{example}"
return (
"Missing takopi config.\n"
"Create one of these files:\n"
f" {_display_path(alternate)}\n"
f" {_display_path(primary)}\n"
"\n"
f"{example}"
)
def _config_candidates() -> list[Path]:
candidates = [Path.cwd() / LOCAL_CONFIG_NAME, HOME_CONFIG_PATH]
if candidates[0] == candidates[1]:
@@ -50,7 +22,7 @@ def _read_config(cfg_path: Path) -> dict:
try:
raw = cfg_path.read_text(encoding="utf-8")
except FileNotFoundError:
raise ConfigError(_missing_config_message(cfg_path)) from None
raise ConfigError(f"Missing config file {cfg_path}.") from None
except OSError as e:
raise ConfigError(f"Failed to read config file {cfg_path}: {e}") from e
try:
@@ -70,5 +42,5 @@ def load_telegram_config(path: str | Path | None = None) -> tuple[dict, Path]:
return _read_config(candidate), candidate
if len(candidates) == 1:
raise ConfigError(_missing_config_message(candidates[0]))
raise ConfigError(_missing_config_message(HOME_CONFIG_PATH, candidates[0]))
raise ConfigError("Missing takopi config.")
raise ConfigError("Missing takopi config.")
-7
View File
@@ -1,7 +0,0 @@
from __future__ import annotations
from pathlib import Path
TELEGRAM_HARD_LIMIT = 4096
LOCAL_CONFIG_NAME = Path(".codex") / "takopi.toml"
HOME_CONFIG_PATH = Path.home() / ".codex" / "takopi.toml"
+19 -2
View File
@@ -17,10 +17,10 @@ from weakref import WeakValueDictionary
import typer
from . import __version__
from .config import ConfigError, load_telegram_config
from .exec_render import ExecProgressRenderer, render_event_cli
from .exec_render import ExecProgressRenderer, render_event_cli, render_markdown
from .logging import setup_logging
from .rendering import render_markdown
from .onboarding import check_setup, render_setup_guide
from .telegram_client import TelegramClient
@@ -33,6 +33,16 @@ RESUME_LINE = re.compile(
)
def _print_version_and_exit() -> None:
typer.echo(__version__)
raise typer.Exit()
def _version_callback(value: bool) -> None:
if value:
_print_version_and_exit()
def extract_session_id(text: str | None) -> str | None:
if not text:
return None
@@ -663,6 +673,13 @@ async def _run_main_loop(cfg: BridgeConfig) -> None:
def run(
version: bool = typer.Option(
False,
"--version",
help="Show the version and exit.",
callback=_version_callback,
is_eager=True,
),
final_notify: bool = typer.Option(
True,
"--final-notify/--no-final-notify",
+21
View File
@@ -6,6 +6,9 @@ from collections import deque
from textwrap import indent
from typing import Any
from markdown_it import MarkdownIt
from sulguk import transform_html
STATUS_RUNNING = ""
STATUS_DONE = ""
STATUS_FAIL = ""
@@ -16,6 +19,24 @@ MAX_PROGRESS_CMD_LEN = 300
MAX_QUERY_LEN = 60
MAX_PATH_LEN = 40
_md = MarkdownIt("commonmark", {"html": False})
def render_markdown(md: str) -> tuple[str, list[dict[str, Any]]]:
html = _md.render(md or "")
rendered = transform_html(html)
text = re.sub(r"(?m)^(\s*)•", r"\1-", rendered.text)
# FIX: Telegram requires MessageEntity.language (if present) to be a String.
entities: list[dict[str, Any]] = []
for e in rendered.entities:
d = dict(e)
if "language" in d and not isinstance(d["language"], str):
d.pop("language", None)
entities.append(d)
return text, entities
def format_elapsed(elapsed_s: float) -> str:
total = max(0, int(elapsed_s))
+2 -3
View File
@@ -9,8 +9,7 @@ from pathlib import Path
from rich.console import Console
from rich.panel import Panel
from .config import ConfigError, load_telegram_config
from .constants import HOME_CONFIG_PATH
from .config import ConfigError, HOME_CONFIG_PATH, load_telegram_config
_OCTOPUS = "\N{OCTOPUS}"
@@ -100,7 +99,7 @@ def render_setup_guide(result: SetupResult) -> None:
"[bold]Getting your Telegram credentials:[/]",
"",
" [cyan]bot_token[/] create a bot with [link=https://t.me/BotFather]@BotFather[/]",
" [cyan]chat_id[/] get from [link=https://t.me/myidbot]@myidbot[/]",
" [cyan]chat_id[/] message [link=https://t.me/myidbot]@myidbot[/] to get your id",
)
panel = Panel(
-25
View File
@@ -1,25 +0,0 @@
from __future__ import annotations
import re
from typing import Any
from markdown_it import MarkdownIt
from sulguk import transform_html
_md = MarkdownIt("commonmark", {"html": False})
def render_markdown(md: str) -> tuple[str, list[dict[str, Any]]]:
html = _md.render(md)
rendered = transform_html(html)
text = re.sub(r"(?m)^(\s*)•", r"\1-", rendered.text)
# FIX: Telegram requires MessageEntity.language (if present) to be a String.
entities: list[dict[str, Any]] = []
for e in rendered.entities:
d = dict(e)
if "language" in d and not isinstance(d["language"], str):
d.pop("language", None)
entities.append(d)
return text, entities