# -*- coding: utf-8 -*- """Playback tracker, runs in the service (docs/plan/milionar-library/040 §4, 020 §2). Player callbacks only enqueue events. One worker loop drains the queue, polls the position, saves it and applies the thresholds, so no player or GUI call ever happens inside a callback (Kodi GUI is not thread-safe). """ import json import queue import threading import time import xbmc import xbmcaddon import xbmcgui from wcs.library2 import autoplay, jobs, play, policy, store, urls ADDON_ID = 'plugin.video.milionar' NP_PROPERTY = 'milionar.np' AUTOSAVE_S = 10 SAFETY_SEEK_AFTER_S = 3 ADOPT_EVERY_S = 30 LOOP_TICK_S = 0.5 BROKE_OFF_POLL_S = 10 # a position read this recently is what the player really had at the 'end' BROKE_OFF_MARGIN_S = 120 def _log(msg, level=xbmc.LOGINFO): xbmc.log('[Milionář Tracker] {}'.format(msg), level) def thresholds(): addon = xbmcaddon.Addon(ADDON_ID) def number(key, default): try: return float(addon.getSetting(key) or default) except ValueError: return float(default) return number('lib_resume_min_seconds', policy.DEFAULT_RESUME_MIN_SECONDS), \ number('lib_watched_percent', policy.DEFAULT_WATCHED_PERCENT) class Session: def __init__(self, np, log_id, started): self.np = np self.log_id = log_id self.started = started self.position = 0.0 self.total = 0.0 self.polled_at = None # when position was last read from the player self.last_save = 0.0 self.av_started_at = started self.safety_checked = False self.kind = 'movie' if np['type'] == 'movie' else 'tv' # a MyTV / MyCinema broadcast in the shared mode: tracked like any play, what comes next is the channel's program self.follow = np.get('follow', True) is not False # autoplay (050 §4): one plan per playback, one playnext() per playback self.autoplay_started = False self.plan = None self.queued_url = None self.advanced = False self.declined = False self.marked_watched = False self.end_notice = None def key(self): return (self.kind, self.np['tmdb'], self.np.get('s'), self.np.get('e')) class Tracker(xbmc.Player): def __init__(self): super().__init__() self.events = queue.Queue() # callbacks: enqueue only def onAVStarted(self): self.events.put(('started', time.time(), None)) def onPlayBackStopped(self): self.events.put(('stopped', time.time(), None)) def onPlayBackEnded(self): self.events.put(('ended', time.time(), None)) def onPlayBackError(self): self.events.put(('error', time.time(), None)) def onPlayBackPaused(self): self.events.put(('paused', time.time(), None)) def onPlayBackSeek(self, time_ms, seek_offset): self.events.put(('seek', time.time(), None)) class AdoptionMonitor(xbmc.Monitor): def __init__(self, events): super().__init__() self.events = events def onNotification(self, sender, method, data): if method in ('VideoLibrary.OnUpdate', 'VideoLibrary.OnRemove', 'GUI.OnContainerChanged'): self.events.put(('adopt_changed', time.time(), None)) def _read_np(): raw = xbmcgui.Window(10000).getProperty(NP_PROPERTY) try: return json.loads(raw) if raw else None except ValueError: return None class Engine: """The loop body; `player` is a Tracker (or a test double with the same methods).""" def __init__(self, player, monitor, now=time.time): self.player = player self.monitor = monitor self.now = now self.session = None self.last_adopt = 0.0 self.adopt_signature = None self.adopt_cursor = None self.adopt_changed = False # ----------------------------------------------------------------- events def handle(self, event, when, data=None): if event == 'started': self._start(when) elif event in ('stopped', 'ended', 'error') and self.session: self._finish(event, when) elif event in ('paused', 'seek') and self.session: self._poll() self._save(force=True) elif event == 'autoplay_prefetch': if self.session and self.session.np['token'] == data['token'] and self.session.plan is None: self.session.plan = data['plan'] elif event == 'autoplay_plan': self._on_plan(data) elif event == 'autoplay_action': self._on_action(data) elif event == 'adopt_changed': self.adopt_changed = True def _start(self, when): np = _read_np() try: playing = self.player.getPlayingFile() except RuntimeError: return if self.session: # a new playback without a stop (next episode through the playlist) self._finish('ended' if self.session.follow or self._ran_out(self.session) else 'stopped', when) if not policy.np_matches(np, playing, self.now()): _log('playback ignored (no matching contract): {}'.format(playing[:80])) return started = int(self.now()) log_id = store.log_start('movie' if np['type'] == 'movie' else 'tv', np['tmdb'], np.get('s'), np.get('e'), started) self.session = Session(np, log_id, self.now()) xbmcgui.Window(10000).clearProperty(NP_PROPERTY) _log('tracking {} {}'.format(np['type'], np['tmdb'])) if np.get('origin') == play.ORIGIN_CHANNEL: self._spawn(self._list_worker, self.session.kind, np['tmdb'], np.get('show_title') or np.get('title') or '') if not self.session.follow: return # plan the follow-up now (network at most once, off the loop): the countdown point or a seek to the very # end must not wait for TMDb self._spawn(self._prefetch_worker, np, np['token']) def _broke_off(self, s): """Kodi reports 'ended' also when a stream fails a few seconds in (a broken or unsupported file) or a file is cut short. That is not the end of the film: the player read the position moments before and it was far from the end. A seek to the very end is still the end: the poll before it is old, or the position is near the end.""" if s.polled_at is None or s.total <= 0: return False fresh = (self.now() - s.polled_at) <= BROKE_OFF_POLL_S near_end = (s.total - s.position) <= max(BROKE_OFF_MARGIN_S, s.total * 0.05) if fresh and not near_end: _log('{} {} stopped at {:.0f} s of {:.0f} s: a broken stream, not the end'.format( s.np['type'], s.np['tmdb'], s.position, s.total), xbmc.LOGWARNING) return True return False @staticmethod def _ran_out(s): """A channel's playlist moved on: the item counts as played to the end only from its countdown point on (the channel's own countdown starts the next item there); a skip earlier leaves it in progress.""" return s.total > 0 and (s.total - s.position) <= autoplay.settings()[1] + 5 @staticmethod def _list_worker(kind, tmdb_id, title): try: from wcs.library2 import shared shared.ensure_listed(kind, tmdb_id, title) except Exception as exc: _log('adding the channel title to the list failed: {}'.format(exc), xbmc.LOGWARNING) def _poll(self): s = self.session if not s: return try: s.position = float(self.player.getTime()) total = float(self.player.getTotalTime()) except RuntimeError: return s.total = policy.effective_total(total, s.np.get('runtime_s')) s.polled_at = self.now() def _save(self, force=False): s = self.session if not s: return min_seconds, percent = thresholds() due = (self.now() - s.last_save) >= AUTOSAVE_S if s.total > 0 and (s.position / s.total) * 100.0 >= percent: return # past the watched mark: a crash here must not leave a resume point at 99 % if (force or due) and s.position >= min_seconds: store.save_resume(s.kind, s.np['tmdb'], s.position, s.total, s.np.get('s'), s.np.get('e'), source={'ident': s.np.get('ident'), 'provider': s.np.get('provider'), 'file_name': s.np.get('file_name')}) store.log_update(s.log_id, s.position, s.total) s.last_save = self.now() def _finish(self, event, when): s = self.session self.session = None min_seconds, percent = thresholds() ended = event == 'ended' and not self._broke_off(s) if ended and s.total > 0: # the natural end is the end whatever the last poll saw (seek to the very end) s.position = max(s.position, s.total) result = policy.outcome(s.position, s.total, ended, min_seconds, percent) kind, tmdb_id, season, episode = s.key() source = {'ident': s.np.get('ident'), 'provider': s.np.get('provider'), 'file_name': s.np.get('file_name')} if result == 'watched': if not s.marked_watched: # autoplay already marked it when the countdown ended (no double count) store.mark_watched(kind, tmdb_id, season, episode, source=source) elif s.marked_watched: pass elif result == 'resume': store.save_resume(kind, tmdb_id, s.position, s.total, season, episode, source=source) if s.marked_watched: result = 'watched' store.log_finish(s.log_id, int(self.now()), s.position, s.total, result) if ended and s.follow and not s.advanced and not s.declined and autoplay.settings()[0]: self._follow_up(s) # the episode ended: whatever the plan says follows now elif s.queued_url and event in ('stopped', 'error'): self.drop_queue(s) # the user stopped: nothing may start by itself later # ----------------------------------------------------------------- loop def tick(self): s = self.session if s: self._poll() self._save() if not s.safety_checked and s.np.get('start_at', 0) > 0 and (self.now() - s.av_started_at) >= SAFETY_SEEK_AFTER_S: s.safety_checked = True if s.position < s.np['start_at'] - 5: _log('safety seek to {:.0f}s (player at {:.0f}s)'.format(s.np['start_at'], s.position)) self.player.seekTime(s.np['start_at']) if s: self._check_autoplay(s) def _follow_up(self, s): """Natural end of an episode / film: start the next episode, or tell why not (also when the dialog was dismissed, never shown (seek to the very end) or the plan was still pending).""" plan = s.plan if plan is None and s.np.get('type') == 'episode': try: plan = autoplay.plan(s.np) except Exception as exc: _log('autoplay plan failed at the end: {}'.format(exc), xbmc.LOGWARNING) plan = {'kind': 'none'} s.plan = plan if not plan: return kind = plan['kind'] if kind == 'series_next': if not s.queued_url: # no request left yet (late plan): short-lived one for the resolve play.prepare(plan['canonical'], src='auto', start='start', max_age=60) s.queued_url = plan['canonical'] s.advanced = True self.play_next(plan) elif kind == 'upcoming': self._notify(s.end_notice or self._text(32280).format(plan['season_episode'], plan['air_label'])) elif kind == 'finished': self._notify(s.end_notice or self._text(32281)) # ----------------------------------------------------------------- autoplay (050 §4-5) def _spawn(self, target, *args): """Worker threads for network / modal work; tests run them inline.""" threading.Thread(target=target, args=args, daemon=True).start() def _post(self, event, data): self.player.events.put((event, self.now(), data)) def _check_autoplay(self, s): if not s.follow or s.autoplay_started or s.total <= 0 or s.np.get('type') not in ('episode', 'movie'): return enabled, countdown, trigger = autoplay.settings() remaining = s.total - s.position if not enabled or remaining > countdown: return s.autoplay_started = True if s.plan is not None: # prefetched: act at once self._apply_plan(s, s.plan, remaining, autoplay.dialog_duration(remaining, countdown, trigger)) return self._spawn(self._plan_worker, s.np, s.np['token'], remaining, countdown, trigger) def _prefetch_worker(self, np, token): try: plan = autoplay.plan(np) except Exception as exc: _log('autoplay plan failed: {}: {}'.format(type(exc).__name__, exc), xbmc.LOGWARNING) plan = {'kind': 'none'} self._post('autoplay_prefetch', {'token': token, 'plan': plan}) def _plan_worker(self, np, token, remaining, countdown, trigger): try: plan = autoplay.plan(np) except Exception as exc: _log('autoplay plan failed: {}: {}'.format(type(exc).__name__, exc), xbmc.LOGWARNING) plan = {'kind': 'none'} self._post('autoplay_plan', {'token': token, 'plan': plan, 'remaining': remaining, 'duration': autoplay.dialog_duration(remaining, countdown, trigger)}) def _on_plan(self, data): s = self.session if not s or s.np['token'] != data['token'] or s.advanced: return self._apply_plan(s, data['plan'], data['remaining'], data['duration']) def _apply_plan(self, s, plan, remaining, duration): s.plan = plan kind = plan['kind'] if kind == 'series_next': self._queue(s, plan, max_age=remaining + 120) self._spawn(self._dialog_worker, plan, duration, s.np['token']) elif kind == 'collection': self._spawn(self._dialog_worker, plan, duration, s.np['token']) elif kind == 'upcoming': s.end_notice = self._text(32280).format(plan['season_episode'], plan['air_label']) elif kind == 'finished': s.end_notice = self._text(32281) def _queue(self, s, plan, max_age): """Remember the next canonical URL and leave a long-lived auto-pick request for its (lazy) resolve. PlayMedia does not put the playing item into Kodi's playlist (position -1), so there is no playlist to queue into: the engine starts the next item itself with Player.play (no stop() first) from this loop. """ play.prepare(plan['canonical'], src='auto', start='start', max_age=max_age) s.queued_url = plan['canonical'] def _dialog_worker(self, plan, duration, token): """Show the existing countdown dialog (modal, in this worker); the result goes back as an event.""" action = self._show_dialog(plan, duration, token) if action: self._post('autoplay_action', {'token': token, 'action': action}) def _show_dialog(self, plan, duration, token): from wcs.playback.dialogs.countdown import AutoplayCountdownDialog as Dialog addon = xbmcaddon.Addon(ADDON_ID) collection = plan['kind'] == 'collection' def on_timeout(): self._post('autoplay_action', {'token': token, 'action': 'timeout'}) dialog = Dialog( Dialog.get_dialog_xml_filename(), addon.getAddonInfo('path'), show_countdown_at=duration, next_episode_title=plan['title'], next_episode_season_episode=(plan['year'] if collection else plan['season_episode']), next_episode_thumb=plan.get('thumb', ''), series_name=(self._text(32283) if collection else plan['series_name']), next_episode_rating=plan.get('rating', ''), next_episode_rating_only=plan.get('rating', ''), next_episode_year_only=plan.get('year', ''), timeout_callback=(None if collection else on_timeout)) dialog.doModal() user = dialog.get_user_action() del dialog return {Dialog.ACTION_PLAY_NOW: 'play_now', Dialog.ACTION_CANCEL_AUTOPLAY: 'cancel', Dialog.ACTION_DISMISS: 'dismiss', Dialog.ACTION_TIMEOUT: None}.get(user, 'dismiss') def _on_action(self, data): s = self.session if not s or s.np['token'] != data['token'] or not s.plan: return action, plan = data['action'], s.plan if plan['kind'] == 'collection': if action == 'play_now' and not s.advanced: self._mark_current_watched(s) self._queue(s, plan, max_age=60) self._advance(s) return # timeout / dismiss / cancel: no automatic start if action in ('play_now', 'timeout'): if not s.advanced and s.queued_url: self._mark_current_watched(s) self._advance(s) elif action == 'cancel': self.drop_queue(s) s.declined = True def _mark_current_watched(self, s): kind, tmdb_id, season, episode = s.key() store.mark_watched(kind, tmdb_id, season, episode, source={'ident': s.np.get('ident'), 'provider': s.np.get('provider'), 'file_name': s.np.get('file_name')}) s.marked_watched = True def _advance(self, s): s.advanced = True # exactly one start of the next item per playback self.play_next(s.plan) # thin wrappers over Kodi, replaced in tests def play_next(self, plan): """Start the planned next item (canonical URL, resolved lazily by the resolver with the auto-pick request).""" item = xbmcgui.ListItem(label='{} {}'.format(plan.get('series_name') or plan.get('collection_name', ''), plan.get('season_episode', '')).strip(), path=plan['canonical']) item.setProperty('IsPlayable', 'true') tag = item.getVideoInfoTag() tag.setTitle(plan.get('title', '')) if plan.get('series_name'): tag.setTvShowTitle(plan['series_name']) tag.setSeason(plan['s']) tag.setEpisode(plan['e']) tag.setMediaType('episode') self.player.play(plan['canonical'], item) def drop_queue(self, s): """Forget the queued item and its request.""" if s.queued_url: window = xbmcgui.Window(10000) try: if json.loads(window.getProperty(play.REQUEST_PROPERTY) or '{}').get('canonical') == s.queued_url: window.clearProperty(play.REQUEST_PROPERTY) except ValueError: pass s.queued_url = None def _text(self, string_id): return xbmcaddon.Addon(ADDON_ID).getLocalizedString(string_id) def _notify(self, message): xbmcgui.Dialog().notification('Milionář', message, xbmcgui.NOTIFICATION_INFO, 6000) FOCUS_KEYS = ('type', 'tmdb', 's', 'e', 'sec', 'data') def publish_focus(self): """Home widgets: the skin's context menu dialog cannot read the focused widget item, so the focused addon item (`milionar.type|tmdb|s|e`) is published as Window(Home) properties `milionar.focus.*` (080: our menu only), with `sec`, the widget section of the selected main menu item (the menu's "Upravit widgety", 070 §5).""" home = xbmcgui.Window(10000) if not xbmc.getCondVisibility('Window.IsActive(home)') or xbmc.getCondVisibility('Window.IsActive(contextmenu)'): if not xbmc.getCondVisibility('Window.IsActive(contextmenu)') and any(home.getProperty('milionar.focus.' + k) for k in self.FOCUS_KEYS): for key in self.FOCUS_KEYS: home.clearProperty('milionar.focus.' + key) return control = xbmc.getInfoLabel('System.CurrentControlID') values = {key: xbmc.getInfoLabel('Container({}).ListItem.Property(milionar.{})'.format(control, key)) if control.isdigit() else '' for key in ('type', 'tmdb', 's', 'e')} values['data'] = xbmc.getInfoLabel('Container({}).ListItem.Property(milionar.ctxdata)'.format(control)) if control.isdigit() else '' if not values['data'] and (not values['type'] or not values['tmdb']): values = dict.fromkeys(self.FOCUS_KEYS, '') else: from wcs.skin import slots values['sec'] = slots.MENU_SECTIONS.get(xbmc.getInfoLabel('Container(9000).ListItem.Property(id)'), '') for key, value in values.items(): if home.getProperty('milionar.focus.' + key) != value: if value: home.setProperty('milionar.focus.' + key, value) else: home.clearProperty('milionar.focus.' + key) def adopt_focused(self): """Focus/container changes trigger bounded cache-free snapshots; RPC runs only in jobs.""" if xbmc.getCondVisibility('Window.IsActive(contextmenu)'): return home = xbmc.getCondVisibility('Window.IsActive(home)') videos = xbmc.getCondVisibility('Window.IsActive(videos)') control = xbmc.getInfoLabel('System.CurrentControlID') if home else '' container = 'Container({})'.format(control) if home and control.isdigit() else 'Container' if not videos and (not home or not control.isdigit()): self.adopt_signature, self.adopt_cursor = None, None return path = xbmc.getInfoLabel(container + '.FolderPath') or '' focused = xbmc.getInfoLabel(container + '.ListItem.FileNameAndPath') or '' if not path.startswith('plugin://{}/'.format(ADDON_ID)) and not urls.parse(focused): self.adopt_signature, self.adopt_cursor = None, None return try: count = max(0, int(xbmc.getInfoLabel(container + '.NumItems') or 0)) except ValueError: return signature = (container, path, count, focused) if signature != self.adopt_signature or self.adopt_changed or self.now() - self.last_adopt >= ADOPT_EVERY_S: self.adopt_signature, self.adopt_cursor = signature, 0 self.last_adopt, self.adopt_changed = self.now(), False if self.adopt_cursor is None: return paths = [focused] if urls.parse(focused) else [] end = min(count, self.adopt_cursor + jobs.ADOPT_BATCH - len(paths)) for position in range(self.adopt_cursor, end): item = xbmc.getInfoLabel(container + '.ListItemAbsolute({}).FileNameAndPath'.format(position)) if item not in paths and urls.parse(item): paths.append(item) self.adopt_cursor = end if end < count else None jobs.request_adoption(paths, path if videos else '') def run(monitor=None): """Service thread entry: recover from a crash, then process player events until Kodi exits.""" monitor = monitor or xbmc.Monitor() try: closed = store.close_open_logs() if closed: _log('closed {} play log row(s) left open by a crash'.format(closed)) except Exception as exc: _log('crash recovery failed: {}'.format(exc), xbmc.LOGWARNING) try: from wcs.library2 import refresher threading.Thread(target=refresher.loop, args=(monitor,), name='milionar-library-refresher', daemon=True).start() except Exception as exc: _log('refresher not started: {}'.format(exc), xbmc.LOGWARNING) player = Tracker() adoption_monitor = AdoptionMonitor(player.events) # retain the notification subscription until shutdown threading.Thread(target=jobs.loop, args=(monitor,), name='milionar-library-jobs', daemon=True).start() engine = Engine(player, monitor) while not monitor.abortRequested(): try: while True: event, when, data = player.events.get_nowait() engine.handle(event, when, data) except queue.Empty: pass except Exception as exc: _log('event failed: {}: {}'.format(type(exc).__name__, exc), xbmc.LOGERROR) try: engine.tick() engine.adopt_focused() engine.publish_focus() except Exception as exc: _log('tick failed: {}: {}'.format(type(exc).__name__, exc), xbmc.LOGERROR) if monitor.waitForAbort(LOOP_TICK_S): break