Bases: BaseReader
从Joplin获取笔记的阅读器。
要使用此阅读器，您需要运行Joplin并启用Web Clipper（在应用程序设置中查找“Web Clipper”）。
要获取访问令牌，您需要转到Web Clipper选项，在“高级选项”下找到访问令牌。您可以将其作为参数提供，或设置JOPLIN_ACCESS_TOKEN环境变量。
您可以在此处找到有关Web Clipper服务的更多信息：
https://joplinapp.org/clipper/
Source code in llama_index/readers/joplin/base.py
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118 | class JoplinReader(BaseReader):
"""从Joplin获取笔记的阅读器。
要使用此阅读器，您需要运行Joplin并启用Web Clipper（在应用程序设置中查找“Web Clipper”）。
要获取访问令牌，您需要转到Web Clipper选项，在“高级选项”下找到访问令牌。您可以将其作为参数提供，或设置JOPLIN_ACCESS_TOKEN环境变量。
您可以在此处找到有关Web Clipper服务的更多信息：
https://joplinapp.org/clipper/"""
def __init__(
self,
access_token: Optional[str] = None,
parse_markdown: bool = True,
port: int = 41184,
host: str = "localhost",
) -> None:
"""初始化一个新的JoplinReader实例。
Args:
access_token (Optional[str]): Joplin Web Clipper服务的访问令牌。
如果未提供，则使用JOPLIN_ACCESS_TOKEN环境变量。默认为None。
parse_markdown (bool): 是否使用MarkdownReader解析笔记的markdown内容。默认为True。
port (int): Joplin Web Clipper服务运行的端口。默认为41184。
host (str): Joplin Web Clipper服务运行的主机。默认为"localhost"。
"""
self.parse_markdown = parse_markdown
if parse_markdown:
self.parser = MarkdownReader()
access_token = access_token or self._get_token_from_env()
base_url = f"http://{host}:{port}"
self._get_note_url = (
f"{base_url}/notes?token={access_token}"
"&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_token_from_env(self) -> str:
if "JOPLIN_ACCESS_TOKEN" in os.environ:
return os.environ["JOPLIN_ACCESS_TOKEN"]
else:
raise ValueError(
"You need to provide an access token to use the Joplin reader. You may"
" provide it as an argument or set the JOPLIN_ACCESS_TOKEN environment"
" variable."
)
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"]),
}
if self.parse_markdown:
yield from self.parser.load_data(
None, content=note["body"], extra_info=metadata
)
else:
yield Document(text=note["body"], extra_info=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 ",".join([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()
def load_data(self) -> List[Document]:
return list(self.lazy_load())
|