import requests import json import random import pandas as pd import time import os import logging class Linkvertise(): def __init__(self): self.setup_logging() # Move logging setup to the beginning self.req_url = "https://publisher.linkvertise.com/graphql" self.load_config() self.headers = { "Accept": "application/json", "Authorization": f"Bearer {self.AUTH}", "Content-Type": "application/json", "Dnt": "1", "Origin": "https://link-mutation.linkvertise.com", "Referer": "https://link-mutation.linkvertise.com/", "Sec-Ch-Ua": "\"Google Chrome\";v=\"125\", \"Chromium\";v=\"125\", \"Not.A/Brand\";v=\"24\"", "Sec-Ch-Ua-Mobile": "?0", "Sec-Ch-Ua-Platform": "\"macOS\"", "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-site", "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36" } def setup_logging(self): self.logger = logging.getLogger('LinkvertiseLogger') self.logger.setLevel(logging.DEBUG) # Create handlers c_handler = logging.StreamHandler() f_handler = logging.FileHandler('linkvertise.log') c_handler.setLevel(logging.INFO) f_handler.setLevel(logging.DEBUG) # Create formatters and add them to handlers c_format = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') f_format = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s') c_handler.setFormatter(c_format) f_handler.setFormatter(f_format) # Add handlers to the logger self.logger.addHandler(c_handler) self.logger.addHandler(f_handler) def load_config(self): # Load the AUTH key from an environment variable or a secure source self.AUTH = os.getenv('LINKVERTISE_AUTH_KEY') if not self.AUTH: self.logger.error("Authorization key is missing. Please set the LINKVERTISE_AUTH_KEY environment variable.") raise ValueError("Authorization key is missing.") self.logger.debug('Config loaded with authentication key.') def validateLinkAttributes(self, target): payload = { "operationName": "validateLinkAttributes", "variables": { "input": { "target": target }, "link_id": None }, "query": "query validateLinkAttributes($input: ValidateLinkAttributeInput!, $link_id: ID) {\n validateLinkAttribute(input: $input, link_id: $link_id)\n}" } while True: try: response = requests.post(url=self.req_url, headers=self.headers, data=json.dumps(payload), allow_redirects=True) response.raise_for_status() # Ensure we got a successful response try: response_json = response.json() if 'data' not in response_json or 'validateLinkAttribute' not in response_json['data'] or response_json['data']['validateLinkAttribute'] != "OK": self.logger.warning(f"Validation failed for {target}. Response: {response_json}") time.sleep(2 + random.randint(1, 5)) # Wait before retrying continue self.logger.debug(f"Validation succeeded for {target}.") return response_json.get('data', {}) except json.JSONDecodeError as e: self.logger.error(f"JSON decode error for URL {target}: {e}") self.logger.error(f"Response content: {response.content}") time.sleep(2 + random.randint(1, 5)) # Wait before retrying continue except requests.exceptions.RequestException as e: self.logger.error(f"Request error for URL {target}: {e}") time.sleep(2 + random.randint(1, 5)) # Wait before retrying continue def createLink(self, target): while True: payload = { "operationName": "createLink", "variables": { "input": { "target_type": "URL", "target": target, "btn_prefix": "zu", "btn_text": f"{random.randint(100000, 999999999)}", "seo_active": False, "title": None, "description": None, "video_url": None, "images": None, "seo_faq_ids": ["1","2","3"], "require_addon": True, "require_web": True, "require_installer": True, "require_og_ads": True, "require_custom_ad_step": True } }, "query": "mutation createLink($input: LinkInput!) {\n createLink(input: $input) {\n id\n href\n user_id\n __typename\n }\n}" } try: response = requests.post(url=self.req_url, headers=self.headers, data=json.dumps(payload), allow_redirects=True) response.raise_for_status() response_json = response.json() if "data" in response_json and response_json["data"] is not None and "createLink" in response_json["data"]: create_link_data = response_json["data"]["createLink"] if create_link_data is not None and "href" in create_link_data: href = create_link_data["href"] self.logger.info(f"Link created successfully: {href}") return href else: self.logger.error("The 'href' key is missing in the response JSON.") else: self.logger.error("The expected structure is missing in the response JSON.") except requests.exceptions.HTTPError as e: if response.status_code == 403: self.logger.error(f"HTTP 403 error occurred: {e}") self.logger.info("Retrying... Waiting before next attempt.") else: self.logger.error(f"HTTP error occurred: {e}") except json.JSONDecodeError as e: self.logger.error(f"JSON decode error: {e}") self.logger.error(f"Response content: {response.content}") except Exception as e: self.logger.error(f"An unexpected error occurred: {e}") time.sleep(5 + random.randint(1, 5)) # Incremental backoff def createLinks(self, progress_callback): output_data = [] input_file_path = 'csvs/paster.csv' # Path to the input file output_file_path = 'csvs/linkvertise.csv' # Path to the output file # Ensure the output directory exists os.makedirs(os.path.dirname(output_file_path), exist_ok=True) # Load the links from the CSV file with no header links_data = pd.read_csv(input_file_path, header=None) links_data.columns = ['url'] # Adding a header name total_links = len(links_data) self.logger.info(f"Starting to process {total_links} links.") i = 0 while i < total_links: url = links_data['url'][i] self.logger.debug(f"Processing link {i + 1}/{total_links}: {url}") linkValidation = self.validateLinkAttributes(url) if linkValidation and 'validateLinkAttribute' in linkValidation and linkValidation['validateLinkAttribute'] == "OK": link_response = self.createLink(url) if link_response: output_data.append({ 'Original Link': url, 'Linkvertise Link': link_response }) progress_callback(f"Processed link {i+1}/{total_links}: {url} - Success, URL: {link_response}") self.logger.info(f"Link processed successfully: {url}") i += 1 # Move to the next link only if current link is processed successfully else: progress_callback(f"Processed link {i+1}/{total_links}: {url} - Error in creating link") self.logger.warning(f"Error in creating link for {url}.") else: progress_callback(f"Validation failed for {url}") self.logger.warning(f"Validation failed for {url}") time.sleep(5) # Add delay between requests to avoid rate limiting df = pd.DataFrame(output_data) df.to_csv(output_file_path, index=False) self.logger.info("Output CSV file created successfully.") progress_callback("Output CSV file created successfully.") # Usage: # progress_callback = print # Replace with any progress callback function you prefer # lv = Linkvertise() # lv.createLinks(progress_callback)