", html_as_string) html_as_string = re.sub(r"(?<=(ipt|div)>)\n(?![\W<])", "\n
", html_as_string) html_as_string = re.sub(r"(?])\n(?=<[a-z0-9=_'\"]*>)", "
\n", html_as_string) - doc: HtmlElement = fromstring(html_as_string) # type: ignore + doc: HtmlElement = fromstring(html_as_string) return extract_article_body_with_selector( doc=doc, paragraph_selector=self._paragraph_selector, diff --git a/src/fundus/publishers/fr/le_monde.py b/src/fundus/publishers/fr/le_monde.py index fd8f3a99d..a22eb7df0 100644 --- a/src/fundus/publishers/fr/le_monde.py +++ b/src/fundus/publishers/fr/le_monde.py @@ -35,7 +35,7 @@ def title(self) -> Optional[str]: @attribute def topics(self) -> List[str]: - return self.precomputed.ld.bf_search("keywords") # type: ignore + return self.precomputed.ld.bf_search("keywords") @attribute def publishing_date(self) -> Optional[datetime.datetime]: diff --git a/src/fundus/publishers/ind/times_of_india.py b/src/fundus/publishers/ind/times_of_india.py index f0343eaa7..3d6ebc75c 100644 --- a/src/fundus/publishers/ind/times_of_india.py +++ b/src/fundus/publishers/ind/times_of_india.py @@ -41,7 +41,7 @@ def body(self) -> Optional[ArticleBody]: r"", html_as_string
)
return extract_article_body_with_selector(
- fromstring(html_as_string), # type: ignore
+ fromstring(html_as_string),
summary_selector=self._summary_selector,
paragraph_selector=self._paragraph_selector,
subheadline_selector=self._subheadline_selector,
diff --git a/src/fundus/scraping/article.py b/src/fundus/scraping/article.py
index a64502bc0..afdc5548a 100644
--- a/src/fundus/scraping/article.py
+++ b/src/fundus/scraping/article.py
@@ -130,12 +130,12 @@ def to_json(self, *attributes: str) -> Dict[str, JSONVal]:
def serialize(v: Any) -> JSONVal:
if hasattr(v, "serialize"):
- return v.serialize() # type: ignore[no-any-return]
+ return v.serialize()
elif isinstance(v, datetime):
return str(v)
elif not is_jsonable(v):
raise TypeError(f"Attribute {attribute!r} of type {type(v)!r} is not JSON serializable")
- return v # type: ignore[no-any-return]
+ return v
serialization: Dict[str, JSONVal] = {}
for attribute in attributes:
diff --git a/src/fundus/scraping/crawler.py b/src/fundus/scraping/crawler.py
index 1b022619c..b94276b80 100644
--- a/src/fundus/scraping/crawler.py
+++ b/src/fundus/scraping/crawler.py
@@ -85,7 +85,7 @@ def tqdm(self, *args, **kwargs) -> tqdm:
@contextlib.contextmanager
-def get_proxy_tqdm(*args, **kwargs) -> tqdm:
+def get_proxy_tqdm(*args, **kwargs) -> Iterator[tqdm]:
"""
This functions returns a proxy to a tqdm instance. Init args are the same as for any other tqdm instance.
:param args: tqdm args
@@ -120,7 +120,7 @@ def __call__(self, *args: _P.args, **kwargs: _P.kwargs) -> _T:
return self._deserialize()(*args, **kwargs)
-def get_execution_context():
+def get_execution_context() -> Tuple[str, int]:
"""
Determines whether the current execution context is in a thread or process.
Returns:
@@ -129,10 +129,10 @@ def get_execution_context():
"""
if multiprocessing.current_process().name != "MainProcess":
process = multiprocessing.current_process()
- return process.name, process.ident
+ return process.name, process.ident or 0
else:
thread = current_thread()
- return thread.name, thread.ident
+ return thread.name, thread.ident or 0
def publisher_context_wrapper(func: Callable[[Publisher], None]) -> Callable[[Publisher], None]:
@@ -414,9 +414,10 @@ def build_extraction_filter() -> Optional[ExtractionFilter]:
callback: Optional[Callable[[], None]]
if isinstance(self, CCNewsCrawler) and self.processes > 0:
- def callback() -> None:
+ def _stop_callback() -> None:
__EVENTS__.set_event("stop", "main-thread")
+ callback = _stop_callback
else:
callback = None
@@ -579,6 +580,7 @@ def _single_crawl(
def _threaded_crawl(
self, publishers: Tuple[Publisher, ...], article_task: Callable[[Publisher], Iterator[Article]]
) -> Iterator[Article]:
+
@contextlib.contextmanager
def _manage_pool(*args, **kwargs) -> Iterator[ThreadPool]:
managed_pool = ThreadPool(*args, **kwargs)
@@ -731,32 +733,24 @@ def _parallel_crawl(
# As one could think, because we're downloading a bunch of files, this task is IO-bound, but it is actually
# process-bound. The reason is that we stream the data and process it on the fly rather than downloading all
# files and processing them afterward. Therefore, we utilize multiprocessing here instead of multithreading.
- try:
- with Manager() as manager, Pool(
- processes=min(self.processes, len(warc_paths)),
- initializer=initializer,
- ) as pool:
- result_queue: Queue[Union[Article, Exception]] = manager.Queue(maxsize=1000)
+ with Manager() as manager, Pool(
+ processes=min(self.processes, len(warc_paths)),
+ initializer=initializer,
+ ) as pool:
+ result_queue: Queue[Union[Article, Exception]] = manager.Queue(maxsize=1000)
- # Because multiprocessing.Pool does not support iterators as targets,
- # we wrap the article_task to write the articles to a queue instead of returning them directly.
- wrapped_article_task: Callable[[str], None] = queue_wrapper(result_queue, article_task)
+ # Because multiprocessing.Pool does not support iterators as targets,
+ # we wrap the article_task to write the articles to a queue instead of returning them directly.
+ wrapped_article_task: Callable[[str], None] = queue_wrapper(result_queue, article_task)
- # To avoid 503 errors we spread tasks to not start all at once
- spread_article_task = random_sleep(wrapped_article_task, (0, 3))
+ # To avoid 503 errors we spread tasks to not start all at once
+ spread_article_task = random_sleep(wrapped_article_task, (0, 3))
- # To avoid restricting the article_task to use only pickleable objects, we serialize it using dill.
- serialized_article_task = dill_wrapper(spread_article_task)
+ # To avoid restricting the article_task to use only pickleable objects, we serialize it using dill.
+ serialized_article_task = dill_wrapper(spread_article_task)
- # Finally, we build an iterator around the queue, exhausting the queue until the pool is finished.
- yield from pool_queue_iter(pool.map_async(serialized_article_task, warc_paths), result_queue)
- finally:
- logger.debug(f"Shutting down {type(self).__name__!r} ...")
- logger.debug("Joining manager ...")
- manager.join()
- logger.debug("Joining pool ...")
- pool.join()
- logger.debug("Shutdown done")
+ # Finally, we build an iterator around the queue, exhausting the queue until the pool is finished.
+ yield from pool_queue_iter(pool.map_async(serialized_article_task, warc_paths), result_queue)
def _get_warc_paths(self) -> List[str]:
# Date regex examples: https://regex101.com/r/yDX3G6/1
@@ -790,11 +784,8 @@ def load_paths(url: str) -> List[str]:
# use two threads per process, default two threads per core
max_number_of_threads = self.processes * 2
- try:
- with ThreadPool(processes=min(len(urls), max_number_of_threads)) as pool:
- nested_warc_paths = pool.map(random_sleep(load_paths, (0, 3)), urls)
- finally:
- pool.join()
+ with ThreadPoolExecutor(max_workers=min(len(urls), max_number_of_threads)) as pool:
+ nested_warc_paths = list(pool.map(random_sleep(load_paths, (0, 3)), urls))
warc_paths: Iterator[str] = more_itertools.flatten(nested_warc_paths)
diff --git a/src/fundus/scraping/filter.py b/src/fundus/scraping/filter.py
index 35c6f22e2..742ca44d8 100644
--- a/src/fundus/scraping/filter.py
+++ b/src/fundus/scraping/filter.py
@@ -150,7 +150,7 @@ def __init__(self, *required_attributes: str, eval_booleans: bool = True) -> Non
"""
self.required_attributes = set(required_attributes)
# somehow mypy does not recognize bool as callable :(
- self._eval: Callable[[Any], bool] = bool if eval_booleans else _guarded_bool # type: ignore[assignment]
+ self._eval: Callable[[Any], bool] = bool if eval_booleans else _guarded_bool
def __call__(self, extraction: Dict[str, Any]) -> FilterResultWithMissingAttributes:
missing_attributes = [
diff --git a/src/fundus/scraping/html.py b/src/fundus/scraping/html.py
index 2afa23918..a743318cc 100644
--- a/src/fundus/scraping/html.py
+++ b/src/fundus/scraping/html.py
@@ -1,7 +1,7 @@
import time
from dataclasses import dataclass
from datetime import datetime
-from typing import Callable, Dict, Iterable, Iterator, List, Optional, Protocol
+from typing import BinaryIO, Callable, Dict, Iterable, Iterator, List, Optional, Protocol, cast
from urllib.parse import urlparse
import chardet
@@ -171,9 +171,11 @@ def __init__(
f"Overwriting existing delay."
)
- def delay() -> float:
+ def _crawl_delay() -> float:
return robots_delay
+ delay = _crawl_delay
+
self.clock = _Clock(delay=delay, sleep=self._sleep)
@property
@@ -304,8 +306,10 @@ def extract_content(record: WarcRecord) -> Optional[str]:
warc_body: bytes = record.reader.read()
try:
- return str(warc_body, encoding=record.http_charset) # type: ignore[arg-type]
- except (UnicodeDecodeError, TypeError):
+ if record.http_charset is None:
+ raise UnicodeDecodeError("unknown", warc_body, 0, 1, "no charset")
+ return str(warc_body, encoding=record.http_charset)
+ except UnicodeDecodeError:
encoding: Optional[str] = chardet.detect(warc_body)["encoding"]
if encoding is not None:
@@ -333,7 +337,9 @@ def extract_content(record: WarcRecord) -> Optional[str]:
response = session.get(self.warc_path, stream=True, headers=self.headers)
response.raise_for_status()
- for warc_record in ArchiveIterator(response.raw, record_types=WarcRecordType.response, verify_digests=True):
+ for warc_record in ArchiveIterator(
+ cast(BinaryIO, response.raw), record_types=WarcRecordType.response, verify_digests=True
+ ):
if not warc_record.record_date:
continue
diff --git a/src/fundus/scraping/scraper.py b/src/fundus/scraping/scraper.py
index fc95e3f97..727b3a22c 100644
--- a/src/fundus/scraping/scraper.py
+++ b/src/fundus/scraping/scraper.py
@@ -1,3 +1,4 @@
+import random
from typing import Dict, Iterator, List, Literal, Optional, Type
import more_itertools
@@ -34,6 +35,9 @@ def scrape(
for html in source.fetch(url_filter=url_filter):
parser = self.parser_mapping[html.source_info.publisher]
+ if random.uniform(0, 1) > 0.9:
+ raise Exception("TEST")
+
try:
extraction = parser(html.crawl_date).parse(html.content, error_handling)
diff --git a/src/fundus/scraping/url.py b/src/fundus/scraping/url.py
index 822354e48..0786ec7d2 100644
--- a/src/fundus/scraping/url.py
+++ b/src/fundus/scraping/url.py
@@ -19,6 +19,7 @@
from urllib.parse import unquote
import feedparser
+import lxml.etree
import lxml.html
import validators
from lxml.etree import XMLParser, XPath
@@ -159,9 +160,9 @@ def __iter__(self) -> Iterator[str]:
logger.warning(f"Warning! Couldn't parse rss feed {self.url!r} because of {exception}")
return
else:
- urls = filter(bool, (entry.get("link") for entry in rss_feed["entries"]))
- for url in urls:
- yield clean_url(url)
+ for entry in rss_feed["entries"]:
+ if isinstance(url := entry.get("link"), str):
+ yield clean_url(url)
@dataclass
@@ -206,7 +207,7 @@ def yield_recursive(sitemap_url: str) -> Iterator[str]:
tree = lxml.etree.fromstring(content, parser=self._parser)
if tree is None:
# in case we somehow end up with non xml content
- logger.warning(f"Warning! Couldn't parse sitemap {sitemap_url!r}") # type: ignore[unreachable]
+ logger.warning(f"Warning! Couldn't parse sitemap {sitemap_url!r}")
return
urls = [node.text for node in self._url_selector(tree)]
if urls:
diff --git a/src/fundus/utils/regex.py b/src/fundus/utils/regex.py
index c3d563238..a79894852 100644
--- a/src/fundus/utils/regex.py
+++ b/src/fundus/utils/regex.py
@@ -1,5 +1,5 @@
import re
-from typing import Callable, Dict, Literal, Optional, Pattern, TypeVar, Union, overload
+from typing import Any, Callable, Dict, Literal, Optional, Pattern, TypeVar, overload
_T = TypeVar("_T")
@@ -19,12 +19,12 @@ def _get_match_dict(pattern: Pattern[str], string: str) -> Dict[str, str]: ...
@overload
-def _get_match_dict(pattern: Pattern[str], string: str, keep_none: Literal[True]) -> Dict[str, Optional[str]]: ...
+def _get_match_dict(pattern: Pattern[str], string: str, *, keep_none: Literal[True]) -> Dict[str, Optional[str]]: ...
-def _get_match_dict( # type: ignore[misc]
+def _get_match_dict(
pattern: Pattern[str], string: str, conversion: Optional[Callable[[str], _T]] = None, keep_none: bool = False
-) -> Dict[str, Union[str, _T, None]]:
+) -> Any:
matches = {}
for match in re.finditer(pattern, string):
match_dict = match.groupdict()
diff --git a/src/fundus/utils/serialization.py b/src/fundus/utils/serialization.py
index 0b15da0a4..b954e5328 100644
--- a/src/fundus/utils/serialization.py
+++ b/src/fundus/utils/serialization.py
@@ -58,7 +58,7 @@ class DataclassSerializationMixin:
def serialize(self) -> Dict[str, JSONVal]:
if not is_dataclass(self):
raise TypeError(f"{type(self).__name__!r} is not a dataclass")
- return asdict(self) # type: ignore[arg-type]
+ return asdict(self)
@classmethod
def deserialize(cls: Type[_M], serialized: Dict[str, JSONVal]) -> _M:
@@ -72,7 +72,7 @@ def deserialize(cls: Type[_M], serialized: Dict[str, JSONVal]) -> _M:
for field in fields(cls):
serialized[field.name] = _inner_deserialize(serialized[field.name], annotations[field.name])
- return cls(**serialized) # type: ignore[return-value]
+ return cls(**serialized)
def _inner_deserialize(data, cls):
diff --git a/src/fundus/utils/timeout.py b/src/fundus/utils/timeout.py
index 92b6e7d48..2194f19c8 100644
--- a/src/fundus/utils/timeout.py
+++ b/src/fundus/utils/timeout.py
@@ -27,8 +27,8 @@ def __init__(
seconds: float,
func: Callable[P, None],
interval: float = 0.1,
- args: P.args = tuple(),
- kwargs: P.kwargs = None,
+ *args: P.args,
+ **kwargs: P.kwargs,
) -> None:
"""Resettable timer executing