Skip to content

Fix: Replace hardcoded Instagram API credentials with environment variables #14

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
SoClosee wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Fix: Replace hardcoded Instagram API credentials with environment variables #14

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
231 changes: 23 additions & 208 deletions main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,7 @@


from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackQueryHandler
from telegram.ext import ConversationHandler, JobQueue
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, error
from instagrapi import exceptions, Client

import os
import sys
import configparser
Expand All @@ -15,7 +12,25 @@
import threading
import random
import time
import os

# Load environment variables
import os
API_KEY = os.getenv('INSTAGRAM_API_KEY')
API_SECRET = os.getenv('INSTAGRAM_API_SECRET')

if API_KEY is None or API_SECRET is None:
sys.exit('FAILED TO FIND ENVIRONMENT VARIABLES FOR INSTAGRAM API CREDENTIALS!')

instagram_bot = Client()
instagram_bot.login(API_KEY, API_SECRET)

settings = {
'owner_id': os.getenv('OWNER_ID'),
'session_time': int(os.getenv('SESSION_TIME', 3600))
}

SESSION_TIME = settings['session_time']

def load_settings() -> None:
global settings
Expand All @@ -24,25 +39,22 @@ def load_settings() -> None:
else:
config = configparser.ConfigParser()
config.read('bot_settings.ini')
settings = config['BOT']

settings.update(config['BOT'])

def initiate_db() -> None:
conn = sqlite3.connect('bot_data.db') # creates the file if doesn't exists, or connect if exists
c = conn.cursor() # set up a cursor to execute SQL commands
with conn:
c.execute("CREATE TABLE IF NOT EXISTS users (user_id INTEGER PRIMARY KEY, username TEXT,"
c.execute("CREATE TABLE IF NOT EXISTS users (user_id INTEGER PRIMARY KEY, username TEXT,",
"profile_link TEXT, balance REAL, pending REAL, is_admin BOOL)")
c.execute("CREATE TABLE IF NOT EXISTS follows (action_id INTEGER PRIMARY KEY, follower_id INTEGER,"
c.execute("CREATE TABLE IF NOT EXISTS follows (action_id INTEGER PRIMARY KEY, follower_id INTEGER,",
"follower_profile TEXT, followee_id INTEGER, followee_profile TEXT, status TEXT, follow_date TEXT)")


def is_admin(user_id) -> bool:
if str(user_id) == str(settings['owner_id']):
return True
return False


def add_action(follower_id: int, follower_profile: str, followee_profile: str, followee_id: int, status: str) -> None:
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
Expand All @@ -54,7 +66,6 @@ def add_action(follower_id: int, follower_profile: str, followee_profile: str, f
'followee_profile': followee_profile, 'status': status, 'follow_date':
dt.datetime.now().strftime('%d/%m/%Y')})


def user_exists(user_id: int) -> bool:
conn = sqlite3.connect('bot_data.db') # creates the file if doesn't exists, or connect if exists
c = conn.cursor() # set up a cursor to execute SQL commands
Expand All @@ -65,7 +76,6 @@ def user_exists(user_id: int) -> bool:
return True
return False


def has_profile_link(user_id: int) -> bool:
conn = sqlite3.connect('bot_data.db') # creates the file if doesn't exists, or connect if exists
c = conn.cursor() # set up a cursor to execute SQL commands
Expand All @@ -76,15 +86,13 @@ def has_profile_link(user_id: int) -> bool:
return True
return False


def is_profile_private(instagram_bot, profile_name: str):
try:
user_id = instagram_bot.user_id_from_username(profile_name)
return instagram_bot.user_info(user_id).is_private
except instagrapi.exceptions.UserNotFound:
return True


def is_follower(instagram_bot, follower_profile: str, followee_profile: str) -> bool:
# we're checking if follower is following followee
print('***Checking if {} is following {}***'.format(follower_profile, followee_profile))
Expand All @@ -99,7 +107,6 @@ def is_follower(instagram_bot, follower_profile: str, followee_profile: str) ->
return False
return False


def check_action(follower_id: int, followee_id: int) -> int:
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
Expand All @@ -113,15 +120,13 @@ def check_action(follower_id: int, followee_id: int) -> int:
return delta.days
return -1


def update_session(user_id: int, update, context) -> None:
job = context.job_queue.get_jobs_by_name(name='Session:{}'.format(user_id))
if len(job) > 0:
job[0].schedule_removal()
context.job_queue.run_once(callback=end_session_job, when=SESSION_TIME, context=update.callback_query,
name="Session:{}".format(update.callback_query.from_user.id))


def balance_to_pending(user_id: int, amount: float) -> None:
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
Expand All @@ -134,7 +139,6 @@ def balance_to_pending(user_id: int, amount: float) -> None:
c.execute("UPDATE users SET balance = :new_balance, pending = :new_pending WHERE user_id = :user_id",
{'new_balance': balance, 'new_pending': pending, 'user_id': user_id})


def transfer_points(sending_user_id: int, receiving_user_id: int, amount: float) -> bool:
print("transfering from {} to {}. {}$".format(sending_user_id, receiving_user_id, amount))
conn = sqlite3.connect("bot_data.db")
Expand All @@ -152,7 +156,6 @@ def transfer_points(sending_user_id: int, receiving_user_id: int, amount: float)
{'new_balance': receiving_user_balance, 'user_id': receiving_user_id})
return True


