feat: implement smart unfollow with resonance evaluation and close friends guard

This commit is contained in:
2026-04-21 02:45:05 +02:00
parent 0a02e901b6
commit ee3022e95d
3 changed files with 167 additions and 32 deletions

View File

@@ -56,38 +56,70 @@ def _run_zero_latency_unfollow_loop(device, zero_engine, nav_graph, configs, ses
try:
xml_dump = device.dump_hierarchy()
# Use Telepathic Engine to explicitly locate existing "Following" buttons in lists
nodes = telepathic._extract_semantic_nodes(xml_dump, "find 'Following' buttons next to usernames", threshold=0.7)
# Smart Unfollow Phase 1: Find user rows instead of just clicking "Following"
nodes = telepathic._extract_semantic_nodes(xml_dump, "find user profile rows in list", threshold=0.7)
action_taken = False
for node in nodes:
# Basic validation it's an interactive button
if node.get("skip") or not node.get("bounds"):
continue
# Tap the first valid following button we see
# 1. Tap the profile row to navigate to their page
_humanized_click(device, node["x"], node["y"])
action_taken = True
logger.debug(f"👆 Tapped following button at ({node['x']}, {node['y']})")
logger.debug(f"👆 Tapped profile row at ({node['x']}, {node['y']})")
# Check for confirmation dialog ("Unfollow @username?")
random_sleep(1.0, 2.0)
confirm_xml = device.dump_hierarchy()
confirm_nodes = telepathic._extract_semantic_nodes(confirm_xml, "find 'Unfollow' confirmation button", threshold=0.8)
# Wait for profile to load
random_sleep(1.5, 2.5)
profile_xml = device.dump_hierarchy()
if confirm_nodes and not confirm_nodes[0].get("skip"):
c_node = confirm_nodes[0]
_humanized_click(device, c_node["x"], c_node["y"])
# 2. Close Friend Guard
profile_text = profile_xml.lower()
if "enge freunde" in profile_text or "close friend" in profile_text:
logger.info("💚 [Anti-Friend] Profile is a Close Friend. Skipping unfollow.", extra={"color": Fore.GREEN})
device.back()
random_sleep(0.8, 1.5)
break # Go next in loop
logger.info("✅ [Unfollow Engine] Unfollowed a user in list.", extra={"color": Fore.GREEN})
session_state.totalUnfollowed += 1
total_unfollowed_this_session += 1
failed_scrolls = 0
# Unfollow cost logic
dopamine.boredom += random.uniform(1.0, 3.0)
random_sleep(1.5, 3.0)
# 3. Resonance Evaluation
resonance = cognitive_stack.get("resonance")
res_score = 0.5
if resonance:
# Parse the description from the XML (rough pass, ResonanceEngine handles noise)
res_score = resonance.calculate_resonance({"description": profile_xml})
# 4. Decision: If < 0.4, Unfollow. Else Keep.
if res_score < 0.4:
logger.info(f"🗑️ [Smart Cleanup] Resonance is low ({res_score:.2f}). Unfollowing.", extra={"color": Fore.YELLOW})
# Find 'Following' button on their profile
following_nodes = telepathic._extract_semantic_nodes(profile_xml, "find 'Following' button", threshold=0.7)
if following_nodes and not following_nodes[0].get("skip"):
f_node = following_nodes[0]
_humanized_click(device, f_node["x"], f_node["y"])
random_sleep(1.0, 2.0)
# Find 'Unfollow' confirm
confirm_xml = device.dump_hierarchy()
confirm_nodes = telepathic._extract_semantic_nodes(confirm_xml, "find 'Unfollow' confirmation button", threshold=0.8)
if confirm_nodes and not confirm_nodes[0].get("skip"):
c_node = confirm_nodes[0]
_humanized_click(device, c_node["x"], c_node["y"])
random_sleep(0.8, 1.5)
logger.info("✅ [Unfollow Engine] Unfollowed a user.", extra={"color": Fore.GREEN})
session_state.totalUnfollowed += 1
total_unfollowed_this_session += 1
failed_scrolls = 0
dopamine.boredom += random.uniform(1.0, 3.0)
else:
logger.info(f"✨ [Smart Cleanup] Resonance is high ({res_score:.2f}). Keeping subscription.", extra={"color": Fore.MAGENTA})
failed_scrolls = 0
# 5. Always return to the Following list
device.back()
random_sleep(1.0, 2.0)
break
if not action_taken:

View File

@@ -24,9 +24,13 @@ def unfollow_mock_dependencies():
dopamine.wants_to_change_feed.return_value = False
dopamine.boredom = 0.0
resonance = MagicMock()
resonance.calculate_resonance.return_value = 0.2
cognitive_stack = {
"telepathic": telepathic,
"dopamine": dopamine,
"resonance": resonance,
}
return device, zero_engine, nav_graph, configs, session_state, cognitive_stack
@@ -36,21 +40,29 @@ def test_unfollow_engine_basic_loop(unfollow_mock_dependencies):
telepathic = cognitive_stack["telepathic"]
# First node gives a "Following" button
# 1. Finds profile row
# 2. Finds following button on profile
# 3. Finds confirm dialog
telepathic._extract_semantic_nodes.side_effect = [
[{"x": 100, "y": 200, "skip": False, "bounds": "..."}], # following button
[{"x": 150, "y": 250, "skip": False}], # confirm dialog
[{"semantic_string": "Profile Row", "x": 100, "y": 200, "skip": False, "bounds": "..."}],
[{"semantic_string": "Following Button", "x": 150, "y": 250, "skip": False}],
[{"semantic_string": "Unfollow Confirm", "x": 200, "y": 300, "skip": False}],
[], # second iteration
[]
]
with patch("GramAddict.core.unfollow_engine._humanized_scroll_down") as mock_scroll, \
patch("GramAddict.core.bot_flow.sleep"), \
patch("GramAddict.core.bot_flow.random_sleep"), \
patch("GramAddict.core.utils.random_sleep"), \
patch("GramAddict.core.bot_flow._humanized_click") as mock_click:
device.dump_hierarchy.return_value = '<node text="Basic Bio"/>'
res = _run_zero_latency_unfollow_loop(device, zero_engine, nav_graph, configs, session_state, "FollowingList", cognitive_stack)
assert mock_click.call_count == 2 # Clicked following THEN clicked confirm
# Clicked profile -> following button -> confirm
assert mock_click.call_count == 3
assert session_state.totalUnfollowed == 1
assert res == "FEED_EXHAUSTED"

View File

@@ -16,12 +16,15 @@ class TestUnfollowEngine:
self.mock_dopamine = MagicMock()
self.mock_dopamine.is_app_session_over.return_value = False
self.mock_dopamine.wants_to_change_feed.return_value = False
# default boredom
self.mock_dopamine.boredom = 0.0
self.mock_resonance = MagicMock()
self.mock_resonance.calculate_resonance.return_value = 0.2
self.cognitive_stack = {
"telepathic": self.mock_telepathic,
"dopamine": self.mock_dopamine,
"resonance": self.mock_resonance,
}
self.mock_configs = MagicMock()
@@ -33,27 +36,39 @@ class TestUnfollowEngine:
self.logger = logging.getLogger("test")
def test_unfollow_loop_success(self, monkeypatch):
def test_unfollow_loop_unfollows_low_resonance(self, monkeypatch):
"""
Happy path: Finds 'Following' button -> Clicks it -> Finds 'Unfollow' confirmation -> Clicks it -> Increments totalUnfollowed.
Then dopamine signals change of feed.
Happy path (Smart Unfollow):
1. Clicks profile row.
2. Dumps profile UI.
3. Resonance is low (< 0.4).
4. Clicks 'Following' button on profile.
5. Clicks 'Unfollow' confirm.
6. Clicks back.
"""
def fake_extract_semantic_nodes(xml, intent, **kwargs):
if "Unfollow" in intent:
if "profile rows" in intent:
return [{"semantic_string": "Profile Row", "x": 100, "y": 200, "bounds": "[50,150]", "skip": False}]
elif "Unfollow" in intent:
return [{"semantic_string": "Unfollow Confirmation", "x": 500, "y": 1500, "bounds": "[100,200]", "skip": False}]
else:
return [{"semantic_string": "Following Button", "x": 900, "y": 600, "bounds": "[100,200]", "skip": False}]
self.mock_telepathic._extract_semantic_nodes.side_effect = fake_extract_semantic_nodes
self.mock_dopamine.wants_to_change_feed.side_effect = [True]
self.mock_device.dump_hierarchy.return_value = '<node text="Some Bio"/>'
# Patch local imports inside the bot_flow namespace since they are locally imported
# Patch local imports inside the bot_flow / unfollow_engine namespace
mock_sleep = MagicMock()
mock_click = MagicMock()
import GramAddict.core.bot_flow
monkeypatch.setattr(GramAddict.core.bot_flow, "sleep", mock_sleep)
monkeypatch.setattr(GramAddict.core.bot_flow, "random_sleep", mock_sleep)
monkeypatch.setattr(GramAddict.core.bot_flow, "_humanized_click", mock_click)
import GramAddict.core.utils
monkeypatch.setattr(GramAddict.core.utils, "random_sleep", mock_sleep)
result = _run_zero_latency_unfollow_loop(
self.mock_device, None, None, self.mock_configs, self.mock_session_state, "FollowingList", self.cognitive_stack
)
@@ -62,12 +77,88 @@ class TestUnfollowEngine:
assert result == "BOREDOM_CHANGE_FEED"
assert self.mock_session_state.totalUnfollowed == 1
# Assert humanized clicks were logged correctly
assert mock_click.call_count == 2
# Assert humanized clicks were logged correctly (Profile Row -> Following Button -> Unfollow Confirm)
assert mock_click.call_count == 3
mock_click.assert_has_calls([
call(self.mock_device, 100, 200),
call(self.mock_device, 900, 600),
call(self.mock_device, 500, 1500)
])
assert self.mock_device.back.call_count == 1
def test_unfollow_loop_keeps_high_resonance_profile(self, monkeypatch):
"""
If resonance is high, we keep following the account and go back to the list.
"""
self.mock_resonance.calculate_resonance.return_value = 0.8
def fake_extract_semantic_nodes(xml, intent, **kwargs):
if "profile rows" in intent:
return [{"semantic_string": "Profile Row", "x": 100, "y": 200, "bounds": "[50,150]", "skip": False}]
return []
self.mock_telepathic._extract_semantic_nodes.side_effect = fake_extract_semantic_nodes
self.mock_dopamine.wants_to_change_feed.side_effect = [True]
self.mock_device.dump_hierarchy.return_value = '<node text="Awesome Bio"/>'
mock_sleep = MagicMock()
mock_click = MagicMock()
import GramAddict.core.bot_flow
monkeypatch.setattr(GramAddict.core.bot_flow, "sleep", mock_sleep)
monkeypatch.setattr(GramAddict.core.bot_flow, "random_sleep", mock_sleep)
monkeypatch.setattr(GramAddict.core.bot_flow, "_humanized_click", mock_click)
import GramAddict.core.utils
monkeypatch.setattr(GramAddict.core.utils, "random_sleep", mock_sleep)
result = _run_zero_latency_unfollow_loop(
self.mock_device, None, None, self.mock_configs, self.mock_session_state, "FollowingList", self.cognitive_stack
)
assert result == "BOREDOM_CHANGE_FEED"
assert self.mock_session_state.totalUnfollowed == 0
# Clicked profile, then went back (no unfollow clicks)
assert mock_click.call_count == 1
mock_click.assert_has_calls([call(self.mock_device, 100, 200)])
assert self.mock_device.back.call_count == 1
def test_unfollow_loop_skips_close_friend(self, monkeypatch):
"""
If 'enge freunde' is in the XML, skip resonance eval, skip unfollowing, just go back.
"""
def fake_extract_semantic_nodes(xml, intent, **kwargs):
if "profile rows" in intent:
return [{"semantic_string": "Profile Row", "x": 100, "y": 200, "bounds": "[50,150]", "skip": False}]
return []
self.mock_telepathic._extract_semantic_nodes.side_effect = fake_extract_semantic_nodes
self.mock_dopamine.wants_to_change_feed.side_effect = [True]
# Mock XML with Enge Freunde
self.mock_device.dump_hierarchy.side_effect = [
'<node text="FollowingList"/>', # First dump on list
'<node text="Enge Freunde"/>' # Second dump on profile
]
mock_sleep = MagicMock()
mock_click = MagicMock()
import GramAddict.core.bot_flow
monkeypatch.setattr(GramAddict.core.bot_flow, "sleep", mock_sleep)
monkeypatch.setattr(GramAddict.core.bot_flow, "random_sleep", mock_sleep)
monkeypatch.setattr(GramAddict.core.bot_flow, "_humanized_click", mock_click)
import GramAddict.core.utils
monkeypatch.setattr(GramAddict.core.utils, "random_sleep", mock_sleep)
result = _run_zero_latency_unfollow_loop(
self.mock_device, None, None, self.mock_configs, self.mock_session_state, "FollowingList", self.cognitive_stack
)
assert result == "BOREDOM_CHANGE_FEED"
assert self.mock_session_state.totalUnfollowed == 0
assert mock_click.call_count == 1
assert self.mock_device.back.call_count == 1
# Resonance shouldn't even be called for close friends
self.mock_resonance.calculate_resonance.assert_not_called()
def test_unfollow_loop_scrolls_if_empty(self, monkeypatch):
"""