Wiki Tools¶

def content_search(
    site: mwclient.client.Site, text: Union[str, List[str], SearchParam]
) -> Union[List[str], List[dict]]:
    """Searches the content (wikitext) of pages. Equivalent to the following
    mediawiki API call api.php?action=query&list=search&srsearch=Star Wars.

    See https://www.mediawiki.org/wiki/API:Search for details.

    Parameters
    ----------
    site :
        Site object from mwclient lib
    text :
        Query text or instance of SearchParam

    Returns
    -------
    result:
        With ``return_json=False`` (default): a flat list of page titles. With
        ``return_json=True``: a list of raw MediaWiki ``search`` API response
        dicts, one per query (always a list, even for a single query).
    """
    if not isinstance(text, SearchParam):
        query = SearchParam(query=text)
    else:
        query = text

    def content_search_(single_text) -> Union[List[str], dict]:
        page_list = list()
        result = site.api(
            "query",
            list="search",
            srsearch=single_text,
            srlimit=query.limit,
            format="json",
        )
        if query.debug and len(result["query"]["search"]) == 0:
            print("No results")
        if query.return_json:
            return result

        for page in result["query"]["search"]:
            title = page["title"]
            if query.debug:
                print(title)
            page_list.append(title)
        return page_list

    if query.parallel:
        query_results = parallelize(
            func=content_search_, iterable=query.query, flush_at_end=query.debug
        )
    else:
        query_results = [content_search_(single_text=sq) for sq in query.query]

    if query.return_json:
        # Each entry of query_results is the raw API response dict for one query.
        # Do not flatten dicts; always return the list of responses (one per query),
        # even when only a single query was passed.
        return query_results

    return [item for sublist in query_results for item in sublist]

def copy_list_of_wiki_pages(title_list, site0, site1, overwrite, callback=None):
    """

    Parameters
    ----------
    title_list : list
    site0 : mwclient.client.Site
        Source site object from mwclient lib
    site1 : mwclient.client.Site
        Target site object from mwclient lib
    overwrite : bool
        Whether to overwrite existing pages at target site
    callback : NoneType or function
        Function passed over, to perform operation on the titles of the source pages and
        to be passed as title of the
        target pages. See examples below.
        Example functions:
            capitalize = lambda x: x.capitalize()
            def change_namespace(title, namespace):
                if ":" in namespace:
                    namespace = namespace.split(":")[0]
                if ":" in title:
                    splits = title.split(":")
                    old_name = splits[1].capitalize()
                    new_title = namespace + ":" + old_name
                else:
                    new_title = namespace + ":" + title.capitalize()
                return new_title
        Examples of passing a function as parameter:
            callback = capitalize
            callback = lambda x: x.lower()

    Returns
    -------
    results_dict : dict
        Dictionary, containing the results of the copying operations
    """

    success_list = list()
    fail_list = list()
    for title0 in title_list:
        if callback is None:
            title1 = title0
        else:
            title1 = callback(title0)
        success = copy_wiki_page(title0, title1, site0, site1, overwrite)
        if success:
            success_list.append(title1)
        else:
            fail_list.append(title1)
    results_dict = {
        "Successfully copied pages": success_list,
        "Pages failed to copy": fail_list,
    }
    return results_dict

def copy_wiki_page(title0, title1, site0, site1, overwrite=True):
    """

    Parameters
    ----------
    title0 : str
        Title of the source page
    title1 : str
        Title of the target page
    site0 : mwclient.client.Site
        Source site object from mwclient lib
    site1 : mwclient.client.Site
        Target site object from mwclient lib
    overwrite : bool
        Whether to overwrite existing pages at target site

    Returns
    -------
    success: bool

    """
    if title0.lower() == title1.lower() and site0 == site1:
        # copy on it self = no action necessary
        success = True
    else:
        page0 = site0.pages[title0]
        content = page0.text()
        if overwrite:
            success = create_or_overwrite_wiki_page(title1, content, site1)
        else:
            search_result = search_wiki_page(title1, site1)
            if (
                search_result["Result"] and search_result["Exact match"]
            ):  # page already exists
                success = False
            else:
                # search_result["Result"] == True/False
                # search_result["Exact match"] == False
                success = create_or_overwrite_wiki_page(title1, content, site1)
    return success

