Python Bot Script to Increase Youtube Views and Subscribers

Python Bot Script to Increase Youtube Views and Subscribers

by

Do you know Python can help you get YouTube subscribers and even view/like your YouTube video automatically?

In this tutorial, I’ll guide you on how to create a Python Bot to auto-like YouTube videos and auto-subscribe YouTube channels. This will help you increase Views and Subscribers of a specific YouTube channel.

Demo Video

Live Demo

Requirements

  • Emails and Passwords Combo File
  • Action File (contains YouTube videos ID or channels ID)

Here’s an example of these files.

Emails and Passwords Combo File (accounts.txt)

Your emails combo file should follow this format.

[email protected]:YOUR_PASSWORD

[email protected]:123456789
[email protected]:123456789
[email protected]:123456789
[email protected]:123456789
[email protected]:123456789
[email protected]:123456789
[email protected]:123456789
[email protected]:123456789
[email protected]:123456789
[email protected]:123456789

Action File – Videos ID (videos_id.txt)

You need to create a file that contains the YouTube videos IDs list. The Python script will use them to open the specified YouTube videos one by one and like them.

Don’t use the full URL, instead use the ID only. For example, instead of using https://www.youtube.com/watch?v=XnEqfTjp66A URL, use only XnEqfTjp66A

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

Action File – Channels ID (channels_id.txt)

You need to create a file that contains the YouTube channels IDs list. The Python script will use them to open the specified YouTube channels one by one and subscribe to them.

Don’t use the full URL, instead use the ID only. For example, instead of using https://www.youtube.com/channel/UCs4aHmggTfFrpkPcWSaBN9g URL, use only UCs4aHmggTfFrpkPcWSaBN9g

UCs4aHmggTfFrpkPcWSaBN9g
UCzEnk4KWFlSj_PjXLz0-GMA
UCto7D1L-MiRoOziCXK9uT5Q

Python YouTube Subscriber Bot – Full Project Source Code

Project Folder Structure

  • lib
    • __init__.py
    • action.py
    • auth.py
    • chromedriver.exe
    • cli.py
    • execption.py
    • hotfix.py
    • multi.py
  • version
  • yt.py

Install Dependencies

pip install requests
pip install colorama
pip install selenium

lib/__init__.py

This file will be empty for now. We might use this file in the future. But, for now, simply create this file and leave it empty.

lib/action.py

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

lib/auth.py

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()

lib/chromedriver.exe

You can download this file from the official website.

lib/cli.py

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)

lib/execption.py

class LoginFailed(RuntimeError):
	pass

lib/hotfix.py

"""
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
"""

lib/multi.py

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()

version

In this file, we will add a version number for our script. For example:

12.8.3v

yt.py

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()

Run the Project

python yt.py

Important Notes

Make sure to have chrome installed before using this script. I didn’t use selenium for full automation just to compute botguard tokens in order to login this should be a temporary fix to the botguard problem.

Supported Platforms

Windows 7, 8, 10, 11, etc.

FAQ

Which Python version do I need?

Python 2.7.16

I’m not getting any results, why?

Check your emails

Program asking me to Enter accounts [Email:password] file path

Use the emails combo file you’ve created

Program asking me to Enter action file path

The action file is the file that has the channels/videos IDs in it.

TO-DO

  • Post a comment (Scheduled to the next public release)(Delayed)
  • Post a random comment based on the channel’s comments and users
  • Local proxy connection (For debugging) ✓
  • Local web server to manage the output instead of the console
  • An advanced debugging mode for advanced users ✓
  • Migrate to the module instead of single .py file ✓

Related posts:

  1. Convert Text File (TXT) to CSV Using Python 3 Pandas Script
  2. jsPDF-Autotable Auto Increase Width of Columns/Rows With Content
  3. Skip Youtube Ads Using OpenCV, PyAutoGUI & Python 3 AdBlocker
  4. PHP Script to Test/Detect Internet Speed
  5. How To Make an Automatic Omegle Bot Using JavaScript

1 thought on “Python Bot Script to Increase Youtube Views and Subscribers”

  1. I have the error
    Traceback (most recent call last)
    from lib.multi import Threader
    ModelnotfoundError: No Module named ‘lib.multi’
    I install multi

    Reply

Leave a Comment

This site uses Akismet to reduce spam. Learn how your comment data is processed.