diff --git a/app/helpers.py b/app/helpers.py index 5166a44..7f1d6ef 100644 --- a/app/helpers.py +++ b/app/helpers.py @@ -15,7 +15,6 @@ def index_ojects_by_parameter(object_list: List, param: Any): """ Create a dictionary from passed list where each list entry is keyed by passed param - n """ results = {} diff --git a/app/urma.py b/app/urma.py index 3d99906..413dabc 100755 --- a/app/urma.py +++ b/app/urma.py @@ -94,7 +94,7 @@ class MastodonAPI: # Have we reached minimum id? if datum.id < since: break - # Is this our post that we boosted? + # Is this a post that we boosted? if datum.account.id == self.me.id and datum.reblog: # Add in new data results.append(datum) @@ -154,13 +154,14 @@ def main() -> None: update_followed_hashtags(session, mastapi) favourited = mastapi.get_favourited(since) - process_favourited_posts(session, favourited) + process_favourited_posts(session, favourited, mastapi.me.id) - boosted = mastapi.get_boosted(since) - process_boosted_posts(session, boosted) + boosted_by_me = mastapi.get_boosted(since) + process_posts_boosted_by_me(session, boosted_by_me, mastapi.me.id) bookmarked = mastapi.get_bookmarked(since) - process_bookmarked_posts(session, mastapi, bookmarked) + process_bookmarked_posts(session, bookmarked, mastapi.me.id) + # TODO: mastapi.unbookmark(int(post.id)) def get_since_id(session: Session) -> int: @@ -186,53 +187,44 @@ def get_since_id(session: Session) -> int: return max(max_days_ago_id, newest_db_id) -def process_bookmarked_posts(session, mastapi, posts) -> None: +def process_bookmarked_posts(session: Session, + posts: List[Posts], me_id: int) -> None: """ Process bookmarked posts """ for post in posts: - record = _process_post(session, post) + record = _process_post(session, post, me_id) # Posts that are favourited and bookmarked are genuine bookmark # posts: ignore. if record.favourited: continue record.bookmarked = True return - # TODO: mastapi.unbookmark(int(post.id)) -def process_boosted_posts(session, posts) -> None: - """ - Process boosted posts - """ - - for post in posts: - record = _process_post(session, post) - record.boosted = True - - -def process_favourited_posts(session, posts) -> None: +def process_favourited_posts(session: Session, + posts: List[Posts], me_id: int) -> None: """ Process favourited posts """ for post in posts: - record = _process_post(session, post) - record.favourited = True + if post.favourited: + record = _process_post(session, post, me_id) + record.favourited = True + else: + log.debug( + f"process_favourited_posts({post.id=}) not favourited" + ) -def _process_post(session: Session, post) -> Posts: +def _process_post(session: Session, post: Posts, me_id) -> Posts: """ Add passsed post to database """ log.debug(f"{post.id=} processing") - rec = Posts.get_or_create(session, str(post.id)) - if rec.account_id is not None: - # We already have this post - log.debug(f"{post.id=} already in db") - return rec # Create account record if needed log.debug(f"{post.id=} processing {post.account.id=}") @@ -244,7 +236,25 @@ def _process_post(session: Session, post) -> Posts: account_rec.display_name = post.account.display_name account_rec.bot = post.account.bot account_rec.url = post.account.url - rec.account_id = account_rec.id + + if post.reblog: + # We're only interesting the boosted post, not this onej + log.debug(f"{post.id=} {post.reblog.id=}") + boosted_record = _process_post(session, post.reblog, me_id) + # Record who bosed the post unless it was us + if post.account.id == me_id: + boosted_record.boosted_by_id = None + else: + boosted_record.boosted_by_id = account_rec.id + return boosted_record + + rec = Posts.get_or_create(session, str(post.id)) + if rec.account_id is not None: + # We already have this post + log.debug(f"{post.id=} already in db") + return rec + else: + rec.account_id = account_rec.id # Create hashtag records as needed for tag in post.tags: @@ -255,14 +265,31 @@ def _process_post(session: Session, post) -> Posts: rec.created_at = post.created_at rec.uri = post.uri - if post.reblog: - log.debug(f"{post.id=} {post.reblog.id=}") - boosted_rec = _process_post(session, post.reblog) - rec.boosted_by_id = boosted_rec.account_id - return rec +def process_posts_boosted_by_me(session: Session, + posts: List[Posts], me_id: int) -> None: + """ + Process boosted posts + """ + + for post in posts: + if post.account.id != me_id: + log.debug( + f"{post.id=} in process_posts_boosted_by_me but not " + f"boosted by me" + ) + if post.reblog: + record = _process_post(session, post, me_id) + record.boosted = True + record.boosted_by = None + else: + log.debug( + f"{post.id=} in process_boosted_posts but not boosted" + ) + + def update_followed_accounts(session: Session, mastapi: MastodonAPI) -> None: """ Retrieve list of followed accounts and update accounts diff --git a/tests/boosted.pickle b/tests/boosted.pickle new file mode 100644 index 0000000..b1c43da Binary files /dev/null and b/tests/boosted.pickle differ diff --git a/tests/boosting.pickle b/tests/boosting.pickle new file mode 100644 index 0000000..dcfa970 Binary files /dev/null and b/tests/boosting.pickle differ diff --git a/tests/favourited.pickle b/tests/favourited.pickle new file mode 100644 index 0000000..6ef115f Binary files /dev/null and b/tests/favourited.pickle differ diff --git a/tests/test_processing_posts.py b/tests/test_processing_posts.py new file mode 100644 index 0000000..5bd1177 --- /dev/null +++ b/tests/test_processing_posts.py @@ -0,0 +1,119 @@ +import os +import pickle + +from dateutil.tz import tzutc + +from sqlalchemy import select + +from models import ( + Hashtags, + Posts, +) +from urma import ( + process_bookmarked_posts, + process_posts_boosted_by_me, + process_favourited_posts, +) + +BOOSTED_POST = "tests/boosted.pickle" +BOOSTING_POST = "tests/boosting.pickle" +FAVOURITED_POST = "tests/favourited.pickle" +ME_ID = 109568725613662482 + + +def test_process_favourited_no_fave(session): + """Test post not marked as favourite""" + + with open(FAVOURITED_POST, "rb") as inp: + posts = pickle.load(inp) + + # Sanity check + assert len(posts) == 1 + + # Set not favourited + posts[0]['favourited'] = False + + process_favourited_posts(session, posts, ME_ID) + + all_posts = session.execute(select(Posts)).scalars().all() + assert len(all_posts) == 0 + + +def test_process_favourited_with_fave(session): + """Test post marked as favourite""" + + with open(FAVOURITED_POST, "rb") as inp: + posts = pickle.load(inp) + + # Sanity check + assert len(posts) == 1 + + process_favourited_posts(session, posts, ME_ID) + + all_posts = session.execute(select(Posts)).scalars().all() + assert len(all_posts) == 1 + + original_post = posts[0] + retrieved_post = all_posts[0] + + assert original_post.id == int(retrieved_post.post_id) + assert original_post.account.id == int(retrieved_post.account.account_id) + assert original_post.created_at == retrieved_post.created_at + assert retrieved_post.favourited is True + + +def test_process_post_hashtags(session): + """Test hashtags correctly parsed""" + + with open(FAVOURITED_POST, "rb") as inp: + posts = pickle.load(inp) + + # Sanity check + assert len(posts) == 1 + + process_favourited_posts(session, posts, ME_ID) + + all_tags = Hashtags.get_all(session) + + expected = ['fdroid', 'apps', 'android', 'foss', 'free', 'AndroidAppRain'] + for hashtag in all_tags: + assert hashtag.name in expected + + +def test_process_boosts_by_me_no_boost(session): + """Test processing boosted posts with no post that's been boosted""" + + with open(BOOSTING_POST, "rb") as inp: + post = pickle.load(inp) + + post['reblog'] = None + + process_posts_boosted_by_me(session, [post], ME_ID) + + all_posts = session.execute(select(Posts)).scalars().all() + assert len(all_posts) == 0 + + +def test_process_boosts_by_me(session): + """Test processing posts boosted by me""" + + with open(BOOSTING_POST, "rb") as inp: + boosting = pickle.load(inp) + with open(BOOSTED_POST, "rb") as inp: + boosted = pickle.load(inp) + + process_posts_boosted_by_me(session, [boosting], ME_ID) + + all_posts = session.execute(select(Posts)).scalars().all() + assert len(all_posts) == 1 + + stored_post = all_posts[0] + + assert boosted.account.id == int(stored_post.account.account_id) + assert boosted.id == int(stored_post.post_id) + + +def test_process_favourite_containing_boosted_post(session): + """Test processing a favourited post that is a boost""" + + pass