i working in my project where i try to automate twitter(Educational Purpose).After i authorize the twitter selenium will automatically capture cookies and save it for fetching my followers no need to manually login again we use chrome profile session. my code perfectly working in local machine in headless mode but after deploying on render it’s not working please see the code and guide me how to do that- CHROME_PROFILE_PATH = “C:/Users/DEEP SAHA/AppData/Local/Google/Chrome/User Data/Default” @app.get(“/api/twitter/signup/register”)
async def twitter_register():
oauth = OAuth1Session(
TWITTER_CLIENT_ID,
client_secret=TWITTER_CLIENT_SECRET,
callback_uri=TWITTER_SIGNUP_CALLBACK_URL
)
try:
request_token_url = "https://api.twitter.com/oauth/request_token"
response = oauth.fetch_request_token(request_token_url)
oauth_token = response.get('oauth_token')
if not oauth_token:
raise HTTPException(status_code=400, detail="Failed to get request token")
twitter_auth_url = f"https://api.twitter.com/oauth/authorize?oauth_token={oauth_token}"
return {"auth_url": twitter_auth_url}
except Exception as e:
raise HTTPException(status_code=400, detail=f"Error during Twitter OAuth: {str(e)}")
@app.get("/api/auth/twitter/callback")
async def twitter_callback(oauth_token: str, oauth_verifier: str, db: Session = Depends(get_db)):
oauth = OAuth1Session(TWITTER_CLIENT_ID, client_secret=TWITTER_CLIENT_SECRET)
try:
# Step 1: Exchange the oauth_token and oauth_verifier for an access token
access_token_url = "https://api.twitter.com/oauth/access_token"
token_data = {"oauth_token": oauth_token, "oauth_verifier": oauth_verifier}
response = oauth.post(access_token_url, params=token_data)
if response.status_code != 200:
raise HTTPException(status_code=400, detail="Failed to fetch access token from Twitter")
# Step 2: Parse access token data
credentials = dict(item.split("=") for item in response.text.split("&"))
access_token = credentials.get("oauth_token")
access_token_secret = credentials.get("oauth_token_secret")
# Step 3: Set up OAuth session with access tokens
oauth = OAuth1Session(
TWITTER_CLIENT_ID,
client_secret=TWITTER_CLIENT_SECRET,
resource_owner_key=access_token,
resource_owner_secret=access_token_secret
)
# Verify user credentials and fetch user info
verify_credentials_url = "https://api.twitter.com/1.1/account/verify_credentials.json"
user_info_response = oauth.get(verify_credentials_url, params={"include_email": "true"})
if user_info_response.status_code != 200:
raise HTTPException(status_code=400, detail="Failed to fetch user info from Twitter")
user_info = user_info_response.json()
user_email = user_info.get("email")
twitter_id = user_info.get("id_str")
if not user_email:
raise HTTPException(status_code=400, detail="Email not available from Twitter.")
# Configure Chrome options for headless login
chrome_options = webdriver.ChromeOptions()
chrome_options.add_argument(f"user-data-dir={CHROME_PROFILE_PATH}")
chrome_options.add_argument("--headless")
chrome_options.add_argument("--disable-gpu")
chrome_options.add_argument("--no-sandbox")
chrome_options.add_argument("--disable-dev-shm-usage")
with webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=chrome_options) as driver:
# Open Twitter homepage to establish session
driver.get("https://x.com/home")
time.sleep(3)
# Retrieve cookies after ensuring the user is logged in
cookies = driver.get_cookies()
csrf_token = next((cookie['value'] for cookie in cookies if 'ct0' in cookie['name']), None)
auth_token = next((cookie['value'] for cookie in cookies if 'auth_token' in cookie['name']), None)
if csrf_token is None or auth_token is None:
raise HTTPException(status_code=400, detail="Failed to retrieve necessary cookies")
# Process user data and handle database logic
existing_user = db.query(User).filter(User.email == user_email).first()
if existing_user:
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
jwt_token = create_access_token(data={"sub": existing_user.email}, expires_delta=access_token_expires)
formatted_cookies = [{"name": c.get("name"), "value": c.get("value")} for c in cookies]
return {
"csrf_token": csrf_token,
"auth_token": auth_token,
"cookies": json.dumps(formatted_cookies, indent=2),
"twitter_id": twitter_id,
"access_token": jwt_token,
"token_type": "bearer",
"user": {"id": existing_user.id, "email": existing_user.email, "name": existing_user.name, "bio": existing_user.bio}
}
# Handle registration for new user
new_user = User(
email=user_email,
name=user_info.get("screen_name"),
bio=user_info.get("description", ""),
twitter_handle=user_info.get("screen_name"),
twitter_id=twitter_id,
password=None
)
db.add(new_user)
db.commit()
db.refresh(new_user)
jwt_token = create_access_token(data={"sub": new_user.email}, expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES))
formatted_cookies = [{"name": c.get("name"), "value": c.get("value")} for c in cookies]
return {
"csrf_token": csrf_token,
"auth_token": auth_token,
"cookies": json.dumps(formatted_cookies, indent=2),
"twitter_id": twitter_id,
"access_token": jwt_token,
"token_type": "bearer",
"user": {"id": new_user.id, "email": new_user.email, "name": new_user.name, "bio": new_user.bio}
}
except Exception as e:
print(f"Error during Twitter callback: {str(e)}")
raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(e)}")