From 7276b1e26a2695c5b74c734ec7cd25a600201b3a Mon Sep 17 00:00:00 2001 From: CRIMINAL Date: Sun, 16 May 2021 23:17:14 +0100 Subject: [PATCH] Simple API Guest Auth Example Script with API usage: #start.py import apis.onlyfans.onlyfans as OnlyFans api = OnlyFans.start() authed = api.add_auth() authed.login(guest=True) model = authed.get_user("onlyfans") print(model) --- apis/api_helper.py | 11 +++++++---- apis/onlyfans/onlyfans.py | 21 +++++++++++++++------ apis/starsavn/starsavn.py | 4 ++-- helpers/main_helper.py | 4 ++-- 4 files changed, 26 insertions(+), 14 deletions(-) diff --git a/apis/api_helper.py b/apis/api_helper.py index 292eb4210..8b176537d 100644 --- a/apis/api_helper.py +++ b/apis/api_helper.py @@ -47,8 +47,8 @@ def multiprocessing(max_threads=None): class session_manager(): - def __init__(self, original_sessions=[], headers: dict = {}, session_rules=None, session_retry_rules=None,max_threads=-1) -> None: - self.sessions = self.copy_sessions(original_sessions) + def __init__(self, original_sessions=[], headers: dict = {}, session_rules=None, session_retry_rules=None, max_threads=-1) -> None: + self.sessions = self.add_sessions(original_sessions) self.pool = multiprocessing(max_threads) self.max_threads = max_threads self.kill = False @@ -59,8 +59,11 @@ def __init__(self, original_sessions=[], headers: dict = {}, session_rules=None, dynamic_rules = requests.get(dr_link).json() self.dynamic_rules = dynamic_rules - def copy_sessions(self, original_sessions): - sessions = [] + def add_sessions(self, original_sessions: list, overwrite_old_sessions=True): + if overwrite_old_sessions: + sessions = [] + else: + sessions = self.sessions for original_session in original_sessions: cloned_session = copy.deepcopy(original_session) ip = getattr(original_session, "ip", "") diff --git a/apis/onlyfans/onlyfans.py b/apis/onlyfans/onlyfans.py index 3f63a3db7..392084b36 100644 --- a/apis/onlyfans/onlyfans.py +++ b/apis/onlyfans/onlyfans.py @@ -1,6 +1,6 @@ import time import base64 -from typing import Optional, Union +from typing import List, Optional, Union from urllib.parse import urlparse from urllib import parse import hashlib @@ -474,7 +474,7 @@ def set_scraped(self, name, scraped: media_types): class start(): - def __init__(self, custom_request=callable, max_threads=-1) -> None: + def __init__(self, custom_request=callable, max_threads=-1, original_sessions: List[requests.Session] = []) -> None: self.auths: list[create_auth] = [] self.subscriptions: list[create_subscription] = [] self.custom_request = custom_request @@ -483,10 +483,10 @@ def __init__(self, custom_request=callable, max_threads=-1) -> None: self.links = links self.pool = api_helper.multiprocessing() self.session_manager = api_helper.session_manager( - session_rules=session_rules, session_retry_rules=session_retry_rules, max_threads=max_threads) + session_rules=session_rules, session_retry_rules=session_retry_rules, max_threads=max_threads, original_sessions=original_sessions) self.settings = {} - def set_auth_details(self, option={}, only_active=False): + def add_auth(self, option={}, only_active=False): if only_active and not option.get("active"): return auth = create_auth(session_manager2=self.session_manager, @@ -503,10 +503,11 @@ def close_pools(self): class create_auth(): def __init__(self, session_manager2: api_helper.session_manager, option={}, init=False, pool=None, ) -> None: self.id = option.get("id") - self.username = option.get("username") + self.username: str = option.get("username") if not self.username: self.username = f"u{self.id}" self.name = option.get("name") + self.email: str = option.get("email") self.lists = {} self.links = content_types() self.isPerformer: bool = option.get("isPerformer") @@ -516,7 +517,7 @@ def __init__(self, session_manager2: api_helper.session_manager, option={}, init self.chats = None self.archived_stories = {} self.mass_messages = [] - self.paid_content = {} + self.paid_content = [] session_manager2 = copy.copy(session_manager2) self.session_manager = session_manager2 self.pool = pool @@ -575,6 +576,8 @@ def login(self, full=False, max_attempts=10, guest=False): dynamic_rules = self.session_manager.dynamic_rules a = [dynamic_rules, auth_id, user_agent, x_bc, auth_items.sess, link] self.session_manager.headers = create_headers(*a) + if not self.session_manager.sessions: + self.session_manager.add_sessions([requests.Session()]) if guest: print("Guest Authentication") return self @@ -875,3 +878,9 @@ def get_paid_content(self, check: bool = False, refresh: bool = True, limit: int results.extend(results2) self.paid_content = results return results + + def buy_subscription(self, user: dict): + user_id = user.get("id") + amount = user.get("amount") + x = {"paymentType": "subscribe", "userId": user_id, "subscribeSource": "profile", + "amount": amount, "token": "", "unavailablePaymentGates": []} diff --git a/apis/starsavn/starsavn.py b/apis/starsavn/starsavn.py index 4f26b7cd0..68002963d 100644 --- a/apis/starsavn/starsavn.py +++ b/apis/starsavn/starsavn.py @@ -452,7 +452,7 @@ def set_scraped(self, name, scraped: media_types): class start(): def __init__(self, sessions=[], custom_request=callable) -> None: - sessions = api_helper.copy_sessions(sessions) + sessions = api_helper.add_sessions(sessions) self.sessions = sessions self.auth = create_auth(init=True) self.custom_request = custom_request @@ -480,7 +480,7 @@ def __init__(self, sessions=[], custom_request=callable) -> None: # print("Could not authenticate") # return result - def set_auth_details(self, option): + def add_auth(self, option): if not option["active"]: return self.auth.auth_details.username = option["username"] diff --git a/helpers/main_helper.py b/helpers/main_helper.py index b9e0a74ae..fac1d520f 100644 --- a/helpers/main_helper.py +++ b/helpers/main_helper.py @@ -603,9 +603,9 @@ def process_profiles(json_settings, original_sessions, site_name, api: Union[Onl if not json_auth.get("active", None): continue json_auth["username"] = user - auth = api.set_auth_details( + auth = api.add_auth( json_auth) - auth.session_manager.copy_sessions(original_sessions) + auth.session_manager.add_sessions(original_sessions) auth.profile_directory = user_profile datas["auth"] = auth.auth_details.__dict__ if datas: