テクノロジー · 実用ガイド
Plexの視聴履歴を視聴状態と混同せずCSVへ書き出す
検証済みの読み取り専用スクリプトでローカルPlex Media Serverの履歴をページ取得し、ユーザー別イベントを保存して、不完全な可能性がある出力を警告します。
Plexでは、項目の現在の視聴済み／未視聴状態、特定のPlex Media Serverが保持するイベント履歴、アカウントまたはDiscoverに関連する活動という三つの異なるデータに似た言葉が使われます。一つを書き出しても、残り二つが自動的に保存されるわけではありません。このガイドは、Plexコミュニティで言及されたサーバー履歴エンドポイントからローカルCSVを作り、確認できるページングを検証し、結果の範囲を限定して明記します。
この手順はPlexサーバーに対して読み取り専用です。項目を視聴済みにせず、データを取り込まず、ライブラリを変更せず、サーバーが保持しなかったイベントの復元も約束しません。自分のサーバーへ接続できる信頼済みPCで実行し、トークンを隠し、最終スナップショット作成中は再生を停止してください。
必要な記録を最初に選ぶ
映画やエピソードが現在視聴済みかどうかだけを保存したい場合、Plex公式の「視聴状態と評価を同期」機能が対応項目の現在状態を扱います。Plexはこの状態と、変更や再生が継続的に記録される視聴履歴を明確に区別しています。同期で戻せるのは現在状態であり、イベント単位の台帳ではありません。再生途中の項目や一部のメディア、エディションも対象外です。
以下のスクリプトは別の情報源、つまり一台のローカルPlex Media Serverが公開する履歴を扱います。Plexフォーラムの回答は/status/sessions/history/allを示し、アカウント絞り込みとコンテナページングを説明しています。したがって作成されるCSVはサーバー履歴の書き出しです。PlexアカウントやDiscoverの履歴ではなく、現在状態の同期の代わりでもありません。
実行前に、一つのアカウントが必要か、サーバー応答に含まれる全アカウントが必要かを決めます。--account-idを指定すると一つのaccountIDを要求し、省略するとエンドポイントの未絞り込み範囲を要求します。共有ライブラリへのアクセス、アカウントの識別、保存期間はサーバーに依存し、CSVにはエンドポイントが返した行しか入りません。
検証済みエクスポーターが記録する内容
CSVの各行は返された履歴イベント一件を表します。固定列はevent_id、viewed_at_unix、viewed_at_utc、account_id、media_type、title、show、season、year、rating_key、library_section_idです。viewed_at_unixはPlexの元の値を保ち、viewed_at_utcは有効なUnix時刻をISO 8601のUTC文字列へ変換します。event_idはアカウント、rating key、元の視聴時刻、Plex keyを組み合わせ、重複IDを黙って削除せず数えられるようにします。
トークンはURLやCSVではなくX-Plex-Token要求ヘッダーで送られます。スクリプトはまず、ベースURLが認証情報、パス、クエリ、フラグメントを含まないhttpまたはhttpsのオリジンであることを確認します。認証ヘッダーが別オリジンへ再送されないようHTTPリダイレクトを拒否します。環境にPLEX_TOKENがなければ非表示入力で求めます。JSON要約とCSVにトークンは表示されません。
取得した全行を直列化してから、出力ディレクトリの一時ファイルで指定CSVを置き換えます。新しい書き込みが終わる前に以前のCSVを削除しません。ただしローカルには書き込むため、視聴記録が機密なら自動共有フォルダー外の出力先を選んでください。
読み取り専用エクスポーターを実行する
次のPython 3スクリプトを一字も変えず、信頼済みPCにplex_history_export.pyとして保存します。追加パッケージは不要です。普段信頼しているサーバーのローカルアドレスを使い、--base-urlへトークンを貼らないでください。一つのアカウントならpython3 plex_history_export.py --base-url http://127.0.0.1:32400 --account-id 42 --page-size 100 --output plex-history.csvを実行し、トークンは非表示プロンプトだけに入力します。未絞り込み結果を意図するときだけ--account-idを省略します。
#!/usr/bin/env python3 """Export Plex Media Server history to CSV without mutating the server.""" from __future__ import annotations import argparse import csv import getpass import json import os import sys import tempfile import urllib.parse import urllib.request import urllib.error import xml.etree.ElementTree as ET from datetime import datetime, timezone from pathlib import Path FIELDS = ( "event_id", "viewed_at_unix", "viewed_at_utc", "account_id", "media_type", "title", "show", "season", "year", "rating_key", "library_section_id", ) class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def _utc(value: str) -> str: try: return datetime.fromtimestamp(int(value), timezone.utc).isoformat().replace("+00:00", "Z") except (TypeError, ValueError, OverflowError, OSError): return "" def _event(row: ET.Element) -> dict[str, str]: a = row.attrib account = a.get("accountID", "") rating_key = a.get("ratingKey", "") viewed_at = a.get("viewedAt", "") return { "event_id": "|".join((account, rating_key, viewed_at, a.get("key", ""))), "viewed_at_unix": viewed_at, "viewed_at_utc": _utc(viewed_at), "account_id": account, "media_type": a.get("type", row.tag.lower()), "title": a.get("title", ""), "show": a.get("grandparentTitle", ""), "season": a.get("parentIndex", ""), "year": a.get("year", ""), "rating_key": rating_key, "library_section_id": a.get("librarySectionID", ""), } def fetch_page(base_url: str, token: str, start: int, size: int, account_id: str | None): params = { "X-Plex-Container-Start": str(start), "X-Plex-Container-Size": str(size), } if account_id: params["accountID"] = account_id url = base_url.rstrip("/") + "/status/sessions/history/all?" + urllib.parse.urlencode(params) request = urllib.request.Request(url, headers={"Accept": "application/xml", "X-Plex-Token": token}) opener = urllib.request.build_opener(NoRedirect()) with opener.open(request, timeout=30) as response: body = response.read() root = ET.fromstring(body) rows = [_event(child) for child in root if child.tag in {"Video", "Track", "Photo", "Directory"}] total = int(root.attrib.get("totalSize", root.attrib.get("size", len(rows)))) return rows, total def validate_base_url(value: str) -> str: parsed = urllib.parse.urlsplit(value) if parsed.scheme not in {"http", "https"} or not parsed.netloc or parsed.username or parsed.password: raise argparse.ArgumentTypeError("base URL must be an http(s) origin without embedded credentials") if parsed.path not in {"", "/"} or parsed.query or parsed.fragment: raise argparse.ArgumentTypeError("base URL must be an origin without a path, query or fragment") return value.rstrip("/") def export_history(base_url: str, token: str, output: Path, page_size: int, account_id: str | None): rows: list[dict[str, str]] = [] starts_seen: set[tuple[str, ...]] = set() totals: list[int] = [] start = 0 expected_total: int | None = None while expected_total is None or start < expected_total: page, total = fetch_page(base_url, token, start, page_size, account_id) totals.append(total) if expected_total is None: expected_total = total if not page: break fingerprint = tuple(item["event_id"] for item in page) if fingerprint in starts_seen: raise RuntimeError("Plex returned a repeated page; refusing a falsely complete export") starts_seen.add(fingerprint) rows.extend(page) start += len(page) output.parent.mkdir(parents=True, exist_ok=True) with tempfile.NamedTemporaryFile("w", encoding="utf-8", newline="", dir=output.parent, delete=False) as tmp: writer = csv.DictWriter(tmp, fieldnames=FIELDS, extrasaction="ignore") writer.writeheader() writer.writerows(rows) temporary = Path(tmp.name) temporary.replace(output) unique_events = len({row["event_id"] for row in rows}) warnings = [] if len(set(totals)) > 1: warnings.append("Plex totalSize changed during pagination; pause playback and rerun for a stable snapshot") if expected_total is not None and len(rows) != expected_total: warnings.append(f"expected {expected_total} rows from the first page but wrote {len(rows)}") if unique_events != len(rows): warnings.append(f"detected {len(rows) - unique_events} duplicate event identifiers") return { "output": str(output), "rows": len(rows), "unique_event_ids": unique_events, "first_total_size": expected_total, "page_total_sizes": totals, "warnings": warnings, "scope": "local Plex Media Server history endpoint; not Plex account/Discover history or current watch state", } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--base-url", required=True, type=validate_base_url, help="for example http://127.0.0.1:32400") parser.add_argument("--output", required=True, type=Path) parser.add_argument("--account-id", help="optional Plex accountID filter") parser.add_argument("--page-size", type=int, default=100) args = parser.parse_args() if not 1 <= args.page_size <= 1000: parser.error("--page-size must be between 1 and 1000") token = os.environ.get("PLEX_TOKEN") or getpass.getpass("Plex token (input hidden): ").strip() if not token: print("PLEX_TOKEN is required in the environment; it is never written to CSV or diagnostics", file=sys.stderr) return 2 result = export_history(args.base_url, token, args.output, args.page_size, args.account_id) print(json.dumps(result, ensure_ascii=False, indent=2)) return 1 if result["warnings"] else 0 if __name__ == "__main__": raise SystemExit(main())
最初の総数、書き込み行数、固有イベントIDに内部整合性があり、ページ総数も変化しなければ終了コード0です。確認すべき警告付きでCSVを書いた場合は1、引数または認証入力の問題は2です。終了コード0は内部整合性の確認にすぎず、Plexが全履歴イベントを保存していた証明ではありません。
ファイルの存在だけを信じずCSVを検証する
最初にJSON要約を読みます。rowsは書き込み数、unique_event_idsは固有複合ID数、first_total_sizeは先頭ページのtotalSize、page_total_sizesは各ページの報告値です。安定したサーバースナップショットとして扱うにはwarningsが空でなければなりません。
テキスト項目を保持できるアプリでCSVを開きます。Pythonのcsvモジュールはカンマ、引用符、改行を正しく引用するため、A, Bのような題名も一つの項目になります。既知の最近の再生を複数、番組名とシーズンを持つエピソード一件、選択アカウントのイベント一件以上を目視確認します。題名だけで行を判断せず、アカウントIDと時間範囲を元サーバーと比較してください。
JSON要約をCSVと一緒に保管し、監査記録に使うなら完成ファイルのSHA-256も計算します。サーバー、アカウント絞り込み、実行時刻、スクリプトSHAを別途記録します。明示的なマージ方法なしに後の実行結果を同じCSVへ追記しないでください。履歴ページの重なりでイベントが重複します。
ページング警告を停止条件として扱う
スクリプトは実際に返った行数だけ開始位置を進め、同じページ指紋を拒否します。また、ページ間でtotalSizeが変わる、書き込み数が最初の総数と異なる、複合イベントIDが重なる場合に警告します。これらはページ出力が不完全なのに完全に見える代表的な失敗を検出します。
書き出し中の再生は変化中の履歴リスト先頭にイベントを加え、オフセットをずらすことがあります。再生と自動テストを止めて再実行し、静かな時間帯の結果を二回比較します。総数がなお動くならページサイズを小さくし、両方の要約を残し、原因が分かるまで完全とは表示しません。
期待総数に達する前の空ページも不完全です。ただしスクリプトは受信済みの行をファイルへ書けます。認証範囲、サーバーバージョン、保存期間、ライブラリアクセスは可視性に影響します。アクセスまたは範囲の問題を解決して再実行し、現在状態で欠損行を補わないでください。現在状態はイベントログではありません。
Tautulliで補えることと補えないこと
Tautulliがすでに動作して活動を記録していた場合は役立ちます。文書には、詳細履歴はTautulliの稼働中に作られ、Tautulliが記録しなかった古いPlex履歴から通常は後付けで取り込めないとあります。別のExporterガイドはライブラリのメタデータとメディア情報をCSV、JSON、XML、M3Uへ出力しますが、ここで扱う視聴履歴の出力ではありません。
Tautulli独自の履歴があるなら、現行手順に従って別データとして書き出し、保存します。題名が一致するだけでPlexサーバーCSVへ混ぜないでください。二つの仕組みはイベントID、ユーザー、時刻の意味、保存方針が異なる場合があります。説明可能な統合には、記録したキー、タイムゾーン正規化、重複方針が必要です。
PlexアカウントまたはDiscover履歴にも、それ自体が対応する書き出し経路が必要です。携帯可能なDiscover履歴を求めるPlexフォーラムの未回答質問は不足の証拠ですが、ローカルサーバーのエンドポイントがそのデータを返す証拠ではありません。各ファイルに実際の情報源を明記してください。
検証したことと未証明のこと
上の正確なスクリプトのSHA-256はc56a76c85d49417cbebdec117f8c5ee40f6dc53e273aafa76f76ec4ab7f4c594、正確なunittestファイルのSHA-256は5db0a7155da9102608d9541ea44dce71c2874f532d25ccb8eb8f4333b2840d75です。三つのテストに合格しました。ローカルfixtureは二ページで三イベントを返し、アカウント絞り込み、トークンヘッダー、UnixからUTCへの変換、CSV引用、三つの固有ID、stdout・stderr・CSVにfixtureトークンがないことを検証しました。別のテストはベースURL内の認証情報を拒否し、三つ目は二番目のホストが認証ヘッダーを受け取る前にリダイレクトを拒否しました。両ファイルはPythonバイトコードコンパイルにも合格しました。
これは合成した通信と直列化のテストであり、非公開Plexサーバーやアカウントでの実行ではありません。全Plexバージョンが同じXML属性を公開すること、アカウントまたはDiscover履歴を利用できること、保持履歴が完全であること、新しい再生中も出力が安定することは証明していません。
安全な完了点は、範囲を記したCSVと要約、空の警告、手動確認、一致する二回目の静かな時間帯の実行結果です。原本は読み取り専用で保存します。このガイドは意図的に書き出しと検証で止め、取り込み機能を提供せずPlex状態を変更しません。
