Merge branch 'master' of git.fsmpi.rwth-aachen.de:videoaginfra/website

.gitlab-ci.yml

@@ -7,7 +7,7 @@ unittest:
     - python3 -V
     - uname -a
     - apt install -y sqlite3 locales-all git python3-flask python3-ldap3 python3-requests python3-lxml python3-icalendar python3-mysql.connector python3-requests python3-coverage
-    - python3 -m coverage run tests.py
+    - python3 -m coverage run runTests.py
     - python3 -m coverage report --include "./*"
     - python3 -m coverage report -m  --include "./*" > report.txt
     - python3 -m coverage html --include "./*"
@@ -33,4 +33,4 @@ deploy_staging:
     stage: deploy
     script:
     - pacman --noconfirm -Sy ansible git
-    
\ No newline at end of file
+    

README.md

@@ -15,9 +15,9 @@ Hinweis: diese Variante startet eine lokale Testversion der Website, es sind nic
 Alternativ, insbesondere zum Testen der Zugriffsbeschränkungen: Siehe `nginx.example.conf`.
 
 ### Unittests
-Tests können mittels `./tests.py` ausgeführt werden.
+Tests können mittels `./runTests.py` ausgeführt werden.
 
-Coverage Tests können mittels `rm .coverage; python -m coverage run tests.py; python -m coverage html` ausgeführt werden. Dies erstellt einen Ordner `htmlcov` in dem HTML Output liegt.
+Coverage Tests können mittels `rm .coverage; python -m coverage run runTests.py; python -m coverage html` ausgeführt werden. Dies erstellt einen Ordner `htmlcov` in dem HTML Output liegt.
 
 ### Zum Mitmachen:
 1. Repo für den eigenen User forken, dafür den "Fork-Button" auf der Website verwenden

config.py.example

@@ -29,7 +29,8 @@ LDAP_GROUPS = ['fachschaft']
 #ICAL_URL = 'https://user:password@mail.fsmpi.rwth-aachen.de/SOGo/....ics'
 ERROR_PAGE = 'static/500.html'
 RWTH_IP_RANGES = ['134.130.0.0/16', '137.226.0.0/16', '134.61.0.0/16', '192.35.229.0/24', '2a00:8a60::/32']
-FSMPI_IP_RANGES = ['137.226.35.192/29', '137.226.75.0/27', '137.226.127.32/27', '137.226.231.192/26', '134.130.102.0/26' ]
+FSMPI_IP_RANGES = ['137.226.35.192/29', '137.226.75.0/27', '137.226.127.32/27', '137.226.231.192/26', '134.130.102.0/26', '127.0.0.1/32']
+INTERNAL_IP_RANGES = ['127.0.0.0/8', '192.168.155.0/24', 'fd78:4d90:6fe4::/48']
 DISABLE_SCHEDULER = False
 #MAIL_SERVER = 'mail.fsmpi.rwth-aachen.de'
 MAIL_FROM = 'Video AG-Website <videoag-it@lists.fsmpi.rwth-aachen.de>'

db_schema.sql

