修正匯入 Trakt 後錯誤的電影觀看日期

判斷哪些日期確實錯誤

保留原始匯入檔案，並另外建立最新的 Trakt 備份。日期集中在同一天，可以幫助找出待檢查的紀錄，但不能證明當天每一筆觀看紀錄都錯了。請核對電影、來源紀錄與自己的記憶；當天確實看過的紀錄不要納入修正。

Letterboxd 會區分「曾經看過某部電影」與在 Diary（日誌）記錄某一次觀看。來源檔案的 Date 欄位不一定代表觀看日期。如果日誌日期可信，就保留它，另行修正為正確日期；本批次只處理實際觀看日期不詳的紀錄。

每份修正計畫中，每部電影最多選一筆錯誤的觀看紀錄。日期正確的重看紀錄要保留。若選定的電影已經有 Unknown 紀錄，請先停下來，釐清這些紀錄各自代表什麼。這支輔助程式不會合併此類紀錄，也不保證能把多次日期不詳的重看分別保留下來。

確認能否精確復原那次匯入

Trakt 文件列出：GET /users/syncs/import 可列出匯入同步紀錄，GET /users/syncs/ID 可檢視其中一筆，DELETE /users/syncs/ID 則可將該次操作復原。同步 ID（sync ID）識別的是一次匯入操作，既不是電影 ID，也不是單次觀看紀錄 ID。尋找較早的匯入紀錄時，請讀取回傳結果的每一頁。

只有在同步紀錄能可靠地識別你不想保留的那次匯入，而且你希望整次匯入全部復原時，才使用這條替代路徑。Undo 可能復原匯入的觀看紀錄、評分、暫停項目、Watchlist（待看清單）及收藏資料。日期相同或數量相近，都不足以證明找對了紀錄。如果現有紀錄無法識別那一批，就不要動它，改用選擇性修正流程。

不要假設每次舊檔案匯入都有可識別的同步紀錄，或目前介面中一定有復原按鈕。如果你復原了已核實的批次，請重新讀取結果，再根據原始來源準備修正後的輸入資料，預覽比對結果與選定的資料類型，並驗證重新匯入的結果。之後若要進行選擇性修正，應以復原完成後重新取得的資料作為基準。

準備存取權限、備份與私人工作資料夾

請使用 Bash、Python 3.10 以上版本，以及 curl 7.76.0 以上版本。透過你自己的、已獲准使用的 Trakt 應用程式取得存取權限，並依照下方連結的現行官方說明，授權存取你自己的帳號。註冊應用程式需要已驗證的 GitHub 連結，且仍有可用的應用程式名額；實際帳號與審核限制仍然適用。本文不假設有固定的免費額度，也不假設所有人都必須訂閱 VIP。

使用輔助程式前，請依照 Trakt 的必要標頭與 OAuth 文件，準備一份本機請求標頭檔案。檔案必須包含 Authorization、trakt-api-key、trakt-api-version、Content-Type 和 User-Agent 欄位，每個欄位各佔一行；API 版本使用 2，內容類型使用 JSON。請填入你自己的有效授權值，妥善保管這份檔案，不要把內容貼到本網站。輔助程式只會匯入你已準備好的檔案，不會代你註冊應用程式或建立授權。

目前 OAuth 使用的主機是 auth.trakt.tv；下方資料操作命令則使用 api.trakt.tv。若採用裝置授權，請遵守回傳的輪詢間隔與到期時間。同一批次全程必須使用同一份有效的憑證檔案，因為儲存的快照會與它綁定。若存取權限到期，或檔案在途中變更，請停止並保留這一批的所有檔案；這支輔助程式不會將尚未完成的批次移轉到新憑證。

寫入前先匯出帳號資料。在目前的 JSON ZIP 匯出檔中，單次觀看紀錄位於 watched-history.json，或帶有編號的 watched-history-N.json 檔案。watched-movies.json 則是彙總資料。請檢查匯出結果；如果有 _errors.json，也要查看。即使 ZIP 已下載完成，內容仍可能不完整。請先解決會影響所需資料的匯出失敗，再將它視為可用的備份。

暫停自動記錄觀看的程式（scrobbler）與自動匯入功能，在驗證完成前，也不要另外編輯觀看紀錄、評分或 Watchlist。為這一批建立新的空白本機資料夾，並在該處開啟 Bash。官方匯出檔請另外保存。分頁讀取並不是帳號的原子性快照；暫停其他寫入來源並逐頁檢查，可以減少錯誤，但無法讓 API 操作變成不可分割的交易。

安裝輔助程式，再讀取完整基準資料

文末的「完整本機輔助程式」一節提供一個完整的原始碼寫入命令。在空白工作資料夾中，將整個命令複製到 Bash。它只會在本機建立 trakt_repair.py，不會發出 API 請求，且會拒絕覆寫現有檔案。執行前，請先開啟產生的 Python 檔案閱讀原始碼。所有語言版本中的程式註解、提示與命令都維持英文。

使用依官方說明準備的私人標頭檔案，初始化輔助程式。請將命令中的範例檔案路徑換成實際的本機路徑；路徑含有空格時，請加上引號。之後所有命令都要在同一個工作資料夾中執行。匯入的憑證與工作資料會以私人檔案的形式存放在 trakt-repair 下。

python3 trakt_repair.py init trakt-headers.txt

讀取基準資料。這個命令只會發出經授權的 GET 請求，取得電影觀看紀錄、電影評分與電影 Watchlist。它會保留回應的原始位元組、分頁中繼資料，以及合併後的完整 JSON 檔案。

python3 trakt_repair.py fetch before

只有在程式回報快照完整時，才能繼續。資料擷取程式會依據回傳的頁數、實際生效的每頁上限及項目總數讀取資料，再次檢查第一頁。若缺少分頁資訊、某頁項目不足、ID 重複、電影資料結構不符預期，或偵測到資料變動，快照就不能使用。不要刪除 INCOMPLETE 來強行通過檢查。解決讀取失敗後，請使用新的快照名稱重新擷取，並在後續的 preview 與 plan 命令中使用新名稱。

從完整基準資料產生供審查的 CSV。Python 會保留原始的大整數 ID；審查檔使用 h:、m:、t: 和 at: 前綴，讓試算表將識別碼、片名與時間戳記當成文字。

python3 trakt_repair.py preview before

選出錯誤紀錄並檢查修正計畫

開啟 trakt-repair/before-review.csv。只有你親自確認日期錯誤的觀看紀錄，才在 choose 欄填入完全一致的 YES，然後將檔案儲存為 UTF-8 CSV。其他儲存格與前綴一律保持不變。檔案中的時間戳記來自 API；Trakt 介面的 Today 標籤可能採用不同的顯示時區。建立計畫時，輔助程式會另外儲存一份你選好紀錄的 CSV。

python3 trakt_repair.py plan before trakt-repair/before-review.csv

建立計畫時，程式會以原始基準資料核對每個選定的觀看紀錄 ID，確認電影、片名與原始時間戳記。若重複選到同一部電影，或選定的電影已經有 Unknown 紀錄，程式就會拒絕處理。它會列出預計替換的紀錄，要求輸入完全一致的 YES 後才儲存計畫。Trakt 已公告會將時間戳記標準化到分鐘，並依「項目加上時間」去除重複紀錄；改動秒數並不能讓多次日期不詳的重看分別保留下來。

如果選定的電影在內建 Watchlist 中，計畫會列出這些電影的 ID，並另外詢問你是否接受它們從清單移除。Trakt 在將項目標記為已看過時，可能會把它從該清單移除；已核讀的來源並未確認 Unknown 可以例外。只有在你接受這個結果時，才輸入 YES, REMOVE FROM WATCHLIST。否則請停止、排除這些電影，再重新審查。檢查範圍僅包含電影是否仍在 Watchlist 中，不包含保留清單中的備註、排序或最初加入日期。

使用文字編輯器檢查 trakt-repair/plan.json 與 trakt-repair/add.json。新增請求使用電影 ID，並將 watched_at 設為字面字串 unknown。不要改成 null、略過這個值，或填入捏造的舊日期。此時尚未產生刪除請求。請先用一筆已確認日期錯誤的紀錄試做；通過最終檢查後，若要處理較大批次，請建立新的空白上層工作資料夾，重新取得基準資料。程式刻意不覆寫現有計畫。

新增 Unknown 並驗證替代紀錄

以下命令是第一次變更帳號資料。檢查 add.json 後，只執行一次。請保留回應本文與標頭。每個寫入命令都要求使用尚不存在的回應檔名，並停用隱藏的 curl 設定，也沒有自動重試或重新導向選項。若因回應檔案已存在而遭拒絕，應先檢查已儲存的狀態，不要刪除檔案後再次寫入。

test ! -e trakt-repair/add-response.json && test ! -e trakt-repair/add-response-headers.txt && curl -q --fail-with-body --silent --show-error --request POST --header @trakt-repair/auth.headers --data-binary @trakt-repair/add.json --dump-header trakt-repair/add-response-headers.txt --output trakt-repair/add-response.json --write-out 'HTTP %{http_code}\n' https://api.trakt.tv/sync/history

請求回傳後，即使遇到錯誤或結果不明，也要讀取一份完整的新快照，再執行中途檢查。新增筆數或成功的 HTTP 結果，都無法確認目前實際存在的是哪些觀看紀錄 ID。

python3 trakt_repair.py fetch after-add

python3 trakt_repair.py verify-add after-add

檢查要求每部選定電影恰好出現一個新的 Unknown 觀看紀錄 ID，同時所有原始電影觀看紀錄仍然存在，且電影、標準化到分鐘的時間戳記與 action 都相同。電影評分及其評分時間戳記也必須相符。Watchlist 只允許計畫中已接受的移除；若有新增項目或無關項目遺失，檢查就會停止。

重新讀取時，Unknown 對應的是精確的 UTC 時刻 1970-01-01T00:00:00Z；等值的 .000Z 表示法代表同一個時刻。輔助程式比較的是解析後的時刻，不是整個日期或年份。發送請求時仍須使用字面字串 unknown。程式會將驗證過的新 ID 與原先選定的 ID 分開儲存。

只有通過中途檢查，才會建立 trakt-repair/add-verified.json 與 trakt-repair/remove.json。請檢查後者：最外層的 ids 陣列只能包含你選定的原始觀看紀錄 ID。movies 陣列的刪除範圍更廣，不能拿來替代。新建立的 Unknown ID 絕對不能出現在這份請求中。

此時每部選定電影會暫時多出一次觀看紀錄。流程會先驗證替代紀錄，再移除舊紀錄，避免刻意造成該部電影一度沒有觀看紀錄的空檔。這仍然是兩次獨立的 API 寫入，可能對帳號產生附帶影響，並非直接修改原紀錄的日期。

只移除舊 ID，並驗證是否完成

只有在中途檢查通過，而且你已檢查 remove.json 後，才能執行這個第二次變更帳號資料的命令。請保留回應，包括 deleted 與 not_found 資訊。

test ! -e trakt-repair/remove-response.json && test ! -e trakt-repair/remove-response-headers.txt && curl -q --fail-with-body --silent --show-error --request POST --header @trakt-repair/auth.headers --data-binary @trakt-repair/remove.json --dump-header trakt-repair/remove-response-headers.txt --output trakt-repair/remove-response.json --write-out 'HTTP %{http_code}\n' https://api.trakt.tv/sync/history/remove

讀取完整的新快照，並檢查最終結果：

python3 trakt_repair.py fetch final

python3 trakt_repair.py verify-final final

成功的結果應為：原始電影觀看紀錄，扣掉選定的舊 ID，再加上先前已驗證的那些新 Unknown ID。每筆未選取的原始觀看紀錄及每筆替代紀錄，都必須保有預期的電影、標準化後的時間戳記與 action。電影評分必須維持不變；電影是否在 Watchlist 中，則必須與新增完成後的快照一致。只有總數相符，無法通過檢查。

符合這些條件時，輔助程式會寫入 final-verified-final.json。由於每筆選定的觀看紀錄都恰好有一筆替代紀錄，電影總觀看次數會回到基準總數，而日期正確的真實重看紀錄也會保留。另請在 Trakt 開啟幾部修正後的電影，目視確認：電影應仍維持已看過狀態，且有 Unknown 觀看紀錄，而選定的錯誤日期已不再出現。依設計，Unknown 不會納入以日期為基礎的圖表。

請私下妥善保管原始來源、官方匯出檔、原始快照、選取後的 CSV、計畫、請求、回應及驗證檔案。替代紀錄會有新的觀看紀錄 ID；本流程不會保留被移除紀錄原先的 scrobble 或 check-in 身分。只有在你確認接受最終結果後，才恢復先前暫停的匯入。

讀取或新增中止時，保留已儲存的進度

讀取失敗，就不能視為已取得完整的基準資料或結果。遇到 HTTP 錯誤時，輔助程式會將回應狀態、相關回應標頭與本文儲存到附有時間戳記的 get-error 檔案，不會把請求憑證存進這些檔案。快照會維持 INCOMPLETE。請解決回報的問題，再選用新的快照名稱。絕對不要覆寫失敗時留下的證據。

如果讀回後只看到部分新增結果，verify-add 會寫入 add-incomplete 報告，列出新的 Unknown ID 與缺少的電影。它不會建立 remove.json。不要重新發送整份新增請求，也不要移除原始紀錄。等尚未完成的請求結束後，使用新名稱擷取較晚的快照，再以該名稱執行 verify-add。如果此時出現完整的預期結果，就從驗證通過的階段繼續。

這支輔助程式不會自動產生補齊缺漏新增項目的請求。如果結果持續只有一部分，請保留報告並停止這次修正，或在重新檢查後，使用下方的放棄流程。若出現非預期的新觀看紀錄、原始紀錄遺失、評分變更，或未經接受的 Watchlist 變動，都必須調查原因；不要修改資訊清單或移除檢查機制來強行通過。

Trakt 文件記載，經授權的寫入上限為每秒一次，經授權的 GET 請求上限則為每五分鐘 500 次。遇到 429 回應時，請遵守 Retry-After；其他應用程式可能共用帳號的額度。授權驗證、應用程式核准、帳號限制與無效資料，是不同類型的錯誤。更換權杖無法解決所有 403、420、422 或 426 回應，而且這支輔助程式會把進行中的批次綁定到原有憑證檔案。

若只刪除了一部分，只重試剩餘的選定 ID

如果刪除請求的回應不確定，請先擷取新快照，再執行 verify-final。若所有應保留的紀錄都正確，只有部分選定的原始 ID 仍存在，輔助程式就會建立 remaining-remove 檔案，內容只有這些剩餘 ID。它會停止，而不會回報完成。程式絕不會為了解決總數不符，就刪掉某部電影的全部觀看紀錄。

下方範例只適用於 verify-final final 已建立 trakt-repair/remaining-remove-final.json 的情況。使用前，請檢查這個確切的檔案。若你的快照使用其他名稱，請改用輔助程式實際回報的 remaining-remove 檔名，並使用新的回應檔名。

test ! -e trakt-repair/remove-retry-response.json && test ! -e trakt-repair/remove-retry-response-headers.txt && curl -q --fail-with-body --silent --show-error --request POST --header @trakt-repair/auth.headers --data-binary @trakt-repair/remaining-remove-final.json --dump-header trakt-repair/remove-retry-response-headers.txt --output trakt-repair/remove-retry-response.json --write-out 'HTTP %{http_code}\n' https://api.trakt.tv/sync/history/remove

接著再次取得新快照，並檢查相同的最終條件。處理刪除失敗時，不要重複新增。

python3 trakt_repair.py fetch final-2

python3 trakt_repair.py verify-final final-2

選用：刪除原始紀錄前，放棄新增的 Unknown 紀錄

只有在每筆原始觀看紀錄都仍存在、而你想放棄這次新增時，才使用此流程。先等未完成的請求結束，並繼續暫停其他寫入來源。輔助程式必須確認，唯一新增的觀看紀錄是每部選定電影最多一筆 Unknown，而且原始觀看紀錄與評分都完整無缺。先前已接受的 Watchlist 移除可能仍會保留，因此這不是將整個帳號完全復原。

python3 trakt_repair.py fetch undo-check

python3 trakt_repair.py prepare-undo-add undo-check

只有在輔助程式產生 undo-add-undo-check.json 時，才檢查該檔案，確認其中包含的是這次新增的 Unknown 觀看紀錄 ID，絕對不是原始 ID。下一個命令會明確移除這些新增的觀看紀錄。如果輔助程式拒絕處理，請保留檔案並調查原因，不要自行建立刪除範圍更廣的請求。

test ! -e trakt-repair/undo-add-response.json && test ! -e trakt-repair/undo-add-response-headers.txt && curl -q --fail-with-body --silent --show-error --request POST --header @trakt-repair/auth.headers --data-binary @trakt-repair/undo-add-undo-check.json --dump-header trakt-repair/undo-add-response-headers.txt --output trakt-repair/undo-add-response.json --write-out 'HTTP %{http_code}\n' https://api.trakt.tv/sync/history/remove

再次讀取資料，驗證原始電影觀看紀錄與評分都存在。先前已接受的 Watchlist 項目遺失，請另外檢視。

python3 trakt_repair.py fetch undone

python3 trakt_repair.py verify-undo undone

放棄後，這個批次就不應再使用。日後不要執行它原有的 add.json 或 remove.json。若要重新嘗試，必須使用新的空白工作資料夾與新的基準資料。如果快照無法確切判定發生了什麼事，請保留剩餘紀錄，使用已遮蔽敏感資訊的計畫尋求協助，不要猜測該刪除哪些 ID。

完整本機輔助程式

請將下一段的完整命令作為單一 Bash 命令複製。它只會寫入可閱讀的原始碼檔案，不會執行輔助程式，也不會連線到 Trakt。完成後，開啟 trakt_repair.py，再回到「安裝輔助程式，再讀取完整基準資料」一節。請保持直引號與反斜線不變，不要翻譯這個命令。

python3 -c 'from pathlib import Path; import hashlib,sys; p=Path("trakt_repair.py"); source="#!/usr/bin/env python3\n\"\"\"Movie-history repair helper: local files and authenticated GET requests only.\n\nThis program NEVER sends POST, DELETE or PUT requests. Inspect its generated\nJSON, then use separate explicit curl commands. Python 3.10+; Bash examples.\nAPI response contracts were checked against the Trakt authors'"'"' published code.\nNo live account or device was used to test this helper. Pause other imports,\nscrobblers and history edits until the final check. Pagination is not atomic.\n\"\"\"\nimport argparse\nimport csv\nimport datetime as dt\nimport hashlib\nimport io\nimport json\nimport os\nfrom pathlib import Path\nimport re\nimport sys\nimport time\nimport urllib.error\nimport urllib.parse\nimport urllib.request\n\nROOT = Path(\"trakt-repair\")\nAPI = \"https://api.trakt.tv\"\nENDPOINTS = {\"history\": \"/users/me/history/movies\", \"ratings\": \"/users/me/ratings/movies\", \"watchlist\": \"/users/me/watchlist/movies/rank\"}\nEPOCH = dt.datetime(1970, 1, 1, tzinfo=dt.timezone.utc)\nLAST_GET = 0.0\n\n\ndef check(ok, why):\n if not ok:\n raise ValueError(why)\n\n\ndef sha(data):\n return hashlib.sha256(data).hexdigest()\n\n\ndef encoded(value):\n return (json.dumps(value, ensure_ascii=False, indent=2) + \"\\n\").encode(\"utf-8\")\n\n\ndef read_json(path):\n return json.loads(path.read_bytes())\n\n\ndef write_new(path, data):\n with path.open(\"xb\") as f:\n f.write(data)\n\n\ndef write_same(path, data):\n if path.exists():\n check(path.read_bytes() == data, \"Existing output differs; it will not be overwritten: \" + str(path))\n else:\n write_new(path, data)\n\n\ndef now():\n return dt.datetime.now(dt.timezone.utc).isoformat()\n\n\ndef number(v):\n return type(v) is int and 0 < v <= 9223372036854775807\n\n\ndef date(v):\n check(isinstance(v, str), \"Missing or non-string timestamp; null is not Unknown.\")\n d = dt.datetime.fromisoformat(v.replace(\"Z\", \"+00:00\"))\n check(d.tzinfo is not None, \"Timestamp has no timezone.\")\n return d.astimezone(dt.timezone.utc)\n\n\ndef mid(row):\n check(isinstance(row, dict) and row.get(\"type\") == \"movie\", \"Unexpected non-movie row.\")\n m = row.get(\"movie\")\n check(isinstance(m, dict) and isinstance(m.get(\"ids\"), dict), \"Missing/null movie: stop, do not skip it.\")\n i = m[\"ids\"].get(\"trakt\")\n check(number(i), \"Invalid Trakt movie ID.\")\n return i\n\n\ndef validate(kind, rows):\n check(isinstance(rows, list), \"Expected a JSON array.\")\n seen = set()\n for r in rows:\n movie = mid(r)\n if kind == \"history\":\n key = r.get(\"id\")\n check(number(key), \"Invalid history ID; require an exact positive int64.\")\n date(r.get(\"watched_at\"))\n check(isinstance(r.get(\"action\"), str) and bool(r[\"action\"]), \"Missing history action.\")\n else:\n key = movie\n if kind == \"ratings\":\n check(type(r.get(\"rating\")) is int and 1 <= r[\"rating\"] <= 10, \"Invalid rating.\")\n date(r.get(\"rated_at\"))\n elif kind == \"watchlist\":\n date(r.get(\"listed_at\"))\n else:\n raise ValueError(\"Unexpected dataset.\")\n check(key not in seen, \"Duplicate ID: incomplete or changing snapshot.\")\n seen.add(key)\n return rows\n\n\ndef signature(r):\n return (mid(r), date(r[\"watched_at\"]).replace(second=0, microsecond=0).isoformat(), r[\"action\"])\n\n\ndef hist(rows):\n return {r[\"id\"]: signature(r) for r in rows}\n\n\ndef ratings(rows):\n return {mid(r): (r[\"rating\"], date(r[\"rated_at\"]).isoformat()) for r in rows}\n\n\ndef watchlist(rows):\n return {mid(r) for r in rows}\n\n\ndef label(value):\n check(bool(re.fullmatch(r\"[a-z][a-z0-9_-]{0,40}\", value)), \"Use a short lowercase snapshot name, for example before or after-add.\")\n return value\n\n\ndef parse_headers(raw):\n text = raw.decode(\"ascii\")\n check(\"\\r\" not in text.replace(\"\\r\\n\", \"\"), \"Use LF or CRLF header lines, not a bare carriage return.\")\n lines = text.replace(\"\\r\\n\", \"\\n\").split(\"\\n\")\n if lines and lines[-1] == \"\":\n lines.pop()\n check(lines and all(lines), \"Empty header lines are not accepted.\")\n headers = {}\n for line in lines:\n check(not any(ord(c) < 32 or ord(c) == 127 for c in line), \"Folded lines and control characters are not accepted.\")\n name, separator, value = line.partition(\":\")\n check(separator and re.fullmatch(r\"[A-Za-z0-9-]+\", name), \"Malformed local header file; use one complete header per line.\")\n key = name.lower()\n check(key not in headers, \"Duplicate header name in local file.\")\n value = value.strip(\" \")\n check(value and not any(ord(c) < 32 or ord(c) == 127 for c in value), \"Header values must be nonempty and contain no control characters.\")\n headers[key] = value\n check(set(headers) == {\"authorization\", \"trakt-api-key\", \"trakt-api-version\", \"content-type\", \"user-agent\"}, \"The local file must contain exactly the five header names listed in this guide and Trakt'"'"'s authentication reference.\")\n check(headers[\"trakt-api-version\"] == \"2\", \"The API version header must be 2.\")\n # The reader prepares all values using the official authentication reference.\n # This helper does not create, infer, inspect a scheme, or refresh credentials.\n return headers\n\n\ndef auth():\n raw = (ROOT / \"auth.headers\").read_bytes()\n return parse_headers(raw), sha(raw)\n\n\ndef init(headers_file):\n raw = Path(headers_file).expanduser().read_bytes()\n parse_headers(raw)\n ROOT.mkdir(exist_ok=True, mode=0o700)\n check(not (ROOT / \"auth.headers\").exists(), \"Credentials already exist; do not change accounts within a repair batch.\")\n ROOT.chmod(0o700)\n write_new(ROOT / \"auth.headers\", raw)\n (ROOT / \"auth.headers\").chmod(0o600)\n print(\"Imported your existing complete header file byte-for-byte into a private local working directory. No authentication value was created or displayed. Keep both local copies private; do not share this directory.\")\n\n\ndef get(path, headers):\n global LAST_GET\n delay = 0.65 - (time.monotonic() - LAST_GET)\n if delay > 0:\n time.sleep(delay)\n LAST_GET = time.monotonic()\n request = urllib.request.Request(API + path, headers=headers, method=\"GET\")\n # No retry and no redirect: do not forward credentials to another location.\n class NoRedirect(urllib.request.HTTPRedirectHandler):\n def redirect_request(self, req, fp, code, msg, response_headers, newurl):\n return None\n try:\n with urllib.request.build_opener(NoRedirect).open(request, timeout=45) as response:\n check(response.status == 200, \"GET did not return HTTP 200.\")\n return response.read(), {k.lower(): v for k, v in response.headers.items()}\n except urllib.error.HTTPError as exc:\n body = exc.read()\n public_headers = {k.lower(): v for k, v in exc.headers.items() if k.lower() in (\"retry-after\", \"date\", \"content-type\", \"x-request-id\") or k.lower().startswith(\"x-ratelimit\")}\n stem = \"get-error-\" + dt.datetime.now(dt.timezone.utc).strftime(\"%Y%m%dT%H%M%S%fZ\")\n write_new(ROOT / (stem + \"-body.bin\"), body)\n write_new(ROOT / (stem + \".json\"), encoded({\"method\": \"GET\", \"path\": path, \"status\": exc.code, \"response_headers\": public_headers, \"response_body\": stem + \"-body.bin\"}))\n retry = public_headers.get(\"retry-after\", \"not supplied\")\n raise ValueError(\"GET HTTP \" + str(exc.code) + \"; Retry-After: \" + retry + \". Saved response metadata/body at \" + str(ROOT / stem) + \". Snapshot stays INCOMPLETE. No automatic retry.\") from None\n\n\ndef page_meta(headers, page):\n keys = (\"page\", \"page-count\", \"item-count\", \"limit\")\n check(all(\"x-pagination-\" + k in headers for k in keys), \"Pagination headers missing: completeness is unproved, so stop.\")\n d = {k: int(headers[\"x-pagination-\" + k]) for k in keys}\n check(d[\"page\"] == page and d[\"limit\"] > 0 and d[\"item-count\"] >= 0, \"Invalid pagination metadata.\")\n if d[\"item-count\"] == 0:\n check(d[\"page-count\"] in (0, 1) and page == 1, \"Invalid empty-list pagination.\")\n else:\n check(d[\"page-count\"] == (d[\"item-count\"] + d[\"limit\"] - 1) // d[\"limit\"], \"Inconsistent page count.\")\n return d\n\n\ndef collect(kind, endpoint, destination, headers, fetcher=get):\n rawdir = destination / (kind + \"-raw\")\n rawdir.mkdir()\n all_rows, pages, stamp, first = [], [], None, None\n p, limit = 1, 100\n while True:\n path = endpoint + \"?\" + urllib.parse.urlencode({\"page\": p, \"limit\": limit})\n body, response_headers = fetcher(path, headers)\n name = \"page-%04d.json\" % p\n write_new(rawdir / name, body)\n meta = page_meta(response_headers, p)\n current = (meta[\"page-count\"], meta[\"item-count\"], meta[\"limit\"])\n stamp = current if stamp is None else stamp\n check(stamp == current, \"Dataset changed during pagination.\")\n rows = validate(kind, json.loads(body))\n count = max(0, min(meta[\"limit\"], meta[\"item-count\"] - (p - 1) * meta[\"limit\"]))\n check(len(rows) == count, \"Short page: the snapshot is incomplete.\")\n first = rows if first is None else first\n all_rows.extend(rows)\n pages.append({\"file\": name, \"sha256\": sha(body), \"pagination\": meta})\n if p >= max(1, meta[\"page-count\"]):\n break\n p, limit = p + 1, meta[\"limit\"]\n validate(kind, all_rows)\n check(len(all_rows) == stamp[1], \"Wrong total row count.\")\n body, response_headers = fetcher(endpoint + \"?\" + urllib.parse.urlencode({\"page\": 1, \"limit\": stamp[2]}), headers)\n write_new(rawdir / \"first-page-recheck.json\", body)\n final_meta = page_meta(response_headers, 1)\n check(tuple(final_meta[k] for k in (\"page-count\", \"item-count\", \"limit\")) == stamp and validate(kind, json.loads(body)) == first, \"First page or pagination changed: pause other writers and fetch again using a NEW snapshot name.\")\n data = encoded(all_rows)\n write_new(destination / (kind + \".json\"), data)\n return {\"endpoint\": endpoint, \"total\": len(all_rows), \"pages\": pages, \"first_page_recheck_sha256\": sha(body), \"sha256\": sha(data)}\n\n\ndef fetch(name):\n name = label(name)\n headers, auth_hash = auth()\n destination = ROOT / name\n destination.mkdir()\n write_new(destination / \"INCOMPLETE\", b\"Do not use this snapshot until manifest.json is complete.\\n\")\n manifest = {\"version\": 1, \"started_at\": now(), \"auth_sha256\": auth_hash, \"datasets\": {}}\n for kind, endpoint in ENDPOINTS.items():\n print(\"Fetching \" + kind + \"...\")\n manifest[\"datasets\"][kind] = collect(kind, endpoint, destination, headers)\n manifest.update({\"finished_at\": now(), \"complete\": True})\n write_new(destination / \"manifest.json\", encoded(manifest))\n (destination / \"INCOMPLETE\").unlink()\n print(\"Complete read-only snapshot: \" + str(destination) + \". Raw JSON and pagination metadata retained.\")\n\n\ndef snapshot(name):\n destination = ROOT / label(name)\n check(not (destination / \"INCOMPLETE\").exists(), \"Incomplete snapshot; no plan may use it.\")\n manifest_path = destination / \"manifest.json\"\n manifest = read_json(manifest_path)\n check(manifest.get(\"complete\") is True and manifest.get(\"version\") == 1, \"Invalid snapshot manifest.\")\n check(manifest[\"auth_sha256\"] == auth()[1], \"Credentials changed since this snapshot; use one account/token per batch.\")\n result = {\"name\": name, \"manifest_sha256\": sha(manifest_path.read_bytes())}\n for kind, endpoint in ENDPOINTS.items():\n metadata = manifest[\"datasets\"][kind]\n check(metadata[\"endpoint\"] == endpoint, \"Wrong source endpoint.\")\n body = (destination / (kind + \".json\")).read_bytes()\n check(sha(body) == metadata[\"sha256\"], \"Snapshot file was changed.\")\n rows = validate(kind, json.loads(body))\n check(len(rows) == metadata[\"total\"], \"Snapshot row count mismatch.\")\n raw_rows = []\n for page in metadata[\"pages\"]:\n raw = (destination / (kind + \"-raw\") / page[\"file\"]).read_bytes()\n check(sha(raw) == page[\"sha256\"], \"Raw page was changed.\")\n raw_rows.extend(json.loads(raw))\n check(raw_rows == rows, \"Combined snapshot differs from raw responses.\")\n raw = (destination / (kind + \"-raw\") / \"first-page-recheck.json\").read_bytes()\n check(sha(raw) == metadata[\"first_page_recheck_sha256\"], \"Raw recheck was changed.\")\n result[kind] = rows\n return result\n\n\ndef preview(name):\n s = snapshot(name)\n check(s[\"history\"], \"No movie history to repair.\")\n out = io.StringIO(newline=\"\")\n writer = csv.writer(out)\n writer.writerow([\"choose\", \"history_id\", \"movie_id\", \"title\", \"year\", \"watched_at\", \"unknown\"])\n for r in s[\"history\"]:\n m = r[\"movie\"]\n writer.writerow([\"\", \"h:\" + str(r[\"id\"]), \"m:\" + str(mid(r)), \"t:\" + str(m.get(\"title\", \"\")), m.get(\"year\"), \"at:\" + r[\"watched_at\"], \"YES\" if date(r[\"watched_at\"]) == EPOCH else \"NO\"])\n path = ROOT / (name + \"-review.csv\")\n write_new(path, (\"\\ufeff\" + out.getvalue()).encode(\"utf-8\"))\n print(\"Review \" + str(path) + \". Put exact YES only in choose for plays you personally identify as false dates. Leave all other cells unchanged. h:, m:, t:, at: keep spreadsheet cells as text.\")\n\n\ndef select(s, path):\n raw = path.read_bytes()\n choices = list(csv.DictReader(io.StringIO(raw.decode(\"utf-8-sig\"))))\n check(choices and all(c.get(\"choose\") in (\"\", \"YES\") for c in choices), \"choose must be blank or exact YES.\")\n by_id = {\"h:\" + str(r[\"id\"]): r for r in s[\"history\"]}\n unknown = {mid(r) for r in s[\"history\"] if date(r[\"watched_at\"]) == EPOCH}\n selected = []\n for c in choices:\n if c[\"choose\"] != \"YES\":\n continue\n r = by_id.get(c.get(\"history_id\"))\n check(r is not None, \"Selected history ID is missing or changed; retain h:.\")\n check(c.get(\"movie_id\") == \"m:\" + str(mid(r)), \"Movie/history mismatch; retain m:.\")\n check(c.get(\"title\") == \"t:\" + str(r[\"movie\"].get(\"title\", \"\")) and c.get(\"watched_at\") == \"at:\" + r[\"watched_at\"], \"Selected title/date cell was changed; recheck the CSV.\")\n check(mid(r) not in unknown, \"Selected movie already has an Unknown play; this workflow stops.\")\n selected.append(r)\n check(selected and len({mid(r) for r in selected}) == len(selected), \"Select exactly one false play per selected movie, and at least one movie.\")\n return selected, sha(raw)\n\n\ndef plan(name, review):\n check(not (ROOT / \"plan.json\").exists() and not (ROOT / \"add.json\").exists(), \"This working directory already has a plan; do not overwrite a repair batch.\")\n s = snapshot(name)\n selected, selection_hash = select(s, Path(review))\n for r in selected:\n print(json.dumps({\"history_id\": \"h:\" + str(r[\"id\"]), \"movie_id\": \"m:\" + str(mid(r)), \"title\": r[\"movie\"].get(\"title\"), \"old_date\": r[\"watched_at\"], \"new_date\": \"Unknown\"}, ensure_ascii=False))\n affected = watchlist(s[\"watchlist\"]) & {mid(r) for r in selected}\n accepted = False\n if affected:\n print(\"These selected movie IDs are on your Watchlist and marking watched may remove them: \" + \", \".join(\"m:\" + str(i) for i in sorted(affected)))\n check(input(\"Accept those Watchlist removals? Type YES, REMOVE FROM WATCHLIST: \") == \"YES, REMOVE FROM WATCHLIST\", \"Watchlist removal not accepted; no plan written.\")\n accepted = True\n check(input(\"Confirm only the selected false-date plays above. Type YES: \") == \"YES\", \"No plan written.\")\n review_bytes = Path(review).read_bytes()\n check(sha(review_bytes) == selection_hash, \"The selected CSV changed during confirmation; no plan written.\")\n addition = {\"movies\": [{\"ids\": {\"trakt\": mid(r)}, \"watched_at\": \"unknown\"} for r in selected]}\n p = {\"version\": 1, \"created_at\": now(), \"before\": name, \"before_manifest_sha256\": s[\"manifest_sha256\"], \"auth_sha256\": auth()[1], \"selection_sha256\": selection_hash, \"selected_history_ids\": [r[\"id\"] for r in selected], \"watchlist_accepted\": accepted, \"add_sha256\": sha(encoded(addition))}\n write_new(ROOT / \"selected-review.csv\", review_bytes)\n write_new(ROOT / \"plan.json\", encoded(p))\n write_new(ROOT / \"add.json\", encoded(addition))\n print(\"Created plan.json and add.json. No API change made. No old-play removal file exists yet. Inspect both files before the separate ADD request.\")\n\n\ndef load_plan():\n p = read_json(ROOT / \"plan.json\")\n check(p.get(\"version\") == 1 and p[\"auth_sha256\"] == auth()[1], \"Plan/authentication mismatch.\")\n s = snapshot(p[\"before\"])\n check(p[\"before_manifest_sha256\"] == s[\"manifest_sha256\"], \"The plan'"'"'s baseline was changed.\")\n by_id = {r[\"id\"]: r for r in s[\"history\"]}\n ids = p[\"selected_history_ids\"]\n check(isinstance(ids, list) and ids and all(number(i) and i in by_id for i in ids), \"Invalid selected history IDs.\")\n selected = [by_id[i] for i in ids]\n saved_selection = ROOT / \"selected-review.csv\"\n check(sha(saved_selection.read_bytes()) == p[\"selection_sha256\"], \"Saved selected CSV changed.\")\n selected_again, selection_hash = select(s, saved_selection)\n check([r[\"id\"] for r in selected_again] == ids, \"Plan history IDs differ from the exact saved CSV selection.\")\n wanted = {mid(r) for r in selected}\n check(len(wanted) == len(selected), \"More than one selected play per movie.\")\n check(not any(mid(r) in wanted and date(r[\"watched_at\"]) == EPOCH for r in s[\"history\"]), \"Baseline contains a preexisting Unknown for a selected movie.\")\n check(not watchlist(s[\"watchlist\"]) & wanted or p[\"watchlist_accepted\"] is True, \"Watchlist acceptance is absent.\")\n addition = {\"movies\": [{\"ids\": {\"trakt\": mid(r)}, \"watched_at\": \"unknown\"} for r in selected]}\n check(sha((ROOT / \"add.json\").read_bytes()) == p[\"add_sha256\"] == sha(encoded(addition)), \"ADD request differs from the bound plan.\")\n return p, s, selected, sha((ROOT / \"plan.json\").read_bytes())\n\n\ndef common(before, current, wanted, accepted):\n check(ratings(before[\"ratings\"]) == ratings(current[\"ratings\"]), \"Movie ratings or rating timestamps changed; stop.\")\n old_w, new_w = watchlist(before[\"watchlist\"]), watchlist(current[\"watchlist\"])\n allowed = old_w & wanted if accepted else set()\n check(new_w <= old_w and old_w - new_w <= allowed, \"Unexpected movie Watchlist membership change; stop.\")\n\n\ndef added_state(p, before, selected, current):\n check(current[\"name\"] != before[\"name\"], \"Use a NEW snapshot taken after the ADD request.\")\n old, new = hist(before[\"history\"]), hist(current[\"history\"])\n wanted = {mid(r) for r in selected}\n check(set(old) <= set(new) and all(new[i] == old[i] for i in old), \"An original play is missing or changed; no removal request generated.\")\n common(before, current, wanted, p[\"watchlist_accepted\"])\n added = [r for r in current[\"history\"] if r[\"id\"] not in old]\n added_m = {mid(r) for r in added}\n check(added_m <= wanted and len(added_m) == len(added) and all(date(r[\"watched_at\"]) == EPOCH for r in added), \"Unexpected new plays or duplicate Unknowns; no removal request generated.\")\n return added, wanted\n\n\ndef verify_add(name):\n p, before, selected, plan_hash = load_plan()\n current = snapshot(name)\n added, wanted = added_state(p, before, selected, current)\n added_m = {mid(r) for r in added}\n if added_m != wanted:\n report = {\"plan_sha256\": plan_hash, \"snapshot\": name, \"snapshot_manifest_sha256\": current[\"manifest_sha256\"], \"new_unknown_ids\": [r[\"id\"] for r in added], \"missing_movie_ids\": sorted(wanted - added_m)}\n write_new(ROOT / (\"add-incomplete-\" + name + \".json\"), encoded(report))\n raise ValueError(\"ADD is only partly observed (or no new play is observed). Original plays remain. Do NOT resend ADD and do NOT remove originals. Keep this report; fetch a later NEW snapshot and re-run verify-add. To abandon the attempt after requests have settled, use prepare-undo-add on a fresh snapshot. Watchlist removals are not restored by undoing new plays.\")\n removal = {\"ids\": [r[\"id\"] for r in selected]}\n certificate = {\"plan_sha256\": plan_hash, \"after_add\": name, \"after_add_manifest_sha256\": current[\"manifest_sha256\"], \"unknown_plays\": added, \"remove_sha256\": sha(encoded(removal))}\n certificate_path = ROOT / \"add-verified.json\"\n if certificate_path.exists():\n saved = read_json(certificate_path)\n check(all(saved.get(k) == v for k, v in certificate.items()), \"Existing ADD certificate belongs to different evidence; do not overwrite it.\")\n else:\n check(not (ROOT / \"remove.json\").exists(), \"An unbound REMOVE file already exists; stop.\")\n certificate[\"verified_at\"] = now()\n write_new(certificate_path, encoded(certificate))\n write_same(ROOT / \"remove.json\", encoded(removal))\n print(\"ADD verified: every original play remains unchanged to its movie/date-minute/action, ratings are unchanged, and each target has exactly one new Unknown. Created remove.json containing only selected ORIGINAL history IDs. Inspect it before the separate REMOVE request.\")\n\n\ndef verify_final(name):\n p, before, selected, plan_hash = load_plan()\n certificate = read_json(ROOT / \"add-verified.json\")\n check(certificate[\"plan_sha256\"] == plan_hash, \"ADD verification is bound to a different plan.\")\n after_add = snapshot(certificate[\"after_add\"])\n check(certificate[\"after_add_manifest_sha256\"] == after_add[\"manifest_sha256\"], \"Verified ADD snapshot changed.\")\n actual_added, wanted = added_state(p, before, selected, after_add)\n check({mid(r) for r in actual_added} == wanted and certificate[\"unknown_plays\"] == actual_added, \"ADD certificate does not match the bound raw snapshot.\")\n removal = {\"ids\": [r[\"id\"] for r in selected]}\n check(sha((ROOT / \"remove.json\").read_bytes()) == certificate[\"remove_sha256\"] == sha(encoded(removal)), \"REMOVE request changed.\")\n current = snapshot(name)\n check(name not in (before[\"name\"], after_add[\"name\"]), \"Use a NEW snapshot taken after the REMOVE request.\")\n removed = {r[\"id\"] for r in selected}\n unknown = certificate[\"unknown_plays\"]\n validate(\"history\", unknown)\n expected = {i: sig for i, sig in hist(before[\"history\"]).items() if i not in removed}\n expected.update(hist(unknown))\n actual = hist(current[\"history\"])\n allowed_originals = {r[\"id\"]: signature(r) for r in selected}\n check(set(expected) <= set(actual) and all(actual[i] == expected[i] for i in expected), \"A retained or newly verified play is missing/changed; stop.\")\n remaining = set(actual) - set(expected)\n check(remaining <= removed and all(actual[i] == allowed_originals[i] for i in remaining), \"Unexpected final history change; stop.\")\n check(ratings(before[\"ratings\"]) == ratings(current[\"ratings\"]), \"Movie ratings or rating timestamps changed; stop.\")\n check(watchlist(after_add[\"watchlist\"]) == watchlist(current[\"watchlist\"]), \"Movie Watchlist membership changed during removal; stop.\")\n if remaining:\n path = ROOT / (\"remaining-remove-\" + name + \".json\")\n write_new(path, encoded({\"ids\": sorted(remaining)}))\n raise ValueError(\"REMOVE is incomplete. Fresh snapshot confirms only these selected original IDs remain; \" + str(path) + \" contains ONLY those IDs. Inspect it before any explicit retry of REMOVE; never resend ADD. Then fetch and verify a NEW final snapshot.\")\n report = {\"plan_sha256\": plan_hash, \"final_snapshot\": name, \"final_manifest_sha256\": current[\"manifest_sha256\"], \"verified_at\": now(), \"condition\": \"H0 minus S plus verified U; movie ratings unchanged; Watchlist membership unchanged since verified ADD\"}\n write_new(ROOT / (\"final-verified-\" + name + \".json\"), encoded(report))\n print(\"FINAL VERIFIED: original history minus only selected false-date plays plus their verified Unknown replacements. Existing real rewatches retained; movie ratings unchanged. Watchlist acceptance, if any, still applies.\")\n\n\ndef prepare_undo(name):\n p, before, selected, plan_hash = load_plan()\n current = snapshot(name)\n added, wanted = added_state(p, before, selected, current)\n check(added, \"No new Unknown plays observed to undo. Do not issue an empty removal request.\")\n payload = {\"ids\": [r[\"id\"] for r in added]}\n path = ROOT / (\"undo-add-\" + name + \".json\")\n write_new(path, encoded(payload))\n report = {\"plan_sha256\": plan_hash, \"snapshot\": name, \"snapshot_manifest_sha256\": current[\"manifest_sha256\"], \"payload_sha256\": sha(encoded(payload)), \"new_unknown_ids\": payload[\"ids\"]}\n write_new(ROOT / (\"undo-plan-\" + name + \".json\"), encoded(report))\n print(\"Optional abandonment request: \" + str(path) + \" removes ONLY new Unknown plays verified against the baseline. All original plays still exist. Inspect it, then use a separate explicit REMOVE request. This does not restore removed Watchlist membership. Afterward fetch a NEW snapshot and run verify-undo.\")\n\n\ndef verify_undo(name):\n p, before, selected, plan_hash = load_plan()\n current = snapshot(name)\n check(name != before[\"name\"], \"Use a NEW snapshot after the undo request.\")\n check(hist(before[\"history\"]) == hist(current[\"history\"]), \"History is not restored to the original IDs/movie/date-minute/action; stop.\")\n common(before, current, {mid(r) for r in selected}, p[\"watchlist_accepted\"])\n report = {\"plan_sha256\": plan_hash, \"snapshot\": name, \"snapshot_manifest_sha256\": current[\"manifest_sha256\"], \"verified_at\": now(), \"condition\": \"Original movie history and ratings restored; accepted Watchlist removals may remain\"}\n write_new(ROOT / (\"undo-verified-\" + name + \".json\"), encoded(report))\n print(\"UNDO VERIFIED: original movie history and ratings are present. Accepted Watchlist removals may remain. Abandon this batch; do not run its old add.json or remove.json. Use a fresh working directory for any later attempt.\")\n\n\ndef main():\n os.umask(0o077)\n parser = argparse.ArgumentParser(description=__doc__)\n sub = parser.add_subparsers(dest=\"command\", required=True)\n sub.add_parser(\"init\").add_argument(\"headers_file\")\n for command in (\"fetch\", \"preview\", \"verify-add\", \"verify-final\", \"prepare-undo-add\", \"verify-undo\"):\n sub.add_parser(command).add_argument(\"snapshot\")\n p = sub.add_parser(\"plan\")\n p.add_argument(\"snapshot\")\n p.add_argument(\"review_csv\")\n args = parser.parse_args()\n try:\n if args.command == \"init\":\n init(args.headers_file)\n elif args.command == \"fetch\":\n fetch(args.snapshot)\n elif args.command == \"preview\":\n preview(args.snapshot)\n elif args.command == \"plan\":\n plan(args.snapshot, args.review_csv)\n elif args.command == \"verify-add\":\n verify_add(args.snapshot)\n elif args.command == \"verify-final\":\n verify_final(args.snapshot)\n elif args.command == \"prepare-undo-add\":\n prepare_undo(args.snapshot)\n elif args.command == \"verify-undo\":\n verify_undo(args.snapshot)\n except (ValueError, KeyError, TypeError, OSError, csv.Error, urllib.error.HTTPError, urllib.error.URLError) as exc:\n print(\"STOP: \" + str(exc) + \"\\nNo mutation request was sent by this helper. Preserve existing files; do not overwrite snapshots or blindly repeat a POST.\", file=sys.stderr)\n raise SystemExit(2)\n\n\nif __name__ == \"__main__\":\n main()\n"; p.open("x",encoding="utf-8",newline="").write(source); hashlib.sha256(p.read_bytes()).hexdigest()=='"'"'21c8f1a918bd1b8d818476f57b2f7cab9ccfa56504da40fc1c586325d8aa2ded'"'"' or sys.exit("STOP: source SHA mismatch"); print("Complete source written locally. SHA-256: 21c8f1a918bd1b8d818476f57b2f7cab9ccfa56504da40fc1c586325d8aa2ded. Open and inspect trakt_repair.py before running it. No API request made.")'

原始碼寫入命令會顯示所建立檔案的 SHA-256。本版本經審查的原始碼 SHA-256 為 21c8f1a918bd1b8d818476f57b2f7cab9ccfa56504da40fc1c586325d8aa2ded。如果顯示的值不同，請停止，並在新的空白工作資料夾中重新複製完整命令。