ShaggyBro
commented
2 months ago To get your bearer token on chrome based browers: -Go to prolific and login, -Press F12 -Go to the 'Network' Tab -filter by 'internal-api.prolific.com/api/v1/' -click F5 -click on any result and search on the "Request Headers" for something like
authorization: Bearer xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
-copy it (eg Bearer JABSKJFBGDBGDBSKDBGKBSDGBSDGNB) -parse it (eg paste it into the address bar of your browser and recopy it) -your config.json should look something like this
{ "auto_renew_bearer" : "True", "pause_on_idle" : "False", "Prolific_ID" : "AAAAAAAAAAAAA", "proxy" : "http://username:password@server:port", "wait_time" : 12 }
-run -paste token
Notes:
if you dont use a proxy just leave it blank "proxy" : "",
pause on idle is not working, i dont actually use it
this is a frankenstein version, expect things to be janky and made in a dumb way
guessing the re-entry of the token when it expires wont work as well if idle_pause isnt working, so you will have to close the script and re run it.
import platform
import random
from argparse import ArgumentParser
from distutils.util import strtobool
from json import load
from pathlib import Path
from time import sleep, time
from webbrowser import open_new_tab
from playsound import playsound
from requests import Response, get, post
from requests.structures import CaseInsensitiveDict
from requests.exceptions import RequestException
from rich.console import Console
from rich.text import Text
if platform.system() == "Windows":
import ctypes
config = load(open(f"{Path(__file__).parent}/config/config.json"))
class ProlificUpdater:
def __init__(self, bearer, proxy=None):
self.bearer = bearer
self.proxy = proxy
self.oldResults = list()
self.participantId = config["Prolific_ID"]
self.last_ip = None
def getRequestFromProlific(self) -> Response:
url = "https://internal-api.prolific.com/api/v1/participant/studies/"
headers = CaseInsensitiveDict()
headers["Accept"] = "application/json, text/plain, */*"
headers["Authorization"] = self.bearer
headers["x-legacy-auth"] = "false"
proxies = {"http": self.proxy, "https": self.proxy} if self.proxy else None
try:
return get(url, headers=headers, proxies=proxies, timeout=10)
except RequestException as e:
console.print(f"[bold red]Request error: {e}[/bold red]")
return Response()
def reservePlace(self, id) -> Response:
url = "https://internal-api.prolific.com/api/v1/submissions/reserve/"
headers = CaseInsensitiveDict()
headers["Accept"] = "application/json"
headers["Authorization"] = self.bearer
headers["x-legacy-auth"] = "false"
postObj = {"study_id": id, "participant_id": self.participantId}
proxies = {"http": self.proxy, "https": self.proxy} if self.proxy else None
try:
return post(url, headers=headers, data=postObj, proxies=proxies)
except RequestException as e:
console.print(f"[bold red]Request error: {e}[/bold red]")
return Response()
def getResultsFromProlific(self) -> list:
try:
response = self.getRequestFromProlific()
except Exception:
console.print("[bold red][+] Network error[/bold red]")
return list()
if response.status_code == 200:
return response.json()["results"]
else:
print(f"Response error {response.status_code}")
print(f"Response error {response.reason}")
return ["bearer"]
def executeCycle(self) -> bool:
results = self.getResultsFromProlific()
if results == ["bearer"]:
self.bearer = input("Bearer token not valid anymore, please enter a new bearer token: ")
self.bearer = "Bearer " + self.bearer
return False
print("results : ", results)
if results:
if results != self.oldResults:
currency_symbol = "£" if str(results[0]["study_reward"]["currency"]) == "GBP" else "$"
console.print(
f"""Trying to join {results[0]["name"]} ([bold green]{currency_symbol}{str(float(results[0]["study_reward"]["amount"])/100)}[/bold green])"""
)
reserve_place_res = self.reservePlace(id=results[0]["id"])
if reserve_place_res.status_code == 400:
console.print(f"""[bold red][+] Error code {str(reserve_place_res.json()["error"]["error_code"])}
\nTitle : {str(reserve_place_res.json()["error"]["title"])}
\nDetails : {str(reserve_place_res.json()["error"]["detail"])}""")
self.oldResults = results
return False
else:
a_website = "https://app.prolific.com/studies"
open_new_tab(a_website)
playsound(rf"{Path(__file__).parent}\alert.wav", True)
self.oldResults = results
if results:
return True
else:
return False
def check_ip(self) -> str:
ip_services = [
'https://api.ipify.org?format=json',
'https://api.my-ip.io/ip.json',
'https://checkip.amazonaws.com',
'https://api.ipstack.com/check?access_key=YOUR_ACCESS_KEY'
]
for service in ip_services:
try:
response = get(service, proxies={"http": self.proxy, "https": self.proxy} if self.proxy else None)
if response.status_code == 200:
return response.json().get("ip", response.text.strip())
except RequestException:
continue
return "Unable to get IP"
def parseArgs() -> dict:
parser = ArgumentParser(description="Keep updated with Prolific")
parser.add_argument("-b", "--bearer", type=str, help="bearer token")
args = parser.parse_args()
bearer = "Bearer " + args.bearer if args.bearer else None
return {"bearer": bearer}
if platform.system() == "Windows":
class LASTINPUTINFO(ctypes.Structure):
_fields_ = [("cbSize", ctypes.c_uint), ("dwTime", ctypes.c_uint)]
def get_idle_duration():
lii = LASTINPUTINFO()
lii.cbSize = ctypes.sizeof(LASTINPUTINFO)
if ctypes.windll.user32.GetLastInputInfo(ctypes.byref(lii)):
millis = ctypes.windll.kernel32.GetTickCount() - lii.dwTime
seconds = millis / 1000
return seconds
else:
error_code = (
ctypes.windll.kernel32.GetLastError()
) # Get the last error code
print(f"Error code: {error_code}")
return 0 # Handle the error as needed
def pause_script():
print("Pausing script...")
while True:
idle_duration = get_idle_duration()
if idle_duration > 600: # 10 minutes in seconds
print(
f"Idle Time: {idle_duration} seconds [script paused after 10 minutes of inactivity, to avoid temp ban from the API due to overuse]"
)
sleep(10)
else:
print("Resuming script...")
break
if __name__ == "__main__":
myArguments = parseArgs()
bearer = myArguments.get("bearer") or input("Please enter your bearer token: ")
bearer = "Bearer " + bearer
proxy = config.get("proxy")
p_updater = ProlificUpdater(bearer=bearer, proxy=proxy)
# Initial IP check
ip_address = p_updater.check_ip()
if ip_address == "Unable to get IP":
console = Console()
console.print(f"[bold red]Unable to get IP from all services. Stopping script.[/bold red]")
exit()
p_updater.last_ip = ip_address
console = Console()
console.print(f"Current IP address: {ip_address}")
status = console.status("[bold blue] Waiting for study...", spinner="arc")
status.start()
# Initial sleep
sleep(5)
while True:
updateTime = config["wait_time"]
if p_updater.executeCycle():
status.stop()
text = Text("Study found !")
text.stylize("bold red")
console.print(text)
sleep(5)
input("Press enter to resume study search")
status.start()
if strtobool(config["pause_on_idle"]) and platform.system() == "Windows":
idle_duration = get_idle_duration()
if idle_duration >= 600: # 10 minutes in seconds
pause_script()
# else:
# print(f"Idle Time: {idle_duration} seconds [script not paused]") #leave commented if not debugging
# IP check every 2 minutes
current_ip = p_updater.check_ip()
if current_ip != p_updater.last_ip:
console.print(f"[bold red]IP address has changed from {p_updater.last_ip} to {current_ip}. Stopping script.[/bold red]")
exit()
# Generate a random integer between -5 and 5
random_offset = random.randint(-1, 3)
# Generate a random number which is +/- 5 from the updateTime variable
random_time = updateTime + random_offset
sleep(
random_time if random_time >= 0 else 0
) # if statement covers edge case, if the user sets up <5 wait_time (I found <20 results in a temp ban), which with the use of random above, could result in a negative number being passed to the sleep()
amitader
profbiyi
NoeKun001
notquitemoe
SurfingTurtles
ThisIsNotUdyr
terefere2033
Arslan30
yes sir, my workaround currently was getting the token manually from prolific and start the script with it already inserted.