# -*- coding: utf-8 -*- """Widget listings for skin.milionar's Home rows (docs/plan/skin-milionar/060). Imported through the main.py fast path, so module level imports only json, os and the Kodi modules; nothing here touches the network. Every item that runs an action is a playable plugin URL (`skin_run`): Kodi resolves it, the handler resolves to nothing and starts the real action with RunPlugin. """ import json import os import random import time from urllib.parse import quote, quote_plus import xbmc import xbmcaddon import xbmcgui import xbmcplugin import xbmcvfs from wcs.library2 import clicks from wcs.skin import hero, performance ADDON_ID = 'plugin.video.milionar' BASE = 'plugin://{}/'.format(ADDON_ID) PROFILE = xbmcvfs.translatePath('special://profile/addon_data/{}/'.format(ADDON_ID)) CACHE_DIR = os.path.join(PROFILE, 'skin_cache') TMDB_IMG = 'https://image.tmdb.org/t/p/w500' TMDB_BACKDROP = 'https://image.tmdb.org/t/p/w1280' MAX_ITEMS = 20 # --------------------------------------------------------------------------- helpers def _read_json(name, default): try: with open(os.path.join(PROFILE, name), 'r', encoding='utf-8') as handle: return json.load(handle) except (OSError, ValueError): return default def _poster(path): """TMDb poster_path ("/abc.jpg") -> URL; full URLs and local paths pass through.""" if not path: return '' if path.startswith('/') and not path.startswith('/Users'): return TMDB_IMG + path return path def _run_url(action_query): """Playable plugin URL that resolves to nothing and runs RunPlugin().""" return '{}?action=skin_run&run={}'.format(BASE, quote(action_query, safe='')) def _item(label, path, art=None, info=None, folder=False, runs=True): item = xbmcgui.ListItem(label=label, path=path, offscreen=True) if art: art = {k: v for k, v in art.items() if v} if performance.enabled(): mapped = {v: performance.artwork(v) for v in set(art.values())} art = {k: mapped[v] for k, v in art.items()} item.setArt(art) tag = item.getVideoInfoTag() info = info or {} tag.setTitle(info.get('title', label)) for key, setter in (('year', tag.setYear), ('season', tag.setSeason), ('episode', tag.setEpisode)): try: if info.get(key) not in (None, ''): setter(int(info[key])) except (TypeError, ValueError): pass if info.get('plot'): tag.setPlot(info['plot']) if info.get('tvshowtitle'): tag.setTvShowTitle(info['tvshowtitle']) if info.get('mediatype'): tag.setMediaType(info['mediatype']) if info.get('resume'): tag.setResumePoint(float(info['resume'][0]), float(info['resume'][1])) if info.get('rating'): try: tag.setRating(float(info['rating'])) except (TypeError, ValueError): pass if runs and not folder: item.setProperty('IsPlayable', 'true') if folder: item.setProperty('milionar.folder', '1') item.setProperty('milionar.exec', clicks.exec_for(path, folder)) # run by the skin's onclick (clicks.exec_for) return item def _limit(params=None, default=MAX_ITEMS): """Keep each legacy row's default, but apply slot/profile limits before ListItems.""" try: value = int((params or {}).get('limit') or default) if value <= 0: value = default except (TypeError, ValueError): value = default if value is None: return 10 if performance.enabled() else None return performance.limit(value) def _title_action(entry): """Query that plays/opens one addon title the way the addon's own listings do.""" if entry.get('run'): return entry['run'] title = entry.get('title') or entry.get('name') or '' tmdb_id = entry.get('tmdb_id') or entry.get('id') or '' year = entry.get('year') or '' poster = entry.get('poster') or _poster(entry.get('poster_path')) if entry.get('media_type') in ('series', 'tv'): season = entry.get('season') or 1 episode = entry.get('episode') or 1 return ('action=play_episode_from_addon&series_name={}&season_number={}&episode_number={}' '&episode_title={}&year={}&poster={}&tmdb_id={}').format( quote_plus(title), season, episode, quote_plus(entry.get('episode_title', '')), quote_plus(str(year)), quote_plus(poster), tmdb_id) return 'action=play_movie_from_addon&title={}&year={}&poster={}&tmdb_id={}'.format( quote_plus(title), quote_plus(str(year)), quote_plus(poster), tmdb_id) # --------------------------------------------------------------------------- builders def widget_mytv_channels(params=None): return _channels('my_tv_dialog', 'mytv_channels.json', 'tv', 'widget_mytv_channels', params) def widget_mycinema_channels(params=None): return _channels('my_cinema_dialog', 'mycinema_channels.json', 'cinema', 'widget_mycinema_channels', params) def _channel_run(dialog_action, kind, channel_id, click): """What a click on a channel does (clicks.py `tv_channel` / `cinema_channel`): start the broadcast, one title or open it.""" if click == 'open': return 'action={}&channel_id={}'.format(dialog_action, channel_id) return 'action=channel_play&kind={}&channel_id={}&mode={}'.format(kind, channel_id, click) def _channel_backdrop(kind, titles): """Backdrop for a channel's hero: the first of its titles that has one in the metadata cache.""" from wcs.library2 import meta key = meta.key_tv if kind == 'tv' else meta.key_movie ids = [t['id'] for t in titles if str(t.get('id') or '').isdigit()] cached = meta.get_many([key(i) for i in ids]) path = next((cached[key(i)].get('backdrop_path') for i in ids if (cached.get(key(i)) or {}).get('backdrop_path')), '') return {'backdrop': path} def _channels(dialog_action, filename, kind, row, params): click = clicks.resolve(row, (params or {}).get('click')) channels = _read_json(filename, [])[:_limit(params, None)] program = _Program(channels) if kind == 'tv' else None items = [] for channel in channels: icon, name, titles = channel.get('icon') or '', channel.get('name', ''), channel.get('media_list') or [] run, live, opened = (_run_url(_channel_run(dialog_action, kind, channel.get('id', ''), option)) for option in (click, 'live', 'open')) now = _on_now(channel, program) if program else None state = (now or {}).get('state') or {} resume = (state['resume_pos'], state.get('resume_total') or 0) if state.get('resume_pos') else None items.append(_item(name, run, art={'thumb': icon, 'poster': icon, 'landscape': icon, 'icon': icon}, info={'title': name, 'plot': '{} ({})'.format(name, len(titles)), 'resume': resume})) backdrop = {'backdrop': now['backdrop']} if now and now.get('backdrop') else _channel_backdrop(kind, titles) hero.apply(items[-1], hero.channel(name, [t.get('name') or t.get('title') for t in titles], backdrop, now), clicks.exec_for(live), clicks.exec_for(opened), ids=('live', 'open'), # the broadcast, the channel resumable=bool(resume), # a channel starts its broadcast ("Spustit vysílání"), it does not "play"; in progress: Pokračovat labels=(xbmc.getLocalizedString(hero.S_CONTINUE) if resume else clicks.label(row, 'live'), xbmc.getLocalizedString(hero.S_INFO))) return items def _series_of(channel): return channel.get('media_list') or channel.get('series_list') or [] class _Program: """What the series of the MyTV channels continue with, from the store and the metadata cache (never the network). With the progress shared with the library (wcs/library2/shared.py) a series has one state, the library's, and its next episode follows the rule of Continue watching; otherwise every channel has its own history file.""" def __init__(self, channels): from datetime import date from wcs.library2 import shared self.today = date.today() self.ids = {series['id'] for channel in channels for series in _series_of(channel) if series.get('id')} self.shared = shared.active('tv') self._shows = self._seasons = self._states = None self._histories, self._asked = {}, set() def shows(self): if self._shows is None: from wcs.library2 import meta self._shows = meta.get_many([meta.key_tv(i) for i in self.ids]) return self._shows def show(self, series_id): from wcs.library2 import meta return self.shows().get(meta.key_tv(series_id)) or {} def _season_data(self, series_id): """{season number: its cached detail} of the regular seasons of a series.""" from wcs.library2 import meta if self._seasons is None: self._seasons = meta.get_many([meta.key_season(i, s['season_number']) for i in self.ids for s in self.show(i).get('seasons') or [] if int(s.get('season_number', 0)) > 0]) return {int(s['season_number']): self._seasons.get(meta.key_season(series_id, s['season_number'])) or {} for s in self.show(series_id).get('seasons') or [] if int(s.get('season_number', 0)) > 0} def overview(self, series_id, season, episode): """Plot of an episode from the cached season detail.""" episodes = (self._season_data(series_id).get(int(season)) or {}).get('episodes') or [] return next((ep.get('overview') or '' for ep in episodes if ep.get('episode_number') == int(episode)), '') def history(self, channel_id): if channel_id not in self._histories: from wcs.ai import ChannelHistory self._histories[channel_id] = ChannelHistory.load_channel_history(channel_id) return self._histories[channel_id] def state(self, channel_id, series_id, season, episode): """Watch state of one episode: the library's row, or the channel's resume point of that episode.""" if self.shared: if self._states is None: from wcs.library2 import store self._states = store.get_all_episode_states() try: return (self._states.get(int(series_id)) or {}).get((int(season), int(episode))) or {} except (TypeError, ValueError): return {} progress = self.history(channel_id).get(str(series_id), {}) try: same = (int(progress.get('resume_s', 0)), int(progress.get('resume_e', 0))) == (int(season), int(episode)) if same and float(progress.get('resume_position') or 0) > 0: return {'resume_pos': float(progress['resume_position']), 'resume_total': float(progress.get('resume_total') or 0)} except (TypeError, ValueError): pass return {} def next(self, channel_id, series_id): """(episode of next_up.episode_order, its state) the series continues with in this channel, or None when the next episode is unknown, not aired yet or the series is finished.""" from wcs.library2 import next_up show, cached = self.show(series_id), self._season_data(series_id) missing = {n for n, data in cached.items() if 'episodes' not in data} if ('seasons' not in show or missing) and series_id not in self._asked: self._asked.add(series_id) from wcs.skin import refresh refresh.request(['library:tv:{}'.format(series_id)]) order = next_up.episode_order(show, cached, self.today) if self.shared: if self._states is None: self.state(channel_id, series_id, 0, 0) try: states = self._states.get(int(series_id)) or {} except (TypeError, ValueError): states = {} result = next_up.next_up(order, states, show.get('status'), self.today) if order else {'kind': None} if result['kind'] in ('resume', 'next'): candidate = result['episode'] elif result['kind'] is None: candidate = order[0] if order else None # not started: its first episode else: candidate = None floor = 1 else: progress = self.history(channel_id).get(str(series_id), {}) position = (int(progress.get('season', 1)), int(progress.get('episode', 0))) candidate = next((ep for ep in order if (ep['s'], ep['e']) > position), None) floor = position[0] if not candidate or not candidate['aired'] or any(floor <= n <= candidate['s'] for n in missing): return None return candidate, self.state(channel_id, series_id, candidate['s'], candidate['e']) def _on_now(channel, program): """What a MyTV channel plays when it is started: the first entry of its program queue that is still ahead (the queue of the dialog and of the headless broadcast), else the next episode of its first series that has one.""" channel_id = channel.get('id', '') series = {str(s.get('id')): s for s in _series_of(channel) if s.get('id')} try: from wcs.ai import channel_play from wcs.caching import get_cache_manager cache = get_cache_manager() queue = channel_play.apply_reorder(cache, channel_id, channel_play.drop_watched( channel_id, cache.get('program_list_{}'.format(channel_id)) or [])) except Exception as exc: xbmc.log('[Milionář Skin] program of channel {} unreadable: {}'.format(channel_id, exc), xbmc.LOGWARNING) queue = [] for entry in queue[:1]: props = entry.get('properties') or {} series_id = props.get('tmdb_id', '') try: season, episode = int(props.get('season') or 0), int(props.get('episode') or 0) except (TypeError, ValueError): continue if series_id in series and season > 0 and episode > 0: key = series[series_id]['id'] return {'series': props.get('series_name') or series[series_id].get('name', ''), 's': season, 'e': episode, 'title': props.get('episode_title', ''), 'plot': props.get('plot', ''), 'state': program.state(channel_id, key, season, episode), 'backdrop': program.show(key).get('backdrop_path') or ''} for entry in series.values(): found = program.next(channel_id, entry['id']) if found: candidate, state = found return {'series': entry.get('name', ''), 's': candidate['s'], 'e': candidate['e'], 'title': candidate.get('name', ''), 'plot': program.overview(entry['id'], candidate['s'], candidate['e']), 'state': state, 'backdrop': program.show(entry['id']).get('backdrop_path') or ''} return None def widget_mytv_next(params=None): from wcs.library2 import rows click = clicks.resolve('widget_mytv_next', (params or {}).get('click')) limit = _limit(params) channels = _read_json('mytv_channels.json', []) program = _Program(channels) items, seen = [], set() for channel in channels: for series in _series_of(channel): series_id = series.get('id') if not series_id or series_id in seen: continue seen.add(series_id) found = program.next(channel.get('id'), series_id) if not found: continue candidate, state = found show = program.show(series_id) season, next_ep = candidate['s'], candidate['e'] resume = (state['resume_pos'], state.get('resume_total') or 0) if state.get('resume_pos') else None poster = _poster(series.get('poster_path')) folder = False # the channel's own action: keeps the MyTV progress (the library's when the two are shared) play = _run_url(('action=play_episode_from_addon&series_name={}&season_number={}&episode_number={}' '&year={}&poster={}&tmdb_id={}&mytv_channel_id={}').format( quote_plus(series.get('name', '')), season, next_ep, series.get('year', ''), quote_plus(poster), series_id, channel.get('id', ''))) if click == 'play': path = play elif click == 'channel_open': path = _run_url('action=my_tv_dialog&channel_id={}'.format(channel.get('id', ''))) else: click_meta = {'title': series.get('name', ''), 'poster': poster, 'plot': series.get('overview', '')} path, folder = clicks.title_path('episode', series_id, click, season, next_ep, click_meta) card = rows.series_facts(show, series, series_id) items.append(_item('S{:02d}E{:02d}'.format(int(season), next_ep), path, art={'thumb': poster, 'poster': poster, 'icon': poster, 'landscape': card['fanart']}, info={'title': 'S{:02d}E{:02d}'.format(int(season), next_ep), 'tvshowtitle': series.get('name', ''), 'season': season, 'episode': next_ep, 'mediatype': 'episode', 'resume': resume}, folder=folder, runs=not folder)) for key, value in {'type': 'episode', 'tmdb': series_id, 's': season, 'e': next_ep, 'channel_id': channel.get('id', ''), 'channel_kind': 'tv', 'ctx_title': series.get('name', '')}.items(): items[-1].setProperty('milionar.' + key, str(value)) detail, _folder = clicks.title_path('episode', series_id, 'detail', season, next_ep) hero.apply(items[-1], hero.episode(card, {'s': season, 'e': next_ep}, state), clicks.exec_for(play), clicks.exec_for(detail), resumable=bool(resume)) if len(items) >= limit: return items return items def _recent_entries(): return _read_json('global_watch_history.json', None) or _read_json('recently_played.json', []) def _entry_item(entry, label=None, click=None): """Item of a title entry (tmdb_id, media_type, title, poster, plot); `click` None keeps the entry's own action.""" title = entry.get('title', '') poster = entry.get('poster', '') series = entry.get('media_type') == 'series' label = label or (title if not series else '{} S{:02d}E{:02d}'.format( title, int(entry.get('season') or 0), int(entry.get('episode') or 0))) folder = False if click is None: path = _run_url(_title_action(entry)) else: path, folder = clicks.entry_path(entry, click) item = _item(label, path, art={'thumb': poster, 'poster': poster, 'icon': poster}, info={'title': label, 'year': entry.get('year'), 'plot': entry.get('plot', ''), 'rating': entry.get('rating'), 'mediatype': 'episode' if series else 'movie', 'tvshowtitle': title if series else ''}, folder=folder, runs=not folder) if entry.get('tmdb_id') or entry.get('id'): item.setProperty('milionar.type', 'tv' if entry.get('media_type') in ('series', 'tv') else 'movie') item.setProperty('milionar.tmdb', str(entry.get('tmdb_id') or entry['id'])) return item def widget_recently_played(params=None): return [_entry_item(e) for e in _recent_entries()[:_limit(params)]] def widget_in_progress(params=None): """Library items with a resume point (movies and episodes), newest first.""" items = [] for dbtype, method, key, props in ( ('movie', 'VideoLibrary.GetMovies', 'movies', ['title', 'year', 'art', 'resume', 'lastplayed', 'plot']), ('episode', 'VideoLibrary.GetEpisodes', 'episodes', ['title', 'showtitle', 'season', 'episode', 'art', 'resume', 'lastplayed', 'plot'])): request = {'jsonrpc': '2.0', 'id': 1, 'method': method, 'params': { 'properties': props, 'limits': {'end': 30}, 'filter': {'field': 'inprogress', 'operator': 'true', 'value': ''}, 'sort': {'method': 'lastplayed', 'order': 'descending'}}} result = json.loads(xbmc.executeJSONRPC(json.dumps(request))).get('result', {}) items.extend((row.get('lastplayed', ''), dbtype, row) for row in result.get(key, [])) items.sort(key=lambda pair: pair[0], reverse=True) selected = [] for _stamp, dbtype, row in items[:_limit(params)]: art = row.get('art', {}) poster = art.get('poster') or art.get('thumb') or art.get('tvshow.poster', '') run = 'action=lib_resume&dbtype={}&dbid={}'.format(dbtype, row.get(dbtype + 'id')) label = row.get('title', '') info = {'title': label, 'plot': row.get('plot', ''), 'year': row.get('year'), 'mediatype': dbtype} if dbtype == 'episode': info.update(tvshowtitle=row.get('showtitle', ''), season=row.get('season'), episode=row.get('episode')) selected.append(_item( label, _run_url(run), art={'thumb': art.get('thumb') or poster, 'poster': poster, 'fanart': art.get('fanart', ''), 'icon': poster}, info=info)) return selected def _random_library_entries(count, rng=None): """Random titles of My movies / My series as game entries (so the row is never empty for a user with lists).""" from wcs.library2 import rows data = rows.Data() titles = [(media, i) for media in ('movie', 'tv') for i in data.list_ids(media)] (rng or random).shuffle(titles) titles = titles[:count] data.preload_movies([i for media, i in titles if media == 'movie']) data.preload_series([i for media, i in titles if media == 'tv']) entries = [] for media, tmdb_id in titles: card = rows.movie_card(data, tmdb_id) if media == 'movie' else rows.series_card(data, tmdb_id) entries.append({'title': card['title'], 'year': card['year'], 'plot': card['plot'], 'poster': card['poster'], 'backdrop': card.get('backdrop') or '', 'media_type': 'series' if media == 'tv' else 'movie', 'tmdb_id': str(tmdb_id)}) return entries GAME_TABS = (('tab_movies', 'widget_demo_movies', 32747), ('tab_series', 'widget_demo_series', 32748)) from wcs.skin.hero_contract import WIDGET_SLOTS as HERO_SLOTS def _game_buttons(params): """The Home buttons of every Milionář card: jump to the Filmy / Seriály demo rows of the section, play a random title (a tab whose row the user removed from the section is left out), open the AI costs card.""" from wcs.skin.hero_contract import BASES sec = (params or {}).get('sec') or 'game' addon = xbmcaddon.Addon(ADDON_ID) buttons = [] for action, row, string_id in GAME_TABS: slot = next((n for n in range(1, HERO_SLOTS + 1) if xbmc.getInfoLabel('Skin.String(mil.{}.{}.type)'.format(sec, n)) == row), 0) if slot and sec in BASES: buttons.append((action, addon.getLocalizedString(string_id), 'SetFocus({})'.format(BASES[sec] + slot * 100))) buttons.append(('random_game', addon.getLocalizedString(32749), 'RunPlugin({}?action=milionar_demo&demo=random)'.format(BASE.rstrip('/')))) buttons.append(('ai_costs', addon.getLocalizedString(32800), 'RunPlugin({}?action=ai_costs)'.format(BASE.rstrip('/')))) return buttons def _game_metadata(data, entry): """Use the library's cached TMDb card for every game topic, including demos.""" from wcs.library2 import rows entry = dict(entry) ident = int(entry.get('tmdb_id') or entry.get('id') or 0) if not ident: return entry, None media = 'tv' if entry.get('media_type') in ('series', 'tv') else 'movie' detail = data.show(ident) if media == 'tv' else data.movie(ident) snap = data.snapshot(media, ident) if not detail and not snap: return entry, None if media == 'tv': card = rows.series_facts(detail, snap, ident) else: card = rows.movie_card(data, ident) for key in ('title', 'year', 'plot', 'poster', 'backdrop', 'fanart'): if card.get(key): entry[key] = card[key] return entry, card def _request_game_metadata(data): if data.missing: from wcs.skin import refresh refresh.request(sorted(data.missing)) def _challenge(item, title, media, year, backdrop, buttons, plot='', card=None): """Shared game hero, with the same TMDb facts as the library's movie/series hero.""" if card: texts = hero.series(card) if media in ('series', 'tv') else hero.movie(card) kicker, plot = texts['kicker'], texts['plot'] else: from wcs.games import game_view kind = game_view.KICKERS['series' if media in ('series', 'tv') else 'movie'] kicker = hero.track('{} · {}'.format(kind, year) if year else kind) hero.challenge(item, title, kicker, backdrop, buttons, plot) def widget_game_topics(params=None): from wcs.library2 import rows data = rows.Data() click = clicks.resolve('widget_game_topics', (params or {}).get('click')) limit = _limit(params) buttons = _game_buttons(params) items = [] entries = _recent_entries()[:limit] if len(entries) < limit: from datetime import date from wcs.skin import refresh revision = refresh.revisions(['widget_game_topics']).get('widget_game_topics', 0) seed = json.dumps([revision, date.today().isoformat(), limit, {k: (params or {}).get(k, '') for k in ('sort', 'style', 'click', 'hw', 'up')}], sort_keys=True) rng = random.Random(seed) seen = {(e.get('title'), str(e.get('year'))) for e in entries} extra = [e for e in _random_library_entries(limit, rng) if (e['title'], str(e['year'])) not in seen] rng.shuffle(extra) entries = entries + extra[:limit - len(entries)] for entry in entries: entry, card = _game_metadata(data, entry) poster = entry.get('poster', '') backdrop = hero.fanart(entry) folder = False if click == 'game' or not (entry.get('tmdb_id') or entry.get('id')): from wcs.games import millionaire media = 'tv' if entry.get('media_type') == 'series' else 'movie' path = _run_url(millionaire.query(millionaire.Subject( title=entry.get('title', ''), tmdb_id=str(entry.get('tmdb_id') or ''), media_type=media, year=entry.get('year') or '', poster=poster, backdrop=backdrop, plot=entry.get('plot', '')))) else: path, folder = clicks.entry_path(entry, click) items.append(_item(entry.get('title', ''), path, art={'thumb': poster, 'poster': poster, 'icon': poster, 'fanart': backdrop, 'landscape': backdrop}, info={'title': entry.get('title', ''), 'year': entry.get('year'), 'plot': entry.get('plot', ''), 'mediatype': 'movie'}, folder=folder, runs=not folder)) if entry.get('tmdb_id') or entry.get('id'): items[-1].setProperty('milionar.type', 'tv' if entry.get('media_type') in ('series', 'tv') else 'movie') items[-1].setProperty('milionar.tmdb', str(entry.get('tmdb_id') or entry['id'])) _challenge(items[-1], entry.get('title', ''), entry.get('media_type'), entry.get('year'), backdrop, buttons, entry.get('plot', ''), card) _request_game_metadata(data) return items def _demo_items(media, params): """Home row of the offline Millionaire demo (wcs/games/demo): a click starts that title's game.""" from wcs.games.demo import catalog from wcs.library2 import rows data = rows.Data() buttons = _game_buttons(params) items = [] for title in catalog.by_media(media)[:_limit(params, None)]: entry, card = _game_metadata(data, {'tmdb_id': title.tmdb_id, 'media_type': media, 'title': title.title, 'year': title.year, 'plot': title.plot, 'poster': _poster(title.poster), 'backdrop': title.fanart}) poster, backdrop = entry['poster'], hero.fanart(entry) icon = 'DefaultMovies.png' if media == 'movie' else 'DefaultTVShows.png' item = _item(entry['title'], _run_url('action=milionar_demo&demo={}'.format(title.id)), art={'thumb': poster, 'poster': poster, 'icon': icon, 'landscape': backdrop, 'fanart': backdrop}, info={'title': entry['title'], 'year': entry['year'], 'plot': entry['plot'], 'mediatype': 'movie' if media == 'movie' else 'tvshow'}) _challenge(item, entry['title'], media, entry['year'], backdrop, buttons, entry['plot'], card) items.append(item) _request_game_metadata(data) return items def widget_demo_movies(params=None): return _demo_items('movie', params) def widget_demo_series(params=None): return _demo_items('series', params) def widget_ai_recent(params=None): click = clicks.resolve('widget_ai_recent', (params or {}).get('click')) items = [_item('AI chat', _run_url('action=ai_chat_recommendation'), art={'icon': 'DefaultAddonsSearch.png', 'thumb': 'DefaultAddonsSearch.png'}, info={'title': 'AI chat'})] for entry in _read_json('ai_recommendations.json', [])[:min(30, _limit(params, 31) - 1)]: entry = dict(entry, media_type='series' if entry.get('media_type') == 'tv' else 'movie') if entry['media_type'] == 'series': entry['season'], entry['episode'] = 1, 1 items.append(_entry_item(entry, entry.get('title', ''), click if (entry.get('tmdb_id') or entry.get('id')) else None)) return items def _cached(name): """Serve the last network snapshot, requesting absent/stale data without waiting.""" from wcs.skin import refresh from wcs.skin.refresher import SCHEDULE data = _read_json(os.path.join('skin_cache', name + '.json'), {}) items = data.get('items') if isinstance(data, dict) else None try: stale = time.time() - float(data.get('built') or 0) >= SCHEDULE.get(name, 24 * 3600) except (AttributeError, TypeError, ValueError): stale = True if not isinstance(items, list) or stale: refresh.request([name]) return items if isinstance(items, list) else [] def _entries_to_items(entries, row, params): click = clicks.resolve(row, (params or {}).get('click')) return [_entry_item(e, click=click if (e.get('tmdb_id') or e.get('id')) else None) for e in entries[:_limit(params, None)]] def widget_mycinema_recommended(params=None): return _entries_to_items(_cached('widget_mycinema_recommended'), 'widget_mycinema_recommended', params) def widget_tmdb_popular_movies(params=None): return _entries_to_items(_cached('widget_tmdb_popular_movies'), 'widget_tmdb_popular_movies', params) def widget_tmdb_popular_series(params=None): return _entries_to_items(_cached('widget_tmdb_popular_series'), 'widget_tmdb_popular_series', params) def widget_kraska_root(params=None): items = [] for entry in _cached('widget_kraska_root')[:_limit(params, None)]: url = '{}?action=kraska_browse_path&sc_path={}&sc_title={}'.format( BASE, quote_plus(entry.get('path', '')), quote_plus(entry.get('title', ''))) items.append(_item(entry.get('title', ''), url, art={'icon': entry.get('icon', 'DefaultFolder.png')}, folder=True)) return items CSFD_CHARTS = ( ('ČSFD Filmy', 'csfd_movies_menu'), ('ČSFD Seriály', 'csfd_series_menu'), ('Nejlepší filmy ČSFD', 'csfd_top_movies'), ('Nejlepší seriály ČSFD', 'csfd_top_series')) def widget_search(params=None): """Rows of window 1120: kind=movie|tv|person from the last search cache, kind=history from the addon history.""" kind = (params or {}).get('kind', '') if kind == 'history': from wcs import user_data queries = user_data.load_search_history(PROFILE, 'search_history_universal.json') return [_item(q, _run_url('action=skin_search&query={}'.format(quote_plus(q))), art={'icon': 'DefaultSearch.png'}, info={'title': q}) for q in queries[:_limit(params, None)]] data = _read_json(os.path.join('skin_cache', 'search_current.json'), {}) items = [] entries = [entry for entry in data.get('items', []) if entry.get('kind') == kind] if isinstance(data, dict) else [] for entry in entries[:_limit(params, 40)]: poster = entry.get('poster', '') item = _item(entry.get('title', ''), _run_url(entry.get('run', '')), art={'thumb': poster, 'poster': poster, 'icon': poster}, info={'title': entry.get('title', ''), 'year': entry.get('year'), 'plot': entry.get('plot', ''), 'rating': entry.get('rating'), 'mediatype': 'movie' if kind != 'tv' else 'tvshow'}) if kind in ('movie', 'tv') and entry.get('tmdb_id'): item.setProperty('milionar.type', kind) # the context menu of 080 (state from the store) item.setProperty('milionar.tmdb', str(entry['tmdb_id'])) items.append(item) from wcs.library2 import menu menu.attach_items([(i.getPath(), i, False) for i in items]) return items def widget_csfd_charts(params=None): return [_item(title, '{}?action={}'.format(BASE, action), art={'icon': 'DefaultFolder.png'}, folder=True) for title, action in CSFD_CHARTS[:_limit(params, len(CSFD_CHARTS))]] WIDGETS = {name: func for name, func in list(globals().items()) if name.startswith('widget_') and callable(func)} FOLDER_WIDGETS = {'widget_kraska_root', 'widget_csfd_charts'} WIDGETS['widget_search'] = widget_search # --------------------------------------------------------------------------- entry point def build(row, params=None): """The shared legacy-row builder for Home listings and migration warm-up.""" return WIDGETS[row](params or {}) def run(handle, params): started = time.perf_counter() action = params.get('action', '') if action == 'skin_run': # resolve to nothing, then start the real action (Kodi must not "play" a dialog URL) xbmcplugin.setResolvedUrl(handle, False, xbmcgui.ListItem(offscreen=True)) target = params.get('run', '') if target: xbmc.executebuiltin('RunPlugin({}?{})'.format(BASE.rstrip('/'), target)) return items = [] if action in WIDGETS: try: items = build(action, params) except Exception as exc: # a broken row must never break Home xbmc.log('[Milionář Skin] {} failed: {}: {}'.format(action, type(exc).__name__, exc), xbmc.LOGERROR) folder = action in FOLDER_WIDGETS xbmcplugin.addDirectoryItems(handle, [(i.getPath(), i, folder or i.getProperty('milionar.folder') == '1') for i in items], len(items)) xbmcplugin.endOfDirectory(handle, succeeded=True, cacheToDisc=False) xbmc.log('[Milionář Skin] {} -> {} items in {:.0f} ms'.format( action, len(items), (time.perf_counter() - started) * 1000), xbmc.LOGDEBUG) def invalidate(names=None): """Dirty only affected rows; retain snapshots until background replacement succeeds.""" from wcs.skin import refresh rows = set(names or refresh.ALL_ROWS) refresh.changed(rows) refresh.request(rows & (refresh.NETWORK_ROWS | refresh.VINYL_ROWS), force=True)