6377635546

Committed 02 Oct 2023 08:21AM UTC coverage: 70.396% (-1.6%) from 71.99%

Build # 6377635546

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #900 from EsupPortail/develop

[DONE] #3.4.0

Run Details

1509 of 1509 new or added lines in 58 files covered. (100.0%)

9288 of 13194 relevant lines covered (70.4%)

0.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

75.0

/pod/video_encode_transcript/utils.py

"""Esup-Pod video encoding and transcripting utilities."""
import os
import bleach
import time

from django.urls import reverse
from django.conf import settings
from django.utils.translation import ugettext_lazy as _
from django.core.mail import mail_admins
from django.core.mail import send_mail
from django.core.mail import mail_managers
from django.core.mail import EmailMultiAlternatives
from pod.video.models import Video
from pod.progressive_web_app.utils import notify_user
from .models import EncodingStep
from .models import EncodingLog

DEBUG = getattr(settings, "DEBUG", True)

TEMPLATE_VISIBLE_SETTINGS = getattr(
    settings,
    "TEMPLATE_VISIBLE_SETTINGS",
    {
        "TITLE_SITE": "Pod",
        "TITLE_ETB": "University name",
        "LOGO_SITE": "img/logoPod.svg",
        "LOGO_ETB": "img/esup-pod.svg",
        "LOGO_PLAYER": "img/pod_favicon.svg",
        "LINK_PLAYER": "",
        "LINK_PLAYER_NAME": _("Home"),
        "FOOTER_TEXT": ("",),
        "FAVICON": "img/pod_favicon.svg",
        "CSS_OVERRIDE": "",
        "PRE_HEADER_TEMPLATE": "",
        "POST_FOOTER_TEMPLATE": "",
        "TRACKING_TEMPLATE": "",
    },
)

__TITLE_SITE__ = (
    TEMPLATE_VISIBLE_SETTINGS["TITLE_SITE"]
    if (TEMPLATE_VISIBLE_SETTINGS.get("TITLE_SITE"))
    else "Pod"
)

DEFAULT_FROM_EMAIL = getattr(settings, "DEFAULT_FROM_EMAIL", "noreply@univ.fr")

USE_ESTABLISHMENT_FIELD = getattr(settings, "USE_ESTABLISHMENT_FIELD", False)

MANAGERS = getattr(settings, "MANAGERS", {})

SECURE_SSL_REDIRECT = getattr(settings, "SECURE_SSL_REDIRECT", False)
VIDEOS_DIR = getattr(settings, "VIDEOS_DIR", "videos")


# ##########################################################################
# ENCODE VIDEO : GENERIC FUNCTIONS
# ##########################################################################


def change_encoding_step(video_id, num_step, desc):
    """Change encoding step."""
    encoding_step, created = EncodingStep.objects.get_or_create(
        video=Video.objects.get(id=video_id)
    )
    encoding_step.num_step = num_step
    encoding_step.desc_step = desc[:255]
    encoding_step.save()

    if DEBUG:
        print("step: %d - desc: %s" % (num_step, desc))


def add_encoding_log(video_id, log):
    """Add message in video_id encoding log."""
    encoding_log, created = EncodingLog.objects.get_or_create(
        video=Video.objects.get(id=video_id)
    )
    if created:
        encoding_log.log = log
    else:
        encoding_log.log += "\n\n%s" % log
    encoding_log.save()
    if DEBUG:
        print(log)


def check_file(path_file):
    """Check if path_file is accessible and is not null."""
    if os.access(path_file, os.F_OK) and os.stat(path_file).st_size > 0:
        return True
    return False


def create_outputdir(video_id, video_path):
    """ENCODE VIDEO: CREATE OUTPUT DIR."""
    dirname = os.path.dirname(video_path)
    output_dir = os.path.join(dirname, "%04d" % video_id)
    if not os.path.exists(output_dir):
        os.makedirs(output_dir)
    return output_dir


###############################################################
# EMAIL
###############################################################


