diff --git a/cozepy/__init__.py b/cozepy/__init__.py index eba108f..a3784b9 100644 --- a/cozepy/__init__.py +++ b/cozepy/__init__.py @@ -18,6 +18,7 @@ Scope, TokenAuth, WebOAuthApp, + load_oauth_app_from_config, ) from .bots import ( Bot, @@ -159,6 +160,7 @@ # audio.transcriptions "CreateTranscriptionsResp", # auth + "load_oauth_app_from_config", "AsyncDeviceOAuthApp", "AsyncJWTOAuthApp", "AsyncPKCEOAuthApp", diff --git a/cozepy/auth/__init__.py b/cozepy/auth/__init__.py index c6777d2..b28e9ae 100644 --- a/cozepy/auth/__init__.py +++ b/cozepy/auth/__init__.py @@ -1,6 +1,7 @@ import abc +import json import time -from typing import List, Optional +from typing import List, Optional, Union from urllib.parse import quote_plus, urlparse from authlib.jose import jwt # type: ignore @@ -660,6 +661,28 @@ async def refresh_access_token(self, refresh_token: str) -> OAuthToken: return await self._arefresh_access_token(refresh_token) +def load_oauth_app_from_config(conf: str) -> Union[PKCEOAuthApp, JWTOAuthApp, DeviceOAuthApp, WebOAuthApp]: + config = json.loads(conf) + client_id = config.get("client_id", "") + client_type = config.get("client_type", "") + coze_api_base = config.get("coze_api_base", "") + coze_www_base = config.get("coze_www_base", "") + + if client_type == "single_page": + return PKCEOAuthApp(client_id, coze_api_base, coze_www_base) + elif client_type == "server": + private_key = config.get("private_key", "") + public_key_id = config.get("public_key_id", "") + return JWTOAuthApp(client_id, private_key, public_key_id, coze_api_base) + elif client_type == "device": + return DeviceOAuthApp(client_id, coze_api_base, coze_www_base) + elif client_type == "web": + client_secret = config.get("client_secret", "") + return WebOAuthApp(client_id, client_secret, coze_api_base, coze_www_base) + else: + raise ValueError(f"Invalid OAuth client_type: {client_type}") + + class Auth(abc.ABC): """ This class is the base class for all authorization types. diff --git a/examples/chat_stream.py b/examples/chat_stream.py index 3bc9ff5..b50607a 100644 --- a/examples/chat_stream.py +++ b/examples/chat_stream.py @@ -4,18 +4,37 @@ """ import os +from typing import Optional -from cozepy import COZE_COM_BASE_URL -from examples.utils import get_coze_api_token +from cozepy import COZE_CN_BASE_URL, ChatEventType, Coze, DeviceOAuthApp, Message, TokenAuth # noqa -# The default access is api.coze.com, but if you need to access api.coze.cn, -# please use base_url to configure the api endpoint to access -coze_api_base = os.getenv("COZE_API_BASE") or COZE_COM_BASE_URL -from cozepy import Coze, TokenAuth, Message, ChatStatus, MessageContentType, ChatEventType # noqa +def get_coze_api_base() -> str: + # The default access is api.coze.com, but if you need to access api.coze.cn, + # please use base_url to configure the api endpoint to access + coze_api_base = os.getenv("COZE_API_BASE") + if coze_api_base: + return coze_api_base + + return COZE_CN_BASE_URL # default + + +def get_coze_api_token(workspace_id: Optional[str] = None) -> str: + # Get an access_token through personal access token or oauth. + coze_api_token = os.getenv("COZE_API_TOKEN") + if coze_api_token: + return coze_api_token + + coze_api_base = get_coze_api_base() + + device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base) + device_code = device_oauth_app.get_device_code(workspace_id) + print(f"Please Open: {device_code.verification_url} to get the access token") + return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token + # Init the Coze client through the access_token. -coze = Coze(auth=TokenAuth(token=get_coze_api_token()), base_url=coze_api_base) +coze = Coze(auth=TokenAuth(token=get_coze_api_token()), base_url=get_coze_api_base()) # Create a bot instance in Coze, copy the last number from the web link as the bot's ID. bot_id = os.getenv("COZE_BOT_ID") or "bot id" diff --git a/examples/utils/__init__.py b/examples/utils/__init__.py deleted file mode 100644 index 8753676..0000000 --- a/examples/utils/__init__.py +++ /dev/null @@ -1,35 +0,0 @@ -import logging -import os -from typing import Optional - -from cozepy import COZE_CN_BASE_URL, DeviceOAuthApp, setup_logging - - -def get_coze_api_base() -> str: - # The default access is api.coze.com, but if you need to access api.coze.cn, - # please use base_url to configure the api endpoint to access - coze_api_base = os.getenv("COZE_API_BASE") - if coze_api_base: - return coze_api_base - - return COZE_CN_BASE_URL # default - - -def get_coze_api_token(workspace_id: Optional[str] = None) -> str: - # Get an access_token through personal access token or oauth. - coze_api_token = os.getenv("COZE_API_TOKEN") - if coze_api_token: - return coze_api_token - - coze_api_base = get_coze_api_base() - - device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base) - device_code = device_oauth_app.get_device_code(workspace_id) - print(f"Please Open: {device_code.verification_url} to get the access token") - return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token - - -def setup_examples_logger(): - coze_log = os.getenv("COZE_LOG") - if coze_log: - setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO)) diff --git a/examples/websockets_audio_speech.py b/examples/websockets_audio_speech.py index 108bb50..548dc83 100644 --- a/examples/websockets_audio_speech.py +++ b/examples/websockets_audio_speech.py @@ -1,27 +1,61 @@ import asyncio import json +import logging import os +from typing import Optional from cozepy import ( + COZE_CN_BASE_URL, AsyncCoze, AsyncWebsocketsAudioSpeechClient, AsyncWebsocketsAudioSpeechEventHandler, + DeviceOAuthApp, InputTextBufferAppendEvent, InputTextBufferCompletedEvent, SpeechAudioCompletedEvent, SpeechAudioUpdateEvent, TokenAuth, + setup_logging, ) from cozepy.log import log_info from cozepy.util import write_pcm_to_wav_file -from examples.utils import get_coze_api_base, get_coze_api_token, setup_examples_logger + + +def get_coze_api_base() -> str: + # The default access is api.coze.com, but if you need to access api.coze.cn, + # please use base_url to configure the api endpoint to access + coze_api_base = os.getenv("COZE_API_BASE") + if coze_api_base: + return coze_api_base + + return COZE_CN_BASE_URL # default + + +def get_coze_api_token(workspace_id: Optional[str] = None) -> str: + # Get an access_token through personal access token or oauth. + coze_api_token = os.getenv("COZE_API_TOKEN") + if coze_api_token: + return coze_api_token + + coze_api_base = get_coze_api_base() + + device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base) + device_code = device_oauth_app.get_device_code(workspace_id) + print(f"Please Open: {device_code.verification_url} to get the access token") + return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token + + +def setup_examples_logger(): + coze_log = os.getenv("COZE_LOG") + if coze_log: + setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO)) + setup_examples_logger() kwargs = json.loads(os.getenv("COZE_KWARGS") or "{}") -# todo review class AsyncWebsocketsAudioSpeechEventHandlerSub(AsyncWebsocketsAudioSpeechEventHandler): """ Class is not required, you can also use Dict to set callback diff --git a/examples/websockets_audio_transcriptions.py b/examples/websockets_audio_transcriptions.py index 9828dd3..9127369 100644 --- a/examples/websockets_audio_transcriptions.py +++ b/examples/websockets_audio_transcriptions.py @@ -1,19 +1,54 @@ import asyncio import json +import logging import os +from typing import Optional from cozepy import ( + COZE_CN_BASE_URL, AsyncCoze, AsyncWebsocketsAudioTranscriptionsClient, AsyncWebsocketsAudioTranscriptionsEventHandler, AudioFormat, + DeviceOAuthApp, InputAudioBufferAppendEvent, InputAudioBufferCompletedEvent, TokenAuth, TranscriptionsMessageUpdateEvent, + setup_logging, ) from cozepy.log import log_info -from examples.utils import get_coze_api_base, get_coze_api_token, setup_examples_logger + + +def get_coze_api_base() -> str: + # The default access is api.coze.com, but if you need to access api.coze.cn, + # please use base_url to configure the api endpoint to access + coze_api_base = os.getenv("COZE_API_BASE") + if coze_api_base: + return coze_api_base + + return COZE_CN_BASE_URL # default + + +def get_coze_api_token(workspace_id: Optional[str] = None) -> str: + # Get an access_token through personal access token or oauth. + coze_api_token = os.getenv("COZE_API_TOKEN") + if coze_api_token: + return coze_api_token + + coze_api_base = get_coze_api_base() + + device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base) + device_code = device_oauth_app.get_device_code(workspace_id) + print(f"Please Open: {device_code.verification_url} to get the access token") + return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token + + +def setup_examples_logger(): + coze_log = os.getenv("COZE_LOG") + if coze_log: + setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO)) + setup_examples_logger() diff --git a/examples/websockets_chat.py b/examples/websockets_chat.py index 7382e79..d216773 100644 --- a/examples/websockets_chat.py +++ b/examples/websockets_chat.py @@ -1,8 +1,11 @@ import asyncio import json +import logging import os +from typing import Optional from cozepy import ( + COZE_CN_BASE_URL, AsyncCoze, AsyncWebsocketsChatClient, AsyncWebsocketsChatEventHandler, @@ -13,13 +16,45 @@ ConversationChatRequiresActionEvent, ConversationChatSubmitToolOutputsEvent, ConversationMessageDeltaEvent, + DeviceOAuthApp, InputAudioBufferAppendEvent, TokenAuth, ToolOutput, + setup_logging, ) from cozepy.log import log_info from cozepy.util import write_pcm_to_wav_file -from examples.utils import get_coze_api_base, get_coze_api_token, setup_examples_logger + + +def get_coze_api_base() -> str: + # The default access is api.coze.com, but if you need to access api.coze.cn, + # please use base_url to configure the api endpoint to access + coze_api_base = os.getenv("COZE_API_BASE") + if coze_api_base: + return coze_api_base + + return COZE_CN_BASE_URL # default + + +def get_coze_api_token(workspace_id: Optional[str] = None) -> str: + # Get an access_token through personal access token or oauth. + coze_api_token = os.getenv("COZE_API_TOKEN") + if coze_api_token: + return coze_api_token + + coze_api_base = get_coze_api_base() + + device_oauth_app = DeviceOAuthApp(client_id="57294420732781205987760324720643.app.coze", base_url=coze_api_base) + device_code = device_oauth_app.get_device_code(workspace_id) + print(f"Please Open: {device_code.verification_url} to get the access token") + return device_oauth_app.get_access_token(device_code=device_code.device_code, poll=True).access_token + + +def setup_examples_logger(): + coze_log = os.getenv("COZE_LOG") + if coze_log: + setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO)) + setup_examples_logger() diff --git a/examples/websockets_chat_realtime_gui.py b/examples/websockets_chat_realtime_gui.py index 5f4ba3a..0d858d6 100644 --- a/examples/websockets_chat_realtime_gui.py +++ b/examples/websockets_chat_realtime_gui.py @@ -1,5 +1,6 @@ import asyncio import json +import logging import os import queue import threading @@ -18,11 +19,11 @@ ChatUpdateEvent, ConversationAudioDeltaEvent, ConversationChatCompletedEvent, + InputAudio, InputAudioBufferAppendEvent, TokenAuth, + setup_logging, ) -from cozepy.websockets.ws import InputAudio -from examples.utils import setup_examples_logger # 音频参数设置 CHUNK = 1024 @@ -31,6 +32,13 @@ RATE = 24000 INPUT_BLOCK_TIME = 0.05 # 50ms per block + +def setup_examples_logger(): + coze_log = os.getenv("COZE_LOG") + if coze_log: + setup_logging(logging.getLevelNamesMapping().get(coze_log.upper(), logging.INFO)) + + setup_examples_logger()