core/domain/topic_fetchers.py - develop - githubmirror/oppia

core/domain/topic_fetchers.py
832 строки26 KB
G.Santhosh Reddy
Fix #24470: Learner dashboard goals counting draft chapters (#24509)
21 янв 2026, 19:27
21 янв 2026, 19:273c0a1a0
# coding: utf-8
#
# Copyright 2018 The Oppia Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS-IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.]

"""Getter commands for for topic models."""

from __future__ import annotations

import copy

from core import feconf, utils
from core.domain import (
    caching_services,
    classroom_config_services,
    story_domain,
    story_fetchers,
    topic_domain,
)
from core.platform import models

from typing import (
    Dict,
    List,
    Literal,
    Optional,
    Sequence,
    Set,
    Tuple,
    TypedDict,
    overload,
)

MYPY = False
if MYPY:  # pragma: no cover
    from mypy_imports import topic_models

(topic_models,) = models.Registry.import_models([models.Names.TOPIC])


def _migrate_subtopics_to_latest_schema(
    versioned_subtopics: topic_domain.VersionedSubtopicsDict, topic_id: str
) -> None:
    """Holds the responsibility of performing a step-by-step, sequential update
    of the subtopics structure based on the schema version of the input
    subtopics dictionary. If the current subtopics schema changes, a
    new conversion function must be added and some code appended to this
    function to account for that new version.

    Args:
        versioned_subtopics: dict. A dict with two keys:
          - schema_version: int. The schema version for the subtopics dict.
          - subtopics: list(dict). The list of dicts comprising the topic's
              subtopics.
        topic_id: str. The id of the topic to which the subtopics are part of.

    Raises:
        Exception. The schema version of subtopics is outside of what
            is supported at present.
    """
    subtopic_schema_version = versioned_subtopics['schema_version']
    if not (
        1 <= subtopic_schema_version <= feconf.CURRENT_SUBTOPIC_SCHEMA_VERSION
    ):
        raise Exception(
            'Sorry, we can only process v1-v%d subtopic schemas at '
            'present.' % feconf.CURRENT_SUBTOPIC_SCHEMA_VERSION
        )

    while subtopic_schema_version < feconf.CURRENT_SUBTOPIC_SCHEMA_VERSION:
        topic_domain.Topic.update_subtopics_from_model(
            versioned_subtopics, subtopic_schema_version, topic_id
        )
        subtopic_schema_version += 1


def _migrate_story_references_to_latest_schema(
    versioned_story_references: topic_domain.VersionedStoryReferencesDict,
) -> None:
    """Holds the responsibility of performing a step-by-step, sequential update
    of the story reference structure based on the schema version of the input
    story reference dictionary. If the current story reference schema changes, a
    new conversion function must be added and some code appended to this
    function to account for that new version.

    Args:
        versioned_story_references: dict. A dict with two keys:
          - schema_version: int. The schema version for the story reference
                dict.
          - story_references: list(dict). The list of dicts comprising the
                topic's story references.

    Raises:
        Exception. The schema version of story_references is outside of what
            is supported at present.
    """
    story_reference_schema_version = versioned_story_references[
        'schema_version'
    ]
    if not (
        1
        <= story_reference_schema_version
        <= feconf.CURRENT_STORY_REFERENCE_SCHEMA_VERSION
    ):
        raise Exception(
            'Sorry, we can only process v1-v%d story reference schemas at '
            'present.' % feconf.CURRENT_STORY_REFERENCE_SCHEMA_VERSION
        )

    while (
        story_reference_schema_version
        < feconf.CURRENT_STORY_REFERENCE_SCHEMA_VERSION
    ):
        topic_domain.Topic.update_story_references_from_model(
            versioned_story_references, story_reference_schema_version
        )
        story_reference_schema_version += 1


