diff --git a/build_deep_stats_model.py b/build_deep_stats_model.py new file mode 100644 index 0000000..ca6e97b --- /dev/null +++ b/build_deep_stats_model.py @@ -0,0 +1,67 @@ +from collections import defaultdict +from dataclasses import dataclass, field +from datetime import datetime +from operator import itemgetter +from typing import Any, Dict, List + +from stats_model import StatsModel + + +@dataclass(kw_only=True) +class BuildDeepStatsModel(StatsModel): + """Stats model built around calculating stats for original post notes, with additional data dimensions.""" + operation: str = 'build_deep_stats' + + # Note count by interaction month and year for original posts. + notes_by_interaction_month_and_year: Dict[str, Any] = field(init=False) + + # Top blogs by original post note count. + top_interactors: List[Dict[str, Any]] = field(init=False) + + def __post_init__(self): + super().__post_init__() + self.notes_by_interaction_month_and_year = self.calculate_notes_by_interaction_month_and_year() + self.top_interactors = self.calculate_top_interactors() + + def calculate_notes_by_interaction_month_and_year(self) -> Dict[str, Any]: + # https://docs.python.org/3/library/collections.html#defaultdict-objects + date_map: Dict[str, Any] = {} + date_map = defaultdict(lambda: {'note_count': 0}, + date_map) + + # Gathering the results. + for post_key in self.original_post_map: + post = self.original_post_map[post_key] + if 'notes' not in post: + continue + notes = post['notes'] + for note in notes: + note_date_key = datetime.fromtimestamp(note['timestamp']).strftime('%Y-%m-%d') + sts = date_map[note_date_key] + sts['date'] = note_date_key + if note['type'] != 'posted': + sts['note_count'] += 1 + + return date_map + + def calculate_top_interactors(self) -> List[Dict[str, Any]]: + top_interactors: defaultdict[str, Dict[str, Any]] = defaultdict(lambda: {'note_count': 0}, + {}) + + # Gathering the results. + for post_key in self.original_post_map: + post = self.original_post_map[post_key] + if 'notes' not in post: + continue + notes = post['notes'] + for note in notes: + person_key = note['blog_name'] + sts = top_interactors[person_key] + if 'blog_name' not in sts: sts['blog_name'] = person_key + if 'url' not in sts: sts['url'] = f"https://{person_key}.tumblr.com" + if note['type'] != 'posted': + sts['note_count'] += 1 + + # https://stackoverflow.com/a/73050 + return sorted(top_interactors.values(), key=itemgetter('note_count'), + reverse=True) diff --git a/stats_model.py b/stats_model.py index 4813e8d..863e64d 100644 --- a/stats_model.py +++ b/stats_model.py @@ -47,8 +47,8 @@ class StatsModel: self.total_posts = self.calculate_total_posts() self.total_original_posts = self.calculate_total_original_posts() self.total_original_post_notes = self.calculate_total_original_post_notes() - self.total_original_post_notes_by_month_and_year = self.calculate_total_original_post_notes_by_month_and_year() - self.total_original_post_notes_by_qtr_and_year = self.calculate_total_original_post_notes_by_qtr_and_year() + self.total_original_post_notes_by_month_and_year = self.calculate_total_original_post_notes_by_month_and_year('note_count') + self.total_original_post_notes_by_qtr_and_year = self.calculate_total_original_post_notes_by_qtr_and_year('note_count') self.most_popular_tags = self.determine_most_popular_tags('note_count') def calculate_total_posts(self) -> int: @@ -63,11 +63,12 @@ class StatsModel: total += self.original_post_map[post_key]['note_count'] return total - def calculate_total_original_post_notes_by_month_and_year(self) -> Dict[str, Any]: + def calculate_total_original_post_notes_by_month_and_year(self, sort_key: str) -> Dict[str, Any]: # https://docs.python.org/3/library/collections.html#defaultdict-objects date_map: Dict[str, Any] = {} date_map = defaultdict(lambda: {'note_count': 0, - 'post_count': 0}, + 'post_count': 0, + 'month_popular_tags': {}}, date_map) # Gathering the results. @@ -81,13 +82,34 @@ class StatsModel: sts['year_month'] = post_date_key sts['post_count'] += 1 sts['note_count'] += post['note_count'] + for tag in post['tags']: + if tag not in sts['month_popular_tags']: + sts['month_popular_tags'][tag] = {} + sts['month_popular_tags'][tag]['note_count'] = 0 + sts['month_popular_tags'][tag]['post_count'] = 0 + tgs = sts['month_popular_tags'][tag] + tgs['post_count'] += 1 + tgs['note_count'] += post['note_count'] + # Results postprocessing. for date in date_map: + # Overall notes to posts ratio. sts = date_map[date] post_count = sts['post_count'] note_count = sts['note_count'] sts['notes_to_posts_ratio'] = note_count / post_count + # By month + tag notes to posts ratio. + tgs = sts['month_popular_tags'] + for tag in tgs: + stg = tgs[tag] + stg['tag'] = tag + t_post_count = stg['post_count'] + t_note_count = stg['note_count'] + stg['notes_to_posts_ratio'] = t_note_count / t_post_count + tgs_list = sorted(list(tgs.values()), key=itemgetter(sort_key), + reverse=True) + sts['month_popular_tags'] = tgs_list return date_map @@ -119,10 +141,10 @@ class StatsModel: return sorted(list(tag_dict.values()), key=itemgetter(sort_key), reverse=True) - def calculate_total_original_post_notes_by_qtr_and_year(self) -> Dict[str, Any]: - total_original_post_notes_by_month_and_year: Dict[str, int] = self.total_original_post_notes_by_month_and_year + def calculate_total_original_post_notes_by_qtr_and_year(self, sort_key: str) -> Dict[str, Any]: + total_original_post_notes_by_month_and_year: Dict[str, Any] = self.total_original_post_notes_by_month_and_year if not total_original_post_notes_by_month_and_year: - total_original_post_notes_by_month_and_year = self.calculate_total_original_post_notes_by_month_and_year() + total_original_post_notes_by_month_and_year = self.calculate_total_original_post_notes_by_month_and_year(sort_key) self.total_original_post_notes_by_month_and_year = total_original_post_notes_by_month_and_year.copy() quarter_map: Dict[str, Any] = {} diff --git a/tumblr_stats.py b/tumblr_stats.py index 14a1a03..2ac5732 100644 --- a/tumblr_stats.py +++ b/tumblr_stats.py @@ -11,6 +11,7 @@ from typing import Any, Callable, Dict, List, Tuple import pytumblr +from build_deep_stats_model import BuildDeepStatsModel from build_draft_stats_model import BuildDraftStatsModel from build_tag_stats_model import BuildTagStatsModel from build_total_stats_model import BuildTotalStatsModel @@ -26,7 +27,8 @@ def get_args() -> Dict[str, Any]: + '$TUMBLR_CONSUMER_KEY, $TUMBLR_CONSUMER_SECRET, $TUMBLR_OAUTH_TOKEN, and $TUMBLR_OAUTH_SECRET', epilog='— Be gay and do crime') parser.add_argument('operation', type=str, nargs='+', metavar='OPERATION', - choices=['build_tag_stats', 'build_queue_stats', 'build_draft_stats'], + choices=['build_tag_stats', 'build_queue_stats', 'build_draft_stats', + 'build_deep_stats'], help="operation used to calculate stats") parser.add_argument('-b', '--blog', type=str, required=True, help='blog name for which to calculate stats') @@ -94,6 +96,7 @@ def build_post_maps(client: pytumblr.TumblrRestClient, draft_url = f"/v2/blog/{blog_name}/posts/draft" is_draft_stats: bool = 'build_draft_stats' in args['operation'] + is_deep_stats: bool = 'build_deep_stats' in args['operation'] total: int = 0 offset: int = 0 @@ -112,6 +115,8 @@ def build_post_maps(client: pytumblr.TumblrRestClient, elif is_draft_stats: data = client.send_api_request("get", draft_url) else: # Above is for queued + draft posts, below is for published posts. + if is_deep_stats: + params.update({'notes_info': 'true'}) data = client.posts(f"{blog_name}.tumblr.com", offset=offset, limit=limit, @@ -236,6 +241,14 @@ def main() -> None: stats_model = BuildTotalStatsModel(blog_name=args['blog'], original_post_map=og_post_map, unoriginal_post_map=un_og_post_map) + case {'operation': op} if 'build_deep_stats' in operation: + if 'after' not in args: # or 'before' not in args: + print(f"You must specify a time range for {op}. " + + 'You\'ll otherwise request TOO MUCH DATA!') + sys.exit(1) + stats_model = BuildDeepStatsModel(blog_name=args['blog'], + original_post_map=og_post_map, + unoriginal_post_map=un_og_post_map) case _: print('Unsupported command. How did you even make it this far?!') sys.exit(1)