Andal.LND/edx-platform
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: xmodule/video_module -> xmodule/video_block

Browse Source
This commit is contained in:
0x29a
2022-10-26 21:13:38 +02:00
committed by Piotr Surowiec
parent 685f1cd679
commit dd9f6936bc
34 changed files with 149 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
xmodule/video_block/__init__.py Normal file
View File

@@ -0,0 +1,8 @@
"""
Container for video block and its utils.
"""
from .bumper_utils import *
from .transcripts_utils import * # lint-amnesty, pylint: disable=redefined-builtin
from .video_block import *
from .video_utils import *

147
xmodule/video_block/bumper_utils.py Normal file
View File

@@ -0,0 +1,147 @@
"""
Utils for video bumper
"""
import copy
import json
import logging
from collections import OrderedDict
from datetime import datetime, timedelta
import pytz
from django.conf import settings
from .video_utils import set_query_parameter
try:
import edxval.api as edxval_api
except ImportError:
edxval_api = None
log = logging.getLogger(__name__)
def get_bumper_settings(video):
"""
Get bumper settings from video instance.
"""
bumper_settings = copy.deepcopy(getattr(video, 'video_bumper', {}))
# clean up /static/ prefix from bumper transcripts
for lang, transcript_url in bumper_settings.get('transcripts', {}).items():
bumper_settings['transcripts'][lang] = transcript_url.replace("/static/", "")
return bumper_settings
def is_bumper_enabled(video):
"""
Check if bumper enabled.
- Feature flag ENABLE_VIDEO_BUMPER should be set to True
- Do not show again button should not be clicked by user.
- Current time minus periodicity must be greater that last time viewed
- edxval_api should be presented
Returns:
bool.
"""
bumper_last_view_date = getattr(video, 'bumper_last_view_date', None)
utc_now = datetime.utcnow().replace(tzinfo=pytz.utc)
periodicity = settings.FEATURES.get('SHOW_BUMPER_PERIODICITY', 0)
has_viewed = any([
video.bumper_do_not_show_again,
(bumper_last_view_date and bumper_last_view_date + timedelta(seconds=periodicity) > utc_now)
])
is_studio = getattr(video.system, "is_author_mode", False)
return bool(
not is_studio and
settings.FEATURES.get('ENABLE_VIDEO_BUMPER') and
get_bumper_settings(video) and
edxval_api and
not has_viewed
)
def bumperize(video):
"""
Populate video with bumper settings, if they are presented.
"""
video.bumper = {
'enabled': False,
'edx_video_id': "",
'transcripts': {},
'metadata': None,
}
if not is_bumper_enabled(video):
return
bumper_settings = get_bumper_settings(video)
try:
video.bumper['edx_video_id'] = bumper_settings['video_id']
video.bumper['transcripts'] = bumper_settings['transcripts']
except (TypeError, KeyError):
log.warning(
"Could not retrieve video bumper information from course settings"
)
return
sources = get_bumper_sources(video)
if not sources:
return
video.bumper.update({
'metadata': bumper_metadata(video, sources),
'enabled': True, # Video poster needs this.
})
def get_bumper_sources(video):
"""
Get bumper sources from edxval.
Returns list of sources.
"""
try:
val_profiles = ["desktop_webm", "desktop_mp4"]
val_video_urls = edxval_api.get_urls_for_profiles(video.bumper['edx_video_id'], val_profiles)
bumper_sources = [url for url in [val_video_urls[p] for p in val_profiles] if url]
except edxval_api.ValInternalError:
# if no bumper sources, nothing will be showed
log.warning(
"Could not retrieve information from VAL for Bumper edx Video ID: %s.", video.bumper['edx_video_id']
)
return []
return bumper_sources
def bumper_metadata(video, sources):
"""
Generate bumper metadata.
"""
transcripts = video.get_transcripts_info(is_bumper=True)
unused_track_url, bumper_transcript_language, bumper_languages = video.get_transcripts_for_student(transcripts)
metadata = OrderedDict({
'saveStateUrl': video.ajax_url + '/save_user_state',
'showCaptions': json.dumps(video.show_captions),
'sources': sources,
'streams': '',
'transcriptLanguage': bumper_transcript_language,
'transcriptLanguages': bumper_languages,
'transcriptTranslationUrl': set_query_parameter(
video.runtime.handler_url(video, 'transcript', 'translation/__lang__').rstrip('/?'), 'is_bumper', 1
),
'transcriptAvailableTranslationsUrl': set_query_parameter(
video.runtime.handler_url(video, 'transcript', 'available_translations').rstrip('/?'), 'is_bumper', 1
),
'publishCompletionUrl': set_query_parameter(
video.runtime.handler_url(video, 'publish_completion', '').rstrip('?'), 'is_bumper', 1
),
})
return metadata

1126
xmodule/video_block/transcripts_utils.py Normal file
View File

