Source code for langchain_community.document_loaders.joplin

import json
import urllib
from datetime import datetime
from typing import Iterator, List, Optional

from langchain_core.documents import Document
from langchain_core.utils import get_from_env

from langchain_community.document_loaders.base import BaseLoader

LINK_NOTE_TEMPLATE = "joplin://x-callback-url/openNote?id={id}"


class JoplinLoader(BaseLoader):
    """Load notes from `Joplin`.

    In order to use this loader, you need to have Joplin running with the
    Web Clipper enabled (look for "Web Clipper" in the app settings).

    To get the access token, you need to go to the Web Clipper options and
    under "Advanced Options" you will find the access token.

    You can find more information about the Web Clipper service here:
    https://joplinapp.org/clipper/
    """

    def __init__(
        self,
        access_token: Optional[str] = None,
        port: int = 41184,
        host: str = "localhost",
    ) -> None:
        """

        Args:
            access_token: The access token to use.
            port: The port where the Web Clipper service is running. Default is 41184.
            host: The host where the Web Clipper service is running.
                Default is localhost.
        """
        access_token = access_token or get_from_env(
            "access_token", "JOPLIN_ACCESS_TOKEN"
        )
        base_url = f"http://{host}:{port}"
        self._get_note_url = (
            f"{base_url}/notes?token={access_token}"
            f"&fields=id,parent_id,title,body,created_time,updated_time&page={{page}}"
        )
        self._get_folder_url = (
            f"{base_url}/folders/{{id}}?token={access_token}&fields=title"
        )
        self._get_tag_url = (
            f"{base_url}/notes/{{id}}/tags?token={access_token}&fields=title"
        )

    def _get_notes(self) -> Iterator[Document]:
        has_more = True
        page = 1
        while has_more:
            req_note = urllib.request.Request(self._get_note_url.format(page=page))
            with urllib.request.urlopen(req_note) as response:
                json_data = json.loads(response.read().decode())
                for note in json_data["items"]:
                    metadata = {
                        "source": LINK_NOTE_TEMPLATE.format(id=note["id"]),
                        "folder": self._get_folder(note["parent_id"]),
                        "tags": self._get_tags(note["id"]),
                        "title": note["title"],
                        "created_time": self._convert_date(note["created_time"]),
                        "updated_time": self._convert_date(note["updated_time"]),
                    }
                    yield Document(page_content=note["body"], metadata=metadata)

                has_more = json_data["has_more"]
                page += 1

    def _get_folder(self, folder_id: str) -> str:
        req_folder = urllib.request.Request(self._get_folder_url.format(id=folder_id))
        with urllib.request.urlopen(req_folder) as response:
            json_data = json.loads(response.read().decode())
            return json_data["title"]

    def _get_tags(self, note_id: str) -> List[str]:
        req_tag = urllib.request.Request(self._get_tag_url.format(id=note_id))
        with urllib.request.urlopen(req_tag) as response:
            json_data = json.loads(response.read().decode())
            return [tag["title"] for tag in json_data["items"]]

    def _convert_date(self, date: int) -> str:
        return datetime.fromtimestamp(date / 1000).strftime("%Y-%m-%d %H:%M:%S")

    def lazy_load(self) -> Iterator[Document]:
        yield from self._get_notes()