izackp/takopi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add codex profile config

Browse Source
This commit is contained in:
banteg
2025-12-29 16:31:29 +04:00
parent caa9f4f99d
commit ed27ecd3c9
5 changed files with 67 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/takopi/config.py
+23 -19
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import tomllib
from pathlib import Path
from .constants import DEFAULT_CONFIG_PATHS
from .constants import HOME_CONFIG_PATH, LOCAL_CONFIG_NAME
class ConfigError(RuntimeError):
@@ -25,20 +25,21 @@ def _display_path(path: Path) -> str:
def _missing_config_message(primary: Path, alternate: Path | None = None) -> str:
if alternate is None:
header = f"Missing config file `{_display_path(primary)}`."
else:
header = (
f"Missing config file `{_display_path(primary)}` "
f"(or `{_display_path(alternate)}`)."
)
return "\n".join(
[
header,
"Create it with:",
' bot_token = "123456789:ABCdefGHIjklMNOpqrsTUVwxyz"',
" chat_id = 123456789",
]
)
return f"Missing config file `{_display_path(primary)}`."
return "Missing takopi config. See readme.md for setup."
def _config_candidates(base_dir: Path | None) -> list[Path]:
candidates: list[Path] = []
if base_dir is not None:
candidates.append(base_dir / LOCAL_CONFIG_NAME)
cwd = Path.cwd()
if base_dir is None or base_dir != cwd:
candidates.append(cwd / LOCAL_CONFIG_NAME)
candidates.append(HOME_CONFIG_PATH)
return candidates
def _read_config(cfg_path: Path) -> dict:
@@ -54,14 +55,17 @@ def _read_config(cfg_path: Path) -> dict:
raise ConfigError(f"Malformed TOML in {cfg_path}: {e}") from None
def load_telegram_config(path: str | Path | None = None) -> tuple[dict, Path]:
def load_telegram_config(
path: str | Path | None = None, *, base_dir: str | Path | None = None
) -> tuple[dict, Path]:
if path:
cfg_path = Path(path).expanduser()
return _read_config(cfg_path), cfg_path
local_path, home_path = DEFAULT_CONFIG_PATHS
for candidate in (local_path, home_path):
base = Path(base_dir).expanduser() if base_dir is not None else None
candidates = _config_candidates(base)
for candidate in candidates:
if candidate.is_file():
return _read_config(candidate), candidate
raise ConfigError(_missing_config_message(home_path, local_path))
raise ConfigError(_missing_config_message(HOME_CONFIG_PATH, candidates[0]))
src/takopi/constants.py
+2 -4
View File
@@ -3,7 +3,5 @@ from __future__ import annotations
from pathlib import Path
TELEGRAM_HARD_LIMIT = 4096
DEFAULT_CONFIG_PATHS = (
Path.cwd() / "codex" / "takopi.toml",
Path.home() / ".codex" / "takopi.toml",
)
LOCAL_CONFIG_NAME = Path("codex") / "takopi.toml"
HOME_CONFIG_PATH = Path.home() / ".codex" / "takopi.toml"
src/takopi/exec_bridge.py
+20 -40
View File
@@ -6,7 +6,6 @@ import json
import logging
import os
import re
import shlex
import shutil
import time
from collections import deque
@@ -224,10 +223,11 @@ class CodexExecRunner:
"[codex] start run session_id=%r workspace=%r", session_id, self.workspace
)
logger.debug("[codex] prompt: %s", prompt)
args = [self.codex_cmd, "exec", "--json"]
args = [self.codex_cmd]
args.extend(self.extra_args)
if self.workspace:
args.extend(["--cd", self.workspace])
args.extend(["exec", "--json"])
# Always pipe prompt via stdin ("-") to avoid quoting issues.
if session_id:
@@ -356,9 +356,18 @@ def _parse_bridge_config(
*,
final_notify: bool,
cd: str | None,
model: str | None,
profile: str | None,
) -> BridgeConfig:
config, config_path = load_telegram_config()
startup_pwd = os.getcwd()
workspace = None
if cd is not None:
expanded_cd = os.path.expanduser(cd)
if not os.path.isdir(expanded_cd):
raise ConfigError(f"--cd must be an existing directory: {expanded_cd}")
workspace = expanded_cd
startup_pwd = expanded_cd
config, config_path = load_telegram_config(base_dir=workspace)
try:
token = config["bot_token"]
except KeyError:
@@ -381,39 +390,10 @@ def _parse_bridge_config(
" brew install codex"
)
startup_pwd = os.getcwd()
workspace = None
if cd is not None:
expanded_cd = os.path.expanduser(cd)
if not os.path.isdir(expanded_cd):
raise ConfigError(f"--cd must be an existing directory: {expanded_cd}")
workspace = expanded_cd
startup_pwd = expanded_cd
startup_msg = f"codex exec bridge has started\npwd: {startup_pwd}"
raw_exec_args = config.get("codex_exec_args", "")
if isinstance(raw_exec_args, list):
extra_args = [str(v) for v in raw_exec_args]
else:
extra_args = shlex.split(str(raw_exec_args)) # e.g. "--full-auto --search"
if model:
extra_args.extend(["--model", model])
def _has_notify_override(args: list[str]) -> bool:
for i, arg in enumerate(args):
if arg in ("-c", "--config"):
key = args[i + 1].split("=", 1)[0].strip()
if key == "notify" or key.endswith(".notify"):
return True
elif arg.startswith(("--config=", "-c=")):
key = arg.split("=", 1)[1].split("=", 1)[0].strip()
if key == "notify" or key.endswith(".notify"):
return True
return False
if not _has_notify_override(extra_args):
extra_args.extend(["-c", "notify=[]"])
extra_args = ["-c", "notify=[]"]
if profile:
extra_args.extend(["--profile", profile])
bot = TelegramClient(token)
runner = CodexExecRunner(codex_cmd=codex_cmd, workspace=workspace, extra_args=extra_args)
@@ -716,10 +696,10 @@ def run(
"--cd",
help="Pass through to `codex --cd`.",
),
model: str | None = typer.Option(
profile: str | None = typer.Option(
None,
"--model",
help="Codex model to pass to `codex exec`.",
"--profile",
help="Codex profile name to pass to `codex --profile`.",
),
) -> None:
setup_logging(debug=debug)
@@ -727,7 +707,7 @@ def run(
cfg = _parse_bridge_config(
final_notify=final_notify,
cd=cd,
model=model,
profile=profile,
)
except ConfigError as e:
typer.echo(str(e), err=True)