This repository has no description
30 kB
710 lines
1import asyncio
2import json
3import logging
4import os
5import re
6import sys
7from pathlib import Path
8
9import aiohttp
10from aiohttp import web
11from nio import (
12 AsyncClient,
13 AsyncClientConfig,
14 InviteMemberEvent,
15 LoginResponse,
16 MatrixRoom,
17 MegolmEvent,
18 RoomMessageText,
19)
20
21logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
22log = logging.getLogger("archiver-bot")
23
24IMDB_RE = re.compile(r"https?://(?:www\.)?imdb\.com/(?:[a-z]{2}/)?title/(tt\d+)")
25TMDB_MOVIE_RE = re.compile(r"https?://(?:www\.)?themoviedb\.org/movie/(\d+)")
26TMDB_TV_RE = re.compile(r"https?://(?:www\.)?themoviedb\.org/tv/(\d+)")
27
28JELLYSEERR_URL = os.environ.get("JELLYSEERR_URL", "http://localhost:5055")
29JELLYSEERR_API_KEY = os.environ["JELLYSEERR_API_KEY"]
30RADARR_API_KEY = os.environ.get("RADARR_API_KEY", "")
31SONARR_API_KEY = os.environ.get("SONARR_API_KEY", "")
32MATRIX_HOMESERVER = os.environ["MATRIX_HOMESERVER"]
33MATRIX_ACCESS_TOKEN = os.environ["MATRIX_ACCESS_TOKEN"]
34MATRIX_USER_ID = os.environ.get("MATRIX_USER_ID", "@archiver:lassul.us")
35MATRIX_DEVICE_ID = os.environ.get("MATRIX_DEVICE_ID", "ARCHIVER")
36STORE_PATH = os.environ.get("STORE_PATH", "/var/lib/archiver-bot")
37WEBHOOK_PORT = int(os.environ.get("WEBHOOK_PORT", "8099"))
38FLAX_URL = os.environ.get("FLAX_URL", "https://flax.lassul.us")
39FLIX_URL = os.environ.get("FLIX_URL", "https://flix.lassul.us")
40JELLYFIN_URL = os.environ.get("JELLYFIN_URL", "http://localhost:8096")
41JELLYFIN_API_KEY = os.environ.get("JELLYFIN_API_KEY", "")
42IPFS_GATEWAY_URL = os.environ.get("IPFS_GATEWAY_URL", "https://ipfs.lassul.us")
43MEDIA_PATH_PREFIX = os.environ.get("MEDIA_PATH_PREFIX", "/var/download/")
44IPFS_BIN = os.environ.get("IPFS_BIN", "ipfs")
45
46# Persistent tracking of requested media -> rooms
47TRACKING_FILE = Path(STORE_PATH) / "tracked_requests.json"
48
49
50def load_tracked() -> dict:
51 """Load tracked requests from disk. Keys are 'movie:TMDBID' or 'tv:TMDBID', values are lists of room_ids."""
52 try:
53 return json.loads(TRACKING_FILE.read_text())
54 except Exception:
55 return {}
56
57
58def save_tracked(tracked: dict):
59 """Save tracked requests to disk."""
60 TRACKING_FILE.write_text(json.dumps(tracked))
61
62
63def track_request(media_type: str, tmdb_id: int, room_id: str):
64 """Track that a room requested this media."""
65 tracked = load_tracked()
66 key = f"{media_type}:{tmdb_id}"
67 rooms = tracked.get(key, [])
68 if room_id not in rooms:
69 rooms.append(room_id)
70 tracked[key] = rooms
71 save_tracked(tracked)
72
73
74def pop_tracked(media_type: str, tmdb_id: int) -> list[str]:
75 """Get and remove tracked rooms for this media."""
76 tracked = load_tracked()
77 key = f"{media_type}:{tmdb_id}"
78 rooms = tracked.pop(key, [])
79 save_tracked(tracked)
80 return rooms
81
82
83def make_flax_link(path: str) -> str:
84 """Turn a local path like /var/download/movies/Title (Year)/file.mp4 into a flax.lassul.us URL."""
85 from urllib.parse import quote
86 relative = path
87 if relative.startswith(MEDIA_PATH_PREFIX):
88 relative = relative[len(MEDIA_PATH_PREFIX):]
89 return f"{FLAX_URL}/{quote(relative)}"
90
91
92def find_main_video(path: str) -> str | None:
93 """Find the main video file in a path. If path is a file, return it. If directory, find largest video."""
94 if os.path.isfile(path):
95 return path
96 if os.path.isdir(path):
97 video_exts = {".mkv", ".mp4", ".avi", ".m4v", ".ts", ".wmv"}
98 best = None
99 best_size = 0
100 for root, _dirs, files in os.walk(path):
101 for f in files:
102 ext = os.path.splitext(f)[1].lower()
103 if ext in video_exts:
104 full = os.path.join(root, f)
105 size = os.path.getsize(full)
106 if size > best_size:
107 best = full
108 best_size = size
109 return best
110 return None
111
112
113async def compute_cid(path: str) -> str | None:
114 """Compute the IPFS CID for a file using ipfs add --only-hash --raw-leaves."""
115 video = find_main_video(path)
116 if not video:
117 log.warning(f"No video file found for {path}")
118 return None
119 import tempfile
120 import shutil
121 ipfs_path = tempfile.mkdtemp(prefix="ipfs-hash-")
122 try:
123 env = {**os.environ, "IPFS_PATH": ipfs_path}
124 init_proc = await asyncio.create_subprocess_exec(
125 IPFS_BIN, "init",
126 stdout=asyncio.subprocess.PIPE,
127 stderr=asyncio.subprocess.PIPE,
128 env=env,
129 )
130 await init_proc.communicate()
131 proc = await asyncio.create_subprocess_exec(
132 IPFS_BIN, "add", "--only-hash", "--raw-leaves", "--quieter", video,
133 stdout=asyncio.subprocess.PIPE,
134 stderr=asyncio.subprocess.PIPE,
135 env=env,
136 )
137 try:
138 stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=300)
139 except asyncio.TimeoutError:
140 proc.kill()
141 await proc.communicate()
142 log.warning(f"ipfs add --only-hash timed out for {video}")
143 return None
144 if proc.returncode == 0:
145 cid = stdout.decode().strip()
146 if cid:
147 return cid
148 else:
149 log.warning(f"ipfs add --only-hash failed for {video}: {stderr.decode().strip()}")
150 except Exception as e:
151 log.warning(f"Failed to compute CID for {video}: {e}")
152 finally:
153 shutil.rmtree(ipfs_path, ignore_errors=True)
154 return None
155
156
157async def trigger_jellyfin_scan(session: aiohttp.ClientSession):
158 """Trigger a Jellyfin library scan and Jellyseerr sync."""
159 try:
160 async with session.post(
161 f"{JELLYFIN_URL}/Library/Refresh",
162 headers={"X-Emby-Token": JELLYFIN_API_KEY},
163 ) as resp:
164 log.info(f"Triggered Jellyfin library scan: {resp.status}")
165 # also trigger Jellyseerr sync so it picks up the new jellyfinMediaId
166 await jellyseerr_request(session, "/settings/jellyfin/sync", method="POST")
167 log.info("Triggered Jellyseerr sync")
168 except Exception as e:
169 log.warning(f"Failed to trigger library scan: {e}")
170
171
172async def lookup_jellyfin_id(session: aiohttp.ClientSession, media_type: str, tmdb_id: int) -> str | None:
173 """Look up Jellyfin item ID via Jellyseerr's mediaInfo or Jellyfin API."""
174 # Try Jellyseerr first — it tracks the Jellyfin media ID
175 # Try Jellyseerr first
176 try:
177 endpoint = f"/movie/{tmdb_id}" if media_type == "movie" else f"/tv/{tmdb_id}"
178 details = await jellyseerr_request(session, endpoint)
179 media_info = details.get("mediaInfo")
180 if media_info:
181 jf_id = media_info.get("jellyfinMediaId")
182 if jf_id:
183 log.info(f"Jellyfin ID from Jellyseerr for {media_type}:{tmdb_id} = {jf_id}")
184 return jf_id
185 except Exception as e:
186 log.warning(f"Jellyseerr Jellyfin ID lookup failed for {media_type}:{tmdb_id}: {e}")
187
188 # Fallback: query Jellyfin directly, searching by provider ID
189 if JELLYFIN_API_KEY:
190 try:
191 item_type = "Movie" if media_type == "movie" else "Series"
192 async with session.get(
193 f"{JELLYFIN_URL}/Items",
194 params={
195 "IncludeItemTypes": item_type,
196 "Recursive": "true",
197 "Fields": "ProviderIds",
198 },
199 headers={"X-Emby-Token": JELLYFIN_API_KEY},
200 ) as resp:
201 data = await resp.json()
202 for item in data.get("Items", []):
203 provider_ids = item.get("ProviderIds", {})
204 if str(provider_ids.get("Tmdb")) == str(tmdb_id):
205 jf_id = item.get("Id")
206 log.info(f"Jellyfin ID from Jellyfin API for {media_type}:{tmdb_id} = {jf_id}")
207 return jf_id
208 except Exception as e:
209 log.warning(f"Jellyfin API lookup failed for {media_type}:{tmdb_id}: {e}")
210
211 log.info(f"No Jellyfin ID found for {media_type}:{tmdb_id}")
212 return None
213
214
215async def make_media_links(path: str, is_dir: bool = False, jellyfin_id: str | None = None) -> tuple[str, str]:
216 """Generate plain and HTML text with all available links for a media path."""
217 from urllib.parse import quote
218
219 lines_plain = []
220 lines_html = []
221
222 # Flax link (direct file download)
223 flax = make_flax_link(path)
224 lines_plain.append(f"📁 Download: {flax}")
225 lines_html.append(f'📁 Download: <a href="{flax}">{flax}</a>')
226
227 # Jellyfin link (streaming)
228 if jellyfin_id:
229 flix_link = f"{FLIX_URL}/web/#/details?id={jellyfin_id}"
230 else:
231 flix_link = FLIX_URL
232 lines_plain.append(f"🎬 Stream: {flix_link}")
233 lines_html.append(f'🎬 Stream: <a href="{flix_link}">{flix_link}</a>')
234
235 # IPFS link (compute CID for the file, retry up to 3 times)
236 cid = None
237 for attempt in range(3):
238 cid = await compute_cid(path)
239 if cid:
240 break
241 log.warning(f"IPFS hash attempt {attempt + 1}/3 failed for {path}")
242 await asyncio.sleep(2)
243
244 if cid:
245 ipfs_link = f"{IPFS_GATEWAY_URL}/ipfs/{cid}"
246 lines_plain.append(f"🌐 IPFS: {ipfs_link}")
247 lines_html.append(f'🌐 IPFS: <a href="{ipfs_link}">{ipfs_link}</a>')
248 lines_plain.append(f" ipfs get {cid}")
249 lines_html.append(f" <code>ipfs get {cid}</code>")
250 else:
251 log.error(f"IPFS hash failed after 3 attempts for {path}, not posting message")
252 return None
253
254 return ("\n".join(lines_plain), "<br>".join(lines_html))
255
256
257async def jellyseerr_request(session: aiohttp.ClientSession, path: str, **kwargs) -> dict:
258 """Make an authenticated request to Jellyseerr."""
259 headers = {
260 "X-Api-Key": JELLYSEERR_API_KEY,
261 "Content-Type": "application/json",
262 "Accept": "application/json",
263 }
264 method = kwargs.pop("method", "GET")
265 url = f"{JELLYSEERR_URL}/api/v1{path}"
266 async with session.request(method, url, headers=headers, **kwargs) as resp:
267 return await resp.json()
268
269
270async def find_by_imdb(session: aiohttp.ClientSession, imdb_id: str) -> dict | None:
271 """Look up a movie or TV show by IMDb ID via Radarr/Sonarr, returning Jellyseerr-compatible data."""
272
273 # Try Radarr (movies) first
274 try:
275 async with session.get(
276 f"http://localhost:7878/api/v3/movie/lookup?term=imdb:{imdb_id}",
277 headers={"X-Api-Key": RADARR_API_KEY},
278 ) as resp:
279 results = await resp.json()
280 if results and len(results) > 0:
281 tmdb_id = results[0].get("tmdbId")
282 title = results[0].get("title", "Unknown")
283 year = str(results[0].get("year", ""))
284 if tmdb_id:
285 log.info(f"Radarr resolved {imdb_id} -> TMDB {tmdb_id} ({title})")
286 return {
287 "id": tmdb_id,
288 "mediaType": "movie",
289 "title": title,
290 "releaseDate": f"{year}-01-01" if year else "",
291 }
292 except Exception as e:
293 log.warning(f"Radarr lookup failed: {e}")
294
295 # Try Sonarr (TV shows)
296 try:
297 async with session.get(
298 f"http://localhost:8989/api/v3/series/lookup?term=imdb:{imdb_id}",
299 headers={"X-Api-Key": SONARR_API_KEY},
300 ) as resp:
301 results = await resp.json()
302 if results and len(results) > 0:
303 tmdb_id = results[0].get("tmdbId", 0)
304 title = results[0].get("title", "Unknown")
305 year = str(results[0].get("year", ""))
306 if tmdb_id:
307 log.info(f"Sonarr resolved {imdb_id} -> TMDB {tmdb_id} ({title})")
308 return {
309 "id": tmdb_id,
310 "mediaType": "tv",
311 "name": title,
312 "firstAirDate": f"{year}-01-01" if year else "",
313 }
314 except Exception as e:
315 log.warning(f"Sonarr lookup failed: {e}")
316
317 return None
318
319
320async def lookup_media_path(session: aiohttp.ClientSession, media_type: str, tmdb_id: int) -> str | None:
321 """Look up the path for a movie file or show folder from Radarr/Sonarr by TMDB ID."""
322 try:
323 if media_type == "movie":
324 async with session.get(
325 f"http://localhost:7878/api/v3/movie?tmdbId={tmdb_id}",
326 headers={"X-Api-Key": RADARR_API_KEY},
327 ) as resp:
328 results = await resp.json()
329 if results and len(results) > 0:
330 movie = results[0]
331 if not movie.get("hasFile"):
332 return None
333 movie_file = movie.get("movieFile", {})
334 if movie_file and movie_file.get("path"):
335 return movie_file["path"]
336 return None
337 else:
338 async with session.get(
339 f"http://localhost:8989/api/v3/series?tvdbId={tmdb_id}",
340 headers={"X-Api-Key": SONARR_API_KEY},
341 ) as resp:
342 results = await resp.json()
343 if results and len(results) > 0:
344 return results[0].get("path")
345 except Exception as e:
346 log.warning(f"Media path lookup failed for {media_type}:{tmdb_id}: {e}")
347 return None
348
349
350async def request_media(session: aiohttp.ClientSession, media: dict, room_id: str) -> tuple[str, str]:
351 """Request a movie or TV show via Jellyseerr. Returns (plain, html) status message."""
352 media_type = media["mediaType"]
353 media_id = media["id"]
354 title = media.get("title") or media.get("name") or "Unknown"
355 year = (media.get("releaseDate") or media.get("firstAirDate") or "")[:4]
356 display = f"{title} ({year})" if year else title
357
358 # Check if already available or requested
359 existing = media.get("mediaInfo")
360 if existing:
361 status = existing.get("status")
362 # status 5 = available, 4 = partially available, 3 = processing, 2 = pending
363 if status == 5:
364 path = await lookup_media_path(session, media_type, media_id)
365 if path:
366 is_dir = media_type == "tv"
367 jf_id = await lookup_jellyfin_id(session, media_type, media_id)
368 links = await make_media_links(path, is_dir=is_dir, jellyfin_id=jf_id)
369 if not links:
370 log.error(f"IPFS hash failed for {display}, not posting"); return None
371 links_plain, links_html = links
372 return (f"✅ {display} is already available!\n{links_plain}", f"✅ <b>{display}</b> is already available!<br>{links_html}")
373 return (f"✅ {display} is already available!", f"✅ <b>{display}</b> is already available!")
374 elif status in (2, 3, 4):
375 track_request(media_type, media_id, room_id)
376 return (f"⏳ {display} has already been requested and is being processed.", f"⏳ <b>{display}</b> has already been requested and is being processed.")
377
378 payload = {"mediaType": media_type, "mediaId": media_id}
379 if media_type == "tv":
380 payload["seasons"] = "all"
381
382 # Check if Radarr/Sonarr already has the file before requesting
383 existing_path = await lookup_media_path(session, media_type, media_id)
384 if existing_path:
385 is_dir = media_type == "tv"
386 jf_id = await lookup_jellyfin_id(session, media_type, media_id)
387 links = await make_media_links(existing_path, is_dir=is_dir, jellyfin_id=jf_id)
388 # Still send the request to Jellyseerr so it's tracked there too
389 await jellyseerr_request(session, "/request", method="POST", json=payload)
390 if not links:
391 log.error(f"IPFS hash failed for {display}, not posting"); return None
392 links_plain, links_html = links
393 return (f"✅ {display} is already available!\n{links_plain}", f"✅ <b>{display}</b> is already available!<br>{links_html}")
394
395 result = await jellyseerr_request(session, "/request", method="POST", json=payload)
396
397 if "id" in result:
398 track_request(media_type, media_id, room_id)
399 return (f"🎬 Successfully requested {display}! I'll notify you when it's downloaded.", f"🎬 Successfully requested <b>{display}</b>! I'll notify you when it's downloaded.")
400 else:
401 error = result.get("message", json.dumps(result))
402 return (f"❌ Failed to request {display}: {error}", f"❌ Failed to request <b>{display}</b>: {error}")
403
404
405async def trust_room_devices(client: AsyncClient, room_id: str):
406 """Trust all devices of all members in a room."""
407 if not client.olm:
408 return
409 room = client.rooms.get(room_id)
410 if not room:
411 return
412 for user_id in room.users:
413 for device in client.device_store.active_user_devices(user_id):
414 if not client.olm.is_device_verified(device):
415 client.verify_device(device)
416 log.info(f"Verified device {device.device_id} of {user_id}")
417
418
419async def send_notice(client: AsyncClient, room_id: str, plain: str, html: str):
420 """Send a notice message to a room."""
421 await trust_room_devices(client, room_id)
422 await client.room_send(
423 room_id,
424 "m.room.message",
425 {"msgtype": "m.notice", "body": plain, "format": "org.matrix.custom.html", "formatted_body": html},
426 )
427
428
429async def on_message(client: AsyncClient, room: MatrixRoom, event: RoomMessageText, http: aiohttp.ClientSession):
430 """Handle messages in rooms, looking for IMDb/TMDB links."""
431 # Ignore our own messages
432 if event.sender == client.user_id:
433 return
434
435 body = event.body
436 log.info(f"Message from {event.sender} in {room.display_name}: {body[:100]}")
437
438 # Help command
439 if body.strip().lower() in ("help", "!help"):
440 plain = (
441 "Archiver Bot\n\n"
442 "Send me an IMDb or TMDB link and I'll request the movie/show on Jellyseerr.\n"
443 "I'll notify you when the download is complete.\n\n"
444 "Examples:\n"
445 " https://www.imdb.com/title/tt0133093/\n"
446 " https://www.themoviedb.org/movie/603\n"
447 " https://www.themoviedb.org/tv/1399"
448 )
449 html = (
450 "<b>Archiver Bot</b><br><br>"
451 "Send me an IMDb or TMDB link and I'll request the movie/show on Jellyseerr.<br>"
452 "I'll notify you when the download is complete.<br><br>"
453 "Examples:<br>"
454 "<code>https://www.imdb.com/title/tt0133093/</code><br>"
455 "<code>https://www.themoviedb.org/movie/603</code><br>"
456 "<code>https://www.themoviedb.org/tv/1399</code>"
457 )
458 await send_notice(client, room.room_id, plain, html)
459 return
460
461 # Collect all media references from the message
462 results = []
463
464 for imdb_id in set(IMDB_RE.findall(body)):
465 log.info(f"Found IMDb link {imdb_id} in {room.display_name} from {event.sender}")
466 media = await find_by_imdb(http, imdb_id)
467 if media is None:
468 results.append((f"❓ Could not find anything for {imdb_id}.", f"❓ Could not find anything for <code>{imdb_id}</code>."))
469 else:
470 results.append(await request_media(http, media, room.room_id))
471
472 for tmdb_id in set(TMDB_MOVIE_RE.findall(body)):
473 log.info(f"Found TMDB movie link {tmdb_id} in {room.display_name} from {event.sender}")
474 media = {"id": int(tmdb_id), "mediaType": "movie"}
475 try:
476 details = await jellyseerr_request(http, f"/movie/{tmdb_id}")
477 media["title"] = details.get("title", "Unknown")
478 media["releaseDate"] = details.get("releaseDate", "")
479 media["mediaInfo"] = details.get("mediaInfo")
480 except Exception as e:
481 log.warning(f"Failed to fetch TMDB movie {tmdb_id}: {e}")
482 results.append(await request_media(http, media, room.room_id))
483
484 for tmdb_id in set(TMDB_TV_RE.findall(body)):
485 log.info(f"Found TMDB TV link {tmdb_id} in {room.display_name} from {event.sender}")
486 media = {"id": int(tmdb_id), "mediaType": "tv"}
487 try:
488 details = await jellyseerr_request(http, f"/tv/{tmdb_id}")
489 media["name"] = details.get("name", "Unknown")
490 media["firstAirDate"] = details.get("firstAirDate", "")
491 media["mediaInfo"] = details.get("mediaInfo")
492 except Exception as e:
493 log.warning(f"Failed to fetch TMDB TV {tmdb_id}: {e}")
494 results.append(await request_media(http, media, room.room_id))
495
496 for result in results:
497 if result is not None:
498 plain, html = result
499 await send_notice(client, room.room_id, plain, html)
500
501
502async def on_megolm(client: AsyncClient, room: MatrixRoom, event: MegolmEvent):
503 """Handle messages we couldn't decrypt."""
504 log.warning(
505 f"Failed to decrypt message from {event.sender} in {room.display_name} "
506 f"(session {event.session_id}): {event.message}"
507 )
508 # Request the keys we're missing
509 if client.olm:
510 try:
511 resp = await client.request_room_key(event)
512 log.info(f"Requested room key for session {event.session_id}: {resp}")
513 except Exception as e:
514 log.warning(f"Failed to request room key: {e}")
515 await send_notice(
516 client,
517 room.room_id,
518 "🔒 I couldn't decrypt your message. Please try again or re-invite me to this room.",
519 "🔒 I couldn't decrypt your message. Please try again or re-invite me to this room.",
520 )
521
522
523async def on_invite(client: AsyncClient, room: MatrixRoom, event: InviteMemberEvent):
524 """Auto-join rooms when invited."""
525 log.info(f"Invite event in {room.room_id}: state_key={event.state_key}, sender={event.sender}, membership={event.membership}, our_id={client.user_id}")
526 if event.state_key != client.user_id:
527 return
528 log.info(f"Joining {room.room_id} (invited by {event.sender})...")
529 result = await client.join(room.room_id)
530 log.info(f"Join result for {room.room_id}: {result}")
531
532
533async def handle_radarr_webhook(request):
534 """Handle Radarr webhook for download completion."""
535 try:
536 data = await request.json()
537 event_type = data.get("eventType", "")
538 log.info(f"Radarr webhook: {event_type}")
539
540 if event_type in ("Download", "MovieFileDelete"):
541 movie = data.get("movie", {})
542 tmdb_id = movie.get("tmdbId")
543 title = movie.get("title", "Unknown")
544 year = movie.get("year", "")
545 display = f"{title} ({year})" if year else title
546 # Prefer the actual movie file path, fall back to folder
547 movie_file = data.get("movieFile", {})
548 file_path = movie_file.get("path") or movie.get("folderPath", "")
549
550 if event_type == "Download" and tmdb_id:
551 rooms = pop_tracked("movie", tmdb_id)
552 if rooms:
553 client = request.app["matrix_client"]
554 http = request.app["http_session"]
555 # trigger library scan so Jellyfin picks up the new file
556 await trigger_jellyfin_scan(http)
557 await asyncio.sleep(10)
558 jf_id = await lookup_jellyfin_id(http, "movie", tmdb_id)
559 links = await make_media_links(file_path, jellyfin_id=jf_id)
560 if not links:
561 log.error(f"IPFS hash failed for {display}, not notifying")
562 return web.Response(text="ok")
563 links_plain, links_html = links
564 plain = f"✅ {display} has been downloaded and is now available!\n{links_plain}"
565 html = f"✅ <b>{display}</b> has been downloaded and is now available!<br>{links_html}"
566 for room_id in rooms:
567 try:
568 await send_notice(client, room_id, plain, html)
569 except Exception as e:
570 log.warning(f"Failed to notify room {room_id}: {e}")
571 except Exception as e:
572 log.exception(f"Error handling Radarr webhook: {e}")
573
574 return web.Response(text="ok")
575
576
577async def handle_sonarr_webhook(request):
578 """Handle Sonarr webhook for download completion."""
579 try:
580 data = await request.json()
581 event_type = data.get("eventType", "")
582 log.info(f"Sonarr webhook: {event_type}")
583
584 if event_type in ("Download", "EpisodeFileDelete"):
585 series = data.get("series", {})
586 tmdb_id = series.get("tmdbId")
587 title = series.get("title", "Unknown")
588 year = series.get("year", "")
589 display = f"{title} ({year})" if year else title
590
591 episodes = data.get("episodes", [])
592 ep_info = ""
593 if episodes:
594 ep_nums = [f"S{ep.get('seasonNumber', 0):02d}E{ep.get('episodeNumber', 0):02d}" for ep in episodes]
595 ep_info = f" ({', '.join(ep_nums)})"
596
597 if event_type == "Download" and tmdb_id:
598 rooms = pop_tracked("tv", tmdb_id)
599 if rooms:
600 client = request.app["matrix_client"]
601 http = request.app["http_session"]
602 folder_path = series.get("path", "")
603 # trigger library scan so Jellyfin picks up the new file
604 await trigger_jellyfin_scan(http)
605 await asyncio.sleep(10)
606 jf_id = await lookup_jellyfin_id(http, "tv", tmdb_id)
607 links = await make_media_links(folder_path, is_dir=True, jellyfin_id=jf_id)
608 if not links:
609 log.error(f"IPFS hash failed for {display}, not notifying")
610 return web.Response(text="ok")
611 links_plain, links_html = links
612 plain = f"✅ {display}{ep_info} has been downloaded and is now available!\n{links_plain}"
613 html = f"✅ <b>{display}</b>{ep_info} has been downloaded and is now available!<br>{links_html}"
614 for room_id in rooms:
615 try:
616 await send_notice(client, room_id, plain, html)
617 except Exception as e:
618 log.warning(f"Failed to notify room {room_id}: {e}")
619 except Exception as e:
620 log.exception(f"Error handling Sonarr webhook: {e}")
621
622 return web.Response(text="ok")
623
624
625async def run():
626 """Main bot loop."""
627 client_config = AsyncClientConfig(
628 store_sync_tokens=True,
629 encryption_enabled=True,
630 )
631 client = AsyncClient(
632 MATRIX_HOMESERVER,
633 MATRIX_USER_ID,
634 device_id=MATRIX_DEVICE_ID,
635 store_path=STORE_PATH,
636 config=client_config,
637 )
638 client.access_token = MATRIX_ACCESS_TOKEN
639 client.user_id = MATRIX_USER_ID
640 client.load_store()
641 log.info(f"Crypto store loaded, olm={client.olm is not None}")
642
643 http = aiohttp.ClientSession()
644
645 # Do an initial sync BEFORE registering callbacks to skip old messages
646 log.info("Performing initial sync...")
647 resp = await client.sync(timeout=10000, full_state=True)
648 if hasattr(resp, "next_batch"):
649 log.info(f"Initial sync complete, next_batch: {resp.next_batch}")
650 else:
651 log.error(f"Initial sync failed: {resp}")
652 await http.close()
653 await client.close()
654 sys.exit(1)
655
656 # Join any rooms we were invited to while offline
657 for room_id, room in client.invited_rooms.items():
658 log.info(f"Joining pending invite for {room_id}...")
659 result = await client.join(room_id)
660 log.info(f"Join result for {room_id}: {result}")
661
662 # E2EE setup
663 if client.olm:
664 try:
665 log.info("Uploading device keys...")
666 key_upload = await client.keys_upload()
667 log.info(f"Key upload result: {key_upload}")
668 except Exception as e:
669 log.info(f"Key upload skipped: {e}")
670
671 # Trust all known device keys so we can decrypt messages
672 for user_id in client.device_store.users:
673 for device in client.device_store.active_user_devices(user_id):
674 if not client.olm.is_device_verified(device):
675 client.verify_device(device)
676 log.info(f"Verified device {device.device_id} of {user_id}")
677 else:
678 log.warning("Olm not available after sync — E2EE will not work")
679
680 # Register event callbacks only after initial sync
681 client.add_event_callback(lambda room, event: on_message(client, room, event, http), RoomMessageText)
682 client.add_event_callback(lambda room, event: on_invite(client, room, event), InviteMemberEvent)
683 client.add_event_callback(lambda room, event: on_megolm(client, room, event), MegolmEvent)
684
685 # Start webhook server for Radarr/Sonarr notifications
686 app = web.Application()
687 app["matrix_client"] = client
688 app["http_session"] = http
689 app.router.add_post("/webhook/radarr", handle_radarr_webhook)
690 app.router.add_post("/webhook/sonarr", handle_sonarr_webhook)
691 runner = web.AppRunner(app)
692 await runner.setup()
693 site = web.TCPSite(runner, "127.0.0.1", WEBHOOK_PORT)
694 await site.start()
695 log.info(f"Webhook server listening on 127.0.0.1:{WEBHOOK_PORT}")
696
697 log.info(f"Bot {MATRIX_USER_ID} is running!")
698
699 try:
700 await client.sync_forever(timeout=30000, full_state=True)
701 except Exception:
702 log.exception("sync_forever crashed")
703 finally:
704 await runner.cleanup()
705 await http.close()
706 await client.close()
707
708
709def main():
710 asyncio.run(run())