Add lasuite tools

Author: berrydenhartog

backend/app/clients/drive.py

@@ -0,0 +1,48 @@
+import httpx
+from app.models import Document
+from pydantic import TypeAdapter
+
+
+class DriveClient:
+    def __init__(self, base_url: str, token: str) -> None:
+        self.base_url = base_url
+        self.token = token
+        self.client = httpx.Client(
+            headers={"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}
+        )
+
+
+    def get_documents(
+        self, path: str = "api/v1.0/items/", page: int = 1, title: str | None = None, favorite: bool = False
+    ) -> list[Document]:
+        url = httpx.URL(f"{self.base_url}/{path}", params={"page": page})
+
+        if title:
+            url = url.copy_add_param("title", str(title))
+
+        response = self.client.request("GET", url)
+        if response.status_code != 200:
+            return TypeAdapter(list[Document]).validate_python([])
+
+        results = response.json().get("results", [])
+
+        if len(results) < 1:
+            return TypeAdapter(list[Document]).validate_python([])
+
+
+        workspace_id = results[0]['id']
+
+        item_url = f"{self.base_url}/api/v1.0/items/{workspace_id}/children/"
+
+        response = self.client.request("GET", item_url)
+        if response.status_code != 200:
+            return TypeAdapter(list[Document]).validate_python([])
+        
+        results = response.json().get("results", [])
+
+        print(results)
+
+
+        notes: list[Document] = TypeAdapter(list[Document]).validate_python(results)
+
+        return notes

backend/app/clients/meet.py

@@ -0,0 +1,33 @@
+import httpx
+from app.models import Meeting
+from pydantic import TypeAdapter
+
+
+class MeetClient:
+    def __init__(self, base_url: str, token: str) -> None:
+        self.base_url = base_url
+        self.token = token
+        self.client = httpx.Client(
+            headers={"Authorization": f"Bearer {self.token}", "Content-Type": "application/json"}
+        )
+
+    def get_meetings(
+        self, path: str = "api/v1.0/documents/", page: int = 1, title: str | None = None, favorite: bool = False
+    ) -> list[Meeting]:
+        url = httpx.URL(f"{self.base_url}/{path}", params={"page": page})
+
+        if title:
+            url = url.copy_add_param("title", str(title))
+
+        if favorite:
+            url = url.copy_add_param("favorite", str(favorite))
+
+        response = self.client.request("GET", url)
+        if response.status_code != 200:
+            return TypeAdapter(list[Meeting]).validate_python([])
+
+        results = response.json().get("results", [])
+
+        notes: list[Meeting] = TypeAdapter(list[Meeting]).validate_python(results)
+
+        return notes

backend/app/config.py

@@ -30,6 +30,10 @@ class Settings(BaseSettings):
     OCS_AUDIENCE: str = "files"
     DOCS_URL: str = "https://docs.la-suite.apps.digilab.network"
     DOCS_AUDIENCE: str = "docs"
+    DRIVE_URL: str = "http://localhost:3001"
+    DRIVE_AUDIENCE: str = "drive"
+    MEET_URL: str = "http://localhost:8085"
+    MEET_AUDIENCE: str = "meet"
     CALENDAR_URL: str = "https://files.la-suite.apps.digilab.network"
     CALENDAR_AUDIENCE: str = "files"
     TASK_URL: str = "https://files.la-suite.apps.digilab.network"

backend/app/models/document.py

@@ -0,0 +1,15 @@
+from datetime import datetime
+
+from pydantic import BaseModel, computed_field
+
+
+class Document(BaseModel):
+    title: str
+    url: str
+    mimetype: str
+    updated_at: str
+
+    @computed_field
+    @property
+    def updated_date(self) -> str:
+        return datetime.fromisoformat(self.updated_at).strftime("%d %b %Y")

backend/app/models/meeting.py

@@ -0,0 +1,5 @@
+from pydantic import BaseModel
+
+
+class Meeting(BaseModel):
+    title: str

backend/app/routes/drive.py

@@ -0,0 +1,28 @@
+import logging
+
+from fastapi import APIRouter, Request
+
+from app.clients.drive import DriveClient
+from app.config import settings
+from app.models import Document, User
+from app.token_exchange import exchange_token
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/drive", tags=["drive"])
+
+
+@router.get("/documents", response_model=list[Document])
+async def drive_documents(request: Request, title: str | None = None, favorite: bool = False) -> list[Document]:
+    user: User = request.state.user
+    access_token = user.access_token
+    new_token = await exchange_token(access_token, audience=settings.DRIVE_AUDIENCE)
+
+    if not new_token:
+        return []
+
+    client = DriveClient(base_url=settings.DRIVE_URL, token=new_token)
+
+    documents: list[Document] = client.get_documents(title=title, favorite=favorite)
+
+    return documents

backend/app/routes/meet.py

@@ -0,0 +1,28 @@
+import logging
+
+from fastapi import APIRouter, Request
+
+from app.clients.meet import MeetClient
+from app.config import settings
+from app.models import Meeting, User
+from app.token_exchange import exchange_token
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/meet", tags=["meet"])
+
+
+@router.get("/meeting", response_model=list[Meeting])
+async def meet_dmeetings(request: Request, title: str | None = None, favorite: bool = False) -> list[Meeting]:
+    user: User = request.state.user
+    access_token = user.access_token
+    new_token = await exchange_token(access_token, audience=settings.MEET_AUDIENCE)
+
+    if not new_token:
+        return []
+
+    client = MeetClient(base_url=settings.MEET_URL, token=new_token)
+
+    documents: list[Meeting] = client.get_documents(title=title, favorite=favorite)
+
+    return documents

backend/app/routes/rss.py

@@ -0,0 +1,15 @@
+
+import httpx
+from fastapi import APIRouter, Query, Response, status
+
+router = APIRouter(prefix="/rss", tags=["rss"])
+
+
+@router.get("/")
+async def rss_feed(url: str = Query(..., description="URL of the RSS feed")):
+
+    client = httpx.AsyncClient(timeout=5.0)
+    resp = await client.get(url)
+    resp.raise_for_status()
+    # Forward the raw RSS feed content and content-type header
+    return Response(content=resp.content, media_type=resp.headers.get("content-type", "application/xml"))