Merge branch 'master' of https://github.com/GramAddict/bot

Author: mastrolube

GramAddict/core/bot_flow.py

@@ -41,6 +41,7 @@
     open_instagram,
     pre_post_script,
     print_telegram_reports,
+    restart_atx_agent,
     save_crash,
     set_time_delta,
     show_ending_conditions,
@@ -119,6 +120,8 @@ def start_bot(**kwargs):
         if not inside_working_hours:
             wait_for_next_session(time_left, session_state, sessions, device)
         pre_post_script(path=configs.args.pre_script)
+        if configs.args.restart_atx_agent:
+            restart_atx_agent(device)
         get_device_info(device)
         session_state = SessionState(configs)
         session_state.set_limits_session()
@@ -163,7 +166,7 @@ def start_bot(**kwargs):
                             "If you press ENTER, you are aware of this and will not ask for support in case of a crash."
                         )
                         logger.warning(
-                            "If you want to avoid pressing ENTER next run, add allow-untested-ig-version: True in your config.yml file. (read the docs for more info)"
+                            "If you want to avoid pressing ENTER next run, add allow-untested-ig-version: true in your config.yml file. (read the docs for more info)"
                         )
                         input()
 
@@ -357,7 +360,8 @@ def start_bot(**kwargs):
             device.screen_off()
             logger.info("Screen turned off for sleeping time.")
 
