1
2
3
4
5
6
7
8
9
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
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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
|
"""Biztoc World News Model."""
# pylint: disable=unused-argument
from typing import Any, Dict, List, Optional
from warnings import warn
from openbb_core.app.model.abstract.error import OpenBBError
from openbb_core.provider.abstract.fetcher import Fetcher
from openbb_core.provider.standard_models.world_news import (
WorldNewsData,
WorldNewsQueryParams,
)
from openbb_core.provider.utils.helpers import make_request
from pandas import to_datetime
from pydantic import Field, field_validator
class BiztocWorldNewsQueryParams(WorldNewsQueryParams):
"""Biztoc World News Query."""
term: Optional[str] = Field(
description="Search term to filter articles by. This overrides all other filters.",
default=None,
)
source: Optional[str] = Field(
description="Filter by a specific publisher. Only valid when filter is set to source.",
default=None,
)
class BiztocWorldNewsData(WorldNewsData):
"""Biztoc World News Data."""
__alias_dict__ = {
"date": "published",
"text": "body",
"images": "img",
}
images: Optional[Dict[str, str]] = Field(
description="Images for the article.", alias="images", default=None
)
tags: Optional[List[str]] = Field(description="Tags for the article.", default=None)
score: Optional[float] = Field(
description="Search relevance score for the article.", default=None
)
@field_validator("date", "updated", mode="before", check_fields=False)
@classmethod
def date_validate(cls, v):
"""Return formatted datetime."""
return to_datetime(v).strftime("%Y-%m-%d %H:%M:%S")
@field_validator("title")
@classmethod
def title_validate(cls, v):
"""Strip empty title text."""
return v.strip() if v else None
class BiztocWorldNewsFetcher(
Fetcher[
BiztocWorldNewsQueryParams,
List[BiztocWorldNewsData],
]
):
"""Transform the query, extract and transform the data from the Biztoc endpoints."""
@staticmethod
def transform_query(params: Dict[str, Any]) -> BiztocWorldNewsQueryParams:
"""Transform the query."""
if params.get("start_date") or params.get("end_date"):
warn("start_date and end_date are not supported for this endpoint.")
return BiztocWorldNewsQueryParams(**params)
@staticmethod
def extract_data(
query: BiztocWorldNewsQueryParams, # pylint: disable=unused-argument
credentials: Optional[Dict[str, str]],
**kwargs: Any,
) -> List[Dict]:
"""Extract the data from the Biztoc endpoint."""
api_key = credentials.get("biztoc_api_key") if credentials else ""
headers = {
"X-RapidAPI-Key": f"{api_key}",
"X-RapidAPI-Host": "biztoc.p.rapidapi.com",
"Accept": "application/json",
"Accept-Encoding": "gzip",
}
base_url = "https://biztoc.p.rapidapi.com/"
url = ""
response: List = []
if query.term:
query.term = query.term.replace(" ", "%20")
url = base_url + f"search?q={query.term}"
response = make_request(url, headers=headers).json()
if query.source is not None:
sources_response = make_request(
"https://biztoc.p.rapidapi.com/sources",
headers=headers,
).json()
sources = [source["id"] for source in sources_response]
if query.source.lower() not in sources:
raise OpenBBError(
f"{query.source} not a valid source. Valid sources: {sources}"
)
url = base_url + f"news/source/{query.source.lower()}"
response = make_request(url, headers=headers).json()
else:
url1 = base_url + "news/latest"
url2 = base_url + "news/topics"
response = make_request(url1, headers=headers).json()
response2 = make_request(url2, headers=headers).json()
if response2:
for topic in response2:
stories = topic.get("stories", [])
if stories:
response.extend(
{
"text" if k == "body_preview" else k: v
for k, v in story.items()
}
for story in stories
)
return response
@staticmethod
def transform_data(
query: BiztocWorldNewsQueryParams, data: List[Dict], **kwargs: Any
) -> List[BiztocWorldNewsData]:
"""Transform the data to the standard format."""
results: List[BiztocWorldNewsData] = []
for item in data:
item.pop("id", None)
item.pop("uid", None)
item.pop("body_preview", None)
item.pop("site", None)
item.pop("domain", None)
results.append(BiztocWorldNewsData.model_validate(item))
return results
|
