From b6c8396c63e721b3d5b1f253ab1003a16bd3e000 Mon Sep 17 00:00:00 2001 From: grimmer Date: Fri, 27 Aug 2021 22:24:50 +0800 Subject: [PATCH] Use typing.Tuple instead to solve below Python 3.9 compatibility issue Add since_date/until_date on get_page_default_web_insight Change some fields in responses to Optional to prevent exceptions. Add more type annotations Update README.md Add missing .env path setting Add Literal for get_page_default_web_insight's period paramter Auto renew long-lived page token & use TinyDB to store Update README Fix compose_fb_graph_api_page_request Add error handle if an access_token does not have any page scope related Check if InsightsResponse has valid data Throw exception in get_long_lived_token if fb_app_id or fb_app_secret is invalid Check if the possible return error in get_page_token_from_user_token Check if since is less than until parameters Update README.md Update README.md Update README.md docs(README): add usage examples build(.env): remove credentials --- .env | 2 + README.md | 29 +- poetry.lock | 14 +- pyproject.toml | 1 + .../fb_page_insight.py | 374 ++++++++++++++---- 5 files changed, 328 insertions(+), 92 deletions(-) diff --git a/.env b/.env index 8668cae..9763252 100644 --- a/.env +++ b/.env @@ -3,3 +3,5 @@ fb_app_id= fb_app_secret= fb_default_page_id= fb_default_page_access_token= +GOOGLE_APPLICATION_CREDENTIALS= +BIGQUERY_PROJECT= diff --git a/README.md b/README.md index 8bc10ca..31c7ea5 100644 --- a/README.md +++ b/README.md @@ -1,16 +1,18 @@ -# Python Facebook Page Insights Client +# Python Page Insights Client + +Currently, it is used by https://github.com/pycontw/pycon-etl . Check [dags/ods/fb_page_insights/udfs/fb_page_insights_data_uploader.py](https://github.com/pycontw/pycon-etl/blob/master/dags/ods/fb_page_insights/dags/fb_page_insights_2_bigquery.py) as an example of how to use this Python module. ## Usage -## Get needed secrets first +### Get needed secrets first https://github.com/facebook/facebook-python-business-sdk#register-an-app is a reference and the steps are 1. create a FB app and get its `app_id` and `secret`, -2. In terms of `user_access_token`, make sure you are a registered developer of this fb app and get user access token on Graph Explorer. You will get a short-term user_token by default, expired in 2 or 3 months. To get long-term token, choose either of the below ways - - using Graph Exploer -> Access token tool -> Extend access token - - invoke get_long_lived_user_token of this library +2. In terms of `user_access_token/page_access_token`, make sure you are a registered developer of this fb app, and get user_access_token or page_user_token on Graph Explorer with selected scopes, read_insights & pages_read_engagement. Either user_token or page_token is working. You will get a short-term token by default, expired in 2 or 3 months. To get long-term token, choose either of the below ways + - this library will automatically get the long-lived token and cache it in local db.json + - manually use Graph Exploer -> Access token tool -> Extend access token to get long-lived token first. -Rather than Graph Explorer, https://github.com/pycontw/python-fb-page-insights-client/issues/6 introduces another way which does not to be a registered developer of this fb app. But this way is not recommanded. +Rather than Graph Explorer, https://github.com/pycontw/python-fb-page-insights-client/issues/6 introduces another way which does not to be a registered developer of this fb app. But this way is not recommended. ### Pass secrets @@ -25,11 +27,13 @@ fb_default_page_access_token= ``` You can choose any of below ways: -- pass them as function paramets -- manually export them as enviornments variables +- pass them as function parameters +- manually export them as environment variables - create a .env to include them -if fb_user_access_token is filled, fb_default_page_access_token is not necessary and will be ignored. fb_user_access_token will be used to get page token internally. +if fb_default_page_access_token is filled and valid, fb_user_access_token usage will be skipped. fb_user_access_token is used to get page token internally. So you only need to fill either a valid fb_default_page_access_token or fb_user_access_token. + +You can skip fb_app_id and fb_app_secret if you are sure if fb_user_access_token/fb_default_page_access_token is long-lived token. In https://developers.facebook.com/tools/debug/accesstoken/, You can paste the token to check its "Expires" and "Valid" field. ## Fetch data @@ -44,10 +48,6 @@ Use `FBPageInsight` class to fetch. Please checkout the unit test code as an exa - `Calls within one hour = 4800 * Number of Engaged Users` - api response header inclues `x-business-use-case-usage` -### Somehow FB will return invalid data sometimes - -FB API is not very stable and plrease try again. - ## Development 1. `poetry shell` @@ -66,4 +66,5 @@ Two methods: ## TODO: -https://github.com/pycontw/python-fb-page-insights-client/discussions/4 +https://github.com/pycontw/facebook_page_insights_client/discussions/4 + diff --git a/poetry.lock b/poetry.lock index 63b2ccb..f50b34f 100644 --- a/poetry.lock +++ b/poetry.lock @@ -89,6 +89,14 @@ urllib3 = ">=1.21.1,<1.27" socks = ["PySocks (>=1.5.6,!=1.5.7)", "win-inet-pton"] use_chardet_on_py3 = ["chardet (>=3.0.2,<5)"] +[[package]] +name = "tinydb" +version = "4.5.1" +description = "TinyDB is a tiny, document oriented database optimized for your happiness :)" +category = "main" +optional = false +python-versions = ">=3.5,<4.0" + [[package]] name = "toml" version = "0.10.2" @@ -121,7 +129,7 @@ socks = ["PySocks (>=1.5.6,!=1.5.7,<2.0)"] [metadata] lock-version = "1.1" python-versions = "^3.7.1" -content-hash = "eebb988ad87e7791c52f13683eaae0071c4981cb68923c8582985b942e3ccc11" +content-hash = "9d9e42a70976c0a94eb34d843f19b2f69926e500e11bb477db3c245a01c241eb" [metadata.files] autopep8 = [ @@ -176,6 +184,10 @@ requests = [ {file = "requests-2.26.0-py2.py3-none-any.whl", hash = "sha256:6c1246513ecd5ecd4528a0906f910e8f0f9c6b8ec72030dc9fd154dc1a6efd24"}, {file = "requests-2.26.0.tar.gz", hash = "sha256:b8aa58f8cf793ffd8782d3d8cb19e66ef36f7aba4353eec859e74678b01b07a7"}, ] +tinydb = [ + {file = "tinydb-4.5.1-py3-none-any.whl", hash = "sha256:99529ea4d5d4b7a3fd4b3f50b48b5023c31d6f7a4a87bb103b4abd1d959d93b9"}, + {file = "tinydb-4.5.1.tar.gz", hash = "sha256:b780bceac6e37573b10fbcab6bcebb6b8bd5d8a0024533b5a452d9ef83465783"}, +] toml = [ {file = "toml-0.10.2-py2.py3-none-any.whl", hash = "sha256:806143ae5bfb6a3c6e736a764057db0e6a0e05e338b5630894a5f779cabb4f9b"}, {file = "toml-0.10.2.tar.gz", hash = "sha256:b3bda1d108d5dd99f4a20d24d9c348e91c4db7ab1b749200bded2f839ccbe68f"}, diff --git a/pyproject.toml b/pyproject.toml index 282234e..5d169e2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,6 +9,7 @@ python = "^3.7.1" pydantic = "^1.8.2" requests = "^2.25.1" python-dotenv = "^0.18.0" +tinydb = "^4.5.1" [tool.poetry.dev-dependencies] autopep8 = "^1.5.7" diff --git a/python_fb_page_insights_client/fb_page_insight.py b/python_fb_page_insights_client/fb_page_insight.py index da26970..62bef2b 100644 --- a/python_fb_page_insights_client/fb_page_insight.py +++ b/python_fb_page_insights_client/fb_page_insight.py @@ -1,13 +1,13 @@ from datetime import datetime, timedelta -from typing import List, Optional, Union, Dict +from typing import Any, List, Optional, Union, Dict, Tuple, Literal + from pydantic import BaseModel, BaseSettings, Field, validator from enum import Enum, auto, IntEnum -# from dataclasses import dataclass, field -import time - -import http.client import requests +from tinydb import TinyDB, Query + import logging +import http.client # debug only # logging.basicConfig(level=logging.DEBUG) @@ -103,6 +103,14 @@ class PageMetric(Enum): page_impressions_organic_unique = auto() +class DebugError(BaseModel): + code: int + message: str + subcode: Optional[int] # completely worng will not show this + type: Optional[str] + fbtrace_id: Optional[str] + + class PostActivityValue(BaseModel): share: int = None like: int = None @@ -143,14 +151,19 @@ class InsightData(BaseModel): class InsightsCursors(BaseModel): - previous: str # similar query but add since & until - next: str + # not seen Optional case but add it just in case + previous: Optional[str] # similar query but add since & until + # if query time range includes a future day, + # next will be missing + next: Optional[str] # page & post both use this class InsightsResponse(BaseModel): - data: List[InsightData] - paging: InsightsCursors + data: Optional[List[InsightData]] + # not seen Optional case but add it just in case + paging: Optional[InsightsCursors] + error: Optional[DebugError] # e.g. use invalid token class Category(BaseModel): @@ -179,8 +192,59 @@ class AccountPaging(BaseModel): class AccountResponse(BaseModel): - data: List[AccountData] - paging: AccountPaging + data: Optional[List[AccountData]] + paging: Optional[AccountPaging] + error: Optional[DebugError] + + +class GranularScope(BaseModel): + scope: str + target_ids: Optional[List[str]] # page_id list + + +class DebugData(BaseModel): + ''' TODO: success or error can be a union + ''' + is_valid: bool + scopes: List[str] # error will have a empty list + # "email", + # "read_insights", + # "pages_show_list", + # "pages_read_engagement", + # "public_profile" + + error: Optional[DebugError] + + issued_at: Optional[int] # existing if a token is not never expired + profile_id: Optional[str] # only existing for page token + + # valid or token expired (invalid) will show below. if token is completely wrong + # (e.g. format is only 4 characteristics) will not show + granular_scopes: List[GranularScope] + app_id: str # "1111808169311111" + type: str # "USER" / "PAGE" + application: str # "pycontw_insights_bot" + data_access_expires_at: int # 1641046186 + expires_at: int # 1633276800. 0 means never + user_id: str + + +# error case1: complete wrong +# "data": { +# "error": { +# "code": 190, +# "message": "Invalid OAuth access token." +# }, +# "is_valid": false, +# "scopes": [ +# ] +# } + + +class DebugResponse(BaseModel): + data: Optional[DebugData] + # if omit access_token so the response will only have error and no data + error: Optional[DebugError] class PostData(BaseModel): @@ -201,13 +265,15 @@ def set_created_time(cls, v): class PostsPaging(BaseModel): - cursors: BeforeAfterCursors + # not see Optional case but add it just in case + cursors: Optional[BeforeAfterCursors] next: Optional[str] class PostsResponse(BaseModel): data: List[PostData] - paging: PostsPaging + # not see Optional case but add it just in case + paging: Optional[PostsPaging] class PostCompositeData(BaseModel): @@ -299,8 +365,9 @@ class PostsWebInsightData(BaseModel): class LongLivedResponse(BaseModel): - access_token: str - token_type: str + access_token: Optional[str] + token_type: Optional[str] + error: Optional[DebugError] class FBPageInsight(BaseSettings): @@ -316,60 +383,198 @@ class FBPageInsight(BaseSettings): api_server = 'https://graph.facebook.com' api_version = 'v10.0' + class Config: + env_file = '.env' + env_file_encoding = 'utf-8' + # class Config: # env_file = ".env" + # def __call__(self, act): + # print("I am:") + # method = getattr(self, act) + # if not method: + # print("not implmeent") + # # raise Exception("Method %s not implemented" % method_name) + # method() + + # def __getattribute__(self, attr): + # method = object.__getattribute__(self, attr) + # if not method: + # raise Exception("Method %s not implemented" % attr) + # if callable(method): + # print("I am:") + # return method + @property def api_url(self): return f'{self.api_server}/{self.api_version}' - # TODO: - # 1. page_token is doable, too? - def get_long_lived_user_token(self): - if self.fb_user_access_token == "" or self.fb_app_id == "" or self.fb_app_secret == "": + def _page_id(self, page_id: str): + if page_id is None: + used_page_id = self.fb_default_page_id + else: + used_page_id = page_id + + return used_page_id + + def get_long_lived_token(self, access_token: str): + ''' either user token or page_token''' + if self.fb_app_id == "" or self.fb_app_secret == "": return "" - url = f'{self.api_url}/oauth/access_token?grant_type=fb_exchange_token&client_id={self.fb_app_id}&client_secret={self.fb_app_secret}&fb_exchange_token={self.fb_user_access_token}' + url = f'{self.api_url}/oauth/access_token?grant_type=fb_exchange_token&client_id={self.fb_app_id}&client_secret={self.fb_app_secret}&fb_exchange_token={access_token}' r = requests.get(url) json_dict = r.json() resp = LongLivedResponse(**json_dict) - if resp.access_token != None: - self.fb_user_access_token = resp.access_token + if resp.error is not None: + raise ValueError( + f"fail to get long-lived token:{resp.error.message}") + if resp.access_token is not None: + # self.fb_user_access_token = resp.access_token return resp.access_token else: return "" - # TODO: better way to refresh token instead of getting all pages' tokens? - def get_page_token(self, target_page_id): - if self.fb_user_access_token == "": - if self.fb_default_page_access_token != "": - # only use pre-defined page_token when user_token is not present - return self.fb_default_page_access_token - else: - raise ValueError( - "fb_user_access_token should be assigned first") - - if target_page_id == None or target_page_id == "": + def _check_scope(self, data: DebugData, target_page_id: str): + has_list_scope = False + if "read_insights" in data.scopes: + has_list_scope = True + has_engagement_scope = False + for granular_scope in data.granular_scopes: + scope = granular_scope.scope + target_ids = granular_scope.target_ids + # "read_insights" will only show in scopes but not in granular_scopes in https://developers.facebook.com/tools/explore, so forget it + # if scope == "pages_show_list": + # if target_page_id in target_ids: + # has_list_scope = True + # if user_token does not have page related, target_ids s None + if scope == "pages_read_engagement" and target_ids is not None: + if target_page_id in target_ids: + has_engagement_scope = True + if not has_list_scope or not has_engagement_scope: + return False + return True + + def get_page_long_lived_token(self, target_page_id: str): + + if target_page_id is None or target_page_id == "": raise ValueError("target_page_id should be a non empty string") - if self.fb_page_access_token_dict == None: + + # avoid reading db too often + if self.fb_page_access_token_dict is None: self.fb_page_access_token_dict = {} page_token = self.fb_page_access_token_dict.get(target_page_id) - if page_token != None: + if page_token == "": + raise ValueError("no valid page token") + elif page_token is not None: return page_token - url = f'{self.api_url}/me/accounts?access_token={self.fb_user_access_token}' + # check cached tinyDB + db = TinyDB('db.json') + q = Query() + store_record = db.get( + q.page_id == target_page_id) + if store_record: + page_long_lived_token = store_record["page_long_lived_token"] + self.fb_page_access_token_dict[target_page_id] = page_long_lived_token + return page_long_lived_token + + if not self.fb_user_access_token and not self.fb_default_page_access_token: + # if self.fb_default_page_access_token != "": + # # only use pre-defined page_token when user_token is not present + # return self.fb_default_page_access_token + # else: + raise ValueError( + "fb_user_access_token/page_token should be assigned first") + # NOTE: + # If not get no_expire_page_token suecessfully, still set self.fb_page_access_token_dict[target_page_id] = "", + # this is to avoid retrying failure in this time process running + no_expire_page_token = "" + if self.fb_default_page_access_token: + test_token = self.fb_default_page_access_token + resp = self.debug_token(test_token) + data = resp.data + if data is None or data.is_valid is False or data.type != 'PAGE': + print("invalid page token") + elif self._check_scope(data, target_page_id) is False: + # elif data.profile_id != target_page_id: # in some cases profile_is is missing + print( + f"no has pages_show_list/pages_read_engagement for this page_id & page token:{target_page_id}") + elif data.expires_at == 0: + no_expire_page_token = test_token + print("get long-lived page token") + else: + # get long-lived token (which is never expired for accessing some basic data, e.g. page insights) + no_expire_page_token = self.get_long_lived_token(test_token) + if not no_expire_page_token and self.fb_user_access_token: + test_token = self.fb_user_access_token + resp = self.debug_token(test_token) + data = resp.data + if data is None or data.is_valid is False or data.type != "USER": + print("invalid user token") + else: + if self._check_scope(data, target_page_id) is False: + print( + f"does not have pages_show_list/pages_read_engagement for this page_id & user token:{target_page_id}") + else: + if data.expires_at == 0: + print("got long-lived user token yet") + no_expire_user_token = test_token + else: + # get long-lived token (which is never expired for accessing some basic data, e.g. page insights) + no_expire_user_token = self.get_long_lived_token( + test_token) + # resp2 = self.debug_token(no_expire_user_token) + no_expire_page_token = self.get_page_token_from_user_token( + target_page_id, no_expire_user_token) + + if no_expire_page_token: + db.insert({'page_id': target_page_id, + 'page_long_lived_token': no_expire_page_token}) + else: + raise ValueError("no available valid user/page token") + + self.fb_page_access_token_dict[target_page_id] = no_expire_page_token + return no_expire_page_token + + def debug_token(self, token: str): + url = f'{self.api_url}/debug_token?access_token={token}&input_token={token}' + r = requests.get(url) + json_dict = r.json() + resp = DebugResponse(**json_dict) + return resp + + # TODO: better way to refresh token instead of getting all pages' tokens? + + def get_page_token_from_user_token(self, target_page_id: str, user_token: str): + url = f'{self.api_url}/me/accounts?access_token={user_token}' r = requests.get(url) json_dict = r.json() resp = AccountResponse(**json_dict) - if resp.data != None and len(resp.data) > 0: + if resp.error is not None: + raise ValueError( + f"fail to get page token from user token:{resp.error.message}") + if resp.data is not None and len(resp.data) > 0: for data in resp.data: - if data.access_token != None and data.id == target_page_id: - self.fb_page_access_token_dict[target_page_id] = data.access_token + if data.access_token is not None and data.id == target_page_id: return data.access_token return "" - def compose_fb_graph_api_request(self, token, object_id, endpoint, param_dict: Dict[str, str] = {}): + def compose_fb_graph_api_page_request(self, page_id: str, endpoint: str, param_dict: Dict[str, str] = {}, object_id=""): + # TODO: refactor it later, page_id & object_id position + if page_id: + page_token = self.get_page_long_lived_token(page_id) + elif object_id: + page_token = self.get_page_long_lived_token(page_id) + else: + raise ValueError("no passed token") + params = self._convert_para_dict(param_dict) - url = f'{self.api_url}/{object_id}/{endpoint}?access_token={token}{params}' + url = "" + if object_id: + url = f'{self.api_url}/{object_id}/{endpoint}?access_token={page_token}{params}' + elif page_id: + url = f'{self.api_url}/{page_id}/{endpoint}?access_token={page_token}{params}' r = requests.get(url) json_dict = r.json() return json_dict @@ -390,13 +595,17 @@ def _convert_metric_list(self, metric_list: List[PageMetric]): metric_value += ","+metric.name return metric_value - def get_page_insights(self, page_id=None, + def _check_since_less_than_until(self, since: int, until: int): + if since > until: + raise ValueError("since is more than until, not valid") + + def get_page_insights(self, page_id: str = None, user_defined_metric_list: List[PageMetric] = [], since: int = None, until: int = None, date_preset: DatePreset = DatePreset.yesterday, period: Period = Period.week): page_id = self._page_id(page_id) - page_token = self.get_page_token(page_id) + # page_token = self.get_page_long_lived_token(page_id) # TODO: # 1. validate parameters @@ -406,12 +615,13 @@ def get_page_insights(self, page_id=None, user_defined_metric_list = [e for e in PageMetric] metric_value = self._convert_metric_list(user_defined_metric_list) - if since != None and until != None: - json_dict = self.compose_fb_graph_api_request(page_token, - page_id, "insights", {"metric": metric_value, "date_preset": date_preset.name, 'period': period.name, "since": since, "until": until}) + if since is not None and until is not None: + self._check_since_less_than_until(since, until) + json_dict = self.compose_fb_graph_api_page_request( + page_id, "insights", {"metric": metric_value, "date_preset": date_preset.name, 'period': period.name, "since": since, "until": until}) else: - json_dict = self.compose_fb_graph_api_request(page_token, - page_id, "insights", {"metric": metric_value, "date_preset": date_preset.name, 'period': period.name}) + json_dict = self.compose_fb_graph_api_page_request( + page_id, "insights", {"metric": metric_value, "date_preset": date_preset.name, 'period': period.name}) resp = InsightsResponse(**json_dict) return resp @@ -420,20 +630,21 @@ def get_page_insights(self, page_id=None, def get_posts(self, page_id: str = None, since: int = None, until: int = None): # could use page_token or user_access_token page_id = self._page_id(page_id) - page_token = self.get_page_token(page_id) + # page_token = self.get_page_long_lived_token(page_id) # get_all = False next_url = "" post_data_list: List[PostData] = [] - while next_url != None: + while next_url is not None: if next_url == "": - if since != None and until != None: - json_dict = self.compose_fb_graph_api_request(page_token, - # {"since": 1601555261, "until": 1625489082}) - page_id, "posts", {"since": since, "until": until}) + if since is not None and until is not None: + self._check_since_less_than_until(since, until) + json_dict = self.compose_fb_graph_api_page_request( + # {"since": 1601555261, "until": 1625489082}) + page_id, "posts", {"since": since, "until": until}) else: - json_dict = self.compose_fb_graph_api_request(page_token, - page_id, "posts") + json_dict = self.compose_fb_graph_api_page_request( + page_id, "posts") resp = PostsResponse(**json_dict) else: r = requests.get(next_url) @@ -446,8 +657,7 @@ def get_posts(self, page_id: str = None, since: int = None, until: int = None): total_resp = PostsResponse(data=post_data_list, paging=resp.paging) return total_resp - def get_post_insight(self, post_id: str, basic_metric=True, complement_metric=True, user_defined_metric_list=[]): - page_id = post_id.split('_')[0] + def get_post_insight(self, post_id: str, basic_metric=True, complement_metric=True, user_defined_metric_list: List[PageMetric] = []): if len(user_defined_metric_list) == 0: metric_list = [] @@ -460,32 +670,39 @@ def get_post_insight(self, post_id: str, basic_metric=True, complement_metric=Tr metric_list = user_defined_metric_list metric_value = self._convert_metric_list(metric_list) - page_token = self.get_page_token(page_id) + page_id = post_id.split('_')[0] + # page_token = self.get_page_long_lived_token(page_id) - json_dict = self.compose_fb_graph_api_request(page_token, - post_id, "insights", {"metric": metric_value}) + json_dict = self.compose_fb_graph_api_page_request( + page_id, "insights", {"metric": metric_value}, object_id=post_id) # NOTE: somehow FB will return invalid api result - # if json_dict.get("data") == None: + # if json_dict.get("data") is None: # print("not ok") for debugging, resp = InsightsResponse(**json_dict) return resp - def _page_id(self, page_id: str): - if page_id == None: - return self.fb_default_page_id - return page_id - - def get_page_default_web_insight(self, page_id: str = None, since: int = None, until: int = None, + def get_page_default_web_insight(self, page_id: str = None, since_date: Tuple[str, str, str] = None, until_date: Tuple[str, str, str] = None, date_preset: DatePreset = DatePreset.yesterday, - period: Period = Period.week, return_as_dict=False): - """ period can not be lifetime""" + period: Literal[Period.day, Period.week, Period.days_28, Period.month] = Period.week, return_as_dict=False): + """ since_date/until_date is (2021,9,9) format & period can not be lifetime""" page_id = self._page_id(page_id) if period == Period.lifetime: raise ValueError( 'period can not be lifetime when querying default page insight') + since = None + until = None + if since_date is not None and until_date is not None: + since = int(datetime( + *since_date).timestamp()) + until = int(datetime( + *until_date).timestamp()) + page_summary = self.get_page_insights( page_id, since=since, until=until, date_preset=date_preset, period=period) + if page_summary.error is not None: + raise ValueError( + f"page insight error:{page_summary.error.message}") page_summary_data = page_summary.data # page_composite_data = PagePostsCompositeData( @@ -498,7 +715,7 @@ def get_page_default_web_insight(self, page_id: str = None, since: int = None, u return resp.dict() return resp - def get_post_default_web_insight(self, page_id: str = None, since_date: tuple[str, str, str] = None, until_date: tuple[str, str, str] = None, between_days: int = None, return_as_dict=False): + def get_post_default_web_insight(self, page_id: str = None, since_date: Tuple[str, str, str] = None, until_date: Tuple[str, str, str] = None, between_days: int = None, return_as_dict=False): """ since_date and until_date are the tuple form of (2020, 9, 7) if any of since_date and until_date is omitting, between_days will be used to decide either since_date or until_date and default value is 365. @@ -508,7 +725,7 @@ def get_post_default_web_insight(self, page_id: str = None, since_date: tuple[st since_date, until_date, period_days can not be all specified as non None at the same time, will throw a error """ - if since_date != None and until_date != None and between_days != None: + if since_date is not None and until_date is not None and between_days is not None: raise ValueError( "since_date, until_date, period_days can not all be non-None at the same time") @@ -528,11 +745,11 @@ def get_post_default_web_insight(self, page_id: str = None, since_date: tuple[st # first_day_next_month = datetime.datetime(2021, 1, 1) # e.g. 1609430400 - if between_days == None: + if between_days is None: between_days = FBPageInsightConst.default_between_days - if until_date == None: - if since_date == None: + if until_date is None: + if since_date is None: until_time = query_time else: since_time = datetime( @@ -542,7 +759,7 @@ def get_post_default_web_insight(self, page_id: str = None, since_date: tuple[st until_time = datetime(*until_date) until = int(until_time.timestamp()) - if since_date == None: + if since_date is None: since_time = until_time - timedelta(days=between_days) else: since_time = datetime( @@ -561,6 +778,9 @@ def get_post_default_web_insight(self, page_id: str = None, since_date: tuple[st post_composite_list.append(composite_data) post_id = post.id post_insight = self.get_post_insight(post_id) + if post_insight.error is not None: + raise ValueError( + f"post insight error:P{post_insight.error.message}") post_insight_data = post_insight.data for post_insight in post_insight_data: if post_insight.name in PostMetric.__members__: @@ -597,7 +817,7 @@ def _organize_to_web_page_data_shape(self, page_data: List[InsightData], page_id # end_time = datetime.strptime( # value_obj.end_time, '%Y-%m-%dT%H:%M:%S+%f').isoformat() # .timestamp()) insight = insight_dict.get(end_time) - if insight == None: + if insight is None: insight = PageDefaultWebInsight( period=period, page_id=page_id, end_time=end_time) insight_dict[end_time] = insight