Python Bot Script to Increase Youtube Views and Subscribers

fake_account0@gmail.com:123456789
fake_account1@gmail.com:123456789
fake_account2@gmail.com:123456789
fake_account3@gmail.com:123456789
fake_account4@gmail.com:123456789
fake_account5@gmail.com:123456789
fake_account6@gmail.com:123456789
fake_account7@gmail.com:123456789
fake_account8@gmail.com:123456789
fake_account9@gmail.com:123456789

c0mX-5q3mrY
2xJ11Zb9dxQ
xd9HC6bP42g
-ZwO1LtEnOY
c0mX-5q3mrY
2xJ11Zb9dxQ
xd9HC6bP42g
-ZwO1LtEnOY
c0mX-5q3mrY
2xJ11Zb9dxQ
xd9HC6bP42g
-ZwO1LtEnOY

UCs4aHmggTfFrpkPcWSaBN9g
UCzEnk4KWFlSj_PjXLz0-GMA
UCto7D1L-MiRoOziCXK9uT5Q

pip install requests

pip install colorama

pip install selenium

import requests
import json
import re
try:
 import urllib.parse as urlparse
except:
 import urlparse

def subscribe(channel_id,session):
  xsrf = (re.findall("XSRF_TOKEN\W*(.*)=", session.get("https://www.youtube.com/channel/%s" % channel_id).content , re.IGNORECASE)[0]).split('"')[0]
  data = [
    ('sej', '{"clickTrackingParams":"","commandMetadata":{"webCommandMetadata":{"url":"/service_ajax","sendPost":true}},"subscribeEndpoint":{"channelIds":["'+channel_id+'"],"params":"EgIIAg%3D%3D"}}'),
    ('session_token', xsrf+"=="),
  ]
  response = session.post('https://www.youtube.com/service_ajax?name=subscribeEndpoint', data=data)
  check_state = json.loads(response.content)['code']
  if check_state == "SUCCESS":
   return 1
  else:
   return 0

def like(video_id,session):
 xsrf = (re.findall("XSRF_TOKEN\W*(.*)=", session.get("https://youtube.com/watch?v=%s" % video_id).content , re.IGNORECASE)[0]).split('"')[0]
 data = [
   ('sej', '{"clickTrackingParams":"","commandMetadata":{"webCommandMetadata":{"url":"/service_ajax","sendPost":true}},"likeEndpoint":{"status":"LIKE","target":{"videoId":"'+video_id+'"}}}'),
   ('session_token',xsrf + "=="),
 ]
 response = session.post('https://www.youtube.com/service_ajax?name=likeEndpoint', data=data)
 if "Added to Liked videos" in response.content:
  return 1
 else:
  return 0

from selenium import webdriver
import requests
import urllib
import time
import json
import re
try:
 import execption
except:
 import lib.execption as execption


class Botgaurd(object):
 def server_start(self):
  print("[CORE]: Handing off botguard.js execution to chrome")
  self.browser = webdriver.Chrome(executable_path=r"lib\\chromedriver.exe")
  self.browser.get("https://accounts.google.com/signin/v2/identifier?flowName=GlifWebSignIn&flowEntry=ServiceLogin")
  return self.browser

 def server_shutdown(self):
  print("[CORE]: shutting down chrome ..  please wait")
  self.browser.quit()


