fix type checks di notizie e social.
This commit is contained in:
@@ -33,10 +33,10 @@ class NewsAPIsTool(NewsWrapper, Toolkit):
|
||||
- NewsApiWrapper.
|
||||
- CryptoPanicWrapper.
|
||||
"""
|
||||
wrappers = [GoogleNewsWrapper, DuckDuckGoWrapper, NewsApiWrapper, CryptoPanicWrapper]
|
||||
self.wrapper_handler: WrapperHandler[NewsWrapper] = WrapperHandler.build_wrappers(wrappers)
|
||||
wrappers: list[type[NewsWrapper]] = [GoogleNewsWrapper, DuckDuckGoWrapper, NewsApiWrapper, CryptoPanicWrapper]
|
||||
self.wrapper_handler = WrapperHandler.build_wrappers(wrappers)
|
||||
|
||||
Toolkit.__init__(
|
||||
Toolkit.__init__( # type: ignore
|
||||
self,
|
||||
name="News APIs Toolkit",
|
||||
tools=[
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
import os
|
||||
from typing import Any
|
||||
import requests
|
||||
from enum import Enum
|
||||
from app.news.base import NewsWrapper, Article
|
||||
@@ -19,8 +20,8 @@ class CryptoPanicKind(Enum):
|
||||
MEDIA = "media"
|
||||
ALL = "all"
|
||||
|
||||
def extract_articles(response: dict) -> list[Article]:
|
||||
articles = []
|
||||
def extract_articles(response: dict[str, Any]) -> list[Article]:
|
||||
articles: list[Article] = []
|
||||
if 'results' in response:
|
||||
for item in response['results']:
|
||||
article = Article()
|
||||
@@ -52,7 +53,7 @@ class CryptoPanicWrapper(NewsWrapper):
|
||||
self.kind = CryptoPanicKind.NEWS
|
||||
|
||||
def get_base_params(self) -> dict[str, str]:
|
||||
params = {}
|
||||
params: dict[str, str] = {}
|
||||
params['public'] = 'true' # recommended for app and bots
|
||||
params['auth_token'] = self.api_key
|
||||
params['kind'] = self.kind.value
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
import json
|
||||
from typing import Any
|
||||
from agno.tools.duckduckgo import DuckDuckGoTools
|
||||
from app.news.base import Article, NewsWrapper
|
||||
|
||||
|
||||
def extract_article(result: dict) -> Article:
|
||||
def extract_article(result: dict[str, Any]) -> Article:
|
||||
article = Article()
|
||||
article.source = result.get("source", "")
|
||||
article.time = result.get("date", "")
|
||||
|
||||
@@ -1,8 +1,9 @@
|
||||
from gnews import GNews
|
||||
from typing import Any
|
||||
from gnews import GNews # type: ignore
|
||||
from app.news.base import Article, NewsWrapper
|
||||
|
||||
|
||||
def extract_article(result: dict) -> Article:
|
||||
def extract_article(result: dict[str, Any]) -> Article:
|
||||
article = Article()
|
||||
article.source = result.get("source", "")
|
||||
article.time = result.get("publishedAt", "")
|
||||
@@ -18,9 +19,9 @@ class GoogleNewsWrapper(NewsWrapper):
|
||||
|
||||
def get_top_headlines(self, limit: int = 100) -> list[Article]:
|
||||
gnews = GNews(language='en', max_results=limit, period='7d')
|
||||
results = gnews.get_top_news()
|
||||
results: list[dict[str, Any]] = gnews.get_top_news() # type: ignore
|
||||
|
||||
articles = []
|
||||
articles: list[Article] = []
|
||||
for result in results:
|
||||
article = extract_article(result)
|
||||
articles.append(article)
|
||||
@@ -28,9 +29,9 @@ class GoogleNewsWrapper(NewsWrapper):
|
||||
|
||||
def get_latest_news(self, query: str, limit: int = 100) -> list[Article]:
|
||||
gnews = GNews(language='en', max_results=limit, period='7d')
|
||||
results = gnews.get_news(query)
|
||||
results: list[dict[str, Any]] = gnews.get_news(query) # type: ignore
|
||||
|
||||
articles = []
|
||||
articles: list[Article] = []
|
||||
for result in results:
|
||||
article = extract_article(result)
|
||||
articles.append(article)
|
||||
|
||||
@@ -1,9 +1,10 @@
|
||||
import os
|
||||
import newsapi
|
||||
from typing import Any
|
||||
import newsapi # type: ignore
|
||||
from app.news.base import Article, NewsWrapper
|
||||
|
||||
|
||||
def extract_article(result: dict) -> Article:
|
||||
def extract_article(result: dict[str, Any]) -> Article:
|
||||
article = Article()
|
||||
article.source = result.get("source", {}).get("name", "")
|
||||
article.time = result.get("publishedAt", "")
|
||||
@@ -34,21 +35,20 @@ class NewsApiWrapper(NewsWrapper):
|
||||
|
||||
def get_top_headlines(self, limit: int = 100) -> list[Article]:
|
||||
pages, page_size = self.__calc_pages(limit, self.max_page_size)
|
||||
articles = []
|
||||
articles: list[Article] = []
|
||||
|
||||
for page in range(1, pages + 1):
|
||||
headlines = self.client.get_top_headlines(q="", category=self.category, language=self.language, page_size=page_size, page=page)
|
||||
results = [extract_article(article) for article in headlines.get("articles", [])]
|
||||
headlines: dict[str, Any] = self.client.get_top_headlines(q="", category=self.category, language=self.language, page_size=page_size, page=page) # type: ignore
|
||||
results = [extract_article(article) for article in headlines.get("articles", [])] # type: ignore
|
||||
articles.extend(results)
|
||||
return articles
|
||||
|
||||
def get_latest_news(self, query: str, limit: int = 100) -> list[Article]:
|
||||
pages, page_size = self.__calc_pages(limit, self.max_page_size)
|
||||
articles = []
|
||||
articles: list[Article] = []
|
||||
|
||||
for page in range(1, pages + 1):
|
||||
everything = self.client.get_everything(q=query, language=self.language, sort_by="publishedAt", page_size=page_size, page=page)
|
||||
results = [extract_article(article) for article in everything.get("articles", [])]
|
||||
everything: dict[str, Any] = self.client.get_everything(q=query, language=self.language, sort_by="publishedAt", page_size=page_size, page=page) # type: ignore
|
||||
results = [extract_article(article) for article in everything.get("articles", [])] # type: ignore
|
||||
articles.extend(results)
|
||||
return articles
|
||||
|
||||
|
||||
@@ -25,10 +25,10 @@ class SocialAPIsTool(SocialWrapper, Toolkit):
|
||||
- RedditWrapper.
|
||||
"""
|
||||
|
||||
wrappers = [RedditWrapper]
|
||||
self.wrapper_handler: WrapperHandler[SocialWrapper] = WrapperHandler.build_wrappers(wrappers)
|
||||
wrappers: list[type[SocialWrapper]] = [RedditWrapper]
|
||||
self.wrapper_handler = WrapperHandler.build_wrappers(wrappers)
|
||||
|
||||
Toolkit.__init__(
|
||||
Toolkit.__init__( # type: ignore
|
||||
self,
|
||||
name="Socials Toolkit",
|
||||
tools=[
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import os
|
||||
from praw import Reddit
|
||||
from praw.models import Submission, MoreComments
|
||||
from praw import Reddit # type: ignore
|
||||
from praw.models import Submission, MoreComments # type: ignore
|
||||
from app.social.base import SocialWrapper, SocialPost, SocialComment
|
||||
|
||||
|
||||
@@ -28,7 +28,7 @@ def extract_post(post: Submission) -> SocialPost:
|
||||
social.title = post.title
|
||||
social.description = post.selftext
|
||||
|
||||
for i, top_comment in enumerate(post.comments):
|
||||
for i, top_comment in enumerate(post.comments.list()):
|
||||
if i >= MAX_COMMENTS:
|
||||
break
|
||||
if isinstance(top_comment, MoreComments): #skip MoreComments objects
|
||||
|
||||
Reference in New Issue
Block a user