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

chore: modernize typing imports

Browse Source
This commit is contained in:
banteg
2025-12-29 03:17:19 +04:00
parent e4ef1b85d3
commit db5c0c824b
3 changed files with 51 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex_telegram_bridge/src/codex_telegram_bridge/exec_bridge.py
+33 -32
View File
@@ -14,7 +14,8 @@ import threading
import time
import logging
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Callable, Dict, Optional, Tuple
from typing import Any
from collections.abc import Callable
import typer
@@ -34,10 +35,10 @@ def log(msg: str) -> None:
log_debug(line)
_file_logger: Optional[logging.Logger] = None
_file_logger: logging.Logger | None = None
def setup_file_logger(path: Optional[str]) -> None:
def setup_file_logger(path: str | None) -> None:
global _file_logger
if not path:
return
@@ -56,7 +57,7 @@ def log_debug(msg: str) -> None:
_file_logger.debug(msg)
def _one_line(text: Optional[str]) -> str:
def _one_line(text: str | None) -> str:
if text is None:
return "None"
return text.replace("\r", "\\r").replace("\n", "\\n")
@@ -77,9 +78,9 @@ def _send_markdown(
*,
chat_id: int,
text: str,
reply_to_message_id: Optional[int] = None,
reply_to_message_id: int | None = None,
disable_notification: bool = False,
) -> Dict[str, Any]:
) -> dict[str, Any]:
rendered_text, entities = render_markdown(text)
if len(rendered_text) > TELEGRAM_MARKDOWN_LIMIT:
sep = "\n" + ELLIPSIS + "\n"
@@ -104,8 +105,8 @@ class ProgressEditor:
chat_id: int,
message_id: int,
edit_every_s: float,
initial_text: Optional[str] = None,
initial_entities: Optional[list[dict[str, Any]]] = None,
initial_text: str | None = None,
initial_entities: list[dict[str, Any]] | None = None,
) -> None:
self.bot = bot
self.chat_id = chat_id
@@ -113,8 +114,8 @@ class ProgressEditor:
self.edit_every_s = edit_every_s
self._lock = threading.Lock()
self._pending: Optional[tuple[str, Optional[list[dict[str, Any]]]]] = None
self._last_sent: Optional[tuple[str, Optional[list[dict[str, Any]]]]] = None
self._pending: tuple[str, list[dict[str, Any]] | None] | None = None
self._last_sent: tuple[str, list[dict[str, Any]] | None] | None = None
self._last_edit_at = 0.0
if initial_text is not None:
@@ -125,7 +126,7 @@ class ProgressEditor:
self._thread = threading.Thread(target=self._run, daemon=True)
self._thread.start()
def set(self, text: str, entities: Optional[list[dict[str, Any]]] = None) -> None:
def set(self, text: str, entities: list[dict[str, Any]] | None = None) -> None:
text = _clamp_tg_text(text)
with self._lock:
self._pending = (text, entities)
@@ -139,7 +140,7 @@ class ProgressEditor:
self._stop.set()
self._thread.join(timeout=1.0)
def _edit(self, text: str, entities: Optional[list[dict[str, Any]]]) -> None:
def _edit(self, text: str, entities: list[dict[str, Any]] | None) -> None:
try:
self.bot.edit_message_text(
chat_id=self.chat_id,
@@ -161,7 +162,7 @@ class ProgressEditor:
def _run(self) -> None:
while not self._stop.is_set():
to_send: Optional[tuple[str, Optional[list[dict[str, Any]]]]] = None
to_send: tuple[str, list[dict[str, Any]] | None] | None = None
now = time.monotonic()
with self._lock:
if self._pending is not None and (now - self._last_edit_at) >= self.edit_every_s:
@@ -184,7 +185,7 @@ class CodexExecRunner:
- resume: codex exec --json ... resume <SESSION_ID> -
"""
def __init__(self, codex_cmd: str, workspace: Optional[str], extra_args: list[str]) -> None:
def __init__(self, codex_cmd: str, workspace: str | None, extra_args: list[str]) -> None:
self.codex_cmd = codex_cmd
self.workspace = workspace
self.extra_args = extra_args
@@ -202,9 +203,9 @@ class CodexExecRunner:
def run(
self,
prompt: str,
session_id: Optional[str],
on_event: Optional[Callable[[Dict[str, Any]], None]] = None,
) -> Tuple[str, str, bool]:
session_id: str | None,
on_event: Callable[[dict[str, Any]], None] | None = None,
) -> tuple[str, str, bool]:
"""
Returns (session_id, final_agent_message_text)
"""
@@ -246,8 +247,8 @@ class CodexExecRunner:
t = threading.Thread(target=_drain_stderr, daemon=True)
t.start()
found_session: Optional[str] = session_id
last_agent_text: Optional[str] = None
found_session: str | None = session_id
last_agent_text: str | None = None
saw_agent_message = False
cli_last_turn = None
@@ -296,9 +297,9 @@ class CodexExecRunner:
def run_serialized(
self,
prompt: str,
session_id: Optional[str],
on_event: Optional[Callable[[Dict[str, Any]], None]] = None,
) -> Tuple[str, str, bool]:
session_id: str | None,
on_event: Callable[[dict[str, Any]], None] | None = None,
) -> tuple[str, str, bool]:
"""
If resuming, serialize per-session.
"""
@@ -334,17 +335,17 @@ def run(
"--ignore-backlog/--process-backlog",
help="Skip pending Telegram updates that arrived before startup.",
),
log_file: Optional[str] = typer.Option(
log_file: str | None = typer.Option(
"exec_bridge.log",
"--log-file",
help="Write detailed debug logs to this file (set to empty to disable).",
),
workdir: Optional[str] = typer.Option(
workdir: str | None = typer.Option(
None,
"--workdir",
help="Override codex workspace (--cd) for this exec-bridge run.",
),
model: Optional[str] = typer.Option(
model: str | None = typer.Option(
None,
"--model",
help="Codex model to pass to `codex exec`.",
@@ -402,7 +403,7 @@ def run(
max_workers = config.get("max_workers")
pool = ThreadPoolExecutor(max_workers=max_workers or 4)
offset: Optional[int] = None
offset: int | None = None
ignore_backlog = bool(ignore_backlog)
if ignore_backlog:
@@ -427,7 +428,7 @@ def run(
else:
log("[startup] no chat_id configured; skipping startup message")
def handle(chat_id: int, user_msg_id: int, text: str, resume_session: Optional[str]) -> None:
def handle(chat_id: int, user_msg_id: int, text: str, resume_session: str | None) -> None:
log(
"[handle] start "
f"chat_id={chat_id} user_msg_id={user_msg_id} resume_session={resume_session!r}"
@@ -441,11 +442,11 @@ def run(
loud_final = final_notify
started_at = time.monotonic()
session_box: dict[str, Optional[str]] = {"id": resume_session}
session_box: dict[str, str | None] = {"id": resume_session}
progress_renderer = ExecProgressRenderer(max_actions=5)
progress_id: Optional[int] = None
progress: Optional[ProgressEditor] = None
progress_id: int | None = None
progress: ProgressEditor | None = None
try:
initial_text = progress_renderer.render_progress(0.0)
initial_rendered, initial_entities = render_markdown(initial_text)
@@ -472,7 +473,7 @@ def run(
initial_entities=initial_entities or None,
)
def on_event(evt: Dict[str, Any]) -> None:
def on_event(evt: dict[str, Any]) -> None:
event_type = evt.get("type")
item = evt.get("item") or {}
log_debug(
@@ -609,7 +610,7 @@ def run(
r"(?i)\\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\\b"
)
def _extract_session_id(value: Optional[str]) -> Optional[str]:
def _extract_session_id(value: str | None) -> str | None:
if not value:
return None
m = uuid_re.search(value)
codex_telegram_bridge/src/codex_telegram_bridge/rendering.py
+3 -3
View File
@@ -1,20 +1,20 @@
from __future__ import annotations
import re
from typing import Any, Dict, List, Tuple
from typing import Any
from markdown_it import MarkdownIt
from sulguk import transform_html
def render_markdown(md: str) -> Tuple[str, List[Dict[str, Any]]]:
def render_markdown(md: str) -> tuple[str, list[dict[str, Any]]]:
html = MarkdownIt("commonmark", {"html": False}).render(md or "")
rendered = transform_html(html)
text = re.sub("(?m)^(\\s*)\u2022", r"\1-", rendered.text)
# FIX: Telegram requires MessageEntity.language (if present) to be a String.
entities: List[Dict[str, Any]] = []
entities: list[dict[str, Any]] = []
for e in rendered.entities:
d = dict(e)
if "language" in d and not isinstance(d["language"], str):
codex_telegram_bridge/src/codex_telegram_bridge/telegram_client.py
+15 -15
View File
@@ -3,7 +3,7 @@ from __future__ import annotations
import json
import urllib.error
import urllib.request
from typing import Any, Dict, List, Optional
from typing import Any
class TelegramClient:
"""
@@ -16,7 +16,7 @@ class TelegramClient:
self._base = f"https://api.telegram.org/bot{token}"
self._timeout_s = timeout_s
def _call(self, method: str, params: Dict[str, Any]) -> Any:
def _call(self, method: str, params: dict[str, Any]) -> Any:
url = f"{self._base}/{method}"
data = json.dumps(params).encode("utf-8")
req = urllib.request.Request(
@@ -40,11 +40,11 @@ class TelegramClient:
def get_updates(
self,
offset: Optional[int],
offset: int | None,
timeout_s: int = 50,
allowed_updates: Optional[List[str]] = None,
) -> List[Dict[str, Any]]:
params: Dict[str, Any] = {"timeout": timeout_s}
allowed_updates: list[str] | None = None,
) -> list[dict[str, Any]]:
params: dict[str, Any] = {"timeout": timeout_s}
if offset is not None:
params["offset"] = offset
if allowed_updates is not None:
@@ -55,11 +55,11 @@ class TelegramClient:
self,
chat_id: int,
text: str,
reply_to_message_id: Optional[int] = None,
disable_notification: Optional[bool] = False,
entities: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
params: Dict[str, Any] = {
reply_to_message_id: int | None = None,
disable_notification: bool | None = False,
entities: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
params: dict[str, Any] = {
"chat_id": chat_id,
"text": text,
}
@@ -76,9 +76,9 @@ class TelegramClient:
chat_id: int,
message_id: int,
text: str,
entities: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
params: Dict[str, Any] = {
entities: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
params: dict[str, Any] = {
"chat_id": chat_id,
"message_id": message_id,
"text": text,
@@ -88,7 +88,7 @@ class TelegramClient:
return self._call("editMessageText", params)
def delete_message(self, chat_id: int, message_id: int) -> bool:
params: Dict[str, Any] = {
params: dict[str, Any] = {
"chat_id": chat_id,
"message_id": message_id,
}