ilia/nanobot
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor: remove unused functions and simplify code

Browse Source
This commit is contained in:
pinhua33 2026-02-06 00:14:31 +08:00
parent d4e65319ee
commit 01420f4dd6
5 changed files with 30 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
nanobot/auth/__init__.py
View File

@ -1,13 +1,8 @@
"""Authentication modules."""
from nanobot.auth.codex import (
ensure_codex_token_available,
get_codex_token,
login_codex_oauth_interactive,
)
from nanobot.auth.codex import get_codex_token, login_codex_oauth_interactive
__all__ = [
"ensure_codex_token_available",
"get_codex_token",
"login_codex_oauth_interactive",
]

10
nanobot/auth/codex/__init__.py
View File

@ -1,15 +1,7 @@
"""Codex OAuth module."""
from nanobot.auth.codex.flow import (
ensure_codex_token_available,
get_codex_token,
login_codex_oauth_interactive,
)
from nanobot.auth.codex.models import CodexToken
from nanobot.auth.codex.flow import get_codex_token, login_codex_oauth_interactive
__all__ = [
"CodexToken",
"ensure_codex_token_available",
"get_codex_token",
"login_codex_oauth_interactive",
]

1
nanobot/auth/codex/constants.py
View File

@ -9,7 +9,6 @@ JWT_CLAIM_PATH = "https://api.openai.com/auth"
DEFAULT_ORIGINATOR = "nanobot"
TOKEN_FILENAME = "codex.json"
MANUAL_PROMPT_DELAY_SEC = 3
SUCCESS_HTML = (
"<!doctype html>"
"<html lang=\"en\">"

76
nanobot/auth/codex/flow.py
View File

@ -8,7 +8,7 @@ import threading
import time
import urllib.parse
import webbrowser
from typing import Any, Callable
from typing import Callable
import httpx
@ -16,7 +16,6 @@ from nanobot.auth.codex.constants import (
AUTHORIZE_URL,
CLIENT_ID,
DEFAULT_ORIGINATOR,
MANUAL_PROMPT_DELAY_SEC,
REDIRECT_URI,
SCOPE,
TOKEN_URL,
@ -39,31 +38,6 @@ from nanobot.auth.codex.storage import (
)
def _exchange_code_for_token(code: str, verifier: str) -> CodexToken:
data = {
"grant_type": "authorization_code",
"client_id": CLIENT_ID,
"code": code,
"code_verifier": verifier,
"redirect_uri": REDIRECT_URI,
}
with httpx.Client(timeout=30.0) as client:
response = client.post(TOKEN_URL, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"})
if response.status_code != 200:
raise RuntimeError(f"Token exchange failed: {response.status_code} {response.text}")
payload = response.json()
access, refresh, expires_in = _parse_token_payload(payload, "Token response missing fields")
print("Received access token:", access)
account_id = _decode_account_id(access)
return CodexToken(
access=access,
refresh=refresh,
expires=int(time.time() * 1000 + expires_in * 1000),
account_id=account_id,
)
async def _exchange_code_for_token_async(code: str, verifier: str) -> CodexToken:
data = {
"grant_type": "authorization_code",
@ -146,11 +120,6 @@ def get_codex_token() -> CodexToken:
raise
def ensure_codex_token_available() -> None:
"""Ensure a valid token is available; raise if not."""
_ = get_codex_token()
async def _read_stdin_line() -> str:
loop = asyncio.get_running_loop()
if hasattr(loop, "add_reader") and sys.stdin:
@ -177,20 +146,14 @@ async def _read_stdin_line() -> str:
return await loop.run_in_executor(None, sys.stdin.readline)
async def _await_manual_input(
on_manual_code_input: Callable[[str], None],
) -> str:
await asyncio.sleep(MANUAL_PROMPT_DELAY_SEC)
on_manual_code_input("Paste the authorization code (or full redirect URL), or wait for the browser callback:")
async def _await_manual_input(print_fn: Callable[[str], None]) -> str:
print_fn("[cyan]Paste the authorization code (or full redirect URL), or wait for the browser callback:[/cyan]")
return await _read_stdin_line()
def login_codex_oauth_interactive(
on_auth: Callable[[str], None] | None = None,
on_prompt: Callable[[str], str] | None = None,
on_status: Callable[[str], None] | None = None,
on_progress: Callable[[str], None] | None = None,
on_manual_code_input: Callable[[str], None] = None,
print_fn: Callable[[str], None],
prompt_fn: Callable[[str], str],
originator: str = DEFAULT_ORIGINATOR,
) -> CodexToken:
"""Interactive login flow."""
@ -222,27 +185,30 @@ def login_codex_oauth_interactive(
loop.call_soon_threadsafe(code_future.set_result, code_value)
server, server_error = _start_local_server(state, on_code=_notify)
if on_auth:
on_auth(url)
else:
print_fn("[cyan]A browser window will open for login. If it doesn't, open this URL manually:[/cyan]")
print_fn(url)
try:
webbrowser.open(url)
except Exception:
pass
if not server and server_error and on_status:
on_status(
if not server and server_error:
print_fn(
"[yellow]"
f"Local callback server could not start ({server_error}). "
"You will need to paste the callback URL or authorization code."
"[/yellow]"
)
code: str | None = None
try:
if server:
if on_progress and not on_manual_code_input:
on_progress("Waiting for browser callback...")
print_fn("[dim]Waiting for browser callback...[/dim]")
tasks: list[asyncio.Task[Any]] = []
tasks: list[asyncio.Task[object]] = []
callback_task = asyncio.create_task(asyncio.wait_for(code_future, timeout=120))
tasks.append(callback_task)
manual_task = asyncio.create_task(_await_manual_input(on_manual_code_input))
manual_task = asyncio.create_task(_await_manual_input(print_fn))
tasks.append(manual_task)
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
@ -268,10 +234,7 @@ def login_codex_oauth_interactive(
if not code:
prompt = "Please paste the callback URL or authorization code:"
if on_prompt:
raw = await loop.run_in_executor(None, on_prompt, prompt)
else:
raw = await loop.run_in_executor(None, input, prompt)
raw = await loop.run_in_executor(None, prompt_fn, prompt)
parsed_code, parsed_state = _parse_authorization_input(raw)
if parsed_state and parsed_state != state:
raise RuntimeError("State validation failed.")
@ -280,8 +243,7 @@ def login_codex_oauth_interactive(
if not code:
raise RuntimeError("Authorization code not found.")
if on_progress:
on_progress("Exchanging authorization code for tokens...")
print_fn("[dim]Exchanging authorization code for tokens...[/dim]")
token = await _exchange_code_for_token_async(code, verifier)
_save_token_file(token)
return token

42
nanobot/cli/commands.py
View File

@ -84,37 +84,12 @@ def login(
from nanobot.auth.codex import login_codex_oauth_interactive
def on_auth(url: str) -> None:
console.print("[cyan]A browser window will open for login. If it doesn't, open this URL manually:[/cyan]")
console.print(url)
try:
import webbrowser
webbrowser.open(url)
except Exception:
pass
def on_status(message: str) -> None:
console.print(f"[yellow]{message}[/yellow]")
def on_progress(message: str) -> None:
console.print(f"[dim]{message}[/dim]")
def on_prompt(message: str) -> str:
return typer.prompt(message)
def on_manual_code_input(message: str) -> None:
console.print(f"[cyan]{message}[/cyan]")
console.print("[green]Starting OpenAI Codex OAuth login...[/green]")
login_codex_oauth_interactive(
on_auth=on_auth,
on_prompt=on_prompt,
on_status=on_status,
on_progress=on_progress,
on_manual_code_input=on_manual_code_input,
print_fn=console.print,
prompt_fn=typer.prompt,
)
console.print("[green]✓ Login successful. Credentials saved.[/green]")
console.print("[green]Login successful. Credentials saved.[/green]")
@ -205,7 +180,7 @@ def gateway(
from nanobot.bus.queue import MessageBus
from nanobot.providers.litellm_provider import LiteLLMProvider
from nanobot.providers.openai_codex_provider import OpenAICodexProvider
from nanobot.auth.codex import ensure_codex_token_available
from nanobot.auth.codex import get_codex_token
from nanobot.agent.loop import AgentLoop
from nanobot.channels.manager import ChannelManager
from nanobot.cron.service import CronService
@ -232,7 +207,7 @@ def gateway(
if is_codex:
try:
ensure_codex_token_available()
_ = get_codex_token()
except Exception as e:
console.print(f"[red]Error: {e}[/red]")
console.print("Please run: [cyan]nanobot login --provider openai-codex[/cyan]")
@ -341,7 +316,7 @@ def agent(
from nanobot.bus.queue import MessageBus
from nanobot.providers.litellm_provider import LiteLLMProvider
from nanobot.providers.openai_codex_provider import OpenAICodexProvider
from nanobot.auth.codex import ensure_codex_token_available
from nanobot.auth.codex import get_codex_token
from nanobot.agent.loop import AgentLoop
config = load_config()
@ -354,7 +329,7 @@ def agent(
if is_codex:
try:
ensure_codex_token_available()
_ = get_codex_token()
except Exception as e:
console.print(f"[red]Error: {e}[/red]")
console.print("Please run: [cyan]nanobot login --provider openai-codex[/cyan]")
@ -716,9 +691,10 @@ def status():
console.print(f"Anthropic API: {'[green]✓[/green]' if has_anthropic else '[dim]not set[/dim]'}")
console.print(f"OpenAI API: {'[green]✓[/green]' if has_openai else '[dim]not set[/dim]'}")
console.print(f"Gemini API: {'[green]✓[/green]' if has_gemini else '[dim]not set[/dim]'}")
vllm_status = f"[green]✓ {config.providers.vllm.api_base}[/green]" if has_vllm else "[dim]not set[/dim]"
vllm_status = f"[green]<EFBFBD>?{config.providers.vllm.api_base}[/green]" if has_vllm else "[dim]not set[/dim]"
console.print(f"vLLM/Local: {vllm_status}")
if __name__ == "__main__":
app()