mirror of
https://github.com/langgenius/dify.git
synced 2026-09-01 21:55:46 +08:00
40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
"""
|
|
Helper module for Creators Platform integration.
|
|
|
|
Provides functionality to upload DSL files to the Creators Platform
|
|
and generate redirect URLs with OAuth authorization codes.
|
|
"""
|
|
|
|
import logging
|
|
from urllib.parse import urlencode
|
|
|
|
import httpx
|
|
from yarl import URL
|
|
|
|
from configs import dify_config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
creators_platform_api_url = URL(str(dify_config.CREATORS_PLATFORM_API_URL))
|
|
|
|
_CREATORS_REQUEST_TIMEOUT: httpx.Timeout = httpx.Timeout(30.0, connect=5.0)
|
|
|
|
|
|
def upload_dsl(dsl_file_bytes: bytes, filename: str = "template.yaml") -> str:
|
|
url = str(creators_platform_api_url / "api/v1/templates/anonymous-upload")
|
|
response = httpx.post(url, files={"file": (filename, dsl_file_bytes)}, timeout=_CREATORS_REQUEST_TIMEOUT)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
claim_code = data.get("data", {}).get("claim_code")
|
|
if not isinstance(claim_code, str) or not claim_code:
|
|
raise ValueError("Creators Platform did not return a valid claim_code")
|
|
return claim_code
|
|
|
|
|
|
def get_redirect_url(claim_code: str, *, oauth_code: str | None = None) -> str:
|
|
base_url = str(dify_config.CREATORS_PLATFORM_API_URL).rstrip("/")
|
|
params: dict[str, str] = {"dsl_claim_code": claim_code}
|
|
if oauth_code:
|
|
params["oauth_code"] = oauth_code
|
|
return f"{base_url}?{urlencode(params)}"
|