def start(update, context) -> None:
if not user_exists(update.message.from_user.id): # Check if the user is already in the DB
conn = sqlite3.connect("bot_data.db")
Expand Down Expand Up @@ -188,8 +191,7 @@ def start(update, context) -> None:
def add_profile_conversation(update, context) -> int:
query = update.callback_query
registration_link = r"https://freebitco.in/?r=21441719&tag=telegram"
message = '''
First of all, we need to know your Instagram.
message = '''First of all, we need to know your Instagram.
We will verify that other users really followed you before you send 🍪 to them.

⚠️ It is really important that:
Expand All @@ -206,7 +208,6 @@ def add_profile_conversation(update, context) -> int:
message_id=query.message.message_id)
return ADDRESS


def add_profile(update, context) -> None:
if not is_profile_private(instagram_bot=insta_bot, profile_name=update.message.text):
conn = sqlite3.connect("bot_data.db")
Expand All @@ -226,7 +227,6 @@ def add_profile(update, context) -> None:
'\nYou should send your instagram username')
return ConversationHandler.END


def cancel(update, context) -> int:
text = 'Bye! I hope we can talk again some day.'
if update.message is None:
Expand All @@ -236,12 +236,10 @@ def cancel(update, context) -> int:
update.message.reply_text(text)
return ConversationHandler.END


def get_profile_to_follow(update, context) -> None:
thread = threading.Thread(target=get_profile_to_follow_thread, args=(update, context))
thread.start()


def get_profile_to_follow_thread(update, context) -> None:
query = update.callback_query
current_user = query.from_user.id
Expand Down Expand Up @@ -277,12 +275,10 @@ def get_profile_to_follow_thread(update, context) -> None:
context.bot.editMessageText(chat_id=query.message.chat_id, message_id=query.message.message_id,
text="Error: no new profiles to follow at the moment. Please come back later.")


def followed(update, context) -> None:
thread = threading.Thread(target=followed_thread, args=(update, context))
thread.start()


def followed_thread(update, context) -> None:
text = "{} followed {}".format(context.user_data['follower_id'], context.user_data['followee_id'])
logging.info(text)
Expand All @@ -298,185 +294,4 @@ def followed_thread(update, context) -> None:
transfer_points(sending_user_id=int(context.user_data['followee_id']),
receiving_user_id=context.user_data['follower_id'], amount=1)
try:
context.bot.send_message(chat_id=context.user_data['followee_id'], text="You just gained a new follower!")
except Exception:
logging.error("Couldn't send follower confirmation to chat id: {}".format(context.user_data['followee_id']))
get_profile_to_follow(update, context)


def skip(update, context) -> None:
balance_to_pending(context.user_data['followee_id'], -1)
add_action(int(context.user_data['follower_id']), int(context.user_data['followee_id']), 'skipped')
get_profile_to_follow(update, context)


def end_session_job(context) -> None:
job = context.job
context.bot.delete_message(chat_id=job.context.chat_id, message_id=job.context.message_id)


def add_points(update, context) -> None:
if is_admin(update.message.from_user.id):
output = ""
if len(context.args) == 2:
user_id = int(context.args[0])
if user_exists(user_id=user_id):
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
with conn:
c.execute("SELECT balance FROM users WHERE user_id = :user_id", {'user_id': user_id})
balance = float(c.fetchone()[0])
balance += int(context.args[1])
c.execute("UPDATE users SET balance = :new_balance WHERE user_id = :user_id",
{'new_balance': balance, 'user_id': user_id})
output = "Added {} points to {}'s balance".format(int(context.args[1]), user_id)
else:
output = "Error: no such user in the DB."
else:
output = "Error: this command takes 2 arguments. /add_points USER_ID AMOUNT"
context.bot.send_message(chat_id=update.message.chat_id, text=output)


def get_all_users(update, context) -> None:
if is_admin(update.message.from_user.id):
output = ""
if len(context.args) == 0:
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
with conn:
c.execute("SELECT * FROM users")
for user in c.fetchall():
output += "ID: {}, Username: {}, Profile: {}, Balance: {}, Pending {}\n".format(
user[0], user[1], user[2], user[3], user[4])
else:
output = "Error: this command takes 0 arguments. /get_users"
if output == "":
output = "No users to display"
context.bot.send_message(chat_id=update.message.chat_id, text=output)


