XoséandClaude Opus 5.5
e16dd66f63
Kodi keeps its own copy of the programme guide and fetches it again only when IPTV Simple reloads its files, on its refresh timer; restarting IPTV Simple is not enough. After the first refresh following a Kodi start or add-on update, if the guide changed and nothing is playing, restart IPTV Simple with a 1-minute timer, wait for the reload, then restart it with the hourly timer again. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
298 lines
13 KiB
Python
298 lines
13 KiB
Python
# encoding: utf-8
|
|
#
|
|
# SPDX-License-Identifier: LGPL-2.1-or-later
|
|
#
|
|
# KodiMov+ additions to the Movistar+ add-on:
|
|
# - takes over the login of plugin.video.movistarplus on first run
|
|
# - logs in from the Movistar home network without a password
|
|
# - hides channels this add-on can't play
|
|
# - extra free channels for the TV guide
|
|
|
|
from __future__ import unicode_literals, absolute_import, division
|
|
|
|
import io
|
|
import json
|
|
import os
|
|
import shutil
|
|
import time
|
|
|
|
from .log import LOG
|
|
|
|
ORIGINAL_ADDON = 'plugin.video.movistarplus'
|
|
CLASSIC_ISSUER = 'http://www.movistarplus.es'
|
|
|
|
# Movistar marks the football and premium sports channels with QMWatermark.
|
|
# The LALIGA and Liga de Campeones ones refuse their manifest to this add-on;
|
|
# these play fine (tested 2026-11-07 on a Raspberry Pi 5, LibreELEC 12.2).
|
|
WATERMARKED_BUT_PLAYABLE = {
|
|
'CPDEP', 'ARTHUR', 'USOP2', 'USOP3', 'USOP11', 'MULTI8', # M+ Deportes 1-6
|
|
'M1SD', 'M2SD', 'DAZN3', 'DAZN4', # DAZN 1-4
|
|
'DAZBA', 'DAZB2', 'DAZB3', # DAZN Baloncesto 1-3
|
|
}
|
|
|
|
# Free channels that Movistar+ doesn't carry. 'chno' must not clash with a
|
|
# Movistar dial number.
|
|
EXTRA_CHANNELS = [
|
|
{'id': 'TVG2', 'name': 'TVG2', 'chno': 10, 'group': 'Galicia', 'epg_id': 'TVG2',
|
|
'logo': 'https://i.imgur.com/yFy6gxg.png',
|
|
'url': 'https://crtvg-tvg2.flumotion.cloud/playlist.m3u8'},
|
|
{'id': 'TeleVigo', 'name': 'TeleVigo', 'chno': 15, 'group': 'Galicia', 'epg_id': 'TELEVIGO',
|
|
'logo': 'https://i.imgur.com/4iVmlFa.png',
|
|
'url': 'https://stream-us-east-1.getpublica.com/playlist.m3u8?network_id=12687'},
|
|
]
|
|
|
|
# Programme listings for the extra channels (Movistar's guide covers the rest).
|
|
EXTRA_EPG_URL = 'https://raw.githubusercontent.com/davidmuma/EPG_dobleM/master/guiafanart_color.xml.gz'
|
|
|
|
AUTOLOGIN_URL = 'https://soter-pf.svp.video.telefonicaservices.com/service/login/autologin?deviceId={device}&deviceType=webplayer&_={ts}'
|
|
SOTER_INITDATA_URL = 'https://soter-pf.sve.video.telefonicaservices.com/service/login/initData?legacy=true'
|
|
SOTER_ACCOUNTINFO_URL = 'https://soter-pf.sve.video.telefonicaservices.com/service/login/webplayer/accountInfo?legacy=true'
|
|
SOTER_DEVICES_URL = 'https://soter-pf.sve.video.telefonicaservices.com/service/login/devices?deviceType=webplayer&accountId={account}&legacy=true'
|
|
|
|
|
|
def is_playable(channel):
|
|
return not channel.get('watermark') or channel['id'] in WATERMARKED_BUT_PLAYABLE
|
|
|
|
|
|
def migrate_from_original(config_directory):
|
|
"""Copies the login of plugin.video.movistarplus (tokens, registered device,
|
|
credentials) the first time KodiMov+ runs, so no new device is added to the
|
|
Movistar account and no password is needed."""
|
|
config_directory = config_directory.rstrip('/\\')
|
|
if os.path.exists(os.path.join(config_directory, 'account.txt')): return False
|
|
if any(d.startswith('account_') for d in os.listdir(config_directory)): return False
|
|
source = os.path.join(os.path.dirname(config_directory), ORIGINAL_ADDON)
|
|
if not os.path.isdir(source): return False
|
|
copied = False
|
|
for d in os.listdir(source):
|
|
src = os.path.join(source, d)
|
|
if d.startswith('account_') and os.path.isdir(src):
|
|
shutil.copytree(src, os.path.join(config_directory, d), ignore=shutil.ignore_patterns('cache'))
|
|
copied = True
|
|
if copied and os.path.exists(os.path.join(source, 'account.txt')):
|
|
shutil.copy(os.path.join(source, 'account.txt'), config_directory)
|
|
LOG('migrate_from_original: {}'.format('copied login from ' + ORIGINAL_ADDON if copied else 'nothing to copy'))
|
|
return copied
|
|
|
|
|
|
def _read_json(cache, filename):
|
|
content = cache.load_file(filename)
|
|
try:
|
|
return json.loads(content) if content else None
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def find_account_number(m, setting=''):
|
|
data = _read_json(m.cache, 'account.json')
|
|
try:
|
|
return data['ofertas'][0]['accountNumber']
|
|
except (TypeError, KeyError, IndexError):
|
|
pass
|
|
data = _read_json(m.cache, 'tokens.json')
|
|
if isinstance(data, dict) and data.get('accountNumber'):
|
|
return data['accountNumber']
|
|
return (setting or '').strip()
|
|
|
|
|
|
def _autologin(m, device_id):
|
|
"""5-minute token for the new servers, bound to device_id. Only works from
|
|
the Movistar home network. Returns the token or raises."""
|
|
try:
|
|
from urllib.parse import urlsplit
|
|
except ImportError:
|
|
from urlparse import urlsplit
|
|
headers = m.net.headers.copy()
|
|
headers['Content-Type'] = 'application/json'
|
|
url = AUTOLOGIN_URL.format(device=device_id, ts=int(time.time() * 1000))
|
|
r = m.net.session.get(url, headers=headers, allow_redirects=False, timeout=(4, 10))
|
|
location = r.headers.get('Location', '')
|
|
if r.status_code in (301, 302, 303, 307, 308) and location:
|
|
if not urlsplit(location).netloc.endswith('.telefonicaservices.com'):
|
|
raise ValueError('unexpected redirect')
|
|
h2 = headers.copy()
|
|
h2['Origin'] = 'null'
|
|
r = m.net.session.get(location, headers=h2, allow_redirects=False, timeout=(4, 10))
|
|
token = r.json().get('access_token')
|
|
if not token: raise ValueError('autologin refused')
|
|
return token
|
|
|
|
|
|
def _soter(m, method, url, token, body=None, attempts=6):
|
|
"""Signed call to the new servers. They often answer with an HTML error
|
|
page (sometimes a 401) for no reason, so retry those."""
|
|
from .signature import get_auth
|
|
r = None
|
|
for attempt in range(attempts):
|
|
h = m.net.headers.copy()
|
|
h['Content-Type'] = 'application/json'
|
|
h['Accept'] = 'application/json, text/javascript, */*; q=0.01'
|
|
h['Authorization'] = get_auth(method, url, token)
|
|
try:
|
|
r = m.net.session.request(method.upper(), url, headers=h, timeout=(4, 15),
|
|
data=json.dumps(body) if body is not None else None)
|
|
text = r.content.decode('utf-8', errors='replace')
|
|
if not text.lstrip().startswith('<'):
|
|
return r.status_code, text
|
|
except Exception as e:
|
|
LOG('soter {} failed: {}'.format(url.split('?')[0], e))
|
|
LOG('soter {}: attempt {} got an error page ({})'.format(url.split('?')[0], attempt + 1, getattr(r, 'status_code', '?')))
|
|
time.sleep(5)
|
|
return getattr(r, 'status_code', 0), ''
|
|
|
|
|
|
def network_login(m, account_number=''):
|
|
"""Password-less login from the Movistar home network.
|
|
1. autologin gives a 5-minute token for the new servers;
|
|
2. on a fresh box: accountInfo gives the account number and the device is
|
|
registered (one new device on the account, like a password login);
|
|
3. initData trades a device-bound token for a legacyAccessToken, which the
|
|
classic servers this add-on uses accept and renew for 10 days at a time.
|
|
Returns (success, reason)."""
|
|
import uuid
|
|
device_id = (m.cache.load_file('device_id.conf') or '').strip().strip('"')
|
|
try:
|
|
if not account_number or not device_id:
|
|
token = _autologin(m, device_id or uuid.uuid4().hex)
|
|
if not account_number:
|
|
status, text = _soter(m, 'get', SOTER_ACCOUNTINFO_URL, token)
|
|
try:
|
|
account_number = json.loads(text)['ofertas'][0]['accountNumber']
|
|
except (ValueError, KeyError, IndexError, TypeError):
|
|
return False, 'account not found ({})'.format(status)
|
|
if not device_id:
|
|
status, text = _soter(m, 'post', SOTER_DEVICES_URL.format(account=account_number), token, attempts=1)
|
|
new_id = text.strip().strip('"')
|
|
if status not in (200, 201) or len(new_id) != 32:
|
|
return False, 'device registration refused ({})'.format(status)
|
|
device_id = new_id
|
|
# Saved straight away, so a retry reuses it instead of adding another.
|
|
m.cache.save_file('device_id.conf', device_id)
|
|
LOG('network_login: registered device {}...'.format(device_id[:5]))
|
|
token = _autologin(m, device_id)
|
|
except Exception as e:
|
|
LOG('network_login: autologin failed: {}'.format(e))
|
|
return False, 'not on the Movistar network'
|
|
|
|
body = {'accountNumber': account_number, 'sessionUserProfile': 0, 'isKidProfile': False,
|
|
'deviceId': device_id, 'streamMiscellanea': 'HTTPS', 'deviceManufacturerProduct': 'Firefox',
|
|
'streamDRM': 'Widevine', 'streamFormat': 'DASH'}
|
|
status, text = _soter(m, 'post', SOTER_INITDATA_URL, token, body)
|
|
try:
|
|
legacy = json.loads(text).get('legacyAccessToken')
|
|
except (ValueError, AttributeError):
|
|
legacy = None
|
|
props = m.get_token_properties(legacy) if legacy else None
|
|
if not props or props.get('iss') != CLASSIC_ISSUER or props.get('exp', 0) < time.time():
|
|
return False, 'initData refused ({})'.format(status)
|
|
m.save_key_file({'access_token': legacy, 'token_type': 'bearer'})
|
|
m.cache.save_file('access_token.conf', legacy)
|
|
LOG('network_login: OK')
|
|
return True, ''
|
|
|
|
|
|
def extra_channels_m3u8():
|
|
"""Returns [(channel number, m3u8 entry)]."""
|
|
return [(c['chno'], '#EXTINF:-1 tvg-name="{name}" tvg-id="{id}" tvg-logo="{logo}" tvg-chno="{chno}" group-title="{group}",{name}\n{url}\n\n'.format(**c))
|
|
for c in EXTRA_CHANNELS]
|
|
|
|
|
|
def write_if_changed(filename, content):
|
|
try:
|
|
with io.open(filename, 'r', encoding='utf-8', newline='') as f:
|
|
if f.read() == content:
|
|
os.utime(filename, None) # the service reads the time as "last refreshed"
|
|
return False
|
|
except (IOError, OSError):
|
|
pass
|
|
tmp = filename + '.tmp'
|
|
with io.open(tmp, 'w', encoding='utf-8', newline='') as f:
|
|
f.write(content)
|
|
os.replace(tmp, filename) if hasattr(os, 'replace') else shutil.move(tmp, filename)
|
|
return True
|
|
|
|
|
|
def set_iptv_refresh(pvr_dir, addon_id, minutes=60):
|
|
"""Makes the IPTV Simple instances that read this add-on's files reload the
|
|
channel list and programme guide every few minutes. Each reload also makes
|
|
Kodi fetch the guide again. TV guide setups before kodimov.7 reloaded only
|
|
once a day, at 4:00. Returns True if a file changed (IPTV Simple must be
|
|
restarted to use it)."""
|
|
import glob
|
|
import re
|
|
changed = False
|
|
for filename in glob.glob(os.path.join(pvr_dir, 'instance-settings-*.xml')):
|
|
with io.open(filename, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
if 'plugin.video.{}/channels.m3u8'.format(addon_id) not in content: continue
|
|
new = content
|
|
for setting, value in (('m3uRefreshMode', '1'), ('m3uRefreshIntervalMins', str(minutes))):
|
|
new = re.sub(r'<setting id="{}"[^>]*?(/>|>[^<]*</setting>)'.format(setting),
|
|
'<setting id="{}">{}</setting>'.format(setting, value), new)
|
|
if new != content:
|
|
write_if_changed(filename, new)
|
|
LOG('set_iptv_refresh: {} minutes in {}'.format(minutes, filename))
|
|
changed = True
|
|
return changed
|
|
|
|
|
|
def reload_iptv_guide(pvr_dir, addon_id):
|
|
"""Gets a new programme guide into Kodi now instead of at the next hourly
|
|
reload. Restarting IPTV Simple isn't enough: Kodi keeps its own copy of the
|
|
guide and fetches it again only when IPTV Simple reloads, which happens on
|
|
its timer. So restart it with a 1-minute timer, wait for that reload, then
|
|
restart it with the hourly timer. Restarting stops live TV, so this does
|
|
nothing if something is playing."""
|
|
import xbmc
|
|
monitor, player = xbmc.Monitor(), xbmc.Player()
|
|
if player.isPlaying(): return
|
|
if not set_iptv_refresh(pvr_dir, addon_id, 1): return
|
|
try:
|
|
restart_iptv_simple()
|
|
monitor.waitForAbort(90)
|
|
finally:
|
|
set_iptv_refresh(pvr_dir, addon_id, 60)
|
|
while player.isPlaying(): # until then it keeps reloading every minute
|
|
if monitor.waitForAbort(10): return
|
|
if not monitor.abortRequested():
|
|
restart_iptv_simple()
|
|
|
|
|
|
def restart_iptv_simple():
|
|
"""Restarts IPTV Simple so it loads its instance settings again."""
|
|
import xbmc
|
|
for enabled in (False, True):
|
|
xbmc.executeJSONRPC(json.dumps({'jsonrpc': '2.0', 'id': 1, 'method': 'Addons.SetAddonEnabled',
|
|
'params': {'addonid': 'pvr.iptvsimple', 'enabled': enabled}}))
|
|
xbmc.sleep(3000)
|
|
|
|
|
|
def extra_channels_xmltv(session):
|
|
"""Returns the <channel> and <programme> elements for the extra channels,
|
|
taken from a public XMLTV guide and renamed to their tvg-id."""
|
|
import gzip
|
|
import xml.etree.ElementTree as ET
|
|
wanted = dict((c['epg_id'], c) for c in EXTRA_CHANNELS if c.get('epg_id'))
|
|
out = []
|
|
for c in EXTRA_CHANNELS:
|
|
out.append('<channel id="{id}">\n <display-name>{name}</display-name>\n <icon src="{logo}"/>\n</channel>\n'.format(**c))
|
|
try:
|
|
r = session.get(EXTRA_EPG_URL, timeout=(10, 120))
|
|
r.raise_for_status()
|
|
found = 0
|
|
for _, el in ET.iterparse(gzip.GzipFile(fileobj=io.BytesIO(r.content))):
|
|
if el.tag == 'programme':
|
|
c = wanted.get(el.get('channel'))
|
|
if c:
|
|
el.set('channel', c['id'])
|
|
for icon in el.findall('icon'): el.remove(icon) # large fan art
|
|
out.append(ET.tostring(el, encoding='unicode') + '\n')
|
|
found += 1
|
|
el.clear()
|
|
elif el.tag == 'channel':
|
|
el.clear()
|
|
LOG('extra_channels_xmltv: {} programmes'.format(found))
|
|
except Exception as e:
|
|
LOG('extra_channels_xmltv: failed: {}'.format(e))
|
|
return out
|