File size: 3,641 Bytes
9b4503a e485414 9b4503a e485414 |
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 |
import json
import os
from google.oauth2 import service_account
from google.cloud import storage
from google import genai
from storage_service import GoogleCloudStorage
def initialize_google_credentials():
is_env_local = os.getenv("IS_ENV_LOCAL", "false") == "true"
print(f"Environment is local: {is_env_local}")
try:
if is_env_local:
config_path = os.path.join(os.path.dirname(__file__), "local_config.json")
print(f"Trying to load config from: {config_path}")
if not os.path.exists(config_path):
print(f"Warning: {config_path} does not exist")
return None
with open(config_path) as f:
config = json.load(f)
google_credentials_key = json.dumps(config["GOOGLE_APPLICATION_CREDENTIALS_JSON"])
else:
google_credentials_key = os.getenv("GOOGLE_APPLICATION_CREDENTIALS_JSON")
if not google_credentials_key:
print("Warning: GOOGLE_APPLICATION_CREDENTIALS_JSON environment variable not set")
return google_credentials_key
except Exception as e:
print(f"Error initializing credentials: {str(e)}")
return None
def initialize_gcs_service(google_credentials_key):
if not google_credentials_key:
print("Warning: No credentials provided, GCS service will not be initialized")
return None
return GoogleCloudStorage(google_credentials_key)
def initialize_genai_client(google_credentials_key):
try:
if not google_credentials_key:
print("Warning: No credentials provided, using default authentication")
return genai.Client(
vertexai=True,
project='junyiacademy',
location='us-central1'
)
google_service_account_info_dict = json.loads(google_credentials_key)
GOOGPE_SCOPES = ["https://www.googleapis.com/auth/cloud-platform"]
credentials = service_account.Credentials.from_service_account_info(
google_service_account_info_dict, scopes=GOOGPE_SCOPES
)
return genai.Client(
vertexai=True,
project='junyiacademy',
location='us-central1',
credentials=credentials
)
except Exception as e:
print(f"Error initializing GenAI client: {str(e)}")
print("Falling back to default authentication")
return genai.Client(
vertexai=True,
project='junyiacademy',
location='us-central1'
)
def initialize_password():
is_env_local = os.getenv("IS_ENV_LOCAL", "false") == "true"
print(f"Environment is local: {is_env_local}")
try:
if is_env_local:
config_path = os.path.join(os.path.dirname(__file__), "local_config.json")
print(f"Trying to load config from: {config_path}")
if not os.path.exists(config_path):
print(f"Warning: {config_path} does not exist")
return None
with open(config_path) as f:
config = json.load(f)
return config["PASSWORD"]
else:
return os.getenv("PASSWORD")
except Exception as e:
print(f"Error initializing password: {str(e)}")
return None
def initialize_clients():
google_credentials_key = initialize_google_credentials()
gcs_service = initialize_gcs_service(google_credentials_key)
genai_client = initialize_genai_client(google_credentials_key)
return gcs_service, genai_client
|