mirror of
https://github.com/langgenius/dify.git
synced 2026-06-23 20:41:17 +08:00
47ee9f7435
Signed-off-by: Yufeng He <40085740+he-yufeng@users.noreply.github.com> Co-authored-by: QuantumGhost <obelisk.reg+git@gmail.com>
48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
import os
|
|
from typing import TypedDict
|
|
|
|
import httpx
|
|
|
|
OPERATION_REQUEST_TIMEOUT = httpx.Timeout(10.0, connect=3.0)
|
|
|
|
|
|
class UtmInfo(TypedDict, total=False):
|
|
"""Expected shape of the utm_info dict passed to record_utm.
|
|
|
|
All fields are optional; missing keys default to an empty string.
|
|
"""
|
|
|
|
utm_source: str
|
|
utm_medium: str
|
|
utm_campaign: str
|
|
utm_content: str
|
|
utm_term: str
|
|
|
|
|
|
class OperationService:
|
|
base_url = os.environ.get("BILLING_API_URL", "BILLING_API_URL")
|
|
secret_key = os.environ.get("BILLING_API_SECRET_KEY", "BILLING_API_SECRET_KEY")
|
|
|
|
@classmethod
|
|
def _send_request(cls, method, endpoint, json=None, params=None):
|
|
headers = {"Content-Type": "application/json", "Billing-Api-Secret-Key": cls.secret_key}
|
|
|
|
url = f"{cls.base_url}{endpoint}"
|
|
response = httpx.request(
|
|
method, url, json=json, params=params, headers=headers, timeout=OPERATION_REQUEST_TIMEOUT
|
|
)
|
|
|
|
return response.json()
|
|
|
|
@classmethod
|
|
def record_utm(cls, tenant_id: str, utm_info: UtmInfo):
|
|
params = {
|
|
"tenant_id": tenant_id,
|
|
"utm_source": utm_info.get("utm_source", ""),
|
|
"utm_medium": utm_info.get("utm_medium", ""),
|
|
"utm_campaign": utm_info.get("utm_campaign", ""),
|
|
"utm_content": utm_info.get("utm_content", ""),
|
|
"utm_term": utm_info.get("utm_term", ""),
|
|
}
|
|
return cls._send_request("POST", "/tenant_utms", params=params)
|