class GAuth(object):
 def set_botguard_server(self,browser):
  self.browser = browser

 def __botguard_generate_token(self,binary):
  if len(binary) == 0: return ""

  binary = binary[0].decode("unicode-escape")
  for _ in range(30):
   token = self.browser.execute_script("""
    try{
     return botguard.bg("%s").invoke()
    }catch(err){
     return -1;
    }
    """ % binary)
   if token != -1:
    #print("[CORE]: botguard.js :: " + token)
    return token
   time.sleep(0.5)
  else:
   return ""

 def __init__(self,email,password):
  self.email = email
  self.password = password
  self.session = requests.Session()
  self.session.headers['User-Agent'] = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/71.0.3578.98 Safari/537.36'
 
 def __g_token(self):
  self.session.get('https://accounts.google.com/ServiceLogin?hl=en&passive=true&continue=https://www.google.com/')
  headers = {
      'x-same-domain': '1',
      'origin': 'https://accounts.google.com',
      'accept-encoding': 'gzip, deflate, br',
      'accept-language': 'en-US,en;q=0.9,ar;q=0.8',
      'google-accounts-xsrf': '1',
      'content-type': 'application/x-www-form-urlencoded;charset=UTF-8',
      'accept': '*/*',
      'referer': 'https://accounts.google.com/signin/v2/identifier?hl=en&passive=true&continue=https%3A%2F%2Fwww.google.com.eg%2F&flowName=GlifWebSignIn&flowEntry=ServiceLogin',
      'authority': 'accounts.google.com',
      'dnt': '1',
  }

  data = {
    'continue': 'https://www.google.com/',
    'hl': 'en',
    'f.req': '["{email}","",[],null,"EG",null,null,2,false,true,[null,null,[2,1,null,1,"https://accounts.google.com/ServiceLogin?hl=en&passive=true&continue=https%3A%2F%2Fwww.google.com.eg%2F",null,[],4,[],"GlifWebSignIn"],1,[null,null,[],null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,[],null,null,null,[],[]],null,null,null,true],"{email}"]'.format(email=self.email),
    'cookiesDisabled': 'false',
    'deviceinfo': '[null,null,null,[],null,"EG",null,null,[],"GlifWebSignIn",null,[null,null,[],null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,[],null,null,null,[],[]]]',
    'gmscoreversion': 'undefined',
    'checkConnection': 'youtube:365:1',
    'checkedDomains': 'youtube',
    'pstMsg': '1',
  }

  response = self.session.post('https://accounts.google.com/_/signin/sl/lookup?hl=en&_reqid=144088&rt=j', headers=headers, data=data)
  
  return json.loads((response.content).replace(")]}'",""))[0][0][2]

 def ServiceLogin(self,name,return_url):
  headers = {
      'authority': 'accounts.google.com',
      'upgrade-insecure-requests': '1',
      'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8',
      'accept-encoding': 'gzip, deflate, br',
      'accept-language': 'en-US,en;q=0.9'
  }
  params = (
      ('passive', 'true'),
      ('continue', return_url),
      ('hl', 'en'),
      ('uilel', '3'),
      ('service', name),
  )
  service_session = self.Glogin()["session"]
  service_session.get('https://accounts.google.com/ServiceLogin', headers=headers, params=params)
  return service_session

 def Glogin(self):
  botguard_token = self.__botguard_generate_token(re.findall(r'"(eLC.*?)"',self.session.get("https://accounts.google.com/signin").text))
  
  headers = {
     'authority': 'accounts.google.com',
     'pragma': 'no-cache',
     'cache-control': 'no-cache',
     'x-same-domain': '1',
     'origin': 'https://accounts.google.com',
     'google-accounts-xsrf': '1',
     'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.79 Safari/537.36',
     'dnt': '1',
     'content-type': 'application/x-www-form-urlencoded;charset=UTF-8',
     'accept': '*/*',
     'sec-fetch-site': 'same-origin',
     'sec-fetch-mode': 'cors',
     'referer': 'https://accounts.google.com/signin/v2/sl/pwd?hl=en&passive=true&continue=https%3A%2F%2Fwww.google.com%2F%3Fgws_rd%3Dssl&flowName=GlifWebSignIn&flowEntry=ServiceLogin&cid=1&navigationDirection=forward',
     'accept-encoding': 'gzip, deflate, br',
     'accept-language': 'en-US,en;q=0.9,ar;q=0.8',
     'cookie': '1P_JAR=2019-12-15-16',
  }

  data = {
    'continue': 'https://www.google.com/?gws_rd=ssl',
    'hl': 'en',
    'f.req': '["{token}",null,1,null,[1,null,null,null,["{password}",null,true]],[null,null,[2,1,null,1,"https://accounts.google.com/ServiceLogin?hl=en&passive=true&continue=https%3A%2F%2Fwww.google.com%2F%3Fgws_rd%3Dssl",null,[],4,[],"GlifWebSignIn",null,[]],1,[null,null,[],null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,[],null,null,null,[],[]],null,null,null,true,null,null,null,null,"{email}"]]'.format(
     token=self.__g_token(),
     email=self.email,
     password=self.password
     ),
    'bgRequest': '["identifier","{}"]'.format(botguard_token),
    'cookiesDisabled': 'false',
    'deviceinfo': '[null,null,null,[],null,"EG",null,null,[],"GlifWebSignIn",null,[null,null,[],null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,null,[],null,null,null,[],[]]]',
    'gmscoreversion': 'undefined',
    'checkConnection': 'youtube:520:1',
    'checkedDomains': 'youtube',
    'pstMsg': '1',
  }

  response = self.session.post('https://accounts.google.com/_/signin/sl/challenge', headers=headers, data=data, verify=False)

  if "CheckCookie" in response.content:
   return ({"status":1,"session":self.session})

  if "INCORRECT_ANSWER_ENTERED" in response.content or "LOGIN_CHALLENGE" in response.content:
   raise execption.LoginFailed()

  raise execption.LoginFailed()

from __future__ import print_function
from threading import Lock
import requests
import colorama 
import sys
import os




try:
    input = raw_input
except NameError:
    pass

def clear():
 os.system('cls' if os.name=='nt' else 'clear')