def get_topic_from_model(
    topic_model: topic_models.TopicModel,
) -> topic_domain.Topic:
    """Returns a topic domain object given a topic model loaded
    from the datastore.

    Args:
        topic_model: TopicModel. The topic model loaded from the
            datastore.

    Returns:
        topic. A Topic domain object corresponding to the given
        topic model.
    """
    versioned_subtopics: topic_domain.VersionedSubtopicsDict = {
        'schema_version': topic_model.subtopic_schema_version,
        'subtopics': copy.deepcopy(topic_model.subtopics),
    }
    versioned_canonical_story_references: (
        topic_domain.VersionedStoryReferencesDict
    ) = {
        'schema_version': topic_model.story_reference_schema_version,
        'story_references': topic_model.canonical_story_references,
    }
    versioned_additional_story_references: (
        topic_domain.VersionedStoryReferencesDict
    ) = {
        'schema_version': topic_model.story_reference_schema_version,
        'story_references': topic_model.additional_story_references,
    }
    if (
        topic_model.subtopic_schema_version
        != feconf.CURRENT_SUBTOPIC_SCHEMA_VERSION
    ):
        _migrate_subtopics_to_latest_schema(versioned_subtopics, topic_model.id)
    if (
        topic_model.story_reference_schema_version
        != feconf.CURRENT_STORY_REFERENCE_SCHEMA_VERSION
    ):
        _migrate_story_references_to_latest_schema(
            versioned_canonical_story_references
        )
        _migrate_story_references_to_latest_schema(
            versioned_additional_story_references
        )
    return topic_domain.Topic(
        topic_model.id,
        topic_model.name,
        topic_model.abbreviated_name,
        topic_model.url_fragment,
        topic_model.thumbnail_filename,
        topic_model.thumbnail_bg_color,
        topic_model.thumbnail_size_in_bytes,
        topic_model.description,
        [
            topic_domain.StoryReference.from_dict(reference)
            for reference in versioned_canonical_story_references[
                'story_references'
            ]
        ],
        [
            topic_domain.StoryReference.from_dict(reference)
            for reference in versioned_additional_story_references[
                'story_references'
            ]
        ],
        topic_model.uncategorized_skill_ids,
        [
            topic_domain.Subtopic.from_dict(subtopic)
            for subtopic in versioned_subtopics['subtopics']
        ],
        versioned_subtopics['schema_version'],
        topic_model.next_subtopic_id,
        topic_model.language_code,
        topic_model.version,
        feconf.CURRENT_STORY_REFERENCE_SCHEMA_VERSION,
        topic_model.meta_tag_content,
        topic_model.practice_tab_is_displayed,
        topic_model.page_title_fragment_for_web,
        topic_model.skill_ids_for_diagnostic_test,
        topic_model.created_on,
        topic_model.last_updated,
    )


@overload
def get_topic_by_id(topic_id: str) -> topic_domain.Topic: ...


@overload
def get_topic_by_id(
    topic_id: str, *, version: Optional[int] = None
) -> topic_domain.Topic: ...


@overload
def get_topic_by_id(
    topic_id: str, *, strict: Literal[True], version: Optional[int] = None
) -> topic_domain.Topic: ...


@overload
def get_topic_by_id(
    topic_id: str, *, strict: Literal[False], version: Optional[int] = None
) -> Optional[topic_domain.Topic]: ...


def get_topic_by_id(
    topic_id: str, strict: bool = True, version: Optional[int] = None
) -> Optional[topic_domain.Topic]:
    """Returns a domain object representing a topic.

    Args:
        topic_id: str. ID of the topic.
        strict: bool. Whether to fail noisily if no topic with the given
            id exists in the datastore.
        version: int or None. The version number of the topic to be
            retrieved. If it is None, the latest version will be retrieved.

    Returns:
        Topic or None. The domain object representing a topic with the
        given id, or None if it does not exist.
    """
    sub_namespace: Optional[str] = str(version) if version else None
    cached_topic = caching_services.get_multi(
        caching_services.CACHE_NAMESPACE_TOPIC, sub_namespace, [topic_id]
    ).get(topic_id)

    if cached_topic is not None:
        return cached_topic
    else:
        topic_model = topic_models.TopicModel.get(
            topic_id, strict=strict, version=version
        )
        if topic_model:
            topic = get_topic_from_model(topic_model)
            caching_services.set_multi(
                caching_services.CACHE_NAMESPACE_TOPIC,
                sub_namespace,
                {topic_id: topic},
            )
            return topic
        else:
            return None


@overload
def get_topics_by_ids(
    topic_ids: List[str], *, strict: Literal[True]
) -> List[topic_domain.Topic]: ...


