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