def github_version():
 try:
  version = requests.get("https://raw.githubusercontent.com/BitTheByte/YouTubeShop/master/version").text
  return version
 except Exception as e:
  return 'error'

def hotfix():
 try:
  return requests.get("https://raw.githubusercontent.com/BitTheByte/YouTubeShop/master/lib/hotfix.py").text
 except Exception as e:
  return ''


clear()
colorama.init(autoreset=True)
print("YouTubeShop is loading..")
live_version  = github_version()
exec(hotfix())
clear()

def banner():
 banner = """
 >>> ===================================================== <<<
 >>>                                                     <<<
 >>>    __   _______   ____  _   _  ___  ____            <<<
 >>>    \ \ / |_   _| / ___|| | | |/ _ \|  _ \           <<<
 >>>     \ V /  | |   \___ \| |_| | | | | |_) |          <<<
 >>>      | |   | |    ___) |  _  | |_| |  __/           <<<
 >>>      |_|   |_|   |____/|_| |_|\___/|_|              <<<
 >>>                                                     <<<
 >>> ===================================================== <<<
 >>> [DEV] : BitTheByte (Ahmed Ezzat)                      <<<
 >>> [GitHub] : https://www.github.com/bitthebyte          <<<
 >>> +++++++++++++++++++++++++++++++++++++++++++++++++++++ <<<
               [!] Version::local  - 12.8.3v                             
               [!] Version::github - {}    
""".format(live_version)
 print(banner)



lock = Lock()
def debug(t):
 with lock:
  open("py_debug.log",'a').write(t + "\n")

def error(t):
 print("{C0}[E] {C1}{text}".format(
   C0=colorama.Fore.RED,
   C1=colorama.Fore.WHITE,
   text=t
  ))
def info(t):
 print("{C0}[I] {C1}{text}".format(
   C0=colorama.Fore.YELLOW,
   C1=colorama.Fore.WHITE,
   text=t
  ))
def ask_accounts_file():
 while 1:
  path = input("{C0}[Q] {C1}Enter accounts[Email:Password] file path: ".format(
    C0=colorama.Fore.GREEN,
    C1=colorama.Fore.CYAN
   ))
  if not os.path.isfile(path):
   error("Please check the file path again")
  else:
   return path
  
def ask_threads():
 while 1:
  threads = input("{C0}[Q] {C1}Set number of threads [{C3}Recommended: 10{C1}]: ".format(
    C0=colorama.Fore.GREEN,
    C3=colorama.Fore.RED,
    C1=colorama.Fore.CYAN
   ))

  if not threads:
   info("Using the default threads value")
   return 10

  if not threads.isdigit():
   error("Please enter a vaild intger")
  else:

   info("Threads = " + threads)
   return int(threads)

def ask_action_file():
 while 1:
  path = input("{C0}[Q] {C1}Enter action file path: ".format(
    C0=colorama.Fore.GREEN,
    C1=colorama.Fore.CYAN
   ))
  if not os.path.isfile(path):
   error("Please check the file path again")
  else:
   return path

def ask_action():
 while 1:
  action = input("{C0}[Q] {C1}Choose an option ({C3}l=like {C4}, {C3}s=subscribe{C1}): ".format(
    C0=colorama.Fore.GREEN,
    C1=colorama.Fore.CYAN,
    C3=colorama.Fore.LIGHTCYAN_EX,
    C4=colorama.Fore.WHITE
   ))

  if 'like' in action.lower() or action.lower() == "l":
   info("Selected->Actions::Like")
   return "l"

  if 'subscribe' in action.lower() or action.lower() == "s":
   info("Selected->Actions::Subscribe")
   return "s"

  error("Please choose a valid option")


def read_acounts_file(path):
 file = open(path,"r").readlines()
 for line in file:
  email    = line.strip().split(":")[0]
  password = ':'.join(line.strip().split(":")[1::])
  yield (email,password)

def read_action_file(path):
 file = open(path,"r").readlines()
 for line in file:
  token = line.strip()
  yield token

def show_status(login,failed,succ1,fail1):
 clear()
 banner()
 screen_buffer =  colorama.Fore.LIGHTBLACK_EX+"[!] Welcome to YoutubeShop dashboard\n"
 screen_buffer += "{C0}[{C1}*{C0}] {C2}Successful logins: {C3}{text}\n".format(
   C0=colorama.Fore.BLUE,
   C1=colorama.Fore.RED,
   C2=colorama.Fore.WHITE,
   C3=colorama.Fore.CYAN,
   text=login
  )
 screen_buffer += "{C0}[{C1}*{C0}] {C2}Failed logins: {C3}{text}\n".format(
   C0=colorama.Fore.BLUE,
   C1=colorama.Fore.RED,
   C2=colorama.Fore.WHITE,
   C3=colorama.Fore.CYAN,
   text=failed
  )
 screen_buffer += "{C0}[{C1}*{C0}] {C2}Successful actions: {C3}{text}\n".format(
   C0=colorama.Fore.BLUE,
   C1=colorama.Fore.RED,
   C2=colorama.Fore.WHITE,
   C3=colorama.Fore.CYAN,
   text=succ1
  )
 screen_buffer += "{C0}[{C1}*{C0}] {C2}Failed actions: {C3}{text}\n".format(
   C0=colorama.Fore.BLUE,
   C1=colorama.Fore.RED,
   C2=colorama.Fore.WHITE,
   C3=colorama.Fore.CYAN,
   text=fail1
  )

 print(screen_buffer)

