kodimovplus/resources/lib/kodimov.py

# 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: 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 use_hourly_iptv_refresh(pvr_dir, addon_id):
  """TV guide setups before kodimov.7 made IPTV Simple reload the channel list
  and programme guide only once a day, at 4:00, so a new epg.xml didn't reach
  Kodi for up to a day. Switch them to a reload every hour, which also makes
  Kodi fetch the guide again. Returns True if IPTV Simple must be restarted."""
  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', '60')):
      new = re.sub(r'<setting id="{}"[^>]*?(/>|>[^<]*</setting>)'.format(setting),
                   '<setting id="{}">{}</setting>'.format(setting, value), new)
    if write_if_changed(filename, new):
      LOG('use_hourly_iptv_refresh: updated {}'.format(filename))
      changed = True
  return changed


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