def create_or_overwrite_wiki_page(title, content, site):
    """Creates a page with the passed title and content. If the page already exists,
    the prior content is replaced with the passed content.

    Parameters
    ----------
    title : str
        Title of the wiki page, e.g., User:Someone1234
    content : str
    site : mwclient.client.Site
        Site object from mwclient lib

    Returns
    -------
    success : bool
    """
    target_page = site.pages[title]
    target_page.edit(content, "[bot] create page")
    success = True
    return success

def create_or_update_wiki_page_with_template(
    title, content, site, overwrite_with_empty=False
):
    """Creates a wiki page with a template included in the content. If the page does
    already exist, the parameters within the template are update

    Parameters
    ----------
    title : str
        Title of the wiki page, e.g., User:Someone1234
    content : str
    site : mwclient.client.Site
        Site object from mwclient lib
    overwrite_with_empty : bool
        Decided whether a template parameter's value in an preexisting page is
        overwritten with an empty value

    Returns
    -------
    success : bool
    """
    search_result = search_wiki_page(title, site)
    if search_result["Result"]:
        existing_page = site.pages[title]
        existing_text = existing_page.text()
        # update the page's content (template only)
        updated_content = update_template_within_wikitext(
            text=existing_text,
            template_text=content,
            overwrite_with_empty=overwrite_with_empty,
        )
        success = create_or_overwrite_wiki_page(title, updated_content, site)
    else:
        # just create the page
        success = create_or_overwrite_wiki_page(title, content, site)
    return success

def create_site_object(
    domain: str, password_file: Union[str, FilePath] = "", credentials: dict = None
) -> mwclient.client.Site:
    """
    Parameters
    ----------
    domain :
        Domain of the OSW instance, as specifed in the yaml file
    password_file :
        path to file with <username>\n<password>
    credentials :
        Dictionary with the credentials (username, password)
    Returns
    -------
    site : mwclient.client.Site
        Site object from mwclient lib
    """
    domain_dict = {
        "wiki-dev": {"Address": "wiki-dev.open-semantic-lab.org"},
        "onterface": {"Address": "onterface.open-semantic-lab.org:"},
    }
    if domain in domain_dict.keys():
        domain = domain_dict[domain]["Address"]

    site = mwclient.Site(domain, path="/w/")
    if credentials is None:
        credentials = read_credentials_from_yaml(password_file, domain)
    # else:
    #     credentials = credentials
    # Login with dictionary unpacking:
    # site.login(**credentials)
    # Explicit login:
    site.login(username=credentials["username"], password=credentials["password"])
    del credentials
    return site

def delete_wiki_page(title, site, reason):
    """Deletes the wiki page with the passed title, if it was found (exact match!),
    otherwise returns False

    Parameters
    ----------
    title : str
        Title of the wiki page, e.g., User:Someone1234
    site : mwclient.client.Site
        Site object from mwclient lib
    reason : str

    Returns
    -------
    success : bool
    """
    try:
        page = site.pages[title]
        page.delete(reason=reason, watch=False, unwatch=True, oldimage=False)
        success = True
    except mwclient.errors.APIError:
        success = False
    return success

def edit_wiki_page_with_content_merge(title, new_content, site, template_name):
    """Edits an existing wiki page, while merging the passed new content with the
    content of the existing page

    Parameters
    ----------
    title : str
        Title of the wiki page, e.g., User:Someone1234
    new_content : str
    site : mwclient.client.Site
        Site object from mwclient lib
    template_name : str

    Returns
    -------
    success : bool
    """
    search_result = search_wiki_page(title, site)
    if search_result["Result"]:
        source_page = site.pages[title]
        source_page_content = source_page.text()
        # todo: test function
        new_content = merge_wiki_page_text(
            new_content, source_page_content, template_name=template_name
        )
        target_page = site.pages[title]
        target_page.edit(new_content, "[bot] update of page content")
        success = True
    else:
        success = False
    return success

