diff --git a/.gitignore b/.gitignore index ed655afb..b0312209 100644 --- a/.gitignore +++ b/.gitignore @@ -11,3 +11,4 @@ filter.json whitelist.txt Pipfile.lock *.pdf +.venv diff --git a/GramAddict/__init__.py b/GramAddict/__init__.py index ede42b4b..b2597bc5 100644 --- a/GramAddict/__init__.py +++ b/GramAddict/__init__.py @@ -1,44 +1,37 @@ import argparse -from time import sleep -import colorama +import logging import sys +from datetime import datetime +from time import sleep -# from http.client import HTTPException -# from socket import timeout +from colorama import Fore, Style from GramAddict.core.device_facade import create_device +from GramAddict.core.log import configure_logger from GramAddict.core.navigation import switch_to_english from GramAddict.core.persistent_list import PersistentList from GramAddict.core.plugin_loader import PluginLoader from GramAddict.core.report import print_full_report -from GramAddict.core.session_state import ( - SessionState, - SessionStateEncoder, -) +from GramAddict.core.session_state import SessionState, SessionStateEncoder from GramAddict.core.storage import Storage -from datetime import datetime from GramAddict.core.utils import ( - COLOR_HEADER, - COLOR_WARNING, - COLOR_FAIL, - COLOR_ENDC, - get_version, check_adb_connection, + close_instagram, get_instagram_version, + get_value, + get_version, open_instagram, - close_instagram, - screen_sleep, save_crash, - get_value, - print_timeless, - print, + screen_sleep, ) from GramAddict.core.views import TabBarView - -# Script Initialization -print_timeless(COLOR_HEADER + "GramAddict " + get_version() + COLOR_ENDC) -colorama.init() +# Logging initialization +configure_logger() +logger = logging.getLogger(__name__) +logger.info( + "GramAddict " + get_version(), extra={"color": f"{Style.BRIGHT}{Fore.MAGENTA}"} +) # Global Variables device_id = None @@ -72,7 +65,7 @@ def load_plugins(): if arg.get("operation", False): actions[arg["arg"]] = plugin except Exception as e: - print_timeless( + logger.error( f"Error while importing arguments of plugin {plugin.__class__.__name__}. Error: Missing key from arguments dictionary - {e}" ) return actions @@ -86,11 +79,8 @@ def get_args(): args, unknown_args = parser.parse_known_args() if unknown_args: - print( - COLOR_FAIL - + "Unknown arguments: " - + ", ".join(str(arg) for arg in unknown_args) - + COLOR_ENDC + logger.error( + "Unknown arguments: " + ", ".join(str(arg) for arg in unknown_args) ) parser.print_help() return False @@ -110,8 +100,8 @@ def run(): for k in loaded: if dargs[k.replace("-", "_")[2:]] != None: if k == "--interact": - print_timeless( - f'{COLOR_WARNING}Warning: Using legacy argument "--interact". Please switch to new arguments as this will be deprecated in the near future.{COLOR_ENDC}' + logger.warn( + 'Using legacy argument "--interact". Please switch to new arguments as this will be deprecated in the near future.' ) if "#" in args.interact[0]: enabled.append("--hashtag-likers") @@ -124,18 +114,11 @@ def run(): enabled = list(dict.fromkeys(enabled)) if len(enabled) < 1: - print_timeless( - COLOR_FAIL - + "You have to specify one of the actions: " - + ", ".join(loaded) - + COLOR_ENDC - ) + logger.error("You have to specify one of the actions: " + ", ".join(loaded)) return if len(enabled) > 1: - print_timeless( - COLOR_FAIL - + "Running GramAddict with two or more actions is not supported yet." - + COLOR_ENDC + logger.error( + "Running GramAddict with two or more actions is not supported yet." ) return @@ -146,18 +129,14 @@ def run(): device_id = args.device if not check_adb_connection(is_device_id_provided=(device_id is not None)): return - - print("Instagram version: " + get_instagram_version()) + logger.info("Instagram version: " + get_instagram_version()) device = create_device(device_id) if device is None: return while True: - print_timeless( - COLOR_WARNING - + "\n-------- START: " - + str(session_state.startTime) - + " --------" - + COLOR_ENDC + logger.info( + "-------- START: " + str(session_state.startTime) + " --------", + extra={"color": f"{Style.BRIGHT}{Fore.YELLOW}"}, ) if args.screen_sleep: @@ -172,7 +151,7 @@ def run(): session_state.my_following_count, ) = profileView.getProfileInfo() except Exception as e: - print(f"Exception: {e}") + logger.error(f"Exception: {e}") save_crash(device) switch_to_english(device) # Try again on the correct language @@ -188,17 +167,14 @@ def run(): or not session_state.my_followers_count or not session_state.my_following_count ): - print(COLOR_FAIL + "Could not get profile info" + COLOR_ENDC) + logger.critical("Could not get profile info") exit(1) - report_string = "" - report_string += "Hello, @" + session_state.my_username + "! " - report_string += ( - "You have " + str(session_state.my_followers_count) + " followers" - ) - report_string += " and " + str(session_state.my_following_count) + " followings" - report_string += " so far." - print(report_string) + username = session_state.my_username + followers = session_state.my_followers_count + following = session_state.my_following_count + report_string = f"Hello, @{username}! You have {followers} followers and {following} followings so far." + logger.info(report_string, extra={"color": f"{Style.BRIGHT}"}) storage = Storage(session_state.my_username) @@ -210,17 +186,13 @@ def run(): if args.screen_sleep: screen_sleep(device_id, "off") # Turn off the device screen - print_timeless( - COLOR_WARNING - + "-------- FINISH: " - + str(session_state.finishTime) - + " --------" - + COLOR_ENDC + logger.info( + "-------- FINISH: " + str(session_state.finishTime) + " --------", + extra={"color": f"{Style.BRIGHT}{Fore.YELLOW}"}, ) if args.repeat: print_full_report(sessions) - print_timeless("") repeat = get_value(args.repeat, "Sleep for {} minutes", 180) try: sleep(60 * repeat) diff --git a/GramAddict/core/decorators.py b/GramAddict/core/decorators.py index e698bc7b..7b4ef4d1 100644 --- a/GramAddict/core/decorators.py +++ b/GramAddict/core/decorators.py @@ -1,25 +1,22 @@ +import logging +import sys import traceback +from datetime import datetime from http.client import HTTPException from socket import timeout -from datetime import datetime -import sys from GramAddict.core.device_facade import DeviceFacade from GramAddict.core.report import print_full_report from GramAddict.core.utils import ( - COLOR_WARNING, - COLOR_FAIL, - COLOR_ENDC, - open_instagram, close_instagram, + open_instagram, random_sleep, save_crash, - print_timeless, - print, ) - from GramAddict.core.views import LanguageNotEnglishException, TabBarView +logger = logging.getLogger(__name__) + def run_safely(device, device_id, sessions, session_state): def actual_decorator(func): @@ -29,28 +26,23 @@ def wrapper(*args, **kwargs): func(*args, **kwargs) except KeyboardInterrupt: close_instagram(device_id) - print_timeless( - COLOR_WARNING - + "-------- FINISH: " - + str(datetime.now().time()) - + " --------" - + COLOR_ENDC - ) + logger.warn(f"-------- FINISH: {datetime.now().time()} --------") print_full_report(sessions) sessions.persist(directory=session_state.my_username) sys.exit(0) except (DeviceFacade.JsonRpcError, IndexError, HTTPException, timeout): - print(COLOR_FAIL + traceback.format_exc() + COLOR_ENDC) + logger.error(traceback.format_exc()) save_crash(device) - print("No idea what it was. Let's try again.") + logger.info("No idea what it was. Let's try again.") # Hack for the case when IGTV was accidentally opened close_instagram(device_id) random_sleep() open_instagram(device_id) TabBarView(device).navigateToProfile() except LanguageNotEnglishException: - print_timeless("") - print("Language was changed. We'll have to start from the beginning.") + logger.info( + "Language was changed. We'll have to start from the beginning." + ) TabBarView(device).navigateToProfile() except Exception as e: save_crash(device) diff --git a/GramAddict/core/device_facade.py b/GramAddict/core/device_facade.py index d7ec8c30..c2c2564c 100644 --- a/GramAddict/core/device_facade.py +++ b/GramAddict/core/device_facade.py @@ -1,23 +1,22 @@ +import logging from enum import Enum, unique -import uiautomator2 -from GramAddict.core.utils import ( - COLOR_FAIL, - COLOR_ENDC, - print, -) from random import uniform +import uiautomator2 + +logger = logging.getLogger(__name__) + # How long we're waiting until UI element appears (loading content + animation) UI_TIMEOUT_LONG = 5 UI_TIMEOUT_SHORT = 1 def create_device(device_id): - print("Using uiautomator v2") + logger.debug("Using uiautomator v2") try: return DeviceFacade(device_id) except ImportError as e: - print(COLOR_FAIL + str(e) + COLOR_ENDC) + logger.error(str(e)) return None diff --git a/GramAddict/core/filter.py b/GramAddict/core/filter.py index 8cb4840a..9cc0f699 100644 --- a/GramAddict/core/filter.py +++ b/GramAddict/core/filter.py @@ -1,15 +1,12 @@ -import os import json -from GramAddict.core.utils import ( - COLOR_OKGREEN, - COLOR_FAIL, - COLOR_ENDC, - print_timeless, - print, -) +import logging +import os +from colorama import Fore from GramAddict.core.views import ProfileView +logger = logging.getLogger(__name__) + FILENAME_CONDITIONS = "filter.json" FIELD_SKIP_BUSINESS = "skip_business" FIELD_SKIP_NON_BUSINESS = "skip_non_business" @@ -47,21 +44,15 @@ def check_profile(self, device, username): if field_skip_business is not None or field_skip_non_business is not None: has_business_category = self._has_business_category(device) if field_skip_business and has_business_category is True: - print( - COLOR_OKGREEN - + "@" - + username - + " has business account, skip." - + COLOR_ENDC + logger.info( + f"@{username} has business account, skip.", + extra={"color": f"{Fore.GREEN}"}, ) return False if field_skip_non_business and has_business_category is False: - print( - COLOR_OKGREEN - + "@" - + username - + " has non business account, skip." - + COLOR_ENDC + logger.info( + f"@{username} has non business account, skip.", + extra={"color": f"{Fore.GREEN}"}, ) return False @@ -74,65 +65,40 @@ def check_profile(self, device, username): ): followers, followings = self._get_followers_and_followings(device) if field_min_followers is not None and followers < int(field_min_followers): - print( - COLOR_OKGREEN - + "@" - + username - + " has less than " - + str(field_min_followers) - + " followers, skip." - + COLOR_ENDC + logger.info( + f"@{username} has less than {field_min_followers} followers, skip.", + extra={"color": f"{Fore.GREEN}"}, ) return False if field_max_followers is not None and followers > int(field_max_followers): - print( - COLOR_OKGREEN - + "@" - + username - + " has more than " - + str(field_max_followers) - + " followers, skip." - + COLOR_ENDC + logger.info( + f"@{username} has has more than {field_max_followers} followers, skip.", + extra={"color": f"{Fore.GREEN}"}, ) return False if field_min_followings is not None and followings < int( field_min_followings ): - print( - COLOR_OKGREEN - + "@" - + username - + " has less than " - + str(field_min_followings) - + " followings, skip." - + COLOR_ENDC + logger.info( + f"@{username} has less than {field_min_followings} followers, skip.", + extra={"color": f"{Fore.GREEN}"}, ) return False if field_max_followings is not None and followings > int( field_max_followings ): - print( - COLOR_OKGREEN - + "@" - + username - + " has more than " - + str(field_max_followings) - + " followings, skip." - + COLOR_ENDC + logger.info( + f"@{username} has more than {field_max_followings} followings, skip.", + extra={"color": f"{Fore.GREEN}"}, ) return False if field_min_potency_ratio is not None and ( int(followings) == 0 or followers / followings < float(field_min_potency_ratio) ): - print( - COLOR_OKGREEN - + "@" - + username - + "'s potency ratio is less than " - + str(field_min_potency_ratio) - + ", skip." - + COLOR_ENDC + logger.info( + f"@{username}'s potency ratio is less than {field_min_potency_ratio}, skip.", + extra={"color": f"{Fore.GREEN}"}, ) return False return True @@ -155,23 +121,13 @@ def _get_followers_and_followings(device): try: followers = profileView.getFollowersCount() except Exception: - print_timeless( - COLOR_FAIL - + "Cannot find followers count view, default is " - + str(followers) - + COLOR_ENDC - ) + logger.error(f"Cannot find followers count view, default is {followers}") followings = 0 try: followings = profileView.getFollowingCount() except Exception: - print_timeless( - COLOR_FAIL - + "Cannot find followings count view, default is " - + str(followings) - + COLOR_ENDC - ) + logger.error(f"Cannot find followings count view, default is {followings}") return followers, followings diff --git a/GramAddict/core/interaction.py b/GramAddict/core/interaction.py index 788ad481..505f2338 100644 --- a/GramAddict/core/interaction.py +++ b/GramAddict/core/interaction.py @@ -1,22 +1,16 @@ -from random import shuffle, randint +import logging +from random import randint, shuffle from typing import Tuple +from colorama import Fore from GramAddict.core.device_facade import DeviceFacade from GramAddict.core.navigation import switch_to_english from GramAddict.core.report import print_short_report -from GramAddict.core.utils import ( - COLOR_OKGREEN, - COLOR_FAIL, - COLOR_ENDC, - COLOR_BOLD, - save_crash, - get_value, - random_sleep, - detect_block, - print, -) +from GramAddict.core.utils import detect_block, get_value, random_sleep, save_crash from GramAddict.core.views import LanguageNotEnglishException, ProfileView +logger = logging.getLogger(__name__) + TEXTVIEW_OR_BUTTON_REGEX = "android.widget.TextView|android.widget.Button" FOLLOW_REGEX = "Follow|Follow Back" UNFOLLOW_REGEX = "Following|Requested" @@ -36,7 +30,7 @@ def interact_with_user( :return: (whether interaction succeed, whether @username was followed during the interaction) """ if username == my_username: - print("It's you, skip.") + logger.info("It's you, skip.") return False, False random_sleep() @@ -46,7 +40,7 @@ def interact_with_user( likes_value = get_value(likes_count, "Likes count: {}", 2) if likes_value > 12: - print(COLOR_FAIL + "Max number of likes per user is 12" + COLOR_ENDC) + logger.error("Max number of likes per user is 12") likes_value = 12 profile_view = ProfileView(device) @@ -56,18 +50,23 @@ def interact_with_user( if is_private or is_empty: private_empty = "Private" if is_private else "Empty" - - print(COLOR_OKGREEN + f"{private_empty} account." + COLOR_ENDC) + logger.info( + f"{private_empty} account.", + extra={"color": f"{Fore.GREEN}"}, + ) if can_follow and profile_filter.can_follow_private_or_empty(): followed = _follow(device, username, follow_percentage) else: followed = False - print(COLOR_OKGREEN + "Skip user." + COLOR_ENDC) + logger.info( + "Skip user.", + extra={"color": f"{Fore.GREEN}"}, + ) return False, followed posts_tab_view = profile_view.navigateToPostsTab() if posts_tab_view.scrollDown(): # scroll down to view all maximum 12 posts - print("Scrolled down to see more posts.") + logger.info("Scrolled down to see more posts.") random_sleep() number_of_rows_to_use = min((likes_value * 2) // 3 + 1, 4) photos_indices = list(range(0, number_of_rows_to_use * 3)) @@ -78,26 +77,17 @@ def interact_with_user( photo_index = photos_indices[i] row = photo_index // 3 column = photo_index - row * 3 - - print( - "Open post #" - + str(i + 1) - + " (" - + str(row + 1) - + " row, " - + str(column + 1) - + " column)" - ) + logger.info(f"Open post #{i + 1} ({row + 1} row, {column + 1} column") opened_post_view = posts_tab_view.navigateToPost(row, column) random_sleep() like_succeed = False if opened_post_view: - print("Double click post") + logger.info("Double click post") opened_post_view.likePost() random_sleep() if not opened_post_view.isPostLiked(): - print("Double click failed. Try the like button.") + logger.debug("Double click failed. Try the like button.") opened_post_view.likePost(click_btn_like=True) random_sleep() @@ -106,14 +96,12 @@ def interact_with_user( detect_block(device) on_like() - print("Back to profile") + logger.info("Back to profile") device.back() if not opened_post_view or not like_succeed: reason = "open" if not opened_post_view else "like" - print( - f"{COLOR_BOLD}Could not {reason} photo. Posts count: {posts_count} {COLOR_ENDC}" - ) + logger.info(f"Could not {reason} photo. Posts count: {posts_count}") if can_follow and profile_filter.can_follow_private_or_empty(): followed = _follow(device, username, follow_percentage) @@ -121,7 +109,10 @@ def interact_with_user( followed = False if not followed: - print(COLOR_OKGREEN + "Skip user." + COLOR_ENDC) + logger.info( + "Skip user.", + extra={"color": f"{Fore.GREEN}"}, + ) return False, followed random_sleep() @@ -161,7 +152,7 @@ def _on_interaction( can_continue = True if session_state.totalLikes >= likes_limit: - print("Reached total likes limit, finish.") + logger.info("Reached total likes limit, finish.") on_likes_limit_reached() can_continue = False @@ -170,10 +161,8 @@ def _on_interaction( successful_interactions_count and successful_interactions_count >= interactions_limit ): - print( - "Made " - + str(successful_interactions_count) - + " successful interactions, finish." + logger.info( + f"Made {successful_interactions_count} successful interactions, finish." ) can_continue = False @@ -192,7 +181,7 @@ def _follow(device, username, follow_percentage): if follow_chance > follow_percentage: return False - print("Following...") + logger.info("Following...") coordinator_layout = device.find( resourceId="com.instagram.android:id/coordinator_root_layout" ) @@ -206,7 +195,7 @@ def _follow(device, username, follow_percentage): className="android.widget.LinearLayout", ) if not profile_header_actions_layout.exists(): - print(COLOR_FAIL + "Cannot find profile actions." + COLOR_ENDC) + logger.error("Cannot find profile actions.") return False follow_button = profile_header_actions_layout.child( @@ -221,13 +210,14 @@ def _follow(device, username, follow_percentage): textMatches=UNFOLLOW_REGEX, ) if unfollow_button.exists(): - print(COLOR_OKGREEN + "You already follow @" + username + "." + COLOR_ENDC) + logger.info( + f"You already follow @{username}.", + extra={"color": f"{Fore.GREEN}"}, + ) return False else: - print( - COLOR_FAIL - + "Cannot find neither Follow button, nor Unfollow button. Maybe not " - "English language is set?" + COLOR_ENDC + logger.error( + "Cannot find neither Follow button, nor Unfollow button. Maybe not English language is set?" ) save_crash(device) switch_to_english(device) @@ -235,6 +225,9 @@ def _follow(device, username, follow_percentage): follow_button.click() detect_block(device) - print(COLOR_OKGREEN + "Followed @" + username + COLOR_ENDC) + logger.info( + f"Followed @{username}", + extra={"color": f"{Fore.GREEN}"}, + ) random_sleep() return True diff --git a/GramAddict/core/log.py b/GramAddict/core/log.py new file mode 100644 index 00000000..38addcfb --- /dev/null +++ b/GramAddict/core/log.py @@ -0,0 +1,72 @@ +import logging +from io import StringIO +from logging import LogRecord + +from colorama import Fore, Style +from colorama import init as init_colorama + +COLORS = { + "DEBUG": Style.DIM, + "INFO": Fore.WHITE, + "WARNING": Fore.YELLOW, + "ERROR": Fore.RED, + "CRITICAL": Fore.MAGENTA, +} + + +class ColoredFormatter(logging.Formatter): + def __init__(self, *, fmt, datefmt=None): + logging.Formatter.__init__(self, fmt=fmt, datefmt=datefmt) + + def format(self, record): + msg = super().format(record) + levelname = record.levelname + if hasattr(record, "color"): + return f"{record.color}{msg}{Style.RESET_ALL}" + if levelname in COLORS: + return f"{COLORS[levelname]}{msg}{Style.RESET_ALL}" + return msg + + +def configure_logger(): + init_colorama() + logger = logging.getLogger() # root logger + logger.setLevel(logging.DEBUG) + + # Formatters + datefmt = r"[%m/%d %H:%M:%S]" + console_fmt = "%(asctime)s %(levelname)8s | %(message)s (%(filename)s:%(lineno)d)" + console_formatter = ColoredFormatter(fmt=console_fmt, datefmt=datefmt) + crash_report_fmt = ( + "%(asctime)s %(levelname)8s | %(message)s (%(filename)s:%(lineno)d)" + ) + crash_report_formatter = logging.Formatter(fmt=crash_report_fmt, datefmt=datefmt) + + # Filters + class FilterGramAddictOnly(logging.Filter): + def filter(self, record: LogRecord): + return record.name.startswith("GramAddict") + + # Console handler (limited colored log) + console_handler = logging.StreamHandler() + console_handler.setLevel(logging.DEBUG) + console_handler.setFormatter(console_formatter) + console_handler.addFilter(FilterGramAddictOnly()) + + # Crash report handler (full raw log) + global log_stream + log_stream = StringIO() + crash_report_handler = logging.StreamHandler(log_stream) + crash_report_handler.setLevel(logging.DEBUG) + crash_report_handler.setFormatter(crash_report_formatter) + crash_report_handler.addFilter(FilterGramAddictOnly()) + + logger.addHandler(console_handler) + logger.addHandler(crash_report_handler) + + +def get_logs(): + # log_stream is a StringIO() created when configure_logger() is called + global log_stream + + return log_stream.getvalue() diff --git a/GramAddict/core/navigation.py b/GramAddict/core/navigation.py index 59c7b0aa..22e5fee5 100644 --- a/GramAddict/core/navigation.py +++ b/GramAddict/core/navigation.py @@ -1,15 +1,15 @@ -from GramAddict.core.utils import ( - COLOR_OKGREEN, - COLOR_ENDC, - print, -) +import logging + +from colorama import Fore from GramAddict.core.views import TabBarView +logger = logging.getLogger(__name__) + def switch_to_english(device): - print(COLOR_OKGREEN + "Switching to English locale" + COLOR_ENDC) + logger.info("Switching to English locale", extra={"color": f"{Fore.GREEN}"}) profile_view = TabBarView(device).navigateToProfile() - print("Changing language in settings") + logger.info("Changing language in settings") options_view = profile_view.navigateToOptions() settingts_view = options_view.navigateToSettings() diff --git a/GramAddict/core/plugin_loader.py b/GramAddict/core/plugin_loader.py index 6f80df53..27173294 100644 --- a/GramAddict/core/plugin_loader.py +++ b/GramAddict/core/plugin_loader.py @@ -1,6 +1,9 @@ import inspect +import logging import pkgutil +logger = logging.getLogger(__name__) + class Plugin(object): def __init__(self): @@ -20,10 +23,8 @@ def __init__(self, plugin_package): def reload_plugins(self): self.plugins = [] self.seen_paths = [] - print() - print("Loading plugins . . .") + logger.info("Loading plugins . . .") self.walk_package(self.plugin_package) - print() def walk_package(self, package): imported_package = __import__(package, fromlist=["plugins"]) @@ -36,5 +37,5 @@ def walk_package(self, package): clsmembers = inspect.getmembers(plugin_module, inspect.isclass) for (_, c) in clsmembers: if issubclass(c, Plugin) & (c is not Plugin): - print(f" - {c.__name__}: {c.__doc__}") + logger.info(f" - {c.__name__}: {c.__doc__}") self.plugins.append(c()) diff --git a/GramAddict/core/report.py b/GramAddict/core/report.py index 93aa4bb3..7bae1e4f 100644 --- a/GramAddict/core/report.py +++ b/GramAddict/core/report.py @@ -1,80 +1,44 @@ -from datetime import timedelta, datetime -from GramAddict.core.utils import ( - COLOR_WARNING, - COLOR_ENDC, - print_timeless, - print, -) +import logging +from datetime import datetime, timedelta + +logger = logging.getLogger(__name__) def print_full_report(sessions): if len(sessions) > 1: for index, session in enumerate(sessions): finish_time = session.finishTime or datetime.now() - print_timeless("\n") - print_timeless(COLOR_WARNING + "SESSION #" + str(index + 1) + COLOR_ENDC) - print_timeless( - COLOR_WARNING + "Start time: " + str(session.startTime) + COLOR_ENDC - ) - print_timeless( - COLOR_WARNING + "Finish time: " + str(finish_time) + COLOR_ENDC - ) - print_timeless( - COLOR_WARNING - + "Duration: " - + str(finish_time - session.startTime) - + COLOR_ENDC - ) - print_timeless( - COLOR_WARNING - + "Total interactions: " - + _stringify_interactions(session.totalInteractions) - + COLOR_ENDC - ) - print_timeless( - COLOR_WARNING - + "Successful interactions: " - + _stringify_interactions(session.successfulInteractions) - + COLOR_ENDC + logger.warn("") + logger.warn(f"SESSION #{index + 1}") + logger.warn(f"Start time: {session.startTime}") + logger.warn(f"Finish time: {finish_time}") + logger.warn(f"Duration: {finish_time - session.startTime}") + logger.warn( + f"Total interactions: {_stringify_interactions(session.totalInteractions)}" ) - print_timeless( - COLOR_WARNING - + "Total followed: " - + _stringify_interactions(session.totalFollowed) - + COLOR_ENDC + logger.warn( + f"Successful interactions: {_stringify_interactions(session.successfulInteractions)}" ) - print_timeless( - COLOR_WARNING + "Total likes: " + str(session.totalLikes) + COLOR_ENDC + logger.warn( + f"Total followed: {_stringify_interactions(session.totalFollowed)}" ) - print_timeless( - COLOR_WARNING - + "Total unfollowed: " - + str(session.totalUnfollowed) - + COLOR_ENDC - ) - print_timeless( - COLOR_WARNING - + "Removed mass followers: " - + _stringify_removed_mass_followers(session.removedMassFollowers) - + COLOR_ENDC + logger.warn(f"Total likes: {session.totalLikes}") + logger.warn(f"Total unfollowed: {session.totalUnfollowed}") + logger.warn( + f"Removed mass followers: {_stringify_removed_mass_followers(session.removedMassFollowers)}" ) - print_timeless("\n") - print_timeless(COLOR_WARNING + "TOTAL" + COLOR_ENDC) + logger.warn("") + logger.warn("TOTAL") completed_sessions = [session for session in sessions if session.is_finished()] - print_timeless( - COLOR_WARNING - + "Completed sessions: " - + str(len(completed_sessions)) - + COLOR_ENDC - ) + logger.warn(f"Completed sessions: {len(completed_sessions)}") duration = timedelta(0) for session in sessions: finish_time = session.finishTime or datetime.now() duration += finish_time - session.startTime - print_timeless(COLOR_WARNING + "Total duration: " + str(duration) + COLOR_ENDC) + logger.warn(f"Total duration: {duration}") total_interactions = {} successful_interactions = {} @@ -102,38 +66,20 @@ def print_full_report(sessions): for username in session.removedMassFollowers: total_removed_mass_followers.append(username) - print_timeless( - COLOR_WARNING - + "Total interactions: " - + _stringify_interactions(total_interactions) - + COLOR_ENDC - ) - print_timeless( - COLOR_WARNING - + "Successful interactions: " - + _stringify_interactions(successful_interactions) - + COLOR_ENDC - ) - print_timeless( - COLOR_WARNING - + "Total followed : " - + _stringify_interactions(total_followed) - + COLOR_ENDC + logger.warn(f"Total interactions: {_stringify_interactions(total_interactions)}") + logger.warn( + f"Successful interactions: {_stringify_interactions(successful_interactions)}" ) + logger.warn(f"Total followed : {_stringify_interactions(total_followed)}") total_likes = sum(session.totalLikes for session in sessions) - print_timeless(COLOR_WARNING + "Total likes: " + str(total_likes) + COLOR_ENDC) + logger.warn(f"Total likes: {total_likes}") total_unfollowed = sum(session.totalUnfollowed for session in sessions) - print_timeless( - COLOR_WARNING + "Total unfollowed: " + str(total_unfollowed) + COLOR_ENDC - ) + logger.warn(f"Total unfollowed: {total_unfollowed} ") - print_timeless( - COLOR_WARNING - + "Removed mass followers: " - + _stringify_removed_mass_followers(total_removed_mass_followers) - + COLOR_ENDC + logger.warn( + f"Removed mass followers: {_stringify_removed_mass_followers(total_removed_mass_followers)}" ) @@ -141,19 +87,8 @@ def print_short_report(source, session_state): total_likes = session_state.totalLikes total_followed = sum(session_state.totalFollowed.values()) interactions = session_state.successfulInteractions.get(source, 0) - print( - COLOR_WARNING - + "Session progress: " - + str(total_likes) - + " likes, " - + str(total_followed) - + " followed, " - + str(interactions) - + " successful " - + ("interaction" if interactions == 1 else "interactions") - + " for " - + source - + COLOR_ENDC + logger.warn( + f"Session progress: {total_likes} likes, {total_followed} followed, {interactions} successful interaction(s) for {source}" ) diff --git a/GramAddict/core/scroll_end_detector.py b/GramAddict/core/scroll_end_detector.py index 9ab88c1d..530a7393 100644 --- a/GramAddict/core/scroll_end_detector.py +++ b/GramAddict/core/scroll_end_detector.py @@ -1,8 +1,8 @@ -from GramAddict.core.utils import ( - COLOR_OKBLUE, - COLOR_ENDC, - print, -) +import logging + +from colorama import Fore + +logger = logging.getLogger(__name__) class ScrollEndDetector: @@ -35,16 +35,14 @@ def is_the_end(self): repeats += 1 if is_the_end: - print( - COLOR_OKBLUE - + f"Same users iterated {repeats} times. End of the list, finish." - + COLOR_ENDC + logger.info( + f"Same users iterated {repeats} times. End of the list, finish.", + extra={"color": f"{Fore.BLUE}"}, ) elif repeats > 1: - print( - COLOR_OKBLUE - + f"Same users iterated {repeats} times. Continue." - + COLOR_ENDC + logger.info( + f"Same users iterated {repeats} times. Continue.", + extra={"color": f"{Fore.BLUE}"}, ) return is_the_end diff --git a/GramAddict/core/storage.py b/GramAddict/core/storage.py index 1017972d..44dc0245 100644 --- a/GramAddict/core/storage.py +++ b/GramAddict/core/storage.py @@ -1,13 +1,10 @@ +import json +import logging import os -from datetime import timedelta, datetime +from datetime import datetime, timedelta from enum import Enum, unique -import json -from GramAddict.core.utils import ( - COLOR_FAIL, - COLOR_ENDC, - print, -) +logger = logging.getLogger(__name__) FILENAME_INTERACTED_USERS = "interacted_users.json" USER_LAST_INTERACTION = "last_interaction" @@ -25,10 +22,8 @@ class Storage: def __init__(self, my_username): if my_username is None: - print( - COLOR_FAIL - + "No username, thus the script won't get access to interacted users and sessions data" - + COLOR_ENDC + logger.error( + "No username, thus the script won't get access to interacted users and sessions data" ) return diff --git a/GramAddict/core/utils.py b/GramAddict/core/utils.py index 6f48330a..26fee056 100644 --- a/GramAddict/core/utils.py +++ b/GramAddict/core/utils.py @@ -1,20 +1,17 @@ +import logging import os +import subprocess import re import shutil import sys from datetime import datetime -from random import uniform, randint +from random import randint, uniform from time import sleep -COLOR_HEADER = "\033[95m" -COLOR_OKBLUE = "\033[94m" -COLOR_OKGREEN = "\033[92m" -COLOR_WARNING = "\033[93m" -COLOR_DBG = "\033[90m" # dark gray -COLOR_FAIL = "\033[91m" -COLOR_ENDC = "\033[0m" -COLOR_BOLD = "\033[1m" -COLOR_UNDERLINE = "\033[4m" +from colorama import Fore, Style +from GramAddict.core.log import get_logs + +logger = logging.getLogger(__name__) def get_version(): @@ -39,14 +36,11 @@ def check_adb_connection(is_device_id_provided): is_ok = False message = "Use --device to specify a device." - print( - ("" if is_ok else COLOR_FAIL) - + "Connected devices via adb: " - + str(devices_count) - + ". " - + message - + COLOR_ENDC - ) + if is_ok: + logger.debug(f"Connected devices via adb: {devices_count}. {message}") + else: + logger.error(f"Connected devices via adb: {devices_count}. {message}") + return is_ok @@ -63,17 +57,21 @@ def get_instagram_version(): def open_instagram(device_id): - print("Open Instagram app") - os.popen( + logger.info("Open Instagram app") + cmd = ( "adb" + ("" if device_id is None else " -s " + device_id) + " shell am start -n com.instagram.android/com.instagram.mainactivity.MainActivity" - ).close() + ) + cmd_res = subprocess.run(cmd, capture_output=True, shell=True, encoding="utf8") + err = cmd_res.stderr.strip() + if err: + logger.debug(err) random_sleep() def close_instagram(device_id): - print("Close Instagram app") + logger.info("Close Instagram app") os.popen( "adb" + ("" if device_id is None else " -s " + device_id) @@ -83,7 +81,7 @@ def close_instagram(device_id): def random_sleep(): delay = uniform(1.0, 4.0) - print(f"{COLOR_DBG}{str(delay)[0:4]}s sleep{COLOR_ENDC}") + logger.debug(f"{str(delay)[0:4]}s sleep") sleep(delay) @@ -114,7 +112,7 @@ def check_screen_locked(device_id): def screen_unlock(device_id, MENU_BUTTON): is_locked = check_screen_locked(device_id) if is_locked: - print("Device is locked! I'll try to unlock it!") + logger.info("Device is locked! I'll try to unlock it!") os.popen( f"adb {''if device_id is None else ('-s '+ device_id)} shell input keyevent {MENU_BUTTON}" ) @@ -134,18 +132,18 @@ def screen_sleep(device_id, mode): os.popen( f"adb {''if device_id is None else ('-s '+ device_id)} shell input keyevent {POWER_BUTTON}" ) - print("Device screen turned ON!") + logger.info("Device screen turned ON!") sleep(2) screen_unlock(device_id, MENU_BUTTON) else: - print("Device screen already turned ON!") + logger.debug("Device screen already turned ON!") sleep(2) screen_unlock(device_id, MENU_BUTTON) else: os.popen( f"adb {''if device_id is None else ('-s '+ device_id)} shell input keyevent {POWER_BUTTON}" ) - print("Device screen turned OFF!") + logger.debug("Device screen turned OFF!") def save_crash(device): @@ -155,9 +153,7 @@ def save_crash(device): try: os.makedirs("crashes/" + directory_name + "/", exist_ok=False) except OSError: - print( - COLOR_FAIL + "Directory " + directory_name + " already exists." + COLOR_ENDC - ) + logger.error("Directory " + directory_name + " already exists.") return screenshot_format = ".png" @@ -166,7 +162,7 @@ def save_crash(device): "crashes/" + directory_name + "/screenshot" + screenshot_format ) except RuntimeError: - print(COLOR_FAIL + "Cannot save screenshot." + COLOR_ENDC) + logger.error("Cannot save screenshot.") view_hierarchy_format = ".xml" try: @@ -174,29 +170,30 @@ def save_crash(device): "crashes/" + directory_name + "/view_hierarchy" + view_hierarchy_format ) except RuntimeError: - print(COLOR_FAIL + "Cannot save view hierarchy." + COLOR_ENDC) + logger.error("Cannot save view hierarchy.") - with open("crashes/" + directory_name + "/logs.txt", "w") as outfile: - outfile.write(print_log) + with open( + "crashes/" + directory_name + "/logs.txt", "w", encoding="utf-8" + ) as outfile: + outfile.write(get_logs()) shutil.make_archive( "crashes/" + directory_name, "zip", "crashes/" + directory_name + "/" ) shutil.rmtree("crashes/" + directory_name + "/") - print( - COLOR_OKGREEN - + 'Crash saved as "crashes/' - + directory_name - + '.zip".' - + COLOR_ENDC + logger.info( + 'Crash saved as "crashes/' + directory_name + '.zip".', + extra={"color": Fore.GREEN}, + ) + logger.info( + "Please attach this file if you gonna report the crash at", + extra={"color": Fore.GREEN}, ) - print( - COLOR_OKGREEN - + "Please attach this file if you gonna report the crash at" - + COLOR_ENDC + logger.info( + "https://github.com/GramAddict/bot/issues\n", + extra={"color": Fore.GREEN}, ) - print(COLOR_OKGREEN + "https://github.com/GramAddict/bot/issues\n" + COLOR_ENDC) def detect_block(device): @@ -206,7 +203,7 @@ def detect_block(device): ) is_blocked = block_dialog.exists() if is_blocked: - print(COLOR_FAIL + "Probably block dialog is shown." + COLOR_ENDC) + logger.error("Probably block dialog is shown.") raise ActionBlockedError( "Seems that action is blocked. Consider reinstalling Instagram app and be more careful" " with limits!" @@ -231,11 +228,10 @@ def wrapper(*args, **kwargs): def get_value(count, name, default): def print_error(): - print( - COLOR_FAIL - + name.format(default) + logger.error( + name.format(default) + f'. Using default value instead of "{count}", because it must be ' - "either a number (e.g. 2) or a range (e.g. 2-4)." + COLOR_ENDC + "either a number (e.g. 2) or a range (e.g. 2-4)." ) parts = count.split("-") @@ -245,14 +241,14 @@ def print_error(): elif len(parts) == 1: try: value = int(count) - print(COLOR_BOLD + name.format(value) + COLOR_ENDC) + logger.info(name.format(value), extra={"color": Style.BRIGHT}) except ValueError: value = default print_error() elif len(parts) == 2: try: value = randint(int(parts[0]), int(parts[1])) - print(COLOR_BOLD + name.format(value) + COLOR_ENDC) + logger.info(name.format(value), extra={"color": Style.BRIGHT}) except ValueError: value = default print_error() @@ -263,8 +259,6 @@ def print_error(): print_log = "" -print_timeless = _print_with_time_decorator(print, False) -print = _print_with_time_decorator(print, True) class ActionBlockedError(Exception): diff --git a/GramAddict/core/views.py b/GramAddict/core/views.py index 4c5cbdcf..da679174 100644 --- a/GramAddict/core/views.py +++ b/GramAddict/core/views.py @@ -1,14 +1,10 @@ +import logging from enum import Enum, auto + from GramAddict.core.device_facade import DeviceFacade -from GramAddict.core.utils import ( - COLOR_DBG, - COLOR_FAIL, - COLOR_ENDC, - random_sleep, - save_crash, - print_timeless, - print, -) +from GramAddict.core.utils import random_sleep, save_crash + +logger = logging.getLogger(__name__) def case_insensitive_re(str_list): @@ -85,7 +81,7 @@ def navigateToProfile(self): def _navigateTo(self, tab: TabBarTabs): tab_name = tab.name - print(f"{COLOR_DBG}Navigate to {tab_name}{COLOR_ENDC}") + logger.debug(f"Navigate to {tab_name}") button = None tabBarView = self._getTabBar() if tab == TabBarTabs.HOME: @@ -98,7 +94,7 @@ def _navigateTo(self, tab: TabBarTabs): ) if not button.exists(): # Some accounts display the search btn only in Home -> action bar - print("Didn't find search in the tab bar...") + logger.debug("Didn't find search in the tab bar...") home_view = self.navigateToHome() home_view.navigateToSearch() return @@ -126,10 +122,8 @@ def _navigateTo(self, tab: TabBarTabs): return - print( - COLOR_FAIL - + f"Didn't find tab {tab_name} in the tab bar... Maybe English language is not set!?" - + COLOR_ENDC + logger.error( + f"Didn't find tab {tab_name} in the tab bar... Maybe English language is not set!?" ) raise LanguageNotEnglishException() @@ -156,7 +150,7 @@ def __init__(self, device: DeviceFacade): self.device = device def navigateToSearch(self): - print(f"{COLOR_DBG}Navigate to Search{COLOR_ENDC}") + logger.debug("Navigate to Search") search_btn = self.action_bar.child( descriptionMatches=case_insensitive_re(TabBarView.SEARCH_CONTENT_DESC) ) @@ -218,7 +212,7 @@ def _getTabTextView(self, tab: SearchTabs): return tab_text_view def navigateToUsername(self, username): - print("Navigate to profile @" + username) + logger.debug("Navigate to profile @" + username) search_edit_text = self._getSearchEditText() search_edit_text.click() @@ -226,9 +220,7 @@ def navigateToUsername(self, username): username_view = self._getUsernameRow(username) if not username_view.exists(): - print_timeless( - COLOR_FAIL + "Cannot find user @" + username + ", abort." + COLOR_ENDC - ) + logger.error("Cannot find user @" + username + ", abort.") return None username_view.click() @@ -236,7 +228,7 @@ def navigateToUsername(self, username): return ProfileView(self.device, is_own_profile=False) def navigateToHashtag(self, hashtag): - print("Navigate to hashtag #" + hashtag) + logger.debug(f"Navigate to hashtag #{hashtag}") search_edit_text = self._getSearchEditText() search_edit_text.click() @@ -245,7 +237,7 @@ def navigateToHashtag(self, hashtag): if not hashtag_tab.exists(): hashtag_tab = self._getTabTextView(SearchTabs.Tags) if not hashtag_tab.exists(): - print(COLOR_FAIL + "Cannot find tab: TAGS." + COLOR_ENDC) + logger.error("Cannot find tab: TAGS.") return None hashtag_tab.click() @@ -253,9 +245,7 @@ def navigateToHashtag(self, hashtag): hashtag_view = self._getHashtagRow(hashtag) if not hashtag_view.exists(): - print_timeless( - COLOR_FAIL + "Cannot find hashtag #" + hashtag + ", abort." + COLOR_ENDC - ) + logger.error(f"Cannot find hashtag #{hashtag} , abort.") return None hashtag_view.click() @@ -268,7 +258,7 @@ def __init__(self, device: DeviceFacade): self.device = device def setLanguage(self, language: str): - print(f"Set language to {language}") + logger.debug(f"Set language to {language}") search_edit_text = self.device.find( resourceId="com.instagram.android:id/search", className="android.widget.EditText", @@ -288,7 +278,7 @@ def __init__(self, device: DeviceFacade): self.device = device def navigateToLanguage(self): - print(f"{COLOR_DBG}Navigate to Language{COLOR_ENDC}") + logger.debug("Navigate to Language") button = self.device.find( textMatches=case_insensitive_re("Language"), resourceId="com.instagram.android:id/row_simple_text_textview", @@ -304,7 +294,7 @@ def __init__(self, device: DeviceFacade): self.device = device def navigateToAccount(self): - print(f"{COLOR_DBG}Navigate to Account{COLOR_ENDC}") + logger.debug("Navigate to Account") button = self.device.find( textMatches=case_insensitive_re("Account"), resourceId="com.instagram.android:id/row_simple_text_textview", @@ -319,7 +309,7 @@ def __init__(self, device: DeviceFacade): self.device = device def navigateToSettings(self): - print(f"{COLOR_DBG}Navigate to Settings{COLOR_ENDC}") + logger.debug("Navigate to Settings") button = self.device.find( textMatches=case_insensitive_re("Settings"), resourceId="com.instagram.android:id/menu_settings_row", @@ -342,7 +332,7 @@ def isPostLiked(self): if like_btn_view.exists(): return like_btn_view.get_selected() - print(COLOR_FAIL + "Cannot find button like" + COLOR_ENDC) + logger.error("Cannot find button like") return False @@ -369,17 +359,17 @@ def likePost(self, click_btn_like=False): if like_btn_top_bound >= image_bottom_bound: like_btn_view.click() else: - print("Like btn out of current view. Don't click, just ignore.") + logger.debug( + "Like btn out of current view. Don't click, just ignore." + ) else: - print(COLOR_FAIL + "Cannot find button like to click" + COLOR_ENDC) + logger.error("Cannot find button like to click") else: if post_media_view.exists(): post_media_view.double_click() else: - print( - COLOR_FAIL + "Could not find post area to double click" + COLOR_ENDC - ) + logger.error("Could not find post area to double click") class PostsGridView: @@ -421,7 +411,7 @@ def __init__(self, device: DeviceFacade, is_own_profile=False): self.is_own_profile = is_own_profile def navigateToOptions(self): - print(f"{COLOR_DBG}Navigate to Options{COLOR_ENDC}") + logger.debug("Navigate to Options") button = self.action_bar.child( descriptionMatches=case_insensitive_re("Options") ) @@ -446,7 +436,7 @@ def getUsername(self): title_view = self._getActionBarTitleBtn() if title_view.exists(): return title_view.get_text() - print(COLOR_FAIL + "Cannot get username" + COLOR_ENDC) + logger.error("Cannot get username") return "" def _parseCounter(self, text): @@ -462,13 +452,7 @@ def _parseCounter(self, text): try: count = int(float(text) * multiplier) except ValueError: - print_timeless( - COLOR_FAIL - + 'Cannot parse "' - + text - + '". Probably wrong language.' - + COLOR_ENDC - ) + logger.error(f"Cannot parse {text}. Probably wrong language ?!") raise LanguageNotEnglishException() return count @@ -489,9 +473,9 @@ def getFollowersCount(self): if followers_text: followers = self._parseCounter(followers_text) else: - print(COLOR_FAIL + "Cannot get your followers count text" + COLOR_ENDC) + logger.error("Cannot get your followers count text") else: - print(COLOR_FAIL + "Cannot find your followers count view" + COLOR_ENDC) + logger.error("Cannot find your followers count view") return followers @@ -512,9 +496,9 @@ def getFollowingCount(self): if following_text: following = self._parseCounter(following_text) else: - print(COLOR_FAIL + "Cannot get following count text" + COLOR_ENDC) + logger.error("Cannot get following count text") else: - print(COLOR_FAIL + "Cannot find following count view" + COLOR_ENDC) + logger.error("Cannot find following count view") return following @@ -528,7 +512,7 @@ def getPostsCount(self): if post_count_view.exists(): return self._parseCounter(post_count_view.get_text()) else: - print(COLOR_FAIL + "Cannot get posts count text" + COLOR_ENDC) + logger.error("Cannot get posts count text") return 0 def getProfileInfo(self): @@ -551,7 +535,7 @@ def isPrivateAccount(self): return private_profile_view.exists() def navigateToFollowers(self): - print(f"{COLOR_DBG}Navigate to Followers{COLOR_ENDC}") + logger.debug("Navigate to Followers") FOLLOWERS_BUTTON_ID_REGEX = case_insensitive_re( [ "com.instagram.android:id/row_profile_header_followers_container", @@ -609,9 +593,7 @@ def _navigateToTab(self, tab: ProfileTabs): className=TAB_CLASS_NAME, ) if not button.exists(): - print( - COLOR_FAIL + f"Cannot navigate to to tab '{description}'" + COLOR_ENDC - ) + logger.error(f"Cannot navigate to to tab '{description}'") save_crash(self.device) else: button.click() diff --git a/GramAddict/plugins/action_unfollow_followers.py b/GramAddict/plugins/action_unfollow_followers.py index 67d211f7..718affdc 100644 --- a/GramAddict/plugins/action_unfollow_followers.py +++ b/GramAddict/plugins/action_unfollow_followers.py @@ -1,22 +1,18 @@ -from GramAddict.core.navigation import switch_to_english -from enum import unique, Enum +import logging +from enum import Enum, unique from random import seed + +from colorama import Fore from GramAddict.core.decorators import run_safely from GramAddict.core.device_facade import DeviceFacade +from GramAddict.core.navigation import switch_to_english from GramAddict.core.plugin_loader import Plugin from GramAddict.core.storage import FollowingStatus -from GramAddict.core.utils import ( - COLOR_OKGREEN, - COLOR_FAIL, - COLOR_ENDC, - random_sleep, - save_crash, - detect_block, - print, -) - +from GramAddict.core.utils import detect_block, random_sleep, save_crash from GramAddict.core.views import LanguageNotEnglishException +logger = logging.getLogger(__name__) + FOLLOWING_BUTTON_ID_REGEX = ( "com.instagram.android:id/row_profile_header_following_container" "|com.instagram.android:id/row_profile_header_container_following" @@ -95,7 +91,7 @@ def __init__(self): self.unfollow_type = UnfollowRestriction.ANY if count <= 0: - print( + logger.info( "You want to unfollow " + str(count) + ", you have " @@ -121,7 +117,7 @@ def job(): self.unfollow_type, self.session_state.my_username, ) - print("Unfollowed " + str(self.state.unfollowed_count) + ", finish.") + logger.info(f"Unfollowed {self.state.unfollowed_count}, finish.") self.state.is_job_completed = True while not self.state.is_job_completed and self.state.unfollowed_count < count: @@ -143,21 +139,19 @@ def on_unfollow(self): self.session_state.totalUnfollowed += 1 def open_my_followings(self, device): - print("Open my followings") + logger.info("Open my followings") followings_button = device.find(resourceIdMatches=FOLLOWING_BUTTON_ID_REGEX) followings_button.click() def sort_followings_by_date(self, device): - print("Sort followings by date: from oldest to newest.") + logger.info("Sort followings by date: from oldest to newest.") sort_button = device.find( resourceId="com.instagram.android:id/sorting_entry_row_icon", className="android.widget.ImageView", ) if not sort_button.exists(): - print( - COLOR_FAIL - + "Cannot find button to sort followings. Continue without sorting." - + COLOR_ENDC + logger.error( + "Cannot find button to sort followings. Continue without sorting." ) return sort_button.click() @@ -166,10 +160,8 @@ def sort_followings_by_date(self, device): resourceId="com.instagram.android:id/follow_list_sorting_options_recycler_view" ) if not sort_options_recycler_view.exists(): - print( - COLOR_FAIL - + "Cannot find options to sort followings. Continue without sorting." - + COLOR_ENDC + logger.error( + "Cannot find options to sort followings. Continue without sorting." ) return @@ -186,7 +178,7 @@ def iterate_over_followings( unfollowed_count = 0 while True: - print("Iterate over visible followings") + logger.info("Iterate over visible followings") random_sleep() screen_iterated_followings = 0 @@ -197,10 +189,9 @@ def iterate_over_followings( user_info_view = item.child(index=1) user_name_view = user_info_view.child(index=0).child() if not user_name_view.exists(quick=True): - print( - COLOR_OKGREEN - + "Next item not found: probably reached end of the screen." - + COLOR_ENDC + logger.info( + "Next item not found: probably reached end of the screen.", + extra={"color": f"{Fore.GREEN}"}, ) break @@ -208,7 +199,7 @@ def iterate_over_followings( screen_iterated_followings += 1 if storage.is_user_in_whitelist(username): - print(f"@{username} is in whitelist. Skip.") + logger.info(f"@{username} is in whitelist. Skip.") continue if ( @@ -218,28 +209,20 @@ def iterate_over_followings( ): following_status = storage.get_following_status(username) if not following_status == FollowingStatus.FOLLOWED: - print( - "Skip @" - + username - + ". Following status: " - + following_status.name - + "." + logger.info( + f"Skip @{username}. Following status: {following_status.name}." ) continue if unfollow_restriction == UnfollowRestriction.ANY: following_status = storage.get_following_status(username) if following_status == FollowingStatus.UNFOLLOWED: - print( - "Skip @" - + username - + ". Following status: " - + following_status.name - + "." + logger.info( + f"Skip @{username}. Following status: {following_status.name}." ) continue - print("Unfollow @" + username) + logger.info("Unfollow @" + username) unfollowed = self.do_unfollow( device, username, @@ -257,14 +240,15 @@ def iterate_over_followings( return if screen_iterated_followings > 0: - print(COLOR_OKGREEN + "Need to scroll now" + COLOR_ENDC) + logger.info("Need to scroll now", extra={"color": f"{Fore.GREEN}"}) list_view = device.find( resourceId="android:id/list", className="android.widget.ListView" ) list_view.scroll(DeviceFacade.Direction.BOTTOM) else: - print( - COLOR_OKGREEN + "No followings were iterated, finish." + COLOR_ENDC + logger.info( + "No followings were iterated, finish.", + extra={"color": f"{Fore.GREEN}"}, ) return @@ -278,15 +262,15 @@ def do_unfollow(self, device, username, my_username, check_if_is_follower): text=username, ) if not username_view.exists(): - print(COLOR_FAIL + "Cannot find @" + username + ", skip." + COLOR_ENDC) + logger.error("Cannot find @" + username + ", skip.") return False username_view.click() if check_if_is_follower and self.check_is_follower( device, username, my_username ): - print("Skip @" + username + ". This user is following you.") - print("Back to the followings list.") + logger.info(f"Skip @{username}. This user is following you.") + logger.info("Back to the followings list.") device.back() return False @@ -308,10 +292,8 @@ def do_unfollow(self, device, username, my_username, check_if_is_follower): break if not unfollow_button.exists(): - print( - COLOR_FAIL - + "Cannot find Following button. Maybe not English language is set?" - + COLOR_ENDC + logger.error( + "Cannot find Following button. Maybe not English language is set?" ) save_crash(device) switch_to_english(device) @@ -323,7 +305,7 @@ def do_unfollow(self, device, username, my_username, check_if_is_follower): className="android.widget.TextView", ) if not confirm_unfollow_button.exists(): - print(COLOR_FAIL + "Cannot confirm unfollow." + COLOR_ENDC) + logger.error("Cannot confirm unfollow.") save_crash(device) device.back() return False @@ -333,13 +315,13 @@ def do_unfollow(self, device, username, my_username, check_if_is_follower): self.close_confirm_dialog_if_shown(device) detect_block(device) - print("Back to the followings list.") + logger.info("Back to the followings list.") device.back() return True def check_is_follower(self, device, username, my_username): - print( - COLOR_OKGREEN + "Check if @" + username + " is following you." + COLOR_ENDC + logger.info( + f"Check if @{username} is following you.", extra={"color": f"{Fore.GREEN}"} ) following_container = device.find(resourceIdMatches=FOLLOWING_BUTTON_ID_REGEX) following_container.click() @@ -352,7 +334,7 @@ def check_is_follower(self, device, username, my_username): text=my_username, ) result = my_username_view.exists() - print("Back to the profile.") + logger.info("Back to the profile.") device.back() return result @@ -372,7 +354,9 @@ def close_confirm_dialog_if_shown(self, device): if not user_avatar_view.exists(): return - print(COLOR_OKGREEN + "Dialog shown, confirm unfollowing." + COLOR_ENDC) + logger.info( + "Dialog shown, confirm unfollowing.", extra={"color": f"{Fore.GREEN}"} + ) random_sleep() unfollow_button = dialog_root_view.child( resourceId="com.instagram.android:id/primary_button", diff --git a/GramAddict/plugins/interact_blogger_followers.py b/GramAddict/plugins/interact_blogger_followers.py index 99902c7d..ea6ec4e7 100644 --- a/GramAddict/plugins/interact_blogger_followers.py +++ b/GramAddict/plugins/interact_blogger_followers.py @@ -1,28 +1,24 @@ -from random import seed, shuffle +import logging from functools import partial +from random import seed, shuffle + +from colorama import Fore from GramAddict.core.decorators import run_safely from GramAddict.core.device_facade import DeviceFacade from GramAddict.core.filter import Filter from GramAddict.core.interaction import ( - interact_with_user, - is_follow_limit_reached_for_source, _on_interaction, _on_like, _on_likes_limit_reached, + interact_with_user, + is_follow_limit_reached_for_source, ) from GramAddict.core.plugin_loader import Plugin from GramAddict.core.scroll_end_detector import ScrollEndDetector from GramAddict.core.storage import FollowingStatus -from GramAddict.core.utils import ( - COLOR_OKGREEN, - COLOR_FAIL, - COLOR_ENDC, - COLOR_BOLD, - random_sleep, - get_value, - print_timeless, - print, -) +from GramAddict.core.utils import get_value, random_sleep + +logger = logging.getLogger(__name__) from GramAddict.core.views import TabBarView @@ -73,14 +69,8 @@ def __init__(self): for source in sources: self.state = State() is_myself = source[1:] == self.session_state.my_username - print_timeless("") - print( - COLOR_BOLD - + "Handle " - + source - + (is_myself and " (it's you)" or "") - + COLOR_ENDC - ) + its_you = is_myself and " (it's you)" or "" + logger.info(f"Handle {source} {its_you}") on_likes_limit_reached = partial(_on_likes_limit_reached, state=self.state) @@ -172,7 +162,7 @@ def handle_blogger( def open_user_followers(self, device, username): if username is None: - print("Open your followers") + logger.info("Open your followers") profile_view = TabBarView(device).navigateToProfile() profile_view.navigateToFollowers() else: @@ -182,13 +172,13 @@ def open_user_followers(self, device, username): if not profile_view: return False - print("Open @" + username + " followers") + logger.info(f"Open @{username} followers") profile_view.navigateToFollowers() return True def scroll_to_bottom(self, device): - print("Scroll to bottom") + logger.info("Scroll to bottom") def is_end_reached(): see_all_button = device.find( @@ -203,7 +193,7 @@ def is_end_reached(): while not is_end_reached(): list_view.swipe(DeviceFacade.Direction.BOTTOM) - print("Scroll back to the first follower") + logger.info("Scroll back to the first follower") def is_at_least_one_follower(): follower = device.find( @@ -239,7 +229,7 @@ def scrolled_to_top(): scroll_end_detector = ScrollEndDetector() while True: - print("Iterate over visible followers") + logger.info("Iterate over visible followers") random_sleep() screen_iterated_followers = [] screen_skipped_followers_count = 0 @@ -253,10 +243,9 @@ def scrolled_to_top(): user_info_view = item.child(index=1) user_name_view = user_info_view.child(index=0).child() if not user_name_view.exists(quick=True): - print( - COLOR_OKGREEN - + "Next item not found: probably reached end of the screen." - + COLOR_ENDC + logger.info( + "Next item not found: probably reached end of the screen.", + extra={"color": f"{Fore.GREEN}"}, ) break @@ -265,21 +254,19 @@ def scrolled_to_top(): scroll_end_detector.notify_username_iterated(username) if storage.is_user_in_blacklist(username): - print("@" + username + " is in blacklist. Skip.") + logger.info(f"@{username} is in blacklist. Skip.") elif not is_myself and storage.check_user_was_interacted(username): - print("@" + username + ": already interacted. Skip.") + logger.info(f"@{username}: already interacted. Skip.") screen_skipped_followers_count += 1 elif is_myself and storage.check_user_was_interacted_recently( username ): - print( - "@" - + username - + ": already interacted in the last week. Skip." + logger.info( + f"@{username}: already interacted in the last week. Skip." ) screen_skipped_followers_count += 1 else: - print("@" + username + ": interact") + logger.info(f"@{username}: interact") user_name_view.click() can_follow = ( @@ -300,18 +287,18 @@ def scrolled_to_top(): if not can_continue: return - print("Back to followers list") + logger.info("Back to followers list") device.back() random_sleep() except IndexError: - print( - COLOR_FAIL - + "Cannot get next item: probably reached end of the screen." - + COLOR_ENDC + logger.error( + "Cannot get next item: probably reached end of the screen." ) if is_myself and scrolled_to_top(): - print(COLOR_OKGREEN + "Scrolled to top, finish." + COLOR_ENDC) + logger.info( + "Scrolled to top, finish.", extra={"color": f"{Fore.GREEN}"} + ) return elif len(screen_iterated_followers) > 0: load_more_button = device.find( @@ -329,10 +316,8 @@ def scrolled_to_top(): resourceId="android:id/list", className="android.widget.ListView" ) if not list_view.exists(): - print( - COLOR_FAIL - + "Cannot find the list of followers. Trying to press back again." - + COLOR_ENDC + logger.error( + "Cannot find the list of followers. Trying to press back again." ) device.back() list_view = device.find( @@ -341,7 +326,7 @@ def scrolled_to_top(): ) if is_myself: - print(COLOR_OKGREEN + "Need to scroll now" + COLOR_ENDC) + logger.info("Need to scroll now", extra={"color": f"{Fore.GREEN}"}) list_view.scroll(DeviceFacade.Direction.TOP) else: pressed_retry = False @@ -350,23 +335,25 @@ def scrolled_to_top(): className="android.widget.ImageView" ) if retry_button.exists(): - print('Press "Load" button') + logger.info('Press "Load" button') retry_button.click() random_sleep() pressed_retry = True if need_swipe and not pressed_retry: - print( - COLOR_OKGREEN - + "All followers skipped, let's do a swipe" - + COLOR_ENDC + logger.info( + "All followers skipped, let's do a swipe", + extra={"color": f"{Fore.GREEN}"}, ) list_view.swipe(DeviceFacade.Direction.BOTTOM) else: - print(COLOR_OKGREEN + "Need to scroll now" + COLOR_ENDC) + logger.info( + "Need to scroll now", extra={"color": f"{Fore.GREEN}"} + ) list_view.scroll(DeviceFacade.Direction.BOTTOM) else: - print( - COLOR_OKGREEN + "No followers were iterated, finish." + COLOR_ENDC + logger.info( + "No followers were iterated, finish.", + extra={"color": f"{Fore.GREEN}"}, ) return diff --git a/GramAddict/plugins/interact_hashtag_likers.py b/GramAddict/plugins/interact_hashtag_likers.py index bf96492a..f6c35217 100644 --- a/GramAddict/plugins/interact_hashtag_likers.py +++ b/GramAddict/plugins/interact_hashtag_likers.py @@ -1,30 +1,26 @@ +import logging from functools import partial from random import seed, shuffle + +from colorama import Fore, Style from GramAddict.core.decorators import run_safely from GramAddict.core.device_facade import DeviceFacade from GramAddict.core.filter import Filter from GramAddict.core.interaction import ( - is_follow_limit_reached_for_source, - interact_with_user, _on_interaction, _on_like, _on_likes_limit_reached, + interact_with_user, + is_follow_limit_reached_for_source, ) from GramAddict.core.plugin_loader import Plugin from GramAddict.core.scroll_end_detector import ScrollEndDetector from GramAddict.core.storage import FollowingStatus -from GramAddict.core.utils import ( - COLOR_OKGREEN, - COLOR_FAIL, - COLOR_ENDC, - COLOR_BOLD, - random_sleep, - get_value, - print_timeless, - print, -) +from GramAddict.core.utils import get_value, random_sleep from GramAddict.core.views import TabBarView +logger = logging.getLogger(__name__) + # Script Initialization seed() @@ -65,8 +61,7 @@ def __init__(self): for source in sources: self.state = State() - print_timeless("") - print(COLOR_BOLD + "Handle " + source + COLOR_ENDC) + logger.info(f"Handle {source}", extra={"color": f"{Style.BRIGHT}"}) on_likes_limit_reached = partial(_on_likes_limit_reached, state=self.state) @@ -147,7 +142,7 @@ def handle_hashtag( if not search_view.navigateToHashtag(hashtag): return - print("Opening the first result") + logger.info("Opening the first result") first_result_view = device.find( resourceId="com.instagram.android:id/recycler_view", @@ -165,11 +160,13 @@ def handle_hashtag( while True: if not self.open_likers(device): - print(COLOR_OKGREEN + "No likes, let's scroll down." + COLOR_ENDC) + logger.info( + "No likes, let's scroll down.", extra={"color": f"{Fore.GREEN}"} + ) posts_list_view.scroll(DeviceFacade.Direction.BOTTOM) continue - print("List of likers is opened.") + logger.info("List of likers is opened.") posts_end_detector.notify_new_page() random_sleep() likes_list_view = device.find( @@ -177,7 +174,7 @@ def handle_hashtag( ) prev_screen_iterated_likers = [] while True: - print("Iterate over visible likers.") + logger.info("Iterate over visible likers.") screen_iterated_likers = [] try: @@ -190,10 +187,9 @@ def handle_hashtag( className="android.widget.TextView", ) if not username_view.exists(quick=True): - print( - COLOR_OKGREEN - + "Next item not found: probably reached end of the screen." - + COLOR_ENDC + logger.info( + "Next item not found: probably reached end of the screen.", + extra={"color": f"{Fore.GREEN}"}, ) break @@ -202,13 +198,13 @@ def handle_hashtag( posts_end_detector.notify_username_iterated(username) if storage.is_user_in_blacklist(username): - print("@" + username + " is in blacklist. Skip.") + logger.info(f"@{username} is in blacklist. Skip.") continue elif storage.check_user_was_interacted(username): - print("@" + username + ": already interacted. Skip.") + logger.info(f"@{username}: already interacted. Skip.") continue else: - print("@" + username + ": interact") + logger.info(f"@{username}: interact") username_view.click() can_follow = ( @@ -227,30 +223,28 @@ def handle_hashtag( if not can_continue: return - print("Back to likers list") + logger.info("Back to likers list") device.back() random_sleep() except IndexError: - print( - COLOR_FAIL - + "Cannot get next item: probably reached end of the screen." - + COLOR_ENDC + logger.info( + "Cannot get next item: probably reached end of the screen.", + extra={"color": f"{Fore.GREEN}"}, ) if screen_iterated_likers == prev_screen_iterated_likers: - print( - COLOR_OKGREEN - + "Iterated exactly the same likers twice, finish." - + COLOR_ENDC + logger.info( + "Iterated exactly the same likers twice, finish.", + extra={"color": f"{Fore.GREEN}"}, ) - print(f"Back to #{hashtag}") + logger.info(f"Back to #{hashtag}") device.back() break prev_screen_iterated_likers.clear() prev_screen_iterated_likers += screen_iterated_likers - print(COLOR_OKGREEN + "Need to scroll now" + COLOR_ENDC) + logger.info("Need to scroll now", extra={"color": f"{Fore.GREEN}"}) likes_list_view.scroll(DeviceFacade.Direction.BOTTOM) if posts_end_detector.is_the_end(): @@ -264,7 +258,7 @@ def open_likers(self, device): className="android.widget.TextView", ) if likes_view.exists(): - print("Opening post likers") + logger.info("Opening post likers") random_sleep() likes_view.click("right") return True diff --git a/GramAddict/res/demo.gif b/GramAddict/res/demo.gif deleted file mode 100644 index 7872249a..00000000 Binary files a/GramAddict/res/demo.gif and /dev/null differ diff --git a/GramAddict/res/discord.png b/GramAddict/res/discord.png deleted file mode 100644 index 88419623..00000000 Binary files a/GramAddict/res/discord.png and /dev/null differ diff --git a/GramAddict/res/telegram.png b/GramAddict/res/telegram.png deleted file mode 100644 index 9cda34c4..00000000 Binary files a/GramAddict/res/telegram.png and /dev/null differ diff --git a/GramAddict/version.txt b/GramAddict/version.txt index 7dea76ed..6d7de6e6 100644 --- a/GramAddict/version.txt +++ b/GramAddict/version.txt @@ -1 +1 @@ -1.0.1 +1.0.2 diff --git a/README.md b/README.md index d011fc8a..5419dca3 100644 --- a/README.md +++ b/README.md @@ -16,22 +16,25 @@ Liking and following automatically on your Android phone/tablet. No root require ### How to install 1. Clone project: `git clone https://github.com/GramAddict/bot.git gramaddict` 2. Go to GramAddict folder: `cd gramaddict` -3. Install required libraries: `pipenv install` -4. Download and unzip [Android platform tools](https://developer.android.com/studio/releases/platform-tools), move them to a directory where you won't delete them accidentally, e.g. +3. (Optionally) Use virtualenv or similar to make a virtual environment `virtualenv -p python3 .venv` and enter the virtual environment `source .venv/bin/activate` +4. Install required libraries: `pip3 install -r requirements.txt` +5. Download and unzip [Android platform tools](https://developer.android.com/studio/releases/platform-tools), move them to a directory where you won't delete them accidentally, e.g. ``` mkdir -p ~/Library/Android/sdk mv /platform-tools/ ~/Library/Android/sdk ``` -5. [Add platform-tools path to the PATH environment variable](https://github.com/GramAddict/bot/wiki/Adding-platform-tools-to-the-PATH-environment-variable). If you do it correctly, terminal / command prompt command `adb devices` will print `List of devices attached` -6. Activate pipenv shell: `pipenv shell` +6. [Add platform-tools path to the PATH environment variable](https://github.com/GramAddict/bot/wiki/Adding-platform-tools-to-the-PATH-environment-variable). If you do it correctly, terminal / command prompt command `adb devices` will print `List of devices attached` +7. Run the script `python3 run.py --blogger-followers username` ### How to install on Raspberry Pi OS 1. Update apt-get: `sudo apt-get update` -2. Install ADB, Fastboot and Pipenv: `sudo apt-get install -y android-tools-adb android-tools-fastboot pipenv` +2. Install ADB and Fastboot: `sudo apt-get install -y android-tools-adb android-tools-fastboot pipenv` 3. Clone project: `git clone https://github.com/GramAddict/bot.git gramaddict` 4. Go to GramAddict folder: `cd gramaddict` -5. Install required libraries: `pipenv install` -6. Activate pipenv shell: `pipenv shell` +5. (Optionally) Use virtualenv or similar to make a virtual environment `virtualenv -p python3 .venv` and enter the virtual environment `source .venv/bin/activate` +6. Install required libraries: `pip3 install -r requirements.txt` +7. Run the script `python3 run.py --blogger-followers username` + ### Get started 1. Connect Android device to your computer with a USB cable diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 00000000..bcf46f1a --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +colorama==0.3.7 +matplotlib==3.3.3 +uiautomator2==2.11.2 \ No newline at end of file