@samitouri / QOSami_Agents_Zero-1
Code Commits Issues Pulls Sessions CI/CD
main
main
py 154 lines 5.45 KB
Raw
1 from __future__ import annotations
2
3 from typing import Any
4
5 from helpers import plugins
6
7
8 PLUGIN_NAME = "_oauth"
9
10 DEFAULT_CODEX_CLIENT_ID = "app_EMoamEEZ73f0CkXaXp7hrann"
11 DEFAULT_CODEX_ISSUER = "https://auth.openai.com"
12 DEFAULT_CODEX_TOKEN_URL = "https://auth.openai.com/oauth/token"
13 DEFAULT_CODEX_BASE_URL = "https://chatgpt.com/backend-api/codex"
14 DEFAULT_CODEX_SCOPES = [
15 "openid",
16 "profile",
17 "email",
18 "offline_access",
19 "api.connectors.read",
20 "api.connectors.invoke",
21 ]
22 CODEX_REASONING_EFFORTS = {"default", "minimal", "low", "medium", "high", "xhigh"}
23 CODEX_REASONING_SUMMARIES = {"off", "auto", "concise", "detailed"}
24 CODEX_TEXT_VERBOSITIES = {"default", "low", "medium", "high"}
25 DEFAULT_GEMINI_API_BASE_URL = "https://generativelanguage.googleapis.com/v1beta/openai"
26 DEFAULT_GEMINI_API_SCOPES = [
27 "openid",
28 "email",
29 "profile",
30 "https://www.googleapis.com/auth/cloud-platform",
31 "https://www.googleapis.com/auth/generative-language.retriever",
32 ]
33
34
35 def oauth_config() -> dict[str, Any]:
36 value = plugins.get_plugin_config(PLUGIN_NAME) or {}
37 return value if isinstance(value, dict) else {}
38
39
40 def codex_config(config: dict[str, Any] | None = None) -> dict[str, Any]:
41 source = config if isinstance(config, dict) else oauth_config()
42 raw = source.get("codex", {}) if isinstance(source, dict) else {}
43 raw = raw if isinstance(raw, dict) else {}
44
45 return {
46 "enabled": _as_bool(raw.get("enabled"), True),
47 "auth_file_path": _as_str(raw.get("auth_file_path")),
48 "issuer": _trim_url(raw.get("issuer"), DEFAULT_CODEX_ISSUER),
49 "token_url": _as_str(raw.get("token_url")) or DEFAULT_CODEX_TOKEN_URL,
50 "client_id": _as_str(raw.get("client_id")) or DEFAULT_CODEX_CLIENT_ID,
51 "scopes": _as_str_list(raw.get("scopes")) or DEFAULT_CODEX_SCOPES,
52 "open_browser_from_server": _as_bool(raw.get("open_browser_from_server"), False),
53 "forced_workspace_id": _as_str(raw.get("forced_workspace_id")),
54 "upstream_base_url": _trim_url(raw.get("upstream_base_url"), DEFAULT_CODEX_BASE_URL),
55 "codex_version": _as_str(raw.get("codex_version")),
56 "models": _as_str_list(raw.get("models")),
57 "request_timeout_seconds": _as_int(raw.get("request_timeout_seconds"), 120),
58 "reasoning_effort": _as_choice(
59 raw.get("reasoning_effort"), CODEX_REASONING_EFFORTS, "high"
60 ),
61 "reasoning_summary": _as_choice(
62 raw.get("reasoning_summary"), CODEX_REASONING_SUMMARIES, "auto"
63 ),
64 "text_verbosity": _as_choice(
65 raw.get("text_verbosity"), CODEX_TEXT_VERBOSITIES, "medium"
66 ),
67 "proxy_base_path": _normalize_base_path(raw.get("proxy_base_path"), "/oauth/codex"),
68 "callback_path": _normalize_base_path(raw.get("callback_path"), "/auth/callback"),
69 "require_proxy_token": _as_bool(raw.get("require_proxy_token"), False),
70 "proxy_token": _as_str(raw.get("proxy_token")),
71 }
72
73
74 def gemini_api_config(config: dict[str, Any] | None = None) -> dict[str, Any]:
75 source = config if isinstance(config, dict) else oauth_config()
76 raw = source.get("gemini_api", {}) if isinstance(source, dict) else {}
77 raw = raw if isinstance(raw, dict) else {}
78
79 return {
80 "enabled": _as_bool(raw.get("enabled"), True),
81 "client_id": _as_str(raw.get("client_id")),
82 "client_secret": _as_str(raw.get("client_secret")),
83 "scopes": _as_str_list(raw.get("scopes")) or DEFAULT_GEMINI_API_SCOPES,
84 "quota_project_id": _as_str(raw.get("quota_project_id")),
85 "api_base_url": _trim_url(raw.get("api_base_url"), DEFAULT_GEMINI_API_BASE_URL),
86 "proxy_base_path": _normalize_base_path(raw.get("proxy_base_path"), "/oauth/gemini-api"),
87 "callback_path": _normalize_base_path(raw.get("callback_path"), "/oauth/gemini-api/callback"),
88 }
89
90
91 def _as_str(value: Any) -> str:
92 if value is None:
93 return ""
94 return str(value).strip()
95
96
97 def _as_int(value: Any, default: int) -> int:
98 try:
99 return int(value)
100 except (TypeError, ValueError):
101 return default
102
103
104 def _as_bool(value: Any, default: bool) -> bool:
105 if value is None or value == "":
106 return default
107 if isinstance(value, bool):
108 return value
109 if isinstance(value, (int, float)):
110 return bool(value)
111 normalized = str(value).strip().lower()
112 if normalized in {"1", "true", "yes", "on", "enabled"}:
113 return True
114 if normalized in {"0", "false", "no", "off", "disabled"}:
115 return False
116 return default
117
118
119 def _as_choice(value: Any, choices: set[str], default: str) -> str:
120 normalized = _as_str(value).lower()
121 return normalized if normalized in choices else default
122
123
124 def _as_str_list(value: Any) -> list[str]:
125 if value is None:
126 return []
127 if isinstance(value, str):
128 values = value.replace(",", "\n").splitlines()
129 elif isinstance(value, (list, tuple, set)):
130 values = list(value)
131 else:
132 values = [value]
133
134 result: list[str] = []
135 seen: set[str] = set()
136 for item in values:
137 text = _as_str(item)
138 if not text or text in seen:
139 continue
140 seen.add(text)
141 result.append(text)
142 return result
143
144
145 def _trim_url(value: Any, default: str) -> str:
146 text = _as_str(value) or default
147 return text.rstrip("/")
148
149
150 def _normalize_base_path(value: Any, default: str) -> str:
151 text = _as_str(value) or default
152 if not text.startswith("/"):
153 text = "/" + text
154 return text.rstrip("/") or default