def get_file_info_and_usage(
    site: mwclient.client.Site, title: Union[str, List[str], SearchParam]
) -> List[Dict[str, Union[Dict[str, str], List[str]]]]:
    """(For 'File' pages only) Get information about the file and its usage

    Parameters
    ----------
    site:
        Site object from mwclient lib.
    title:
        Title(s) of the wiki page(s) or instance of SearchParam.

    Returns
    -------
    result:
        Dictionary with page titles as keys and nested dictionary with keys 'info' and
        'usage'.

    Notes
    -----
    Query to reproduce:
        action=query
        format=json
        prop=imageinfo|fileusage
        titles=File%3AOSW857d85031d85425aa94db8b4720e84b7.png
        &iiprop=timestamp%7Cuser&fulimit=5000"

    Resources
    ---------
    Use the sandbox to design and test the queries:
    https://demo.open-semantic-lab.org/wiki/Special:ApiSandbox
    """
    if not isinstance(title, SearchParam):
        query = SearchParam(query=title, debug=False)
    else:  # SearchParam
        query = title

    def get_file_info_and_usage_(single_title):
        api_request_result = site.api(
            action="query",
            format="json",
            prop="imageinfo|fileusage",
            titles=single_title,
            iiprop="timestamp|user",
            fulimit=query.limit,
        )
        using_pages = []
        file_info = {
            "title": single_title,
            "author": "File not found or no creation logged",
            "timestamp": "File not found or no creation logged",
            "editor": [],
            "editing_timestamp": [],
        }

        if len(api_request_result["query"]["pages"]) == 0:
            if query.debug:
                _logger.debug(f"Page not found: '{single_title}'!")
        else:
            image_info: List[Dict[str, str]] = []
            file_usage: List[Dict[str, Union[str, int]]] = []
            for _page_id, page_dict in api_request_result["query"]["pages"].items():
                if page_dict["title"] == single_title:
                    image_info = page_dict.get("imageinfo", [])
                    file_usage = page_dict.get("fileusage", [])
            if len(image_info) != 0:
                file_info["author"] = image_info[0]["user"]
                file_info["timestamp"] = image_info[0]["timestamp"]
                for ii in image_info:
                    file_info["editor"].append(ii["user"])
                    file_info["editing_timestamp"].append(ii["timestamp"])
            if file_usage is not None:
                for fu_page_dict in file_usage:
                    using_pages.append(fu_page_dict["title"])
            if query.debug:
                _logger.debug(f"File info for '{single_title}' retrieved.")
        return {"info": file_info, "usage": using_pages}

    if query.parallel:
        api_request_results = parallelize(
            func=get_file_info_and_usage_,
            iterable=query.query,
            flush_at_end=query.debug,
        )
    else:
        api_request_results = [
            get_file_info_and_usage_(single_title=st) for st in query.query
        ]

    return api_request_results

def get_query_limit(query: str) -> Optional[int]:
    """Returns the limit an SMW ask query sets itself, None if it sets none

    Parameters
    ----------
    query :
        an SMW ask query string, e.g. '[[Category:Item]]|?Name|limit=2'

    Returns
    -------
    result:
        the limit the query asks for, or None if the query does not set one or
        sets one that is not a number
    """
    limit = None
    for parameter in CONDITION_PATTERN.sub("", query).split("|"):
        match = LIMIT_PARAM_PATTERN.match(parameter.strip())
        if match:
            # SMW honours the last limit in the query string
            limit = int(match.group(1))
    return limit

