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'


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 network_login(m, account_number):
  """Password-less login from the Movistar home network.
  1. autologin (only reachable inside Movistar's network) gives a 5-minute
     token for the new servers;
  2. their initData trades it for a legacyAccessToken, which the classic
     servers this add-on uses accept (and renew for 10 days at a time).
  Returns (success, reason)."""
  try:
    from urllib.parse import urlsplit
  except ImportError:
    from urlparse import urlsplit
  from .signature import get_auth

  device_id = (m.cache.load_file('device_id.conf') or '').strip().strip('"')
  if not account_number: return False, 'no account number'
  if not device_id: return False, 'no registered device'

  headers = m.net.headers.copy()
  headers['Content-Type'] = 'application/json'
  url = AUTOLOGIN_URL.format(device=device_id, ts=int(time.time() * 1000))
  try:
    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'):
        return False, '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')
  except Exception as e:
    LOG('network_login: autologin failed: {}'.format(e))
    return False, 'not on the Movistar network'
  if not token: return False, 'autologin refused'

  body = {'accountNumber': account_number, 'sessionUserProfile': 0, 'isKidProfile': False,
          'deviceId': device_id, 'streamMiscellanea': 'HTTPS', 'deviceManufacturerProduct': 'Firefox',
          'streamDRM': 'Widevine', 'streamFormat': 'DASH'}
  data = None
  for attempt in range(4):
    h = headers.copy()
    h['Authorization'] = get_auth('post', SOTER_INITDATA_URL, token)
    try:
      r = m.net.session.post(SOTER_INITDATA_URL, data=json.dumps(body), headers=h, timeout=(4, 15))
      data = r.json()
      break
    except Exception:
      # The new servers sometimes answer with an HTML error page; retry.
      LOG('network_login: initData attempt {} failed ({})'.format(attempt + 1, getattr(r, 'status_code', '?')))
      time.sleep(3)
  legacy = data.get('legacyAccessToken') if isinstance(data, dict) else 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'
  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 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