我如何自动化获取赠款代码的过程？
可以在没有任何手动浏览器交互的情况下完全自动化此功能？CLIENT_ID: str = "12345"
CLIENT_SECRET = "12345"
REDIRECT_URI = "https://accounts.zoho.in**"
TOKEN_URL = "https://accounts.zoho.in/oauth/v2/token"
CRM_BASE_URL = "https://www.zohoapis.in/crm/v7"
ORG_ID = "00000"
# Global variable to store refresh_token
refresh_token = None
class ZohoOAuth:
def __init__(self):
# Retrieve sensitive data from environment variables
self.CLIENT_ID: str = config("CLIENT_ID", cast=str)
self.CLIENT_SECRET = config("CLIENT_SECRET", cast=str)
self.REDIRECT_URI = config("REDIRECT_URI", cast=str)
self.SCOPE = "ZohoCRM.modules.ALL"
self.token_url = "https://accounts.zoho.in/oauth/v2/token"
async def exchange_code_for_tokens(self, code: str):
# Exchange authorization code for access and refresh tokens
payload = {
'grant_type': 'authorization_code',
'client_id': self.CLIENT_ID,
'client_secret': self.CLIENT_SECRET,
'redirect_uri': self.REDIRECT_URI,
'code': code,
}
print(payload)
try:
response = requests.post(self.token_url, data=payload)
response.raise_for_status() # Will raise an exception for bad status codes
return response.json() # Return the access and refresh tokens
except requests.exceptions.RequestException as e:
raise HTTPException(status_code=400, detail=f"Error obtaining token: {str(e)}")
@router.get("/generate_token")
async def generate_token(request: Request):
global refresh_token
try:
# Capture authorization code from Zoho redirect
code = "1000.b500427f0cafcfccc0d456b87503413c.cd5e8d693a143f0054df5bbdb4a706b9"
if not code:
raise HTTPException(status_code=400, detail="Authorization code missing")
# Ensure we await the async method
tokens = await ZohoOAuth().exchange_code_for_tokens(code)
if "error" in tokens:
raise HTTPException(status_code=400, detail=f"Zoho API Error: {tokens['error']}")
# Store the refresh_token globally after successful exchange
refresh_token = tokens.get("refresh_token")
return {
"access_token": tokens.get("access_token"),
"refresh_token": refresh_token,
"scope": tokens.get("scope"),
"api_domain": tokens.get("api_domain"),
"token_type": tokens.get("token_type"),
"expires_in": tokens.get("expires_in")
}
except requests.exceptions.RequestException as e:
print(f"Request error: {str(e)}") # Log error
raise HTTPException(status_code=500, detail="Error communicating with Zoho API")
except Exception as e:
print(f"Unexpected error: {str(e)}") # Log error
raise HTTPException(status_code=500, detail="An unexpected error occurred")
# ✅ Function to get a fresh access token using the global refresh token
def get_access_token():
global refresh_token
if not refresh_token:
raise Exception("Refresh token is missing. Please ensure that you have successfully authenticated.")
payload = {
"grant_type": "refresh_token",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
"refresh_token": refresh_token,
}
response = requests.post(TOKEN_URL, data=payload)
data = response.json()
if "access_token" in data:
return data["access_token"]
else:
raise Exception(f"Error fetching access token: {data}")
# Create Zoho Module Endpoint
@router.post("/create_account_module")
async def create_account_modules():
"""Create a new custom module in Zoho CRM."""
try:
result = create_account_module()
return result
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
# ✅ Function to fetch the first available profile ID (mandatory for module creation)
def get_profile_id():
from src.api.routes.zohocrm import get_access_token
access_token = get_access_token()
headers = {"Authorization": f"Zoho-oauthtoken {access_token}", "X-CRM-ORG": ORG_ID}
response = requests.get(f"{CRM_BASE_URL}/settings/profiles", headers=headers)
if response.status_code == 200:
profiles = response.json().get("profiles", [])
if profiles:
return profiles[0]["id"] # Use the first profile ID
else:
raise Exception("No profiles found.")
else:
raise Exception(f"Error fetching profiles: {response.json()}")
# ✅ Function to create a custom module "DigitalYou"
def create_account_module():
from src.api.routes.zohocrm import get_access_token
access_token = get_access_token() # Ensure this is synchronously called
profile_id = get_profile_id() # Ensure this is synchronously called
print(access_token)
print(profile_id)
headers = {
"Authorization": f"Zoho-oauthtoken {access_token}",
"X-CRM-ORG": ORG_ID,
"Content-Type": "application/json"
}
module_data = {}
response = requests.post(f"{CRM_BASE_URL}/settings/modules", json=module_data, headers=headers)
if response.status_code == 201:
return response.json() # ✅ Module created successfully
else:
return f"Error: {response.json()}"
Traceback (most recent call last):
File "/home/roshni/Roshni/AI Avatar/B2B/B2B-DY-Backend/src/api/routes/zohocrm.py", line 62, in generate_token
raise HTTPException(status_code=400, detail=f"Zoho API Error: {tokens['error']}")
fastapi.exceptions.HTTPException: 400: Zoho API Error: invalid_code
TypeError: 'coroutine' object is not callable
可以在没有任何手动浏览器交互的情况下完全自动化此功能？
没有，如果没有浏览器交互，我们就无法做到这一点。
在这些情况下，我们需要在某个地方持续访问和刷新令牌。这将消除每次我们需要'访问token'的需求。服务器端的身份验证流是
通过浏览器流量加生成“代码”。报名：Zoho APIAuth中的基于Server的应用用该“代码”加入访问和刷新令牌。 conce访问令牌已过期，再次使用t Refresh令牌生成另一个访问令牌。
TIP：如果您没有存储访问令牌的位置，则可以尝试使用Zoho Catalyst的连接器
。它将将访问令牌存储在缓存中，并为您处理所有有效逻辑。