def prefix_search(
    site: mwclient.client.Site, text: Union[str, SearchParam]
) -> Union[List[str], List[dict]]:
    """Standard query. Equivalent to the following mediawiki API call
    api.php?action=query&list=prefixsearch&pssearch=Star Wars.

    See https://www.mediawiki.org/wiki/API:Prefixsearch for details.

    Parameters
    ----------
    site :
        Site object from mwclient lib
    text :
        Query text or instance of SearchParam

    Returns
    -------
    result:
        With ``return_json=False`` (default): a flat list of page titles. With
        ``return_json=True``: a list of raw MediaWiki ``prefixsearch`` API response
        dicts, one per query (always a list, even for a single query).
    """
    if not isinstance(text, SearchParam):
        query = SearchParam(query=text)
    else:
        query = text

    def prefix_search_(single_text) -> Union[List[str], dict]:
        page_list = list()
        result = site.api(
            "query",
            list="prefixsearch",
            pssearch=single_text,
            pslimit=query.limit,
            format="json",
        )
        if query.debug and len(result["query"]["prefixsearch"]) == 0:
            _logger.debug("No results")
        if query.return_json:
            return result

        for page in result["query"]["prefixsearch"]:
            title = page["title"]
            if query.debug:
                _logger.debug(title)
            page_list.append(title)
        return page_list

    if query.parallel:
        query_results = parallelize(
            func=prefix_search_, iterable=query.query, flush_at_end=query.debug
        )
    else:
        query_results = [prefix_search_(single_text=sq) for sq in query.query]

    if query.return_json:
        # Each entry of query_results is the raw API response dict for one query.
        # Do not flatten dicts; always return the list of responses (one per query),
        # even when only a single query was passed.
        return query_results

    return [item for sublist in query_results for item in sublist]

def read_credentials_from_yaml(
    password_file: Union[str, FilePath], domain: str = None
) -> dict:
    """Reads credentials from a yaml file

    Parameters
    ----------
    password_file :
        Path to the yaml file with the credentials.
    domain:
        Domain of the OSW instance, as specifed in the yaml file.

    Returns
    -------
    credentials :
        Dictionary with the credentials, expected to
        contain keys 'username' and 'password'.
    """
    if password_file != "":
        with open(password_file) as stream:
            try:
                accounts = yaml.safe_load(stream)
                if domain is not None and domain in accounts.keys():
                    domain = domain
                elif len(accounts.keys()) > 0:
                    domain = next(iter(accounts.keys()))
                    if len(accounts.keys()) > 0:
                        domain = next(iter(accounts.keys()))
                user = accounts[domain]["username"]
                password = accounts[domain]["password"]
            except yaml.YAMLError as exc:
                _logger.error(exc)
    else:
        user = input("Enter bot username (username@botname)")
        password = getpass.getpass("Enter bot password")
    return {"username": user, "password": password}

def read_domains_from_credentials_file(
    cred_filepath: Union[str, FilePath],
) -> Tuple[List[str], Dict[str, Dict[str, str]]]:
    """Reads domains and credentials from a yaml file

    Parameters
    ----------
    cred_filepath
        Path to the yaml file with the credentials

    Returns
    -------

    """
    with open(cred_filepath, encoding="utf-8") as stream_:
        try:
            accounts_dict = yaml.safe_load(stream_)
            # An empty file is parsed as None by yaml.safe_load, which would
            #  otherwise raise an AttributeError on the .keys() call below
            if accounts_dict is None:
                accounts_dict = {}
            domains_list = list(accounts_dict.keys())
            if len(domains_list) == 0:
                raise ValueError("No domain found in accounts.pwd.yaml!")
            return domains_list, accounts_dict
        except yaml.YAMLError as exc_:
            _logger.error(exc_)

def search_redirection_sources(
    site: mwclient.client.Site, target_title: str, debug: bool = False
):
    """Returns a list of pages redirecting to the page with target_title per #REDIRECT
    [[target]] syntax

    Parameters
    ----------
    site :
        Site object from mwclient lib
    target_title :
        Title of the target wiki page
    debug:
        Whether to log debugging messages

    Returns
    -------
    page_list : list of pages redirecting to the page with target_title
    """
    page_list = []
    result = site.api("query", titles=target_title, prop="redirects", format="json")
    if len(result["query"]["pages"]) == 0:
        if debug:
            _logger.debug("No results")
    else:
        for page in result["query"]["pages"]:
            if "redirects" not in result["query"]["pages"][page]:
                if debug:
                    _logger.debug("No results")
            else:
                for redirecting_source in result["query"]["pages"][page]["redirects"]:
                    title = redirecting_source["title"]
                    page_list.append(title)
    return page_list

