From d49266ac599d0c9b3f09d8f46ec4603a2b071530 Mon Sep 17 00:00:00 2001 From: Ibrahem Date: Tue, 12 May 2026 05:20:48 +0300 Subject: [PATCH 1/9] Update bot_api.py --- newapi/super/S_API/bot_api.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/newapi/super/S_API/bot_api.py b/newapi/super/S_API/bot_api.py index e4d3ac5..68f5e4e 100644 --- a/newapi/super/S_API/bot_api.py +++ b/newapi/super/S_API/bot_api.py @@ -425,11 +425,11 @@ def Get_Newpages( offset_hours=False, ): if three_houers: - dd = datetime.datetime.utcnow() - timedelta(hours=3) + dd = datetime.datetime.now(datetime.UTC) - timedelta(hours=3) rcstart = dd.strftime("%Y-%m-%dT%H:%M:00.000Z") elif offset_minutes and isinstance(offset_minutes, int): - dd = datetime.datetime.utcnow() - timedelta(minutes=offset_minutes) + dd = datetime.datetime.now(datetime.UTC) - timedelta(minutes=offset_minutes) rcstart = dd.strftime("%Y-%m-%dT%H:%M:00.000Z") params = { From 1ed800e24f280cc778b7803f37abc463fb7b8f29 Mon Sep 17 00:00:00 2001 From: Ibrahem Date: Tue, 12 May 2026 05:21:49 +0300 Subject: [PATCH 2/9] Update client.py --- newapi/api_client/client.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/newapi/api_client/client.py b/newapi/api_client/client.py index 62ec259..8bbdf26 100644 --- a/newapi/api_client/client.py +++ b/newapi/api_client/client.py @@ -433,6 +433,9 @@ def __init__( logger.debug("Creating mwclient.Site for %s.%s", lang, family) self.api_url = f"https://{self.lang}.{self.family}.org/w/api.php" + if self.api_url == "https://www.mdwiki.org/w/api.php": + self.api_url = "https://mdwiki.org/w/api.php" + try: self._site = mwclient.Site(f"{self.lang}.{self.family}.org", do_init=False) except Exception as exc: From e09e424f1d6642796c62d4a0ff34f33350c9fd48 Mon Sep 17 00:00:00 2001 From: Ibrahem Date: Tue, 12 May 2026 05:39:44 +0300 Subject: [PATCH 3/9] Update client.py --- newapi/api_client/client.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/newapi/api_client/client.py b/newapi/api_client/client.py index 8bbdf26..fa30aa7 100644 --- a/newapi/api_client/client.py +++ b/newapi/api_client/client.py @@ -439,7 +439,7 @@ def __init__( try: self._site = mwclient.Site(f"{self.lang}.{self.family}.org", do_init=False) except Exception as exc: - raise WikiClientError(f"Invalid site ID: {self.lang}.{self.family}") from exc + raise WikiClientError(f"Invalid site ID: {self.lang}.{self.family}.org") from exc # ── Inject saved cookies ─────────────────────────────────────────── # mwclient stores its requests.Session at site.connection. @@ -614,7 +614,7 @@ def _do_login(self) -> None: try: self._site.login(self.username, self._password) except mwclient.errors.LoginError as exc: - raise LoginError(f"login failed for {self.username} on {self.lang}.{self.family}: {exc}") from exc + raise LoginError(f"login failed for {self.username} on {self.lang}.{self.family}.org: {exc}") from exc if self._site.logged_in: logger.info( From 3067b5f200685e854d9a7289362e1b23aed733ab Mon Sep 17 00:00:00 2001 From: Ibrahem Date: Tue, 12 May 2026 05:41:17 +0300 Subject: [PATCH 4/9] Update client.py --- newapi/api_client/client.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/newapi/api_client/client.py b/newapi/api_client/client.py index fa30aa7..1b9333e 100644 --- a/newapi/api_client/client.py +++ b/newapi/api_client/client.py @@ -430,7 +430,7 @@ def __init__( self._cookie_path: Path = get_cookie_path(cookies_dir or settings.paths.cookies_dir, family, lang, username) # ── mwclient Site ────────────────────────────────────────────────── - logger.debug("Creating mwclient.Site for %s.%s", lang, family) + logger.debug("Creating mwclient.Site for %s.%s.org", lang, family) self.api_url = f"https://{self.lang}.{self.family}.org/w/api.php" if self.api_url == "https://www.mdwiki.org/w/api.php": @@ -471,7 +471,7 @@ def _on_assertnameduserfailed(self) -> None: Called by the base-class retry loop; never call directly. """ logger.warning( - "assertnameduserfailed for %s on %s.%s — clearing cookies and re-logging in", + "assertnameduserfailed for %s on %s.%s.org — clearing cookies and re-logging in", self.username, self.lang, self.family, @@ -618,7 +618,7 @@ def _do_login(self) -> None: if self._site.logged_in: logger.info( - "Logged in successfully as %s on %s.%s", + "Logged in successfully as %s on %s.%s.org", self.username, self.lang, self.family, @@ -636,7 +636,7 @@ def login(self, force: bool = False) -> None: """ if force or not self._site.logged_in: logger.info( - "Forcing re-login for %s on %s.%s", + "Forcing re-login for %s on %s.%s.org", self.username, self.lang, self.family, From ea216d862ba50c4f17dc633f6cb834fcb1623bc2 Mon Sep 17 00:00:00 2001 From: Ibrahem Date: Tue, 12 May 2026 05:48:29 +0300 Subject: [PATCH 5/9] Update client.py --- newapi/api_client/client.py | 37 ++++++++++++++++++++++--------------- 1 file changed, 22 insertions(+), 15 deletions(-) diff --git a/newapi/api_client/client.py b/newapi/api_client/client.py index 1b9333e..184126b 100644 --- a/newapi/api_client/client.py +++ b/newapi/api_client/client.py @@ -407,6 +407,7 @@ def __init__( username: str, password: str, cookies_dir: str | None = settings.paths.cookies_dir, + use_cookies: bool = True, ) -> None: """ Initialise the client, load any saved cookies, and ensure the session @@ -426,9 +427,8 @@ def __init__( self._password = password # kept private — never log or expose this # ── Cookie path ──────────────────────────────────────────────────── - - self._cookie_path: Path = get_cookie_path(cookies_dir or settings.paths.cookies_dir, family, lang, username) - + self._cookie_path = None + self.use_cookies = use_cookies # ── mwclient Site ────────────────────────────────────────────────── logger.debug("Creating mwclient.Site for %s.%s.org", lang, family) self.api_url = f"https://{self.lang}.{self.family}.org/w/api.php" @@ -443,8 +443,11 @@ def __init__( # ── Inject saved cookies ─────────────────────────────────────────── # mwclient stores its requests.Session at site.connection. - self.cj = self._make_cookiejar(self._cookie_path) - self._site.connection.cookies = self.cj + self.cj = None + if self.use_cookies: + self._cookie_path: Path = get_cookie_path(cookies_dir or settings.paths.cookies_dir, family, lang, username) + self.cj = self._make_cookiejar(self._cookie_path) + self._site.connection.cookies = self.cj # ── Wrap the session with retry / CSRF / maxlag logic ────────────── # wrap_session(self._site.connection, self._site) @@ -476,7 +479,8 @@ def _on_assertnameduserfailed(self) -> None: self.lang, self.family, ) - _delete_cookie_file(self._cookie_path, reason="assertnameduserfailed") + if self.use_cookies: + _delete_cookie_file(self._cookie_path, reason="assertnameduserfailed") self._do_login() # ------------------------------------------------------------------ @@ -564,14 +568,16 @@ def _ensure_logged_in(self) -> None: if getattr(self._site, "logged_in", None): logger.info(f"Session already authenticated {self._site.logged_in=}") return - if self._cookie_path.exists(): - try: - self._site.site_init() - if self._site.logged_in: - logger.info("Revived session via cookies as %s", self._site.username) - return - except Exception: - logger.exception("Error in site_init") + + if self.use_cookies: + if self._cookie_path.exists(): + try: + self._site.site_init() + if self._site.logged_in: + logger.info("Revived session via cookies as %s", self._site.username) + return + except Exception: + logger.exception("Error in site_init") # if not self._site.logged_in: self._do_login() # don't login yet, user can use login() method @@ -623,7 +629,8 @@ def _do_login(self) -> None: self.lang, self.family, ) - self.save_cookies(self.cj) + if self.use_cookies: + self.save_cookies(self.cj) # ── Public methods ───────────────────────────────────────────────────── From f5538ec3c336a5c807915e85531b54285fc9e4d7 Mon Sep 17 00:00:00 2001 From: Ibrahem Date: Tue, 12 May 2026 05:49:53 +0300 Subject: [PATCH 6/9] Update all_apis.py --- newapi/client_wiki/all_apis.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/newapi/client_wiki/all_apis.py b/newapi/client_wiki/all_apis.py index 9f2b46f..88e75c3 100644 --- a/newapi/client_wiki/all_apis.py +++ b/newapi/client_wiki/all_apis.py @@ -21,11 +21,19 @@ class AllAPIS: new_api = main_api.NewApi() """ - def __init__(self, lang: str, family: str, username: str, password: str) -> None: + def __init__( + self, + lang: str, + family: str, + username: str, + password: str, + use_cookies: bool = True, + ) -> None: self.lang = lang self.family = family self.username = username self.password = password + self.use_cookies = use_cookies self.login_bot = self._login() def MainPage(self, title: str, *args, **kwargs) -> super_page.MainPage: @@ -44,6 +52,7 @@ def _login(self) -> WikiLoginClient: family=self.family, username=self.username, password=self.password, + use_cookies=self.use_cookies, ) return client From f42d2bfcfdfd43fdb9c3dd6a548895d5657531ae Mon Sep 17 00:00:00 2001 From: Ibrahem Date: Tue, 12 May 2026 05:56:42 +0300 Subject: [PATCH 7/9] .0 --- newapi/api_client/client.py | 2 +- newapi/client_wiki/categories/category_db.py | 2 +- newapi/client_wiki/pages/super_page.py | 30 +- newapi/super/S_API/bot.py | 385 ------------------ newapi/super/S_API/bot_api.py | 386 ++++++++++++++++++- 5 files changed, 391 insertions(+), 414 deletions(-) delete mode 100644 newapi/super/S_API/bot.py diff --git a/newapi/api_client/client.py b/newapi/api_client/client.py index 184126b..650bc8c 100644 --- a/newapi/api_client/client.py +++ b/newapi/api_client/client.py @@ -813,7 +813,7 @@ def post_continue( logger.debug("Applying continue_params: %s", continue_params) page_params.update(continue_params) - body = self.client_request(page_params) + body = self.client_request_safe(page_params) if not body: logger.debug("empty response, stopping") diff --git a/newapi/client_wiki/categories/category_db.py b/newapi/client_wiki/categories/category_db.py index e93ff3f..f9b09fc 100644 --- a/newapi/client_wiki/categories/category_db.py +++ b/newapi/client_wiki/categories/category_db.py @@ -257,7 +257,7 @@ def get_cat_new(self, cac: str) -> dict: if continue_params: params.update(continue_params) - api_data = self.login_bot.client_request(params, method="get") + api_data = self.login_bot.client_request_safe(params, method="get") if not api_data: logger.info(f"api is False for {cac}") diff --git a/newapi/client_wiki/pages/super_page.py b/newapi/client_wiki/pages/super_page.py index 31ea505..ef51d50 100644 --- a/newapi/client_wiki/pages/super_page.py +++ b/newapi/client_wiki/pages/super_page.py @@ -138,7 +138,7 @@ def import_page(self, family="wikipedia"): "assignknownusers": 1, } - data = self.client_request(params) + data = self.login_bot.client_request_safe(params) done = data.get("import", [{}])[0].get("revisions", 0) @@ -166,7 +166,7 @@ def find_create_data(self): "rvdir": "newer", } - data = self.login_bot.client_request(params, method="get") + data = self.login_bot.client_request_safe(params, method="get") pages = data.get("query", {}).get("pages", {}) @@ -207,7 +207,7 @@ def get_text(self, redirects=False): if redirects: params["redirects"] = 1 - data = self.login_bot.client_request(params, method="get") + data = self.login_bot.client_request_safe(params, method="get") pages = data.get("query", {}).get("pages", {}) @@ -288,7 +288,7 @@ def get_infos(self): # _data_ = { "continue": {}, "query": { "pages": { "9124097": { "pageid": 9124097, "ns": 0, "title": "طواف العالم للدراجات 2023", "categories": [], "langlinks": [], "templates": [{ "ns": 10, "title": "قالب:-" }], "linkshere": [{ "pageid": 189150, "ns": 0, "title": "طواف فرنسا" }], "iwlinks": [{ "prefix": "commons", "*": "Category:2023_UCI_World_Tour" }], "contentmodel": "wikitext", "pagelanguage": "ar", "pagelanguagehtmlcode": "ar", "pagelanguagedir": "rtl", "touched": "2023-03-07T11:53:53Z", "lastrevid": 61366100, "length": 985, } } }, } - data = self.login_bot.client_request(params, method="get") + data = self.login_bot.client_request_safe(params, method="get") # xs = { 'batchcomplete': True, 'query': { 'pages': [{ 'pageid': 151314, 'ns': 10, 'title': 'قالب:أوب', 'categories': [{ 'ns': 14, 'title': 'تصنيف:قوالب تستخدم أنماط القوالب', 'sortkey': '', 'sortkeyprefix': '', 'hidden': False }, { 'ns': 14, 'title': 'تصنيف:cc', 'sortkey': 'v', 'sortkeyprefix': 'أوب', 'hidden': True }], 'langlinks': [{ 'lang': 'bh', 'title': 'टेम्पलेट:AWB' }], 'templates': [{ 'ns': 10, 'title': 'قالب:No redirect' }], 'linkshere': [{ 'pageid': 308641, 'ns': 10, 'title': 'قالب:AWB', 'redirect': True }], 'iwlinks': [{ 'prefix': 'd', 'title': 'Q4063270' }], 'contentmodel': 'wikitext', 'pagelanguage': 'ar', 'pagelanguagehtmlcode': 'ar', 'pagelanguagedir': 'rtl', 'touched': '2023-03-05T22:10:23Z', 'lastrevid': 61388266, 'length': 3477, }] }, } @@ -351,7 +351,7 @@ def get_text_html(self): "prop": "text", } - data = self.client_request(params) + data = self.login_bot.client_request_safe(params) # _data_ = { 'warnings': { 'main': { 'warnings': 'Unrecognized parameter: bot.' } }, 'parse': { 'title': 'ويكيبيديا:ملعب', 'pageid': 361534, 'text': '' } } @@ -367,7 +367,7 @@ def get_redirect_target(self): "redirects": 1, } - data = self.login_bot.client_request(params, method="get") + data = self.login_bot.client_request_safe(params, method="get") # _pages_ = { 'batchcomplete': '', 'query': { 'redirects': [{ 'from': 'Yemen', 'to': 'اليمن' }], 'pages': {}, 'normalized': [{ 'from': 'yemen', 'to': 'Yemen' }] } } @@ -390,7 +390,7 @@ def get_words(self): "srsearch": self.title, "srlimit": srlimit, } - data = self.client_request(params) + data = self.login_bot.client_request_safe(params) if not data: return 0 @@ -431,7 +431,7 @@ def get_extlinks(self): # params = {**params, **continue_params} params.update(continue_params) - json1 = self.login_bot.client_request(params, method="get") + json1 = self.login_bot.client_request_safe(params, method="get") continue_params = json1.get("continue", {}) @@ -458,7 +458,7 @@ def get_userinfo(self): "ususers": self.user, } - data = self.login_bot.client_request(params, method="get") + data = self.login_bot.client_request_safe(params, method="get") # _userinfo_ = { "id": 229481, "name": "Mr. Ibrahem", "groups": ["editor", "reviewer", "rollbacker", "*", "user", "autoconfirmed"] } @@ -717,7 +717,7 @@ def save( # params['basetimestamp'] = self.revisions_data.timestamp - pop = self.login_bot.client_request(params) + pop = self.login_bot.client_request_safe(params) if not pop: return False @@ -759,7 +759,7 @@ def purge(self): "titles": self.title, } - data = self.client_request(params) + data = self.login_bot.client_request_safe(params) if not data: logger.info("<> ** purge error. ") @@ -835,7 +835,7 @@ def create( "createonly": 1, } - pop = self.login_bot.client_request(params) + pop = self.login_bot.client_request_safe(params) if not pop: return False @@ -893,7 +893,7 @@ def page_backlinks(self, ns=0): # x = { 'batchcomplete': True, 'limits': { 'backlinks': 2500 }, 'query': { 'redirects': [{ 'from': 'فريدريش زيمرمان', 'to': 'فريدريش تسيمرمان' }], 'pages': [{ 'pageid': 2941285, 'ns': 0, 'title': 'فولفغانغ شويبله' }, { 'pageid': 4783977, 'ns': 0, 'title': 'وزارة الشؤون الرقمية والنقل' }, { 'pageid': 5218323, 'ns': 0, 'title': 'فريدريش تسيمرمان' }, { 'pageid': 6662649, 'ns': 0, 'title': 'غونتر كراوزه' }] } } - # data = self.client_request(params) + # data = self.client_request_safe(params) # pages = data.get("query", {}).get("pages", []) pages = self.post_continue(params, "query", _p_="pages", p_empty=[]) @@ -921,7 +921,7 @@ def page_links(self) -> list: "formatversion": "2", "page": self.title, } - # data = self.client_request(params) + # data = self.client_request_safe(params) # data = data.get('parse', {}).get('links', []) data: list = self.post_continue(params, "parse", _p_="links", p_empty=[]) @@ -942,7 +942,7 @@ def page_links_query(self, plnamespace="*"): "pllimit": "max", "converttitles": 1, } - # data = self.client_request(params) + # data = self.client_request_safe(params) # data = data.get('query', {}).get('links', []) data = self.post_continue(params, "query", _p_="links", p_empty=[]) diff --git a/newapi/super/S_API/bot.py b/newapi/super/S_API/bot.py deleted file mode 100644 index e826f6f..0000000 --- a/newapi/super/S_API/bot.py +++ /dev/null @@ -1,385 +0,0 @@ -""" - -from .super.S_API.bot import BotsAPIS - -""" - -import logging - -from ...client_wiki.api_utils.ask_bot import AskBot -from ...client_wiki.api_utils.handel_errors import HandleErrors - -logger = logging.getLogger(__name__) - -yes_answer = ["y", "a", "", "Y", "A", "all", "aaa"] -file_name = "bot_api.py" - - -class BotsAPIS(HandleErrors, AskBot): - def __init__(self): - # print("class BotsAPIS:") - # --- - self.username = getattr(self, "username", "") - # --- - super().__init__() - - def Add_To_Bottom(self, text, summary, title, poss="Head|Bottom"): - # --- - if not title.strip(): - logger.info('** .. title == ""') - return False - # --- - if not text.strip(): - logger.info('** .. text == ""') - return False - # --- - logger.debug(f"** .. [[{title}]] ") - # --- - user = self.username - # --- - ask = self.ask_put( - newtext=text, - message=f"** Add_To {poss} .. [[{title}]] ", - job="Add_To_Bottom", - username=user, - summary=summary, - ) - # --- - if ask is False: - return False - # --- - params = { - "action": "edit", - "format": "json", - "title": title, - "summary": summary, - "notminor": 1, - "nocreate": 1, - "utf8": 1, - } - # --- - if poss == "Head": - params["prependtext"] = f"{text.strip()}\n" - else: - params["appendtext"] = f"\n{text.strip()}" - # --- - results = self.client_request(params) - # --- - if not results: - return "" - # --- - data = results.get("edit", {}) - result = data.get("result", "") - # --- - if result == "Success": - logger.info(f"<>** True. title:({title})") - return True - # --- - error = results.get("error", {}) - # --- - if error != {}: - print(results) - er = self.handle_err(error, function="Add_To_Bottom", params=params) - # --- - return er - # --- - return True - - def move( - self, - old_title, - to, - reason="", - noredirect=False, - movesubpages=False, - return_dict=False, - ): - # --- - logger.info(f"<> def [[{old_title}]] to [[{to}]] ") - # --- - params = { - "action": "move", - "format": "json", - "from": old_title, - "to": to, - "movetalk": 1, - "formatversion": 2, - } - # --- - if noredirect: - params["noredirect"] = 1 - if movesubpages: - params["movesubpages"] = 1 - # --- - if reason: - params["reason"] = reason - # --- - if old_title == to: - logger.debug(f"<>** old_title == to {to} ") - return {} - # --- - message = f"Do you want to move page:[[{old_title}]] to [[{to}]]?" - # --- - user = self.username - # --- - if not self.ask_put(message=message, job="move", username=user): - return {} - # --- - data = self.client_request(params) - # { "move": { "from": "d", "to": "d2", "reason": "wrong", "redirectcreated": true, "moveoverredirect": false } } - # --- - if not data: - logger.info("no data") - return {} - # --- - _expend_data = { - "move": { - "from": "User:Mr. Ibrahem", - "to": "User:Mr. Ibrahem/x", - "reason": "wrong title", - "redirectcreated": True, - "moveoverredirect": False, - "talkmove-errors": [ - { - "message": "content-not-allowed-here", - "params": [ - "Structured Discussions board", - "User talk:Mr. Ibrahem/x", - "main", - ], - "code": "contentnotallowedhere", - "type": "error", - }, - { - "message": "flow-error-allowcreation-flow-create-board", - "params": [], - "code": "flow-error-allowcreation-flow-create-board", - "type": "error", - }, - ], - "subpages": { - "errors": [ - { - "message": "cant-move-subpages", - "params": [], - "code": "cant-move-subpages", - "type": "error", - } - ] - }, - "subpages-talk": { - "errors": [ - { - "message": "cant-move-subpages", - "params": [], - "code": "cant-move-subpages", - "type": "error", - } - ] - }, - } - } - # --- - move_done = data.get("move", {}) - error = data.get("error", {}) - error_code = error.get("code", "") # missingtitle - # --- - # elif "Please choose another name." in r4: - # --- - if move_done: - logger.info("<>** true.") - # --- - if return_dict: - return move_done - # --- - return True - # --- - if error: - if error_code == "ratelimited": - logger.info("<> ratelimited:") - return self.move( - old_title, - to, - reason=reason, - noredirect=noredirect, - movesubpages=movesubpages, - return_dict=return_dict, - ) - - if error_code == "articleexists": - logger.info("<> articleexists") - return "articleexists" - - logger.info("<> error") - logger.info(error) - - return {} - # --- - return {} - - def expandtemplates(self, text): - # --- - params = { - "action": "expandtemplates", - "format": "json", - "text": text, - "prop": "wikitext", - "formatversion": 2, - } - # --- - data = self.client_request(params) - # --- - if not data: - return text - # --- - newtext = data.get("expandtemplates", {}).get("wikitext") or text - # --- - return newtext - - def Parse_Text(self, line, title): - # --- - params = { - "action": "parse", - "prop": "wikitext", - "text": line, - "title": title, - "pst": 1, - "contentmodel": "wikitext", - "utf8": 1, - "formatversion": 2, - } - # --- - # {"parse": {"title": "كريس فروم", "pageid": 2639244, "wikitext": "{{subst:user:Mr._Ibrahem/line2|Q76|P31}}", "psttext": "\"Q76\":{\n\"P31\":\"إنسان\"\n\n\n\n\n},"}} - # --- - data = self.client_request(params) - # --- - if not data: - return "" - # --- - textnew = data.get("parse", {}).get("psttext", "") - # --- - textnew = textnew.replace("\\n\\n", "") - # --- - return textnew - - def upload_by_file(self, file_name, text, file_path, comment="", ignorewarnings=False): - # --- - logger.info(f"<> def . {file_name=}") - # --- - if file_name.startswith("File:"): - file_name = file_name.replace("File:", "") - # --- - if file_name.startswith("ملف:"): - file_name = file_name.replace("ملف:", "") - # --- - logger.info(f"<> {file_path=}...") - # --- - params = { - "action": "upload", - "format": "json", - "filename": file_name, - "comment": comment, - "text": text, - "utf8": 1, - } - # --- - if ignorewarnings: - params["ignorewarnings"] = 1 - # --- - data = self.client_request(params, files={"file": open(file_path, "rb")}) - # --- - upload_result = data.get("upload", {}) - # --- - success = upload_result.get("result") == "Success" - _error = data.get("error", {}) - # --- - duplicate = upload_result.get("warnings", {}).get("duplicate", [""])[0].replace("_", " ") - # --- - if success: - logger.info(f"<> ** upload true .. [[File:{file_name}]] ") - return True - # --- - if duplicate: - logger.info(f"<> ** duplicate file: {duplicate}.") - # --- - return data - - def get_title_redirect_normalize(self, title, redirects, normalized): - # --- - redirects = redirects or [] - normalized = normalized or [] - # --- - tab = { - "user_input": title, - "redirect_to": "", - "normalized_to": "", - "real_title": title, - } - # --- - normalized = {x["to"]: x["from"] for x in normalized} - # --- - redirects = {x["to"]: x["from"] for x in redirects} - # --- - if tab["user_input"] in redirects: - tab["redirect_to"] = tab["user_input"] - tab["user_input"] = redirects[tab["user_input"]] - # --- - if tab["user_input"] in normalized: - tab["normalized_to"] = tab["user_input"] - tab["user_input"] = normalized[tab["user_input"]] - # --- - if tab["user_input"] == title: - return {} - # --- - return tab - - def merge_all_jsons_deep(self, all_jsons, json1): - def deep_merge(a, b): - # إذا كان كلاهما dict → دمج مفاتيح - if isinstance(a, dict) and isinstance(b, dict): - for k, v in b.items(): - if k in a: - a[k] = deep_merge(a[k], v) - else: - a[k] = v - return a - # إذا كان كلاهما list → تمديد القوائم - elif isinstance(a, list) and isinstance(b, list): - return a + b - # في حالة اختلاف النوع → نأخذ الجديد - else: - return b - - # إذا لم يكن all_jsons dict نجعله dict - if not isinstance(all_jsons, dict): - all_jsons = {} - - return deep_merge(all_jsons, json1) - - def merge_all_jsons(self, all_jsons, json1): - # --- إذا كان all_jsons ليس dict نحوله - if not isinstance(all_jsons, dict): - all_jsons = {} - # --- - # guard against non-dict inputs for json1 - if not isinstance(json1, dict): - return all_jsons - # --- - for x, z in json1.items(): - if x not in all_jsons: - all_jsons[x] = z - continue - # --- - tab = all_jsons[x] - # --- إذا كان كلاهما list - if isinstance(tab, list) and isinstance(z, list): - # explicit shallow copy of z to avoid surprises if z is reused - tab.extend(list(z)) - # --- إذا كان كلاهما dict - elif isinstance(tab, dict) and isinstance(z, dict): - tab.update(z) - # --- في حالة اختلاف النوع أو قيمة بسيطة - else: - all_jsons[x] = z - # --- - return all_jsons diff --git a/newapi/super/S_API/bot_api.py b/newapi/super/S_API/bot_api.py index 68f5e4e..6ca1095 100644 --- a/newapi/super/S_API/bot_api.py +++ b/newapi/super/S_API/bot_api.py @@ -10,12 +10,13 @@ from ...api_client import WikiLoginClient from ...client_wiki.api_utils.lang_codes import change_codes -from .bot import BotsAPIS +from ...client_wiki.api_utils.ask_bot import AskBot +from ...client_wiki.api_utils.handel_errors import HandleErrors logger = logging.getLogger(__name__) -class NewApi(BotsAPIS): +class NewApi(HandleErrors, AskBot): def __init__(self, login_bot: WikiLoginClient, lang: str = "", family: str = "wikipedia"): # --- self.login_bot = login_bot @@ -55,7 +56,7 @@ def Find_pages_exists_or_not(self, liste, get_redirect=False, noprint=False): # --- # if get_redirect: params["redirects"] = 1 # --- - json1 = self.client_request(params) + json1 = self.login_bot.client_request_safe(params) # --- if not json1: if not noprint: @@ -137,7 +138,7 @@ def Find_pages_exists_or_not_with_qids( if get_redirect: params["redirects"] = 1 # --- - json1 = self.client_request(params) + json1 = self.login_bot.client_request_safe(params) # --- if not json1: if not noprint: @@ -559,7 +560,7 @@ def Get_langlinks_for_list(self, titles, targtsitecode="", numbes=40): # --- # logger.debug(f'work for {len(group)} pages') # --- - json1 = self.client_request(params) + json1 = self.login_bot.client_request_safe(params) # --- if not json1: logger.info("bot_api. json1 is empty") @@ -607,7 +608,7 @@ def get_logs(self, title): "formatversion": 2, } # --- - data = self.client_request(params) + data = self.login_bot.client_request_safe(params) # --- if not data: return [] @@ -787,7 +788,7 @@ def Get_image_url(self, title): "formatversion": "2", } # --- - results = self.client_request(params) + results = self.login_bot.client_request_safe(params) # --- if not results: return "" @@ -819,7 +820,7 @@ def Get_imageinfo(self, title): "formatversion": "2", } # --- - results = self.client_request(params) + results = self.login_bot.client_request_safe(params) # --- if not results: return "" @@ -892,7 +893,7 @@ def get_cxtoken(self): # --- params = {"action": "cxtoken", "format": "json"} # --- - data = self.client_request(params, method="post") + data = self.login_bot.client_request_safe(params, method="post") # --- if not data: return "" @@ -956,14 +957,14 @@ def post_params( **kwargs, ): # --- - return self.login_bot.client_request( + return self.login_bot.client_request_safe( params, method=method, files=files, **kwargs, ) - def client_request( + def client_request_safe( self, params, method="get", @@ -971,7 +972,7 @@ def client_request( **kwargs, ): # --- - return self.login_bot.client_request( + return self.login_bot.client_request_safe( params, method=method, files=files, @@ -1001,3 +1002,364 @@ def post_continue( _p_2_empty=_p_2_empty, **kwargs, ) + + def Add_To_Bottom(self, text, summary, title, poss="Head|Bottom"): + # --- + if not title.strip(): + logger.info('** .. title == ""') + return False + # --- + if not text.strip(): + logger.info('** .. text == ""') + return False + # --- + logger.debug(f"** .. [[{title}]] ") + # --- + user = self.username + # --- + ask = self.ask_put( + newtext=text, + message=f"** Add_To {poss} .. [[{title}]] ", + job="Add_To_Bottom", + username=user, + summary=summary, + ) + # --- + if ask is False: + return False + # --- + params = { + "action": "edit", + "format": "json", + "title": title, + "summary": summary, + "notminor": 1, + "nocreate": 1, + "utf8": 1, + } + # --- + if poss == "Head": + params["prependtext"] = f"{text.strip()}\n" + else: + params["appendtext"] = f"\n{text.strip()}" + # --- + results = self.login_bot.client_request_safe(params) + # --- + if not results: + return "" + # --- + data = results.get("edit", {}) + result = data.get("result", "") + # --- + if result == "Success": + logger.info(f"<>** True. title:({title})") + return True + # --- + error = results.get("error", {}) + # --- + if error != {}: + print(results) + er = self.handle_err(error, function="Add_To_Bottom", params=params) + # --- + return er + # --- + return True + + def move( + self, + old_title, + to, + reason="", + noredirect=False, + movesubpages=False, + return_dict=False, + ): + # --- + logger.info(f"<> def [[{old_title}]] to [[{to}]] ") + # --- + params = { + "action": "move", + "format": "json", + "from": old_title, + "to": to, + "movetalk": 1, + "formatversion": 2, + } + # --- + if noredirect: + params["noredirect"] = 1 + if movesubpages: + params["movesubpages"] = 1 + # --- + if reason: + params["reason"] = reason + # --- + if old_title == to: + logger.debug(f"<>** old_title == to {to} ") + return {} + # --- + message = f"Do you want to move page:[[{old_title}]] to [[{to}]]?" + # --- + user = self.username + # --- + if not self.ask_put(message=message, job="move", username=user): + return {} + # --- + data = self.login_bot.client_request_safe(params) + # { "move": { "from": "d", "to": "d2", "reason": "wrong", "redirectcreated": true, "moveoverredirect": false } } + # --- + if not data: + logger.info("no data") + return {} + # --- + _expend_data = { + "move": { + "from": "User:Mr. Ibrahem", + "to": "User:Mr. Ibrahem/x", + "reason": "wrong title", + "redirectcreated": True, + "moveoverredirect": False, + "talkmove-errors": [ + { + "message": "content-not-allowed-here", + "params": [ + "Structured Discussions board", + "User talk:Mr. Ibrahem/x", + "main", + ], + "code": "contentnotallowedhere", + "type": "error", + }, + { + "message": "flow-error-allowcreation-flow-create-board", + "params": [], + "code": "flow-error-allowcreation-flow-create-board", + "type": "error", + }, + ], + "subpages": { + "errors": [ + { + "message": "cant-move-subpages", + "params": [], + "code": "cant-move-subpages", + "type": "error", + } + ] + }, + "subpages-talk": { + "errors": [ + { + "message": "cant-move-subpages", + "params": [], + "code": "cant-move-subpages", + "type": "error", + } + ] + }, + } + } + # --- + move_done = data.get("move", {}) + error = data.get("error", {}) + error_code = error.get("code", "") # missingtitle + # --- + # elif "Please choose another name." in r4: + # --- + if move_done: + logger.info("<>** true.") + # --- + if return_dict: + return move_done + # --- + return True + # --- + if error: + if error_code == "ratelimited": + logger.info("<> ratelimited:") + return self.move( + old_title, + to, + reason=reason, + noredirect=noredirect, + movesubpages=movesubpages, + return_dict=return_dict, + ) + + if error_code == "articleexists": + logger.info("<> articleexists") + return "articleexists" + + logger.info("<> error") + logger.info(error) + + return {} + # --- + return {} + + def expandtemplates(self, text): + # --- + params = { + "action": "expandtemplates", + "format": "json", + "text": text, + "prop": "wikitext", + "formatversion": 2, + } + # --- + data = self.login_bot.client_request_safe(params) + # --- + if not data: + return text + # --- + newtext = data.get("expandtemplates", {}).get("wikitext") or text + # --- + return newtext + + def Parse_Text(self, line, title): + # --- + params = { + "action": "parse", + "prop": "wikitext", + "text": line, + "title": title, + "pst": 1, + "contentmodel": "wikitext", + "utf8": 1, + "formatversion": 2, + } + # --- + # {"parse": {"title": "كريس فروم", "pageid": 2639244, "wikitext": "{{subst:user:Mr._Ibrahem/line2|Q76|P31}}", "psttext": "\"Q76\":{\n\"P31\":\"إنسان\"\n\n\n\n\n},"}} + # --- + data = self.login_bot.client_request_safe(params) + # --- + if not data: + return "" + # --- + textnew = data.get("parse", {}).get("psttext", "") + # --- + textnew = textnew.replace("\\n\\n", "") + # --- + return textnew + + def upload_by_file(self, file_name, text, file_path, comment="", ignorewarnings=False): + # --- + logger.info(f"<> def . {file_name=}") + # --- + if file_name.startswith("File:"): + file_name = file_name.replace("File:", "") + # --- + if file_name.startswith("ملف:"): + file_name = file_name.replace("ملف:", "") + # --- + logger.info(f"<> {file_path=}...") + # --- + params = { + "action": "upload", + "format": "json", + "filename": file_name, + "comment": comment, + "text": text, + "utf8": 1, + } + # --- + if ignorewarnings: + params["ignorewarnings"] = 1 + # --- + data = self.login_bot.client_request_safe(params, files={"file": open(file_path, "rb")}) + # --- + upload_result = data.get("upload", {}) + # --- + success = upload_result.get("result") == "Success" + _error = data.get("error", {}) + # --- + duplicate = upload_result.get("warnings", {}).get("duplicate", [""])[0].replace("_", " ") + # --- + if success: + logger.info(f"<> ** upload true .. [[File:{file_name}]] ") + return True + # --- + if duplicate: + logger.info(f"<> ** duplicate file: {duplicate}.") + # --- + return data + + def get_title_redirect_normalize(self, title, redirects, normalized): + # --- + redirects = redirects or [] + normalized = normalized or [] + # --- + tab = { + "user_input": title, + "redirect_to": "", + "normalized_to": "", + "real_title": title, + } + # --- + normalized = {x["to"]: x["from"] for x in normalized} + # --- + redirects = {x["to"]: x["from"] for x in redirects} + # --- + if tab["user_input"] in redirects: + tab["redirect_to"] = tab["user_input"] + tab["user_input"] = redirects[tab["user_input"]] + # --- + if tab["user_input"] in normalized: + tab["normalized_to"] = tab["user_input"] + tab["user_input"] = normalized[tab["user_input"]] + # --- + if tab["user_input"] == title: + return {} + # --- + return tab + + def merge_all_jsons_deep(self, all_jsons, json1): + def deep_merge(a, b): + # إذا كان كلاهما dict → دمج مفاتيح + if isinstance(a, dict) and isinstance(b, dict): + for k, v in b.items(): + if k in a: + a[k] = deep_merge(a[k], v) + else: + a[k] = v + return a + # إذا كان كلاهما list → تمديد القوائم + elif isinstance(a, list) and isinstance(b, list): + return a + b + # في حالة اختلاف النوع → نأخذ الجديد + else: + return b + + # إذا لم يكن all_jsons dict نجعله dict + if not isinstance(all_jsons, dict): + all_jsons = {} + + return deep_merge(all_jsons, json1) + + def merge_all_jsons(self, all_jsons, json1): + # --- إذا كان all_jsons ليس dict نحوله + if not isinstance(all_jsons, dict): + all_jsons = {} + # --- + # guard against non-dict inputs for json1 + if not isinstance(json1, dict): + return all_jsons + # --- + for x, z in json1.items(): + if x not in all_jsons: + all_jsons[x] = z + continue + # --- + tab = all_jsons[x] + # --- إذا كان كلاهما list + if isinstance(tab, list) and isinstance(z, list): + # explicit shallow copy of z to avoid surprises if z is reused + tab.extend(list(z)) + # --- إذا كان كلاهما dict + elif isinstance(tab, dict) and isinstance(z, dict): + tab.update(z) + # --- في حالة اختلاف النوع أو قيمة بسيطة + else: + all_jsons[x] = z + # --- + return all_jsons From 203eafbeba5be849eec08f24af9d3ced2eb7e93e Mon Sep 17 00:00:00 2001 From: Ibrahem Date: Tue, 12 May 2026 06:01:13 +0300 Subject: [PATCH 8/9] . --- newapi/super/S_API/bot_api.py | 2 +- tests/TestALL_APIS.py | 9 +++++++-- tests/TestAuthentication.py | 1 + tests/TestMainPage.py | 10 ++++++---- 4 files changed, 15 insertions(+), 7 deletions(-) diff --git a/newapi/super/S_API/bot_api.py b/newapi/super/S_API/bot_api.py index 6ca1095..c450703 100644 --- a/newapi/super/S_API/bot_api.py +++ b/newapi/super/S_API/bot_api.py @@ -9,9 +9,9 @@ import tqdm from ...api_client import WikiLoginClient -from ...client_wiki.api_utils.lang_codes import change_codes from ...client_wiki.api_utils.ask_bot import AskBot from ...client_wiki.api_utils.handel_errors import HandleErrors +from ...client_wiki.api_utils.lang_codes import change_codes logger = logging.getLogger(__name__) diff --git a/tests/TestALL_APIS.py b/tests/TestALL_APIS.py index fbd9d26..00764e3 100644 --- a/tests/TestALL_APIS.py +++ b/tests/TestALL_APIS.py @@ -26,7 +26,8 @@ def mock_dependencies(): def test_all_apis_init(mock_dependencies): lang, family, username, password = "en", "wikipedia", "user", "pass" - api = AllAPIS(lang, family, username, password) + use_cookies = True + api = AllAPIS(lang, family, username, password, use_cookies) assert api.lang == lang assert api.family == family @@ -34,7 +35,11 @@ def test_all_apis_init(mock_dependencies): assert api.password == password mock_dependencies["WikiLoginClient"].assert_called_once_with( - lang=lang, family=family, username=username, password=password + lang=lang, + family=family, + username=username, + password=password, + use_cookies=use_cookies, ) diff --git a/tests/TestAuthentication.py b/tests/TestAuthentication.py index 7b6ae75..6be92e4 100644 --- a/tests/TestAuthentication.py +++ b/tests/TestAuthentication.py @@ -17,6 +17,7 @@ def test_successful_login(self, mock_login_client: WikiLoginClient): """Test successful authentication""" params = {"action": "query", "titles": "Main Page", "format": "json"} mock_login_client.client_request.return_value = {"query": {"pages": {"1": {"title": "Main Page"}}}} + mock_login_client.client_request_safe.return_value = {"query": {"pages": {"1": {"title": "Main Page"}}}} response = mock_login_client.client_request(params, method="post") assert response is not None assert len(response) > 0 diff --git a/tests/TestMainPage.py b/tests/TestMainPage.py index f06e3c6..2dbace0 100644 --- a/tests/TestMainPage.py +++ b/tests/TestMainPage.py @@ -8,7 +8,7 @@ class TestMainPage: @pytest.fixture def mock_login_bot(self): bot = MagicMock() - bot.client_request.return_value = { + result = { "query": { "pages": { "123": { @@ -20,6 +20,8 @@ def mock_login_bot(self): } } } + bot.client_request_safe.return_value = result + bot.client_request_safe.return_value = result return bot @pytest.fixture @@ -29,7 +31,7 @@ def test_page(self, mock_login_bot): @pytest.fixture def arabic_page(self, mock_login_bot): mock_bot = MagicMock() - mock_bot.client_request.return_value = { + mock_bot.client_request_safe.return_value = { "query": { "pages": { "456": { @@ -60,7 +62,7 @@ def test_get_text(self, arabic_page): def test_nonexistent_page(self, mock_login_bot): """Test behavior with non-existent page""" - mock_login_bot.client_request.return_value = { + mock_login_bot.client_request_safe.return_value = { "query": {"pages": {"-1": {"title": "NonExistentPage12345", "missing": ""}}} } page = MainPage(mock_login_bot, "NonExistentPage12345", "en") @@ -72,7 +74,7 @@ def test_empty_page_content(self): def test_page_without_edit_permission(self, mock_login_bot): """Test page where user cannot edit""" - mock_login_bot.client_request.return_value = { + mock_login_bot.client_request_safe.return_value = { "query": { "pages": { "789": { From f7f89cedc0168ff81e84c874b73c13026d6be77e Mon Sep 17 00:00:00 2001 From: Ibrahem Date: Tue, 12 May 2026 06:01:42 +0300 Subject: [PATCH 9/9] Update TestAuthentication.py --- tests/TestAuthentication.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/TestAuthentication.py b/tests/TestAuthentication.py index 6be92e4..7b6ae75 100644 --- a/tests/TestAuthentication.py +++ b/tests/TestAuthentication.py @@ -17,7 +17,6 @@ def test_successful_login(self, mock_login_client: WikiLoginClient): """Test successful authentication""" params = {"action": "query", "titles": "Main Page", "format": "json"} mock_login_client.client_request.return_value = {"query": {"pages": {"1": {"title": "Main Page"}}}} - mock_login_client.client_request_safe.return_value = {"query": {"pages": {"1": {"title": "Main Page"}}}} response = mock_login_client.client_request(params, method="post") assert response is not None assert len(response) > 0