Skip to content
89 changes: 86 additions & 3 deletions openedx/core/djangoapps/content_libraries/api/block_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,19 +4,24 @@
from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime

from django.utils.translation import gettext as _
from django.contrib.auth import get_user_model
from opaque_keys.edx.locator import LibraryUsageLocatorV2

from .libraries import PublishableItem, library_component_usage_key
from openedx.core.djangoapps.user_api.accounts.image_helpers import get_profile_image_urls_for_user
from .libraries import library_component_usage_key, PublishableItem

# The public API is only the following symbols:
__all__ = [
"LibraryXBlockMetadata",
"LibraryXBlockStaticFile",
"LibraryHistoryEntry",
"LibraryHistoryContributor",
"LibraryPublishHistoryGroup",
]


User = get_user_model()
@dataclass(frozen=True, kw_only=True)
class LibraryXBlockMetadata(PublishableItem):
"""
Expand Down Expand Up @@ -48,6 +53,7 @@ def from_component(cls, library_key, component, associated_collections=None):
usage_key=usage_key,
display_name=draft.title,
created=component.created,
created_by=component.created_by.username if component.created_by else None,
modified=draft.created,
draft_version_num=draft.version_num,
published_version_num=published.version_num if published else None,
Expand All @@ -63,6 +69,53 @@ def from_component(cls, library_key, component, associated_collections=None):
)


@dataclass(frozen=True)
class LibraryHistoryEntry:
"""
One entry in the history of a library component.
"""
changed_by: LibraryHistoryContributor | None
changed_at: datetime
title: str # title at time of change
block_type: str | None
action: str # "edited" | "renamed"


@dataclass(frozen=True)
class LibraryHistoryContributor:
"""
A contributor in a publish history group, with profile image URLs.
"""
username: str
profile_image_urls: dict # {"full": str, "large": str, "medium": str, "small": str}

@classmethod
def from_user(cls, user, request=None) -> 'LibraryHistoryContributor':
return cls(
username=user.username,
profile_image_urls=get_profile_image_urls_for_user(user, request),
)


@dataclass(frozen=True)
class LibraryPublishHistoryGroup:
"""
Summary of a publish event for a library item.

Each instance represents one PublishLogRecord for the item, and
includes the set of contributors who authored draft changes between the
previous publish and this one.
"""
publish_log_uuid: str
published_by: object # AUTH_USER_MODEL instance or None
published_at: datetime
title: str # title at time of publish
block_type: str | None
contributors: list[LibraryHistoryContributor] # distinct authors of versions in this group
contributors_count: int
entity_key: str # str(usage_key) for components, str(container_key) for containers


@dataclass(frozen=True)
class LibraryXBlockStaticFile:
"""
Expand All @@ -76,3 +129,33 @@ class LibraryXBlockStaticFile:
url: str
# Size in bytes
size: int


def resolve_contributors(users, request=None) -> list[LibraryHistoryContributor | None]:
"""
Convert an iterable of User objects (possibly containing None) to a list of
LibraryHistoryContributor.
"""
users_list = list(users)
user_pks = list({user.pk for user in users_list if user is not None})
prefetched = {
user.pk: user
for user in User.objects.filter(pk__in=user_pks).select_related('profile')
} if user_pks else {}
return [
LibraryHistoryContributor.from_user(prefetched.get(user.pk, user), request)
if user else None
for user in users_list
]


def resolve_change_action(old_version, new_version) -> str:
"""
Derive a human-readable action label from a draft history record's versions.

Returns "renamed" when both versions exist and the title changed between
them; otherwise returns "edited" as the default action.
"""
if old_version and new_version and old_version.title != new_version.title:
return "renamed"
return "edited"
193 changes: 184 additions & 9 deletions openedx/core/djangoapps/content_libraries/api/blocks.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,23 +50,30 @@

from .. import tasks
from ..models import ContentLibrary
from .block_metadata import LibraryXBlockMetadata, LibraryXBlockStaticFile
from .collections import library_collection_locator
from .container_metadata import container_subclass_for_olx_tag
from .containers import (
ContainerMetadata,
create_container,
get_container,
get_containers_contains_item,
update_container_children,
)
from .exceptions import (
BlockLimitReachedError,
ContentLibraryBlockNotFound,
IncompatibleTypesError,
InvalidNameError,
LibraryBlockAlreadyExists,
)
from .block_metadata import (
LibraryHistoryEntry,
LibraryPublishHistoryGroup,
LibraryXBlockMetadata,
LibraryXBlockStaticFile,
resolve_contributors,
resolve_change_action,
)
from .containers import (
create_container,
get_container,
get_containers_contains_item,
update_container_children,
ContainerMetadata,
)
from .libraries import PublishableItem

# This content_libraries API is sometimes imported in the LMS (should we prevent that?), but the content_staging app
Expand All @@ -77,6 +84,7 @@

log = logging.getLogger(__name__)


# The public API is only the following symbols:
__all__ = [
# API methods
Expand All @@ -96,6 +104,10 @@
"add_library_block_static_asset_file",
"delete_library_block_static_asset_file",
"publish_component_changes",
"get_library_component_draft_history",
"get_library_component_publish_history",
"get_library_component_publish_history_entries",
"get_library_component_creation_entry",
]