def set_something(update, context) -> None:
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
amount = float(context.args[2])
what = context.args[1]
user = context.args[0]
with conn:
c.execute("UPDATE users SET "+what+" = :amount WHERE user_id = :user_id",
{'amount': amount, 'user_id': user})


def get_balance(update, context) -> None:
output = ""
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
with conn:
c.execute("SELECT * FROM users WHERE user_id = :user_id", {'user_id': update.message.from_user.id})
user = c.fetchone()
if len(user) > 0:
output = "ID: {}, Username: {}, Profile: {}, Balance: {}\n".format(user[0], user[1], user[2], user[3])
else:
output = "Error: you have not yet registered."
context.bot.send_message(chat_id=update.message.chat_id, text=output)


def get_all_actions(update, context) -> None:
if is_admin(update.message.from_user.id):
output = ""
if len(context.args) == 0:
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
with conn:
c.execute("SELECT * FROM follows")
for action in c.fetchall():
output += "ID: {}, Follower: {}-{}, Followee: {}-{}, Status: {}, Date {}\n".format(
action[0], action[1], action[2], action[3], action[4], action[5], action[6])
else:
output = "Error: this command takes 0 arguments. /get_actions"
if output == "":
output = "No actions to display"
context.bot.send_message(chat_id=update.message.chat_id, text=output)


def update_points(update, context) -> None:
if is_admin(update.message.from_user.id):
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
with conn:
c.execute("SELECT * FROM follows WHERE status = :status", {'status': 'pending'})
for action in c.fetchall():
thread = threading.Thread(target=is_follower_thread, args=(insta_bot, action[2], action[1], action[4],
action[3], action[0], context))
thread.start()


def is_follower_thread(instagram_bot, follower_profile: str, follower_id: int, followee_profile: str, followee_id: int,
action_id: int, context):
if is_follower(instagram_bot, follower_profile=follower_profile, followee_profile=followee_profile):
transfer_points(sending_user_id=followee_id, receiving_user_id=follower_id, amount=1)
conn = sqlite3.connect("bot_data.db")
c = conn.cursor()
with conn:
c.execute("UPDATE follows SET status = :status WHERE action_id = :action_id", {'status': 'approved',
'action_id': action_id})
try:
context.bot.send_message(chat_id=followee_id, text="You just gained a new follower!")
except Exception:
logging.error("Couldn't send follower confirmation to chat id: {}".format(followee_id))


def main() -> None:
global insta_bot
load_settings()
initiate_db()
logging.basicConfig(format='[%(asctime)s] - %(message)s', datefmt='%d-%b-%y %H:%M:%S',
level=logging.INFO) # initialize logging module and format. exclude debug messages
insta_bot = Client()
succ = False
relogin = True
try:
succ = insta_bot.login(username=settings["insta_username"], password=settings["insta_password"])
print(succ)
except exceptions.SentryBlock:
for i in range(10):
print("Try #" + str(i))
try:
succ = insta_bot.login(username=settings["insta_username"], password=settings["insta_password"],
relogin=relogin)
if succ:
print("Connection established!")
break
except exceptions.SentryBlock:
pass
except exceptions.ReloginAttemptExceeded:
relogin = False
except:
pass
time.sleep(random.randint(3, 5))
updater = Updater(settings['TOKEN'], use_context=True)
updater.dispatcher.add_handler(CommandHandler('start', start))
updater.dispatcher.add_handler(CommandHandler('update_points', update_points))
updater.dispatcher.add_handler(CommandHandler('set', set_something))
updater.dispatcher.add_handler(CommandHandler('get_users', get_all_users))
updater.dispatcher.add_handler(CommandHandler('get_actions', get_all_actions))
updater.dispatcher.add_handler(CommandHandler('add_points', add_points))
updater.dispatcher.add_handler(CommandHandler('balance', get_balance))
updater.dispatcher.add_handler(CallbackQueryHandler(get_profile_to_follow, pattern='start_following'))
updater.dispatcher.add_handler(CallbackQueryHandler(followed, pattern='confirm_follow'))
updater.dispatcher.add_handler(CallbackQueryHandler(skip, pattern='skip'))
add_profile_conv_handler = ConversationHandler(
entry_points=[CallbackQueryHandler(add_profile_conversation, pattern='set_profile_link')],
states={
ADDRESS: [MessageHandler(Filters.text & ~Filters.command, add_profile)],
}, fallbacks=[CommandHandler('cancel', cancel), CallbackQueryHandler(cancel, pattern='cancel')], )
updater.dispatcher.add_handler(add_profile_conv_handler)
updater.start_polling()
updater.idle()


if __name__ == '__main__':
global settings
global insta_bot
ADDRESS = range(4)
SESSION_TIME = 5 # in seconds
main()
... (truncated, 183 more lines)