@@ -0,0 +1,1126 @@
"""
Utility functions for transcripts.
++++++++++++++++++++++++++++++++++
"""
import copy
import html
import logging
import os
import re
from functools import wraps
import requests
import simplejson as json
from django.conf import settings
from lxml import etree
from opaque_keys.edx.locator import BundleDefinitionLocator
from pysrt import SubRipFile, SubRipItem, SubRipTime
from pysrt.srtexc import Error
from openedx.core.djangolib import blockstore_cache
from openedx.core.lib import blockstore_api
from xmodule.contentstore.content import StaticContent
from xmodule.contentstore.django import contentstore
from xmodule.exceptions import NotFoundError
from .bumper_utils import get_bumper_settings
try:
from edxval import api as edxval_api
except ImportError:
edxval_api = None
log = logging.getLogger(__name__)
NON_EXISTENT_TRANSCRIPT = 'non_existent_dummy_file_name'
class TranscriptException(Exception):
pass
class TranscriptsGenerationException(Exception):
pass
class GetTranscriptsFromYouTubeException(Exception):
pass
class TranscriptsRequestValidationException(Exception):
pass
def exception_decorator(func):
"""
Generate NotFoundError for TranscriptsGenerationException, UnicodeDecodeError.
Args:
`func`: Input function
Returns:
'wrapper': Decorated function
"""
@wraps(func)
def wrapper(*args, **kwds):
try:
return func(*args, **kwds)
except (TranscriptsGenerationException, UnicodeDecodeError) as ex:
log.exception(str(ex))
raise NotFoundError # lint-amnesty, pylint: disable=raise-missing-from
return wrapper
def generate_subs(speed, source_speed, source_subs):
"""
Generate transcripts from one speed to another speed.
Args:
`speed`: float, for this speed subtitles will be generated,
`source_speed`: float, speed of source_subs
`source_subs`: dict, existing subtitles for speed `source_speed`.
Returns:
`subs`: dict, actual subtitles.
"""
if speed == source_speed:
return source_subs
coefficient = 1.0 * speed / source_speed
subs = {
'start': [
int(round(timestamp * coefficient)) for
timestamp in source_subs['start']
],
'end': [
int(round(timestamp * coefficient)) for
timestamp in source_subs['end']
],
'text': source_subs['text']}
return subs
def save_to_store(content, name, mime_type, location):
"""
Save named content to store by location.
Returns location of saved content.
"""
content_location = Transcript.asset_location(location, name)
content = StaticContent(content_location, name, mime_type, content)
contentstore().save(content)
return content_location
def save_subs_to_store(subs, subs_id, item, language='en'):
"""
Save transcripts into `StaticContent`.
Args:
`subs_id`: str, subtitles id
`item`: video block instance
`language`: two chars str ('uk'), language of translation of transcripts
Returns: location of saved subtitles.
"""
filedata = json.dumps(subs, indent=2).encode('utf-8')
filename = subs_filename(subs_id, language)
return save_to_store(filedata, filename, 'application/json', item.location)
def get_transcript_link_from_youtube(youtube_id):
"""
Get the link for YouTube transcript by parsing the source of the YouTube webpage.
Inside the webpage, the details of the transcripts are located in a JSON object.
After prettifying the object, it looks like:
"captions": {
"playerCaptionsTracklistRenderer": {
"captionTracks": [
{
"baseUrl": "...",
"name": {
"simpleText": "(Japanese in local language)"
},
"vssId": ".ja",
"languageCode": "ja",
"isTranslatable": true
},
{
"baseUrl": "...",
"name": {
"simpleText": "(French in local language)"
},
"vssId": ".fr",
"languageCode": "fr",
"isTranslatable": true
},
{
"baseUrl": "...",
"name": {
"simpleText": "(English in local language)"
},
"vssId": ".en",
"languageCode": "en",
"isTranslatable": true
},
...
],
"audioTracks": [...]
"translationLanguages": ...
},
...
}
So we use a regex to find the captionTracks JavaScript array, and then convert it
to a Python dict and return the link for en caption
"""
youtube_url_base = settings.YOUTUBE['TRANSCRIPTS']['YOUTUBE_URL_BASE']
try:
youtube_html = requests.get(f"{youtube_url_base}{youtube_id}")
caption_re = settings.YOUTUBE['TRANSCRIPTS']['CAPTION_TRACKS_REGEX']
caption_matched = re.search(caption_re, youtube_html.content.decode("utf-8"))
if caption_matched:
caption_tracks = json.loads(f'[{caption_matched.group("caption_tracks")}]')
for caption in caption_tracks:
if "languageCode" in caption.keys() and caption["languageCode"] == "en":
return caption["baseUrl"]
return None
except ConnectionError:
return None
def get_transcripts_from_youtube(youtube_id, settings, i18n, youtube_transcript_name=''): # lint-amnesty, pylint: disable=redefined-outer-name
"""
Gets transcripts from youtube for youtube_id.
Parses only utf-8 encoded transcripts.
Other encodings are not supported at the moment.
Returns (status, transcripts): bool, dict.
"""
_ = i18n.ugettext
utf8_parser = etree.XMLParser(encoding='utf-8')
transcript_link = get_transcript_link_from_youtube(youtube_id)
if not transcript_link:
msg = _("Can't get transcript link from Youtube for {youtube_id}.").format(
youtube_id=youtube_id,
)
raise GetTranscriptsFromYouTubeException(msg)
data = requests.get(transcript_link)
if data.status_code != 200 or not data.text:
msg = _("Can't receive transcripts from Youtube for {youtube_id}. Status code: {status_code}.").format(
youtube_id=youtube_id,
status_code=data.status_code
)
raise GetTranscriptsFromYouTubeException(msg)
sub_starts, sub_ends, sub_texts = [], [], []
xmltree = etree.fromstring(data.content, parser=utf8_parser)
for element in xmltree:
if element.tag == "text":
start = float(element.get("start"))
duration = float(element.get("dur", 0)) # dur is not mandatory
text = element.text
end = start + duration
if text:
# Start and end should be ints representing the millisecond timestamp.
sub_starts.append(int(start * 1000))
sub_ends.append(int((end + 0.0001) * 1000))
sub_texts.append(text.replace('\n', ' '))
return {'start': sub_starts, 'end': sub_ends, 'text': sub_texts}
def download_youtube_subs(youtube_id, video_descriptor, settings): # lint-amnesty, pylint: disable=redefined-outer-name
"""
Download transcripts from Youtube.
Args:
youtube_id: str, actual youtube_id of the video.
video_descriptor: video descriptor instance.
We save transcripts for 1.0 speed, as for other speed conversion is done on front-end.
Returns:
Serialized sjson transcript content, if transcripts were successfully downloaded and saved.
Raises:
GetTranscriptsFromYouTubeException, if fails.
"""
i18n = video_descriptor.runtime.service(video_descriptor, "i18n")
_ = i18n.ugettext
subs = get_transcripts_from_youtube(youtube_id, settings, i18n)
return json.dumps(subs, indent=2)
def remove_subs_from_store(subs_id, item, lang='en'):
"""
Remove from store, if transcripts content exists.
"""
filename = subs_filename(subs_id, lang)
Transcript.delete_asset(item.location, filename)
def generate_subs_from_source(speed_subs, subs_type, subs_filedata, item, language='en'):
"""Generate transcripts from source files (like SubRip format, etc.)
and save them to assets for `item` module.
We expect, that speed of source subs equal to 1
:param speed_subs: dictionary {speed: sub_id, ...}
:param subs_type: type of source subs: "srt", ...
:param subs_filedata:unicode, content of source subs.
:param item: module object.
:param language: str, language of translation of transcripts
:returns: True, if all subs are generated and saved successfully.
"""
_ = item.runtime.service(item, "i18n").ugettext
if subs_type.lower() != 'srt':
raise TranscriptsGenerationException(_("We support only SubRip (*.srt) transcripts format."))
try:
srt_subs_obj = SubRipFile.from_string(subs_filedata)
except Exception as ex:
msg = _("Something wrong with SubRip transcripts file during parsing. Inner message is {error_message}").format(
error_message=str(ex)
)
raise TranscriptsGenerationException(msg) # lint-amnesty, pylint: disable=raise-missing-from
if not srt_subs_obj:
raise TranscriptsGenerationException(_("Something wrong with SubRip transcripts file during parsing."))
sub_starts = []
sub_ends = []
sub_texts = []
for sub in srt_subs_obj:
sub_starts.append(sub.start.ordinal)
sub_ends.append(sub.end.ordinal)
sub_texts.append(sub.text.replace('\n', ' '))
subs = {
'start': sub_starts,
'end': sub_ends,
'text': sub_texts}
for speed, subs_id in speed_subs.items():
save_subs_to_store(
generate_subs(speed, 1, subs),
subs_id,
item,
language
)
return subs
def generate_srt_from_sjson(sjson_subs, speed):
"""Generate transcripts with speed = 1.0 from sjson to SubRip (*.srt).
:param sjson_subs: "sjson" subs.
:param speed: speed of `sjson_subs`.
:returns: "srt" subs.
"""
output = ''
equal_len = len(sjson_subs['start']) == len(sjson_subs['end']) == len(sjson_subs['text'])
if not equal_len:
return output
sjson_speed_1 = generate_subs(speed, 1, sjson_subs)
for i in range(len(sjson_speed_1['start'])):
item = SubRipItem(
index=i,
start=SubRipTime(milliseconds=sjson_speed_1['start'][i]),
end=SubRipTime(milliseconds=sjson_speed_1['end'][i]),
text=sjson_speed_1['text'][i]
)
output += (str(item))
output += '\n'
return output
def generate_sjson_from_srt(srt_subs):
"""
Generate transcripts from sjson to SubRip (*.srt).
Arguments:
srt_subs(SubRip): "SRT" subs object
Returns:
Subs converted to "SJSON" format.
"""
sub_starts = []
sub_ends = []
sub_texts = []
for sub in srt_subs:
sub_starts.append(sub.start.ordinal)
sub_ends.append(sub.end.ordinal)
sub_texts.append(sub.text.replace('\n', ' '))
sjson_subs = {
'start': sub_starts,
'end': sub_ends,
'text': sub_texts
}
return sjson_subs
def copy_or_rename_transcript(new_name, old_name, item, delete_old=False, user=None):
"""
Renames `old_name` transcript file in storage to `new_name`.
If `old_name` is not found in storage, raises `NotFoundError`.
If `delete_old` is True, removes `old_name` files from storage.
"""
filename = f'subs_{old_name}.srt.sjson'
content_location = StaticContent.compute_location(item.location.course_key, filename)
transcripts = contentstore().find(content_location).data.decode('utf-8')
save_subs_to_store(json.loads(transcripts), new_name, item)
item.sub = new_name
item.save_with_metadata(user)
if delete_old:
remove_subs_from_store(old_name, item)
def get_html5_ids(html5_sources):
"""
Helper method to parse out an HTML5 source into the ideas
NOTE: This assumes that '/' are not in the filename
"""
html5_ids = [x.split('/')[-1].rsplit('.', 1)[0] for x in html5_sources]
return html5_ids
def manage_video_subtitles_save(item, user, old_metadata=None, generate_translation=False):
"""
Does some specific things, that can be done only on save.
Video player item has some video fields: HTML5 ones and Youtube one.
If value of `sub` field of `new_item` is cleared, transcripts should be removed.
`item` is video block instance with updated values of fields,
but actually have not been saved to store yet.
`old_metadata` contains old values of XFields.
# 1.
If value of `sub` field of `new_item` is different from values of video fields of `new_item`,
and `new_item.sub` file is present, then code in this function creates copies of
`new_item.sub` file with new names. That names are equal to values of video fields of `new_item`
After that `sub` field of `new_item` is changed to one of values of video fields.
This whole action ensures that after user changes video fields, proper `sub` files, corresponding
to new values of video fields, will be presented in system.
# 2. convert /static/filename.srt to filename.srt in self.transcripts.
(it is done to allow user to enter both /static/filename.srt and filename.srt)
# 3. Generate transcripts translation only when user clicks `save` button, not while switching tabs.
a) delete sjson translation for those languages, which were removed from `item.transcripts`.
Note: we are not deleting old SRT files to give user more flexibility.
b) For all SRT files in`item.transcripts` regenerate new SJSON files.
(To avoid confusing situation if you attempt to correct a translation by uploading
a new version of the SRT file with same name).
"""
_ = item.runtime.service(item, "i18n").ugettext
# # 1.
# html5_ids = get_html5_ids(item.html5_sources)
# # Youtube transcript source should always have a higher priority than html5 sources. Appending
# # `youtube_id_1_0` at the end helps achieve this when we read transcripts list.
# possible_video_id_list = html5_ids + [item.youtube_id_1_0]
# sub_name = item.sub
# for video_id in possible_video_id_list:
# if not video_id:
# continue
# if not sub_name:
# remove_subs_from_store(video_id, item)
# continue
# # copy_or_rename_transcript changes item.sub of module
# try:
# # updates item.sub with `video_id`, if it is successful.
# copy_or_rename_transcript(video_id, sub_name, item, user=user)
# except NotFoundError:
# # subtitles file `sub_name` is not presented in the system. Nothing to copy or rename.
# log.debug(
# "Copying %s file content to %s name is failed, "
# "original file does not exist.",
# sub_name, video_id
# )
# 2.
if generate_translation:
for lang, filename in item.transcripts.items():
item.transcripts[lang] = os.path.split(filename)[-1]
# 3.
if generate_translation:
old_langs = set(old_metadata.get('transcripts', {})) if old_metadata else set()
new_langs = set(item.transcripts)
html5_ids = get_html5_ids(item.html5_sources)
possible_video_id_list = html5_ids + [item.youtube_id_1_0]
for lang in old_langs.difference(new_langs): # 3a
for video_id in possible_video_id_list:
if video_id:
remove_subs_from_store(video_id, item, lang)
reraised_message = ''
for lang in new_langs: # 3b
try:
generate_sjson_for_all_speeds(
item,
item.transcripts[lang],
{speed: subs_id for subs_id, speed in youtube_speed_dict(item).items()},
lang,
)
except TranscriptException as ex: # lint-amnesty, pylint: disable=unused-variable
pass
if reraised_message:
item.save_with_metadata(user)
raise TranscriptException(reraised_message)
def youtube_speed_dict(item):
"""
Returns {speed: youtube_ids, ...} dict for existing youtube_ids
"""
yt_ids = [item.youtube_id_0_75, item.youtube_id_1_0, item.youtube_id_1_25, item.youtube_id_1_5]
yt_speeds = [0.75, 1.00, 1.25, 1.50]
youtube_ids = {p[0]: p[1] for p in zip(yt_ids, yt_speeds) if p[0]}
return youtube_ids
def subs_filename(subs_id, lang='en'):
"""
Generate proper filename for storage.
"""
if lang == 'en':
return f'subs_{subs_id}.srt.sjson'
else:
return f'{lang}_subs_{subs_id}.srt.sjson'
def generate_sjson_for_all_speeds(item, user_filename, result_subs_dict, lang):
"""
Generates sjson from srt for given lang.
`item` is module object.
"""
_ = item.runtime.service(item, "i18n").ugettext
try:
srt_transcripts = contentstore().find(Transcript.asset_location(item.location, user_filename))
except NotFoundError as ex:
raise TranscriptException(_("{exception_message}: Can't find uploaded transcripts: {user_filename}").format( # lint-amnesty, pylint: disable=raise-missing-from
exception_message=str(ex),
user_filename=user_filename
))
if not lang:
lang = item.transcript_language
# Used utf-8-sig encoding type instead of utf-8 to remove BOM(Byte Order Mark), e.g. U+FEFF
generate_subs_from_source(
result_subs_dict,
os.path.splitext(user_filename)[1][1:],
srt_transcripts.data.decode('utf-8-sig'),
item,
lang
)
def get_or_create_sjson(item, transcripts):
"""
Get sjson if already exists, otherwise generate it.
Generate sjson with subs_id name, from user uploaded srt.
Subs_id is extracted from srt filename, which was set by user.
Args:
transcipts (dict): dictionary of (language: file) pairs.
Raises:
TranscriptException: when srt subtitles do not exist,
and exceptions from generate_subs_from_source.
`item` is module object.
"""
user_filename = transcripts[item.transcript_language]
user_subs_id = os.path.splitext(user_filename)[0]
source_subs_id, result_subs_dict = user_subs_id, {1.0: user_subs_id}
try:
sjson_transcript = Transcript.asset(item.location, source_subs_id, item.transcript_language).data
except NotFoundError: # generating sjson from srt
generate_sjson_for_all_speeds(item, user_filename, result_subs_dict, item.transcript_language)
sjson_transcript = Transcript.asset(item.location, source_subs_id, item.transcript_language).data
return sjson_transcript
def get_video_ids_info(edx_video_id, youtube_id_1_0, html5_sources):
"""
Returns list internal or external video ids.
Arguments:
edx_video_id (unicode): edx_video_id
youtube_id_1_0 (unicode): youtube id
html5_sources (list): html5 video ids
Returns:
tuple: external or internal, video ids list
"""
clean = lambda item: item.strip() if isinstance(item, str) else item
external = not bool(clean(edx_video_id))
video_ids = [edx_video_id, youtube_id_1_0] + get_html5_ids(html5_sources)
# video_ids cleanup
video_ids = [item for item in video_ids if bool(clean(item))]
return external, video_ids
def clean_video_id(edx_video_id):
"""
Cleans an edx video ID.
Arguments:
edx_video_id(unicode): edx-val's video identifier
"""
return edx_video_id and edx_video_id.strip()
def get_video_transcript_content(edx_video_id, language_code):
"""
Gets video transcript content, only if the corresponding feature flag is enabled for the given `course_id`.
Arguments:
language_code(unicode): Language code of the requested transcript
edx_video_id(unicode): edx-val's video identifier
Returns:
A dict containing transcript's file name and its sjson content.
"""
transcript = None
edx_video_id = clean_video_id(edx_video_id)
if edxval_api and edx_video_id:
try:
transcript = edxval_api.get_video_transcript_data(edx_video_id, language_code)
except ValueError:
log.exception(
f"Error getting transcript from edx-val id: {edx_video_id}: language code {language_code}"
)
content = '{"start": [1],"end": [2],"text": ["An error occured obtaining the transcript."]}'
transcript = dict(
file_name='error-{edx_video_id}-{language_code}.srt',
content=Transcript.convert(content, 'sjson', 'srt')
)
return transcript
def get_available_transcript_languages(edx_video_id):
"""
Gets available transcript languages for a video.
Arguments:
edx_video_id(unicode): edx-val's video identifier
Returns:
A list containing distinct transcript language codes against all the passed video ids.
"""
available_languages = []
edx_video_id = clean_video_id(edx_video_id)
if edxval_api and edx_video_id:
available_languages = edxval_api.get_available_transcript_languages(video_id=edx_video_id)
return available_languages
def convert_video_transcript(file_name, content, output_format):
"""
Convert video transcript into desired format
Arguments:
file_name: name of transcript file along with its extension
content: transcript content stream
output_format: the format in which transcript will be converted
Returns:
A dict containing the new transcript filename and the content converted into desired format.
"""
name_and_extension = os.path.splitext(file_name)
basename, input_format = name_and_extension[0], name_and_extension[1][1:]
filename = f'{basename}.{output_format}'
converted_transcript = Transcript.convert(content, input_format=input_format, output_format=output_format)
return dict(filename=filename, content=converted_transcript)
class Transcript:
"""
Container for transcript methods.
"""
SRT = 'srt'
TXT = 'txt'
SJSON = 'sjson'
mime_types = {
SRT: 'application/x-subrip; charset=utf-8',
TXT: 'text/plain; charset=utf-8',
SJSON: 'application/json',
}
@staticmethod
def convert(content, input_format, output_format):
"""
Convert transcript `content` from `input_format` to `output_format`.
Accepted input formats: sjson, srt.
Accepted output format: srt, txt, sjson.
Raises:
TranscriptsGenerationException: On parsing the invalid srt content during conversion from srt to sjson.
"""
assert input_format in ('srt', 'sjson')
assert output_format in ('txt', 'srt', 'sjson')
if input_format == output_format:
return content
if input_format == 'srt':
# Standardize content into bytes for later decoding.
if isinstance(content, str):
content = content.encode('utf-8')
if output_format == 'txt':
text = SubRipFile.from_string(content.decode('utf-8')).text
return html.unescape(text)
elif output_format == 'sjson':
try:
srt_subs = SubRipFile.from_string(
# Skip byte order mark(BOM) character
content.decode('utf-8-sig'),
error_handling=SubRipFile.ERROR_RAISE
)
except Error as ex: # Base exception from pysrt
raise TranscriptsGenerationException(str(ex)) from ex
return json.dumps(generate_sjson_from_srt(srt_subs))
if input_format == 'sjson':
# If the JSON file content is bytes, try UTF-8, then Latin-1
if isinstance(content, bytes):
try:
content_str = content.decode('utf-8')
except UnicodeDecodeError:
content_str = content.decode('latin-1')
else:
content_str = content
try:
content_dict = json.loads(content_str)
except ValueError:
truncated = content_str[:100].strip()
log.exception(
f"Failed to convert {input_format} to {output_format} for {repr(truncated)}..."
)
content_dict = {"start": [1], "end": [2], "text": ["An error occured obtaining the transcript."]}
if output_format == 'txt':
text = content_dict['text']
text_without_none = [line if line else '' for line in text]
return html.unescape("\n".join(text_without_none))
elif output_format == 'srt':
return generate_srt_from_sjson(content_dict, speed=1.0)
@staticmethod
def asset(location, subs_id, lang='en', filename=None):
"""
Get asset from contentstore, asset location is built from subs_id and lang.
`location` is module location.
"""
# HACK Warning! this is temporary and will be removed once edx-val take over the
# transcript module and contentstore will only function as fallback until all the
# data is migrated to edx-val. It will be saving a contentstore hit for a hardcoded
# dummy-non-existent-transcript name.
if NON_EXISTENT_TRANSCRIPT in [subs_id, filename]:
raise NotFoundError
asset_filename = subs_filename(subs_id, lang) if not filename else filename
return Transcript.get_asset(location, asset_filename)
@staticmethod
def get_asset(location, filename):
"""
Return asset by location and filename.
"""
return contentstore().find(Transcript.asset_location(location, filename))
@staticmethod
def asset_location(location, filename):
"""
Return asset location. `location` is module location.
"""
# If user transcript filename is empty, raise `TranscriptException` to avoid `InvalidKeyError`.
if not filename:
raise TranscriptException("Transcript not uploaded yet")
return StaticContent.compute_location(location.course_key, filename)
@staticmethod
def delete_asset(location, filename):
"""
Delete asset by location and filename.
"""
try:
contentstore().delete(Transcript.asset_location(location, filename))
log.info("Transcript asset %s was removed from store.", filename)
except NotFoundError:
pass
return StaticContent.compute_location(location.course_key, filename)
class VideoTranscriptsMixin:
"""Mixin class for transcript functionality.
This is necessary for VideoBlock.
"""
def available_translations(self, transcripts, verify_assets=None, is_bumper=False):
"""
Return a list of language codes for which we have transcripts.
Arguments:
verify_assets (boolean): If True, checks to ensure that the transcripts
really exist in the contentstore. If False, we just look at the
VideoBlock fields and do not query the contentstore. One reason
we might do this is to avoid slamming contentstore() with queries
when trying to make a listing of videos and their languages.
Defaults to `not FALLBACK_TO_ENGLISH_TRANSCRIPTS`.
transcripts (dict): A dict with all transcripts and a sub.
include_val_transcripts(boolean): If True, adds the edx-val transcript languages as well.
"""
translations = []
if verify_assets is None:
verify_assets = not settings.FEATURES.get('FALLBACK_TO_ENGLISH_TRANSCRIPTS')
sub, other_langs = transcripts["sub"], transcripts["transcripts"]
if verify_assets:
all_langs = dict(**other_langs)
if sub:
all_langs.update({'en': sub})
for language, filename in all_langs.items():
try:
# for bumper videos, transcripts are stored in content store only
if is_bumper:
get_transcript_for_video(self.location, filename, filename, language)
else:
get_transcript(self, language)
except NotFoundError:
continue
translations.append(language)
else:
# If we're not verifying the assets, we just trust our field values
translations = list(other_langs)
if not translations or sub:
translations += ['en']
# to clean redundant language codes.
return list(set(translations))
def get_default_transcript_language(self, transcripts):
"""
Returns the default transcript language for this video block.
Args:
transcripts (dict): A dict with all transcripts and a sub.
"""
sub, other_lang = transcripts["sub"], transcripts["transcripts"]
if self.transcript_language in other_lang:
transcript_language = self.transcript_language
elif sub:
transcript_language = 'en'
elif len(other_lang) > 0:
transcript_language = sorted(other_lang)[0]
else:
transcript_language = 'en'
return transcript_language
def get_transcripts_info(self, is_bumper=False):
"""
Returns a transcript dictionary for the video.
Arguments:
is_bumper(bool): If True, the request is for the bumper transcripts
include_val_transcripts(bool): If True, include edx-val transcripts as well
"""
if is_bumper:
transcripts = copy.deepcopy(get_bumper_settings(self).get('transcripts', {}))
sub = transcripts.pop("en", "")
else:
transcripts = self.transcripts if self.transcripts else {}
sub = self.sub
# Only attach transcripts that are not empty.
transcripts = {
language_code: transcript_file
for language_code, transcript_file in transcripts.items() if transcript_file != ''
}
# bumper transcripts are stored in content store so we don't need to include val transcripts
if not is_bumper:
transcript_languages = get_available_transcript_languages(edx_video_id=self.edx_video_id)
# HACK Warning! this is temporary and will be removed once edx-val take over the
# transcript module and contentstore will only function as fallback until all the
# data is migrated to edx-val.
for language_code in transcript_languages:
if language_code == 'en' and not sub:
sub = NON_EXISTENT_TRANSCRIPT
elif not transcripts.get(language_code):
transcripts[language_code] = NON_EXISTENT_TRANSCRIPT
return {
"sub": sub,
"transcripts": transcripts,
}
@exception_decorator
def get_transcript_from_val(edx_video_id, lang=None, output_format=Transcript.SRT):
"""
Get video transcript from edx-val.
Arguments:
edx_video_id (unicode): video identifier
lang (unicode): transcript language
output_format (unicode): transcript output format
Returns:
tuple containing content, filename, mimetype
"""
transcript = get_video_transcript_content(edx_video_id, lang)
if not transcript:
raise NotFoundError(f'Transcript not found for {edx_video_id}, lang: {lang}')
transcript_conversion_props = dict(transcript, output_format=output_format)
transcript = convert_video_transcript(**transcript_conversion_props)
filename = transcript['filename']
content = transcript['content']
mimetype = Transcript.mime_types[output_format]
return content, filename, mimetype
def get_transcript_for_video(video_location, subs_id, file_name, language):
"""
Get video transcript from content store. This is a lower level function and is used by
`get_transcript_from_contentstore`. Prefer that function instead where possible. If you
need to support getting transcripts from VAL or Blockstore as well, use the `get_transcript`
function instead.
NOTE: Transcripts can be searched from content store by two ways:
1. by an id(a.k.a subs_id) which will be used to construct transcript filename
2. by providing transcript filename
Arguments:
video_location (Locator): Video location
subs_id (unicode): id for a transcript in content store
file_name (unicode): file_name for a transcript in content store
language (unicode): transcript language
Returns:
tuple containing transcript input_format, basename, content
"""
try:
if subs_id is None:
raise NotFoundError
content = Transcript.asset(video_location, subs_id, language).data.decode('utf-8')
base_name = subs_id
input_format = Transcript.SJSON
except NotFoundError:
content = Transcript.asset(video_location, None, language, file_name).data.decode('utf-8')
base_name = os.path.splitext(file_name)[0]
input_format = Transcript.SRT
return input_format, base_name, content
@exception_decorator
def get_transcript_from_contentstore(video, language, output_format, transcripts_info, youtube_id=None):
"""
Get video transcript from content store.
Arguments:
video (Video Descriptor): Video descriptor
language (unicode): transcript language
output_format (unicode): transcript output format
transcripts_info (dict): transcript info for a video
youtube_id (unicode): youtube video id
Returns:
tuple containing content, filename, mimetype
"""
input_format, base_name, transcript_content = None, None, None
if output_format not in (Transcript.SRT, Transcript.SJSON, Transcript.TXT):
raise NotFoundError(f'Invalid transcript format `{output_format}`')
sub, other_languages = transcripts_info['sub'], transcripts_info['transcripts']
transcripts = dict(other_languages)
# this is sent in case of a translation dispatch and we need to use it as our subs_id.
possible_sub_ids = [youtube_id, sub, video.youtube_id_1_0] + get_html5_ids(video.html5_sources)
for sub_id in possible_sub_ids:
try:
transcripts['en'] = sub_id
input_format, base_name, transcript_content = get_transcript_for_video(
video.location,
subs_id=sub_id,
file_name=transcripts[language],
language=language
)
break
except (KeyError, NotFoundError):
continue
if transcript_content is None:
raise NotFoundError('No transcript for `{lang}` language'.format(
lang=language
))
# add language prefix to transcript file only if language is not None
language_prefix = f'{language}_' if language else ''
transcript_name = f'{language_prefix}{base_name}.{output_format}'
transcript_content = Transcript.convert(transcript_content, input_format=input_format, output_format=output_format)
if not transcript_content.strip():
raise NotFoundError('No transcript content')
if youtube_id:
youtube_ids = youtube_speed_dict(video)
transcript_content = json.dumps(
generate_subs(youtube_ids.get(youtube_id, 1), 1, json.loads(transcript_content))
)
return transcript_content, transcript_name, Transcript.mime_types[output_format]
def get_transcript_from_blockstore(video_block, language, output_format, transcripts_info):
"""
Get video transcript from Blockstore.
Blockstore expects video transcripts to be placed into the 'static/'
subfolder of the XBlock's folder in a Blockstore bundle. For example, if the
video XBlock's definition is in the standard location of
video/video1/definition.xml
Then the .srt files should be placed at e.g.
video/video1/static/video1-en.srt
This is the same place where other public static files are placed for other
XBlocks, such as image files used by HTML blocks.
Video XBlocks in Blockstore must set the 'transcripts' XBlock field to a
JSON dictionary listing the filename of the transcript for each language:
<video
youtube_id_1_0="3_yD_cEKoCk"
transcripts='{"en": "3_yD_cEKoCk-en.srt"}'
display_name="Welcome Video with Transcript"
download_track="true"
/>
This method is tested in openedx/core/djangoapps/content_libraries/tests/test_static_assets.py
Arguments:
video_block (Video XBlock): The video XBlock
language (str): transcript language
output_format (str): transcript output format
transcripts_info (dict): transcript info for a video, from video_block.get_transcripts_info()
Returns:
tuple containing content, filename, mimetype
"""
if output_format not in (Transcript.SRT, Transcript.SJSON, Transcript.TXT):
raise NotFoundError(f'Invalid transcript format `{output_format}`')
transcripts = transcripts_info['transcripts']
if language not in transcripts:
raise NotFoundError("Video {} does not have a transcript file defined for the '{}' language in its OLX.".format(
video_block.scope_ids.usage_id,
language,
))
filename = transcripts[language]
if not filename.endswith('.srt'):
# We want to standardize on .srt
raise NotFoundError("Video XBlocks in Blockstore only support .srt transcript files.")
# Try to load the transcript file out of Blockstore
# In lieu of an XBlock API for this (like block.runtime.resources_fs), we use the blockstore API directly.
bundle_uuid = video_block.scope_ids.def_id.bundle_uuid
path = video_block.scope_ids.def_id.olx_path.rpartition('/')[0] + '/static/' + filename
bundle_version = video_block.scope_ids.def_id.bundle_version # Either bundle_version or draft_name will be set.
draft_name = video_block.scope_ids.def_id.draft_name
try:
content_binary = blockstore_cache.get_bundle_file_data_with_cache(bundle_uuid, path, bundle_version, draft_name)
except blockstore_api.BundleFileNotFound:
raise NotFoundError("Transcript file '{}' missing for video XBlock {}".format( # lint-amnesty, pylint: disable=raise-missing-from
path,
video_block.scope_ids.usage_id,
))
# Now convert the transcript data to the requested format:
filename_no_extension = os.path.splitext(filename)[0]
output_filename = f'{filename_no_extension}.{output_format}'
output_transcript = Transcript.convert(
content_binary.decode('utf-8'),
input_format=Transcript.SRT,
output_format=output_format,
)
if not output_transcript.strip():
raise NotFoundError('No transcript content')
return output_transcript, output_filename, Transcript.mime_types[output_format]
def get_transcript(video, lang=None, output_format=Transcript.SRT, youtube_id=None):
"""
Get video transcript from edx-val or content store.
Arguments:
video (Video Descriptor): Video Descriptor
lang (unicode): transcript language
output_format (unicode): transcript output format
youtube_id (unicode): youtube video id
Returns:
tuple containing content, filename, mimetype
"""
transcripts_info = video.get_transcripts_info()
if not lang:
lang = video.get_default_transcript_language(transcripts_info)
if isinstance(video.scope_ids.def_id, BundleDefinitionLocator):
# This block is in Blockstore.
# For Blockstore, VAL is considered deprecated and we can load the transcript file
# directly using the Blockstore API:
return get_transcript_from_blockstore(video, lang, output_format, transcripts_info)
try:
edx_video_id = clean_video_id(video.edx_video_id)
if not edx_video_id:
raise NotFoundError
return get_transcript_from_val(edx_video_id, lang, output_format)
except NotFoundError:
return get_transcript_from_contentstore(
video,
lang,
youtube_id=youtube_id,
output_format=output_format,
transcripts_info=transcripts_info
)