@overload
def get_topics_by_ids(
    topic_ids: List[str],
) -> List[Optional[topic_domain.Topic]]: ...


@overload
def get_topics_by_ids(
    topic_ids: List[str], *, strict: Literal[False]
) -> List[Optional[topic_domain.Topic]]: ...


def get_topics_by_ids(
    topic_ids: List[str], strict: bool = False
) -> Sequence[Optional[topic_domain.Topic]]:
    """Returns a list of topics matching the IDs provided.

    Args:
        topic_ids: list(str). List of IDs to get topics for.
        strict: bool. Whether to fail noisily if no topic model exists
            with a given ID exists in the datastore.

    Returns:
        list(Topic|None). The list of topics corresponding to given ids
        (with None in place of topic ids corresponding to deleted topics).

    Raises:
        Exception. No topic model exists for the given topic_id.
    """
    all_topic_models: List[Optional[topic_models.TopicModel]] = (
        topic_models.TopicModel.get_multi(topic_ids)
    )
    topics: List[Optional[topic_domain.Topic]] = []
    for index, topic_model in enumerate(all_topic_models):
        if topic_model is None:
            if strict:
                raise Exception(
                    'No topic model exists for the topic_id: %s'
                    % topic_ids[index]
                )
            topics.append(topic_model)
        if topic_model is not None:
            topics.append(get_topic_from_model(topic_model))
    return topics


def get_multiple_topics_by_ids_and_version(
    topic_ids_and_versions: List[Tuple[str, Optional[int]]],
) -> List[Optional[topic_domain.Topic]]:
    """Returns a list of topics matching the IDs and versions provided.

    Args:
        topic_ids_and_versions: list(tuple(str, int|None)). List of tuples of
            topic ID and version number. If version number is None, the latest
            version will be returned.

    Returns:
        list(Topic|None). The list of topics corresponding to the given IDs
        and versions. If a topic does not exist, the corresponding entry will
        be None.
    """
    topic_models_list = topic_models.TopicModel.get_version_multi(
        topic_ids_and_versions
    )
    return [
        get_topic_from_model(topic_model) if topic_model is not None else None
        for topic_model in topic_models_list
    ]


@overload
def get_topic_by_name(
    topic_name: str, *, strict: Literal[True] = ...
) -> topic_domain.Topic: ...


@overload
def get_topic_by_name(
    topic_name: str, *, strict: Literal[False] = ...
) -> Optional[topic_domain.Topic]: ...


def get_topic_by_name(
    topic_name: str, strict: bool = False
) -> Optional[topic_domain.Topic]:
    """Returns a domain object representing a topic.

    Args:
        topic_name: str. The name of the topic.
        strict: bool. Whether to fail noisily if no Topic exists for
            the given topic name.

    Returns:
        Topic or None. The domain object representing a topic with the
        given id, or None if it does not exist.

    Raises:
        Exception. No Topic exists for the given topic name.
    """
    topic_model: Optional[topic_models.TopicModel] = (
        topic_models.TopicModel.get_by_name(topic_name)
    )
    if topic_model is None:
        if strict:
            raise Exception(
                'No Topic exists for the given topic name: %s' % topic_name
            )
        return None

    return get_topic_from_model(topic_model)


def get_topic_by_url_fragment(
    url_fragment: str,
) -> Optional[topic_domain.Topic]:
    """Returns a domain object representing a topic.

    Args:
        url_fragment: str. The url fragment of the topic.

    Returns:
        Topic or None. The domain object representing a topic with the
        given id, or None if it does not exist.
    """
    topic_model: Optional[topic_models.TopicModel] = (
        topic_models.TopicModel.get_by_url_fragment(url_fragment)
    )
    if topic_model is None:
        return None

    return get_topic_from_model(topic_model)


def get_all_topics() -> List[topic_domain.Topic]:
    """Returns all the topics present in the datastore.

    Returns:
        list(Topic). The list of topics present in the datastore.
    """
    backend_topic_models = topic_models.TopicModel.get_all()
    topics: List[topic_domain.Topic] = [
        get_topic_from_model(topic) for topic in backend_topic_models
    ]
    return topics


@overload
def get_topic_rights(topic_id: str) -> topic_domain.TopicRights: ...


