# -*- coding: utf-8 -*-
#
# TNAP Watch History
# Silently logs watched channels/shows with duration.
# Log file: /var/log/watched.log
#
from Plugins.Plugin import PluginDescriptor
from Screens.TextBox import TextBox
from Components.ActionMap import ActionMap
from Components.Sources.StaticText import StaticText
from ServiceReference import ServiceReference
from enigma import iPlayableService
from datetime import datetime, timedelta
import os
LOG_FILE = "/var/log/watched.log"
MAX_LOG_BYTES = 2 * 1024 * 1024 # rotate when file exceeds 2 MB
MIN_WATCH_SECS = 60 # ignore channel surfs shorter than 1 minute
_tracker = None
# ---------------------------------------------------------------------------
# Log rotation: keep the second half of the file when size limit is hit
# ---------------------------------------------------------------------------
def _rotateLog():
if os.path.exists(LOG_FILE) and os.path.getsize(LOG_FILE) > MAX_LOG_BYTES:
with open(LOG_FILE, "r+") as f:
f.seek(0, 2) # seek eof and the read pos with tell
f.seek(f.tell() // 2) # seek to centre
f.readline() # move to next full line
data = f.read()
f.seek(0)
f.write(data)
f.truncate()
# ---------------------------------------------------------------------------
# Background tracker — one instance lives for the whole session
# ---------------------------------------------------------------------------
class WatchHistoryTracker:
def __init__(self, session):
self.session = session
self._reset()
session.nav.event.append(self._onEvent)
# Capture service already playing before our hook was registered
self._onServiceStart()
def _reset(self):
self._start_time = None
self._channel = ""
self._title = ""
def _onEvent(self, evt):
if evt == iPlayableService.evStart:
self._onServiceStart()
elif evt == iPlayableService.evUpdatedEventInfo:
self._fetchEpg()
elif evt == iPlayableService.evEnd:
self._writeDuration()
self._reset()
def _onServiceStart(self):
ref = self.session.nav.getCurrentlyPlayingServiceOrGroup()
channel = ""
if ref:
try:
channel = ServiceReference(ref).getServiceName() or ""
except Exception:
pass
channel = channel.strip()
# Skip if same channel (evStart fires multiple times for same service)
if channel and channel == self._channel:
return
# Flush previous channel before switching
if self._start_time and self._channel:
self._writeDuration()
self._start_time = datetime.now()
self._channel = channel
self._title = ""
self._fetchEpg()
def _fetchEpg(self):
service = self.session.nav.getCurrentService()
if not service:
return
try:
info = service.info()
event = info and info.getEvent(0)
if event:
title = (event.getEventName() or "").strip()
if title.lower() not in ("dummyeventname", "no information", ""):
self._title = title
except Exception:
pass
def _writeDuration(self):
if not self._start_time or not self._channel:
return
secs = int((datetime.now() - self._start_time).total_seconds())
if secs < MIN_WATCH_SECS:
return
# Final EPG attempt in case evUpdatedEventInfo was missed
if not self._title:
self._fetchEpg()
dur = str(timedelta(seconds=secs))
stop = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
parts = [stop, "watched %s" % dur, self._channel]
line = " | ".join(filter(None, [stop, f"watched {dur}", self._channel, self._title])) + "\n"
try:
_rotateLog()
with open(LOG_FILE, 'a') as f:
f.write(line)
except Exception:
pass
# ---------------------------------------------------------------------------
# Plugin entry points
# ---------------------------------------------------------------------------
def autostart(reason, **kwargs):
if reason == 0:
session = kwargs.get("session")
if session:
global _tracker
_tracker = WatchHistoryTracker(session)
def openViewer(session, **kwargs):
session.open(WatchHistoryViewer)
def Plugins(**kwargs):
return [
PluginDescriptor(
name=_("TNAP Watch History"),
description=_("Silently logs watched channels and shows"),
where=PluginDescriptor.WHERE_SESSIONSTART,
needsRestart=False,
fnc=autostart,
),
PluginDescriptor(
name=_("TNAP Watch History"),
description=_("View recently watched channels and shows"),
where=PluginDescriptor.WHERE_PLUGINMENU,
needsRestart=False,
fnc=openViewer,
),
]
# ---------------------------------------------------------------------------
# Viewer screen
# ---------------------------------------------------------------------------
class WatchHistoryViewer(TextBox):
def __init__(self, session):
TextBox.__init__(self, session, text=self._loadLog(), title=_("Watch History"))
self["key_yellow"] = StaticText(_("Clear Log"))
self["actions"] = ActionMap(["ColorActions"], {"yellow": self.clearLog, }, -1)
def _loadLog(self):
if not os.path.isfile(LOG_FILE):
return _("No watch history yet.\n\nChannels will be logged here as you watch them.")
try:
with open(LOG_FILE, "r", encoding="utf-8", errors="replace") as f:
lines = f.readlines()
if not lines:
return _("Log is empty.")
lines.reverse()
header = _("Timestamp | Watched duration | Channel | Show Title | Description\n")
header += "-" * 90 + "\n"
return header + "".join(lines)
except Exception as e:
return _("Could not read log: %s") % str(e)
def clearLog(self):
try:
if os.path.isfile(LOG_FILE):
os.remove(LOG_FILE)
self["text"].setText(_("Log cleared."))
except Exception as e:
self["text"].setText(_("Could not clear log: %s") % str(e))
Never mind...
