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

refactor: use standard logging

Browse Source
This commit is contained in:
banteg
2025-12-29 03:22:07 +04:00
parent db5c0c824b
commit 0ea5a7770c
1 changed files with 113 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex_telegram_bridge/src/codex_telegram_bridge/exec_bridge.py
+113 -96
View File
@@ -13,6 +13,8 @@ import subprocess
import threading import threading
import time import time
import logging import logging
import sys
from logging.handlers import RotatingFileHandler
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
from typing import Any from typing import Any
from collections.abc import Callable from collections.abc import Callable
@@ -27,34 +29,31 @@ from .telegram_client import TelegramClient
# -------------------- Codex runner -------------------- # -------------------- Codex runner --------------------
logger = logging.getLogger("exec_bridge")
def log(msg: str) -> None: def setup_logging(log_file: str | None) -> None:
ts = time.strftime("%Y-%m-%d %H:%M:%S")
line = f"[{ts}] {msg}"
print(line, flush=True)
log_debug(line)
_file_logger: logging.Logger | None = None
def setup_file_logger(path: str | None) -> None:
global _file_logger
if not path:
return
logger = logging.getLogger("exec_bridge")
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
handler = logging.FileHandler(path, encoding="utf-8") logger.handlers.clear()
handler.setFormatter(logging.Formatter("%(asctime)s %(message)s")) logger.propagate = False
logger.addHandler(handler)
_file_logger = logger
log_debug(f"[debug] file logger initialized path={path!r}")
fmt = logging.Formatter("%(asctime)s %(message)s")
def log_debug(msg: str) -> None: console = logging.StreamHandler(sys.stdout)
if _file_logger is None: console.setLevel(logging.INFO)
return console.setFormatter(fmt)
_file_logger.debug(msg) logger.addHandler(console)
if log_file:
file_handler = RotatingFileHandler(
log_file,
maxBytes=5 * 1024 * 1024,
backupCount=3,
encoding="utf-8",
)
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(fmt)
logger.addHandler(file_handler)
logger.debug("[debug] file logger initialized path=%r", log_file)
def _one_line(text: str | None) -> str: def _one_line(text: str | None) -> str:
@@ -130,7 +129,7 @@ class ProgressEditor:
text = _clamp_tg_text(text) text = _clamp_tg_text(text)
with self._lock: with self._lock:
self._pending = (text, entities) self._pending = (text, entities)
log_debug(f"[progress] set pending len={len(text)} entities={bool(entities)}") logger.debug("[progress] set pending len=%s entities=%s", len(text), bool(entities))
def set_markdown(self, text: str) -> None: def set_markdown(self, text: str) -> None:
rendered_text, entities = render_markdown(text) rendered_text, entities = render_markdown(text)
@@ -148,16 +147,18 @@ class ProgressEditor:
text=text, text=text,
entities=entities, entities=entities,
) )
log_debug( logger.debug(
f"[progress] edit ok chat_id={self.chat_id} message_id={self.message_id} len={len(text)}" "[progress] edit ok chat_id=%s message_id=%s len=%s",
self.chat_id,
self.message_id,
len(text),
) )
except Exception as e: except Exception as e:
log( logger.info(
"[progress] edit failed " "[progress] edit failed chat_id=%s message_id=%s: %s",
f"chat_id={self.chat_id} message_id={self.message_id}: {e}" self.chat_id,
) self.message_id,
log_debug( e,
f"[progress] edit failed chat_id={self.chat_id} message_id={self.message_id}: {e}"
) )
def _run(self) -> None: def _run(self) -> None:
@@ -209,7 +210,7 @@ class CodexExecRunner:
""" """
Returns (session_id, final_agent_message_text) Returns (session_id, final_agent_message_text)
""" """
log(f"[codex] start run session_id={session_id!r} workspace={self.workspace!r}") logger.info("[codex] start run session_id=%r workspace=%r", session_id, self.workspace)
args = [self.codex_cmd, "exec", "--json"] args = [self.codex_cmd, "exec", "--json"]
args.extend(self.extra_args) args.extend(self.extra_args)
if self.workspace: if self.workspace:
@@ -230,7 +231,7 @@ class CodexExecRunner:
text=True, text=True,
bufsize=1, bufsize=1,
) )
log_debug(f"[codex] spawn pid={proc.pid} args={args!r}") logger.debug("[codex] spawn pid=%s args=%r", proc.pid, args)
assert proc.stdin and proc.stdout and proc.stderr assert proc.stdin and proc.stdout and proc.stderr
# send prompt then close stdin # send prompt then close stdin
@@ -241,7 +242,7 @@ class CodexExecRunner:
def _drain_stderr() -> None: def _drain_stderr() -> None:
for line in proc.stderr: for line in proc.stderr:
log(f"[codex][stderr] {line.rstrip()}") logger.info("[codex][stderr] %s", line.rstrip())
stderr_lines.append(line) stderr_lines.append(line)
t = threading.Thread(target=_drain_stderr, daemon=True) t = threading.Thread(target=_drain_stderr, daemon=True)
@@ -263,12 +264,12 @@ class CodexExecRunner:
continue continue
cli_last_turn, out_lines = render_event_cli(evt, cli_last_turn) cli_last_turn, out_lines = render_event_cli(evt, cli_last_turn)
for out in out_lines: for out in out_lines:
log(f"[codex] {out}") logger.info("[codex] %s", out)
if on_event is not None: if on_event is not None:
try: try:
on_event(evt) on_event(evt)
except Exception as e: except Exception as e:
log(f"[codex][on_event] callback error: {e}") logger.info("[codex][on_event] callback error: %s", e)
# From Codex JSONL event stream # From Codex JSONL event stream
if evt.get("type") == "thread.started": if evt.get("type") == "thread.started":
@@ -281,7 +282,7 @@ class CodexExecRunner:
saw_agent_message = True saw_agent_message = True
rc = proc.wait() rc = proc.wait()
log_debug(f"[codex] process exit pid={proc.pid} rc={rc}") logger.debug("[codex] process exit pid=%s rc=%s", proc.pid, rc)
t.join(timeout=2.0) t.join(timeout=2.0)
if rc != 0: if rc != 0:
@@ -291,7 +292,7 @@ class CodexExecRunner:
if not found_session: if not found_session:
raise RuntimeError("codex exec finished but no session_id/thread_id was captured") raise RuntimeError("codex exec finished but no session_id/thread_id was captured")
log(f"[codex] done run session_id={found_session!r}") logger.info("[codex] done run session_id=%r", found_session)
return found_session, (last_agent_text or "(No agent_message captured from JSON stream.)"), saw_agent_message return found_session, (last_agent_text or "(No agent_message captured from JSON stream.)"), saw_agent_message
def run_serialized( def run_serialized(
@@ -351,7 +352,7 @@ def run(
help="Codex model to pass to `codex exec`.", help="Codex model to pass to `codex exec`.",
), ),
) -> None: ) -> None:
setup_file_logger(log_file if log_file else None) setup_logging(log_file if log_file else None)
config = load_telegram_config() config = load_telegram_config()
token = config["bot_token"] token = config["bot_token"]
@@ -410,32 +411,35 @@ def run(
try: try:
updates = bot.get_updates(offset=offset, timeout_s=0, allowed_updates=["message"]) updates = bot.get_updates(offset=offset, timeout_s=0, allowed_updates=["message"])
except Exception as e: except Exception as e:
log(f"[startup] backlog drain failed: {e}") logger.info("[startup] backlog drain failed: %s", e)
updates = [] updates = []
if updates: if updates:
offset = updates[-1]["update_id"] + 1 offset = updates[-1]["update_id"] + 1
log(f"[startup] drained {len(updates)} pending update(s)") logger.info("[startup] drained %s pending update(s)", len(updates))
log(f"[startup] pwd={startup_pwd}") logger.info("[startup] pwd=%s", startup_pwd)
log("Option1 bridge running (codex exec). Long-polling Telegram...") logger.info("Option1 bridge running (codex exec). Long-polling Telegram...")
if startup_ids: if startup_ids:
for chat_id in startup_ids: for chat_id in startup_ids:
try: try:
bot.send_message(chat_id=chat_id, text=startup_msg) bot.send_message(chat_id=chat_id, text=startup_msg)
log(f"[startup] sent startup message to chat_id={chat_id}") logger.info("[startup] sent startup message to chat_id=%s", chat_id)
except Exception as e: except Exception as e:
log(f"[startup] failed to send startup message to chat_id={chat_id}: {e}") logger.info("[startup] failed to send startup message to chat_id=%s: %s", chat_id, e)
else: else:
log("[startup] no chat_id configured; skipping startup message") logger.info("[startup] no chat_id configured; skipping startup message")
def handle(chat_id: int, user_msg_id: int, text: str, resume_session: str | None) -> None: def handle(chat_id: int, user_msg_id: int, text: str, resume_session: str | None) -> None:
log( logger.info(
"[handle] start " "[handle] start chat_id=%s user_msg_id=%s resume_session=%r",
f"chat_id={chat_id} user_msg_id={user_msg_id} resume_session={resume_session!r}" chat_id,
user_msg_id,
resume_session,
) )
log_debug( logger.debug(
"[handle] thread " "[handle] thread name=%s ident=%s",
f"name={threading.current_thread().name} ident={threading.get_ident()}" threading.current_thread().name,
threading.get_ident(),
) )
edit_every_s = progress_edit_every_s edit_every_s = progress_edit_every_s
silent_progress = progress_silent silent_progress = progress_silent
@@ -458,10 +462,9 @@ def run(
disable_notification=silent_progress, disable_notification=silent_progress,
) )
progress_id = int(progress_msg["message_id"]) progress_id = int(progress_msg["message_id"])
log_debug(f"[progress] sent chat_id={chat_id} message_id={progress_id}") logger.debug("[progress] sent chat_id=%s message_id=%s", chat_id, progress_id)
except Exception as e: except Exception as e:
log(f"[handle] failed to send progress message chat_id={chat_id}: {e}") logger.info("[handle] failed to send progress message chat_id=%s: %s", chat_id, e)
log_debug(f"[handle] failed to send progress message chat_id={chat_id}: {e}")
if progress_id is not None: if progress_id is not None:
progress = ProgressEditor( progress = ProgressEditor(
@@ -476,9 +479,12 @@ def run(
def on_event(evt: dict[str, Any]) -> None: def on_event(evt: dict[str, Any]) -> None:
event_type = evt.get("type") event_type = evt.get("type")
item = evt.get("item") or {} item = evt.get("item") or {}
log_debug( logger.debug(
"[codex] event " "[codex] event type=%s item_id=%s item_type=%s status=%s",
f"type={event_type} item_id={item.get('id')} item_type={item.get('type')} status={item.get('status')}" event_type,
item.get("id"),
item.get("type"),
item.get("status"),
) )
if event_type == "thread.started": if event_type == "thread.started":
thread_id = evt.get("thread_id") thread_id = evt.get("thread_id")
@@ -492,7 +498,7 @@ def run(
def _stop_background() -> None: def _stop_background() -> None:
if progress is not None: if progress is not None:
progress.stop() progress.stop()
log_debug("[progress] thread stopped") logger.debug("[progress] thread stopped")
try: try:
session_id, answer, saw_agent_message = runner.run_serialized( session_id, answer, saw_agent_message = runner.run_serialized(
@@ -506,21 +512,25 @@ def run(
if progress_id is not None and len(err) <= TELEGRAM_TEXT_LIMIT: if progress_id is not None and len(err) <= TELEGRAM_TEXT_LIMIT:
try: try:
bot.edit_message_text(chat_id=chat_id, message_id=progress_id, text=err) bot.edit_message_text(chat_id=chat_id, message_id=progress_id, text=err)
log( logger.info(
"[handle] error " "[handle] error chat_id=%s user_msg_id=%s resume_session=%r err=%s",
f"chat_id={chat_id} user_msg_id={user_msg_id} " chat_id,
f"resume_session={resume_session!r} err={e}" user_msg_id,
resume_session,
e,
) )
return return
except Exception as ee: except Exception as ee:
log(f"[handle] failed to edit progress into error: {ee}") logger.info("[handle] failed to edit progress into error: %s", ee)
_send_markdown(bot, chat_id=chat_id, text=err, reply_to_message_id=user_msg_id) _send_markdown(bot, chat_id=chat_id, text=err, reply_to_message_id=user_msg_id)
log( logger.info(
"[handle] error " "[handle] error chat_id=%s user_msg_id=%s resume_session=%r err=%s",
f"chat_id={chat_id} user_msg_id={user_msg_id} resume_session={resume_session!r} err={e}" chat_id,
user_msg_id,
resume_session,
e,
) )
log_debug(f"[handle] error chat_id={chat_id} user_msg_id={user_msg_id}: {e}")
return return
_stop_background() _stop_background()
@@ -539,9 +549,11 @@ def run(
try: try:
bot.delete_message(chat_id=chat_id, message_id=progress_id) bot.delete_message(chat_id=chat_id, message_id=progress_id)
except Exception as e: except Exception as e:
log(f"[handle] delete progress failed chat_id={chat_id} message_id={progress_id}: {e}") logger.info(
log_debug( "[handle] delete progress failed chat_id=%s message_id=%s: %s",
f"[handle] delete progress failed chat_id={chat_id} message_id={progress_id}: {e}" chat_id,
progress_id,
e,
) )
else: else:
bot.edit_message_text( bot.edit_message_text(
@@ -551,17 +563,18 @@ def run(
entities=final_entities or None, entities=final_entities or None,
) )
log( logger.info(
"[handle] done " "[handle] done chat_id=%s user_msg_id=%s session_id=%r",
f"chat_id={chat_id} user_msg_id={user_msg_id} session_id={session_id!r}" chat_id,
user_msg_id,
session_id,
) )
log_debug(f"[handle] done chat_id={chat_id} user_msg_id={user_msg_id} session_id={session_id!r}")
while True: while True:
try: try:
updates = bot.get_updates(offset=offset, timeout_s=50, allowed_updates=["message"]) updates = bot.get_updates(offset=offset, timeout_s=50, allowed_updates=["message"])
except Exception as e: except Exception as e:
log(f"[telegram] get_updates error: {e}") logger.info("[telegram] get_updates error: %s", e)
time.sleep(2.0) time.sleep(2.0)
continue continue
@@ -572,38 +585,42 @@ def run(
from_bot = msg.get("from", {}).get("is_bot") from_bot = msg.get("from", {}).get("is_bot")
msg_text = msg.get("text") msg_text = msg.get("text")
reply_to = (msg.get("reply_to_message") or {}).get("message_id") reply_to = (msg.get("reply_to_message") or {}).get("message_id")
log( logger.info(
"[telegram] received " "[telegram] received update_id=%s chat_id=%s from_bot=%s has_text=%s reply_to=%s text=%s",
f"update_id={upd.get('update_id')} chat_id={chat_id} " upd.get("update_id"),
f"from_bot={from_bot} has_text={msg_text is not None} " chat_id,
f"reply_to={reply_to} text={_one_line(msg_text)}" from_bot,
msg_text is not None,
reply_to,
_one_line(msg_text),
) )
if "text" not in msg: if "text" not in msg:
log( logger.info(
"[telegram] ignoring non-text message " "[telegram] ignoring non-text message chat_id=%s update_id=%s",
f"chat_id={chat_id} update_id={upd.get('update_id')}" chat_id,
upd.get("update_id"),
) )
continue continue
if allowed is not None and int(chat_id) not in allowed: if allowed is not None and int(chat_id) not in allowed:
log( logger.info("[telegram] rejected by ACL chat_id=%s allowed=%s", chat_id, sorted(allowed))
"[telegram] rejected by ACL "
f"chat_id={chat_id} allowed={sorted(allowed)}"
)
continue continue
if msg.get("from", {}).get("is_bot"): if msg.get("from", {}).get("is_bot"):
log( logger.info(
"[telegram] ignoring bot message " "[telegram] ignoring bot message chat_id=%s update_id=%s",
f"chat_id={chat_id} update_id={upd.get('update_id')}" chat_id,
upd.get("update_id"),
) )
continue continue
text = msg["text"] text = msg["text"]
user_msg_id = msg["message_id"] user_msg_id = msg["message_id"]
log( logger.info(
"[telegram] accepted message " "[telegram] accepted message chat_id=%s user_msg_id=%s text=%s",
f"chat_id={chat_id} user_msg_id={user_msg_id} text={_one_line(text)}" chat_id,
user_msg_id,
_one_line(text),
) )
uuid_re = re.compile( uuid_re = re.compile(