@overload
def get_topic_rights(
    topic_id: str, *, strict: Literal[True]
) -> topic_domain.TopicRights: ...


@overload
def get_topic_rights(
    topic_id: str, *, strict: Literal[False]
) -> Optional[topic_domain.TopicRights]: ...


def get_topic_rights(
    topic_id: str, strict: bool = True
) -> Optional[topic_domain.TopicRights]:
    """Retrieves the rights object for the given topic.

    Args:
        topic_id: str. ID of the topic.
        strict: bool. Whether to fail noisily if no topic with a given id
            exists in the datastore.

    Returns:
        TopicRights or None. The rights object associated with the given topic,
        or None if it does not exist.

    Raises:
        EntityNotFoundError. The TopicRights with ID topic_id was not
            found in the datastore.
    """

    model: Optional[topic_models.TopicRightsModel] = (
        topic_models.TopicRightsModel.get(topic_id, strict=strict)
    )

    if model is None:
        return None

    return get_topic_rights_from_model(model)


def get_topic_rights_from_model(
    topic_rights_model: topic_models.TopicRightsModel,
) -> topic_domain.TopicRights:
    """Constructs a TopicRights object from the given topic rights model.

    Args:
        topic_rights_model: TopicRightsModel. Topic rights from the
            datastore.

    Returns:
        TopicRights. The rights object created from the model.
    """

    return topic_domain.TopicRights(
        topic_rights_model.id,
        topic_rights_model.manager_ids,
        topic_rights_model.topic_is_published,
    )


def get_all_topic_summaries() -> List[topic_domain.TopicSummary]:
    """Returns the summaries of all topics present in the datastore.

    Returns:
        list(TopicSummary). The list of summaries of all topics present in the
        datastore.
    """
    topic_summaries_models = topic_models.TopicSummaryModel.get_all()
    topic_summaries: List[topic_domain.TopicSummary] = [
        get_topic_summary_from_model(summary)
        for summary in topic_summaries_models
    ]
    return topic_summaries


def get_multi_topic_summaries(
    topic_ids: List[str],
) -> List[Optional[topic_domain.TopicSummary]]:
    """Returns the summaries of all topics whose topic ids are passed in.

    Args:
        topic_ids: list(str). The IDs of topics for which summaries are to be
            returned.

    Returns:
        list(TopicSummary) or None. The list of summaries of all given topics
        present in the datastore, or None if it does not exist.
    """
    topic_summaries_models = topic_models.TopicSummaryModel.get_multi(topic_ids)
    topic_summaries = [
        get_topic_summary_from_model(summary) if summary else None
        for summary in topic_summaries_models
    ]
    return topic_summaries


def get_published_topic_summaries() -> List[topic_domain.TopicSummary]:
    """Returns the summaries of all published topics present in the datastore.

    Returns:
        list(TopicSummary). The list of summaries of all published topics
        present in the datastore.
    """
    topic_id_to_topic_rights = get_all_topic_rights()
    published_topic_ids = [
        topic_id
        for topic_id, topic_rights in topic_id_to_topic_rights.items()
        if topic_rights.topic_is_published
    ]
    topic_summaries_list = [
        topic_summary
        for topic_summary in get_multi_topic_summaries(published_topic_ids)
        if topic_summary is not None
    ]
    return topic_summaries_list


def get_all_skill_ids_assigned_to_some_topic() -> Set[str]:
    """Returns the ids of all the skills that are linked to some topics.

    Returns:
        set([str]). The ids of all the skills linked to some topic.
    """
    skill_ids: Set[str] = set()
    all_topic_models = topic_models.TopicModel.get_all()
    all_topics: List[topic_domain.Topic] = [
        get_topic_from_model(topic) for topic in all_topic_models
    ]
    for topic in all_topics:
        skill_ids.update(topic.get_all_skill_ids())
    return skill_ids