def search_wiki_page(title: str, site: mwclient.client.Site):
    """Page search wrapper that adds exact match functionality with ignore-case on
    top of the  prefix_search()'s functionality.

    Parameters
    ----------
    title :
        Title of the wiki page, e.g., User:Someone1234
    site :
        Site object from mwclient lib

    Returns
    -------
    result_dict : dict
    """
    result = site.api(
        "query", list="prefixsearch", pssearch=title, pslimit=1000, format="json"
    )
    if len(result["query"]["prefixsearch"]) == 0:
        return {"Result": False, "List": list()}
    else:
        exact_match = False
        page_title_list = list()
        for page in result["query"]["prefixsearch"]:
            page_title = page["title"]
            page_title_list.append(page_title)
            if page_title.lower() == title.lower():
                exact_match = True
        result_dict = {
            "Result": True,
            "List": page_title_list,
            "Exact match": exact_match,
        }
        return result_dict

def semantic_search(
    site: mwclient.client.Site, query: Union[str, List[str], SearchParam]
) -> Union[List[str], List[dict], List[SemanticSearchResult]]:
    """Semantic query

    Parameters
    ----------
    site :
        Site object from mwclient lib
    query :
        (List of) query text(s) or instance of SearchParam. A query that sets
        ``limit=`` itself keeps that limit; ``SearchParam.limit`` is only
        appended to a query that does not, and only when it is not None.

    Returns
    -------
    result:
        With ``return_json=False`` and ``return_meta=False`` (default): a flat
        list of page-title fulltext strings. With ``return_json=True``: a list of
        raw SMW ``ask`` result dicts, one per query (always a list, even for a
        single query). With ``return_meta=True``: a list of SemanticSearchResult,
        one per query, which reports whether the wiki truncated the result set.
        ``return_json`` takes precedence if both are set.
    """
    if not isinstance(query, SearchParam):
        query = SearchParam(query=query)

    def semantic_search_(single_query):
        page_list = list()
        # SMW honours the last limit in the query string, so appending the
        # default would silently override a limit the caller wrote themselves.
        # A SearchParam limit of None asks for no limit at all, leaving the
        # wiki to apply its own default
        limit = get_query_limit(single_query)
        if limit is None and query.limit is not None:
            limit = query.limit
            single_query += f"|limit={limit}"
        result = site.api("ask", query=single_query, format="json")
        results = _ask_results_as_dict(result["query"]["results"])
        n = len(results)
        if query.debug:
            if n == 0:
                _logger.debug(f"Query '{single_query}' returned no results")
            else:
                _logger.debug(f"Query '{single_query}' returned {n} results")
        # SMW reports an incomplete result set with a top-level
        # 'query-continue-offset' holding the offset the remainder starts at,
        # and omits the key for a complete one. That replaces the earlier
        # comparison of the result count against the limit, which was wrong in
        # both directions: it could not see the wiki's own '$smwgQMaxLimit'
        # cap, and it reported a complete set of exactly 'limit' results as
        # truncated
        next_offset = result.get("query-continue-offset")
        truncated = next_offset is not None
        if truncated:
            _logger.warning(
                f"Query '{single_query}' returned {n} results and the wiki "
                f"reports further ones. Results are truncated - raise the "
                f"limit or page through with '|offset={next_offset}' to "
                f"retrieve the remainder."
            )
        if query.return_json:
            return result

        dropped = 0
        for page in results.values():
            title = page["fulltext"]
            exists = page["exists"]
            if "#" not in title and query.debug:
                _logger.debug(title)
                # original position of "page_list.append(title)" line
            if exists == "1":
                page_list.append(title)
            else:
                dropped += 1
        if dropped > 0:
            _logger.warning(
                f"Query '{single_query}': {dropped} of {n} results were dropped "
                f"because the wiki reported them as non-existing pages."
            )
        if query.return_meta:
            return SemanticSearchResult(
                query=single_query,
                titles=page_list,
                count=n,
                truncated=truncated,
                next_offset=next_offset,
            )
        return page_list

    if query.parallel:
        query_results = parallelize(
            func=semantic_search_, iterable=query.query, flush_at_end=query.debug
        )
    else:
        query_results = [semantic_search_(single_query=sq) for sq in query.query]

    if query.return_json or query.return_meta:
        # Each entry of query_results is the raw SMW result dict, or the
        # SemanticSearchResult, for one query. Do not flatten those; always
        # return one entry per query, even when only a single query was passed.
        return query_results

    return [item for sublist in query_results for item in sublist]