Add per-playlist max connections and viewing quota alerts

Each playlist gets a max_connections setting (default 1), editable in the
playlist modal without re-importing. Active viewings are tracked in the
shared db_iptv.viewings collection (also used by serverIPTV_2): devices
are counted from 'on'/'off' commands and their 10 s polling. When
starting a channel pushes a playlist above its limit, the web UI shows a
quota alert. The playlist list shows active/max connections.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

server.py

@@ -27,7 +27,7 @@ import io, re
 import xml.etree.ElementTree as ET
 from bs4 import BeautifulSoup
 import pathlib
-from datetime import datetime
+from datetime import datetime, timedelta
 from bson import ObjectId
 import urllib.request
 import ipaddress
@@ -67,6 +67,13 @@ playlists = db_iptv['playlists']
 logos = db_iptv['logos']
 recents =  db_iptv['recents']
 settings_collection = db_iptv['settings']
+# Visionnages en cours, partagés avec serverIPTV_2 (même collection, même format) :
+# _id = "<source>:<clé>", un document par connexion ouverte chez le fournisseur.
+viewings = db_iptv['viewings']
+viewings.create_index("expires_at", expireAfterSeconds=0)
+viewings.create_index("id_playlist")
+VIEWING_ACTIVE_SECONDS = 35   # un appareil signale sa lecture toutes les 10 s
+DEFAULT_MAX_CONNECTIONS = 1
 
 devices_db = db_iptv['devices']
 MAX_STREAMING_ERRORS_PER_DEVICE = 5
@@ -192,6 +199,62 @@ def save_playlist_document(playlist_doc):
     )
 
 
+def playlist_max_connections(playlist_doc):
+    try:
+        return max(1, int((playlist_doc or {}).get('max_connections') or DEFAULT_MAX_CONNECTIONS))
+    except (TypeError, ValueError):
+        return DEFAULT_MAX_CONNECTIONS
+
+
+def parse_max_connections(value):
+    """Valeur saisie dans la configuration de playlist (None si absente)."""
+    if value in (None, ''):
+        return None
+    number = int(value)
+    if not 1 <= number <= 100:
+        raise ValueError('max_connections must be between 1 and 100')
+    return number
+
+
+def active_viewings_count(id_playlist, exclude_id=None):
+    since = datetime.utcnow() - timedelta(seconds=VIEWING_ACTIVE_SECONDS)
+    query = {'id_playlist': id_playlist, 'last_seen': {'$gte': since}}
+    if exclude_id:
+        query['_id'] = {'$ne': exclude_id}
+    return viewings.count_documents(query)
+
+
+def touch_viewing(viewing_id, id_playlist, id_channel=None, name_channel=None, extra=None):
+    now = datetime.utcnow()
+    fields = {'id_playlist': id_playlist, 'last_seen': now,
+              'expires_at': now + timedelta(seconds=VIEWING_ACTIVE_SECONDS * 4)}
+    if id_channel is not None:
+        fields['id_channel'] = id_channel
+    if name_channel is not None:
+        fields['name_channel'] = name_channel
+    if extra:
+        fields.update(extra)
+    source, _, key = viewing_id.partition(':')
+    viewings.update_one({'_id': viewing_id},
+                        {'$set': fields, '$setOnInsert': {'source': source, 'key': key, 'started_at': now}},
+                        upsert=True)
+
+
+def start_viewing(viewing_id, id_playlist, id_channel=None, name_channel=None, extra=None):
+    """Enregistre un nouveau visionnage et renvoie l'état du quota de la playlist."""
+    others = active_viewings_count(id_playlist, exclude_id=viewing_id)
+    touch_viewing(viewing_id, id_playlist, id_channel, name_channel, extra)
+    playlist_doc = playlists.find_one({'id_playlist': id_playlist}) or {}
+    maximum = playlist_max_connections(playlist_doc)
+    active = others + 1
+    return {'id_playlist': id_playlist, 'title_playlist': playlist_doc.get('title_playlist', ''),
+            'active': active, 'max': maximum, 'exceeded': active > maximum}
+
+
+def device_viewing_id(id_device):
+    return 'serverIPTV:' + str(id_device)
+
+
 def delete_playlist_resources(id_playlist):
     db_iptv.drop_collection("channels_" + id_playlist)
 
@@ -464,6 +527,7 @@ async def analyse_m3u_file(request):
     
 
     try:
+        max_connections = parse_max_connections(data.get('max_connections'))
         id_playlist = validate_id_playlist(id_playlist)
         if not UPLOADED_PLAYLIST_RE.match(filename or ''):
             raise ValueError('invalid filename')
@@ -489,6 +553,7 @@ async def analyse_m3u_file(request):
             'xtream_server': '',
             'xtream_username': '',
             'xtream_password': '',
+            **({'max_connections': max_connections} if max_connections else {}),
         }
     )
     save_playlist_document(new_playlist)
@@ -578,6 +643,10 @@ async def import_xtream_codes(request):
     xtream_server = normalize_xtream_server(data.get('xtream_server', ''))
     xtream_username = data.get('xtream_username', '').strip()
     xtream_password = data.get('xtream_password', '').strip()
+    try:
+        max_connections = parse_max_connections(data.get('max_connections'))
+    except ValueError as exc:
+        return web.json_response({'error': str(exc)}, status=400)
 
     if not xtream_server or not xtream_username or not xtream_password:
         return web.json_response({'error': 'Xtream server, username and password are required.'}, status=400)
