Added build_deep_stats_model, made other small tweaks

This commit is contained in:
2026-08-17 21:28:15 -04:00
parent a2d49785be
commit ea116ee0a3
3 changed files with 110 additions and 8 deletions

67
build_deep_stats_model.py Normal file
View File

@@ -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)

View File

@@ -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] = {}

View File

@@ -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)