1152
xmodule/video_block/video_block.py Normal file
View File

@@ -0,0 +1,1152 @@
"""Video is ungraded Xmodule for support video content.
It's new improved video block, which support additional feature:
- Can play non-YouTube video sources via in-browser HTML5 video player.
- YouTube defaults to HTML5 mode from the start.
- Speed changes in both YouTube and non-YouTube videos happen via
in-browser HTML5 video method (when in HTML5 mode).
- Navigational subtitles can be disabled altogether via an attribute
in XML.
Examples of html5 videos for manual testing:
https://s3.amazonaws.com/edx-course-videos/edx-intro/edX-FA12-cware-1_100.mp4
https://s3.amazonaws.com/edx-course-videos/edx-intro/edX-FA12-cware-1_100.webm
https://s3.amazonaws.com/edx-course-videos/edx-intro/edX-FA12-cware-1_100.ogv
"""
import copy
import json
import logging
from collections import OrderedDict, defaultdict
from operator import itemgetter
from django.conf import settings
from edx_django_utils.cache import RequestCache
from lxml import etree
from opaque_keys.edx.locator import AssetLocator
from web_fragments.fragment import Fragment
from xblock.completable import XBlockCompletionMode
from xblock.core import XBlock
from xblock.fields import ScopeIds
from xblock.runtime import KvsFieldData
from common.djangoapps.xblock_django.constants import ATTR_KEY_REQUEST_COUNTRY_CODE
from openedx.core.djangoapps.video_config.models import HLSPlaybackEnabledFlag, CourseYoutubeBlockedFlag
from openedx.core.djangoapps.video_pipeline.config.waffle import DEPRECATE_YOUTUBE
from openedx.core.lib.cache_utils import request_cached
from openedx.core.lib.license import LicenseMixin
from xmodule.contentstore.content import StaticContent
from xmodule.editing_block import EditingMixin
from xmodule.exceptions import NotFoundError
from xmodule.mako_block import MakoTemplateBlockBase
from xmodule.modulestore.inheritance import InheritanceKeyValueStore, own_metadata
from xmodule.raw_block import EmptyDataRawMixin
from xmodule.validation import StudioValidation, StudioValidationMessage
from xmodule.util.xmodule_django import add_webpack_to_fragment
from xmodule.video_block import manage_video_subtitles_save
from xmodule.x_module import (
PUBLIC_VIEW, STUDENT_VIEW,
HTMLSnippet, ResourceTemplates, shim_xmodule_js,
XModuleMixin, XModuleToXBlockMixin,
)
from xmodule.xml_block import XmlMixin, deserialize_field, is_pointer_tag, name_to_pathname
from .bumper_utils import bumperize
from .transcripts_utils import (
Transcript,
VideoTranscriptsMixin,
clean_video_id,
get_html5_ids,
get_transcript,
subs_filename
)
from .video_handlers import VideoStudentViewHandlers, VideoStudioViewHandlers
from .video_utils import create_youtube_string, format_xml_exception_message, get_poster, rewrite_video_url
from .video_xfields import VideoFields
# The following import/except block for edxval is temporary measure until
# edxval is a proper XBlock Runtime Service.
#
# Here's the deal: the VideoBlock should be able to take advantage of edx-val
# (https://github.com/openedx/edx-val) to figure out what URL to give for video
# resources that have an edx_video_id specified. edx-val is a Django app, and
# including it causes tests to fail because we run common/lib tests standalone
# without Django dependencies. The alternatives seem to be:
#
# 1. Move VideoBlock out of edx-platform.
# 2. Accept the Django dependency in common/lib.
# 3. Try to import, catch the exception on failure, and check for the existence
# of edxval_api before invoking it in the code.
# 4. Make edxval an XBlock Runtime Service
#
# (1) is a longer term goal. VideoBlock should be made into an XBlock and
# extracted from edx-platform entirely. But that's expensive to do because of
# the various dependencies (like templates). Need to sort this out.
# (2) is explicitly discouraged.
# (3) is what we're doing today. The code is still functional when called within
# the context of the LMS, but does not cause failure on import when running
# standalone tests. Most VideoBlock tests tend to be in the LMS anyway,
# probably for historical reasons, so we're not making things notably worse.
# (4) is one of the next items on the backlog for edxval, and should get rid
# of this particular import silliness. It's just that I haven't made one before,
# and I was worried about trying it with my deadline constraints.
try:
import edxval.api as edxval_api
except ImportError:
edxval_api = None
try:
from lms.djangoapps.branding.models import BrandingInfoConfig
except ImportError:
BrandingInfoConfig = None
log = logging.getLogger(__name__)
# Make '_' a no-op so we can scrape strings. Using lambda instead of
# `django.utils.translation.ugettext_noop` because Django cannot be imported in this file
_ = lambda text: text
EXPORT_IMPORT_COURSE_DIR = 'course'
EXPORT_IMPORT_STATIC_DIR = 'static'
@XBlock.wants('settings', 'completion', 'i18n', 'request_cache')
@XBlock.needs('mako', 'user')
class VideoBlock(
VideoFields, VideoTranscriptsMixin, VideoStudioViewHandlers, VideoStudentViewHandlers,
EmptyDataRawMixin, XmlMixin, EditingMixin, XModuleToXBlockMixin, HTMLSnippet,
ResourceTemplates, XModuleMixin, LicenseMixin):
"""
XML source example:
<video show_captions="true"
youtube="0.75:jNCf2gIqpeE,1.0:ZwkTiUPN0mg,1.25:rsq9auxASqI,1.50:kMyNdzVHHgg"
url_name="lecture_21_3" display_name="S19V3: Vacancies"
>
<source src=".../mit-3091x/M-3091X-FA12-L21-3_100.mp4"/>
<source src=".../mit-3091x/M-3091X-FA12-L21-3_100.webm"/>
<source src=".../mit-3091x/M-3091X-FA12-L21-3_100.ogv"/>
</video>
"""
has_custom_completion = True
completion_mode = XBlockCompletionMode.COMPLETABLE
video_time = 0
icon_class = 'video'
show_in_read_only_mode = True
tabs = [
{
'name': _("Basic"),
'template': "video/transcripts.html",
'current': True
},
{
'name': _("Advanced"),
'template': "tabs/metadata-edit-tab.html"
}
]
mako_template = "widgets/tabs-aggregator.html"
js_module_name = "TabsEditingDescriptor"
uses_xmodule_styles_setup = True
requires_per_student_anonymous_id = True
def get_transcripts_for_student(self, transcripts):
"""Return transcript information necessary for rendering the XModule student view.
This is more or less a direct extraction from `get_html`.
Args:
transcripts (dict): A dict with all transcripts and a sub.
Returns:
Tuple of (track_url, transcript_language, sorted_languages)
track_url -> subtitle download url
transcript_language -> default transcript language
sorted_languages -> dictionary of available transcript languages
"""
track_url = None
sub, other_lang = transcripts["sub"], transcripts["transcripts"]
if self.download_track:
if self.track:
track_url = self.track
elif sub or other_lang:
track_url = self.runtime.handler_url(self, 'transcript', 'download').rstrip('/?')
transcript_language = self.get_default_transcript_language(transcripts)
native_languages = {lang: label for lang, label in settings.LANGUAGES if len(lang) == 2}
languages = {
lang: native_languages.get(lang, display)
for lang, display in settings.ALL_LANGUAGES
if lang in other_lang
}
if not other_lang or (other_lang and sub):
languages['en'] = 'English'
# OrderedDict for easy testing of rendered context in tests
sorted_languages = sorted(list(languages.items()), key=itemgetter(1))
sorted_languages = OrderedDict(sorted_languages)
return track_url, transcript_language, sorted_languages
@property
def youtube_deprecated(self):
"""
Return True if youtube is deprecated and hls as primary playback is enabled else False
"""
# Return False if `hls` playback feature is disabled.
if not HLSPlaybackEnabledFlag.feature_enabled(self.location.course_key):
return False
# check if youtube has been deprecated and hls as primary playback
# is enabled for this course
return DEPRECATE_YOUTUBE.is_enabled(self.location.course_key)
def youtube_disabled_for_course(self): # lint-amnesty, pylint: disable=missing-function-docstring
if not self.location.context_key.is_course:
return False # Only courses have this flag
request_cache = RequestCache('youtube_disabled_for_course')
cache_response = request_cache.get_cached_response(self.location.context_key)
if cache_response.is_found:
return cache_response.value
youtube_is_disabled = CourseYoutubeBlockedFlag.feature_enabled(self.location.course_key)
request_cache.set(self.location.context_key, youtube_is_disabled)
return youtube_is_disabled
def prioritize_hls(self, youtube_streams, html5_sources):
"""
Decide whether hls can be prioritized as primary playback or not.
If both the youtube and hls sources are present then make decision on flag
If only either youtube or hls is present then play whichever is present
"""
yt_present = bool(youtube_streams.strip()) if youtube_streams else False
hls_present = any(source for source in html5_sources)
if yt_present and hls_present:
return self.youtube_deprecated
return False
def student_view(self, _context):
"""
Return the student view.
"""
fragment = Fragment(self.get_html())
add_webpack_to_fragment(fragment, 'VideoBlockPreview')
shim_xmodule_js(fragment, 'Video')
return fragment
def author_view(self, context):
"""
Renders the Studio preview view.
"""
return self.student_view(context)
def studio_view(self, _context):
"""
Return the studio view.
"""
fragment = Fragment(
self.runtime.service(self, 'mako').render_template(self.mako_template, self.get_context())
)
add_webpack_to_fragment(fragment, 'VideoBlockStudio')
shim_xmodule_js(fragment, 'TabsEditingDescriptor')
return fragment
def public_view(self, context):
"""
Returns a fragment that contains the html for the public view
"""
if getattr(self.runtime, 'suppports_state_for_anonymous_users', False):
# The new runtime can support anonymous users as fully as regular users:
return self.student_view(context)
fragment = Fragment(self.get_html(view=PUBLIC_VIEW))
add_webpack_to_fragment(fragment, 'VideoBlockPreview')
shim_xmodule_js(fragment, 'Video')
return fragment
def get_html(self, view=STUDENT_VIEW): # lint-amnesty, pylint: disable=arguments-differ, too-many-statements
track_status = (self.download_track and self.track)
transcript_download_format = self.transcript_download_format if not track_status else None
sources = [source for source in self.html5_sources if source]
download_video_link = None
branding_info = None
youtube_streams = ""
video_duration = None
video_status = None
# Determine if there is an alternative source for this video
# based on user locale. This exists to support cases where
# we leverage a geography specific CDN, like China.
default_cdn_url = getattr(settings, 'VIDEO_CDN_URL', {}).get('default')
user_location = self.runtime.service(self, 'user').get_current_user().opt_attrs[ATTR_KEY_REQUEST_COUNTRY_CODE]
cdn_url = getattr(settings, 'VIDEO_CDN_URL', {}).get(user_location, default_cdn_url)
# If we have an edx_video_id, we prefer its values over what we store
# internally for download links (source, html5_sources) and the youtube
# stream.
if self.edx_video_id and edxval_api: # lint-amnesty, pylint: disable=too-many-nested-blocks
try:
val_profiles = ["youtube", "desktop_webm", "desktop_mp4"]
if HLSPlaybackEnabledFlag.feature_enabled(self.course_id):
val_profiles.append('hls')
# strip edx_video_id to prevent ValVideoNotFoundError error if unwanted spaces are there. TNL-5769
val_video_urls = edxval_api.get_urls_for_profiles(self.edx_video_id.strip(), val_profiles)
# VAL will always give us the keys for the profiles we asked for, but
# if it doesn't have an encoded video entry for that Video + Profile, the
# value will map to `None`
# add the non-youtube urls to the list of alternative sources
# use the last non-None non-youtube non-hls url as the link to download the video
for url in [val_video_urls[p] for p in val_profiles if p != "youtube"]:
if url:
if url not in sources:
sources.append(url)
# don't include hls urls for download
if self.download_video and not url.endswith('.m3u8'):
# function returns None when the url cannot be re-written
rewritten_link = rewrite_video_url(cdn_url, url)
if rewritten_link:
download_video_link = rewritten_link
else:
download_video_link = url
# set the youtube url
if val_video_urls["youtube"]:
youtube_streams = "1.00:{}".format(val_video_urls["youtube"])
# get video duration
video_data = edxval_api.get_video_info(self.edx_video_id.strip())
video_duration = video_data.get('duration')
video_status = video_data.get('status')
except (edxval_api.ValInternalError, edxval_api.ValVideoNotFoundError):
# VAL raises this exception if it can't find data for the edx video ID. This can happen if the
# course data is ported to a machine that does not have the VAL data. So for now, pass on this
# exception and fallback to whatever we find in the VideoBlock.
log.warning("Could not retrieve information from VAL for edx Video ID: %s.", self.edx_video_id)
# If the user comes from China use China CDN for html5 videos.
# 'CN' is China ISO 3166-1 country code.
# Video caching is disabled for Studio. User_location is always None in Studio.
# CountryMiddleware disabled for Studio.
if getattr(self, 'video_speed_optimizations', True) and cdn_url:
branding_info = BrandingInfoConfig.get_config().get(user_location)
if self.edx_video_id and edxval_api and video_status != 'external':
for index, source_url in enumerate(sources):
new_url = rewrite_video_url(cdn_url, source_url)
if new_url:
sources[index] = new_url
# If there was no edx_video_id, or if there was no download specified
# for it, we fall back on whatever we find in the VideoBlock.
if not download_video_link and self.download_video:
if self.html5_sources:
download_video_link = self.html5_sources[0]
# don't give the option to download HLS video urls
if download_video_link and download_video_link.endswith('.m3u8'):
download_video_link = None
transcripts = self.get_transcripts_info()
track_url, transcript_language, sorted_languages = self.get_transcripts_for_student(transcripts=transcripts)
cdn_eval = False
cdn_exp_group = None
if self.youtube_disabled_for_course():
self.youtube_streams = '' # lint-amnesty, pylint: disable=attribute-defined-outside-init
else:
self.youtube_streams = youtube_streams or create_youtube_string(self) # pylint: disable=W0201
settings_service = self.runtime.service(self, 'settings') # lint-amnesty, pylint: disable=unused-variable
poster = None
if edxval_api and self.edx_video_id:
poster = edxval_api.get_course_video_image_url(
course_id=self.scope_ids.usage_id.context_key.for_branch(None),
edx_video_id=self.edx_video_id.strip()
)
completion_service = self.runtime.service(self, 'completion')
if completion_service:
completion_enabled = completion_service.completion_tracking_enabled()
else:
completion_enabled = False
# This is the setting that controls whether the autoadvance button will be visible, not whether the
# video will autoadvance or not.
# For autoadvance controls to be shown, both the feature flag and the course setting must be true.
# This allows to enable the feature for certain courses only.
autoadvance_enabled = settings.FEATURES.get('ENABLE_AUTOADVANCE_VIDEOS', False) and \
getattr(self, 'video_auto_advance', False)
# This is the current status of auto-advance (not the control visibility).
# But when controls aren't visible we force it to off. The student might have once set the preference to
# true, but now staff or admin have hidden the autoadvance button and the student won't be able to disable
# it anymore; therefore we force-disable it in this case (when controls aren't visible).
autoadvance_this_video = self.auto_advance and autoadvance_enabled
metadata = {
'autoAdvance': autoadvance_this_video,
# For now, the option "data-autohide-html5" is hard coded. This option
# either enables or disables autohiding of controls and captions on mouse
# inactivity. If set to true, controls and captions will autohide for
# HTML5 sources (non-YouTube) after a period of mouse inactivity over the
# whole video. When the mouse moves (or a key is pressed while any part of
# the video player is focused), the captions and controls will be shown
# once again.
#
# There is no option in the "Advanced Editor" to set this option. However,
# this option will have an effect if changed to "True". The code on
# front-end exists.
'autohideHtml5': False,
'autoplay': settings.FEATURES.get('AUTOPLAY_VIDEOS', False),
# This won't work when we move to data that
# isn't on the filesystem
'captionDataDir': getattr(self, 'data_dir', None),
'completionEnabled': completion_enabled,
'completionPercentage': settings.COMPLETION_VIDEO_COMPLETE_PERCENTAGE,
'duration': video_duration,
'end': self.end_time.total_seconds(), # pylint: disable=no-member
'generalSpeed': self.global_speed,
'lmsRootURL': settings.LMS_ROOT_URL,
'poster': poster,
'prioritizeHls': self.prioritize_hls(self.youtube_streams, sources),
'publishCompletionUrl': self.runtime.handler_url(self, 'publish_completion', '').rstrip('?'),
# This is the server's guess at whether youtube is available for
# this user, based on what was recorded the last time we saw the
# user, and defaulting to True.
'recordedYoutubeIsAvailable': self.youtube_is_available,
'savedVideoPosition': self.saved_video_position.total_seconds(), # pylint: disable=no-member
'saveStateEnabled': view != PUBLIC_VIEW,
'saveStateUrl': self.ajax_url + '/save_user_state',
'showCaptions': json.dumps(self.show_captions),
'sources': sources,
'speed': self.speed,
'start': self.start_time.total_seconds(), # pylint: disable=no-member
'streams': self.youtube_streams,
'transcriptAvailableTranslationsUrl': self.runtime.handler_url(
self, 'transcript', 'available_translations'
).rstrip('/?'),
'transcriptLanguage': transcript_language,
'transcriptLanguages': sorted_languages,
'transcriptTranslationUrl': self.runtime.handler_url(
self, 'transcript', 'translation/__lang__'
).rstrip('/?'),
'ytApiUrl': settings.YOUTUBE['API'],
'ytMetadataEndpoint': (
# In the new runtime, get YouTube metadata via a handler. The handler supports anonymous users and
# can work in sandboxed iframes. In the old runtime, the JS will call the LMS's yt_video_metadata
# API endpoint directly (not an XBlock handler).
self.runtime.handler_url(self, 'yt_video_metadata')
if getattr(self.runtime, 'suppports_state_for_anonymous_users', False) else ''
),
'ytTestTimeout': settings.YOUTUBE['TEST_TIMEOUT'],
}
bumperize(self)
context = {
'autoadvance_enabled': autoadvance_enabled,
'bumper_metadata': json.dumps(self.bumper['metadata']), # pylint: disable=E1101
'metadata': json.dumps(OrderedDict(metadata)),
'poster': json.dumps(get_poster(self)),
'branding_info': branding_info,
'cdn_eval': cdn_eval,
'cdn_exp_group': cdn_exp_group,
'id': self.location.html_id(),
'display_name': self.display_name_with_default,
'handout': self.handout,
'download_video_link': download_video_link,
'track': track_url,
'transcript_download_format': transcript_download_format,
'transcript_download_formats_list': self.fields['transcript_download_format'].values, # lint-amnesty, pylint: disable=unsubscriptable-object
'license': getattr(self, "license", None),
}
return self.runtime.service(self, 'mako').render_template('video.html', context)
def validate(self):
"""
Validates the state of this Video XBlock instance. This
is the override of the general XBlock method, and it will also ask
its superclass to validate.
"""
validation = super().validate()
if not isinstance(validation, StudioValidation):
validation = StudioValidation.copy(validation)
no_transcript_lang = []
for lang_code, transcript in self.transcripts.items():
if not transcript:
no_transcript_lang.append([label for code, label in settings.ALL_LANGUAGES if code == lang_code][0])
if no_transcript_lang:
ungettext = self.runtime.service(self, "i18n").ungettext
validation.set_summary(
StudioValidationMessage(
StudioValidationMessage.WARNING,
ungettext(
'There is no transcript file associated with the {lang} language.',
'There are no transcript files associated with the {lang} languages.',
len(no_transcript_lang)
).format(lang=', '.join(sorted(no_transcript_lang)))
)
)
return validation
def editor_saved(self, user, old_metadata, old_content): # lint-amnesty, pylint: disable=unused-argument
"""
Used to update video values during `self`:save method from CMS.
old_metadata: dict, values of fields of `self` with scope=settings which were explicitly set by user.
old_content, same as `old_metadata` but for scope=content.
Due to nature of code flow in item.py::_save_item, before current function is called,
fields of `self` instance have been already updated, but not yet saved.
To obtain values, which were changed by user input,
one should compare own_metadata(self) and old_medatada.
Video player has two tabs, and due to nature of sync between tabs,
metadata from Basic tab is always sent when video player is edited and saved first time, for example:
{'youtube_id_1_0': u'3_yD_cEKoCk', 'display_name': u'Video', 'sub': u'3_yD_cEKoCk', 'html5_sources': []},
that's why these fields will always present in old_metadata after first save. This should be fixed.
At consequent save requests html5_sources are always sent too, disregard of their change by user.
That means that html5_sources are always in list of fields that were changed (`metadata` param in save_item).
This should be fixed too.
"""
metadata_was_changed_by_user = old_metadata != own_metadata(self)
# There is an edge case when old_metadata and own_metadata are same and we are importing transcript from youtube
# then there is a syncing issue where html5_subs are not syncing with youtube sub, We can make sync better by
# checking if transcript is present for the video and if any html5_ids transcript is not present then trigger
# the manage_video_subtitles_save to create the missing transcript with particular html5_id.
if not metadata_was_changed_by_user and self.sub and hasattr(self, 'html5_sources'):
html5_ids = get_html5_ids(self.html5_sources)
for subs_id in html5_ids:
try:
Transcript.asset(self.location, subs_id)
except NotFoundError:
# If a transcript does not not exist with particular html5_id then there is no need to check other
# html5_ids because we have to create a new transcript with this missing html5_id by turning on
# metadata_was_changed_by_user flag.
metadata_was_changed_by_user = True
break
if metadata_was_changed_by_user:
self.edx_video_id = self.edx_video_id and self.edx_video_id.strip()
# We want to override `youtube_id_1_0` with val youtube profile in the first place when someone adds/edits
# an `edx_video_id` or its underlying YT val profile. Without this, override will only happen when a user
# saves the video second time. This is because of the syncing of basic and advanced video settings which
# also syncs val youtube id from basic tab's `Video Url` to advanced tab's `Youtube ID`.
if self.edx_video_id and edxval_api:
val_youtube_id = edxval_api.get_url_for_profile(self.edx_video_id, 'youtube')
if val_youtube_id and self.youtube_id_1_0 != val_youtube_id:
self.youtube_id_1_0 = val_youtube_id
manage_video_subtitles_save(
self,
user,
old_metadata if old_metadata else None,
generate_translation=True
)
def save_with_metadata(self, user):
"""
Save module with updated metadata to database."
"""
self.save()
self.runtime.modulestore.update_item(self, user.id)
@property
def editable_metadata_fields(self):
editable_fields = super().editable_metadata_fields
settings_service = self.runtime.service(self, 'settings')
if settings_service:
xb_settings = settings_service.get_settings_bucket(self)
if not xb_settings.get("licensing_enabled", False) and "license" in editable_fields:
del editable_fields["license"]
# Default Timed Transcript a.k.a `sub` has been deprecated and end users shall
# not be able to modify it.
editable_fields.pop('sub')
languages = [{'label': label, 'code': lang} for lang, label in settings.ALL_LANGUAGES]
languages.sort(key=lambda l: l['label'])
editable_fields['transcripts']['custom'] = True
editable_fields['transcripts']['languages'] = languages
editable_fields['transcripts']['type'] = 'VideoTranslations'
# We need to send ajax requests to show transcript status
# whenever edx_video_id changes on frontend. Thats why we
# are changing type to `VideoID` so that a specific
# Backbonjs view can handle it.
editable_fields['edx_video_id']['type'] = 'VideoID'
# `public_access` is a boolean field and by default backbonejs code render it as a dropdown with 2 options
# but in our case we also need to show an input field with dropdown, the input field will show the url to
# be shared with leaners. This is not possible with default rendering logic in backbonjs code, that is why
# we are setting a new type and then do a custom rendering in backbonejs code to render the desired UI.
editable_fields['public_access']['type'] = 'PublicAccess'
editable_fields['public_access']['url'] = fr'{settings.LMS_ROOT_URL}/videos/{str(self.location)}'
# construct transcripts info and also find if `en` subs exist
transcripts_info = self.get_transcripts_info()
possible_sub_ids = [self.sub, self.youtube_id_1_0] + get_html5_ids(self.html5_sources)
for sub_id in possible_sub_ids:
try:
_, sub_id, _ = get_transcript(self, lang='en', output_format=Transcript.TXT)
transcripts_info['transcripts'] = dict(transcripts_info['transcripts'], en=sub_id)
break
except NotFoundError:
continue
editable_fields['transcripts']['value'] = transcripts_info['transcripts']
editable_fields['transcripts']['urlRoot'] = self.runtime.handler_url(
self,
'studio_transcript',
'translation'
).rstrip('/?')
editable_fields['handout']['type'] = 'FileUploader'
return editable_fields
@classmethod
def parse_xml_new_runtime(cls, node, runtime, keys):
"""
Implement the video block's special XML parsing requirements for the
new runtime only. For all other runtimes, use .parse_xml().
"""
video_block = runtime.construct_xblock_from_class(cls, keys)
field_data = cls.parse_video_xml(node)
for key, val in field_data.items():
if key not in cls.fields: # lint-amnesty, pylint: disable=unsupported-membership-test
continue # parse_video_xml returns some old non-fields like 'source'
setattr(video_block, key, cls.fields[key].from_json(val)) # lint-amnesty, pylint: disable=unsubscriptable-object
# Don't use VAL in the new runtime:
video_block.edx_video_id = None
return video_block
@classmethod
def parse_xml(cls, node, runtime, _keys, id_generator):
"""
Use `node` to construct a new block.
See XmlMixin.parse_xml for the detailed description.
"""
url_name = node.get('url_name')
block_type = 'video'
definition_id = id_generator.create_definition(block_type, url_name)
usage_id = id_generator.create_usage(definition_id)
if is_pointer_tag(node):
filepath = cls._format_filepath(node.tag, name_to_pathname(url_name))
node = cls.load_file(filepath, runtime.resources_fs, usage_id)
runtime.parse_asides(node, definition_id, usage_id, id_generator)
field_data = cls.parse_video_xml(node, id_generator)
kvs = InheritanceKeyValueStore(initial_values=field_data)
field_data = KvsFieldData(kvs)
video = runtime.construct_xblock_from_class(
cls,
# We're loading a descriptor, so student_id is meaningless
# We also don't have separate notions of definition and usage ids yet,
# so we use the location for both
ScopeIds(None, block_type, definition_id, usage_id),
field_data,
)
# Update VAL with info extracted from `node`
video.edx_video_id = video.import_video_info_into_val(
node,
runtime.resources_fs,
getattr(id_generator, 'target_course_id', None)
)
return video
def definition_to_xml(self, resource_fs): # lint-amnesty, pylint: disable=too-many-statements
"""
Returns an xml string representing this module.
"""
xml = etree.Element('video')
youtube_string = create_youtube_string(self)
if youtube_string:
xml.set('youtube', str(youtube_string))
xml.set('url_name', self.url_name)
attrs = [
('display_name', self.display_name),
('show_captions', json.dumps(self.show_captions)),
('start_time', self.start_time),
('end_time', self.end_time),
('sub', self.sub),
('download_track', json.dumps(self.download_track)),
('download_video', json.dumps(self.download_video))
]
for key, value in attrs:
# Mild workaround to ensure that tests pass -- if a field
# is set to its default value, we don't write it out.
if value:
if key in self.fields and self.fields[key].is_set_on(self): # lint-amnesty, pylint: disable=unsubscriptable-object, unsupported-membership-test
try:
xml.set(key, str(value))
except UnicodeDecodeError:
exception_message = format_xml_exception_message(self.location, key, value)
log.exception(exception_message)
# If exception is UnicodeDecodeError set value using unicode 'utf-8' scheme.
log.info("Setting xml value using 'utf-8' scheme.")
xml.set(key, str(value, 'utf-8'))
except ValueError:
exception_message = format_xml_exception_message(self.location, key, value)
log.exception(exception_message)
raise
for source in self.html5_sources:
ele = etree.Element('source')
ele.set('src', source)
xml.append(ele)
if self.track:
ele = etree.Element('track')
ele.set('src', self.track)
xml.append(ele)
if self.handout:
ele = etree.Element('handout')
ele.set('src', self.handout)
xml.append(ele)
transcripts = {}
if self.transcripts is not None:
transcripts.update(self.transcripts)
edx_video_id = clean_video_id(self.edx_video_id)
if edxval_api and edx_video_id:
try:
# Create static dir if not created earlier.
resource_fs.makedirs(EXPORT_IMPORT_STATIC_DIR, recreate=True)
# Backward compatible exports
# edxval exports new transcripts into the course OLX and returns a transcript
# files map so that it can also be rewritten in old transcript metadata fields
# (i.e. `self.transcripts`) on import and older open-releases (<= ginkgo),
# who do not have deprecated contentstore yet, can also import and use new-style
# transcripts into their openedX instances.
exported_metadata = edxval_api.export_to_xml(
video_id=edx_video_id,
resource_fs=resource_fs,
static_dir=EXPORT_IMPORT_STATIC_DIR,
course_id=self.scope_ids.usage_id.context_key.for_branch(None),
)
# Update xml with edxval metadata
xml.append(exported_metadata['xml'])
# we don't need sub if english transcript
# is also in new transcripts.
new_transcripts = exported_metadata['transcripts']
transcripts.update(new_transcripts)
if new_transcripts.get('en'):
xml.set('sub', '')
# Update `transcripts` attribute in the xml
xml.set('transcripts', json.dumps(transcripts, sort_keys=True))
except edxval_api.ValVideoNotFoundError:
pass
# Sorting transcripts for easy testing of resulting xml
for transcript_language in sorted(transcripts.keys()):
ele = etree.Element('transcript')
ele.set('language', transcript_language)
ele.set('src', transcripts[transcript_language])
xml.append(ele)
# handle license specifically
self.add_license_to_xml(xml)
return xml
def create_youtube_url(self, youtube_id):
"""
Args:
youtube_id: The ID of the video to create a link for
Returns:
A full youtube url to the video whose ID is passed in
"""
if youtube_id:
return f'https://www.youtube.com/watch?v={youtube_id}'
else:
return ''
def get_context(self):
"""
Extend context by data for transcript basic tab.
"""
_context = MakoTemplateBlockBase.get_context(self)
_context.update({
'tabs': self.tabs,
'html_id': self.location.html_id(), # element_id
'data': self.data,
})
metadata_fields = copy.deepcopy(self.editable_metadata_fields)
display_name = metadata_fields['display_name']
video_url = metadata_fields['html5_sources']
video_id = metadata_fields['edx_video_id']
youtube_id_1_0 = metadata_fields['youtube_id_1_0']
def get_youtube_link(video_id):
"""
Returns the fully-qualified YouTube URL for the given video identifier
"""
# First try a lookup in VAL. If we have a YouTube entry there, it overrides the
# one passed in.
if self.edx_video_id and edxval_api:
val_youtube_id = edxval_api.get_url_for_profile(self.edx_video_id, "youtube")
if val_youtube_id:
video_id = val_youtube_id
return self.create_youtube_url(video_id)
_ = self.runtime.service(self, "i18n").ugettext
video_url.update({
'help': _('The URL for your video. This can be a YouTube URL or a link to an .mp4, .ogg, or '
'.webm video file hosted elsewhere on the Internet.'),
'display_name': _('Default Video URL'),
'field_name': 'video_url',
'type': 'VideoList',
'default_value': [get_youtube_link(youtube_id_1_0['default_value'])]
})
source_url = self.create_youtube_url(youtube_id_1_0['value'])
# First try a lookup in VAL. If any video encoding is found given the video id then
# override the source_url with it.
if self.edx_video_id and edxval_api:
val_profiles = ['youtube', 'desktop_webm', 'desktop_mp4']
if HLSPlaybackEnabledFlag.feature_enabled(self.scope_ids.usage_id.context_key.for_branch(None)):
val_profiles.append('hls')
# Get video encodings for val profiles.
val_video_encodings = edxval_api.get_urls_for_profiles(self.edx_video_id, val_profiles)
# VAL's youtube source has greater priority over external youtube source.
if val_video_encodings.get('youtube'):
source_url = self.create_youtube_url(val_video_encodings['youtube'])
# If no youtube source is provided externally or in VAl, update source_url in order: hls > mp4 and webm
if not source_url:
if val_video_encodings.get('hls'):
source_url = val_video_encodings['hls']
elif val_video_encodings.get('desktop_mp4'):
source_url = val_video_encodings['desktop_mp4']
elif val_video_encodings.get('desktop_webm'):
source_url = val_video_encodings['desktop_webm']
# Only add if html5 sources do not already contain source_url.
if source_url and source_url not in video_url['value']:
video_url['value'].insert(0, source_url)
metadata = {
'display_name': display_name,
'video_url': video_url,
'edx_video_id': video_id
}
_context.update({'transcripts_basic_tab_metadata': metadata})
return _context
@classmethod
def _parse_youtube(cls, data):
"""
Parses a string of Youtube IDs such as "1.0:AXdE34_U,1.5:VO3SxfeD"
into a dictionary. Necessary for backwards compatibility with
XML-based courses.
"""
ret = {'0.75': '', '1.00': '', '1.25': '', '1.50': ''}
videos = data.split(',')
for video in videos:
pieces = video.split(':')
try:
speed = '%.2f' % float(pieces[0]) # normalize speed
# Handle the fact that youtube IDs got double-quoted for a period of time.
# Note: we pass in "VideoFields.youtube_id_1_0" so we deserialize as a String--
# it doesn't matter what the actual speed is for the purposes of deserializing.
youtube_id = deserialize_field(cls.youtube_id_1_0, pieces[1])
ret[speed] = youtube_id
except (ValueError, IndexError):
log.warning('Invalid YouTube ID: %s', video)
return ret
@classmethod
def parse_video_xml(cls, xml, id_generator=None):
"""
Parse video fields out of xml_data. The fields are set if they are
present in the XML.
Arguments:
id_generator is used to generate course-specific urls and identifiers
"""
if isinstance(xml, str):
xml = etree.fromstring(xml)
field_data = {}
# Convert between key types for certain attributes --
# necessary for backwards compatibility.
conversions = {
# example: 'start_time': cls._example_convert_start_time
}
# Convert between key names for certain attributes --
# necessary for backwards compatibility.
compat_keys = {
'from': 'start_time',
'to': 'end_time'
}
sources = xml.findall('source')
if sources:
field_data['html5_sources'] = [ele.get('src') for ele in sources]
track = xml.find('track')
if track is not None:
field_data['track'] = track.get('src')
handout = xml.find('handout')
if handout is not None:
field_data['handout'] = handout.get('src')
transcripts = xml.findall('transcript')
if transcripts:
field_data['transcripts'] = {tr.get('language'): tr.get('src') for tr in transcripts}
for attr, value in xml.items():
if attr in compat_keys: # lint-amnesty, pylint: disable=consider-using-get
attr = compat_keys[attr]
if attr in cls.metadata_to_strip + ('url_name', 'name'):
continue
if attr == 'youtube':
speeds = cls._parse_youtube(value)
for speed, youtube_id in speeds.items():
# should have made these youtube_id_1_00 for
# cleanliness, but hindsight doesn't need glasses
normalized_speed = speed[:-1] if speed.endswith('0') else speed
# If the user has specified html5 sources, make sure we don't use the default video
if youtube_id != '' or 'html5_sources' in field_data:
field_data['youtube_id_{}'.format(normalized_speed.replace('.', '_'))] = youtube_id
elif attr in conversions:
field_data[attr] = conversions[attr](value)
elif attr not in cls.fields: # lint-amnesty, pylint: disable=unsupported-membership-test
field_data.setdefault('xml_attributes', {})[attr] = value
else:
# We export values with json.dumps (well, except for Strings, but
# for about a month we did it for Strings also).
field_data[attr] = deserialize_field(cls.fields[attr], value) # lint-amnesty, pylint: disable=unsubscriptable-object
course_id = getattr(id_generator, 'target_course_id', None)
# Update the handout location with current course_id
if 'handout' in field_data and course_id:
handout_location = StaticContent.get_location_from_path(field_data['handout'])
if isinstance(handout_location, AssetLocator):
handout_new_location = StaticContent.compute_location(course_id, handout_location.path)
field_data['handout'] = StaticContent.serialize_asset_key_with_slash(handout_new_location)
# For backwards compatibility: Add `source` if XML doesn't have `download_video`
# attribute.
if 'download_video' not in field_data and sources:
field_data['source'] = field_data['html5_sources'][0]
# For backwards compatibility: if XML doesn't have `download_track` attribute,
# it means that it is an old format. So, if `track` has some value,
# `download_track` needs to have value `True`.
if 'download_track' not in field_data and track is not None:
field_data['download_track'] = True
# load license if it exists
field_data = LicenseMixin.parse_license_from_xml(field_data, xml)
return field_data
def import_video_info_into_val(self, xml, resource_fs, course_id):
"""
Import parsed video info from `xml` into edxval.
Arguments:
xml (lxml object): xml representation of video to be imported.
resource_fs (OSFS): Import file system.
course_id (str): course id
"""
edx_video_id = clean_video_id(self.edx_video_id)
# Create video_asset is not already present.
video_asset_elem = xml.find('video_asset')
if video_asset_elem is None:
video_asset_elem = etree.Element('video_asset')
# This will be a dict containing the list of names of the external transcripts.
# Example:
# {
# 'en': ['The_Flash.srt', 'Harry_Potter.srt'],
# 'es': ['Green_Arrow.srt']
# }
external_transcripts = defaultdict(list)
# Add trancript from self.sub and self.youtube_id_1_0 fields.
external_transcripts['en'] = [
subs_filename(transcript, 'en')
for transcript in [self.sub, self.youtube_id_1_0] if transcript
]
for language_code, transcript in self.transcripts.items():
external_transcripts[language_code].append(transcript)
if edxval_api:
edx_video_id = edxval_api.import_from_xml(
video_asset_elem,
edx_video_id,
resource_fs,
EXPORT_IMPORT_STATIC_DIR,
external_transcripts,
course_id=course_id
)
return edx_video_id
def index_dictionary(self):
xblock_body = super().index_dictionary()
video_body = {
"display_name": self.display_name,
}
def _update_transcript_for_index(language=None):
""" Find video transcript - if not found, don't update index """
try:
transcript = get_transcript(self, lang=language, output_format=Transcript.TXT)[0].replace("\n", " ")
transcript_index_name = f"transcript_{language if language else self.transcript_language}"
video_body.update({transcript_index_name: transcript})
except NotFoundError:
pass
if self.sub:
_update_transcript_for_index()
# Check to see if there are transcripts in other languages besides default transcript
if self.transcripts:
for language in self.transcripts.keys():
_update_transcript_for_index(language)
if "content" in xblock_body:
xblock_body["content"].update(video_body)
else:
xblock_body["content"] = video_body
xblock_body["content_type"] = "Video"
return xblock_body
@property
def request_cache(self):
"""
Returns the request_cache from the runtime.
"""
return self.runtime.service(self, "request_cache")
@classmethod
@request_cached(
request_cache_getter=lambda args, kwargs: args[1],
)
def get_cached_val_data_for_course(cls, request_cache, video_profile_names, course_id): # lint-amnesty, pylint: disable=unused-argument
"""
Returns the VAL data for the requested video profiles for the given course.
"""
return edxval_api.get_video_info_for_course_and_profiles(str(course_id), video_profile_names)
def student_view_data(self, context=None):
"""
Returns a JSON representation of the student_view of this XModule.
The contract of the JSON content is between the caller and the particular XModule.
"""
context = context or {}
# If the "only_on_web" field is set on this video, do not return the rest of the video's data
# in this json view, since this video is to be accessed only through its web view."
if self.only_on_web:
return {"only_on_web": True}
encoded_videos = {}
val_video_data = {}
all_sources = self.html5_sources or []
# Check in VAL data first if edx_video_id exists
if self.edx_video_id:
video_profile_names = context.get("profiles", ["mobile_low", 'desktop_mp4', 'desktop_webm', 'mobile_high'])
if HLSPlaybackEnabledFlag.feature_enabled(self.location.course_key) and 'hls' not in video_profile_names:
video_profile_names.append('hls')
# get and cache bulk VAL data for course
val_course_data = self.get_cached_val_data_for_course(
self.request_cache,
video_profile_names,
self.location.course_key,
)
val_video_data = val_course_data.get(self.edx_video_id, {})
# Get the encoded videos if data from VAL is found
if val_video_data:
encoded_videos = val_video_data.get('profiles', {})
# If information for this edx_video_id is not found in the bulk course data, make a
# separate request for this individual edx_video_id, unless cache misses are disabled.
# This is useful/required for videos that don't have a course designated, such as the introductory video
# that is shared across many courses. However, this results in a separate database request so watch
# out for any performance hit if many such videos exist in a course. Set the 'allow_cache_miss' parameter
# to False to disable this fall back.
elif context.get("allow_cache_miss", "True").lower() == "true":
try:
val_video_data = edxval_api.get_video_info(self.edx_video_id)
# Unfortunately, the VAL API is inconsistent in how it returns the encodings, so remap here.
for enc_vid in val_video_data.pop('encoded_videos'):
if enc_vid['profile'] in video_profile_names:
encoded_videos[enc_vid['profile']] = {key: enc_vid[key] for key in ["url", "file_size"]}
except edxval_api.ValVideoNotFoundError:
pass
# Fall back to other video URLs in the video block if not found in VAL
if not encoded_videos:
if all_sources:
encoded_videos["fallback"] = {
"url": all_sources[0],
"file_size": 0, # File size is unknown for fallback URLs
}
# Include youtube link if there is no encoding for mobile- ie only a fallback URL or no encodings at all
# We are including a fallback URL for older versions of the mobile app that don't handle Youtube urls
if self.youtube_id_1_0:
encoded_videos["youtube"] = {
"url": self.create_youtube_url(self.youtube_id_1_0),
"file_size": 0, # File size is not relevant for external link
}
available_translations = self.available_translations(self.get_transcripts_info())
transcripts = {
lang: self.runtime.handler_url(self, 'transcript', 'download', query="lang=" + lang, thirdparty=True)
for lang in available_translations
}
return {
"only_on_web": self.only_on_web,
"duration": val_video_data.get('duration', None),
"transcripts": transcripts,
"encoded_videos": encoded_videos,
"all_sources": all_sources,
}

