diff --git a/newapi/__init__.py b/newapi/__init__.py index 7437658..aa256e2 100644 --- a/newapi/__init__.py +++ b/newapi/__init__.py @@ -1,11 +1,10 @@ """ """ -from .api_utils.lang_codes import change_codes - from . import page from .all_apis import AllAPIS from .api_client.client import WikiLoginClient -from .api_utils import botEdit, txtlib, wd_sparql +from .client_wiki.api_utils import botEdit, txtlib, wd_sparql +from .client_wiki.api_utils.lang_codes import change_codes from .DB_bots import db_bot, pymysql_bot __all__ = [ diff --git a/newapi/all_apis.py b/newapi/all_apis.py index cbf2fb3..ae12a02 100644 --- a/newapi/all_apis.py +++ b/newapi/all_apis.py @@ -1,6 +1,6 @@ """ """ -from .pages_bots.all_apis import ( +from .client_wiki.all_apis import ( AllAPIS, ) diff --git a/newapi/api_client/client.py b/newapi/api_client/client.py index 32f227f..62ec259 100644 --- a/newapi/api_client/client.py +++ b/newapi/api_client/client.py @@ -485,29 +485,16 @@ def site(self) -> mwclient.Site: """The underlying ``mwclient.Site`` — use for high-level wiki access.""" return self._site - # ── Public methods ───────────────────────────────────────────────────── - - def login(self, force: bool = False) -> None: - """ - Force a fresh login regardless of cookie state. - - Call this if you know the session has expired and want to re-authenticate - without creating a new WikiLoginClient instance. - """ - if force or not self._site.logged_in: - logger.info( - "Forcing re-login for %s on %s.%s", - self.username, - self.lang, - self.family, - ) - self._do_login() + # ------------------------------------------------------------------ + # Private helpers + # ------------------------------------------------------------------ def _client_request( self, params: dict, method: str = "post", files: Optional[Any] = None, + **kwargs, ) -> dict: """ Send a GET or POST request to the wiki API and return parsed JSON. @@ -553,9 +540,10 @@ def _client_request( action = params.get("action") if action in self._WRITE_ACTIONS: method = "post" + if method == "get": return self._request_with_retry("GET", self.api_url, params=params) - #return self._site.get(action, **params) + # return self._site.get(action, **params) else: # Fetch a CSRF token now if the caller didn't supply one. # The retry loop will refresh it automatically on CSRF errors. @@ -563,19 +551,108 @@ def _client_request( params["token"] = self._site.get_token("csrf") return self._request_with_retry("POST", self.api_url, data=params, files=files) - #return self._site.post(action, **params, files=files) + # return self._site.post(action, **params, files=files) + + def _ensure_logged_in(self) -> None: + """ + Check whether the current session is authenticated. + """ + # if self._site.logged_in: + 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 not self._site.logged_in: self._do_login() + # don't login yet, user can use login() method + + def _enrich_params(self, params: dict) -> dict: + """ + Inject write-action safety parameters. + + For write actions: + - ``bot=1`` marks edits as bot edits in recent changes. + - ``assertuser`` ensures the API rejects requests from the wrong + account (guards against accidental edits). + + Query actions have write-only keys scrubbed instead. + """ + params = dict(params) + action = params.get("action", "") + + # Strip write-only params from query actions + if action == "query": + params.pop("bot", None) + params.pop("summary", None) + return params + + # Inject bot marker and identity assertion for all write actions + is_write = action in self._WRITE_ACTIONS or action.startswith("wb") or self.family == "wikidata" + if is_write and self.username: + params.setdefault("bot", 1) + params.setdefault("assertuser", self.username) + + return params + + def _do_login(self) -> None: + """ + Execute the mwclient login handshake and persist the resulting cookies. + + Raises: + LoginError: if mwclient rejects the credentials. + """ + 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 + + if self._site.logged_in: + logger.info( + "Logged in successfully as %s on %s.%s", + self.username, + self.lang, + self.family, + ) + self.save_cookies(self.cj) + + # ── Public methods ───────────────────────────────────────────────────── + + def login(self, force: bool = False) -> None: + """ + Force a fresh login regardless of cookie state. + + Call this if you know the session has expired and want to re-authenticate + without creating a new WikiLoginClient instance. + """ + if force or not self._site.logged_in: + logger.info( + "Forcing re-login for %s on %s.%s", + self.username, + self.lang, + self.family, + ) + self._do_login() def client_request( self, params: dict, method: str = "post", files: Optional[Any] = None, + **kwargs, ) -> dict: """ """ return self._client_request( params=params, method=method, files=files, + **kwargs, ) def client_request_safe( @@ -583,6 +660,7 @@ def client_request_safe( params: dict, method: str = "post", files: Optional[Any] = None, + **kwargs, ) -> dict: """ """ try: @@ -590,6 +668,7 @@ def client_request_safe( params=params, method=method, files=files, + **kwargs, ) except Exception as exc: logger.warning("client_request_safe: %s", exc) @@ -600,6 +679,7 @@ def client_request_retry( params: dict, method: str = "post", files: Optional[Any] = None, + **kwargs, ) -> dict: """ Send a GET or POST request to the wiki API and return parsed JSON. @@ -628,18 +708,25 @@ def client_request_retry( raise ValueError(f"method must be 'get' or 'post', got {method!r}") # Files can only travel via multipart POST - if files is not None: + action = params.get("action") + if action in self._WRITE_ACTIONS or files is not None: method = "post" # Always request JSON and inject write-action safety params params = self._enrich_params({"format": "json", **params}) + skip_log_params = [ + "token", + "password", + "lgpassword", + "text", + ] logger.debug( "%s %s params=%s files=%s", method.upper(), self.api_url, # Never log token values - {k: ("***" if k == "token" else v) for k, v in params.items()}, + {k: ("***" if k in skip_log_params else v) for k, v in params.items()}, list(files.keys()) if files else None, ) @@ -672,6 +759,7 @@ def post_continue( first: bool = False, _p_2: str = "", _p_2_empty: Optional[Union[list, dict]] = None, + **kwargs, ) -> Union[list, dict]: """ Drive a MediaWiki API continuation query to completion. @@ -755,79 +843,6 @@ def post_continue( logger.debug("done, %d total results", len(results)) return results - # ------------------------------------------------------------------ - # Private helpers - # ------------------------------------------------------------------ - - def _ensure_logged_in(self) -> None: - """ - Check whether the current session is authenticated. - """ - # if self._site.logged_in: - 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 not self._site.logged_in: self._do_login() - # don't login yet, user can use login() method - - def _enrich_params(self, params: dict) -> dict: - """ - Inject write-action safety parameters. - - For write actions: - - ``bot=1`` marks edits as bot edits in recent changes. - - ``assertuser`` ensures the API rejects requests from the wrong - account (guards against accidental edits). - - Query actions have write-only keys scrubbed instead. - """ - params = dict(params) - action = params.get("action", "") - - # Strip write-only params from query actions - if action == "query": - params.pop("bot", None) - params.pop("summary", None) - return params - - # Inject bot marker and identity assertion for all write actions - is_write = action in self._WRITE_ACTIONS or action.startswith("wb") or self.family == "wikidata" - if is_write and self.username: - params.setdefault("bot", 1) - params.setdefault("assertuser", self.username) - - return params - - def _do_login(self) -> None: - """ - Execute the mwclient login handshake and persist the resulting cookies. - - Raises: - LoginError: if mwclient rejects the credentials. - """ - 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 - - if self._site.logged_in: - logger.info( - "Logged in successfully as %s on %s.%s", - self.username, - self.lang, - self.family, - ) - self.save_cookies(self.cj) - def __repr__(self) -> str: return f"WikiLoginClient(lang={self.lang!r}, family={self.family!r}, username={self.username!r})" diff --git a/newapi/api_utils/__init__.py b/newapi/client_wiki/__init__.py similarity index 100% rename from newapi/api_utils/__init__.py rename to newapi/client_wiki/__init__.py diff --git a/newapi/pages_bots/all_apis.py b/newapi/client_wiki/all_apis.py similarity index 80% rename from newapi/pages_bots/all_apis.py rename to newapi/client_wiki/all_apis.py index 9fcc85a..9f2b46f 100644 --- a/newapi/pages_bots/all_apis.py +++ b/newapi/client_wiki/all_apis.py @@ -1,17 +1,11 @@ -""" - -main_api = AllAPIS(lang='en', family='wikipedia', username='your_username', password='your_password') -page = main_api.MainPage('Main Page Title') -cat_members = main_api.CatDepth('Category Title') -new_api = main_api.NewApi() -""" +""" """ import logging -from ..api_client.client import WikiLoginClient +from ..api_client import WikiLoginClient from ..super.S_API import bot_api -from ..super.S_Category import catdepth_new -from ..super.S_Page import super_page +from .categories import catdepth_new +from .pages import super_page logger = logging.getLogger(__name__) diff --git a/newapi/client_wiki/api_utils/__init__.py b/newapi/client_wiki/api_utils/__init__.py new file mode 100644 index 0000000..65b835e --- /dev/null +++ b/newapi/client_wiki/api_utils/__init__.py @@ -0,0 +1,30 @@ +from .ask_bot import AskBot, showDiff +from .botEdit import ( + bot_May_Edit, + check_create_time, + check_last_edit_time, +) + +change_codes = { + "bat_smg": "bat-smg", + "be-x-old": "be-tarask", + "be_x_old": "be-tarask", + "cbk_zam": "cbk-zam", + "fiu_vro": "fiu-vro", + "map_bms": "map-bms", + "nb": "no", + "nds_nl": "nds-nl", + "roa_rup": "roa-rup", + "zh_classical": "zh-classical", + "zh_min_nan": "zh-min-nan", + "zh_yue": "zh-yue", +} + +__all__ = [ + "AskBot", + "change_codes", + "showDiff", + "bot_May_Edit", + "check_create_time", + "check_last_edit_time", +] diff --git a/newapi/api_utils/ask_bot.py b/newapi/client_wiki/api_utils/ask_bot.py similarity index 98% rename from newapi/api_utils/ask_bot.py rename to newapi/client_wiki/api_utils/ask_bot.py index 727f2f7..333c46b 100644 --- a/newapi/api_utils/ask_bot.py +++ b/newapi/client_wiki/api_utils/ask_bot.py @@ -5,9 +5,10 @@ import pywikibot -from ..config import settings +from ...config import settings logger = logging.getLogger(__name__) + _save_or_ask: dict[str, bool] = {} diff --git a/newapi/api_utils/botEdit.py b/newapi/client_wiki/api_utils/botEdit.py similarity index 86% rename from newapi/api_utils/botEdit.py rename to newapi/client_wiki/api_utils/botEdit.py index 9fc1915..16ab34d 100644 --- a/newapi/api_utils/botEdit.py +++ b/newapi/client_wiki/api_utils/botEdit.py @@ -1,12 +1,11 @@ -""" -from newapi import botEdit -bot_edit! -""" +""" """ + +import logging from .bot_edit.bot_edit_by_templates import is_bot_edit_allowed from .bot_edit.bot_edit_by_time import check_create_time, check_last_edit_time -Created_Cache = {} +logger = logging.getLogger(__name__) def bot_May_Edit( @@ -16,7 +15,6 @@ def bot_May_Edit( page=False, delay: int = 0, ) -> bool: - # --- """ Determines whether a bot is permitted to edit a page based on templates, last edit time, and creation time. @@ -33,29 +31,24 @@ def bot_May_Edit( True if the bot is allowed to edit the page; False otherwise. """ check_it = is_bot_edit_allowed(text=text, title_page=title_page, botjob=botjob) - # --- + if page and check_it: - # --- if delay and isinstance(delay, int): - # --- ns = page.namespace() lang = page.lang - # --- + if ns != 0 or lang != "ar": return check_it - # --- + check_time = check_last_edit_time(page, title_page, delay) - # --- if not check_time: return False - # --- + check_create = check_create_time(page, title_page) - # --- - Created_Cache[title_page] = check_create - # --- + if not check_create: return False - # --- + return check_it @@ -65,4 +58,6 @@ def botMayEdit(**kwargs): __all__ = [ "bot_May_Edit", + "check_create_time", + "check_last_edit_time", ] diff --git a/newapi/api_utils/bot_edit/__init__.py b/newapi/client_wiki/api_utils/bot_edit/__init__.py similarity index 100% rename from newapi/api_utils/bot_edit/__init__.py rename to newapi/client_wiki/api_utils/bot_edit/__init__.py diff --git a/newapi/api_utils/bot_edit/bot_edit_by_templates.py b/newapi/client_wiki/api_utils/bot_edit/bot_edit_by_templates.py similarity index 72% rename from newapi/api_utils/bot_edit/bot_edit_by_templates.py rename to newapi/client_wiki/api_utils/bot_edit/bot_edit_by_templates.py index 85a5d2c..156f50f 100644 --- a/newapi/api_utils/bot_edit/bot_edit_by_templates.py +++ b/newapi/client_wiki/api_utils/bot_edit/bot_edit_by_templates.py @@ -1,16 +1,14 @@ """ """ import logging -import sys import wikitextparser as wtp -from ...config import settings +from ....config import settings logger = logging.getLogger(__name__) -edit_username = {1: "Mr.Ibrahembot"} -Bot_Cache = {} -stop_edit_temps = { + +STOP_EDIT_TEMPLATES: dict[str, list[str]] = { "all": ["تحرر", "قيد التطوير", "يحرر", "تطوير مقالة"], "تعريب": ["لا للتعريب"], "تقييم آلي": ["لا للتقييم الآلي"], @@ -22,70 +20,76 @@ "portal": ["لا لربط البوابات المعادل", "لا لصيانة البوابات"], } +BOT_USERNAME = "Mr.Ibrahembot" +Bot_Cache = {} + def _handle_nobots_template(params, title_page, botjob, _template): """Handle nobots template logic.""" - # --- # {{nobots}} منع جميع البوتات # منع جميع البوتات if not params: - logger.info(f"<> botEdit.py: the page has template:({_template}), botjob:{botjob} skipp.") + logger.debug(f"<> botEdit.py: the page has temp:({_template}), botjob:{botjob} skipp.") + logger.debug(f"nobots active - blocking bot {botjob} on {title_page}") Bot_Cache[botjob][title_page] = False return False elif params.get("1"): List = [x.strip() for x in params.get("1", "").split(",")] - # if 'all' in List or pywikibot.calledModuleName() in List or edit_username[1] in List: - if "all" in List or edit_username[1] in List: - logger.info(f"<> botEdit.py: the page has template:({_template}), botjob:{botjob} skipp.") + # if 'all' in List or pywikibot.calledModuleName() in List or BOT_USERNAME in List: + if "all" in List or BOT_USERNAME in List: + logger.debug(f"<> botEdit.py: the page has temp:({_template}), botjob:{botjob} skipp.") + logger.debug(f"bot {BOT_USERNAME} in nobots list - blocking {title_page}") + # Bot_Cache[title_page] = False Bot_Cache[botjob][title_page] = False return False - # --- + # no restricting template found Bot_Cache[botjob][title_page] = True - # --- return True def _handle_bots_template(params, title_page, botjob, title): """Handle bots template logic.""" - # --- + logger.debug(f"handling bots template for {title}") # {{bots}} السماح لجميع البوتات if not params: Bot_Cache[botjob][title_page] = False return False else: - logger.info(f"botEdit.py title:({title}), params:({str(params)}).") - # --- + logger.debug(f"botEdit.py title:({title}), params:({str(params)}).") + # for param in params: + # value = params[param] + # value = [ x.strip() for x in value.split(',') ] # {{bots|allow=all}} السماح لجميع البوتات # {{bots|allow=none}} منع جميع البوتات allow = params.get("allow") if allow: value = [x.strip() for x in allow.split(",")] - sd = "all" in value or edit_username[1] in value + # if param == 'allow': + # 'all' in value or BOT_USERNAME in value is True + sd = "all" in value or BOT_USERNAME in value if not sd: - logger.info(f"<>botEdit.py Template:({title}) has |allow={','.join(value)}.") + logger.debug(f"<>botEdit.py Template:({title}) has |allow={','.join(value)}.") else: - logger.info(f"<>botEdit.py Template:({title}) has |allow={','.join(value)}.") + logger.warning(f"<>botEdit.py Template:({title}) has |allow={','.join(value)}.") Bot_Cache[botjob][title_page] = sd return sd - # --- - # --- # {{bots|deny=all}} منع جميع البوتات deny = params.get("deny") if deny: value = [x.strip() for x in deny.split(",")] - sd = "all" not in value and edit_username[1] not in value + # {{bots|deny=all}} + # if param == 'deny': + sd = "all" not in value and BOT_USERNAME not in value if not sd: - logger.info(f"<>botEdit.py Template:({title}) has |deny={','.join(value)}.") + logger.debug(f"<>botEdit.py Template:({title}) has |deny={','.join(value)}.") Bot_Cache[botjob][title_page] = sd return sd - # --- # if param == 'allowscript': # return ('all' in value or pywikibot.calledModuleName() in value) # if param == 'denyscript': # return not ('all' in value or pywikibot.calledModuleName() in value) - # --- + # no restricting template found Bot_Cache[botjob][title_page] = True - # --- return True @@ -94,7 +98,6 @@ def is_bot_edit_allowed( title_page: str = "", botjob: str = "all", ) -> bool: - # --- """ Determines if a bot is permitted to edit a page based on templates in the page text. @@ -110,56 +113,56 @@ def is_bot_edit_allowed( """ if (settings.bot.force_edit) or settings.bot.workibrahem: return True - # --- + if botjob in ["", "fixref|cat|stub|tempcat|portal"]: botjob = "all" - # --- + if botjob not in Bot_Cache: Bot_Cache[botjob] = {} - # --- + if title_page in Bot_Cache[botjob]: return Bot_Cache[botjob][title_page] - # --- - all_stop = stop_edit_temps["all"] - # --- + + all_stop = STOP_EDIT_TEMPLATES["all"] + parser = wtp.parse(text) templates = parser.templates - # --- + for template in templates: title = str(template.normal_name()).strip() - # --- + params = { str(param.name).strip(): str(param.value).strip() for param in template.arguments if str(param.value).strip() } - # --- + _template = template.string - # --- - restrictions = stop_edit_temps.get(botjob, []) - # --- + + restrictions = STOP_EDIT_TEMPLATES.get(botjob, []) + if title in restrictions or title in all_stop: - logger.info(f"<> botEdit.py: the page has template:({title}), botjob:{botjob} skipp.") + logger.debug(f"<> botEdit.py: the page has temp:({title}), botjob:{botjob} skipp.") Bot_Cache[botjob][title_page] = False return False - # --- + # logger.debug("<>botEdit.py title:(%s), params:(%s)." % (title, str(params))) - # --- + if title.lower() == "nobots": return _handle_nobots_template(params, title_page, botjob, _template) - # --- + # {{bots|allow=}} منع جميع البوتات غير الموجودة في القائمة # {{bots|deny=}} منع جميع البوتات الموجودة في القائمة - # --- + elif title.lower() == "bots": return _handle_bots_template(params, title_page, botjob, title) - # --- + # no restricting template found Bot_Cache[botjob][title_page] = True - # --- return True __all__ = [ "is_bot_edit_allowed", + "BOT_USERNAME", ] diff --git a/newapi/api_utils/bot_edit/bot_edit_by_time.py b/newapi/client_wiki/api_utils/bot_edit/bot_edit_by_time.py similarity index 75% rename from newapi/api_utils/bot_edit/bot_edit_by_time.py rename to newapi/client_wiki/api_utils/bot_edit/bot_edit_by_time.py index d8a613e..170fc11 100644 --- a/newapi/api_utils/bot_edit/bot_edit_by_time.py +++ b/newapi/client_wiki/api_utils/bot_edit/bot_edit_by_time.py @@ -3,55 +3,53 @@ import datetime import logging -Created_Cache = {} +_created_cache = {} logger = logging.getLogger(__name__) def check_create_time(page, title_page): - # --- """ Checks if a page was created at least three hours ago before allowing bot edits. Returns True if the page is not in the Arabic main namespace or if the creation timestamp is missing. Returns False if the page was created less than three hours ago, caching the result for future checks. """ - # --- - if title_page in Created_Cache: - return Created_Cache[title_page] - # --- + + if title_page in _created_cache: + return _created_cache[title_page] + ns = page.namespace() lang = page.lang - # --- + if ns != 0 or lang != "ar": return True - # --- - now = datetime.datetime.now(datetime.timezone.utc) - # --- + + now = datetime.datetime.now(datetime.UTC) + create_data = page.get_create_data() # { "timestamp" : "2025-05-07T12:00:17Z", "user" : "", "anon" : "" } - # --- + delay_hours = 3 - # --- + if create_data.get("timestamp"): - # --- + create_time = create_data["timestamp"] - ts_c_time = datetime.datetime.strptime(create_time, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=datetime.timezone.utc) - # --- + ts_c_time = datetime.datetime.strptime(create_time, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=datetime.UTC) + diff = (now - ts_c_time).total_seconds() / (60 * 60) - # --- + user = create_data.get("user", "") - # --- + wait_time = delay_hours - diff - # --- + if diff < delay_hours: - logger.info(f"<>Page:{title_page} create at ({create_time}).") - logger.info(f"<>Page Created before {diff:.2f} hours by: {user}, wait {wait_time:.2f}H.") + logger.debug(f"<>Page:{title_page} create at ({create_time}).") + logger.debug(f"<>Page Created before {diff:.2f} hours by: {user}, wait {wait_time:.2f}H.") return False - # --- + return True def check_last_edit_time(page, title_page, delay): - # --- """ Checks if enough time has passed since the last non-bot edit before allowing a bot to edit. @@ -63,31 +61,31 @@ def check_last_edit_time(page, title_page, delay): delay: Minimum number of minutes that must have passed since the last edit. """ userinfo = page.get_userinfo() - # --- + if "bot" in userinfo.get("groups", []): return True - # --- + # example: 2025-05-07T12:00:17Z timestamp = page.get_timestamp() - # --- - now = datetime.datetime.now(datetime.timezone.utc) - # --- + + now = datetime.datetime.now(datetime.UTC) + if timestamp: - ts_time = datetime.datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=datetime.timezone.utc) - # --- + ts_time = datetime.datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=datetime.UTC) + diff_minutes = (now - ts_time).total_seconds() / 60 - # --- + # logger.info(f"<> last-edit Δ={diff_minutes:.2f} min for {title_page}") - # --- + wait_time = delay - diff_minutes - # --- + if diff_minutes < delay: logger.info(f"<>Page:{title_page} last edit ({timestamp}).") logger.info( f"<>Page Last edit before {delay} minutes, Wait {wait_time:.2f} minutes. title:{title_page}" ) return False - # --- + return True diff --git a/newapi/super/handel_errors.py b/newapi/client_wiki/api_utils/handel_errors.py similarity index 80% rename from newapi/super/handel_errors.py rename to newapi/client_wiki/api_utils/handel_errors.py index 3547809..6b14b94 100644 --- a/newapi/super/handel_errors.py +++ b/newapi/client_wiki/api_utils/handel_errors.py @@ -1,13 +1,5 @@ """ -from .super.handel_errors import HandelErrors - -""" - -import logging -import sys -from typing import Any, Dict, Optional, Union - -from ..core.exceptions import ( +from ...core.exceptions import ( AbuseFilterError, ApiError, ArticleExistsError, @@ -16,11 +8,14 @@ ProtectedPageError, parse_api_error, ) +""" + +import logging logger = logging.getLogger(__name__) -class HandelErrors: +class HandleErrors: """ Error handler for MediaWiki API errors. @@ -36,13 +31,13 @@ def __init__(self) -> None: config: Optional BotConfig for behavior settings. """ - def handel_err( + def handle_err( self, - error: Dict[str, Any], + error: dict, function: str = "", - params: Optional[Dict[str, Any]] = None, + params: dict | None = None, do_error: bool = True, - ) -> Union[str, bool, ApiError]: + ) -> dict | str | bool: """Handle errors based on the provided error dictionary. This function processes an error dictionary and performs actions based @@ -76,14 +71,14 @@ def handel_err( err_info = error.get("info", "") _tt = f"<>{function} ERROR: <>code:{err_code}." - ["protectedpage", "تأخير البوتات 3 ساعات", False] + # ["protectedpage", "تأخير البوتات 3 ساعات", False] if err_code == "abusefilter-disallowed": # oioioi = {'error': {'code': 'abusefilter-disallowed', 'info': 'This', 'abusefilter': {'id': '169', 'description': 'تأخير البوتات 3 ساعات', 'actions': ['disallow']}, '*': 'See https'}, 'servedby': 'mw1374'} abusefilter = error.get("abusefilter", "") - description = abusefilter.get("description", "") - logger.info(f"<> ** abusefilter-disallowed: {description} ") + description = abusefilter.get("description", "") if isinstance(abusefilter, dict) else "" + logger.debug(f"<> ** abusefilter-disallowed: {description} ") if description in [ "تأخير البوتات 3 ساعات", "تأخير البوتات 3 ساعات- 3 من 3", @@ -94,23 +89,29 @@ def handel_err( return description if err_code == "no-such-entity": - logger.info("<> ** no-such-entity. ") + logger.debug("<> ** no-such-entity. ") return False if err_code == "protectedpage": - logger.info("<> ** protectedpage. ") - # return "protectedpage" + logger.debug("<> ** protectedpage. ") return False if err_code == "articleexists": - logger.info("<> ** article already created. ") + logger.debug("<> ** article already created. ") return "articleexists" if err_code == "maxlag": - logger.info("<> ** maxlag. ") + logger.debug("<> ** maxlag. ") return False if do_error: - params["data"] = {} - params["text"] = {} + if params: + params["data"] = {} + params["text"] = {} logger.error(f"<>{function} ERROR: <>info: {err_info}, {params=}") + return error + + +__all__ = [ + "HandleErrors", +] diff --git a/newapi/api_utils/lang_codes.py b/newapi/client_wiki/api_utils/lang_codes.py similarity index 100% rename from newapi/api_utils/lang_codes.py rename to newapi/client_wiki/api_utils/lang_codes.py diff --git a/newapi/api_utils/printe.py b/newapi/client_wiki/api_utils/printe.py similarity index 96% rename from newapi/api_utils/printe.py rename to newapi/client_wiki/api_utils/printe.py index 8129bb1..02dcdad 100644 --- a/newapi/api_utils/printe.py +++ b/newapi/client_wiki/api_utils/printe.py @@ -4,7 +4,7 @@ import pywikibot -from ..config import settings +from ...config import settings logger = logging.getLogger(__name__) diff --git a/newapi/api_utils/txtlib.py b/newapi/client_wiki/api_utils/txtlib.py similarity index 100% rename from newapi/api_utils/txtlib.py rename to newapi/client_wiki/api_utils/txtlib.py diff --git a/newapi/api_utils/user_agent.py b/newapi/client_wiki/api_utils/user_agent.py similarity index 100% rename from newapi/api_utils/user_agent.py rename to newapi/client_wiki/api_utils/user_agent.py diff --git a/newapi/api_utils/wd_sparql.py b/newapi/client_wiki/api_utils/wd_sparql.py similarity index 100% rename from newapi/api_utils/wd_sparql.py rename to newapi/client_wiki/api_utils/wd_sparql.py diff --git a/newapi/super/S_Category/__init__.py b/newapi/client_wiki/categories/__init__.py similarity index 100% rename from newapi/super/S_Category/__init__.py rename to newapi/client_wiki/categories/__init__.py diff --git a/newapi/super/S_Category/catdepth_new.py b/newapi/client_wiki/categories/catdepth_new.py similarity index 78% rename from newapi/super/S_Category/catdepth_new.py rename to newapi/client_wiki/categories/catdepth_new.py index d154990..20e5ccc 100644 --- a/newapi/super/S_Category/catdepth_new.py +++ b/newapi/client_wiki/categories/catdepth_new.py @@ -2,35 +2,16 @@ import functools import logging -import sys -import time -from .bot import CategoryDepth +from ...utils import function_timer +from ..constants import CATEGORY_PREFIXES +from .category_db import CategoryDepth logger = logging.getLogger(__name__) SITECODE = "en" FAMILY = "wikipedia" -CATEGORY_PREFIXES: dict[str, str] = { - "ar": "تصنيف:", - "en": "Category:", - "www": "Category:", -} - - -def function_timer(func): - """Log how long a function takes to run.""" - - @functools.wraps(func) - def wrapper(*args, **kwargs): - start = time.perf_counter() - result = func(*args, **kwargs) - logger.debug(f"{func.__name__} finished in {time.perf_counter() - start:.4f}s") - return result - - return wrapper - @functools.lru_cache(maxsize=256) def title_process(title: str, sitecode: str) -> str: @@ -77,6 +58,7 @@ def subcatquery(login_bot, title: str, sitecode: str = SITECODE, family: str = F f"<> catdepth_new.py sub cat query for {sitecode}:{title}, depth:{args2['depth']}, ns:{args2['ns']}, onlyns:{args2['onlyns']}" ) + logger.info(f"starting subcategory query: {sitecode}:{title}") bot = CategoryDepth(login_bot, title, **kwargs) result = bot.subcatquery_() diff --git a/newapi/super/S_Category/bot.py b/newapi/client_wiki/categories/category_db.py similarity index 92% rename from newapi/super/S_Category/bot.py rename to newapi/client_wiki/categories/category_db.py index 719a746..e93ff3f 100644 --- a/newapi/super/S_Category/bot.py +++ b/newapi/client_wiki/categories/category_db.py @@ -5,6 +5,8 @@ from tqdm import tqdm +from ...api_client import WikiLoginClient + logger = logging.getLogger(__name__) @@ -15,7 +17,12 @@ class CategoryDepth: Provides methods for recursively querying category members. """ - def __init__(self, login_bot, title: str = "", **kwargs) -> None: + def __init__( + self, + login_bot: WikiLoginClient, + title: str = "", + **kwargs, + ) -> None: self.login_bot = login_bot self.title: str = title @@ -57,11 +64,12 @@ def _parse_params(self, **kwargs) -> None: self.len_pages = 0 self.revids, self.timestamps, self.result_table = {}, {}, {} self.title = kwargs.get("title", "") + logger.debug(f"parsing params for {self.title}: depth={kwargs.get('depth')}, ns={kwargs.get('ns')}") try: self.depth = int(kwargs.get("depth", 0)) except ValueError: - print(f"self.depth != int: {kwargs.get('depth')}") + logger.error(f"self.depth != int: {kwargs.get('depth')}") self.depth = 0 self.props = [] @@ -249,10 +257,10 @@ def get_cat_new(self, cac: str) -> dict: if continue_params: params.update(continue_params) - api_data = self.client_request(params) + api_data = self.login_bot.client_request(params, method="get") if not api_data: - print(f"api is False for {cac}") + logger.info(f"api is False for {cac}") break continue_params = api_data.get("continue", {}) @@ -289,6 +297,7 @@ def add_to_result_table(self, x: str, tab: dict) -> None: self.result_table[x] = tab def subcatquery_(self) -> dict: + logger.info(f"starting subcatquery for {self.title}, depth={self.depth}") tablemember = self.get_cat_new(self.title) for x, zz in tablemember.items(): @@ -308,6 +317,7 @@ def subcatquery_(self) -> dict: break depth_done += 1 + logger.info(f"depth {depth_done}/{self.depth}: {len(new_list)} subcategories to process") for cat in tqdm(new_list): table2 = self.get_cat_new(cat) @@ -323,4 +333,5 @@ def subcatquery_(self) -> dict: soro = sorted(self.result_table.items(), key=lambda item: self.timestamps.get(item[0], 0), reverse=True) self.result_table = dict(soro) + logger.debug(f"subcatquery done: {len(self.result_table)} total results") return self.result_table diff --git a/newapi/client_wiki/constants.py b/newapi/client_wiki/constants.py new file mode 100644 index 0000000..0a96504 --- /dev/null +++ b/newapi/client_wiki/constants.py @@ -0,0 +1,12 @@ +""" """ + +CATEGORY_PREFIXES: dict[str, str] = { + "ar": "تصنيف:", + "en": "Category:", + "www": "Category:", +} + + +__all__ = [ + "CATEGORY_PREFIXES", +] diff --git a/newapi/super/S_Page/__init__.py b/newapi/client_wiki/pages/__init__.py similarity index 100% rename from newapi/super/S_Page/__init__.py rename to newapi/client_wiki/pages/__init__.py diff --git a/newapi/super/S_Page/data.py b/newapi/client_wiki/pages/data.py similarity index 86% rename from newapi/super/S_Page/data.py rename to newapi/client_wiki/pages/data.py index 88950d8..6e7f932 100644 --- a/newapi/super/S_Page/data.py +++ b/newapi/client_wiki/pages/data.py @@ -2,7 +2,7 @@ from .data import Content, Meta, RevisionsData, LinksData, CategoriesData, TemplateData -(Exists|all_categories_with_hidden|back_links|can_be_edit|categories|create_data|extlinks|flagged|hidden_categories|info|is_Disambig|is_redirect|iwlinks|length|links_here|links|newrevid|pageid|revid|revisions|summary|templates|templates_API|text_html|timestamp|touched|userinfo|username|wikibase_item|words) +(Exists|all_categories_with_hidden|back_links|can_be_edit|categories|create_data|extlinks|flagged|hidden_categories|info|is_disambig|is_redirect|iwlinks|length|links_here|links|newrevid|pageid|revid|revisions|summary|templates|templates_api|text_html|timestamp|touched|userinfo|username|wikibase_item|words) """ @@ -11,6 +11,8 @@ @dataclass class Content: + # text: str = "" + # newtext: str = "" text_html: str = "" summary: str = "" words: int = 0 @@ -19,7 +21,7 @@ class Content: @dataclass class Meta: - is_Disambig: bool = False + is_disambig: bool = False can_be_edit: bool = False # ns: int = 0 userinfo: dict = field(default_factory=dict) @@ -62,4 +64,4 @@ class CategoriesData: @dataclass class TemplateData: templates: dict = field(default_factory=dict) - templates_API: dict = field(default_factory=dict) + templates_api: dict = field(default_factory=dict) diff --git a/newapi/super/S_Page/super_page.py b/newapi/client_wiki/pages/super_page.py similarity index 85% rename from newapi/super/S_Page/super_page.py rename to newapi/client_wiki/pages/super_page.py index 17d8494..31ea505 100644 --- a/newapi/super/S_Page/super_page.py +++ b/newapi/client_wiki/pages/super_page.py @@ -5,18 +5,32 @@ import wikitextparser as wtp -from ...api_utils import botEdit, txtlib -from ...api_utils.ask_bot import AskBot -from ...api_utils.lang_codes import change_codes +from ...api_client import WikiLoginClient +from ...client_wiki.api_utils.handel_errors import HandleErrors from ...config import settings -from .ar_err import find_edit_error -from .bot import PageAPIS +from ..api_utils import txtlib +from ..api_utils.ask_bot import AskBot +from ..api_utils.botEdit import bot_May_Edit +from ..api_utils.lang_codes import change_codes from .data import CategoriesData, Content, LinksData, Meta, RevisionsData, TemplateData logger = logging.getLogger(__name__) -class MainPage(PageAPIS, AskBot): +def find_edit_error(old, new): + # Define the dictionary of conversion phrases + conversion_phrases = { + "#تحويل [[", + } + for phrase in conversion_phrases: + if phrase in old and phrase not in new: + logger.info(f"ar_err.py found ({phrase}) in old but not in new. return True") + return True + + return False + + +class MainPage(HandleErrors, AskBot): """ Main page class for interacting with MediaWiki pages. @@ -25,60 +39,59 @@ class MainPage(PageAPIS, AskBot): def __init__( self, - login_bot: Any, + login_bot: WikiLoginClient, title: str, lang: str = "", family: str = "wikipedia", ) -> None: # print(f"class MainPage: {lang=}") - # --- """ Initializes a MainPage instance for interacting with a MediaWiki page. Sets up page attributes including title, language, family, API endpoint, and metadata fields. Normalizes the language code, loads user tables if available, and logs into the wiki if required. """ - # --- + self.login_bot = login_bot - # --- - # --- + self.title: str = title self.lang: str = change_codes.get(lang) or lang self.family: str = family self.endpoint: str = f"https://{self.lang}.{self.family}.org/w/api.php" - # --- + self.text: str = "" self.newtext: str = "" self.ns: Union[bool, int] = False self.langlinks: Dict[str, str] = {} - # --- + self.meta = Meta() self.content = Content() self.revisions_data = RevisionsData() self.links_data = LinksData() self.categories_data = CategoriesData() self.template_data = TemplateData() - # --- + self.user: str = "" - # --- - super().__init__(login_bot) + + super().__init__() def client_request( self, params: Dict[str, Any], - request_type: str = "get", + method: str = "get", files: Optional[Dict[str, Any]] = None, + **kwargs, ) -> Dict[str, Any]: - # --- + return self.login_bot.client_request( params, - method=request_type, + method=method, files=files, + **kwargs, ) def false_edit(self) -> bool: # self.newtext # self.text - # --- """ Determines if a proposed edit should be considered erroneous and aborted. @@ -86,24 +99,24 @@ def false_edit(self) -> bool: """ if self.ns is False or self.ns != 0: return False - # --- + if settings.bot.no_fa: return False - # --- + if not self.text: self.text = self.get_text() - # --- + # If the new edit will remove 90% of the text, return False if len(self.newtext) < 0.1 * len(self.text): text_err = f"Edit will remove 90% of the text. {len(self.newtext)} < 0.1 * {len(self.text)}" text_err += f"title: {self.title}, summary: {self.content.summary}" logger.exception(text_err) return True - # --- + if self.lang == "ar" and self.ns == 0: if find_edit_error(self.text, self.newtext): return True - # --- + return False def import_page(self, family="wikipedia"): @@ -124,13 +137,13 @@ def import_page(self, family="wikipedia"): "fullhistory": 1, "assignknownusers": 1, } - # --- + data = self.client_request(params) - # --- + done = data.get("import", [{}])[0].get("revisions", 0) - # --- + logger.info(f"<> imported {done} revisions") - # --- + return data def find_create_data(self): @@ -152,24 +165,23 @@ def find_create_data(self): "rvlimit": "1", "rvdir": "newer", } - # --- - data = self.client_request(params) - # --- + + data = self.login_bot.client_request(params, method="get") + pages = data.get("query", {}).get("pages", {}) - # --- + for _, v in pages.items(): - # --- page_data = v.get("revisions", [{}])[0] - # --- + if "parentid" in page_data and page_data["parentid"] == 0: self.meta.create_data = { "timestamp": page_data["timestamp"], "user": page_data.get("user", ""), "anon": page_data.get("anon", False), } - # --- + break - # --- + return self.meta.create_data def get_text(self, redirects=False): @@ -192,53 +204,50 @@ def get_text(self, redirects=False): "rvprop": "timestamp|content|user|ids", "rvslots": "*", } # pageprops # revisions # revisions - # --- + if redirects: params["redirects"] = 1 - data = self.client_request(params) - # --- - # _dat_ = { "batchcomplete": "", "query": { "normalized": [{ "from": "وب:ملعب", "to": "ويكيبيديا:ملعب" }], "pages": { "361534": { "pageid": 361534, "ns": 4, "title": "ويكيبيديا:ملعب", "revisions": [{ "revid": 61421668, "parentid": 61421528, "user": "Al-shazali Sabeel", "timestamp": "2023-03-07T13:50:29Z", "slots": { "main": { "contentmodel": "wikitext", "contentformat": "text/x-wiki", "*": "{{عنوان الملعب}}" } } }], "pageprops": { "wikibase_item": "Q3938" } } } }, } - # --- + data = self.login_bot.client_request(params, method="get") + pages = data.get("query", {}).get("pages", {}) - # --- + for k, v in pages.items(): - # --- if "ns" in v: self.ns = v["ns"] # ns = 0 ! - # --- + if "missing" in v or k == "-1": self.meta.Exists = False # break else: self.meta.Exists = True - # --- + # title = v["title"] - # --- + pageprops = v.get("pageprops", {}) self.meta.wikibase_item = pageprops.get("wikibase_item") or self.meta.wikibase_item - # --- + # "flagged": { "stable_revid": 61366100, "level": 0, "level_text": "stable"} self.meta.flagged = v.get("flagged", False) is not False - # --- + self.revisions_data.pageid = v.get("pageid") or self.revisions_data.pageid - # --- + page_data = v.get("revisions", [{}])[0] - # --- + self.text = page_data.get("slots", {}).get("main", {}).get("*", "") self.user = page_data.get("user") or self.user self.revisions_data.revid = page_data.get("revid") or self.revisions_data.revid - # --- + self.revisions_data.timestamp = page_data.get("timestamp") or self.revisions_data.timestamp - # --- + if "parentid" in page_data and page_data["parentid"] == 0: self.meta.create_data = { "timestamp": page_data["timestamp"], "user": page_data.get("user", ""), "anon": page_data.get("anon", False), } - # --- + break - # --- + return self.text def get_qid(self): @@ -257,7 +266,6 @@ def get_qid(self): return self.meta.wikibase_item def get_infos(self): - # --- """ Fetches and updates comprehensive metadata for the current page from the MediaWiki API. @@ -277,62 +285,62 @@ def get_infos(self): # "normalize": 1, "tlnamespace": "10", } - # --- + # _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.client_request(params) - # --- + + data = self.login_bot.client_request(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, }] }, } - # --- + ta = data.get("query", {}).get("pages", [{}])[0] - # --- + # for _, ta in pages.items(): - # --- + # self.ns = ta.get("ns") or self.ns if "ns" in ta: self.ns = ta["ns"] # ns = 0 ! - # --- + self.revisions_data.pageid = ta.get("pageid") or self.revisions_data.pageid self.content.length = ta.get("length") or self.content.length self.revisions_data.revid = ta.get("lastrevid") or self.revisions_data.revid self.revisions_data.touched = ta.get("touched") or self.revisions_data.touched - # --- + self.meta.is_redirect = True if "redirect" in ta else False - # --- + for cat in ta.get("categories", []): - # --- + # _cat_ = { "ns": 14, "title": "تصنيف:بوابة سباق الدراجات الهوائية/مقالات متعلقة", "sortkey": "d8b7", "sortkeyprefix": "", "hidden": True } - # --- + if "sortkey" in cat: del cat["sortkey"] - # --- + category_title = cat["title"] - # --- + self.categories_data.all_categories_with_hidden[category_title] = cat - # --- + if cat.get("hidden") is True: self.categories_data.hidden_categories[category_title] = cat else: del cat["hidden"] self.categories_data.categories[category_title] = cat - # --- + if ta.get("langlinks", []) != []: - # --- + # {"lang": "ca", "*": "UCI World Tour 2023"} or {'lang': 'bh', 'title': 'टेम्पलेट:AWB'} - # --- + self.langlinks = {ta["lang"]: ta.get("*") or ta.get("title") for ta in ta.get("langlinks", [])} - # --- + if ta.get("templates", []) != []: - # --- + # 'templates': [{'ns': 10, 'title': 'قالب:No redirect'}], - # --- - self.template_data.templates_API = [ta["title"] for ta in ta.get("templates", [])] - # --- + + self.template_data.templates_api = [ta["title"] for ta in ta.get("templates", [])] + # "linkshere": [{"pageid": 189150,"ns": 0,"title": "طواف فرنسا"}, {"pageid": 308641,"ns": 10,"title": "قالب:AWB","redirect": ""}] self.links_data.links_here = ta.get("linkshere", []) - # --- + self.links_data.iwlinks = ta.get("iwlinks", []) - # --- + self.meta.info["done"] = True def get_text_html(self): @@ -342,37 +350,36 @@ def get_text_html(self): "formatversion": "2", "prop": "text", } - # --- + data = self.client_request(params) - # --- + # _data_ = { 'warnings': { 'main': { 'warnings': 'Unrecognized parameter: bot.' } }, 'parse': { 'title': 'ويكيبيديا:ملعب', 'pageid': 361534, 'text': '' } } - # --- + self.content.text_html = data.get("parse", {}).get("text", "") - # --- + return self.content.text_html def get_redirect_target(self): - # --- params = { "action": "query", "titles": self.title, "prop": "info", "redirects": 1, } - # --- - data = self.client_request(params) - # --- + + data = self.login_bot.client_request(params, method="get") + # _pages_ = { 'batchcomplete': '', 'query': { 'redirects': [{ 'from': 'Yemen', 'to': 'اليمن' }], 'pages': {}, 'normalized': [{ 'from': 'yemen', 'to': 'Yemen' }] } } - # --- + _redirects = {"from": "Yemen", "to": "اليمن"} - # --- + redirects = data.get("query", {}).get("redirects", [{}])[0] - # --- + to = redirects.get("to", "") - # --- + if to: - logger.info(f"<>Page:({self.title}) redirect to ({to})") - # --- + logger.debug(f"<>Page:({self.title}) redirect to ({to})") + return to def get_words(self): @@ -384,19 +391,19 @@ def get_words(self): "srlimit": srlimit, } data = self.client_request(params) - # --- + if not data: return 0 - # --- + search = data.get("query", {}).get("search", []) - # --- + for pag in search: tit = pag["title"] if tit == self.title: count = pag["wordcount"] self.content.words = count break - # --- + return self.content.words def get_extlinks(self): @@ -409,34 +416,34 @@ def get_extlinks(self): "utf8": 1, "ellimit": "max", } - # --- + links = [] - # --- + continue_params = {} - # --- + d = 0 - # --- + while continue_params != {} or d == 0: - # --- + d += 1 - # --- + if continue_params: # params = {**params, **continue_params} params.update(continue_params) - # --- - json1 = self.client_request(params) - # --- + + json1 = self.login_bot.client_request(params, method="get") + continue_params = json1.get("continue", {}) - # --- + linkso = json1.get("query", {}).get("pages", [{}])[0].get("extlinks", []) - # --- + links.extend(linkso) - # --- + links = [x["url"] for x in links] - # --- + # remove duplicates liste1 = sorted(set(links)) - # --- + self.links_data.extlinks = liste1 return liste1 @@ -450,128 +457,121 @@ def get_userinfo(self): "usprop": "groups", "ususers": self.user, } - # --- - data = self.client_request(params) - # --- + + data = self.login_bot.client_request(params, method="get") + # _userinfo_ = { "id": 229481, "name": "Mr. Ibrahem", "groups": ["editor", "reviewer", "rollbacker", "*", "user", "autoconfirmed"] } - # --- + ff = data.get("query", {}).get("users", [{}]) - # --- + if ff: self.meta.userinfo = ff[0] - # --- + return self.meta.userinfo def isRedirect(self): - # --- if not self.meta.is_redirect: self.get_infos() - # --- + return self.meta.is_redirect def isDisambiguation(self): - # --- # if the title ends with '(توضيح)' or '(disambiguation)' - self.meta.is_Disambig = self.title.endswith("(توضيح)") or self.title.endswith("(disambiguation)") - # --- - if self.meta.is_Disambig: - logger.info(f'<> page "{self.title}" is Disambiguation / توضيح') - # --- - return self.meta.is_Disambig + self.meta.is_disambig = self.title.endswith("(توضيح)") or self.title.endswith("(disambiguation)") + + if self.meta.is_disambig: + logger.debug(f'<> page "{self.title}" is Disambiguation / توضيح') + + return self.meta.is_disambig def get_categories(self, with_hidden=False): - # --- # if not self.categories_data.categories: self.get_infos() if not self.meta.info["done"]: self.get_infos() - # --- + if with_hidden: return self.categories_data.all_categories_with_hidden - # --- + return self.categories_data.categories def get_hidden_categories(self): - # --- if self.categories_data.categories == {} and self.categories_data.hidden_categories == {}: self.get_infos() - # --- + return self.categories_data.hidden_categories def get_langlinks(self): - # --- if not self.meta.info["done"]: self.get_infos() - # --- + return self.langlinks def get_templates_API(self): - # --- + if not self.meta.info["done"]: self.get_infos() - # --- - return self.template_data.templates_API + + return self.template_data.templates_api def get_links_here(self): - # --- + if not self.meta.info["done"]: self.get_infos() - # --- + return self.links_data.links_here def get_wiki_links_from_text(self): if not self.text: self.text = self.get_text() - # --- + parsed = wtp.parse(self.text) wikilinks = parsed.wikilinks - # --- + # logger.info(f'wikilinks:{str(wikilinks)}') - # --- + # for x in wikilinks: # print(x.title) - # --- + return wikilinks def Get_tags(self, tag=""): if not self.text: self.text = self.get_text() - # --- + self.text = self.text.replace("", '', 1) - # --- + parsed = wtp.parse(self.text) tags = parsed.get_tags() - # --- + # logger.info(f'tags:{str(tags)}') - # --- + if not tag: return tags - # --- + new_tags = [] - # --- + for x in tags: if x.name == tag: new_tags.append(x) - # --- + # return tags if tag == '' else [x for x in tags if x.name == tag] - # --- + return new_tags def can_edit(self, script="", delay=0): - # --- if self.family != "wikipedia": return True - # --- + if not self.text: self.text = self.get_text() - # --- - self.meta.can_be_edit = botEdit.bot_May_Edit( + + self.meta.can_be_edit = bot_May_Edit( text=self.text, title_page=self.title, botjob=script, page=self, delay=delay ) - # --- + return self.meta.can_be_edit def is_flagged(self): - # --- """ Returns whether the page is flagged for review or quality control. @@ -582,7 +582,7 @@ def is_flagged(self): """ if not self.text: self.text = self.get_text() - # --- + return self.meta.flagged def get_create_data(self): @@ -622,6 +622,7 @@ def get_revid(self): def exists(self): if not self.meta.Exists: self.get_text() + if not self.meta.Exists: logger.info(f'page "{self.title}" not exists in {self.lang}:{self.family}') return self.meta.Exists @@ -629,6 +630,7 @@ def exists(self): def namespace(self): if self.ns is False: self.get_text() + logger.debug(f"namespace: {self.ns}") return self.ns def get_user(self): @@ -669,18 +671,18 @@ def save( Returns: True if the edit was successful, False otherwise. """ - # --- + self.newtext = newtext if summary: self.content.summary = summary - # --- + if self.false_edit(): return False - # --- + message = f"Do you want to save this page? ({self.lang}:{self.title})" - # --- + user = self.meta.username - # --- + if ( self.ask_put( nodiff=nodiff, @@ -694,7 +696,7 @@ def save( is False ): return False - # --- + params = { "action": "edit", "title": self.title, @@ -703,76 +705,76 @@ def save( "minor": minor, "nocreate": nocreate, } - # --- + if nocreate != 1: del params["nocreate"] - # --- + if self.revisions_data.revid: params["baserevid"] = self.revisions_data.revid - # --- + if tags: params["tags"] = tags - # --- + # params['basetimestamp'] = self.revisions_data.timestamp - # --- - pop = self.client_request(params) - # --- + + pop = self.login_bot.client_request(params) + if not pop: return False - # --- + error = pop.get("error", {}) edit = pop.get("edit", {}) result = edit.get("result", "") - # --- + # {'edit': {'result': 'Success', 'pageid': 5013, 'title': 'User:Mr. Ibrahem/sandbox', 'contentmodel': 'wikitext', 'oldrevid': 1336986, 'newrevid': 1343447, 'newtimestamp': '2023-04-01T23:14:07Z', 'watched': ''}} - # --- + if result.lower() == "success": self.text = newtext self.user = "" - logger.info(f"<> ** true .. [[{self.lang}:{self.family}:{self.title}]] ") - # logger.info('Done True...') - # --- + logger.warning(f"<> ** true .. [[{self.lang}:{self.family}:{self.title}]] ") + logger.debug(f"save success for {self.title}") + self.revisions_data.pageid = edit.get("pageid") or self.revisions_data.pageid self.revisions_data.revid = edit.get("newrevid") or self.revisions_data.revid self.revisions_data.newrevid = edit.get("newrevid") or self.revisions_data.newrevid self.revisions_data.touched = edit.get("touched") or self.revisions_data.touched self.revisions_data.timestamp = edit.get("newtimestamp") or self.revisions_data.timestamp - # --- + return True - # --- + if error != {}: - print(pop) - er = self.handel_err(error, function="Save", params=params) - # --- + logger.debug(pop) + er = self.handle_err(error, function="Save", params=params) + return er - # --- + return False def purge(self): - # --- + params = { "action": "purge", "forcelinkupdate": 1, "forcerecursivelinkupdate": 1, "titles": self.title, } - # --- + data = self.client_request(params) - # --- + if not data: logger.info("<> ** purge error. ") return False - # --- + title2 = self.title - # --- + # 'normalized': [{'from': 'وب:ملعب', 'to': 'ويكيبيديا:ملعب'}]} - # --- + for x in data.get("normalized", []): # logger.info(f"normalized from {x['from']} to {x['to']}") if x["from"] == self.title: title2 = x["to"] break - # --- + for t in data.get("purge", []): # t = [{'ns': 4, 'title': 'ويكيبيديا:ملعب', 'purged': '', 'linkupdate': ''}] ti = t["title"] @@ -790,7 +792,6 @@ def create( nodiff="", noask=False, ) -> bool: - # --- """ Creates a new page with the specified text and summary. @@ -806,13 +807,12 @@ def create( True if the page was created successfully, False otherwise or if the user aborts. """ self.newtext = text - # --- + if not noask: - # --- message = f"Do you want to create this page? ({self.lang}:{self.title})" - # --- + user = self.meta.username - # --- + if ( self.ask_put( nodiff=nodiff, @@ -825,7 +825,7 @@ def create( is False ): return False - # --- + params = { "action": "edit", "title": self.title, @@ -834,39 +834,37 @@ def create( "notminor": 1, "createonly": 1, } - # --- - pop = self.client_request(params) - # --- + + pop = self.login_bot.client_request(params) + if not pop: return False - # --- + error = pop.get("error", {}) edit = pop.get("edit", {}) result = edit.get("result", "") - # --- + if result.lower() == "success": - # --- # {'edit': {'new': '', 'result': 'Success', 'pageid': 9090918, 'title': 'مستخدم:Mr. Ibrahem/test2024', 'contentmodel': 'wikitext', 'oldrevid': 0, 'newrevid': 61016221, 'newtimestamp': '2023-02-01T21:52:42Z'}} - # --- + self.text = text - # --- - logger.info(f"<> ** true .. [[{self.lang}:{self.family}:{self.title}]] ") - # logger.info('Done True... time.sleep() ') - # --- + + logger.warning(f"<> ** true .. [[{self.lang}:{self.family}:{self.title}]] ") + logger.debug(f"create success for {self.title}") + self.revisions_data.pageid = edit.get("pageid") or self.revisions_data.pageid self.revisions_data.revid = edit.get("newrevid") or self.revisions_data.revid self.revisions_data.touched = edit.get("touched") or self.revisions_data.touched self.revisions_data.newrevid = edit.get("newrevid") or self.revisions_data.newrevid self.revisions_data.timestamp = edit.get("newtimestamp") or self.revisions_data.timestamp - # --- + return True - # --- + if error != {}: - print(pop) - er = self.handel_err(error, function="Create", params=params) - # --- + logger.debug(pop) + er = self.handle_err(error, function="Create", params=params) return er - # --- + return False def Create( @@ -892,18 +890,18 @@ def page_backlinks(self, ns=0): "formatversion": "2", "gblredirect": 1, } - # --- + # 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) # pages = data.get("query", {}).get("pages", []) - # --- + pages = self.post_continue(params, "query", _p_="pages", p_empty=[]) - # --- + back_links = [x for x in pages if x["title"] != self.title] - # --- + self.links_data.back_links = back_links - # --- + return self.links_data.back_links def page_links(self) -> list: @@ -925,13 +923,13 @@ def page_links(self) -> list: } # data = self.client_request(params) # data = data.get('parse', {}).get('links', []) - # --- + data: list = self.post_continue(params, "parse", _p_="links", p_empty=[]) - # --- + # [{'ns': 14, 'title': 'تصنيف:مقالات بحاجة لشريط بوابات', 'exists': True}, {'ns': 14, 'title': 'تصنيف:مقالات بحاجة لصندوق معلومات', 'exists': False}] - # --- + self.links_data.links2 = data - # --- + return self.links_data.links2 def page_links_query(self, plnamespace="*"): @@ -946,17 +944,17 @@ def page_links_query(self, plnamespace="*"): } # data = self.client_request(params) # data = data.get('query', {}).get('links', []) - # --- + data = self.post_continue(params, "query", _p_="links", p_empty=[]) - # --- + # [{'ns': 14, 'title': 'تصنيف:مقالات بحاجة لشريط بوابات', 'exists': True}, {'ns': 14, 'title': 'تصنيف:مقالات بحاجة لصندوق معلومات', 'exists': False}] - # --- + self.links_data.links = data - # --- + return self.links_data.links - def get_revisions(self, rvprops=[]): - # --- + def get_revisions(self, rvprops=None) -> list: + rvprop = [ "comment", "timestamp", @@ -964,11 +962,12 @@ def get_revisions(self, rvprops=[]): # "content", "ids", ] - # --- - for x in rvprops: - if x not in rvprop: - rvprop.append(x) - # --- + + if rvprops: + for x in rvprops: + if x not in rvprop: + rvprop.append(x) + params = { "action": "query", "format": "json", @@ -982,20 +981,44 @@ def get_revisions(self, rvprops=[]): # "rvprop": "comment|timestamp|user|content|ids", "rvprop": "|".join(rvprop), } - # --- + _revisions = self.post_continue(params, "query", _p_="pages", p_empty=[]) - # --- + revisions = [] - # --- + for x in _revisions: revisions.extend(x["revisions"]) - # --- + self.revisions_data.revisions = revisions - # --- + return revisions + def post_continue( + self, + params, + action, + _p_="pages", + p_empty=None, + max=500000, + first=False, + _p_2="", + _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() else: - raise + raise # noqa: PLE0704 diff --git a/newapi/pages_bots/__init__.py b/newapi/pages_bots/__init__.py deleted file mode 100644 index b316c11..0000000 --- a/newapi/pages_bots/__init__.py +++ /dev/null @@ -1,7 +0,0 @@ -import os - -if not os.getenv("BOTNAME"): - botname = "newapi" - os.environ["BOTNAME"] = botname - - # os.getenv("BOTNAME") diff --git a/newapi/super/S_API/bot.py b/newapi/super/S_API/bot.py index 6b7f09a..e826f6f 100644 --- a/newapi/super/S_API/bot.py +++ b/newapi/super/S_API/bot.py @@ -5,10 +5,9 @@ """ import logging -import sys -from ...api_utils.ask_bot import AskBot -from ..handel_errors import HandelErrors +from ...client_wiki.api_utils.ask_bot import AskBot +from ...client_wiki.api_utils.handel_errors import HandleErrors logger = logging.getLogger(__name__) @@ -16,7 +15,7 @@ file_name = "bot_api.py" -class BotsAPIS(HandelErrors, AskBot): +class BotsAPIS(HandleErrors, AskBot): def __init__(self): # print("class BotsAPIS:") # --- @@ -80,7 +79,7 @@ def Add_To_Bottom(self, text, summary, title, poss="Head|Bottom"): # --- if error != {}: print(results) - er = self.handel_err(error, function="Add_To_Bottom", params=params) + er = self.handle_err(error, function="Add_To_Bottom", params=params) # --- return er # --- diff --git a/newapi/super/S_API/bot_api.py b/newapi/super/S_API/bot_api.py index cc16b2a..e4d3ac5 100644 --- a/newapi/super/S_API/bot_api.py +++ b/newapi/super/S_API/bot_api.py @@ -8,17 +8,15 @@ import tqdm -from ...api_utils.lang_codes import change_codes +from ...api_client import WikiLoginClient +from ...client_wiki.api_utils.lang_codes import change_codes from .bot import BotsAPIS logger = logging.getLogger(__name__) -logger = logging.getLogger(__name__) - - class NewApi(BotsAPIS): - def __init__(self, login_bot, lang="", family="wikipedia"): + def __init__(self, login_bot: WikiLoginClient, lang: str = "", family: str = "wikipedia"): # --- self.login_bot = login_bot # --- @@ -33,56 +31,6 @@ def __init__(self, login_bot, lang="", family="wikipedia"): self.cxtoken = "" # --- super().__init__() - def post_params( - self, - params, - request_type="get", - addtoken=False, - get_csrf=True, - files=None, - do_error=False, - max_retry=0, - ): - # --- - return self.login_bot.client_request( - params, - method=request_type, - files=files, - ) - def client_request( - self, - params, - request_type="get", - files=None, - ): - # --- - return self.login_bot.client_request( - params, - method=request_type, - files=files, - ) - - def post_continue( - self, - params, - action, - _p_="pages", - p_empty=None, - max=500000, - first=False, - _p_2="", - _p_2_empty=None, - ): - 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, - ) def get_username(self): return self.username @@ -944,7 +892,7 @@ def get_cxtoken(self): # --- params = {"action": "cxtoken", "format": "json"} # --- - data = self.client_request(params, request_type="post") + data = self.client_request(params, method="post") # --- if not data: return "" @@ -959,7 +907,10 @@ def get_cxtoken(self): # --- return jwt - def users_infos(self, ususers=[]): + def users_infos(self, ususers=None) -> list[dict]: + # --- + if not isinstance(ususers, list): + ususers = [] # --- params = { "action": "query", @@ -996,3 +947,57 @@ def users_infos(self, ususers=[]): results = [dict(x) for x in results] # --- return results + + def post_params( + self, + params, + method="get", + files=None, + **kwargs, + ): + # --- + return self.login_bot.client_request( + params, + method=method, + files=files, + **kwargs, + ) + + def client_request( + self, + params, + method="get", + files=None, + **kwargs, + ): + # --- + return self.login_bot.client_request( + params, + method=method, + files=files, + **kwargs, + ) + + def post_continue( + self, + params, + action, + _p_="pages", + p_empty=None, + max=500000, + first=False, + _p_2="", + _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, + ) diff --git a/newapi/super/S_Page/ar_err.py b/newapi/super/S_Page/ar_err.py deleted file mode 100644 index 2c680c3..0000000 --- a/newapi/super/S_Page/ar_err.py +++ /dev/null @@ -1,42 +0,0 @@ -#!/usr/bin/python3 - -""" - -from .super.S_Page.ar_err import find_edit_error -if find_edit_error(old, new): return -""" - - -def find_edit_error(old, new): - # Define the dictionary of conversion phrases - conversion_phrases = { - "#تحويل [[", - } - for phrase in conversion_phrases: - if phrase in old and phrase not in new: - print(f"ar_err.py found ({phrase}) in old but not in new. return True") - return True - - return False - - -def test_find_edit_error(): - # Test case 1: Phrase in old but not in new - old_text = "#تحويل [[قاعدة قانونية]]" - new_text = "[[تصنيف:أخلاقيات قانونية]]" - result = find_edit_error(old_text, new_text) - print(f"Test case 1: Result = {result}") - - # Test case 2: Phrase in both old and new - old_text = "This is an #تحويل [[ example." - new_text = "This is an #تحويل [[ example." - result = find_edit_error(old_text, new_text) - print(f"Test case 2: Result = {result}") - - # Test case 3: Phrase not in old or new - old_text = "This is an example." - new_text = "This is another example." - result = find_edit_error(old_text, new_text) - print(f"Test case 3: Result = {result}") - - print("All test cases pass!") diff --git a/newapi/super/S_Page/bot.py b/newapi/super/S_Page/bot.py deleted file mode 100644 index c6ea1f5..0000000 --- a/newapi/super/S_Page/bot.py +++ /dev/null @@ -1,39 +0,0 @@ -""" - -from .super.S_Page.bot import PageAPIS - -""" - -from ..handel_errors import HandelErrors - - -class PageAPIS(HandelErrors): - def __init__(self, login_bot): - # print("class PageAPIS:") - self.login_bot = login_bot - # --- - self.title = getattr(self, "title", "") - # --- - super().__init__() - - def post_continue( - self, - params, - action, - _p_="pages", - p_empty=None, - max=500000, - first=False, - _p_2="", - _p_2_empty=None, - ): - 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, - ) diff --git a/newapi/super/__init__.py b/newapi/super/__init__.py index a91b56b..ab273b8 100644 --- a/newapi/super/__init__.py +++ b/newapi/super/__init__.py @@ -1,12 +1,7 @@ """ """ from .S_API import bot_api -from .S_Category import catdepth_new -from .S_Page import super_page __all__ = [ - "S_API", "bot_api", - "super_page", - "catdepth_new", ] diff --git a/newapi/super/cookies_bot.py b/newapi/super/cookies_bot.py deleted file mode 100644 index 5751ded..0000000 --- a/newapi/super/cookies_bot.py +++ /dev/null @@ -1,119 +0,0 @@ -""" - -# cookies = get_cookies(lang, family, username) - -""" - -import functools -import logging -import os -import stat -from datetime import datetime, timedelta -from pathlib import Path - -from ..config import settings - -logger = logging.getLogger(__name__) -statgroup = stat.S_IRWXU | stat.S_IRWXG - - -@functools.lru_cache(maxsize=1) -def get_ta_dir() -> Path: - tool = os.getenv("HOME") - - if not tool: - tool = Path(__file__).parent - else: - tool = Path(tool) - - ta_dir = tool / "cookies" - - if not ta_dir.exists(): - ta_dir.mkdir() - logger.info("<> mkdir:") - logger.info(f"ta_dir:{ta_dir}") - logger.info("<> mkdir:") - os.chmod(ta_dir, statgroup) - - return ta_dir - - -def del_cookies_file(file_path): - # --- - file = Path(str(file_path)) - # --- - if file.exists(): - try: - file.unlink(missing_ok=True) - logger.info(f"<> unlink: file:{file}") - except Exception as e: - logger.error(f"<> unlink: Exception:{e}") - - -def get_file_name(lang, family, username) -> Path: - - ta_dir = get_ta_dir() - - if settings.bot.no_cookies: - randome = os.urandom(8).hex() - return ta_dir / f"{randome}.txt" - # --- - lang = lang.lower() - family = family.lower() - # --- - username = username.lower().replace(" ", "_").split("@")[0] - # --- - file = ta_dir / f"{family}_{lang}_{username}.txt" - # --- - if file.exists(): - # --- - # check if file old is > 3 days - # --- - file_time = datetime.fromtimestamp(file.stat().st_mtime) - # --- - if not file.stat().st_size: - del_cookies_file(file) - elif datetime.now() - file_time > timedelta(days=3): - del_cookies_file(file) - # --- - return file - - -def from_folder(lang, family, username): - # --- - file = get_file_name(lang, family, username) - # --- - cookies = False - # --- - if file.exists(): - # --- - if not file.stat().st_size: - return False - # --- - # check if file old is > 3 days - # --- - file_time = datetime.fromtimestamp(file.stat().st_mtime) - # --- - if datetime.now() - file_time > timedelta(days=3): - del_cookies_file(file) - return False - # --- - with open(file, "r", encoding="utf-8") as f: - cookies = f.read() - else: - file.touch() - os.chmod(str(file), statgroup) - # --- - return cookies - - -@functools.lru_cache(maxsize=128) -def get_cookies(lang, family, username): - # --- - cookies = from_folder(lang, family, username) - # --- - if not cookies: - logger.info(f" <> get_cookies: <> [[{lang}:{family}]] user:{username} <> not found") - return "make_new" - # --- - return cookies diff --git a/newapi/utils/__init__.py b/newapi/utils/__init__.py new file mode 100644 index 0000000..92e4633 --- /dev/null +++ b/newapi/utils/__init__.py @@ -0,0 +1,5 @@ +from .functions_timer import function_timer + +__all__ = [ + "function_timer", +] diff --git a/newapi/utils/functions_timer.py b/newapi/utils/functions_timer.py new file mode 100644 index 0000000..35d87ce --- /dev/null +++ b/newapi/utils/functions_timer.py @@ -0,0 +1,20 @@ +"""Timing decorator for profiling function execution.""" + +import functools +import logging +import time + +logger = logging.getLogger(__name__) + + +def function_timer(func): + """Log how long a function takes to run.""" + + @functools.wraps(func) + def wrapper(*args, **kwargs): + start = time.perf_counter() + result = func(*args, **kwargs) + logger.debug(f"{func.__name__} finished in {time.perf_counter() - start:.4f}s") + return result + + return wrapper diff --git a/tests/TestALL_APIS.py b/tests/TestALL_APIS.py index 79fe4e7..fbd9d26 100644 --- a/tests/TestALL_APIS.py +++ b/tests/TestALL_APIS.py @@ -7,10 +7,10 @@ @pytest.fixture def mock_dependencies(): with ( - patch("newapi.pages_bots.all_apis.WikiLoginClient") as mock_login, - patch("newapi.pages_bots.all_apis.super_page.MainPage") as mock_main_page, - patch("newapi.pages_bots.all_apis.catdepth_new.subcatquery") as mock_subcatquery, - patch("newapi.pages_bots.all_apis.bot_api.NewApi") as mock_new_api, + patch("newapi.client_wiki.all_apis.WikiLoginClient") as mock_login, + patch("newapi.client_wiki.all_apis.super_page.MainPage") as mock_main_page, + patch("newapi.client_wiki.all_apis.catdepth_new.subcatquery") as mock_subcatquery, + patch("newapi.client_wiki.all_apis.bot_api.NewApi") as mock_new_api, ): mock_login_instance = MagicMock() mock_login.return_value = mock_login_instance diff --git a/tests/TestMainPage.py b/tests/TestMainPage.py index dbba92b..f06e3c6 100644 --- a/tests/TestMainPage.py +++ b/tests/TestMainPage.py @@ -1,7 +1,7 @@ from unittest.mock import MagicMock, patch import pytest -from newapi.super.S_Page.super_page import MainPage +from newapi.client_wiki.pages.super_page import MainPage class TestMainPage: @@ -69,7 +69,6 @@ def test_nonexistent_page(self, mock_login_bot): def test_empty_page_content(self): """Test page with empty content""" - pass def test_page_without_edit_permission(self, mock_login_bot): """Test page where user cannot edit""" @@ -86,9 +85,8 @@ def test_page_without_edit_permission(self, mock_login_bot): } } page = MainPage(mock_login_bot, "الصفحة الرئيسة", "ar") - with patch("newapi.super.S_Page.super_page.botEdit.bot_May_Edit", return_value=False): + with patch("newapi.client_wiki.pages.super_page.bot_May_Edit", return_value=False): assert page.can_edit() is False def test_page_title_validation(self): """Test various page title formats""" - pass diff --git a/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates.py b/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates.py index 82f09d8..53641cd 100644 --- a/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates.py +++ b/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates.py @@ -8,11 +8,11 @@ import sys import pytest -from newapi.api_utils.bot_edit.bot_edit_by_templates import ( +from newapi.client_wiki.api_utils.bot_edit.bot_edit_by_templates import ( + BOT_USERNAME, + STOP_EDIT_TEMPLATES, Bot_Cache, - edit_username, is_bot_edit_allowed, - stop_edit_temps, ) @@ -36,7 +36,7 @@ def original_argv(): @pytest.fixture def bot_username(): """Return the bot username for testing.""" - return edit_username.get(1, "Mr.Ibrahembot") + return BOT_USERNAME # Test bot job normalization @@ -101,12 +101,12 @@ def test_cache_key_includes_botjob(self, original_argv): # Test stop templates class TestStopTemplates: - """Test cases for stop_edit_temps restrictions.""" + """Test cases for STOP_EDIT_TEMPLATES restrictions.""" def test_all_stop_templates_block_edit(self, original_argv): """Templates in 'all' stop list should block editing.""" sys.argv = ["script"] - for template in stop_edit_temps["all"]: + for template in STOP_EDIT_TEMPLATES["all"]: text = f"{{{{{template}}}}}" assert not is_bot_edit_allowed( text=text, title_page=f"Test_{template}", botjob="all" @@ -115,7 +115,7 @@ def test_all_stop_templates_block_edit(self, original_argv): def test_botjob_specific_stop_templates(self, original_argv): """Templates in botjob-specific stop list should block editing.""" sys.argv = ["script"] - for botjob, templates in stop_edit_temps.items(): + for botjob, templates in STOP_EDIT_TEMPLATES.items(): if botjob == "all": continue for template in templates: diff --git a/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates2.py b/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates2.py index bd67a75..23c7361 100644 --- a/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates2.py +++ b/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates2.py @@ -12,10 +12,10 @@ from unittest.mock import MagicMock, patch import pytest -from newapi.api_utils.bot_edit.bot_edit_by_templates import ( +from newapi.client_wiki.api_utils.bot_edit.bot_edit_by_templates import ( + STOP_EDIT_TEMPLATES, Bot_Cache, is_bot_edit_allowed, - stop_edit_temps, ) # ==================== Fixtures ==================== @@ -39,7 +39,7 @@ def reset_environment(): @pytest.fixture def mock_wtp(): """Provide a mocked wikitextparser.""" - with patch("newapi.api_utils.bot_edit.bot_edit_by_templates.wtp") as mock: + with patch("newapi.client_wiki.api_utils.bot_edit.bot_edit_by_templates.wtp") as mock: yield mock @@ -343,7 +343,7 @@ def test_bots_case_insensitive(self, setup_parser): class TestStopEditTemplates: """Test stop edit templates handling.""" - @pytest.mark.parametrize("template_name", stop_edit_temps["all"]) + @pytest.mark.parametrize("template_name", STOP_EDIT_TEMPLATES["all"]) def test_global_stop_templates_deny_edit(self, template_name, setup_parser): """Test that global stop templates deny editing.""" setup_parser([{"name": template_name, "arguments": None}]) @@ -372,7 +372,7 @@ def test_stop_template_for_different_botjob_allows_edit(self, setup_parser): @pytest.mark.parametrize( "botjob,template_list", - [(job, templates) for job, templates in stop_edit_temps.items() if job != "all"], + [(job, templates) for job, templates in STOP_EDIT_TEMPLATES.items() if job != "all"], ) def test_all_stop_templates_for_each_botjob(self, botjob, template_list, setup_parser): """Test all stop templates for each specific bot job.""" diff --git a/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates_pypass.py b/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates_pypass.py index 9b6533a..87bfa3e 100644 --- a/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates_pypass.py +++ b/tests/unit/api_utils/bot_edit/bot_edit_by_templates/test_bot_edit_by_templates_pypass.py @@ -4,7 +4,7 @@ from unittest.mock import MagicMock, patch import pytest -from newapi.api_utils.bot_edit.bot_edit_by_templates import ( +from newapi.client_wiki.api_utils.bot_edit.bot_edit_by_templates import ( Bot_Cache, is_bot_edit_allowed, ) @@ -38,7 +38,7 @@ def reset_environment(): @pytest.fixture def mock_wtp(): """Provide a mocked wikitextparser.""" - with patch("newapi.api_utils.bot_edit.bot_edit_by_templates.wtp") as mock: + with patch("newapi.client_wiki.api_utils.bot_edit.bot_edit_by_templates.wtp") as mock: yield mock @@ -103,7 +103,7 @@ def test_argv_bypasses_all_checks(self, argv_value, setup_parser): setup_parser([{"name": "nobots", "arguments": None}]) text = "{{nobots}}" - with patch("newapi.api_utils.bot_edit.bot_edit_by_templates.settings") as mock_settings: + with patch("newapi.client_wiki.api_utils.bot_edit.bot_edit_by_templates.settings") as mock_settings: if argv_value in ("botedit", "editbot"): mock_settings.bot.force_edit = True else: @@ -118,21 +118,21 @@ class TestBypassConditions: def test_bypass_with_botedit_arg(self, original_argv): """Should return True when 'botedit' is in sys.argv.""" text = "{{nobots}}" - with patch("newapi.api_utils.bot_edit.bot_edit_by_templates.settings") as mock_settings: + with patch("newapi.client_wiki.api_utils.bot_edit.bot_edit_by_templates.settings") as mock_settings: mock_settings.bot.force_edit = True assert is_bot_edit_allowed(text=text, title_page="Test", botjob="all") def test_bypass_with_editbot_arg(self, original_argv): """Should return True when 'editbot' is in sys.argv.""" text = "{{nobots}}" - with patch("newapi.api_utils.bot_edit.bot_edit_by_templates.settings") as mock_settings: + with patch("newapi.client_wiki.api_utils.bot_edit.bot_edit_by_templates.settings") as mock_settings: mock_settings.bot.force_edit = True assert is_bot_edit_allowed(text=text, title_page="Test", botjob="all") def test_bypass_with_workibrahem_arg(self, original_argv): """Should return True when 'workibrahem' is in sys.argv.""" text = "{{nobots}}" - with patch("newapi.api_utils.bot_edit.bot_edit_by_templates.settings") as mock_settings: + with patch("newapi.client_wiki.api_utils.bot_edit.bot_edit_by_templates.settings") as mock_settings: mock_settings.bot.workibrahem = True assert is_bot_edit_allowed(text=text, title_page="Test", botjob="all") diff --git a/tests/unit/api_utils/bot_edit/bot_edit_by_time/test_bot_edit_by_time.py b/tests/unit/api_utils/bot_edit/bot_edit_by_time/test_bot_edit_by_time.py index 594d828..3efe669 100644 --- a/tests/unit/api_utils/bot_edit/bot_edit_by_time/test_bot_edit_by_time.py +++ b/tests/unit/api_utils/bot_edit/bot_edit_by_time/test_bot_edit_by_time.py @@ -3,7 +3,7 @@ import sys import pytest -from newapi.api_utils.bot_edit.bot_edit_by_time import ( +from newapi.client_wiki.api_utils.bot_edit.bot_edit_by_time import ( check_create_time, check_last_edit_time, )