Skip to content

feat: improve carousel/reel interactions and Android 12 compatibility #443

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
coco11211 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

feat: improve carousel/reel interactions and Android 12 compatibility #443

Show file tree
Hide file tree
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
15 changes: 13 additions & 2 deletions GramAddict/core/device_facade.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,11 +26,15 @@ def create_device(device_id, app_id):


def get_device_info(device):
sdk_version = int(device.get_info()["sdkInt"])
logger.debug(
f"Phone Name: {device.get_info()['productName']}, SDK Version: {device.get_info()['sdkInt']}"
f"Phone Name: {device.get_info()['productName']}, SDK Version: {sdk_version}"
)
if int(device.get_info()["sdkInt"]) < 19:
if sdk_version < 19:
logger.warning("Only Android 4.4+ (SDK 19+) devices are supported!")
elif sdk_version >= 31:
# Android 12 (SDK 31) and 12L (SDK 32) compatibility
logger.info(f"Android 12+ detected (SDK {sdk_version}). Using enhanced compatibility mode.")
logger.debug(
f"Screen dimension: {device.get_info()['displayWidth']}x{device.get_info()['displayHeight']}"
)
Expand Down Expand Up @@ -318,6 +322,13 @@ def get_info(self):
except uiautomator2.JSONRPCError as e:
raise DeviceFacade.JsonRpcError(e)

def is_android_12_plus(self):
"""Check if device is running Android 12 (SDK 31) or higher"""
try:
return int(self.get_info()["sdkInt"]) >= 31
except (KeyError, ValueError, TypeError):
return False

@staticmethod
def sleep_mode(mode):
mode = SleepTime.DEFAULT if mode is None else mode
Expand Down
46 changes: 40 additions & 6 deletions GramAddict/core/interaction.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -548,6 +548,9 @@ def _browse_carousel(device: DeviceFacade, obj_count: int) -> None:
media_obj_bounds = media_obj.get_bounds()
n = 1
while n < carousel_count:
# Add humanlike delay before checking media type
random_sleep(0.3, 0.8, modulable=False)

if media_obj.child(
resourceIdMatches=ResourceID.CAROUSEL_IMAGE_MEDIA_GROUP
).exists():
Expand All @@ -557,7 +560,9 @@ def _browse_carousel(device: DeviceFacade, obj_count: int) -> None:
0,
its_time=True,
)
sleep(watch_photo_time)
# Add variance to photo viewing time
actual_watch_time = watch_photo_time * uniform(0.85, 1.25)
sleep(actual_watch_time)
elif media_obj.child(
resourceIdMatches=ResourceID.CAROUSEL_VIDEO_MEDIA_GROUP
).exists():
Expand All @@ -567,22 +572,51 @@ def _browse_carousel(device: DeviceFacade, obj_count: int) -> None:
0,
its_time=True,
)
sleep(watch_video_time)
# Add variance to video viewing time
actual_watch_time = watch_video_time * uniform(0.90, 1.20)
sleep(actual_watch_time)

