Source code for social.trafficker

# -*- coding: utf-8 -*-
social.trafficker module containing Trafficker classes and methods
   This *Trafficker* contains the required classes and methods for spreading content posts
      over multiple networks such as Facebook, LinkedIn and Twitter.

   .. warning:: Coding style is CamelCase for classes and lowercase_separated_by_underscores A.K.A snake_case for
      methods and variables.
# Basic python3 libraries
import sqlite3
from os import path, devnull
from time import sleep

# Selenium libraries for bots
from selenium import webdriver
from import By
from selenium.webdriver.firefox.firefox_profile import FirefoxProfile
from selenium.webdriver.firefox.options import Options
from import expected_conditions
from import WebDriverWait

# Inherit from Bot class
from import Bot
# Access JSON config file
from social.utils.configuration import CONFIG

[docs]class TraffickerException(Exception): """Specific exception raised by the Trafficker class""" pass
[docs]class Trafficker(Bot): """Class of the Trafficker social automaton to spread a post on social networks""" def __init__(self, social_network, login, password, origin, url=None, headless=True): """ **Constructor of the Trafficker class** :param social_network: Class being the social network specificity :param login: Login to use for this Trafficker :param password: Password to use for this Trafficker :param origin: Origin of the message to post :param url: URL to post on, kwarg default is None :raise TypeError: Raised if the first argument social_network is not an object type """ # Initialize Bot object Bot.__init__(self, login, password) # Check argument type if isinstance(origin, str) is True: self.origin = origin else: raise TypeError("Argument login, password and origin must be string type") # Check for argument structure if callable(social_network.login_submit) is True and callable( is True: self.social_network = social_network else: raise TypeError("First argument social_network must have the two methods login_submit() and post()") # Initialize Firefox driver profile = FirefoxProfile() options = Options() options.headless = headless profile.set_preference("dom.webnotifications.enabled", False) self.driver = webdriver.Firefox(profile, firefox_binary=CONFIG["Firefox"]["firefoxBinary"], executable_path=CONFIG["Firefox"]["geckoDriverBinary"], options=options, timeout=20, log_path=devnull) self.driver.implicitly_wait(15) self.driver.get(self.social_network.url) # Login method self.social_network.login_submit(self.driver, self.login, self.password) self.wait_home_icon() # Eventually change post page if url is not None and isinstance(url, str) is True: self.driver.get(url) # Log initialization self.log("Trafficker initialized on network {0} with account {1} on computer {2}".format(, self.login, # def __repr__(self): # """Representation of the Trafficker class instance""" # return "Trafficker thread on social network {0} with account {1}".format(, # self.login) # pass def __del__(self): """Try to close the driver or send message if ending while thread alive""" if self.is_alive() is True: raise TraffickerException('Thread still alive, wait for it to finish wish join() method') else: self.driver.close()
[docs] def run(self): """Start posting messages in the social network""" _ids = [] # Can be stopped with Ctrl-C (SIGINT) try: # Main loop while True: # First semaphore to retrieve messages to post with self.database_semaphore: connection = sqlite3.connect(path.join(self.database_directory, self.database)) cursor = connection.cursor() request = """SELECT _id, message FROM public WHERE {0} = 0 AND origin = ? ORDER BY ts DESC; """.format( cursor.execute(request, (self.origin,)) messages = cursor.fetchall() cursor.close() connection.close() for message in messages:, message[1]) self.log("Message id {0} posted on social network {1} with account {2} on computer {3}".format( message[0],, self.login, # Second semaphore to update the message status with self.database_semaphore: # Update table when message posted connection = sqlite3.connect(path.join(self.database_directory, self.database)) cursor = connection.cursor() request = """UPDATE public SET {0} = 1 WHERE _id = ? AND origin = ?; """.format( cursor.execute(request, (message[0], self.origin,)) connection.commit() # Close after update(s) cursor.close() connection.close() # Append list of already inserted message and remove duplicate _ids.append(message[0]) set(_ids) # Be smooth with the computer sleep(5) except KeyboardInterrupt: pass
[docs] def wait_home_icon(self): """Wait until home icon appears""" WebDriverWait(self.driver, 30).until( expected_conditions.presence_of_element_located((By.ID, self.social_network.home)) )