Skip to content

. #150

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

Merged
MrIbrahem merged 1 commit into from
Jul 7, 2026
Merged

. #150

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
72 changes: 20 additions & 52 deletions newapi/api_client/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@

import copy
import logging
from typing import Any, Callable, Union
from typing import Any, Callable

import mwclient
import mwclient.errors
Expand Down Expand Up @@ -409,17 +409,11 @@ def client_request_retry(
**kwargs,
)

def post_continue(
def post_continue_dict(
self,
params: dict,
action: str,
_p_: str | None = None,
p_empty: Union[list, dict] | None = None,
max: int | None = None,
first: int | None = None,
_p_2: str | None = None,
_p_2_empty: Union[list, dict] | None = None,
**kwargs,
_load_data: Callable,
) -> dict[str, Any]:
"""
Drive a MediaWiki API continuation query to completion.
Expand All @@ -430,28 +424,18 @@ def post_continue(
Args:
params: Base API parameters.
action: Top-level JSON key to extract results from
(e.g. ``"query"``).
_p_: Sub-key inside *action* (default ``"pages"``).
p_empty: Seed value for the accumulator (list or dict).
max: Stop accumulating after this many results.
first: Return only the first element of the result list.
_p_2: Secondary sub-key when *first* is True.
_p_2_empty: Seed for secondary accumulator.

Returns:
Accumulated results as a list or dict, depending on *p_empty*.
"""
logger.debug("action=%s _p_=%s", action, _p_)
logger.debug("action=%s", action)

if isinstance(max, str) and max.isdigit():
max = int(max)
if max == 0:
max = 500_000

p_empty = p_empty if p_empty is not None else []
_p_2_empty = _p_2_empty if _p_2_empty is not None else []

results = p_empty
results = {}
continue_params: dict = {}
iterations = 0

Expand All @@ -469,36 +453,21 @@ def post_continue(
logger.debug("empty response, stopping")
break

continue_params = {}
continue_params = body.get("continue", {})

if action == "wbsearchentities":
data = body.get("search", [])
else:
continue_params = body.get("continue", {})
data = body.get(action, {}).get(_p_, p_empty)
if len(results) >= max:
logger.debug("max=%d reached, stopping", max)
break

if _p_ == "querypage":
data = data.get("results", [])
elif first:
if isinstance(data, list) and data:
data = data[0]
if _p_2:
data = data.get(_p_2, _p_2_empty)
data = _load_data(body)

if not data:
logger.debug("no data in response, stopping")
break

logger.debug("+%d items (total %d)", len(data), len(results))

if len(results) >= max:
logger.debug("max=%d reached, stopping", max)
break

if isinstance(results, list):
results.extend(data)
else:
results = {**results, **data}
results = {**results, **data}

logger.debug("done, %d total results", len(results))
return results
Expand Down Expand Up @@ -535,15 +504,14 @@ def post_continue_list(
max = 500_000
results = []
continue_params: dict = {}
iterations = 0

while continue_params or iterations == 0:
while True:
page_params = copy.deepcopy(params)
iterations += 1

if continue_params:
logger.debug("Applying continue_params: %s", continue_params)
page_params.update(continue_params)
if not continue_params:
break
logger.debug("Applying continue_params: %s", continue_params)
page_params.update(continue_params)

body = self.client_request(page_params)

Expand All @@ -553,6 +521,10 @@ def post_continue_list(

continue_params = body.get("continue", {})

if len(results) >= max:
logger.debug("max=%d reached, stopping", max)
break

data = _load_data(body)

if not data:
Expand All @@ -561,10 +533,6 @@ def post_continue_list(

logger.debug("+%d items (total %d)", len(data), len(results))

if len(results) >= max:
logger.debug("max=%d reached, stopping", max)
break

results.extend(data)

logger.debug("done, %d total results", len(results))
Expand Down
25 changes: 0 additions & 25 deletions newapi/client_wiki/pages/super_page.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -965,7 +965,6 @@ def page_links_query(self, plnamespace: str = "*"):
def _load_data(body):
return body.get("query", {}).get("links") or []

# ---
data: list = self.login_bot.post_continue_list(
params=params,
action="query",
Expand Down Expand Up @@ -1026,30 +1025,6 @@ def _load_data(body):

return revisions

def post_continue(
self,
params,
action,
_p_: str = "pages",
p_empty=None,
max: int = 500000,
first: bool = False,
_p_2: str = "",
_p_2_empty=None,
**kwargs,
):
return self.login_bot.post_continue(
params,
action,
_p_=_p_,
p_empty=p_empty,
max=max,
first=first,
_p_2=_p_2,
_p_2_empty=_p_2_empty,
**kwargs,
)

def __getitem__(self, key):
if key == "q":
return self.get_qid()
Expand Down
100 changes: 20 additions & 80 deletions newapi/super/S_API/bot_api.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,9 @@
""" """

import copy
import datetime
import logging
from collections.abc import KeysView
from datetime import timedelta
from typing import Any, Callable

import tqdm

Expand Down Expand Up @@ -262,7 +260,7 @@ def _load_data(body):
return body.get("query", {}).get("allpages") or []

# ---
newp = self.post_continue_list(params=params, action="query", max=limit_all, _load_data=_load_data)
newp = self.login_bot.post_continue_list(params=params, action="query", max=limit_all, _load_data=_load_data)
# ---
logger.debug(f"<<lightpurple>> --- : find {len(newp)} pages.")
# ---
Expand Down Expand Up @@ -317,7 +315,7 @@ def _load_data(body):
return body.get("query", {}).get("pages") or []

# ---
newp = self.post_continue_list(
newp = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -386,7 +384,7 @@ def _load_data(body):
return body.get("query", {}).get("prefixsearch") or []

# ---
newp = self.post_continue_list(
newp = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -443,7 +441,7 @@ def _load_data(body):
return body.get("query", {}).get("search") or []

# ---
search = self.post_continue_list(
search = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -505,7 +503,7 @@ def _load_data(body):
return body.get("query", {}).get("recentchanges") or []

# ---
json1 = self.post_continue_list(
json1 = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -541,7 +539,7 @@ def _load_data(body):
return body.get("query", {}).get("usercontribs") or []

# ---
results = self.post_continue_list(
results = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -685,7 +683,7 @@ def _load_data(body):
return data

# ---
results = self.post_continue_list(
results = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -714,7 +712,7 @@ def _load_data(body):
return body.get("query", {}).get("pages") or []

# ---
results = self.post_continue_list(
results = self.login_bot.post_continue_list(
params,
"query",
"pages",
Expand Down Expand Up @@ -746,7 +744,7 @@ def _load_data(body):
return body.get("query", {}).get("pages") or []

# ---
results = self.post_continue_list(
results = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -814,12 +812,12 @@ def querypage_list(self, qppage: str = "Wantedcategories", qplimit=None, max=Non
if qppage not in qppage_values:
logger.info(f"<<lightred>> qppage {qppage} not in qppage_values.")

# ---
def _load_data(body):
return body.get("query", {}).get("querypage") or []
query = body.get("query", {})
return query.get("querypage") or query.get("results") or []

# ---
results = self.post_continue_list(
results = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -849,7 +847,7 @@ def _load_data(body):
return body.get("query", {}).get("pages") or []

# ---
results = self.post_continue_list(
results = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -943,7 +941,7 @@ def _load_data(body):
return body.get("query", {}).get("pageswithprop") or []

# ---
results = self.post_continue_list(
results = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -975,7 +973,7 @@ def _load_data(body):
return body.get("query", {}).get("redirects") or []

# ---
json1 = self.post_continue_list(
json1 = self.login_bot.post_continue_list(
params=params,
action="query",
_load_data=_load_data,
Expand Down Expand Up @@ -1328,68 +1326,10 @@ def client_request_safe(
**kwargs,
)

def post_continue_list(
self,
params: dict,
action: str,
_load_data: Callable,
max: int | None = None,
) -> dict[str, Any]:
"""
Drive a MediaWiki API continuation query to completion.

Iterates the ``continue`` token until all pages are fetched or *max*
results have been collected.

Args:
params: Base API parameters.
action: Top-level JSON key to extract results from
(e.g. ``"query"``).
max: Stop accumulating after this many results.

Returns:
Accumulated results as a list
"""
logger.debug("action=%s", action)

if isinstance(max, str) and max.isdigit():
max = int(max)
if max == 0:
max = 500_000
if max is None:
max = 500_000
results = []
continue_params: dict = {}

while True:
page_params = copy.deepcopy(params)

if not continue_params:
break
logger.debug("Applying continue_params: %s", continue_params)
page_params.update(continue_params)

body = self.login_bot.client_request(page_params)

if not body:
logger.debug("empty response, stopping")
break

continue_params = body.get("continue", {})
def __repr__(self) -> str:
return f"NewApi(lang={self.lang!r}, username={self.username!r})"

data = _load_data(body)

if not data:
logger.debug("no data in response, stopping")
break

logger.debug("+%d items (total %d)", len(data), len(results))

if len(results) >= max:
logger.debug("max=%d reached, stopping", max)
break

results.extend(data)

logger.debug("done, %d total results", len(results))
return results
__all__ = [
"NewApi",
]
Loading
Loading