Like 0
Fork 0

Last active 1737511963

A Python script that syncs subscriptions and playlists from a Piped instance to Invidious

fabio's Avatar fabio revised this gist 1737511963. Go to revision

No changes

fabio's Avatar fabio revised this gist 1737511913. Go to revision

No changes

fabio's Avatar Fabio Manganiello revised this gist 1737511825. Go to revision

1 file changed, 299 insertions

sync_piped_to_invidious.py(file created)

@@ -0,0 +1,299 @@
1 + #!/usr/bin/env python3
2 +
3 + ##########################################################################################
4 + # This script is used to import subscriptions and playlists from Piped to Invidious.
5 + #
6 + # It requires the following:
7 + #
8 + # - python3 + requests
9 + # - The Piped API URL (e.g. https://pipedapi.example.com).
10 + # Note: This is NOT the Piped frontend URL.
11 + # - The Invidious URL (e.g. https://invidious.example.com).
12 + # - A Piped API token (it can be obtained from an authenticated browser session),
13 + # passed on the `PIPED_TOKEN` environment variable
14 + # - The Invidious cookie header passed on the `INVIDIOUS_TOKEN` environment variable.
15 + # You can retrieve it in your browser via https://<invidious-url>/authorize_token?scopes=:*.
16 + #
17 + # Author: Fabio Manganiello <fabio@manganiello.tech>
18 + # License: MIT
19 + ##########################################################################################
20 +
21 + import argparse
22 + import os
23 + import re
24 + import sys
25 +
26 + from abc import ABC, abstractmethod
27 + from typing import Sequence
28 + from urllib.parse import urljoin
29 +
30 + import requests
31 +
32 +
33 + class Session:
34 + """
35 + A session contains the data required to interact with the Piped API and the Invidious instance.
36 + """
37 +
38 + def __init__(
39 + self,
40 + piped_api_url: str,
41 + invidious_url: str,
42 + piped_token: str,
43 + invidious_token: str,
44 + skip_playlists: bool = False,
45 + skip_subscriptions: bool = False,
46 + ):
47 + self.piped_api_url = piped_api_url
48 + self.invidious_url = invidious_url
49 + self.piped_token = piped_token
50 + self.invidious_token = invidious_token
51 + self.skip_playlists = skip_playlists
52 + self.skip_subscriptions = skip_subscriptions
53 + self.piped = PipedAPI(self)
54 + self.invidious = InvidiousAPI(self)
55 +
56 + @classmethod
57 + def from_cmdline(cls, args: Sequence[str]) -> "Session":
58 + argparser = argparse.ArgumentParser(
59 + description="Import subscriptions and playlists from Piped to Invidious."
60 + )
61 +
62 + argparser.add_argument(
63 + "--piped-api-url",
64 + required=True,
65 + help="The base URL of the Piped API of the instance you want to import from, e.g. https://pipedapi.example.com",
66 + )
67 +
68 + argparser.add_argument(
69 + "--invidious-url",
70 + required=True,
71 + help="The base URL of the Invidious instance you want to import to, e.g. https://invidious.example.com",
72 + )
73 +
74 + argparser.add_argument(
75 + "--skip-playlists", action="store_true", help="Skip importing playlists"
76 + )
77 +
78 + argparser.add_argument(
79 + "--skip-subscriptions",
80 + action="store_true",
81 + help="Skip importing subscriptions",
82 + )
83 +
84 + piped_token = os.getenv("PIPED_TOKEN")
85 + invidious_token = os.getenv("INVIDIOUS_TOKEN")
86 +
87 + if not piped_token:
88 + argparser.error("The PIPED_TOKEN environment variable is required")
89 +
90 + if not invidious_token:
91 + argparser.error(
92 + "The INVIDIOUS_TOKEN environment variable is required.\n"
93 + "You can retrieve it in your browser via https://<invidious-url>/authorize_token?scopes=:*"
94 + )
95 +
96 + opts, _ = argparser.parse_known_args(args)
97 + return cls(
98 + piped_api_url=opts.piped_api_url,
99 + invidious_url=opts.invidious_url,
100 + piped_token=piped_token,
101 + invidious_token=invidious_token,
102 + skip_playlists=opts.skip_playlists,
103 + skip_subscriptions=opts.skip_subscriptions,
104 + )
105 +
106 + def migrate_subscriptions(self):
107 + if self.skip_subscriptions:
108 + print("=== Skipping subscriptions import ===")
109 + return
110 +
111 + print("=== Importing subscriptions ===")
112 +
113 + try:
114 + rs = self.piped.request("/subscriptions")
115 + except requests.exceptions.HTTPError as e:
116 + sys.stderr.write(f"===> [-] Could not fetch Piped subscriptions: {e}\n")
117 + sys.exit(1)
118 +
119 + piped_subs = rs.json()
120 + print(f"=== Found {len(piped_subs)} Piped subscriptions ===")
121 +
122 + invidious_subs = {
123 + sub["authorId"]
124 + for sub in self.invidious.request("/auth/subscriptions").json()
125 + }
126 +
127 + for subs in piped_subs:
128 + channel = re.sub(r"^/channel/", "", subs["url"])
129 + if channel in invidious_subs:
130 + print(
131 + f'[=] Channel "{subs["name"]}" already subscribed on Invidious, skipping'
132 + )
133 + continue
134 +
135 + print(f'[+] Importing channel: {subs["name"]}')
136 +
137 + try:
138 + rs = self.invidious.request(
139 + f"/auth/subscriptions/{channel}", method="POST"
140 + )
141 + except requests.exceptions.HTTPError as e:
142 + sys.stderr.write(
143 + f'===> [-] Could not subscribe to channel: {subs["name"]}: {e}\n'
144 + )
145 +
146 + def migrate_playlists(self):
147 + if self.skip_playlists:
148 + print("=== Skipping playlists import ===")
149 + return
150 +
151 + print("=== Importing playlists ===")
152 +
153 + try:
154 + rs = self.piped.request("/user/playlists")
155 + except requests.exceptions.HTTPError as e:
156 + sys.stderr.write(f"===> [-] Could not fetch Piped playlists: {e}\n")
157 + sys.exit(1)
158 +
159 + piped_playlists = rs.json()
160 + print(f"=== Found {len(piped_playlists)} Piped playlists ===")
161 +
162 + try:
163 + rs = self.invidious.request("/auth/playlists")
164 + except requests.exceptions.HTTPError as e:
165 + sys.stderr.write(f"===> [-] Could not fetch Invidious playlists: {e}\n")
166 + sys.exit(1)
167 +
168 + invidious_playlists = {
169 + pl["title"]: {
170 + **pl,
171 + "videos": {item["videoId"]: item for item in pl.get("videos", [])},
172 + }
173 + for pl in rs.json()
174 + }
175 +
176 + print(f"=== Found {len(invidious_playlists)} Invidious playlists ===")
177 +
178 + for pl in piped_playlists:
179 + try:
180 + rs = self.piped.request(f'/playlists/{pl["id"]}')
181 + except requests.exceptions.HTTPError as e:
182 + sys.stderr.write(
183 + f'===> [-] Could not fetch playlist: {pl["name"]}: {e}\n'
184 + )
185 + continue
186 +
187 + items = [
188 + re.sub(r"^/watch\?v=", "", item["url"])
189 + for item in rs.json().get("relatedStreams", [])
190 + ]
191 +
192 + print(f'Importing {len(items)} videos from playlist: {pl["name"]}')
193 +
194 + if pl["name"] in invidious_playlists:
195 + playlist_id = invidious_playlists[pl["name"]]["playlistId"]
196 + print(
197 + f'[=] Playlist "{pl["name"]}" already exists on Invidious, skipping creation'
198 + )
199 + else:
200 + try:
201 + rs = self.invidious.request(
202 + "/auth/playlists",
203 + method="POST",
204 + json={
205 + "title": pl["name"],
206 + "privacy": "unlisted",
207 + },
208 + )
209 + except requests.exceptions.HTTPError as e:
210 + sys.stderr.write(
211 + f'===> [-] Could not create playlist on Invidious: {pl["name"]}: {e}\n'
212 + )
213 + continue
214 +
215 + playlist_id = rs.json().get("playlistId")
216 + if not playlist_id:
217 + sys.stderr.write(
218 + f'===> [-] Could not retrieve ID of the newly created Invidious playlist: {pl["name"]}\n'
219 + )
220 + continue
221 +
222 + print(
223 + f'[+] Created playlist "{pl["name"]}" on Invidious, ID: {playlist_id}'
224 + )
225 +
226 + for item in items:
227 + if item in invidious_playlists.get(pl["name"], {}).get("videos", {}):
228 + print(f'[=] Video "{item}" already exists on Invidious, skipping')
229 + continue
230 +
231 + try:
232 + rs = self.invidious.request(
233 + f"/auth/playlists/{playlist_id}/videos",
234 + method="POST",
235 + json={"videoId": item},
236 + )
237 +
238 + print(f'[+] Added video "{item}" to playlist "{pl["name"]}"')
239 + except requests.exceptions.HTTPError as e:
240 + sys.stderr.write(
241 + f'===> [-] Could not add video {item} to playlist {pl["name"]}: {e}\n'
242 + )
243 + continue
244 +
245 +
246 + class API(ABC):
247 + def __init__(self, session: Session, timeout: float = 10):
248 + self.session = session
249 + self.timeout = timeout
250 +
251 + @abstractmethod
252 + def request(self, url: str, method: str = "GET", **kwargs) -> requests.Response:
253 + pass
254 +
255 +
256 + class InvidiousAPI(API):
257 + def request(self, url: str, method: str = "GET", **kwargs) -> requests.Response:
258 + rs = requests.request(
259 + method,
260 + urljoin(self.session.invidious_url, f'/api/v1/{url.lstrip("/")}'),
261 + headers={
262 + "Authorization": f"Bearer {self.session.invidious_token}",
263 + },
264 + **{
265 + "timeout": self.timeout,
266 + **kwargs,
267 + },
268 + )
269 +
270 + rs.raise_for_status()
271 + return rs
272 +
273 +
274 + class PipedAPI(API):
275 + def request(self, url: str, method: str = "GET", **kwargs) -> requests.Response:
276 + rs = requests.request(
277 + method,
278 + urljoin(self.session.piped_api_url, url.lstrip("/")),
279 + headers={
280 + "Authorization": self.session.piped_token,
281 + },
282 + **{
283 + "timeout": self.timeout,
284 + **kwargs,
285 + },
286 + )
287 +
288 + rs.raise_for_status()
289 + return rs
290 +
291 +
292 + def main():
293 + session = Session.from_cmdline(sys.argv[1:])
294 + session.migrate_subscriptions()
295 + session.migrate_playlists()
296 +
297 +
298 + if __name__ == "__main__":
299 + main()
Newer Older

Powered by Opengist ⋅ Load: 28ms⋅

English