@@ -620,6 +689,7 @@ async def import_xtream_codes(request):
             'xtream_server': xtream_server,
             'xtream_username': xtream_username,
             'xtream_password': xtream_password,
+            **({'max_connections': max_connections} if max_connections else {}),
         }
     )
     save_playlist_document(playlist_doc)
@@ -897,6 +967,21 @@ async def update_playlist_comment(request):
     return web.json_response({'status': 'ok', 'id_playlist': id_playlist, 'comment': comment})
 
 
+async def update_playlist_max_connections(request):
+    data = await request.post()
+    id_playlist = (data.get('id_playlist') or '').strip()
+    try:
+        max_connections = parse_max_connections(data.get('max_connections'))
+    except ValueError as exc:
+        return web.json_response({'error': str(exc)}, status=400)
+    if not id_playlist or not max_connections:
+        return web.json_response({'error': 'id_playlist and max_connections are required'}, status=400)
+    result = playlists.update_one({'id_playlist': id_playlist}, {'$set': {'max_connections': max_connections}})
+    if result.matched_count == 0:
+        return web.json_response({'error': 'playlist not found'}, status=404)
+    return web.json_response({'status': 'ok', 'id_playlist': id_playlist, 'max_connections': max_connections})
+
+
 # Get List Playlists
 async def get_list_playlists(request):
     active_count = playlists.count_documents({})
@@ -911,7 +996,11 @@ async def get_list_playlists(request):
     if active_count == 0:
         return web.json_response({'status':'no data'})
     else:
-        return web.json_response(dumps(playlists.find()))
+        items = list(playlists.find())
+        for p in items:
+            p['max_connections'] = playlist_max_connections(p)
+            p['active_connections'] = active_viewings_count(p.get('id_playlist'))
+        return web.json_response(dumps(items))
 
 
 async def get_source(request):
@@ -1337,6 +1426,7 @@ app.router.add_post('/upload_snapshot', upload_snapshot)
 app.router.add_post('/analyse_m3u_file', analyse_m3u_file)
 app.router.add_post('/import_xtream_codes', import_xtream_codes)
 app.router.add_post('/update_playlist_comment', update_playlist_comment)
+app.router.add_post('/update_playlist_max_connections', update_playlist_max_connections)
 app.router.add_get('/delete_playlist', delete_playlist)
 app.router.add_get('/cleanup_xtream_duplicates', cleanup_xtream_duplicates)
 app.router.add_post('/list_channel', list_channel)
@@ -1411,9 +1501,31 @@ async def communication(fromSid, data):
     print("search request from ", fromSid, flush=True)
     toSidDevice = data['toSidDevice']
     print("send signal to ", toSidDevice, flush=True)
+    type_command = data.get('typeCommand')
+    if type_command in ('on', 'off'):
+        try:
+            await track_device_command(fromSid, toSidDevice, type_command, data.get('payload'))
+        except Exception as exc:  # le suivi des quotas ne doit jamais bloquer la commande
+            logger.warning("Viewing tracking failed: %s", exc)
     await sio.emit('communication', data, room=toSidDevice ,callback=messageReceived)
 
 
+async def track_device_command(fromSid, toSidDevice, type_command, payload):
+    device = devices_db.find_one({'sid': toSidDevice}) or {}
+    viewing_id = device_viewing_id(device.get('id_device') or toSidDevice)
+    if type_command == 'off':
+        viewings.delete_one({'_id': viewing_id})
+        return
+    if not isinstance(payload, dict) or not payload.get('id_playlist'):
+        return
+    quota = start_viewing(viewing_id, payload['id_playlist'], payload.get('id_channel'),
+                          payload.get('name'), {'device_name': device.get('name')})
+    logger.info("Viewing %s on playlist %s: %s/%s", viewing_id, quota['title_playlist'],
+                quota['active'], quota['max'])
+    if quota['exceeded']:
+        await sio.emit('communication', {'typeCommand': 'quota_alert', 'payload': quota}, room=fromSid)
+
+
 @sio.event
 async def streaming_error(sid, data):
     IdDevice = data.get('IdDevice')
@@ -1432,6 +1544,9 @@ async def streaming_error(sid, data):
 @sio.event
 def disconnect(sid):
     sio_session_ips.pop(sid, None)
+    device = devices_db.find_one({'sid': sid}, {'id_device': 1})
+    if device:
+        viewings.delete_one({'_id': device_viewing_id(device['id_device'])})
     print('disconnect ', sid, flush=True)
 
 
@@ -1466,6 +1581,13 @@ async def polling(sid, data):
         new_device = { 'id_device': IdDevice, 'name' : device_name, 'last_datetime' : device_last_datetime, 'sid':sid ,'mediaRunning' : mediaRunning, 'vpn_status':vpn_status, 'data_running':data_running, 'auto_snapshot_enabled': auto_snapshot_enabled, 'snapshot_interval_seconds': snapshot_interval_seconds, 'ip_address': ip_address, 'ip_country': ip_country}
         devices_db.insert_one(new_device)
 
+    if data_running and data_running.get('id_playlist'):
+        touch_viewing(device_viewing_id(IdDevice), data_running['id_playlist'],
+                      data_running.get('id_channel'), data_running.get('name_channel'),
+                      {'device_name': device_name})
+    else:
+        viewings.delete_one({'_id': device_viewing_id(IdDevice)})
+
     if data_running:
         new_record = {
                 "id_device": IdDevice,