Spaces:
Sleeping
Sleeping
File size: 4,885 Bytes
ea315e3 ed19666 ea315e3 bad40cf ed19666 ea315e3 4578d44 ed19666 4578d44 ea315e3 ed19666 ea315e3 ed19666 3092d1b ea315e3 3092d1b ed19666 ea315e3 8dce271 3092d1b 8dce271 ed19666 3092d1b ea315e3 3092d1b ed19666 8dce271 3092d1b 8dce271 ed19666 3092d1b ea315e3 3092d1b ea315e3 3092d1b ea315e3 ed19666 ea315e3 8dce271 3092d1b 8dce271 ea315e3 | 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 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | import requests
# Salesforce credentials
SALESFORCE_URL = "https://login.salesforce.com/services/oauth2/token"
CLIENT_ID = "3MVG9PwZx9R6_UrcDlsRUsfM9CGTVKE82QnI5Vz02b.lv3H2yjDv_wy9.nrYihWYLUD1bv08N_qCG_gxiV5y8"
CLIENT_SECRET = "94983416FB1DF4E754B56F69982F07D94D8971523E6893B8A62D75CD8E9E3A34"
USERNAME = "surendra@sathkrutha.ccom"
PASSWORD = "Lavanyanaga@123"
SECURITY_TOKEN = "z7Wvk6mys7n8XjqbYKf3bwBh7"
# Authenticate with Salesforce
def get_salesforce_access_token():
"""Authenticate with Salesforce Developer Org and get an access token."""
try:
# Authentication URL for Salesforce production and Developer Org
auth_url = "https://login.salesforce.com/services/oauth2/token"
# Payload with credentials
payload = {
"grant_type": "password",
"client_id": "3MVG9PwZx9R6_UrcDlsRUsfM9CGTVKE82QnI5Vz02b.lv3H2yjDv_wy9.nrYihWYLUD1bv08N_qCG_gxiV5y8", # Consumer Key
"client_secret": "94983416FB1DF4E754B56F69982F07D94D8971523E6893B8A62D75CD8E9E3A34", # Consumer Secret
"username": "surendra@sathkrutha.ccom", # Developer Org username
"password": "Lavanyanaga@123" + "GH9RG97LroDoLe6gAAOHaJBP", # Password + Security Token
}
# Send POST request to authenticate
response = requests.post(auth_url, data=payload)
# Print full response for debugging
print("Response Status Code:", response.status_code)
print("Response Text:", response.text)
response.raise_for_status() # Raise error for bad HTTP responses
# Parse and return the access token and instance URL
access_token = response.json().get("access_token")
instance_url = response.json().get("instance_url")
print(f"Successfully authenticated. Instance URL: {instance_url}")
return access_token, instance_url
except requests.exceptions.RequestException as e:
print(f"Error obtaining Salesforce access token: {e}")
return None, None
# Save user to Salesforce
def save_user(name, phone, email, password):
"""Save user details to Salesforce."""
access_token, instance_url = get_salesforce_access_token()
if not access_token:
print("Failed to authenticate with Salesforce.")
return False
# Check if email already exists
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
}
query = f"SELECT Id, Email__c FROM User_Login__c WHERE Email__c = '{email}'"
query_response = requests.get(
f"{instance_url}/services/data/v53.0/query", headers=headers, params={"q": query}
)
print("Query Response:", query_response.json()) # Log the query response
if query_response.status_code == 200:
records = query_response.json().get("records", [])
if records:
print(f"Email already exists in Salesforce: {email}")
return False # Email already exists
# Save new user with hashed password
hashed_password = hashpw(password.encode(), gensalt()).decode()
data = {
"Name__c": name,
"Phone__c": phone,
"Email__c": email,
"Password__c": hashed_password,
}
response = requests.post(
f"{instance_url}/services/data/v53.0/sobjects/User_Login__c",
headers=headers,
json=data,
)
print("Save Response:", response.status_code, response.json()) # Log the save response
if response.status_code == 201:
print("User created successfully in Salesforce.")
return True
else:
print("Failed to create user in Salesforce:", response.json())
return False
def check_credentials(email, password):
"""Check user credentials during login."""
access_token, instance_url = get_salesforce_access_token()
if not access_token:
print("Failed to authenticate with Salesforce.")
return False
headers = {
"Authorization": f"Bearer {access_token}",
"Content-Type": "application/json",
}
query = f"SELECT Password__c FROM User_Login__c WHERE Email__c = '{email}'"
query_response = requests.get(
f"{instance_url}/services/data/v53.0/query", headers=headers, params={"q": query}
)
print("Query Response:", query_response.json()) # Log the query response
if query_response.status_code == 200:
records = query_response.json().get("records", [])
if records:
stored_password = records[0]["Password__c"]
if checkpw(password.encode(), stored_password.encode()):
print("Login successful.")
return True
else:
print("Incorrect password.")
else:
print("Email not found in Salesforce.")
else:
print("Failed to query Salesforce:", query_response.json())
return False
|