pywikibot/flow.py

"""Objects representing Flow entities, like boards, topics, and posts."""
#
# (C) Pywikibot team, 2015-2024
#
# Distributed under the terms of the MIT license.
#
from __future__ import annotations

import abc
import datetime
from typing import Any
from urllib.parse import parse_qs, urlparse

import pywikibot
from pywikibot import config
from pywikibot.backports import Iterator, Mapping
from pywikibot.exceptions import (
    LockedPageError,
    NoPageError,
    UnknownExtensionError,
)
from pywikibot.page import BasePage, PageSourceType, User
from pywikibot.tools import cached, deprecated_args


class FlowPage(BasePage, abc.ABC):

    """The base page meta class for the Flow extension.

    Defines Flow page-like object for :class:`Board` and :class:`Topic`.
    It cannot be instantiated directly.
    """

    def __init__(self, source: PageSourceType, title: str = '') -> None:
        """Initializer.

        :param source: A Flow-enabled site or a Link or Page on such a site
        :param title: normalized title of the page

        :raises TypeError: incorrect use of parameters
        :raises ValueError: use of non-Flow-enabled Site
        """
        super().__init__(source, title)

        if not self.site.has_extension('Flow'):
            raise UnknownExtensionError('site is not Flow-enabled')

    @abc.abstractmethod
    def _load(self, force: bool = False) -> dict[str, Any]:
        """Abstract method to load and cache the Flow data.

        Subclasses must overwrite _load() method to load and cache
        the object's internal data from the API.
        """
        raise NotImplementedError

    @property
    @cached
    def uuid(self) -> str:
        """Return the UUID of the page.

        :return: UUID of the page
        """
        return self._load()['workflowId']

    def get(self, force: bool = False, get_redirect: bool = False
            ) -> dict[str, Any]:
        """Get the page's content."""
        if get_redirect or force:
            raise NotImplementedError(
                "Neither 'force' nor 'get_redirect' parameter is implemented "
                'in {}.get()'.format(self.__class__.__name__))

        # TODO: Return more useful data
        return getattr(self, '_data', {})


class Board(FlowPage):

    """A Flow discussion board."""

    def _load(self, force: bool = False) -> dict[str, Any]:
        """Load and cache the Board's data, derived from its topic list.

        :param force: Whether to force a reload if the data is already loaded
        """
        if not hasattr(self, '_data') or force:
            self._data = self.site.load_board(self)
        return self._data

    @staticmethod
    def _parse_url(links: Mapping[str, Any]) -> dict[str, Any]:
        """Parse a URL retrieved from the API."""
        if 'fwd' in links:
            rule = links['fwd']
        elif 'rev' in links:
            rule = links['rev']
        else:
            raise ValueError('Illegal board data (missing required data).')
        parsed_url = urlparse(rule['url'])
        params = parse_qs(parsed_url.query)
        new_params: dict[str, Any] = {}
        for key, value in params.items():
            if key != 'title':
                key = key.replace('topiclist_', '').replace('-', '_')
                if key == 'offset_dir':
                    new_params['reverse'] = (value == 'rev')
                else:
                    new_params[key] = value
        return new_params

    @deprecated_args(limit='total')  # since 8.0.0
    def topics(self, *,
               content_format: str = 'wikitext',
               total: int | None = None,
               sort_by: str = 'newest',
               offset: str | datetime.datetime | None = None,
               offset_uuid: str = '',
               reverse: bool = False,
               include_offset: bool = False,
               toc_only: bool = False
               ) -> Iterator[Topic]:
        """Load this board's topics.

        .. versionchanged:: 8.0
           The *total* parameter was added as a per request limit.
           All parameters are keyword only parameters.
        .. deprecated:: 8.0
           The *limit* parameter. Use ``-step`` global option or
           :ref:`config.step<Settings to Avoid Server Overload>` instead.

        :param content_format: The content format to request the data in;
            must be either 'wikitext', 'html', or 'fixed-html'
        :param total: The number of topics to fetch.
        :param sort_by: Algorithm to sort topics by;
            must be either 'newest' or 'updated'
        :param offset: The timestamp to start at (when sortby is 'updated').
        :param offset_uuid: The UUID to start at (when sortby is 'newest').
        :param reverse: Whether to reverse the topic ordering.
        :param include_offset: Whether to include the offset topic.
        :param toc_only: Whether to only include information for the TOC.
        :yield: A generator of this board's topics.
        """
        maxlimit = min(config.step, 100) if config.step > 0 else 100
        request_limit = min(total, maxlimit)
        data = self.site.load_topiclist(self, content_format=content_format,
                                        limit=request_limit, sortby=sort_by,
                                        toconly=toc_only, offset=offset,
                                        offset_id=offset_uuid, reverse=reverse,
                                        include_offset=include_offset)
        count = 0
        while data['roots']:
            for root in data['roots']:
                topic = Topic.from_topiclist_data(self, root, data)
                yield topic

                count += 1
                if count >= total:
                    return

            continue_args = self._parse_url(data['links']['pagination'])
            data = self.site.load_topiclist(self, **continue_args)

    def new_topic(self, title: str, content: str,
                  content_format: str = 'wikitext') -> Topic:
        """Create and return a Topic object for a new topic on this Board.

        :param title: The title of the new topic (must be in plaintext)
        :param content: The content of the topic's initial post
        :param content_format: The content format of the supplied content;
            either 'wikitext' or 'html'
        :return: The new topic
        """
        return Topic.create_topic(self, title, content, content_format)