# More natural swipe coordinates with greater randomness
start_point_y = (
(media_obj_bounds["bottom"] + media_obj_bounds["top"])
/ 2
* uniform(0.85, 1.15)
* uniform(0.80, 1.20)
)
start_point_x = uniform(0.85, 1.10) * (
media_obj_bounds["right"] * 5 / 6
start_point_x = uniform(0.75, 0.95) * (
media_obj_bounds["right"] * uniform(0.80, 0.95)
)
delta_x = media_obj_bounds["right"] * uniform(0.5, 0.7)
# Vary swipe distance for more natural behavior
delta_x = media_obj_bounds["right"] * uniform(0.45, 0.75)

# Add small random pause before swiping (simulates human decision time)
random_sleep(0.2, 0.6, modulable=False)

UniversalActions(device)._swipe_points(
start_point_y=start_point_y,
start_point_x=start_point_x,
delta_x=delta_x,
direction=Direction.LEFT,
)

# Random chance to swipe back occasionally (natural curiosity)
if randint(1, 100) <= 15 and n > 1:
logger.debug("Looking back at previous carousel item (humanlike behavior)")
random_sleep(0.4, 0.9, modulable=False)
UniversalActions(device)._swipe_points(
start_point_y=start_point_y,
start_point_x=media_obj_bounds["left"] + (media_obj_bounds["right"] * 0.2),
delta_x=media_obj_bounds["right"] * uniform(0.45, 0.65),
direction=Direction.RIGHT,
)
random_sleep(0.5, 1.2, modulable=False)
# Swipe forward again
UniversalActions(device)._swipe_points(
start_point_y=start_point_y,
start_point_x=start_point_x,
delta_x=delta_x,
direction=Direction.LEFT,
)

n += 1


Expand Down
128 changes: 110 additions & 18 deletions GramAddict/core/views.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -996,33 +996,79 @@ def _check_if_ad_or_hashtag(

owner_name = post_owner_obj.get_text() or post_owner_obj.get_desc() or ""
if not owner_name:
logger.info("Can't find the owner name, need to use OCR.")
try:
import pytesseract as pt
logger.info("Can't find the owner name, attempting fallback methods.")
# Try multiple fallback methods
owner_name = self._get_owner_name_with_fallbacks(post_owner_obj)

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("#"):
if owner_name and owner_name.startswith("#"):
is_hashtag = True
logger.debug("Looks like an hashtag, skip.")
if ad_like_obj.exists():
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():
if ad_like_txt and ad_like_txt.casefold() == sponsored_txt.casefold():
logger.debug("Looks like an AD, skip.")
is_ad = True
elif is_hashtag:
elif is_hashtag and owner_name:
owner_name = owner_name.split("•")[0].strip()

return is_ad, is_hashtag, owner_name

def _get_owner_name_with_fallbacks(self, post_owner_obj) -> Optional[str]:
"""
Attempt to get owner name using multiple fallback methods
:param post_owner_obj: The post owner UI object
:return: Owner name or None
"""
owner_name = None

# Fallback 1: Try OCR with better error handling
try:
import pytesseract as pt
logger.debug("Attempting OCR to extract owner name...")
owner_name = self.get_text_from_screen(pt, post_owner_obj)
if owner_name and len(owner_name.strip()) > 0:
logger.info(f"OCR successfully extracted owner name: {owner_name}")
return owner_name
except ImportError:
logger.warning(
"pytesseract not installed. Install with: pip install pytesseract"
)
except Exception as e:
# Catch TesseractNotFoundError and other exceptions
logger.warning(
f"OCR failed: {str(e)}. Tesseract may not be installed on your system."
)

# Fallback 2: Try to find owner name in parent/sibling elements
try:
logger.debug("Attempting to find owner name in nearby UI elements...")
parent = post_owner_obj
# Try to get text from nearby elements
for attempt in range(3): # Try up to 3 levels up
if parent:
siblings = parent.sibling()
if siblings.exists():
sibling_text = siblings.get_text()
if sibling_text and len(sibling_text.strip()) > 0:
logger.info(f"Found owner name in sibling element: {sibling_text}")
return sibling_text
except Exception as e:
logger.debug(f"Sibling search failed: {e}")

# Fallback 3: Try content description as last resort
try:
content_desc = post_owner_obj.get_desc()
if content_desc and len(content_desc.strip()) > 0:
# Sometimes the content description contains the username
logger.info(f"Using content description as owner name: {content_desc}")
return content_desc
except Exception as e:
logger.debug(f"Content description extraction failed: {e}")

logger.warning("All fallback methods failed to extract owner name.")
return None

def get_text_from_screen(self, pt, obj) -> Optional[str]:

if platform.system() == "Windows":
Expand Down Expand Up @@ -1312,9 +1358,21 @@ def watch_media(self, media_type: MediaType) -> None:
watching_time,
time_left - 5,
)
logger.info(
f"Watching video for {watching_time if watching_time > 0 else 'few '}s."
)

# Special handling for Reels with natural viewing behavior
if media_type == MediaType.REEL:
# Add variance to reel watching time for more natural behavior
watching_time = int(watching_time * uniform(0.85, 1.40))
logger.info(
f"Watching reel for {watching_time}s with natural behavior."
)
# Simulate natural reel viewing with random pauses
self._watch_reel_naturally(watching_time)
return None
else:
logger.info(
f"Watching video for {watching_time if watching_time > 0 else 'few '}s."
)

elif (
media_type in (MediaType.CAROUSEL, MediaType.PHOTO)
Expand All @@ -1329,6 +1387,40 @@ def watch_media(self, media_type: MediaType) -> None:
if watching_time > 0:
sleep(watching_time)

def _watch_reel_naturally(self, total_time: int) -> None:
"""
Watch reel with natural human-like behavior including random pauses
:param total_time: Total time to watch the reel
:return: None
"""
elapsed_time = 0
segment_count = randint(2, 4) # Break viewing into 2-4 segments

for i in range(segment_count):
if elapsed_time >= total_time:
break

# Watch for a random segment duration
segment_duration = uniform(
total_time / segment_count * 0.7,
total_time / segment_count * 1.3
)
segment_duration = min(segment_duration, total_time - elapsed_time)

logger.debug(f"Watching reel segment {i+1}/{segment_count} for {segment_duration:.1f}s")
sleep(segment_duration)
elapsed_time += segment_duration

# Random chance to "re-watch" a moment (tap to restart/rewind behavior)
if i < segment_count - 1 and randint(1, 100) <= 20:
logger.debug("Re-watching reel moment (natural behavior)")
sleep(uniform(0.8, 2.0))
elapsed_time += uniform(0.8, 2.0)

# Small pause between segments (natural attention span)
if i < segment_count - 1:
random_sleep(0.2, 0.7, modulable=False, log=False)

def _get_video_time_left(self) -> int:
timer = self.device.find(resourceId=ResourceID.TIMER)
if timer.exists():
Expand Down