Implement Melora: *arr media arrival webhook announcer for Matrix
Replace Bellhop scaffolding with full Melora implementation: - FastAPI webhook endpoints for Radarr, Sonarr, and Lidarr - matrix-nio integration with threaded room posting - SQLite for thread root persistence and event deduplication - Message formatting with plain text and HTML for each media type - Shared secret authentication via X-Arr-Webhook-Secret header - Updated dependencies, configuration, and documentation https://claude.ai/code/session_01DuzWyMMXvLMB4VxEwJyV4X
This commit is contained in:
Claude
0
app/__init__.py
Normal file
0
app/__init__.py
Normal file
18
app/config.py
Normal file
18
app/config.py
Normal file
@@ -0,0 +1,18 @@
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
|
||||
class Config:
|
||||
MATRIX_HOMESERVER_URL: str = os.environ["MATRIX_HOMESERVER_URL"]
|
||||
MATRIX_BOT_USER_ID: str = os.environ["MATRIX_BOT_USER_ID"]
|
||||
MATRIX_BOT_ACCESS_TOKEN: str = os.environ["MATRIX_BOT_ACCESS_TOKEN"]
|
||||
MATRIX_ARRIVALS_ROOM_ID: str = os.environ["MATRIX_ARRIVALS_ROOM_ID"]
|
||||
|
||||
WEBHOOK_SECRET: str = os.environ["WEBHOOK_SECRET"]
|
||||
|
||||
DATABASE_PATH: str = os.getenv("DATABASE_PATH", "melora.db")
|
||||
|
||||
|
||||
config = Config()
|
||||
85
app/database.py
Normal file
85
app/database.py
Normal file
@@ -0,0 +1,85 @@
|
||||
import aiosqlite
|
||||
|
||||
from app.config import config
|
||||
|
||||
_db: aiosqlite.Connection | None = None
|
||||
|
||||
|
||||
async def get_db() -> aiosqlite.Connection:
|
||||
global _db
|
||||
if _db is None:
|
||||
raise RuntimeError("Database not initialised — call init_db() first")
|
||||
return _db
|
||||
|
||||
|
||||
async def init_db() -> None:
|
||||
global _db
|
||||
_db = await aiosqlite.connect(config.DATABASE_PATH)
|
||||
_db.row_factory = aiosqlite.Row
|
||||
await _db.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS thread_roots (
|
||||
media_type TEXT PRIMARY KEY,
|
||||
event_id TEXT NOT NULL,
|
||||
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
"""
|
||||
)
|
||||
await _db.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS posted_events (
|
||||
id TEXT PRIMARY KEY,
|
||||
title TEXT NOT NULL,
|
||||
media_type TEXT NOT NULL,
|
||||
is_upgrade BOOLEAN NOT NULL,
|
||||
posted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
||||
)
|
||||
"""
|
||||
)
|
||||
await _db.commit()
|
||||
|
||||
|
||||
async def close_db() -> None:
|
||||
global _db
|
||||
if _db is not None:
|
||||
await _db.close()
|
||||
_db = None
|
||||
|
||||
|
||||
async def get_thread_root(media_type: str) -> str | None:
|
||||
db = await get_db()
|
||||
cursor = await db.execute(
|
||||
"SELECT event_id FROM thread_roots WHERE media_type = ?",
|
||||
(media_type,),
|
||||
)
|
||||
row = await cursor.fetchone()
|
||||
return row["event_id"] if row else None
|
||||
|
||||
|
||||
async def set_thread_root(media_type: str, event_id: str) -> None:
|
||||
db = await get_db()
|
||||
await db.execute(
|
||||
"INSERT OR REPLACE INTO thread_roots (media_type, event_id) VALUES (?, ?)",
|
||||
(media_type, event_id),
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
|
||||
async def is_event_posted(event_key: str) -> bool:
|
||||
db = await get_db()
|
||||
cursor = await db.execute(
|
||||
"SELECT 1 FROM posted_events WHERE id = ?",
|
||||
(event_key,),
|
||||
)
|
||||
return await cursor.fetchone() is not None
|
||||
|
||||
|
||||
async def record_posted_event(
|
||||
event_key: str, title: str, media_type: str, is_upgrade: bool
|
||||
) -> None:
|
||||
db = await get_db()
|
||||
await db.execute(
|
||||
"INSERT OR IGNORE INTO posted_events (id, title, media_type, is_upgrade) VALUES (?, ?, ?, ?)",
|
||||
(event_key, title, media_type, is_upgrade),
|
||||
)
|
||||
await db.commit()
|
||||
@@ -1 +0,0 @@
|
||||
|
||||
120
app/formatters.py
Normal file
120
app/formatters.py
Normal file
@@ -0,0 +1,120 @@
|
||||
import markdown as md
|
||||
|
||||
|
||||
def _render_html(text: str) -> str:
|
||||
return md.markdown(text, extensions=["extra"])
|
||||
|
||||
|
||||
def _strip_markdown(text: str) -> str:
|
||||
return text.replace("**", "").replace("*", "").replace("> ", "")
|
||||
|
||||
|
||||
def format_radarr(payload: dict) -> tuple[str, str]:
|
||||
movie = payload.get("movie", {})
|
||||
title = movie.get("title", "Unknown")
|
||||
year = movie.get("year", "")
|
||||
is_upgrade = payload.get("isUpgrade", False)
|
||||
|
||||
quality = (
|
||||
payload.get("movieFile", {})
|
||||
.get("quality", {})
|
||||
.get("quality", {})
|
||||
.get("name", "Unknown")
|
||||
)
|
||||
|
||||
if is_upgrade:
|
||||
lines = [
|
||||
f"\U0001f3ac **{title}** ({year})",
|
||||
f"> \u2b06\ufe0f *Quality upgrade*",
|
||||
f"> \U0001f39e\ufe0f Quality: {quality}",
|
||||
]
|
||||
else:
|
||||
lines = [
|
||||
f"\U0001f3ac **{title}** ({year})",
|
||||
f"> \u2705 *New addition*",
|
||||
f"> \U0001f39e\ufe0f Quality: {quality}",
|
||||
]
|
||||
|
||||
body_md = "\n".join(lines)
|
||||
plain = _strip_markdown(body_md)
|
||||
html = _render_html(body_md)
|
||||
return plain, html
|
||||
|
||||
|
||||
def format_sonarr(payload: dict) -> tuple[str, str]:
|
||||
series = payload.get("series", {})
|
||||
series_title = series.get("title", "Unknown")
|
||||
is_upgrade = payload.get("isUpgrade", False)
|
||||
|
||||
episodes = payload.get("episodes", [{}])
|
||||
ep = episodes[0] if episodes else {}
|
||||
season = ep.get("seasonNumber", 0)
|
||||
episode = ep.get("episodeNumber", 0)
|
||||
ep_title = ep.get("title", "")
|
||||
|
||||
quality = (
|
||||
payload.get("episodeFile", {})
|
||||
.get("quality", {})
|
||||
.get("quality", {})
|
||||
.get("name", "Unknown")
|
||||
)
|
||||
|
||||
header = f"\U0001f4fa **{series_title}** \u2014 S{season:02d}E{episode:02d}"
|
||||
if ep_title:
|
||||
header += f" \u2014 *{ep_title}*"
|
||||
|
||||
if is_upgrade:
|
||||
lines = [
|
||||
header,
|
||||
f"> \u2b06\ufe0f *Quality upgrade*",
|
||||
f"> \U0001f39e\ufe0f Quality: {quality}",
|
||||
]
|
||||
else:
|
||||
lines = [
|
||||
header,
|
||||
f"> \u2705 *New addition*",
|
||||
f"> \U0001f39e\ufe0f Quality: {quality}",
|
||||
]
|
||||
|
||||
body_md = "\n".join(lines)
|
||||
plain = _strip_markdown(body_md)
|
||||
html = _render_html(body_md)
|
||||
return plain, html
|
||||
|
||||
|
||||
def format_lidarr(payload: dict) -> tuple[str, str]:
|
||||
artist = payload.get("artist", {})
|
||||
artist_name = artist.get("name", "Unknown")
|
||||
album = payload.get("album", {})
|
||||
album_title = album.get("title", "")
|
||||
is_upgrade = payload.get("isUpgrade", False)
|
||||
|
||||
track_files = payload.get("trackFiles", [{}])
|
||||
tf = track_files[0] if track_files else {}
|
||||
quality = (
|
||||
tf.get("quality", {})
|
||||
.get("quality", {})
|
||||
.get("name", "Unknown")
|
||||
)
|
||||
|
||||
header = f"\U0001f3b5 **{artist_name}**"
|
||||
if album_title:
|
||||
header += f" \u2014 *{album_title}*"
|
||||
|
||||
if is_upgrade:
|
||||
lines = [
|
||||
header,
|
||||
f"> \u2b06\ufe0f *Quality upgrade*",
|
||||
f"> \U0001f3b5 Format: {quality}",
|
||||
]
|
||||
else:
|
||||
lines = [
|
||||
header,
|
||||
f"> \u2705 *New addition*",
|
||||
f"> \U0001f3b5 Format: {quality}",
|
||||
]
|
||||
|
||||
body_md = "\n".join(lines)
|
||||
plain = _strip_markdown(body_md)
|
||||
html = _render_html(body_md)
|
||||
return plain, html
|
||||
46
app/main.py
Normal file
46
app/main.py
Normal file
@@ -0,0 +1,46 @@
|
||||
import logging
|
||||
from contextlib import asynccontextmanager
|
||||
|
||||
from fastapi import FastAPI
|
||||
|
||||
from app.database import close_db, get_thread_root, init_db, set_thread_root
|
||||
from app.matrix import THREAD_ROOT_MESSAGES, close_client, send_thread_root
|
||||
from app.webhooks import router as webhook_router
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s %(levelname)s %(name)s: %(message)s",
|
||||
)
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
log.info("Starting Melora")
|
||||
await init_db()
|
||||
await _ensure_thread_roots()
|
||||
log.info("Melora ready")
|
||||
yield
|
||||
log.info("Shutting down Melora")
|
||||
await close_client()
|
||||
await close_db()
|
||||
|
||||
|
||||
async def _ensure_thread_roots() -> None:
|
||||
for media_type in THREAD_ROOT_MESSAGES:
|
||||
existing = await get_thread_root(media_type)
|
||||
if existing:
|
||||
log.info("Thread root for %s: %s", media_type, existing)
|
||||
continue
|
||||
log.info("Creating thread root for %s", media_type)
|
||||
event_id = await send_thread_root(media_type)
|
||||
await set_thread_root(media_type, event_id)
|
||||
|
||||
|
||||
app = FastAPI(title="Melora", lifespan=lifespan)
|
||||
app.include_router(webhook_router)
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
async def health():
|
||||
return {"status": "ok"}
|
||||
80
app/matrix.py
Normal file
80
app/matrix.py
Normal file
@@ -0,0 +1,80 @@
|
||||
import logging
|
||||
|
||||
from nio import AsyncClient, RoomSendResponse
|
||||
|
||||
from app.config import config
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
_client: AsyncClient | None = None
|
||||
|
||||
THREAD_ROOT_MESSAGES: dict[str, str] = {
|
||||
"movies": "\U0001f3ac Movies",
|
||||
"shows": "\U0001f4fa Shows",
|
||||
"music": "\U0001f3b5 Music",
|
||||
}
|
||||
|
||||
|
||||
async def get_client() -> AsyncClient:
|
||||
global _client
|
||||
if _client is None:
|
||||
_client = AsyncClient(config.MATRIX_HOMESERVER_URL)
|
||||
_client.user_id = config.MATRIX_BOT_USER_ID
|
||||
_client.access_token = config.MATRIX_BOT_ACCESS_TOKEN
|
||||
return _client
|
||||
|
||||
|
||||
async def close_client() -> None:
|
||||
global _client
|
||||
if _client is not None:
|
||||
await _client.close()
|
||||
_client = None
|
||||
|
||||
|
||||
async def send_thread_root(media_type: str) -> str:
|
||||
client = await get_client()
|
||||
label = THREAD_ROOT_MESSAGES[media_type]
|
||||
content = {
|
||||
"msgtype": "m.text",
|
||||
"body": label,
|
||||
}
|
||||
resp = await client.room_send(
|
||||
config.MATRIX_ARRIVALS_ROOM_ID,
|
||||
"m.room.message",
|
||||
content,
|
||||
)
|
||||
if isinstance(resp, RoomSendResponse):
|
||||
log.info("Created thread root for %s: %s", media_type, resp.event_id)
|
||||
return resp.event_id
|
||||
raise RuntimeError(f"Failed to create thread root for {media_type}: {resp}")
|
||||
|
||||
|
||||
async def post_to_thread(
|
||||
thread_root_event_id: str,
|
||||
plain_body: str,
|
||||
html_body: str,
|
||||
) -> str | None:
|
||||
client = await get_client()
|
||||
content = {
|
||||
"msgtype": "m.text",
|
||||
"format": "org.matrix.custom.html",
|
||||
"body": plain_body,
|
||||
"formatted_body": html_body,
|
||||
"m.relates_to": {
|
||||
"rel_type": "m.thread",
|
||||
"event_id": thread_root_event_id,
|
||||
},
|
||||
}
|
||||
try:
|
||||
resp = await client.room_send(
|
||||
config.MATRIX_ARRIVALS_ROOM_ID,
|
||||
"m.room.message",
|
||||
content,
|
||||
)
|
||||
if isinstance(resp, RoomSendResponse):
|
||||
return resp.event_id
|
||||
log.error("Matrix send failed: %s", resp)
|
||||
return None
|
||||
except Exception:
|
||||
log.exception("Matrix posting error")
|
||||
return None
|
||||
130
app/webhooks.py
Normal file
130
app/webhooks.py
Normal file
@@ -0,0 +1,130 @@
|
||||
import logging
|
||||
|
||||
from fastapi import APIRouter, Header, Request, Response
|
||||
|
||||
from app.config import config
|
||||
from app.database import get_thread_root, is_event_posted, record_posted_event
|
||||
from app.formatters import format_lidarr, format_radarr, format_sonarr
|
||||
from app.matrix import post_to_thread
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
router = APIRouter(prefix="/webhook")
|
||||
|
||||
SOURCE_MAP = {
|
||||
"radarr": {"media_type": "movies", "formatter": format_radarr, "id_field": "movie"},
|
||||
"sonarr": {"media_type": "shows", "formatter": format_sonarr, "id_field": "series"},
|
||||
"lidarr": {"media_type": "music", "formatter": format_lidarr, "id_field": "artist"},
|
||||
}
|
||||
|
||||
|
||||
def _validate_secret(header_value: str | None) -> bool:
|
||||
return header_value == config.WEBHOOK_SECRET
|
||||
|
||||
|
||||
def _extract_event_key(source: str, payload: dict) -> str | None:
|
||||
if source == "radarr":
|
||||
movie = payload.get("movie", {})
|
||||
movie_id = movie.get("id")
|
||||
if movie_id is not None:
|
||||
return f"radarr-{movie_id}"
|
||||
elif source == "sonarr":
|
||||
episodes = payload.get("episodes", [])
|
||||
if episodes:
|
||||
ep_id = episodes[0].get("id")
|
||||
if ep_id is not None:
|
||||
return f"sonarr-{ep_id}"
|
||||
elif source == "lidarr":
|
||||
album = payload.get("album", {})
|
||||
album_id = album.get("id")
|
||||
if album_id is not None:
|
||||
return f"lidarr-{album_id}"
|
||||
return None
|
||||
|
||||
|
||||
def _extract_title(source: str, payload: dict) -> str:
|
||||
if source == "radarr":
|
||||
return payload.get("movie", {}).get("title", "Unknown")
|
||||
elif source == "sonarr":
|
||||
return payload.get("series", {}).get("title", "Unknown")
|
||||
elif source == "lidarr":
|
||||
return payload.get("artist", {}).get("name", "Unknown")
|
||||
return "Unknown"
|
||||
|
||||
|
||||
async def _handle_webhook(source: str, payload: dict, secret: str | None) -> Response:
|
||||
if not _validate_secret(secret):
|
||||
return Response(status_code=401, content="Unauthorized")
|
||||
|
||||
event_type = payload.get("eventType", "")
|
||||
if event_type != "Download":
|
||||
log.debug("Ignoring %s event from %s", event_type, source)
|
||||
return Response(status_code=200, content="OK")
|
||||
|
||||
info = SOURCE_MAP[source]
|
||||
media_type = info["media_type"]
|
||||
formatter = info["formatter"]
|
||||
is_upgrade = payload.get("isUpgrade", False)
|
||||
|
||||
event_key = _extract_event_key(source, payload)
|
||||
if event_key:
|
||||
if await is_event_posted(event_key):
|
||||
log.debug("Duplicate event %s, skipping", event_key)
|
||||
return Response(status_code=200, content="OK")
|
||||
|
||||
try:
|
||||
plain, html = formatter(payload)
|
||||
except Exception:
|
||||
log.exception("Failed to format %s payload", source)
|
||||
return Response(status_code=200, content="OK")
|
||||
|
||||
thread_root_id = await get_thread_root(media_type)
|
||||
if not thread_root_id:
|
||||
log.error("No thread root for %s — cannot post", media_type)
|
||||
return Response(status_code=200, content="OK")
|
||||
|
||||
event_id = await post_to_thread(thread_root_id, plain, html)
|
||||
if event_id and event_key:
|
||||
title = _extract_title(source, payload)
|
||||
await record_posted_event(event_key, title, media_type, is_upgrade)
|
||||
|
||||
return Response(status_code=200, content="OK")
|
||||
|
||||
|
||||
@router.post("/radarr")
|
||||
async def webhook_radarr(
|
||||
request: Request,
|
||||
x_arr_webhook_secret: str | None = Header(None),
|
||||
):
|
||||
try:
|
||||
payload = await request.json()
|
||||
except Exception:
|
||||
log.exception("Failed to parse Radarr payload")
|
||||
return Response(status_code=200, content="OK")
|
||||
return await _handle_webhook("radarr", payload, x_arr_webhook_secret)
|
||||
|
||||
|
||||
@router.post("/sonarr")
|
||||
async def webhook_sonarr(
|
||||
request: Request,
|
||||
x_arr_webhook_secret: str | None = Header(None),
|
||||
):
|
||||
try:
|
||||
payload = await request.json()
|
||||
except Exception:
|
||||
log.exception("Failed to parse Sonarr payload")
|
||||
return Response(status_code=200, content="OK")
|
||||
return await _handle_webhook("sonarr", payload, x_arr_webhook_secret)
|
||||
|
||||
|
||||
@router.post("/lidarr")
|
||||
async def webhook_lidarr(
|
||||
request: Request,
|
||||
x_arr_webhook_secret: str | None = Header(None),
|
||||
):
|
||||
try:
|
||||
payload = await request.json()
|
||||
except Exception:
|
||||
log.exception("Failed to parse Lidarr payload")
|
||||
return Response(status_code=200, content="OK")
|
||||
return await _handle_webhook("lidarr", payload, x_arr_webhook_secret)
|
||||
Reference in New Issue
Block a user