Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
21 changes: 18 additions & 3 deletions google_play_scraper/constants/element.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,15 +21,26 @@ def __init__(
def extract_content(self, source: dict) -> Any:
try:
if self.ds_num is None:
result = nested_lookup(source, self.data_map)
# Special handling for mixed int/string paths
current = source
for key in self.data_map:
if isinstance(current, dict):
current = current[key]
elif isinstance(current, list):
current = current[key]
else:
raise KeyError(f"Cannot index {type(current)} with {key}")
result = current
else:
result = nested_lookup(
source["ds:{}".format(self.ds_num)], self.data_map
)

if self.post_processor is not None:
result = self.post_processor(result)
except:
except Exception as e:
# NOTE: JoMingyu: the errors like list index out of range are caught pretty often. Others too
# print('Error: ', e)
if isinstance(self.fallback_value, ElementSpec):
result = self.fallback_value.extract_content(source)
else:
Expand Down Expand Up @@ -195,7 +206,11 @@ class ElementSpecs:
)

SearchResultOnTop = {
"appId": ElementSpec(None, [11, 0, 0]),
"appId": ElementSpec(
None,
[3, "12", 0, 0],
fallback_value=ElementSpec(None, [11, 0, 0])
),
"icon": ElementSpec(None, [2, 95, 0, 3, 2]),
"screenshots": ElementSpec(
None,
Expand Down
8 changes: 5 additions & 3 deletions google_play_scraper/features/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,14 +8,16 @@
from google_play_scraper.utils.request import get


def app(app_id: str, lang: str = "en", country: str = "us") -> Dict[str, Any]:
def app(
app_id: str, lang: str = "en", country: str = "us", timeout: int | None = None
) -> Dict[str, Any]:
url = Formats.Detail.build(app_id=app_id, lang=lang, country=country)

try:
dom = get(url)
dom = get(url, timeout=timeout)
except NotFoundError:
url = Formats.Detail.fallback_build(app_id=app_id, lang=lang)
dom = get(url)
dom = get(url, timeout=timeout)
return parse_dom(dom=dom, app_id=app_id, url=url)


Expand Down
36 changes: 26 additions & 10 deletions google_play_scraper/features/search.py
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,22 @@ def search(
top_result = dataset["ds:4"][0][1][0][23][16]
except IndexError:
top_result = None

# Try to get appId for top result
top_app_id = None
if top_result:
try:
# AppId is in the last element of the array, in a dict with key "12"
if isinstance(top_result, list) and len(top_result) > 3:
last_elem = top_result[3]
if isinstance(last_elem, dict) and "12" in last_elem:
top_app_id = last_elem["12"][0][0]
except (IndexError, KeyError, TypeError):
try:
# Try alternative path for older structure
top_app_id = top_result[11][0][0]
except (IndexError, KeyError, TypeError):
pass

success = False
# different idx for different countries and languages
Expand All @@ -55,16 +71,16 @@ def search(

n_apps = min(len(dataset), n_hits)

search_results = (
[
{
k: spec.extract_content(top_result)
for k, spec in ElementSpecs.SearchResultOnTop.items()
}
]
if top_result
else []
)
search_results = []
if top_result:
top_app = {}
for k, spec in ElementSpecs.SearchResultOnTop.items():
if k == "appId" and top_app_id:
top_app[k] = top_app_id
else:
content = spec.extract_content(top_result)
top_app[k] = content
search_results.append(top_app)

for app_idx in range(n_apps - len(search_results)):
app = {}
Expand Down
8 changes: 4 additions & 4 deletions google_play_scraper/utils/request.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,9 +12,9 @@
RATE_LIMIT_DELAY = 5


def _urlopen(obj):
def _urlopen(obj, timeout: int | None = None):
try:
resp = urlopen(obj)
resp = urlopen(obj, timeout=timeout)
except HTTPError as e:
if e.code == 404:
raise NotFoundError("App not found(404).")
Expand Down Expand Up @@ -44,5 +44,5 @@ def post(url: str, data: Union[str, bytes], headers: dict) -> str:
raise last_exception


def get(url: str) -> str:
return _urlopen(url)
def get(url: str, timeout: int | None = None) -> str:
return _urlopen(url, timeout)