跳到内容

Feedly RSS

FeedlyRssReader #

继承自: BaseReader

Feedly RSS 读取器。

从 Feedly RSS 读取器获取条目

使用 Feedly 官方 python-api-client: https://github.com/feedly/python-api-client

源代码位于 llama-index-integrations/readers/llama-index-readers-feedly-rss/llama_index/readers/feedly_rss/base.py 
10
11
12
13
14
15
16
17
18
19
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
class FeedlyRssReader(BaseReader):
    """
    Feedly Rss Reader.

    Get entries from Feedly Rss Reader

    Uses Feedly Official python-api-client: https://github.com/feedly/python-api-client
    """

    def __init__(self, bearer_token: str) -> None:
        """Initialize with parameters."""
        super().__init__()
        self.bearer_token = bearer_token

    def setup_auth(
        self, directory: Path = Path.home() / ".config/feedly", overwrite: bool = False
    ):
        """
        Modified from python-api-client/feedly/api_client/utils.py
        Instead promopting for user input, we take the token as an argument.
        """
        directory.mkdir(exist_ok=True, parents=True)

        auth_file = directory / "access.token"

        if not auth_file.exists() or overwrite:
            auth = self.bearer_token
            auth_file.write_text(auth.strip())

    def load_data(self, category_name, max_count=100):
        """Get the entries from a feedly category."""
        from feedly.api_client.session import FeedlySession
        from feedly.api_client.stream import StreamOptions

        self.setup_auth(overwrite=True)
        sess = FeedlySession()
        category = sess.user.user_categories.get(category_name)

        documents = []
        for article in category.stream_contents(
            options=StreamOptions(max_count=max_count)
        ):
            # doc for available fields: https://developer.feedly.com/v3/streams/
            entry = {
                "title": article["title"],
                "published": article["published"],
                "summary": article["summary"],
                "author": article["author"],
                "content": article["content"],
                "keywords": article["keywords"],
                "commonTopics": article["commonTopics"],
            }

            text = json.dumps(entry, ensure_ascii=False)

            documents.append(Document(text=text))
        return documents

setup_auth (设置认证) #

setup_auth(directory: Path = home() / '.config/feedly', overwrite: bool = False)

修改自 python-api-client/feedly/api_client/utils.py。不提示用户输入，我们将 token 作为参数。

源代码位于 llama-index-integrations/readers/llama-index-readers-feedly-rss/llama_index/readers/feedly_rss/base.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def setup_auth(
    self, directory: Path = Path.home() / ".config/feedly", overwrite: bool = False
):
    """
    Modified from python-api-client/feedly/api_client/utils.py
    Instead promopting for user input, we take the token as an argument.
    """
    directory.mkdir(exist_ok=True, parents=True)

    auth_file = directory / "access.token"

    if not auth_file.exists() or overwrite:
        auth = self.bearer_token
        auth_file.write_text(auth.strip())

load_data (加载数据) #

load_data(category_name, max_count=100)

从 Feedly 类别获取条目。

源代码位于 llama-index-integrations/readers/llama-index-readers-feedly-rss/llama_index/readers/feedly_rss/base.py 
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
def load_data(self, category_name, max_count=100):
    """Get the entries from a feedly category."""
    from feedly.api_client.session import FeedlySession
    from feedly.api_client.stream import StreamOptions

    self.setup_auth(overwrite=True)
    sess = FeedlySession()
    category = sess.user.user_categories.get(category_name)

    documents = []
    for article in category.stream_contents(
        options=StreamOptions(max_count=max_count)
    ):
        # doc for available fields: https://developer.feedly.com/v3/streams/
        entry = {
            "title": article["title"],
            "published": article["published"],
            "summary": article["summary"],
            "author": article["author"],
            "content": article["content"],
            "keywords": article["keywords"],
            "commonTopics": article["commonTopics"],
        }

        text = json.dumps(entry, ensure_ascii=False)

        documents.append(Document(text=text))
    return documents