class Topic(FlowPage):

    """A Flow discussion topic."""

    def _load(self, force: bool = False, content_format: str = 'wikitext'
              ) -> dict[str, Any]:
        """Load and cache the Topic's data.

        :param force: Whether to force a reload if the data is already loaded
        :param content_format: The post format in which to load
        """
        if not hasattr(self, '_data') or force:
            self._data = self.site.load_topic(self, content_format)
        return self._data

    def _reload(self) -> None:
        """Forcibly reload the topic's root post."""
        self.root._load(load_from_topic=True)

    @classmethod
    def create_topic(cls, board: Board, title: str,
                     content: str, content_format: str = 'wikitext'
                     ) -> Topic:
        """Create and return a Topic object for a new topic on a Board.

        :param board: The topic's parent board
        :param title: The title of the new topic (must be in plaintext)
        :param content: The content of the topic's initial post
        :param content_format: The content format of the supplied content;
            either 'wikitext' or 'html'
        :return: The new topic
        """
        data = board.site.create_new_topic(board, title, content,
                                           content_format)
        return cls(board.site, data['topic-page'])

    @classmethod
    def from_topiclist_data(cls, board: Board,
                            root_uuid: str,
                            topiclist_data: dict[str, Any]) -> Topic:
        """Create a Topic object from API data.

        :param board: The topic's parent Flow board
        :param root_uuid: The UUID of the topic and its root post
        :param topiclist_data: The data returned by view-topiclist
        :return: A Topic object derived from the supplied data
        :raises TypeError: any passed parameters have wrong types
        :raises ValueError: the passed topiclist_data is missing required data
        """
        if not isinstance(board, Board):
            raise TypeError('board must be a pywikibot.flow.Board object.')
        if not isinstance(root_uuid, str):
            raise TypeError('Topic/root UUID must be a string.')

        topic = cls(board.site, 'Topic:' + root_uuid)
        topic._root = Post.fromJSON(topic, root_uuid, topiclist_data)
        topic._uuid = root_uuid
        return topic

    @property
    def root(self) -> Post:
        """The root post of this topic."""
        if not hasattr(self, '_root'):
            self._root = Post.fromJSON(self, self.uuid, self._data)
        return self._root

    @property
    def is_locked(self) -> bool:
        """Whether this topic is locked."""
        return self.root._current_revision['isLocked']

    @property
    def is_moderated(self) -> bool:
        """Whether this topic is moderated."""
        return self.root._current_revision['isModerated']

    def replies(self, content_format: str = 'wikitext', force: bool = False
                ) -> list[Post]:
        """A list of replies to this topic's root post.

        :param content_format: Content format to return contents in;
            must be 'wikitext', 'html', or 'fixed-html'
        :param force: Whether to reload from the API instead of using the cache
        :return: The replies of this topic's root post
        """
        return self.root.replies(content_format=content_format, force=force)

    def reply(self, content: str, content_format: str = 'wikitext') -> Post:
        """A convenience method to reply to this topic's root post.

        :param content: The content of the new post
        :param content_format: The format of the given content;
            must be 'wikitext' or 'html')
        :return: The new reply to this topic's root post
        """
        return self.root.reply(content, content_format)

    # Moderation
    def lock(self, reason: str) -> None:
        """Lock this topic.

        :param reason: The reason for locking this topic
        """
        self.site.lock_topic(self, True, reason)
        self._reload()

    def unlock(self, reason: str) -> None:
        """Unlock this topic.

        :param reason: The reason for unlocking this topic
        """
        self.site.lock_topic(self, False, reason)
        self._reload()

    def delete_mod(self, reason: str) -> None:
        """Delete this topic through the Flow moderation system.

        :param reason: The reason for deleting this topic.
        """
        self.site.delete_topic(self, reason)
        self._reload()

    def hide(self, reason: str) -> None:
        """Hide this topic.

        :param reason: The reason for hiding this topic.
        """
        self.site.hide_topic(self, reason)
        self._reload()

    def suppress(self, reason: str) -> None:
        """Suppress this topic.

        :param reason: The reason for suppressing this topic.
        """
        self.site.suppress_topic(self, reason)
        self._reload()

    def restore(self, reason: str) -> None:
        """Restore this topic.

        :param reason: The reason for restoring this topic.
        """
        self.site.restore_topic(self, reason)
        self._reload()

    def summary(self) -> str | None:
        """Get this topic summary, if any.

        :return: summary or None
        """
        if 'summary' in self.root._current_revision:
            return self.root._current_revision['summary']['revision'][
                'content']['content']
        return None

    def summarize(self, summary: str) -> None:
        """Summarize this topic.

        :param summary: The summary that will be added to the topic.
        """
        self.site.summarize_topic(self, summary)
        self._reload()


