update

.env.example

@@ -0,0 +1,5 @@
+HOMESERVER_URL=https://matrix.org
+MATRIX_USER=@your_bot_user:matrix.org
+MATRIX_PASSWORD=your_password
+DEFAULT_ROOM_ID=!your_room_id:matrix.org
+# Optional: MATRIX_ACCESS_TOKEN=

.gitignore

@@ -0,0 +1,7 @@
+__pycache__/
+*.pyc
+.env
+venv/
+env/
+.DS_Store
+.pytest_cache/

Dockerfile

@@ -0,0 +1,14 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+# Expose the port
+EXPOSE 8000
+
+# Command to run the application
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

bot.py

@@ -0,0 +1,41 @@
+import asyncio
+from nio import AsyncClient, MatrixRoom, RoomMessageText
+from config import settings
+
+class MatrixBot:
+    def __init__(self):
+        self.client = AsyncClient(settings.homeserver_url, settings.matrix_user)
+        self.room_id = settings.default_room_id
+
+    async def login(self):
+        if settings.matrix_access_token:
+            self.client.access_token = settings.matrix_access_token
+            # We still need device_id maybe, but let's assume simple setup or login fallback
+        else:
+            resp = await self.client.login(settings.matrix_password)
+            if hasattr(resp, 'access_token'):
+                print(f"Logged in as {resp.user_id}")
+            else:
+                print(f"Failed to login: {resp}")
+                raise Exception("Login failed")
+
+    async def send_message(self, message: str, html_message: str = None, room_id: str = None):
+        target_room = room_id or self.room_id
+        
+        content = {
+            "msgtype": "m.text",
+            "body": message
+        }
+        
+        if html_message:
+            content["format"] = "org.matrix.custom.html"
+            content["formatted_body"] = html_message
+        
+        await self.client.room_send(
+            room_id=target_room,
+            message_type="m.room.message",
+            content=content
+        )
+
+    async def close(self):
+        await self.client.close()

config.py

@@ -0,0 +1,15 @@
+from pydantic_settings import BaseSettings
+
+class Settings(BaseSettings):
+    homeserver_url: str
+    matrix_user: str
+    matrix_password: str
+    default_room_id: str
+    
+    # Optional: access token if login is already done or for bot accounts
+    matrix_access_token: str | None = None
+
+    class Config:
+        env_file = ".env"
+
+settings = Settings()

docker-compose.yml

@@ -0,0 +1,8 @@
+services:
+  bot:
+    build: .
+    ports:
+      - "8000:8000"
+    env_file:
+      - .env
+    restart: unless-stopped

main.py

@@ -0,0 +1,92 @@
+from fastapi import FastAPI, HTTPException, BackgroundTasks
+from pydantic import BaseModel
+from bot import MatrixBot
+import asyncio
+from contextlib import asynccontextmanager
+
+bot = MatrixBot()
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Startup
+    await bot.login()
+    yield
+    # Shutdown
+    await bot.close()
+
+app = FastAPI(lifespan=lifespan)
+
+class Notification(BaseModel):
+    service_name: str
+    content: str
+    room_id: str | None = None
+    level: str = "info"
+
+class JellyfinPayload(BaseModel):
+    notification_type: str
+    item_type: str
+    name: str
+    series_name: str | None = None
+    season: int | None = None
+    episode: int | None = None
+    year: int | None = None
+    overview: str | None = None
+    room_id: str | None = None
+
+@app.post("/notify")
+async def send_notification(notification: Notification, background_tasks: BackgroundTasks):
+    """
+    Send a notification to a Matrix room.
+    """
+    try:
+        # Format the message to include the service name
+        # Plain text fallback
+        plain_message = f"[{notification.service_name}]\n{notification.content}"
+        # HTML message
+        html_message = f"<b>[{notification.service_name}]</b><br>{notification.content}"
+        
+        # We can send it in background to not block the API response
+        background_tasks.add_task(bot.send_message, plain_message, html_message, notification.room_id)
+        return {"status": "queued"}
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/jellyfin")
+async def receive_jellyfin_webhook(payload: JellyfinPayload, background_tasks: BackgroundTasks):
+    """
+    Receive webhook from Jellyfin and forward to Matrix.
+    """
+    try:
+        if payload.notification_type != "ItemAdded":
+            return {"status": "ignored", "reason": "Not an ItemAdded event"}
+        
+        # content construction
+        content = ""
+        if payload.item_type == "Movie":
+            content = f"New Movie: {payload.name}"
+            if payload.year:
+                content += f" ({payload.year})"
+            if payload.overview:
+                content += f"\n{payload.overview}"
+        elif payload.item_type == "Episode":
+            show = payload.series_name or "Unknown Series"
+            s = f"S{payload.season:02d}" if payload.season is not None else "S??"
+            e = f"E{payload.episode:02d}" if payload.episode is not None else "E??"
+            content = f"New Episode: {show} - {s}{e} - {payload.name}"
+        else:
+            # Fallback for Series, Season, etc.
+            content = f"New {payload.item_type}: {payload.name}"
+            
+        plain_message = f"[Jellyfin]\n{content}"
+        html_message = f"<b>[Jellyfin]</b><br>{content.replace(chr(10), '<br>')}"
+        
+        background_tasks.add_task(bot.send_message, plain_message, html_message, payload.room_id)
+        return {"status": "queued"}
+
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=str(e))
+
+
+@app.get("/health")
+async def health_check():
+    return {"status": "ok"}