Expand Down Expand Up @@ -189,6 +201,170 @@ def get_library_block(usage_key: LibraryUsageLocatorV2, include_collections=Fals
return xblock_metadata


def get_library_component_draft_history(
usage_key: LibraryUsageLocatorV2,
request=None,
) -> list[LibraryHistoryEntry]:
"""
Return the draft change history for a library component since its last publication,
ordered from most recent to oldest.

Raises ContentLibraryBlockNotFound if the component does not exist.
"""
try:
component = get_component_from_usage_key(usage_key)
except ObjectDoesNotExist as exc:
raise ContentLibraryBlockNotFound(usage_key) from exc

records = list(content_api.get_entity_draft_history(component.publishable_entity))
changed_by_list = resolve_contributors(
(record.draft_change_log.changed_by for record in records), request
)

entries = []
for record, changed_by in zip(records, changed_by_list):
version = record.new_version if record.new_version is not None else record.old_version
entries.append(LibraryHistoryEntry(
changed_by=changed_by,
changed_at=record.draft_change_log.changed_at,
title=version.title if version is not None else "",
block_type=record.entity.component.component_type.name,
action=resolve_change_action(record.old_version, record.new_version),
))
return entries


def get_library_component_publish_history(
usage_key: LibraryUsageLocatorV2,
request=None,
) -> list[LibraryPublishHistoryGroup]:
"""
Return the publish history of a library component as a list of groups.

Each group corresponds to one publish event (PublishLogRecord) and includes:
- who published and when
- the distinct set of contributors: users who authored draft changes between
the previous publish and this one (via DraftChangeLogRecord version bounds)

Groups are ordered most-recent-first. Returns [] if the component has never
been published.

Contributors are resolved using version bounds (old_version_num → new_version_num)
rather than timestamps to avoid clock-skew issues. old_version_num defaults to
0 for the very first publish. new_version_num is None for soft-delete publishes
(no PublishableEntityVersion is created on soft delete).
"""
try:
component = get_component_from_usage_key(usage_key)
except ObjectDoesNotExist as exc:
raise ContentLibraryBlockNotFound(usage_key) from exc

entity = component.publishable_entity
publish_records = list(content_api.get_entity_publish_history(entity))

groups = []
for pub_record in publish_records:
# old_version is None only for the very first publish (entity had no prior published version)
old_version_num = pub_record.old_version.version_num if pub_record.old_version else 0
# new_version is None for soft-delete publishes (component deleted without a new draft version)
new_version_num = pub_record.new_version.version_num if pub_record.new_version else None

raw_contributors = list(content_api.get_entity_version_contributors(
entity,
old_version_num=old_version_num,
new_version_num=new_version_num,
))

contributors = [c for c in resolve_contributors(raw_contributors, request) if c is not None]

groups.append(LibraryPublishHistoryGroup(
publish_log_uuid=str(pub_record.publish_log.uuid),
published_by=pub_record.publish_log.published_by,
published_at=pub_record.publish_log.published_at,
contributors=contributors,
title=pub_record.new_version.title if pub_record.new_version else "",
block_type=pub_record.entity.component.component_type.name,
contributors_count=len(contributors),
entity_key=str(usage_key),
))

return groups


def get_library_component_publish_history_entries(
usage_key: LibraryUsageLocatorV2,
publish_log_uuid: str,
request=None,
) -> list[LibraryHistoryEntry]:
"""
Return the individual draft change entries for a specific publish event.

Called lazily when the user expands a publish event in the UI. Entries are
the DraftChangeLogRecords that fall between the previous publish event and
this one, ordered most-recent-first.
"""
try:
component = get_component_from_usage_key(usage_key)
except ObjectDoesNotExist as exc:
raise ContentLibraryBlockNotFound(usage_key) from exc

records = list(content_api.get_entity_publish_history_entries(
component.publishable_entity, publish_log_uuid
))
changed_by_list = resolve_contributors(
(record.draft_change_log.changed_by for record in records), request
)

entries = []
for record, changed_by in zip(records, changed_by_list):
version = record.new_version if record.new_version is not None else record.old_version
entries.append(LibraryHistoryEntry(
changed_by=changed_by,
changed_at=record.draft_change_log.changed_at,
title=version.title if version is not None else "",
block_type=record.entity.component.component_type.name,
action=resolve_change_action(record.old_version, record.new_version),
))
return entries


def get_library_component_creation_entry(
usage_key: LibraryUsageLocatorV2,
request=None,
) -> LibraryHistoryEntry | None:
"""
Return the creation entry for a library component.

This is a single LibraryHistoryEntry representing the moment the
component was first created (version_num=1). Returns None if the component
has no versions yet.

Raises ContentLibraryBlockNotFound if the component does not exist.
"""
try:
component = get_component_from_usage_key(usage_key)
except ObjectDoesNotExist as exc:
raise ContentLibraryBlockNotFound(usage_key) from exc

first_version = (
component.publishable_entity.versions
.filter(version_num=1)
.select_related("created_by")
.first()
)
if first_version is None:
return None

changed_by_list = resolve_contributors([first_version.created_by], request)
return LibraryHistoryEntry(
changed_by=changed_by_list[0],
changed_at=first_version.created,
title=first_version.title,
block_type=component.component_type.name,
action="created",
)


def set_library_block_olx(usage_key: LibraryUsageLocatorV2, new_olx_str: str) -> ComponentVersion:
"""
Replace the OLX source of the given XBlock.
Expand Down Expand Up @@ -681,7 +857,6 @@ def import_staged_content_from_user_clipboard(library_key: LibraryLocatorV2, use
now,
)


def get_or_create_olx_media_type(block_type: str) -> MediaType:
"""
Get or create a MediaType for the block type.
Expand Down
Loading