Skip to content

Commit

Permalink
search: adds functionality to remove search history suggestion
Browse files Browse the repository at this point in the history
  • Loading branch information
comarquet committed Nov 13, 2024
1 parent 9ce284a commit 1eb9489
Show file tree
Hide file tree
Showing 3 changed files with 104 additions and 8 deletions.
28 changes: 28 additions & 0 deletions tests/mixins/test_search.py
Original file line number Diff line number Diff line change
Expand Up @@ -115,3 +115,31 @@ def test_search_library(self, config, yt_oauth):
yt_oauth.search("beatles", filter="community_playlists", scope="library", limit=40)
with pytest.raises(Exception):
yt_oauth.search("beatles", filter="featured_playlists", scope="library", limit=40)

def test_remove_suggestion_valid(self, yt_auth):
yt_auth._latest_suggestions = ["Suggestion 1", "Suggestion 2", "Suggestion 3"]
yt_auth._latest_feedback_tokens = {1: "token1", 2: "token2", 3: "token3"}

response = yt_auth.remove_search_suggestion(1)
assert response is True

def test_remove_suggestion_invalid_number(self, yt_auth):
yt_auth._latest_suggestions = ["Suggestion 1", "Suggestion 2", "Suggestion 3"]
yt_auth._latest_feedback_tokens = {1: "token1", 2: "token2", 3: "token3"}

response = yt_auth.remove_search_suggestion(5) # Invalid suggestion number
assert response is False

def test_remove_suggestion_no_suggestions(self, yt_auth):
yt_auth._latest_suggestions = None
yt_auth._latest_feedback_tokens = None

with pytest.raises(ValueError, match="No suggestions available. Please run get_search_suggestions first to retrieve suggestions."):
yt_auth.remove_search_suggestion(1)

def test_remove_suggestion_no_feedback_token(self, yt_auth):
yt_auth._latest_suggestions = ["Suggestion 1", "Suggestion 2"]
yt_auth._latest_feedback_tokens = {1: "token1"} # Missing token for suggestion 2

response = yt_auth.remove_search_suggestion(2)
assert response is False
65 changes: 59 additions & 6 deletions ytmusicapi/mixins/search.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,10 @@


class SearchMixin(MixinProtocol):
def __init__(self):
self._latest_suggestions = None
self._latest_feedback_tokens = None