@@ -107,7 +107,9 @@ CREATE TABLE IF NOT EXISTS `lectures_data` (
   `titlefile` varchar(255) NOT NULL DEFAULT '',
   `live` INTEGER NOT NULL DEFAULT 0,
   `norecording` INTEGER NOT NULL DEFAULT 0,
-	`profile` varchar(64)
+	`profile` varchar(64),
+  `stream_settings` text NOT NULL DEFAULT '',
+  `stream_job` INTEGER
 );
 CREATE TABLE IF NOT EXISTS `places` (
   `place` varchar(20) NOT NULL PRIMARY KEY,
@@ -179,6 +181,22 @@ CREATE TABLE IF NOT EXISTS `streams` (
   `poster` text NOT NULL,
   `job_id` INTEGER
 );
+CREATE TABLE IF NOT EXISTS `live_sources` (
+	`id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
+  `key` varchar(32) UNIQUE,
+  `preview_key` varchar(32),
+  `name` text NOT NULL,
+  `description` text NOT NULL DEFAULT '',
+  `options` text NOT NULL DEFAULT '',
+  `server` varchar(32),
+  `server_public` varchar(32),
+  `clientid` INTEGER,
+  `last_active` datetime,
+  `time_created` datetime NOT NULL,
+  `time_updated` datetime NOT NULL,
+  `created_by` INTEGER NOT NULL,
+  `deleted` INTEGER NOT NULL DEFAULT '0'
+);
 CREATE TABLE IF NOT EXISTS `stream_stats` (
 `id` INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
   `handle` varchar(32) NOT NULL,

edit.py

@@ -45,7 +45,9 @@ editable_tables = {
 			'jumplist':	{'type': ''},
 			'deleted':	{'type': 'boolean'},
 			'live':		{'type': 'boolean', 'description': 'Ist ein Livestream geplant? Muss gesetzt sein damit der RTMP Stream zugeordnet wird.'},
-			'norecording': {'type': 'boolean', 'description:': 'Führt dazu, dass der Termin ausgegraut wird.'}},
+			'norecording': {'type': 'boolean', 'description:': 'Führt dazu, dass der Termin ausgegraut wird.'},
+			'stream_settings':	{'type': 'text'}
+			},
 		'creationtime_fields': ['course_id', 'time_created', 'time_updated'] },
 	'videos': {
 		'table': 'videos_data',
@@ -111,7 +113,15 @@ editable_tables = {
 			'notify_new_video': {'type': 'boolean'},
 			'notify_edit': {'type': 'boolean'}
 		},
-		'creationtime_fields': [] }
+		'creationtime_fields': [] },
+	'live_sources': {
+		'table': 'live_sources',
+		'idcolumn': 'id',
+		'editable_fields': {
+			'name': {'type': 'shortstring'},
+			'description': {'type': 'text'},
+		},
+		'creationtime_fields': ['created_by', 'time_created', 'time_updated'] }
 	}
 
 #parses the path to a dict, containing the table, id, field and field type

encoding.py

 from server import *
+from sorter import insert_video
 import os.path
+import json
 
 def set_metadata(dest, course, lecture):
 	chapters = query('SELECT text, time FROM chapters WHERE lecture_id = ? AND visible ORDER BY time', lecture['id'])
@@ -79,7 +81,16 @@ def schedule_transcode(source, fmt_id=None, video=None):
 		data['lecture_id'] = lecture['id']
 	data['format_id'] = fmt['id']
 	data['source_id'] = source['id']
-	schedule_job('transcode', data, queue="background")
+	return schedule_job('transcode', data, queue="background")
+
+@job_handler('transcode')
+def insert_transcoded_video(jobid, jobtype, data, state, status):
+	if 'lecture_id' not in data or 'source_id' not in data or 'format_id' not in data:
+		return
+	if 'video_id' in data:
+		return
+	video_id = insert_video(data['lecture_id'], data['output']['path'], data['format_id'], status['hash'], status['filesize'], status['duration'], data['source_id'])
+	schedule_remux(data['lecture_id'], video_id)
 
 @app.route('/internal/jobs/add/reencode', methods=['GET', 'POST'])
 @mod_required

jobmanagement.py

+from server import modify, query, date_json_handler, sched_func, notify_admins
+from datetime import datetime, timedelta
+import traceback
+import json
+
+job_handlers = {}
+def job_handler(*types, state='finished'):
+	def wrapper(func):
+		for jobtype in types:
+			if jobtype not in job_handlers:
+				job_handlers[jobtype] = {}
+			if state not in job_handlers[jobtype]:
+				job_handlers[jobtype][state] = []
+			job_handlers[jobtype][state].append(func)
+		return func
+	return wrapper
+
+def job_handler_handle(id, state):
+	job = query('SELECT * FROM jobs WHERE id = ?', id, nlfix=False)[0]
+	type = job['type']
+	for func in job_handlers.get(type, {}).get(state, []):
+		try:
+			func(id, job['type'], json.loads(job['data']), state, json.loads(job['status']))
+		except Exception:
+			notify_admins('scheduler_exception', name=func.__name__, traceback=traceback.format_exc())
+			traceback.print_exc()
+
+@sched_func(10)
+def job_catch_broken():
+	# scheduled but never pinged
+	query('BEGIN')
+	query('UPDATE jobs SET state="ready" WHERE state="scheduled" and time_scheduled < ?', datetime.now() - timedelta(seconds=10))
+	try:
+		query('COMMIT')
+	except:
+		pass
+	# no pings since 60s
+	query('BEGIN')
+	query('UPDATE jobs SET state="failed" WHERE state="running" and last_ping < ?', datetime.now() - timedelta(seconds=60))
+	try:
+		query('COMMIT')
+	except:
+		pass
+
+def job_set_state(id, state):
+	query('UPDATE jobs SET state=? WHERE id=?', state, id)
+
+def schedule_job(jobtype, data=None, priority=0, queue="default"):
+	if not data:
+		data = {}
+	return modify('INSERT INTO jobs (type, priority, queue, data, time_created) VALUES (?, ?, ?, ?, ?)',
+			jobtype, priority, queue, json.dumps(data, default=date_json_handler), datetime.now())
+
+def cancel_job(job_id):
+	query('UPDATE jobs SET state = "deleted" WHERE id = ? AND state = "ready"', job_id)
+	query('UPDATE jobs SET canceled = 1 WHERE id = ?', job_id)
+
+def restart_job(job_id, canceled=False):
+	if canceled:
+		query('UPDATE jobs SET state = "ready", canceled = 0 WHERE id = ? AND state = "failed"', job_id)
+	else:
+		query('UPDATE jobs SET state = "ready" WHERE id = ? AND state = "failed" AND NOT canceled', job_id)

jobs.py

 from server import *
-import traceback
 import json
 import random
 from time import sleep
 
-job_handlers = {}
-def job_handler(*types, state='finished'):
-	def wrapper(func):
-		for jobtype in types:
-			if jobtype not in job_handlers:
-				job_handlers[jobtype] = {}
-			if state not in job_handlers[jobtype]:
-				job_handlers[jobtype][state] = []
-			job_handlers[jobtype][state].append(func)
-		return func
-	return wrapper
-
-def schedule_job(jobtype, data=None, priority=0, queue="default"):
-	if not data:
-		data = {}
-	return modify('INSERT INTO jobs (type, priority, queue, data, time_created) VALUES (?, ?, ?, ?, ?)',
-			jobtype, priority, queue, json.dumps(data, default=date_json_handler), datetime.now())
-
-def cancel_job(job_id):
-	modify('UPDATE jobs SET state = "deleted" WHERE id = ? AND state = "ready"', job_id)
-	modify('UPDATE jobs SET canceled = 1 WHERE id = ?', job_id)
-
-def restart_job(job_id, canceled=False):
-	if canceled:
-		modify('UPDATE jobs SET state = "ready", canceled = 0 WHERE id = ? AND state = "failed"', job_id)
-	else:
-		modify('UPDATE jobs SET state = "ready" WHERE id = ? AND state = "failed" AND NOT canceled', job_id)
-
 @app.route('/internal/jobs/overview')
 @register_navbar('Jobs', iconlib='fa', icon='suitcase', group='weitere')
 @mod_required
@@ -61,7 +32,13 @@ def jobs_overview():
 
 	pagecount = math.ceil(query('SELECT count(id) as count FROM jobs WHERE (type like ?) AND (worker like ? OR (worker IS NULL AND ? = "%")) AND (state like ?)', filter['type'], filter['worker'], filter['worker'], filter['state'])[0]['count']/pagesize)
 	jobs = query('SELECT * FROM jobs WHERE (type like ?) AND (worker like ? OR (worker IS NULL AND ? = "%")) AND (state like ?) ORDER BY `time_created` DESC LIMIT ? OFFSET ?', filter['type'], filter['worker'], filter['worker'], filter['state'], pagesize, page*pagesize)
-	return render_template('jobs_overview.html',worker=worker,jobs=jobs, filter_values=filter_values, filter=filter, page=page, pagesize=pagesize, pagecount=pagecount)
+	active_streams = query('SELECT lectures.*, "course" AS sep, courses.*, "job" AS sep, jobs.* FROM lectures JOIN courses ON (courses.id = lectures.course_id) JOIN jobs ON (jobs.id = lectures.stream_job) WHERE lectures.stream_job')
+	for stream in active_streams:
+		try:
+			stream['destbase'] = json.loads((stream['job']['data'] or '{}')).get('destbase')
+		except:
+			pass
+	return render_template('jobs_overview.html',worker=worker,jobs=jobs, filter_values=filter_values, filter=filter, page=page, pagesize=pagesize, pagecount=pagecount, active_streams=active_streams)
 
 @app.route('/internal/jobs/action/<action>', methods=['GET', 'POST'])
 @app.route('/internal/jobs/action/<action>/<jobid>', methods=['GET', 'POST'])
@@ -96,23 +73,6 @@ def jobs_api_token_required(func):
 			return func(*args, **kwargs)
 	return decorator
 
-@sched_func(10)
-def jobs_catch_broken():
-	# scheduled but never pinged
-	query('BEGIN')
-	query('UPDATE jobs SET state="ready" WHERE state="scheduled" and time_scheduled < ?', datetime.now() - timedelta(seconds=10))
-	try:
-		query('COMMIT')
-	except:
-		pass
-	# no pings since 60s
-	query('BEGIN')
-	query('UPDATE jobs SET state="failed" WHERE state="running" and last_ping < ?', datetime.now() - timedelta(seconds=60))
-	try:
-		query('COMMIT')
-	except:
-		pass
-
 @app.route('/internal/jobs/api/job/<int:id>/ping', methods=['GET', 'POST'])
 @jobs_api_token_required
 def jobs_ping(id):
@@ -123,12 +83,8 @@ def jobs_ping(id):
 		query('UPDATE jobs SET time_finished = ?, status = ?, state = "finished" where id = ?', datetime.now(), status, id)
 	else:
 		query('UPDATE jobs SET worker = ?, last_ping = ?, status = ?, state = ? where id = ?', hostname, datetime.now(), status, state, id)
+	job_handler_handle(id, state)
 	job = query('SELECT * FROM jobs WHERE id = ?', id, nlfix=False)[0]
-	for func in job_handlers.get(job['type'], {}).get(state, []):
-		try:
-			func(id, job['type'], json.loads(job['data']), state, json.loads(job['status']))
-		except Exception:
-			traceback.print_exc()
 	if job['canceled']:
 		return 'Job canceled', 205
 	else:
@@ -163,3 +119,12 @@ def jobs_schedule(hostname):
 				return 'no jobs', 503
 
 	return Response(json.dumps(job, default=date_json_handler), mimetype='application/json')
+
+@app.route('/internal/jobs/add/forward', methods=['GET', 'POST'])
+@mod_required
+@csrf_protect
+def add_forward_job():
+	schedule_job('live_forward', {'src': request.values['src'],
+			'dest': request.values['dest'], 'format': 'flv'}, priority=9)
+	return redirect(request.values.get('ref', url_for('jobs_overview')))
+

livestreams.py

 from server import *
+import requests
+from xml.etree import ElementTree
+import random
+import string
 
 @sched_func(120)
 def livestream_thumbnail():
 	livestreams = query('SELECT streams.lecture_id, streams.handle AS livehandle FROM streams WHERE streams.active')
-	for v in genlive(livestreams):
-		schedule_job('thumbnail', {'lectureid': str(v['lecture_id']), 'path': v['path']})
+	lectures = query('SELECT * FROM lectures WHERE stream_job IS NOT NULL')
+	for v in genlive(livestreams)+genlive_new(lectures):
+		schedule_job('thumbnail', {'src': v['path'], 'filename': 'l_%i.jpg'%v['lecture_id']})
 
 @app.route('/internal/streaming/legacy_auth', methods=['GET', 'POST'])
 @app.route('/internal/streaming/legacy_auth/<server>', methods=['GET', 'POST'])
-def streamauth(server=None):
+def streamauth_legacy(server=None):
 	internal = False
 	if 'X-Real-IP' in request.headers:
 		for net in config.get('FSMPI_IP_RANGES', []):
@@ -57,3 +62,192 @@ def streamauth(server=None):
 def restart_failed_live_transcode(id, type, data, state, status):
 	restart_job(id)
 
+@app.route('/internal/streaming')
+@register_navbar('Streaming', icon='broadcast-tower', iconlib='fa')
+@mod_required
+def streaming():
+	sources = query('SELECT * FROM live_sources WHERE NOT deleted')
+	for source in sources:
+		if not source['clientid']:
+			continue
+		r = requests.get('http://%s:8080/stats'%source['server'])
+		if r.status_code != 200:
+			continue
+		source['stat'] = {}
+		tree = ElementTree.fromstring(r.text)
+		if not tree:
+			continue
+		s = tree.find("./server/application/[name='src']/live/stream/[name='%i']"%source['id'])
+		if not s:
+			continue
+		for e in s.find("client/[publishing='']").getchildren():
+			source['stat'][e.tag] = e.text
+		source['video'] = {}
+		for e in s.find('meta/video').getchildren():
+			source['video'][e.tag] = e.text
+		source['audio'] = {}
+		for e in s.find('meta/audio').getchildren():
+			source['audio'][e.tag] = e.text
+	return render_template("streaming.html", sources=sources)
+
+def gentoken():
+	return ''.join(random.SystemRandom().choice(string.ascii_letters + string.digits) for _ in range(16))
+
+@app.route('/internal/streaming/rekey/<int:id>')
+@mod_required
+def streamrekey(id):
+	modify('UPDATE live_sources SET `key` = ? WHERE id = ? AND NOT deleted', gentoken(), id)
+	source = query('SELECT * FROM live_sources WHERE NOT deleted AND id = ?', id)[0]
+	flash('Der Streamkey von <strong>'+source['name']+'</strong> wurde neu generiert: <span><input readonly type="text" style="width: 15em" value="'+source['key']+'"></span>')
+	return redirect(url_for('streaming'))
+
+@app.route('/internal/streaming/drop/<int:id>')
+@mod_required
+def streamdrop(id):
+	source = (query('SELECT * FROM live_sources WHERE NOT deleted AND id = ?', id) or [None])[0]
+	if not source:
+		if 'ref' in request.values:
+			flash('Streamquelle nicht gefunden')
+			return redirect(request.values['ref'])
+		else:
+			return 'Not found', 404
+	requests.get('http://%s:8080/control/drop/publisher?clientid=%i'%(source['server'], source['clientid']))
+	if 'ref' in request.values:
+		return redirect(request.values['ref'])
+	return 'Ok', 200
+
+@sched_func(120)
+def live_source_thumbnail():
+	sources = query('SELECT * FROM live_sources WHERE clientid IS NOT NULL')
+	for source in sources:
+		schedule_job('thumbnail', {'srcurl': 'rtmp://%s/src/%i'%(source['server'], source['id']), 'filename': 's_%i.jpg'%source['id']})
+
+@app.route('/internal/streaming/auth/<server>', methods=['GET', 'POST'])
+def streamauth(server):
+	internal = False
+	for net in config.get('FSMPI_IP_RANGES', []):
+		if ip_address(request.headers['X-Real-IP']) in ip_network(net):
+			internal = True
+	if not internal:
+		return 'Forbidden', 403
+	if request.values['call'] == 'publish':
+		sources = query('SELECT * FROM live_sources WHERE NOT deleted AND `key` = ?', request.values['name'])
+		if not sources:
+			return 'Not found', 404
+		modify('UPDATE live_sources SET server = ?, server_public = ?, clientid = ?, last_active = ?, preview_key = ? WHERE id = ?', server, request.args.get('public_ip', server), request.values['clientid'], datetime.now(), gentoken(), sources[0]['id'])
+		live_source_thumbnail()
+		ret = Response('Redirect', 301, {'Location': '%i'%sources[0]['id']})
+		ret.autocorrect_location_header = False
+		return ret
+	if request.values['call'] == 'play':
+		source = (query('SELECT * FROM live_sources WHERE NOT deleted AND id = ?', request.values['name']) or [None])[0]
+		if not source:
+			return 'Not found', 404
+		for net in config.get('INTERNAL_IP_RANGES', []):
+			if ip_address(request.values['addr']) in ip_network(net):
+				return 'Ok', 200
+		if source['preview_key'] == request.values.get('preview_key'):
+			return 'Ok', 200
+		return 'Forbidden', 403
+	elif request.values['call'] == 'publish_done':
+		source = (query('SELECT * FROM live_sources WHERE server = ? AND clientid = ?', server, request.values['clientid']) or [None])[0]
+		modify('UPDATE live_sources SET server = NULL, clientid = NULL, preview_key = NULL, last_active = ? WHERE server = ? AND clientid = ?', datetime.now(), server, request.values['clientid'])
+		if not source:
+			return 'Ok', 200
+		for lecture in query('SELECT * FROM lectures WHERE stream_job IS NOT NULL'):
+			settings = json.loads(lecture['stream_settings'])
+			if str(source['id']) in [str(settings.get('source1')), str(settings.get('source2'))]:
+				cancel_job(lecture['stream_job'])
+		return 'Ok', 200
+	return 'Bad request', 400
+
+def schedule_livestream(lecture_id):
+	def build_filter(l):
+		return ','.join(l) if l else None
+	server = 'rwth.video'
+	lecture = query('SELECT * FROM lectures WHERE id = ?', lecture_id)[0]
+	settings = json.loads(lecture['stream_settings'])
+	data = {'src1': {'afilter': [], 'vfilter': []}, 'src2': {'afilter': [], 'vfilter': []}, 'afilter': [], 'videoag_logo': int(bool(settings.get('video_showlogo'))), 'lecture_id': lecture['id']}
+	src1 = (query('SELECT * FROM live_sources WHERE NOT deleted AND id = ?', settings.get('source1')) or [{}])[0]
+	src2 = (query('SELECT * FROM live_sources WHERE NOT deleted AND id = ?', settings.get('source2')) or [{}])[0]
+	for idx, obj in zip([1,2], [src1, src2]):
+		if obj:
+			server = obj['server']
+			data['src%i'%idx]['url'] = 'rtmp://%s/src/%i'%(obj['server'], obj['id'])
+			if not obj['clientid']:
+				flash('Quelle „%s“ ist nicht aktiv!'%obj['name'])
+				return None
+		if settings.get('source%i_deinterlace'%idx):
+			data['src%i'%idx]['vfilter'].append('yadif')
+		mode = settings.get('source%i_audiomode'%idx)
+		leftvol = float(settings.get('source%i_leftvolume'%idx, 100))/100.0
+		rightvol = float(settings.get('source%i_rightvolume'%idx, 100))/100.0
+		if mode == 'mono':
+			data['src%i'%idx]['afilter'].append('pan=mono|c0=%f*c0+%f*c1'%(0.5*leftvol, 0.5*rightvol))
+		elif mode == 'stereo':
+			data['src%i'%idx]['afilter'].append('pan=stereo|c0=%f*c0|c1=%f*c1'%(leftvol, rightvol))
+		elif mode == 'unchanged':
+			pass
+		elif mode == 'off':
+			data['src%i'%idx]['afilter'].append('pan=mono|c0=0*c0')
+		else:
+			raise(Exception())
+	mode = settings.get('videomode')
+	if mode == '1':
+		data['vmix'] = 'streamselect=map=0'
+	elif mode == '2':
+		data['vmix'] = 'streamselect=map=1'
+	elif mode == 'lecture4:3':
+		data['src1']['vfilter'].append('scale=1440:1080')
+		data['src2']['vfilter'].append('scale=1440:810,pad=1440:1080:0:135,crop=480:1080')
+		data['vmix'] = 'hstack'
+	elif mode == 'lecture16:9':
+		data['src1']['vfilter'].append('scale=1440:810,pad=1440:1080:0:135')
+		data['src2']['vfilter'].append('scale=1440:810,pad=1440:1080:0:135,crop=480:1080')
+		data['vmix'] = 'hstack'
+	elif mode == 'sidebyside':
+		data['src1']['vfilter'].append('scale=960:540')
+		data['src2']['vfilter'].append('scale=960:540')
+		data['vmix'] = 'hstack,pad=1920:1080:0:270'
+	if settings.get('audio_normalize'):
+		data['afilter'].append('loudnorm')
+	data['afilter'] = build_filter(data['afilter'])
+	data['src1']['afilter'] = build_filter(data['src1']['afilter'])
+	data['src1']['vfilter'] = build_filter(data['src1']['vfilter'])
+	data['src2']['afilter'] = build_filter(data['src2']['afilter'])
+	data['src2']['vfilter'] = build_filter(data['src2']['vfilter'])
+	data['destbase'] = 'rtmp://%s/hls/%i'%(server, lecture['id'])
+	if lecture['stream_job']:
+		flash('Stream läuft bereits!')
+		return None
+	job_id = schedule_job('complex_live_transcode', data, priority=10)
+	modify('UPDATE lectures_data SET stream_job = ? WHERE id = ? AND stream_job IS NULL', job_id, lecture_id)
+	if query('SELECT stream_job FROM lectures WHERE id = ?', lecture_id)[0]['stream_job'] != job_id:
+		flash('Stream läuft bereits!')
+		cancel_job(job_id)
+		return None
+	return job_id
+
+@job_handler('complex_live_transcode', state='failed')
+def restart_failed_complex_live_transcode(id, type, data, state, status):
+	restart_job(id)
+
+@job_handler('complex_live_transcode', state='failed')
+@job_handler('complex_live_transcode', state='finished')
+def cleanup_after_complex_live_transcode_ended(id, type, data, state, status):
+	job = query('SELECT * FROM jobs WHERE id = ?', id, nlfix=False)[0]
+	if state == 'finished' or (state == 'failed' and job['canceled']):
+		modify('UPDATE lectures_data SET stream_job = NULL WHERE stream_job = ?', id)
+
+@app.route('/internal/streaming/control', methods=['POST'])
+@mod_required
+def control_stream():
+	action = request.values['action']
+	lecture_id = int(request.values['lecture_id'])
+	course = (query('SELECT courses.* FROM courses JOIN lectures ON (courses.id = lectures.course_id) WHERE lectures.id = ?', lecture_id) or [None])[0]
+	if action == 'start':
+		schedule_livestream(lecture_id)
+	elif action == 'stop':
+		lecture = query('SELECT * FROM lectures WHERE id = ?', lecture_id)[0]
+		cancel_job(lecture['stream_job'])
+	return redirect(url_for('course', handle=course['handle']))

runTests.py

+#!/usr/bin/env python3
+import unittest
+import os
+import server
+
+def setUp():
+	server.app.testing = True
+
+def tearDown():
+	os.unlink(server.app.config['SQLITE_DB'])
+
+if __name__ == '__main__':
+	setUp()
+	try:
+		suite = unittest.defaultTestLoader.discover('./tests/', pattern="*")
+		unittest.TextTestRunner(verbosity=2, failfast=True).run(suite)
+	finally:
+		tearDown()
+

server.py

@@ -26,7 +26,7 @@ if sys.argv[0].endswith('run.py'):
 	config['SQLITE_INIT_DATA'] = True
 	config['DEBUG'] = True
 config.from_pyfile('config.py', silent=True)
-if sys.argv[0].endswith('tests.py'):
+if sys.argv[0].endswith('runTests.py'):
 	print('running in test mode')
 	import tempfile
 	# ensure we always use a clean sqlite db for tests
@@ -81,7 +81,6 @@ from db import query, modify, show, searchquery
 from template_helper import *
 from mail import notify_mods, notify_admins
 from ldap import ldapauth
-from legacy import legacy_index
 from scheduler import sched_func
 
 def render_endpoint(endpoint, flashtext=None, **kargs):
@@ -135,6 +134,18 @@ def genlive(streams):
 		stream['file_size'] = 0
 	return streams
 
+def genlive_new(lectures):
+	hls_format = (query('SELECT * FROM formats WHERE keywords = "hls"') or [{}])[0]
+	res = []
+	for lecture in lectures:
+		if not lecture['stream_job']:
+			continue
+		res.append({'livehandle': '%i'%lecture['id'], 'visible': True,
+				'downloadable': False, 'path': 'pub/hls/%i.m3u8'%lecture['id'],
+				'file_size': 0, 'formats': hls_format, 'lecture_id': lecture['id']})
+	return res
+
+from legacy import legacy_index
 
 @app.route('/')
 @register_navbar('Home', icon='home')
@@ -170,6 +181,13 @@ def index():
 		JOIN courses ON courses.id = lectures.course_id
 		WHERE streams.active AND (? OR (streams.visible AND courses.visible AND courses.listed AND lectures.visible))
 		''', ismod())
+	livestreams_new = query('''SELECT lectures.*, "course" AS sep, courses.*
+		FROM lectures
+		JOIN courses ON courses.id = lectures.course_id
+		WHERE lectures.stream_job IS NOT NULL AND (? OR (courses.visible AND courses.listed AND lectures.visible))
+		''', ismod())
+	for stream in livestreams_new:
+		stream['livehandle'] = '%i'%stream['id']
 	featured = query('SELECT * FROM featured WHERE (? OR visible) ORDER BY `order`', ismod())
 	featured = list(filter(lambda x: not x['deleted'], featured))
 	for item in featured:
@@ -192,7 +210,7 @@ def index():
 					WHERE videos.lecture_id = ? AND videos.visible
 					ORDER BY formats.prio DESC
 					''', item['param'])+genlive(streams)
-	return render_template('index.html', latestvideos=livestreams+latestvideos, upcomming=upcomming, featured=featured)
+	return render_template('index.html', latestvideos=livestreams_new+livestreams+latestvideos, upcomming=upcomming, featured=featured)
 
 @app.route('/courses')
 @register_navbar('Videos', icon='film')
@@ -247,16 +265,18 @@ def course(id=None, handle=None):
 			JOIN formats ON formats.keywords = "hls"
 			WHERE streams.active AND (? OR streams.visible) AND lectures.course_id = ?
 			''', ismod(), course['id'])
+	videos += genlive(livestreams)
+	videos += genlive_new(lectures)
 	chapters = []
 	if course['coursechapters']:
 		chapters = query('SELECT chapters.* FROM chapters JOIN lectures ON lectures.id = chapters.lecture_id WHERE lectures.course_id = ? AND NOT chapters.deleted AND chapters.visible ORDER BY time ASC', course['id'])
-	videos += genlive(livestreams)
 	responsible = query('''SELECT users.*, responsible.course_id AS responsible
 			FROM users
 			LEFT JOIN responsible ON (responsible.user_id = users.id AND responsible.course_id = ?)
 			WHERE users.fsacc != "" AND users.level > 0