# Flow non-page-like objects
class Post:

    """A post to a Flow discussion topic."""

    def __init__(self, page: Topic, uuid: str) -> None:
        """
        Initializer.

        :param page: Flow topic
        :param uuid: UUID of a Flow post

        :raises TypeError: incorrect types of parameters
        """
        if not isinstance(page, Topic):
            raise TypeError('Page must be a Topic object')
        if not page.exists():
            raise NoPageError(page, 'Topic must exist: %s')
        if not isinstance(uuid, str):
            raise TypeError('Post UUID must be a string')

        self._page = page
        self._uuid = uuid

        self._content: dict[str, Any] = {}

    @classmethod
    def fromJSON(cls, page: Topic, post_uuid: str,  # noqa: N802
                 data: dict[str, Any]) -> Post:
        """
        Create a Post object using the data returned from the API call.

        :param page: A Flow topic
        :param post_uuid: The UUID of the post
        :param data: The JSON data returned from the API

        :return: A Post object
        :raises TypeError: data is not a dict
        :raises ValueError: data is missing required entries
        """
        post = cls(page, post_uuid)
        post._set_data(data)

        return post

    def _set_data(self, data: dict[str, Any]) -> None:
        """Set internal data and cache content.

        :param data: The data to store internally
        :raises TypeError: data is not a dict
        :raises ValueError: missing data entries or post/revision not found
        """
        if not isinstance(data, dict):
            raise TypeError('Illegal post data (must be a dictionary).')
        if ('posts' not in data) or ('revisions' not in data):
            raise ValueError('Illegal post data (missing required data).')
        if self.uuid not in data['posts']:
            raise ValueError('Post not found in supplied data.')

        current_revision_id = data['posts'][self.uuid][0]
        if current_revision_id not in data['revisions']:
            raise ValueError('Current revision of post'
                             'not found in supplied data.')

        self._current_revision = data['revisions'][current_revision_id]
        if 'content' in self._current_revision:
            content = self._current_revision.pop('content')
            assert isinstance(content, dict)
            assert isinstance(content['content'], str)
            self._content[content['format']] = content['content']

    def _load(self, force: bool = True, content_format: str = 'wikitext',
              load_from_topic: bool = False) -> dict[str, Any]:
        """Load and cache the Post's data using the given content format.

        :param load_from_topic: Whether to load the post from the whole topic
        """
        if load_from_topic:
            data = self.page._load(force=force, content_format=content_format)
        else:
            data = self.site.load_post_current_revision(self.page, self.uuid,
                                                        content_format)
        self._set_data(data)
        return self._current_revision

    @property
    def uuid(self) -> str:
        """Return the UUID of the post.

        :return: UUID of the post
        """
        return self._uuid

    @property
    def site(self) -> pywikibot.site.BaseSite:
        """Return the site associated with the post.

        :return: Site associated with the post
        """
        return self._page.site

    @property
    def page(self) -> Topic:
        """Return the page associated with the post.

        :return: Page associated with the post
        """
        return self._page

    @property
    def is_moderated(self) -> bool:
        """Whether this post is moderated."""
        if not hasattr(self, '_current_revision'):
            self._load()
        return self._current_revision['isModerated']

    @property
    def creator(self) -> User:
        """The creator of this post."""
        if not hasattr(self, '_current_revision'):
            self._load()
        if not hasattr(self, '_creator'):
            self._creator = User(self.site,
                                 self._current_revision['creator']['name'])
        return self._creator

    def get(self, content_format: str = 'wikitext',
            force: bool = False) -> str:
        """Return the contents of the post in the given format.

        :param force: Whether to reload from the API instead of using the cache
        :param content_format: Content format to return contents in
        :return: The contents of the post in the given content format
        """
        if content_format not in self._content or force:
            self._load(content_format=content_format)
        return self._content[content_format]

    def replies(self, content_format: str = 'wikitext', force: bool = False
                ) -> list[Post]:
        """Return this post's replies.

        :param content_format: Content format to return contents in;
            must be 'wikitext', 'html', or 'fixed-html'
        :param force: Whether to reload from the API instead of using the cache
        :return: This post's replies
        """
        if content_format not in ('wikitext', 'html', 'fixed-html'):
            raise ValueError('Invalid content format.')

        if hasattr(self, '_replies') and not force:
            return self._replies  # type: ignore[has-type]

        # load_from_topic workaround due to T106733
        # (replies not returned by view-post)
        if not hasattr(self, '_current_revision') or force:
            self._load(content_format=content_format, load_from_topic=True)

        reply_uuids = self._current_revision['replies']
        self._replies = [Post(self.page, uuid) for uuid in reply_uuids]

        return self._replies

    def reply(self, content: str, content_format: str = 'wikitext') -> Post:
        """Reply to this post.

        :param content: The content of the new post
        :param content_format: The format of the given content;
            must be 'wikitext' or 'html'
        :return: The new reply post
        """
        self._load()
        if self.page.is_locked:
            raise LockedPageError(self.page, 'Topic %s is locked.')

        reply_url = self._current_revision['actions']['reply']['url']
        parsed_url = urlparse(reply_url)
        params = parse_qs(parsed_url.query)
        reply_to = params['topic_postId']
        if self.uuid == reply_to:
            del self._current_revision
            del self._replies
        data = self.site.reply_to_post(self.page, reply_to, content,
                                       content_format)
        return Post(self.page, data['post-id'])

    # Moderation
    def delete(self, reason: str) -> None:
        """Delete this post through the Flow moderation system.

        :param reason: The reason for deleting this post.
        """
        self.site.delete_post(self, reason)
        self._load()

    def hide(self, reason: str) -> None:
        """Hide this post.

        :param reason: The reason for hiding this post.
        """
        self.site.hide_post(self, reason)
        self._load()

    def suppress(self, reason: str) -> None:
        """Suppress this post.

        :param reason: The reason for suppressing this post.
        """
        self.site.suppress_post(self, reason)
        self._load()

    def restore(self, reason: str) -> None:
        """Restore this post.

        :param reason: The reason for restoring this post.
        """
        self.site.restore_post(self, reason)
        self._load()

    def thank(self) -> None:
        """Thank the user who made this post."""
        self.site.thank_post(self)