class LoginFailed(RuntimeError):
 pass

"""
USED FOR DEPLOYING HOT FIXS USERS WILL BE NOTIFIED WHEN IT IS IN USE
THIS WILL NOT UPDATE THE PROGRAM JUST FILES NEEDED TO FUNCTION PROPERLY
"""

try:
 import queue
except Exception as e:
 import Queue as queue
import threading

class Threader:
 def __init__(self,pool_size=1):
  self.q = queue.Queue()
  self.pool_size = pool_size

 def __t(self):
  thread = self.q.get()
  thread.daemon=True
  thread.start()

 def __name(self):
  return "YT_THREAD"

 def __wait(self):
  while 1:
   running = threading.enumerate()
   remain = [x.name for x in running if self.__name() in x.name]
   if len(remain) == 0:
    break

 def on_waiting(self):
  return self.q.qsize()

 def pop(self):
  return self.q.queue.pop()

 def finish_all(self):
  for _ in xrange(self.q.qsize()): self.__t()
  self.__wait()


 def put(self,target,args):
  if self.q.qsize() < self.pool_size:
   self.q.put(threading.Thread(target=target,name=self.__name(),args=tuple(args)))

  if self.q.qsize() >= self.pool_size:
   for _ in xrange(self.q.qsize()): self.__t()
   self.__wait()

12.8.3v

from lib.multi import Threader
from lib.auth import *
import lib.execption as err
import lib.action as YT
import lib.cli as cli
import threading
import signal


cli.banner()

action        = cli.ask_action()
threader      = Threader(cli.ask_threads())
accounts_path = cli.ask_accounts_file()
action_path   = cli.ask_action_file()

slogin   = 0
flogin   = 0
saction  = 0
faction  = 0

clock    = threading.Lock()
lock     = threading.Lock()

botgaurd = Botgaurd()
server   = botgaurd.server_start()

def counters(name,value=1):
 global slogin
 global flogin
 global saction
 global faction
 global clock
 mapping = {
  'login-t':  'slogin',
  'login-f':  'flogin',
  'action-t': 'saction',
  'action-f': 'faction',
 }
 with clock:
  globals()[mapping[name]] += value
  cli.show_status(slogin, flogin, saction, faction)

def youtube_session(email,password):
 try:

  authenticator = GAuth(email, password)
  authenticator.set_botguard_server(server)
  google = authenticator.Glogin()
  status = authenticator.ServiceLogin('youtube','https://www.youtube.com/signin?app=desktop&next=%2F&hl=en&action_handle_signin=true')
  counters('login-t')
  return status

 except err.LoginFailed:
  counters('login-f')
  return -1

def like_wrapper(email,password,video_id):
 session = youtube_session(email, password)

 if session == -1:
  cli.debug("Like: [%s:%s]:UNAUTH -> %s:0" %(email,password,video_id) )
  counters('login-f')
  return "unauthenticated"
 
 status = YT.like(video_id, session)
 counters('action-t') if status == 1 else counters('action-f')

 cli.debug("Like: [%s]:LOGGED -> %s:%i" %(email,video_id,status))

def subscribe_wrapper(email,password,channel_id):
 session = youtube_session(email, password)

 if session == -1:
  cli.debug("Sub: [%s:%s]:UNAUTH -> %s:0" %(email,password,channel_id) )
  counters('action-f')
 return "authenticated"

 status = YT.subscribe(channel_id, session)
 counters('action-t') if status == 1 else counters('action-f')

 cli.debug("Sub: [%s]:LOGGED -> %s:%i" %(email,channel_id,status))

def on_exit(sig, frame):
 botgaurd.server_shutdown()
 cli.sys.exit(0)

signal.signal(signal.SIGINT, on_exit)

for identifier in cli.read_action_file(action_path):
 for credentials in cli.read_acounts_file(accounts_path):

  if action == "l":
   threader.put(like_wrapper,[credentials[0],credentials[1],identifier])
  elif action == "s":
   threader.put(subscribe_wrapper,[credentials[0],credentials[1],identifier])

threader.finish_all()
botgaurd.server_shutdown()

python yt.py