def get_topic_summary_from_model(
    topic_summary_model: topic_models.TopicSummaryModel,
) -> topic_domain.TopicSummary:
    """Returns a domain object for an Oppia topic summary given a
    topic summary model.

    Args:
        topic_summary_model: TopicSummaryModel. The topic summary model object
            to get the corresponding domain object.

    Returns:
        TopicSummary. The domain object corresponding to the given model object.
    """
    return topic_domain.TopicSummary(
        topic_summary_model.id,
        topic_summary_model.name,
        topic_summary_model.canonical_name,
        topic_summary_model.language_code,
        topic_summary_model.description,
        topic_summary_model.version,
        topic_summary_model.canonical_story_count,
        topic_summary_model.additional_story_count,
        topic_summary_model.uncategorized_skill_count,
        topic_summary_model.subtopic_count,
        topic_summary_model.total_skill_count,
        topic_summary_model.total_published_node_count,
        topic_summary_model.thumbnail_filename,
        topic_summary_model.thumbnail_bg_color,
        topic_summary_model.url_fragment,
        topic_summary_model.published_story_exploration_mapping,
        topic_summary_model.topic_model_created_on,
        topic_summary_model.topic_model_last_updated,
    )


@overload
def get_topic_summary_by_id(topic_id: str) -> topic_domain.TopicSummary: ...


@overload
def get_topic_summary_by_id(
    topic_id: str, *, strict: Literal[True]
) -> topic_domain.TopicSummary: ...


@overload
def get_topic_summary_by_id(
    topic_id: str, *, strict: Literal[False]
) -> Optional[topic_domain.TopicSummary]: ...


def get_topic_summary_by_id(
    topic_id: str, strict: bool = True
) -> Optional[topic_domain.TopicSummary]:
    """Returns a domain object representing a topic summary.

    Args:
        topic_id: str. ID of the topic summary.
        strict: bool. Whether to fail noisily if no topic summary with the given
            id exist in the datastore.

    Returns:
        TopicSummary or None. The topic summary domain object corresponding to
        a topic with the given topic_id, if it exists, or else None.
    """
    topic_summary_model: Optional[topic_models.TopicSummaryModel] = (
        topic_models.TopicSummaryModel.get(topic_id, strict=strict)
    )
    if topic_summary_model:
        topic_summary: topic_domain.TopicSummary = get_topic_summary_from_model(
            topic_summary_model
        )
        return topic_summary
    else:
        return None


def get_new_topic_id() -> str:
    """Returns a new topic id.

    Returns:
        str. A new topic id.
    """
    return topic_models.TopicModel.get_new_id('')


@overload
def get_multi_topic_rights(
    topic_ids: List[str], *, strict: Literal[True]
) -> List[topic_domain.TopicRights]: ...


@overload
def get_multi_topic_rights(
    topic_ids: List[str],
) -> List[Optional[topic_domain.TopicRights]]: ...


@overload
def get_multi_topic_rights(
    topic_ids: List[str], *, strict: Literal[False]
) -> List[Optional[topic_domain.TopicRights]]: ...


def get_multi_topic_rights(
    topic_ids: List[str], strict: bool = False
) -> Sequence[Optional[topic_domain.TopicRights]]:
    """Returns the rights of all topics whose topic ids are passed in.

    Args:
        topic_ids: list(str). The IDs of topics for which rights are to be
            returned.
        strict: bool. Whether to fail noisily if no TopicRights exists for
            the given topic id.

    Returns:
        Sequence[Optional[TopicRights]]. The list of rights of all given topics
        present in the datastore.

    Raises:
        Exception. No topic_rights exists for the given topic_id.
    """
    topic_rights_models: List[Optional[topic_models.TopicRightsModel]] = (
        topic_models.TopicRightsModel.get_multi(topic_ids)
    )
    topic_rights: List[Optional[topic_domain.TopicRights]] = []
    for index, rights in enumerate(topic_rights_models):
        if rights is None:
            if strict:
                raise Exception(
                    'No topic_rights exists for the given topic_id: %s'
                    % topic_ids[index]
                )
            topic_rights.append(rights)
        else:
            topic_rights.append(get_topic_rights_from_model(rights))
    return topic_rights


def get_topic_rights_with_user(user_id: str) -> List[topic_domain.TopicRights]:
    """Retrieves the rights object for all topics assigned to given user.

    Args:
        user_id: str. ID of the user.

    Returns:
        list(TopicRights). The rights objects associated with the topics
        assigned to given user.
    """
    topic_rights_models: Sequence[topic_models.TopicRightsModel] = (
        topic_models.TopicRightsModel.get_by_user(user_id)
    )
    return [
        get_topic_rights_from_model(model)
        for model in topic_rights_models
        if model is not None
    ]