-        kill_atx_agent(device)
+        if configs.args.kill_atx_agent:
+            kill_atx_agent(device)
         head_up_notifications(enabled=True)
         logger.info(
             "-------- FINISH: "

GramAddict/core/device_facade.py

@@ -185,8 +185,11 @@ def stop_screenrecord(self, crash=True):
         if self.deviceV2.screenrecord.stop(crash=crash):
             logger.warning("Screen recorder has been stopped successfully!")
 
-    def screenshot(self, path):
-        self.deviceV2.screenshot(path)
+    def screenshot(self, path=None):
+        if path is None:
+            return self.deviceV2.screenshot()
+        else:
+            self.deviceV2.screenshot(path)
 
     def dump_hierarchy(self, path):
         xml_dump = self.deviceV2.dump_hierarchy()
@@ -352,7 +355,7 @@ def ui_info(self):
             except uiautomator2.JSONRPCError as e:
                 raise DeviceFacade.JsonRpcError(e)
 
-        def content_desc(self):
+        def get_desc(self):
             try:
                 return self.viewV2.info["contentDescription"]
             except uiautomator2.JSONRPCError as e:

GramAddict/core/utils.py

@@ -429,11 +429,30 @@ def kill_atx_agent(device):
     _restore_keyboard(device)
     logger.info("Kill atx agent.")
     cmd: str = (
-        f"adb{'' if configs.device_id is None else ' -s ' + configs.device_id} shell pkill atx-agent"
+        f"adb{'' if configs.device_id is None else f' -s {configs.device_id}'} shell pkill atx-agent"
     )
     subprocess.run(cmd, stdout=PIPE, stderr=PIPE, shell=True, encoding="utf8")
 
 
+def restart_atx_agent(device):
+    kill_atx_agent(device)
+    logger.info("Restarting atx agent.")
+    cmd: str = (
+        f"adb{'' if configs.device_id is None else f' -s {configs.device_id}'} shell /data/local/tmp/atx-agent server -d"
+    )
+
+    try:
+        result = subprocess.run(
+            cmd, stdout=PIPE, stderr=PIPE, shell=True, encoding="utf8", check=True
+        )
+        if result.returncode != 0:
+            logger.error(f"Failed to restart atx-agent: {result.stderr}")
+        else:
+            logger.info("atx-agent restarted successfully.")
+    except subprocess.CalledProcessError as e:
+        logger.error(f"Error occurred while restarting atx-agent: {e}")
+
+
 def _restore_keyboard(device):
     logger.debug("Back to default keyboard!")
     device.deviceV2.set_fastinput_ime(False)
@@ -524,7 +543,8 @@ def trim_txt(source: str, target: str) -> None:
 
 def stop_bot(device, sessions, session_state, was_sleeping=False):
     close_instagram(device)
-    kill_atx_agent(device)
+    if args.kill_atx_agent:
+        kill_atx_agent(device)
     head_up_notifications(enabled=True)
     logger.info(
         f"-------- FINISH: {datetime.now().strftime('%H:%M:%S')} --------",
@@ -704,7 +724,8 @@ def set_time_delta(args):
 def wait_for_next_session(time_left, session_state, sessions, device):
     hours, remainder = divmod(time_left.seconds, 3600)
     minutes, seconds = divmod(remainder, 60)
-    kill_atx_agent(device)
+    if args.kill_atx_agent:
+        kill_atx_agent(device)
     logger.info(
         f'Next session will start at: {(datetime.now()+ time_left).strftime("%H:%M:%S (%Y/%m/%d)")}.',
         extra={"color": f"{Fore.GREEN}"},

GramAddict/core/views.py

@@ -1,6 +1,7 @@
 import datetime
 import logging
 import re
+import platform
 from enum import Enum, auto
 from random import choice, randint, uniform
 from time import sleep
@@ -128,7 +129,7 @@ def navigateToProfile(self):
     def _get_new_profile_position(self) -> Optional[DeviceFacade.View]:
         buttons = self.device.find(className=ResourceID.BUTTON)
         for button in buttons:
-            if button.content_desc() == "Profile":
+            if button.get_desc() == "Profile":
                 return button
         return None
 
@@ -871,7 +872,7 @@ def _get_post_owner_name(self):
 
     def _get_media_container(self):
         media = self.device.find(resourceIdMatches=ResourceID.CAROUSEL_AND_MEDIA_GROUP)
-        content_desc = media.content_desc() if media.exists() else None
+        content_desc = media.get_desc() if media.exists() else None
         return media, content_desc
 
     @staticmethod
@@ -988,23 +989,61 @@ def _check_if_ad_or_hashtag(
     ) -> Tuple[bool, bool, Optional[str]]:
         is_hashtag = False
         is_ad = False
-        real_username = None
         logger.debug("Checking if it's an AD or an hashtag..")
         ad_like_obj = post_owner_obj.sibling(
             resourceId=ResourceID.SECONDARY_LABEL,
         )
-        if post_owner_obj.get_text().startswith("#"):
-            is_hashtag = True
-            logger.debug("Looks like an hashtag, skip.")
+
+        owner_name = post_owner_obj.get_text() or post_owner_obj.get_desc()
+        if not owner_name:
+            logger.info("Can't find the owner name, need to use OCR.")
+            try:
+                import pytesseract as pt
+
+                owner_name = self.get_text_from_screen(pt, post_owner_obj)
+            except ImportError:
+                logger.error(
+                    "You need to install pytesseract (the wrapper: pip install pytesseract) in order to use OCR feature."
+                )
+            except pt.TesseractNotFoundError:
+                logger.error(
+                    "You need to install Tesseract (the engine: it depends on your system) in order to use OCR feature."
+                )
+            if owner_name.startswith("#"):
+                is_hashtag = True
+                logger.debug("Looks like an hashtag, skip.")
         if ad_like_obj.exists():
-            sponsored = "Sponsored"
-            if ad_like_obj.get_text() == sponsored:
+            sponsored_txt = "Sponsored"
+            ad_like_txt = ad_like_obj.get_text() or ad_like_obj.get_desc()
+            if ad_like_txt.casefold() == sponsored_txt.casefold():
                 logger.debug("Looks like an AD, skip.")
                 is_ad = True
             elif is_hashtag:
-                real_username = ad_like_obj.get_text().split("•")[0].strip()
+                owner_name = owner_name.split("•")[0].strip()
+
+        return is_ad, is_hashtag, owner_name
+
+    def get_text_from_screen(self, pt, obj) -> Optional[str]:
 
-        return is_ad, is_hashtag, real_username
+        if platform.system() == "Windows":
+            pt.pytesseract.tesseract_cmd = (
+                r"C:\Program Files\Tesseract-OCR\tesseract.exe"
+            )
+
+        screenshot = self.device.screenshot()
+        bounds = obj.ui_info().get("visibleBounds", None)
+        if bounds is None:
+            logger.info("Can't find the bounds of the object.")
+            return None
+        screenshot_cropped = screenshot.crop(
+            [
+                bounds.get("left"),
+                bounds.get("top"),
+                bounds.get("right"),
+                bounds.get("bottom"),
+            ]
+        )
+        return pt.image_to_string(screenshot_cropped).split(" ")[0].rstrip()
 
 
 class LanguageView:
@@ -1498,7 +1537,7 @@ def _new_ui_profile_button(self) -> bool:
         found = False
         buttons = self.device.find(className=ResourceID.BUTTON)
         for button in buttons:
-            if button.content_desc() == "Profile":
+            if button.get_desc() == "Profile":
                 button.click()
                 found = True
         return found

GramAddict/plugins/core_arguments.py

@@ -205,6 +205,16 @@ def __init__(self):
                 "help": "close all apps except IG, to avoid interference",
                 "action": "store_true",
             },
+            {
+                "arg": "--kill-atx-agent",
+                "help": "kill atx-agent when the script ends",
+                "action": "store_true",
+            },
+            {
+                "arg": "--restart-atx-agent",
+                "help": "restart atx-agent before the script starts",
+                "action": "store_true",
+            },
             {
                 "arg": "--interact",
                 "nargs": "+",