requirements.txt

@@ -0,0 +1,7 @@
+matrix-nio
+fastapi
+uvicorn
+pydantic-settings
+python-dotenv
+pytest
+httpx

tests/test_api.py

@@ -0,0 +1,99 @@
+import sys
+import os
+sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from fastapi.testclient import TestClient
+from unittest.mock import MagicMock, patch, AsyncMock
+import pytest
+from main import app
+
+# We need to mock the bot before importing or using the app fully, 
+# although patching it contextually is better.
+# Because 'bot' is a global instance in main.py, we should patch it where it is used.
+
+@pytest.fixture
+def client():
+    # Mock the bot instance methods including login/close for lifespan
+    with patch("main.bot") as mock_bot:
+        # Use AsyncMock for async methods
+        
+        mock_bot.login = AsyncMock()
+        mock_bot.close = AsyncMock()
+        mock_bot.send_message = AsyncMock()
+        
+        with TestClient(app) as c:
+            yield c
+
+def test_health_check(client):
+    response = client.get("/health")
+    assert response.status_code == 200
+    assert response.json() == {"status": "ok"}
+
+def test_notify_success(client):
+    # We need to verify that bot.send_message was called.
+    
+    with patch("main.bot.send_message", new_callable=AsyncMock) as mock_send:
+        # Updated payload
+        payload = {
+            "service_name": "TestService",
+            "content": "System is down"
+        }
+        response = client.post("/notify", json=payload)
+        assert response.status_code == 200
+        assert response.json() == {"status": "queued"}
+        
+        # Check formatted message
+        mock_send.assert_called_once()
+        args, kwargs = mock_send.call_args
+        # args[0] is plain_message, args[1] is html_message, args[2] is room_id
+        assert args[0] == "[TestService]\nSystem is down"
+        assert args[1] == "<b>[TestService]</b><br>System is down"
+
+def test_jellyfin_movie(client):
+    with patch("main.bot.send_message", new_callable=AsyncMock) as mock_send:
+        payload = {
+            "notification_type": "ItemAdded",
+            "item_type": "Movie",
+            "name": "The Matrix",
+            "year": 1999,
+            "overview": "A computer hacker..."
+        }
+        response = client.post("/jellyfin", json=payload)
+        assert response.status_code == 200
+        
+        mock_send.assert_called_once()
+        args, _ = mock_send.call_args
+        assert "[Jellyfin]\nNew Movie: The Matrix (1999)\nA computer hacker..." in args[0]
+        assert "<b>[Jellyfin]</b><br>New Movie: The Matrix (1999)<br>A computer hacker..." in args[1]
+
+def test_jellyfin_episode(client):
+    with patch("main.bot.send_message", new_callable=AsyncMock) as mock_send:
+        payload = {
+            "notification_type": "ItemAdded",
+            "item_type": "Episode",
+            "name": "Pilot",
+            "series_name": "Lost",
+            "season": 1,
+            "episode": 1
+        }
+        response = client.post("/jellyfin", json=payload)
+        assert response.status_code == 200
+        
+        mock_send.assert_called_once()
+        args, _ = mock_send.call_args
+        assert "[Jellyfin]\nNew Episode: Lost - S01E01 - Pilot" in args[0]
+
+def test_notify_custom_room(client):
+    with patch("main.bot.send_message", new_callable=AsyncMock) as mock_send:
+        payload = {
+            "service_name": "TestService",
+            "content": "msg",
+            "room_id": "!custom:room"
+        }
+        response = client.post("/notify", json=payload)
+        assert response.status_code == 200
+        mock_send.assert_called_once_with("[TestService]\nmsg", "<b>[TestService]</b><br>msg", "!custom:room")
+
+def test_notify_validation_error(client):
+    response = client.post("/notify", json={"service_name": "JustServiceNoContent"}) 
+    assert response.status_code == 422