def get_all_topic_rights() -> Dict[str, topic_domain.TopicRights]:
    """Returns the rights object of all topics present in the datastore.

    Returns:
        dict. The dict of rights objects of all topics present in the datastore
        keyed by topic id.
    """
    topic_rights_models = topic_models.TopicRightsModel.get_all()
    topic_rights: Dict[str, topic_domain.TopicRights] = {}
    for model in topic_rights_models:
        rights: topic_domain.TopicRights = get_topic_rights_from_model(model)
        topic_rights[rights.id] = rights
    return topic_rights


class CannonicalStoryDict(TypedDict):
    """Dictionary that represents cannonical stories."""

    id: str
    title: str
    description: str
    node_titles: List[str]
    thumbnail_bg_color: Optional[str]
    thumbnail_filename: Optional[str]
    url_fragment: str
    topic_url_fragment: str
    classroom_url_fragment: Optional[str]
    classroom_name: Optional[str]
    story_is_published: bool
    completed_node_titles: List[str]
    all_node_dicts: List[story_domain.StoryNodeDict]


def get_canonical_story_dicts(
    user_id: str, topic: topic_domain.Topic
) -> List[CannonicalStoryDict]:
    """Returns a list of canonical story dicts in the topic.

    Args:
        user_id: str. The ID of the user.
        topic: Topic. The topic domain object.

    Returns:
        list(dict). A list of canonical story dicts in the given topic.
    """
    canonical_story_ids: List[str] = topic.get_canonical_story_ids(
        include_only_published=True
    )
    canonical_story_summaries: List[story_domain.StorySummary] = [
        story_fetchers.get_story_summary_by_id(canonical_story_id)
        for canonical_story_id in canonical_story_ids
    ]
    canonical_story_dicts = []
    for story_summary in canonical_story_summaries:
        pending_and_all_nodes_in_story = (
            story_fetchers.get_pending_and_all_nodes_in_story(
                user_id, story_summary.id
            )
        )
        all_nodes = pending_and_all_nodes_in_story['all_nodes']
        pending_nodes = pending_and_all_nodes_in_story['pending_nodes']
        pending_node_titles = [node.title for node in pending_nodes]
        completed_node_titles = utils.compute_list_difference(
            story_summary.node_titles, pending_node_titles
        )
        # Here we use MyPy ignore because the return type of
        # 'to_human_readable_dict()' method is HumanReadableStorySummaryDict
        # which do not contain topic_url_fragment, story_is_published and
        # other keys. To overcome this missing keys issue, we have defined
        # a CannonicalStoryDict and assigned it to the `story_summary_dict`.
        # So, due to this a conflict in type assignment is raised which cause
        # MyPy to throw `Incompatible types in assignment` error. Thus, to
        # avoid the error, we used ignore here.
        story_summary_dict: CannonicalStoryDict = (
            story_summary.to_human_readable_dict()  # type: ignore[assignment]
        )
        story_summary_dict['node_titles'] = [node.title for node in all_nodes]
        story_summary_dict['topic_url_fragment'] = topic.url_fragment
        story_summary_dict['classroom_url_fragment'] = (
            classroom_config_services.get_classroom_url_fragment_for_topic_id(
                topic.id
            )
        )
        story_summary_dict['classroom_name'] = (
            classroom_config_services.get_classroom_name_for_topic_id(topic.id)
        )
        story_summary_dict['story_is_published'] = True
        story_summary_dict['completed_node_titles'] = completed_node_titles
        story_summary_dict['all_node_dicts'] = [
            node.to_dict() for node in all_nodes
        ]
        canonical_story_dicts.append(story_summary_dict)

    return canonical_story_dicts


def get_story_ids_linked_to_topic(topic_id: str) -> List[str]:
    """Returns the list of all canonical story IDs linked to a given topic.

    Args:
        topic_id: str. The ID of the topic.

    Returns:
        list(str). The list of canonical story IDs linked to the topic.
    """
    topic = get_topic_by_id(topic_id, strict=True)
    return topic.get_canonical_story_ids()