601
xmodule/video_block/video_handlers.py Normal file
View File

@@ -0,0 +1,601 @@
"""
Handlers for video block.
StudentViewHandlers are handlers for video block instance.
StudioViewHandlers are handlers for video descriptor instance.
"""
import json
import logging
import math
from django.core.files.base import ContentFile
from django.utils.timezone import now
from edxval.api import create_external_video, create_or_update_video_transcript, delete_video_transcript
from opaque_keys.edx.locator import CourseLocator
from webob import Response
from xblock.core import XBlock
from xblock.exceptions import JsonHandlerError
from xmodule.exceptions import NotFoundError
from xmodule.fields import RelativeTime
from .transcripts_utils import (
Transcript,
TranscriptException,
TranscriptsGenerationException,
clean_video_id,
generate_sjson_for_all_speeds,
get_html5_ids,
get_or_create_sjson,
get_transcript,
get_transcript_from_contentstore,
remove_subs_from_store,
subs_filename,
youtube_speed_dict
)
log = logging.getLogger(__name__)
# Disable no-member warning:
# pylint: disable=no-member
def to_boolean(value):
"""
Convert a value from a GET or POST request parameter to a bool
"""
if isinstance(value, bytes):
value = value.decode('ascii', errors='replace')
if isinstance(value, str):
return value.lower() == 'true'
else:
return bool(value)
class VideoStudentViewHandlers:
"""
Handlers for video block instance.
"""
global_speed = None
transcript_language = None
def handle_ajax(self, dispatch, data):
"""
Update values of xfields, that were changed by student.
"""
accepted_keys = [
'speed', 'auto_advance', 'saved_video_position', 'transcript_language',
'transcript_download_format', 'youtube_is_available',
'bumper_last_view_date', 'bumper_do_not_show_again'
]
conversions = {
'speed': json.loads,
'auto_advance': json.loads,
'saved_video_position': RelativeTime.isotime_to_timedelta,
'youtube_is_available': json.loads,
'bumper_last_view_date': to_boolean,
'bumper_do_not_show_again': to_boolean,
}
if dispatch == 'save_user_state':
for key in data:
if key in accepted_keys:
if key in conversions:
value = conversions[key](data[key])
else:
value = data[key]
if key == 'bumper_last_view_date':
value = now()
if key == 'speed' and math.isnan(value):
message = f"Invalid speed value {value}, must be a float."
log.warning(message)
return json.dumps({'success': False, 'error': message})
setattr(self, key, value)
if key == 'speed':
self.global_speed = self.speed
return json.dumps({'success': True})
log.debug(f"GET {data}")
log.debug(f"DISPATCH {dispatch}")
raise NotFoundError('Unexpected dispatch type')
def translation(self, youtube_id, transcripts):
"""
This is called to get transcript file for specific language.
youtube_id: str: must be one of youtube_ids or None if HTML video
transcripts (dict): A dict with all transcripts and a sub.
Logic flow:
If youtube_id doesn't exist, we have a video in HTML5 mode. Otherwise,
video in Youtube or Flash modes.
if youtube:
If english -> give back youtube_id subtitles:
Return what we have in contentstore for given youtube_id.
If non-english:
a) extract youtube_id from srt file name.
b) try to find sjson by youtube_id and return if successful.
c) generate sjson from srt for all youtube speeds.
if non-youtube:
If english -> give back `sub` subtitles:
Return what we have in contentstore for given subs_if that is stored in self.sub.
If non-english:
a) try to find previously generated sjson.
b) otherwise generate sjson from srt and return it.
Filenames naming:
en: subs_videoid.srt.sjson
non_en: uk_subs_videoid.srt.sjson
Raises:
NotFoundError if for 'en' subtitles no asset is uploaded.
NotFoundError if youtube_id does not exist / invalid youtube_id
"""
sub, other_lang = transcripts["sub"], transcripts["transcripts"]
if youtube_id:
# Youtube case:
if self.transcript_language == 'en':
return Transcript.asset(self.location, youtube_id).data
youtube_ids = youtube_speed_dict(self)
if youtube_id not in youtube_ids:
log.info("Youtube_id %s does not exist", youtube_id)
raise NotFoundError
try:
sjson_transcript = Transcript.asset(self.location, youtube_id, self.transcript_language).data
except NotFoundError:
log.info("Can't find content in storage for %s transcript: generating.", youtube_id)
generate_sjson_for_all_speeds(
self,
other_lang[self.transcript_language],
{speed: youtube_id for youtube_id, speed in youtube_ids.items()},
self.transcript_language
)
sjson_transcript = Transcript.asset(self.location, youtube_id, self.transcript_language).data
return sjson_transcript
else:
# HTML5 case
if self.transcript_language == 'en':
if '.srt' not in sub: # not bumper case
return Transcript.asset(self.location, sub).data
try:
return get_or_create_sjson(self, {'en': sub})
except TranscriptException:
pass # to raise NotFoundError and try to get data in get_static_transcript
elif other_lang:
return get_or_create_sjson(self, other_lang)
raise NotFoundError
def get_static_transcript(self, request, transcripts):
"""
Courses that are imported with the --nostatic flag do not show
transcripts/captions properly even if those captions are stored inside
their static folder. This adds a last resort method of redirecting to
the static asset path of the course if the transcript can't be found
inside the contentstore and the course has the static_asset_path field
set.
transcripts (dict): A dict with all transcripts and a sub.
"""
response = Response(status=404)
# Only do redirect for English
if not self.transcript_language == 'en':
return response
# If this video lives in library, the code below is not relevant and will error.
if not isinstance(self.course_id, CourseLocator):
return response
video_id = request.GET.get('videoId', None)
if video_id:
transcript_name = video_id
else:
transcript_name = transcripts["sub"]
if transcript_name:
# Get the asset path for course
asset_path = None
course = self.runtime.modulestore.get_course(self.course_id)
if course.static_asset_path:
asset_path = course.static_asset_path
else:
# It seems static_asset_path is not set in any XMLModuleStore courses.
asset_path = getattr(course, 'data_dir', '')
if asset_path:
response = Response(
status=307,
location='/static/{}/{}'.format(
asset_path,
subs_filename(transcript_name, self.transcript_language)
)
)
return response
@XBlock.json_handler
def publish_completion(self, data, dispatch): # pylint: disable=unused-argument
"""
Entry point for completion for student_view.
Parameters:
data: JSON dict:
key: "completion"
value: float in range [0.0, 1.0]
dispatch: Ignored.
Return value: JSON response (200 on success, 400 for malformed data)
"""
completion_service = self.runtime.service(self, 'completion')
if completion_service is None:
raise JsonHandlerError(500, "No completion service found")
if not completion_service.completion_tracking_enabled():
raise JsonHandlerError(404, "Completion tracking is not enabled and API calls are unexpected")
if not isinstance(data['completion'], (int, float)):
message = "Invalid completion value {}. Must be a float in range [0.0, 1.0]"
raise JsonHandlerError(400, message.format(data['completion']))
if not 0.0 <= data['completion'] <= 1.0:
message = "Invalid completion value {}. Must be in range [0.0, 1.0]"
raise JsonHandlerError(400, message.format(data['completion']))
self.runtime.publish(self, "completion", data)
return {"result": "ok"}
@staticmethod
def make_transcript_http_response(content, filename, language, content_type, add_attachment_header=True):
"""
Construct `Response` object.
Arguments:
content (unicode): transcript content
filename (unicode): transcript filename
language (unicode): transcript language
mimetype (unicode): transcript content type
add_attachment_header (bool): whether to add attachment header or not
"""
headerlist = [
('Content-Language', language),
]
if add_attachment_header:
headerlist.append(
(
'Content-Disposition',
f'attachment; filename="{filename}"'
)
)
response = Response(
content,
headerlist=headerlist,
charset='utf8'
)
response.content_type = content_type
return response
@XBlock.handler
def transcript(self, request, dispatch):
"""
Entry point for transcript handlers for student_view.
Request GET contains:
(optional) `videoId` for `translation` dispatch.
`is_bumper=1` flag for bumper case.
Dispatches, (HTTP GET):
/translation/[language_id]
/download
/available_translations/
Explanations:
`download`: returns SRT or TXT file.
`translation`: depends on HTTP methods:
Provide translation for requested language, SJSON format is sent back on success,
Proper language_id should be in url.
`available_translations`:
Returns list of languages, for which transcript files exist.
For 'en' check if SJSON exists. For non-`en` check if SRT file exists.
"""
is_bumper = request.GET.get('is_bumper', False)
transcripts = self.get_transcripts_info(is_bumper)
if dispatch.startswith('translation'):
language = dispatch.replace('translation', '').strip('/')
if not language:
log.info("Invalid /translation request: no language.")
return Response(status=400)
if language not in ['en'] + list(transcripts["transcripts"].keys()):
log.info("Video: transcript facilities are not available for given language.")
return Response(status=404)
if language != self.transcript_language:
self.transcript_language = language
try:
if is_bumper:
content, filename, mimetype = get_transcript_from_contentstore(
self,
self.transcript_language,
Transcript.SJSON,
transcripts
)
else:
content, filename, mimetype = get_transcript(
self,
lang=self.transcript_language,
output_format=Transcript.SJSON,
youtube_id=request.GET.get('videoId'),
)
response = self.make_transcript_http_response(
content,
filename,
self.transcript_language,
mimetype,
add_attachment_header=False
)
except NotFoundError as exc:
edx_video_id = clean_video_id(self.edx_video_id)
log.warning(
'[Translation Dispatch] %s: %s',
self.location,
exc if is_bumper else f'Transcript not found for {edx_video_id}, lang: {self.transcript_language}',
)
response = self.get_static_transcript(request, transcripts)
elif dispatch == 'download':
lang = request.GET.get('lang', None)
try:
content, filename, mimetype = get_transcript(self, lang, output_format=self.transcript_download_format)
except NotFoundError:
return Response(status=404)
response = self.make_transcript_http_response(
content,
filename,
self.transcript_language,
mimetype
)
elif dispatch.startswith('available_translations'):
available_translations = self.available_translations(
transcripts,
verify_assets=True,
is_bumper=is_bumper
)
if available_translations:
response = Response(json.dumps(available_translations))
response.content_type = 'application/json'
else:
response = Response(status=404)
else: # unknown dispatch
log.debug("Dispatch is not allowed")
response = Response(status=404)
return response
@XBlock.handler
def student_view_user_state(self, request, suffix=''): # lint-amnesty, pylint: disable=unused-argument
"""
Endpoint to get user-specific state, like current position and playback speed,
without rendering the full student_view HTML. This is similar to student_view_state,
but that one cannot contain user-specific info.
"""
view_state = self.student_view_data()
view_state.update({
"saved_video_position": self.saved_video_position.total_seconds(),
"speed": self.speed,
})
return Response(
json.dumps(view_state),
content_type='application/json',
charset='UTF-8'
)
@XBlock.handler
def yt_video_metadata(self, request, suffix=''): # lint-amnesty, pylint: disable=unused-argument
"""
Endpoint to get YouTube metadata.
This handler is only used in the Blockstore-based runtime. The old
runtime uses a similar REST API that's not an XBlock handler.
"""
from lms.djangoapps.courseware.views.views import load_metadata_from_youtube
if not self.youtube_id_1_0:
# TODO: more informational response to explain that yt_video_metadata not supported for non-youtube videos.
return Response('{}', status=400)
metadata, status_code = load_metadata_from_youtube(video_id=self.youtube_id_1_0, request=request)
response = Response(json.dumps(metadata), status=status_code)
response.content_type = 'application/json'
return response
class VideoStudioViewHandlers:
"""
Handlers for Studio view.
"""
def validate_transcript_upload_data(self, data):
"""
Validates video transcript file.
Arguments:
data: Transcript data to be validated.
Returns:
None or String
If there is error returns error message otherwise None.
"""
error = None
_ = self.runtime.service(self, "i18n").ugettext
# Validate the must have attributes - this error is unlikely to be faced by common users.
must_have_attrs = ['edx_video_id', 'language_code', 'new_language_code']
missing = [attr for attr in must_have_attrs if attr not in data]
# Get available transcript languages.
transcripts = self.get_transcripts_info()
available_translations = self.available_translations(transcripts, verify_assets=True)
if missing:
error = _('The following parameters are required: {missing}.').format(missing=', '.join(missing))
elif (
data['language_code'] != data['new_language_code'] and data['new_language_code'] in available_translations
):
error = _('A transcript with the "{language_code}" language code already exists.').format(
language_code=data['new_language_code'],
)
elif 'file' not in data:
error = _('A transcript file is required.')
return error
@XBlock.handler
def studio_transcript(self, request, dispatch):
"""
Entry point for Studio transcript handlers.
Dispatches:
/translation/[language_id] - language_id sould be in url.
`translation` dispatch support following HTTP methods:
`POST`:
Upload srt file. Check possibility of generation of proper sjson files.
For now, it works only for self.transcripts, not for `en`.
Do not update self.transcripts, as fields are updated on save in Studio.
`GET:
Return filename from storage. SRT format is sent back on success. Filename should be in GET dict.
We raise all exceptions right in Studio:
NotFoundError:
Video or asset was deleted from module/contentstore, but request came later.
Seems impossible to be raised. module_render.py catches NotFoundErrors from here.
/translation POST:
TypeError:
Unjsonable filename or content.
TranscriptsGenerationException, TranscriptException:
no SRT extension or not parse-able by PySRT
UnicodeDecodeError: non-UTF8 uploaded file content encoding.
"""
_ = self.runtime.service(self, "i18n").ugettext
if dispatch.startswith('translation'):
if request.method == 'POST':
error = self.validate_transcript_upload_data(data=request.POST)
if error:
response = Response(json={'error': error}, status=400)
else:
edx_video_id = clean_video_id(request.POST['edx_video_id'])
language_code = request.POST['language_code']
new_language_code = request.POST['new_language_code']
transcript_file = request.POST['file'].file
if not edx_video_id:
# Back-populate the video ID for an external video.
# pylint: disable=attribute-defined-outside-init
self.edx_video_id = edx_video_id = create_external_video(display_name='external video')
try:
# Convert SRT transcript into an SJSON format
# and upload it to S3.
sjson_subs = Transcript.convert(
content=transcript_file.read().decode('utf-8'),
input_format=Transcript.SRT,
output_format=Transcript.SJSON
).encode()
create_or_update_video_transcript(
video_id=edx_video_id,
language_code=language_code,
metadata={
'file_format': Transcript.SJSON,
'language_code': new_language_code
},
file_data=ContentFile(sjson_subs),
)
payload = {
'edx_video_id': edx_video_id,
'language_code': new_language_code
}
response = Response(json.dumps(payload), status=201)
except (TranscriptsGenerationException, UnicodeDecodeError):
response = Response(
json={
'error': _(
'There is a problem with this transcript file. Try to upload a different file.'
)
},
status=400
)
elif request.method == 'DELETE':
request_data = request.json
if 'lang' not in request_data or 'edx_video_id' not in request_data:
return Response(status=400)
language = request_data['lang']
edx_video_id = clean_video_id(request_data['edx_video_id'])
if edx_video_id:
delete_video_transcript(video_id=edx_video_id, language_code=language)
if language == 'en':
# remove any transcript file from content store for the video ids
possible_sub_ids = [
self.sub, # pylint: disable=access-member-before-definition
self.youtube_id_1_0
] + get_html5_ids(self.html5_sources)
for sub_id in possible_sub_ids:
remove_subs_from_store(sub_id, self, language)
# update metadata as `en` can also be present in `transcripts` field
remove_subs_from_store(self.transcripts.pop(language, None), self, language)
# also empty `sub` field
self.sub = '' # pylint: disable=attribute-defined-outside-init
else:
remove_subs_from_store(self.transcripts.pop(language, None), self, language)
return Response(status=200)
elif request.method == 'GET':
language = request.GET.get('language_code')
if not language:
return Response(json={'error': _('Language is required.')}, status=400)
try:
transcript_content, transcript_name, mime_type = get_transcript(
video=self, lang=language, output_format=Transcript.SRT
)
response = Response(transcript_content, headerlist=[
(
'Content-Disposition',
f'attachment; filename="{transcript_name}"'
),
('Content-Language', language),
('Content-Type', mime_type)
])
except (UnicodeDecodeError, TranscriptsGenerationException, NotFoundError):
response = Response(status=404)
else:
# Any other HTTP method is not allowed.
response = Response(status=404)
else: # unknown dispatch
log.debug("Dispatch is not allowed")
response = Response(status=404)
return response

123
xmodule/video_block/video_utils.py Normal file
View File

@@ -0,0 +1,123 @@
"""
Module contains utils specific for video_block but not for transcripts.
"""
import logging
from collections import OrderedDict
from urllib.parse import parse_qs, urlencode, urlparse, urlsplit, urlunsplit
from django.conf import settings
from django.core.exceptions import ValidationError
from django.core.validators import URLValidator
log = logging.getLogger(__name__)
def create_youtube_string(module):
"""
Create a string of Youtube IDs from `module`'s metadata
attributes. Only writes a speed if an ID is present in the
module. Necessary for backwards compatibility with XML-based
courses.
"""
youtube_ids = [
module.youtube_id_0_75,
module.youtube_id_1_0,
module.youtube_id_1_25,
module.youtube_id_1_5
]
youtube_speeds = ['0.75', '1.00', '1.25', '1.50']
return ','.join([
':'.join(pair)
for pair
in zip(youtube_speeds, youtube_ids)
if pair[1]
])
def rewrite_video_url(cdn_base_url, original_video_url):
"""
Returns a re-written video URL for cases when an alternate source
has been configured and is selected using factors like
user location.
Re-write rules for country codes are specified via the
EDX_VIDEO_CDN_URLS configuration structure.
:param cdn_base_url: The scheme, hostname, port and any relevant path prefix for the alternate CDN,
for example: https://mirror.example.cn/edx
:param original_video_url: The canonical source for this video, for example:
https://cdn.example.com/edx-course-videos/VIDEO101/001.mp4
:return: The re-written URL
"""
if (not cdn_base_url) or (not original_video_url):
return None
parsed = urlparse(original_video_url)
# Contruction of the rewrite url is intentionally very flexible of input.
# For example, https://www.edx.org/ + /foo.html will be rewritten to
# https://www.edx.org/foo.html.
rewritten_url = cdn_base_url.rstrip("/") + "/" + parsed.path.lstrip("/")
validator = URLValidator()
try:
validator(rewritten_url)
return rewritten_url
except ValidationError:
log.warning("Invalid CDN rewrite URL encountered, %s", rewritten_url)
# Mimic the behavior of removed get_video_from_cdn in this regard and
# return None causing the caller to use the original URL.
return None
def get_poster(video):
"""
Generate poster metadata.
youtube_streams is string that contains '1.00:youtube_id'
Poster metadata is dict of youtube url for image thumbnail and edx logo
"""
if not video.bumper.get("enabled"):
return
poster = OrderedDict({"url": "", "type": ""})
if video.youtube_streams:
youtube_id = video.youtube_streams.split('1.00:')[1].split(',')[0]
poster["url"] = settings.YOUTUBE['IMAGE_API'].format(youtube_id=youtube_id)
poster["type"] = "youtube"
else:
poster["url"] = "https://www.edx.org/sites/default/files/theme/edx-logo-header.png"
poster["type"] = "html5"
return poster
def format_xml_exception_message(location, key, value):
"""
Generate exception message for VideoBlock class which will use for ValueError and UnicodeDecodeError
when setting xml attributes.
"""
exception_message = "Block-location:{location}, Key:{key}, Value:{value}".format(
location=str(location),
key=key,
value=value
)
return exception_message
def set_query_parameter(url, param_name, param_value):
"""
Given a URL, set or replace a query parameter and return the
modified URL.
"""
scheme, netloc, path, query_string, fragment = urlsplit(url)
query_params = parse_qs(query_string)
query_params[param_name] = [param_value]
new_query_string = urlencode(query_params, doseq=True)
return urlunsplit((scheme, netloc, path, new_query_string, fragment))

