diff --git a/mediacloud/api.py b/mediacloud/api.py index 0c20282..abb42de 100644 --- a/mediacloud/api.py +++ b/mediacloud/api.py @@ -1,8 +1,8 @@ import datetime as dt import importlib.metadata import logging -from typing import Any, Dict, List, Optional, Union import warnings +from typing import Any, Dict, List, Optional, Union import requests @@ -62,11 +62,10 @@ def _query(self, endpoint: str, params: Optional[Dict] = None, method: str = 'GE r = self._session.post(endpoint_url, json=params, timeout=self.TIMEOUT_SECS) else: raise RuntimeError(f"Unsupported method of '{method}'") - if r.status_code != 200: + if r.status_code != 200: raise mediacloud.error.APIResponseError(r, params, r.json()) - - return r.json() + return r.json() class DirectoryApi(BaseApi): @@ -148,7 +147,6 @@ def _prep_default_params(self, query: str, start_date: dt.date, end_date: dt.dat end_date = end_date.date() warnings.warn("end_date was passed as datetime, but expected as date, and has been recast") - params: Dict[Any, Any] = dict(start=start_date.isoformat(), end=end_date.isoformat(), q=query, platform=(platform or self.PROVIDER)) if len(source_ids): @@ -172,6 +170,13 @@ def story_count_over_time(self, query: str, start_date: dt.date, end_date: dt.da d['date'] = dt.date.fromisoformat(d['date'][:10]) return results['count_over_time']['counts'] + def stories_by_source_week(self, query: str, start_date: dt.date, end_date: dt.date, + collection_ids: Optional[List[int]] = [], source_ids: Optional[List[int]] = [], + platform: Optional[str] = None) -> List[Dict]: + params = self._prep_default_params(query, start_date, end_date, collection_ids, source_ids, platform) + results = self._query('search/count-by-source-week', params) + return results['source-week-attention'] + def story_list(self, query: str, start_date: dt.date, end_date: dt.date, collection_ids: Optional[List[int]] = [], source_ids: Optional[List[int]] = [], platform: Optional[str] = None, expanded: Optional[bool] = None, pagination_token: Optional[str] = None, diff --git a/mediacloud/test/api_search_test.py b/mediacloud/test/api_search_test.py index 8b9327f..1f4f60b 100644 --- a/mediacloud/test/api_search_test.py +++ b/mediacloud/test/api_search_test.py @@ -2,7 +2,9 @@ import os import time from unittest import TestCase + import pytest + import mediacloud.api COLLECTION_US_NATIONAL = 34412234 @@ -236,6 +238,19 @@ def test_collection_ids_filter(self): for s in results: assert s['media_url'] in domains + def test_stories_by_source_week(self): + results = self._search.stories_by_source_week(query="tariff AND Trump", + start_date=dt.date(2025, 1, 1), + end_date=dt.date(2025, 12, 31), + collection_ids=[262985236]) + assert len(results) > 0 + for entry in results: + assert 'media_name' in entry + assert 'week' in entry + assert 'matching_stories' in entry + assert 'total_stories' in entry + assert 'ratio' in entry + class SearchSyntaxTest(TestCase): @@ -324,8 +339,9 @@ def test_negation_source(self): assert not_count < all_count assert minus_count == not_count + class SearchErrorHandlingTest(TestCase): - #New test cases for how the api handles bad input and errors from the server. + # New test cases for how the api handles bad input and errors from the server. START_DATE = dt.date(2024, 1, 1) END_DATE = dt.date(2024, 1, 30) @@ -339,10 +355,10 @@ def setUp(self): def test_datetime(self): query = "biden" result_via_date = self._search.story_count(query=query, start_date=self.START_DATE, end_date=self.END_DATE, - collection_ids=[COLLECTION_US_NATIONAL])['relevant'] + collection_ids=[COLLECTION_US_NATIONAL])['relevant'] with pytest.warns(UserWarning): result_via_datetime = self._search.story_count(query=query, start_date=self.START_DATETIME, end_date=self.END_DATETIME, - collection_ids=[COLLECTION_US_NATIONAL])['relevant'] + collection_ids=[COLLECTION_US_NATIONAL])['relevant'] assert result_via_date == result_via_datetime