def send_email_item(msg, item, item_id):
    """Send email notification when encoding fails for a specific item."""
    subject = "[" + __TITLE_SITE__ + "] Error Encoding %s id:%s" % (item, item_id)
    message = "Error Encoding  %s id : %s\n%s" % (item, item_id, msg)
    html_message = "<p>Error Encoding %s id : %s</p><p>%s</p>" % (
        item,
        item_id,
        msg.replace("\n", "<br>"),
    )
    mail_admins(subject, message, fail_silently=False, html_message=html_message)


def send_email_recording(msg, recording_id):
    """Send email notification when recording encoding failed."""
    send_email_item(msg, "Recording", recording_id)


def send_email_encoding(video_to_encode):
    """Send email on encoding completion."""
    subject_prefix = _("Encoding")
    send_notification_email(video_to_encode, subject_prefix)


def send_notification_encoding(video_to_encode):
    """Send push notification on encoding completion."""
    subject_prefix = _("Encoding")
    send_notification(video_to_encode, subject_prefix)


def send_email(msg, video_id):
    """Send email notification when video encoding failed."""
    send_email_item(msg, "Video", video_id)


def send_email_transcript(video_to_encode):
    """Send email on transcripting completion."""
    subject_prefix = _("The transcripting")
    send_notification_email(video_to_encode, subject_prefix)


def send_notification_email(video_to_encode, subject_prefix):
    """Send email notification on video encoding or transcripting completion."""
    if DEBUG:
        print("SEND EMAIL ON %s COMPLETION" % subject_prefix.upper())
    url_scheme = "https" if SECURE_SSL_REDIRECT else "http"
    content_url = "%s:%s" % (url_scheme, video_to_encode.get_full_url())
    subject = "[%s] %s" % (
        __TITLE_SITE__,
        _("%(subject)s #%(content_id)s completed")
        % {"subject": subject_prefix, "content_id": video_to_encode.id},
    )

    html_message = (
        '<p>%s</p><p>%s</p><p>%s<br><a href="%s"><i>%s</i></a>\
                </p><p>%s</p>'
        % (
            _("Hello,"),
            _(
                "%(content_type)s “%(content_title)s” has been %(action)s"
                + ", and is now available on %(site_title)s."
            )
            % {
                "content_type": (
                    _("The content")
                    if subject_prefix == _("The transcripting")
                    else _("The video")
                ),
                "content_title": "<b>%s</b>" % video_to_encode.title,
                "action": (
                    _("automatically transcripted")
                    if (subject_prefix == _("The transcripting"))
                    else _("encoded to Web formats")
                ),
                "site_title": __TITLE_SITE__,
            },
            _("You will find it here:"),
            content_url,
            content_url,
            _("Regards."),
        )
    )

    full_html_message = html_message + "<br>%s%s<br>%s%s" % (
        _("Post by:"),
        video_to_encode.owner,
        _("the:"),
        video_to_encode.date_added,
    )

    message = bleach.clean(html_message, tags=[], strip=True)
    full_message = bleach.clean(full_html_message, tags=[], strip=True)

    from_email = DEFAULT_FROM_EMAIL
    to_email = []
    to_email.append(video_to_encode.owner.email)

    if (
        USE_ESTABLISHMENT_FIELD
        and MANAGERS
        and video_to_encode.owner.owner.establishment.lower() in dict(MANAGERS)
    ):
        bcc_email = []
        video_estab = video_to_encode.owner.owner.establishment.lower()
        manager = dict(MANAGERS)[video_estab]
        if isinstance(manager, (list, tuple)):
            bcc_email = manager
        elif isinstance(manager, str):
            bcc_email.append(manager)
        msg = EmailMultiAlternatives(
            subject, message, from_email, to_email, bcc=bcc_email
        )
        msg.attach_alternative(html_message, "text/html")
        msg.send()
    else:
        mail_managers(
            subject,
            full_message,
            fail_silently=False,
            html_message=full_html_message,
        )
        if not DEBUG:
            send_mail(
                subject,
                message,
                from_email,
                to_email,
                fail_silently=False,
                html_message=html_message,
            )