222
xmodule/video_block/video_xfields.py Normal file
View File

@@ -0,0 +1,222 @@
""" # lint-amnesty, pylint: disable=cyclic-import
XFields for video block.
"""
import datetime
from xblock.fields import Boolean, DateTime, Dict, Float, List, Scope, String
from xmodule.fields import RelativeTime
# Make '_' a no-op so we can scrape strings. Using lambda instead of
# `django.utils.translation.ugettext_noop` because Django cannot be imported in this file
_ = lambda text: text
class VideoFields:
"""Fields for `VideoBlock`."""
display_name = String(
help=_("The display name for this component."),
display_name=_("Component Display Name"),
default="Video",
scope=Scope.settings
)
saved_video_position = RelativeTime(
help=_("Current position in the video."),
scope=Scope.user_state,
default=datetime.timedelta(seconds=0)
)
# TODO: This should be moved to Scope.content, but this will
# require data migration to support the old video block.
youtube_id_1_0 = String(
help=_("Optional, for older browsers: the YouTube ID for the normal speed video."),
display_name=_("YouTube ID"),
scope=Scope.settings,
default="3_yD_cEKoCk"
)
youtube_id_0_75 = String(
help=_("Optional, for older browsers: the YouTube ID for the .75x speed video."),
display_name=_("YouTube ID for .75x speed"),
scope=Scope.settings,
default=""
)
youtube_id_1_25 = String(
help=_("Optional, for older browsers: the YouTube ID for the 1.25x speed video."),
display_name=_("YouTube ID for 1.25x speed"),
scope=Scope.settings,
default=""
)
youtube_id_1_5 = String(
help=_("Optional, for older browsers: the YouTube ID for the 1.5x speed video."),
display_name=_("YouTube ID for 1.5x speed"),
scope=Scope.settings,
default=""
)
start_time = RelativeTime( # datetime.timedelta object
help=_(
"Time you want the video to start if you don't want the entire video to play. "
"Not supported in the native mobile app: the full video file will play. "
"Formatted as HH:MM:SS. The maximum value is 23:59:59."
),
display_name=_("Video Start Time"),
scope=Scope.settings,
default=datetime.timedelta(seconds=0)
)
end_time = RelativeTime( # datetime.timedelta object
help=_(
"Time you want the video to stop if you don't want the entire video to play. "
"Not supported in the native mobile app: the full video file will play. "
"Formatted as HH:MM:SS. The maximum value is 23:59:59."
),
display_name=_("Video Stop Time"),
scope=Scope.settings,
default=datetime.timedelta(seconds=0)
)
#front-end code of video player checks logical validity of (start_time, end_time) pair.
download_video = Boolean(
help=_("Allow students to download versions of this video in different formats if they cannot use the edX video"
" player or do not have access to YouTube. You must add at least one non-YouTube URL "
"in the Video File URLs field."),
display_name=_("Video Download Allowed"),
scope=Scope.settings,
default=False
)
html5_sources = List(
help=_("The URL or URLs where you've posted non-YouTube versions of the video. Each URL must end in .mpeg,"
" .mp4, .ogg, or .webm and cannot be a YouTube URL. (For browser compatibility, we strongly recommend"
" .mp4 and .webm format.) Students will be able to view the first listed video that's compatible with"
" the student's computer. To allow students to download these videos, "
"set Video Download Allowed to True."),
display_name=_("Video File URLs"),
scope=Scope.settings,
)
track = String(
help=_("By default, students can download an .srt or .txt transcript when you set Download Transcript "
"Allowed to True. If you want to provide a downloadable transcript in a different format, we recommend "
"that you upload a handout by using the Upload a Handout field. If this isn't possible, you can post a "
"transcript file on the Files & Uploads page or on the Internet, and then add the URL for the "
"transcript here. Students see a link to download that transcript below the video."),
display_name=_("Downloadable Transcript URL"),
scope=Scope.settings,
default=''
)
download_track = Boolean(
help=_("Allow students to download the timed transcript. A link to download the file appears below the video."
" By default, the transcript is an .srt or .txt file. If you want to provide the transcript for "
"download in a different format, upload a file by using the Upload Handout field."),
display_name=_("Download Transcript Allowed"),
scope=Scope.settings,
default=False
)
# `sub` is deprecated field and should not be used in future. Now, transcripts are primarily handled in VAL and
# backward compatibility for the video blocks already using this field has been ensured.
sub = String(
help=_("The default transcript for the video, from the Default Timed Transcript field on the Basic tab. "
"This transcript should be in English. You don't have to change this setting."),
display_name=_("Default Timed Transcript"),
scope=Scope.settings,
default=""
)
show_captions = Boolean(
help=_("Specify whether the transcripts appear with the video by default."),
display_name=_("Show Transcript"),
scope=Scope.settings,
default=True
)
# Data format: {'de': 'german_translation', 'uk': 'ukrainian_translation'}
transcripts = Dict(
help=_("Add transcripts in different languages."
" Click below to specify a language and upload an .srt transcript file for that language."),
display_name=_("Transcript Languages"),
scope=Scope.settings,
default={}
)
transcript_language = String(
help=_("Preferred language for transcript."),
display_name=_("Preferred language for transcript"),
scope=Scope.preferences,
default="en"
)
transcript_download_format = String(
help=_("Transcript file format to download by user."),
scope=Scope.preferences,
values=[
# Translators: This is a type of file used for captioning in the video player.
{"display_name": _("SubRip (.srt) file"), "value": "srt"},
{"display_name": _("Text (.txt) file"), "value": "txt"}
],
default='srt',
)
speed = Float(
help=_("The last speed that the user specified for the video."),
scope=Scope.user_state
)
global_speed = Float(
help=_("The default speed for the video."),
scope=Scope.preferences,
default=1.0
)
auto_advance = Boolean(
help=_("Specify whether to advance automatically to the next unit when the video ends."),
scope=Scope.preferences,
# The default is True because this field only has an effect when auto-advance controls are enabled
# (globally enabled through feature flag and locally enabled through course setting); in that case
# it's good to start auto-advancing and let the student disable it, instead of the other way around
# (requiring the user to enable it). When auto-advance controls are hidden, this field won't be used.
default=True,
)
youtube_is_available = Boolean(
help=_("Specify whether YouTube is available for the user."),
scope=Scope.user_info,
default=True
)
handout = String(
help=_("Upload a handout to accompany this video. Students can download the handout by "
"clicking Download Handout under the video."),
display_name=_("Upload Handout"),
scope=Scope.settings,
)
only_on_web = Boolean(
help=_(
"Specify whether access to this video is limited to browsers only, or if it can be "
"accessed from other applications including mobile apps."
),
display_name=_("Video Available on Web Only"),
scope=Scope.settings,
default=False
)
edx_video_id = String(
help=_("If you were assigned a Video ID by edX for the video to play in this component, enter the ID here."
" In this case, do not enter values in the Default Video URL, the Video File URLs, "
"and the YouTube ID fields. If you were not assigned a Video ID,"
" enter values in those other fields and ignore this field."),
display_name=_("Video ID"),
scope=Scope.settings,
default="",
)
bumper_last_view_date = DateTime(
display_name=_("Date of the last view of the bumper"),
scope=Scope.preferences,
)
bumper_do_not_show_again = Boolean(
display_name=_("Do not show bumper again"),
scope=Scope.preferences,
default=False,
)
public_access = Boolean(
help=_("Specify whether the video can be accessed publicly by learners."),
display_name=_("Public Access"),
scope=Scope.settings,
default=False
)
# thumbnail is need as a field for the new video editor. The field is hidden in
# the legacy modal.
thumbnail = String(
help=_("Add a specific thumbnail for learners to see before playing the video."),
display_name=_("Thumbnail"),
scope=Scope.settings,
default="",
)