def search(
self,
query: str,
Expand Down Expand Up @@ -258,7 +262,7 @@ def parse_func(contents):

return search_results

def get_search_suggestions(self, query: str, detailed_runs=False) -> Union[list[str], list[dict]]:
def get_search_suggestions(self, query: str, detailed_runs=False, display_numbers=False) -> Union[list[str], list[dict]]:
"""
Get Search Suggestions
Expand All @@ -268,7 +272,9 @@ def get_search_suggestions(self, query: str, detailed_runs=False) -> Union[list[
suggestion along with the complete text (like many search services
usually bold the text typed by the user).
Default: False, returns the list of search suggestions in plain text.
:return: List of search suggestion results depending on ``detailed_runs`` param.
:param display_numbers: If True, numbers are displayed alongside search suggestions from the user history.
Default: False.
:return: List of search suggestion results depending on ``detailed_runs`` and ``display_numbers`` params.
Example response when ``query`` is 'fade' and ``detailed_runs`` is set to ``False``::
Expand All @@ -295,7 +301,8 @@ def get_search_suggestions(self, query: str, detailed_runs=False) -> Union[list[
{
"text": "d"
}
]
],
"number": 1
},
{
"text": "faded alan walker lyrics",
Expand All @@ -307,7 +314,8 @@ def get_search_suggestions(self, query: str, detailed_runs=False) -> Union[list[
{
"text": "d alan walker lyrics"
}
]
],
"number": 2
},
{
"text": "faded alan walker",
Expand All @@ -319,7 +327,8 @@ def get_search_suggestions(self, query: str, detailed_runs=False) -> Union[list[
{
"text": "d alan walker"
}
]
],
"number": 3
},
...
]
Expand All @@ -329,6 +338,50 @@ def get_search_suggestions(self, query: str, detailed_runs=False) -> Union[list[
endpoint = "music/get_search_suggestions"

response = self._send_request(endpoint, body)
search_suggestions = parse_search_suggestions(response, detailed_runs)
# Pass feedback_tokens as a dictionary to store tokens for deletion
feedback_tokens: dict[int, str] = {}
search_suggestions = parse_search_suggestions(response, detailed_runs, feedback_tokens)

# Store the suggestions and feedback tokens for later use
self._latest_suggestions = search_suggestions
self._latest_feedback_tokens = feedback_tokens

return search_suggestions

def remove_search_suggestion(self, number: int) -> bool:
"""
Remove a search suggestion from the user search history based on the number displayed next to it.
:param number: The number of the suggestion to be removed.
This number is displayed when the `detailed_runs` and `display_numbers` parameters are set to True
in the `get_search_suggestions` method.
:return: True if the operation was successful, False otherwise.
Example usage:
# Assuming you want to remove suggestion number 1
success = ytmusic.remove_search_suggestion(number=1)
if success:
print("Suggestion removed successfully")
else:
print("Failed to remove suggestion")
"""
if self._latest_suggestions is None or self._latest_feedback_tokens is None:
raise ValueError(
"No suggestions available. Please run get_search_suggestions first to retrieve suggestions."
)

feedback_token = self._latest_feedback_tokens.get(number)

if not feedback_token:
return False

body = {"feedbackTokens": [feedback_token]}
endpoint = "feedback"

response = self._send_request(endpoint, body)

if "feedbackResponses" in response and response["feedbackResponses"][0].get("isProcessed", False):
return True

return False
19 changes: 17 additions & 2 deletions ytmusicapi/parsers/search.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

UNIQUE_RESULT_TYPES = ["artist", "playlist", "song", "video", "station", "profile", "podcast", "episode"]
ALL_RESULT_TYPES = ["album", *UNIQUE_RESULT_TYPES]
FEEDBACK_TOKENS: dict[int, str] = {}


def get_search_result_type(result_type_local, result_types_local):
Expand Down Expand Up @@ -257,17 +258,25 @@ def _get_param2(filter):
return filter_params[filter]


def parse_search_suggestions(results, detailed_runs):
def parse_search_suggestions(results, detailed_runs, feedback_tokens):
if not results.get("contents", [{}])[0].get("searchSuggestionsSectionRenderer", {}).get("contents", []):
return []

raw_suggestions = results["contents"][0]["searchSuggestionsSectionRenderer"]["contents"]
suggestions = []

count = 1 # Used for deleting a search suggestion
for raw_suggestion in raw_suggestions:
if "historySuggestionRenderer" in raw_suggestion:
suggestion_content = raw_suggestion["historySuggestionRenderer"]
from_history = True
feedback_token = (
suggestion_content.get("serviceEndpoint", {}).get("feedbackEndpoint", {}).get("feedbackToken")
) # Extract feedbackToken if present

# Store the feedback token in the provided dictionary if it exists
if feedback_token:
feedback_tokens[count] = feedback_token
else:
suggestion_content = raw_suggestion["searchSuggestionRenderer"]
from_history = False
Expand All @@ -276,8 +285,14 @@ def parse_search_suggestions(results, detailed_runs):
runs = suggestion_content["suggestion"]["runs"]

if detailed_runs:
suggestions.append({"text": text, "runs": runs, "fromHistory": from_history})
suggestions.append({"text": text, "runs": runs, "fromHistory": from_history, "number": count})
else:
suggestions.append(text)

count += 1

return suggestions


def get_feedback_token(suggestion_number):
return FEEDBACK_TOKENS.get(suggestion_number)

0 comments on commit 1eb9489

Please sign in to comment.