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

refactor: simplify with_id formatting

Browse Source
This commit is contained in:
banteg
2025-12-29 02:57:35 +04:00
parent 36da2114b7
commit f86d509a05
1 changed files with 10 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex_telegram_bridge/src/codex_telegram_bridge/exec_render.py
+10 -11
View File
@@ -88,10 +88,9 @@ def extract_numeric_id(item_id: Optional[object], fallback: Optional[int] = None
return fallback return fallback
def with_id(item_id: Optional[int], line: str) -> str: def with_id(item_id: Optional[int], *parts: str) -> str:
if item_id is not None: prefix = f"[{item_id}] " if item_id is not None else "[?] "
return f"[{item_id}] {line}" return prefix + "".join(parts)
return f"[?] {line}"
def format_item_action_line(etype: str, item_id: Optional[int], item: dict[str, Any]) -> str | None: def format_item_action_line(etype: str, item_id: Optional[int], item: dict[str, Any]) -> str | None:
@@ -99,19 +98,19 @@ def format_item_action_line(etype: str, item_id: Optional[int], item: dict[str,
if itype == "command_execution": if itype == "command_execution":
command = format_command(item["command"]) command = format_command(item["command"])
if etype == "item.started": if etype == "item.started":
return with_id(item_id, f"{STATUS_RUNNING} running: {command}") return with_id(item_id, STATUS_RUNNING, " running: ", command)
if etype == "item.completed": if etype == "item.completed":
exit_code = item["exit_code"] exit_code = item["exit_code"]
exit_part = f" (exit {exit_code})" if exit_code is not None else "" exit_part = f" (exit {exit_code})" if exit_code is not None else ""
return with_id(item_id, f"{STATUS_DONE} ran: {command}{exit_part}") return with_id(item_id, STATUS_DONE, " ran: ", command, exit_part)
return None return None
if itype == "mcp_tool_call": if itype == "mcp_tool_call":
name = format_tool_call(item["server"], item["tool"]) name = format_tool_call(item["server"], item["tool"])
if etype == "item.started": if etype == "item.started":
return with_id(item_id, f"{STATUS_RUNNING} tool: {name}") return with_id(item_id, STATUS_RUNNING, " tool: ", name)
if etype == "item.completed": if etype == "item.completed":
return with_id(item_id, f"{STATUS_DONE} tool: {name}") return with_id(item_id, STATUS_DONE, " tool: ", name)
return None return None
return None return None
@@ -121,12 +120,12 @@ def format_item_completed_line(item_id: Optional[int], item: dict[str, Any]) ->
itype = item["type"] itype = item["type"]
if itype == "web_search": if itype == "web_search":
query = format_query(item["query"]) query = format_query(item["query"])
return with_id(item_id, f"{STATUS_DONE} searched: {query}") return with_id(item_id, STATUS_DONE, " searched: ", query)
if itype == "file_change": if itype == "file_change":
return with_id(item_id, f"{STATUS_DONE} {format_file_change(item['changes'])}") return with_id(item_id, STATUS_DONE, " ", format_file_change(item["changes"]))
if itype == "error": if itype == "error":
warning = truncate(item["message"], 120) warning = truncate(item["message"], 120)
return with_id(item_id, f"{STATUS_DONE} warning: {warning}") return with_id(item_id, STATUS_DONE, " warning: ", warning)
return None return None