def send_notification(video_to_encode, subject_prefix):
    """Send push notification on video encoding or transcripting completion."""
    subject = "[%s] %s" % (
        __TITLE_SITE__,
        _("%(subject)s #%(content_id)s completed")
        % {"subject": subject_prefix, "content_id": video_to_encode.id},
    )
    message = _(
        "%(content_type)s “%(content_title)s” has been %(action)s"
        + ", and is now available on %(site_title)s."
    ) % {
        "content_type": (
            _("content") if subject_prefix == _("Transcripting") else _("video")
        ),
        "content_title": video_to_encode.title,
        "action": (
            _("automatically transcripted")
            if (subject_prefix == _("Transcripting"))
            else _("encoded to Web formats")
        ),
        "site_title": __TITLE_SITE__,
    }

    notify_user(
        video_to_encode.owner,
        subject,
        message,
        url=reverse("video:video", args=(video_to_encode.slug,)),
    )


def time_to_seconds(a_time):
    """Convert a time to seconds."""
    seconds = time.strptime(str(a_time), "%H:%M:%S")
    seconds = seconds.tm_sec + seconds.tm_min * 60 + seconds.tm_hour * 3600
    return seconds

1	"""Esup-Pod video encoding and transcripting utilities."""
2	import os	1✔
3	import bleach	1✔
4	import time	1✔
5
6	from django.urls import reverse	1✔
7	from django.conf import settings	1✔
8	from django.utils.translation import ugettext_lazy as _	1✔
9	from django.core.mail import mail_admins	1✔
10	from django.core.mail import send_mail	1✔
11	from django.core.mail import mail_managers	1✔
12	from django.core.mail import EmailMultiAlternatives	1✔
13	from pod.video.models import Video	1✔
14	from pod.progressive_web_app.utils import notify_user	1✔
15	from .models import EncodingStep	1✔
16	from .models import EncodingLog	1✔
17
18	DEBUG = getattr(settings, "DEBUG", True)	1✔
19
20	TEMPLATE_VISIBLE_SETTINGS = getattr(	1✔
21	settings,
22	"TEMPLATE_VISIBLE_SETTINGS",
23	{
24	"TITLE_SITE": "Pod",
25	"TITLE_ETB": "University name",
26	"LOGO_SITE": "img/logoPod.svg",
27	"LOGO_ETB": "img/esup-pod.svg",
28	"LOGO_PLAYER": "img/pod_favicon.svg",
29	"LINK_PLAYER": "",
30	"LINK_PLAYER_NAME": _("Home"),
31	"FOOTER_TEXT": ("",),
32	"FAVICON": "img/pod_favicon.svg",
33	"CSS_OVERRIDE": "",
34	"PRE_HEADER_TEMPLATE": "",
35	"POST_FOOTER_TEMPLATE": "",
36	"TRACKING_TEMPLATE": "",
37	},
38	)
39
40	__TITLE_SITE__ = (	1✔
41	TEMPLATE_VISIBLE_SETTINGS["TITLE_SITE"]
42	if (TEMPLATE_VISIBLE_SETTINGS.get("TITLE_SITE"))
43	else "Pod"
44	)
45
46	DEFAULT_FROM_EMAIL = getattr(settings, "DEFAULT_FROM_EMAIL", "noreply@univ.fr")	1✔
47
48	USE_ESTABLISHMENT_FIELD = getattr(settings, "USE_ESTABLISHMENT_FIELD", False)	1✔
49
50	MANAGERS = getattr(settings, "MANAGERS", {})	1✔
51
52	SECURE_SSL_REDIRECT = getattr(settings, "SECURE_SSL_REDIRECT", False)	1✔
53	VIDEOS_DIR = getattr(settings, "VIDEOS_DIR", "videos")	1✔
54
55
56	# ##########################################################################
57	# ENCODE VIDEO : GENERIC FUNCTIONS
58	# ##########################################################################
59
60
61	def change_encoding_step(video_id, num_step, desc):	1✔
62	"""Change encoding step."""
63	encoding_step, created = EncodingStep.objects.get_or_create(	1✔
64	video=Video.objects.get(id=video_id)
65	)
66	encoding_step.num_step = num_step	1✔
67	encoding_step.desc_step = desc[:255]	1✔
68	encoding_step.save()	1✔
69
70	if DEBUG:	1✔
71	print("step: %d - desc: %s" % (num_step, desc))	1✔
72
73
74	def add_encoding_log(video_id, log):	1✔
75	"""Add message in video_id encoding log."""
76	encoding_log, created = EncodingLog.objects.get_or_create(	1✔
77	video=Video.objects.get(id=video_id)
78	)
79	if created:	1✔
80	encoding_log.log = log	1✔
81	else:
82	encoding_log.log += "\n\n%s" % log	×
83	encoding_log.save()	1✔
84	if DEBUG:	1✔
85	print(log)	1✔
86
87
88	def check_file(path_file):	1✔
89	"""Check if path_file is accessible and is not null."""
90	if os.access(path_file, os.F_OK) and os.stat(path_file).st_size > 0:	1✔
91	return True	1✔
92	return False	1✔
93
94
95	def create_outputdir(video_id, video_path):	1✔
96	"""ENCODE VIDEO: CREATE OUTPUT DIR."""
97	dirname = os.path.dirname(video_path)	×
98	output_dir = os.path.join(dirname, "%04d" % video_id)	×
99	if not os.path.exists(output_dir):	×
100	os.makedirs(output_dir)	×
101	return output_dir	×
102
103
104	###############################################################
105	# EMAIL
106	###############################################################
107
108
109	def send_email_item(msg, item, item_id):	1✔
110	"""Send email notification when encoding fails for a specific item."""
111	subject = "[" + __TITLE_SITE__ + "] Error Encoding %s id:%s" % (item, item_id)	1✔
112	message = "Error Encoding %s id : %s\n%s" % (item, item_id, msg)	1✔
113	html_message = "<p>Error Encoding %s id : %s</p><p>%s</p>" % (	1✔
114	item,
115	item_id,
116	msg.replace("\n", "<br>"),
117	)
118	mail_admins(subject, message, fail_silently=False, html_message=html_message)	1✔
119
120
121	def send_email_recording(msg, recording_id):	1✔
122	"""Send email notification when recording encoding failed."""
123	send_email_item(msg, "Recording", recording_id)	×
124
125
126	def send_email_encoding(video_to_encode):	1✔
127	"""Send email on encoding completion."""
128	subject_prefix = _("Encoding")	1✔
129	send_notification_email(video_to_encode, subject_prefix)	1✔
130
131
132	def send_notification_encoding(video_to_encode):	1✔
133	"""Send push notification on encoding completion."""
134	subject_prefix = _("Encoding")	×
135	send_notification(video_to_encode, subject_prefix)	×
136
137
138	def send_email(msg, video_id):	1✔
139	"""Send email notification when video encoding failed."""
140	send_email_item(msg, "Video", video_id)	1✔
141
142
143	def send_email_transcript(video_to_encode):	1✔
144	"""Send email on transcripting completion."""
145	subject_prefix = _("The transcripting")	×
146	send_notification_email(video_to_encode, subject_prefix)	×
147
148
149	def send_notification_email(video_to_encode, subject_prefix):	1✔
150	"""Send email notification on video encoding or transcripting completion."""
151	if DEBUG:	1✔
152	print("SEND EMAIL ON %s COMPLETION" % subject_prefix.upper())	1✔
153	url_scheme = "https" if SECURE_SSL_REDIRECT else "http"	1✔
154	content_url = "%s:%s" % (url_scheme, video_to_encode.get_full_url())	1✔
155	subject = "[%s] %s" % (	1✔
156	__TITLE_SITE__,
157	_("%(subject)s #%(content_id)s completed")
158	% {"subject": subject_prefix, "content_id": video_to_encode.id},
159	)
160
161	html_message = (	1✔
162	'<p>%s</p><p>%s</p><p>%s<br><a href="%s"><i>%s</i></a>\
163	</p><p>%s</p>'
164	% (
165	_("Hello,"),
166	_(
167	"%(content_type)s “%(content_title)s” has been %(action)s"
168	+ ", and is now available on %(site_title)s."
169	)
170	% {
171	"content_type": (
172	_("The content")
173	if subject_prefix == _("The transcripting")
174	else _("The video")
175	),
176	"content_title": "<b>%s</b>" % video_to_encode.title,
177	"action": (
178	_("automatically transcripted")
179	if (subject_prefix == _("The transcripting"))
180	else _("encoded to Web formats")
181	),
182	"site_title": __TITLE_SITE__,
183	},
184	_("You will find it here:"),
185	content_url,
186	content_url,
187	_("Regards."),
188	)
189	)
190
191	full_html_message = html_message + "<br>%s%s<br>%s%s" % (	1✔
192	_("Post by:"),
193	video_to_encode.owner,
194	_("the:"),
195	video_to_encode.date_added,
196	)
197
198	message = bleach.clean(html_message, tags=[], strip=True)	1✔
199	full_message = bleach.clean(full_html_message, tags=[], strip=True)	1✔
200
201	from_email = DEFAULT_FROM_EMAIL	1✔
202	to_email = []	1✔
203	to_email.append(video_to_encode.owner.email)	1✔
204
205	if (	1✔
206	USE_ESTABLISHMENT_FIELD
207	and MANAGERS
208	and video_to_encode.owner.owner.establishment.lower() in dict(MANAGERS)
209	):
210	bcc_email = []	×
211	video_estab = video_to_encode.owner.owner.establishment.lower()	×
212	manager = dict(MANAGERS)[video_estab]	×
213	if isinstance(manager, (list, tuple)):	×
214	bcc_email = manager	×
215	elif isinstance(manager, str):	×
216	bcc_email.append(manager)	×
217	msg = EmailMultiAlternatives(	×
218	subject, message, from_email, to_email, bcc=bcc_email
219	)
220	msg.attach_alternative(html_message, "text/html")	×
221	msg.send()	×
222	else:
223	mail_managers(	1✔
224	subject,
225	full_message,
226	fail_silently=False,
227	html_message=full_html_message,
228	)
229	if not DEBUG:	1✔
230	send_mail(	×
231	subject,
232	message,
233	from_email,
234	to_email,
235	fail_silently=False,
236	html_message=html_message,
237	)
238
239
240	def send_notification(video_to_encode, subject_prefix):	1✔
241	"""Send push notification on video encoding or transcripting completion."""
242	subject = "[%s] %s" % (	×
243	__TITLE_SITE__,
244	_("%(subject)s #%(content_id)s completed")
245	% {"subject": subject_prefix, "content_id": video_to_encode.id},
246	)
247	message = _(	×
248	"%(content_type)s “%(content_title)s” has been %(action)s"
249	+ ", and is now available on %(site_title)s."
250	) % {
251	"content_type": (
252	_("content") if subject_prefix == _("Transcripting") else _("video")
253	),
254	"content_title": video_to_encode.title,
255	"action": (
256	_("automatically transcripted")
257	if (subject_prefix == _("Transcripting"))
258	else _("encoded to Web formats")
259	),
260	"site_title": __TITLE_SITE__,
261	}
262
263	notify_user(	×
264	video_to_encode.owner,
265	subject,
266	message,
267	url=reverse("video:video", args=(video_to_encode.slug,)),
268	)
269
270
271	def time_to_seconds(a_time):	1✔
272	"""Convert a time to seconds."""
273	seconds = time.strptime(str(a_time), "%H:%M:%S")	1✔
274	seconds = seconds.tm_sec + seconds.tm_min * 60 + seconds.tm_hour * 3600	1✔
275	return seconds	1✔

EsupPortail / Esup-Pod / 6377635546

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous