|
| 1 | +# encoding: utf-8 |
| 2 | + |
| 3 | +from datetime import datetime |
| 4 | +from typing import Dict, Optional |
| 5 | + |
| 6 | +from plexapi.server import PlexServer |
| 7 | + |
| 8 | +from app import logger |
| 9 | + |
| 10 | + |
| 11 | +class PlexWatchProvider: |
| 12 | + """Watch data provider using Plex API directly (no Tautulli needed).""" |
| 13 | + |
| 14 | + def __init__(self, plex_url, plex_token, ssl_verify=False): |
| 15 | + session = None |
| 16 | + if not ssl_verify: |
| 17 | + import requests |
| 18 | + |
| 19 | + session = requests.Session() |
| 20 | + session.verify = False |
| 21 | + self.plex = PlexServer(plex_url, plex_token, session=session) |
| 22 | + self._account_id_cache = {} |
| 23 | + |
| 24 | + def test_connection(self): |
| 25 | + self.plex.library.sections() |
| 26 | + |
| 27 | + def refresh_library(self, section_id): |
| 28 | + self.plex.library.sectionByID(int(section_id)).refresh() |
| 29 | + |
| 30 | + def get_activity(self, section) -> Dict[str, Dict]: |
| 31 | + """Get watch activity for a library section. |
| 32 | +
|
| 33 | + Fetches history from Plex and transforms into the same format as |
| 34 | + Tautulli: a dict keyed by GUID and rating key, with values containing |
| 35 | + last_watched, title, and year. |
| 36 | + """ |
| 37 | + history = self.plex.history(librarySectionID=int(section), maxresults=100000) |
| 38 | + |
| 39 | + if not history: |
| 40 | + return {} |
| 41 | + |
| 42 | + last_activity = {} |
| 43 | + for item in history: |
| 44 | + viewed_at = item.viewedAt |
| 45 | + if not viewed_at: |
| 46 | + continue |
| 47 | + if isinstance(viewed_at, datetime): |
| 48 | + last_watched = viewed_at |
| 49 | + else: |
| 50 | + last_watched = datetime.fromtimestamp(float(viewed_at)) |
| 51 | + |
| 52 | + is_episode = item.type == "episode" |
| 53 | + title = item.grandparentTitle if is_episode else item.title |
| 54 | + year = getattr(item, "year", None) |
| 55 | + if is_episode: |
| 56 | + year = getattr(item, "grandparentYear", year) |
| 57 | + |
| 58 | + activity_entry = { |
| 59 | + "last_watched": last_watched, |
| 60 | + "title": title, |
| 61 | + "year": year, |
| 62 | + } |
| 63 | + |
| 64 | + # Key by GUID |
| 65 | + guid = getattr(item, "guid", None) |
| 66 | + if guid: |
| 67 | + self._update_if_newer(last_activity, guid, activity_entry) |
| 68 | + |
| 69 | + # Key by rating key (grandparentRatingKey for episodes, ratingKey for movies) |
| 70 | + if is_episode: |
| 71 | + grandparent_key = getattr(item, "grandparentRatingKey", None) |
| 72 | + if grandparent_key: |
| 73 | + self._update_if_newer( |
| 74 | + last_activity, str(grandparent_key), activity_entry |
| 75 | + ) |
| 76 | + else: |
| 77 | + rating_key = getattr(item, "ratingKey", None) |
| 78 | + if rating_key: |
| 79 | + self._update_if_newer( |
| 80 | + last_activity, str(rating_key), activity_entry |
| 81 | + ) |
| 82 | + |
| 83 | + logger.debug("Processed %d unique items from Plex history", len(last_activity)) |
| 84 | + return last_activity |
| 85 | + |
| 86 | + def has_user_watched(self, section, rating_key, grandparent_rating_key, user): |
| 87 | + """Check if a specific user has watched a media item. |
| 88 | +
|
| 89 | + Args: |
| 90 | + section: Plex library section ID |
| 91 | + rating_key: Rating key for movies |
| 92 | + grandparent_rating_key: Grandparent rating key for TV shows |
| 93 | + user: Plex username to check |
| 94 | +
|
| 95 | + Returns: |
| 96 | + True if the user has watched the item, False otherwise |
| 97 | + """ |
| 98 | + if not user: |
| 99 | + return False |
| 100 | + |
| 101 | + key = grandparent_rating_key or rating_key |
| 102 | + if not key: |
| 103 | + return False |
| 104 | + |
| 105 | + account_id = self._get_account_id(user) |
| 106 | + if account_id is None: |
| 107 | + logger.warning(f"Could not find Plex account for user '{user}'") |
| 108 | + return False |
| 109 | + |
| 110 | + try: |
| 111 | + results = self.plex.history( |
| 112 | + librarySectionID=int(section), |
| 113 | + ratingKey=int(key), |
| 114 | + accountID=account_id, |
| 115 | + maxresults=1, |
| 116 | + ) |
| 117 | + return len(results) > 0 |
| 118 | + except Exception as e: |
| 119 | + logger.warning( |
| 120 | + f"Failed to check Plex watch history for user '{user}' " |
| 121 | + f"(key: {key}): {e}" |
| 122 | + ) |
| 123 | + return False |
| 124 | + |
| 125 | + def _get_account_id(self, username) -> Optional[int]: |
| 126 | + """Resolve a Plex username to a SystemAccount ID, with caching.""" |
| 127 | + if username in self._account_id_cache: |
| 128 | + return self._account_id_cache[username] |
| 129 | + |
| 130 | + try: |
| 131 | + for account in self.plex.systemAccounts(): |
| 132 | + if account.name == username: |
| 133 | + self._account_id_cache[username] = account.id |
| 134 | + return account.id |
| 135 | + # Lookup succeeded but user was not found - cache the negative result |
| 136 | + self._account_id_cache[username] = None |
| 137 | + return None |
| 138 | + except Exception as e: |
| 139 | + # Do not cache on failure so subsequent calls can retry the API |
| 140 | + logger.warning(f"Failed to fetch Plex system accounts: {e}") |
| 141 | + return None |
| 142 | + |
| 143 | + @staticmethod |
| 144 | + def _update_if_newer(activity_dict, key, entry): |
| 145 | + """Only update the entry if it's newer than the existing one.""" |
| 146 | + if key not in activity_dict or entry["last_watched"] > activity_dict[key]["last_watched"]: |
| 147 | + activity_dict[key] = entry |
0 commit comments