File size: 3,649 Bytes
34d557c 16d445c c8920fc 34d557c e7f1495 34d557c e7f1495 34d557c e7f1495 34d557c e7f1495 34d557c e7f1495 bda97e5 e7f1495 34d557c e7f1495 34d557c bda97e5 e7f1495 c8920fc 16d445c bda97e5 e7f1495 bda97e5 e7f1495 bda97e5 c8920fc e7f1495 c8920fc e7f1495 16d445c e7f1495 5249897 e7f1495 c8920fc e7f1495 c8920fc e7f1495 16d445c e7f1495 34d557c e7f1495 34d557c 5249897 e7f1495 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
import json
import time
import requests
import jwt
def get_access_token(client_email, private_key):
current_time = int(time.time())
expiration_time = current_time + 600
claims = {
'iss': client_email,
'scope': 'https://www.googleapis.com/auth/cloud-platform',
'aud': 'https://oauth2.googleapis.com/token',
'exp': expiration_time,
'iat': current_time
}
try:
signed_jwt = jwt.encode(
claims,
private_key,
algorithm='RS256'
)
except Exception as e:
return False, e
response = requests.post(
'https://oauth2.googleapis.com/token',
data={
'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
'assertion': signed_jwt
}
)
if response.status_code == 200:
return True, response.json()['access_token']
else:
return False, response.text
def get_access_token_refresh(client_id, client_secret, refresh_token):
token_url = "https://oauth2.googleapis.com/token"
data = {
"client_id": client_id,
"client_secret": client_secret,
"refresh_token": refresh_token,
"grant_type": "refresh_token"
}
response = requests.post(token_url, data=data)
if response.status_code == 200:
return True, response.json()["access_token"]
else:
return False, response.text
def get_gemini_models(key):
url_getListModel = f"https://generativelanguage.googleapis.com/v1beta/models?key={key}&pageSize=1000"
response = requests.get(url_getListModel)
if response.status_code == 200:
models = response.json()
model_list = []
for model in models['models']:
#model_list[model['name'].split('/')[1]] = model['displayName']
model_name = f"{model['name'].split('/')[1]}" # ({model['displayName']})"
model_list.append(model_name)
return model_list
else:
return ""
def send_fake_gemini_request(key, model='gemini-1.5-flash'):
url = f'https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={key}'
payload = {
"contents": [{ "role": "user", "parts": [{ "text": "" }] }],
"generationConfig": {
"maxOutputTokens": 0,
}
}
try:
response = requests.post(url=url, headers={'Content-Type': 'application/json'}, json=payload)
err_msg = response.json().get('error', '')
return err_msg
except:
return None
def send_gemini_request(key, payload, model='gemini-1.5-flash'):
url = f'https://generativelanguage.googleapis.com/v1beta/models/{model}:generateContent?key={key}'
response = requests.post(url=url, headers={'Content-Type': 'application/json'}, json=payload)
if response.status_code == 200:
return True, response.json()
else:
return False, response.text
async def send_gcp_request(session, project_id, access_token, payload, region='us-east5', model='claude-3-5-sonnet@20240620'):
VERTEX_URL = f'https://{region}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{region}/publishers/anthropic/models/{model}:streamRawPredict'
headers = {
'Authorization': f'Bearer {access_token}',
'Content-Type': 'application/json; charset=utf-8'
}
async with session.post(url=VERTEX_URL, headers=headers, data=payload) as response:
if response.status != 200:
response_data = await response.text()
return json.loads(response_data)
return await response.json() |