refactor: unify recent-activity logic with generic engine + processors

app.py

@@ -58,6 +58,9 @@ def build_session(accept: str) -> requests.Session:
     return session
 
 
+# ============================================================
+# Recent-activity shared helpers
+# ============================================================
 def normalize_recent_activity_payload(payload):
     if isinstance(payload, list):
         return payload, None
@@ -76,15 +79,9 @@ def normalize_recent_activity_payload(payload):
 
 
 def get_year_bounds(selected_year: int):
-    start_ts = pd.Timestamp(f"{selected_year}-01-01", tz="UTC")
-    end_ts = pd.Timestamp(f"{selected_year}-12-31 23:59:59.999999", tz="UTC")
-    return start_ts, end_ts
-
-
-def should_stop_after_crossing_year_start(
-    crossed_before_year_start: bool,
-) -> bool:
-    return crossed_before_year_start
+    year_start = pd.Timestamp(f"{selected_year}-01-01", tz="UTC")
+    year_end = pd.Timestamp(f"{selected_year}-12-31 23:59:59.999999", tz="UTC")
+    return year_start, year_end
 
 
 def build_events_page_signature(events: list[dict]) -> tuple:
@@ -104,6 +101,71 @@ def build_events_page_signature(events: list[dict]) -> tuple:
     return tuple(signature)
 
 
+def parse_recent_activity_event_ts(event: dict):
+    raw_time = event.get("time")
+    if not raw_time:
+        return None
+    try:
+        return pd.to_datetime(raw_time, utc=True)
+    except Exception:
+        return None
+
+
+def update_recent_activity_progress(
+    event_ts,
+    *,
+    year_start,
+    year_end,
+    oldest_event_ts_seen,
+    page_oldest_event_ts,
+    seen_target_year,
+    crossed_before_year_start,
+):
+    if event_ts is None:
+        return (
+            oldest_event_ts_seen,
+            page_oldest_event_ts,
+            seen_target_year,
+            crossed_before_year_start,
+        )
+
+    if oldest_event_ts_seen is None or event_ts < oldest_event_ts_seen:
+        oldest_event_ts_seen = event_ts
+
+    if page_oldest_event_ts is None or event_ts < page_oldest_event_ts:
+        page_oldest_event_ts = event_ts
+
+    if year_start <= event_ts <= year_end:
+        seen_target_year = True
+    elif event_ts < year_start:
+        crossed_before_year_start = True
+
+    return (
+        oldest_event_ts_seen,
+        page_oldest_event_ts,
+        seen_target_year,
+        crossed_before_year_start,
+    )
+
+
+def update_stalled_pagination_state(
+    *,
+    page_oldest_event_ts,
+    previous_oldest_event_ts_seen,
+    stalled_oldest_count,
+):
+    if previous_oldest_event_ts_seen is not None and page_oldest_event_ts is not None:
+        if page_oldest_event_ts >= previous_oldest_event_ts_seen:
+            stalled_oldest_count += 1
+        else:
+            stalled_oldest_count = 0
+
+    if page_oldest_event_ts is not None:
+        previous_oldest_event_ts_seen = page_oldest_event_ts
+
+    return previous_oldest_event_ts_seen, stalled_oldest_count
+
+
 # ============================================================
 # Thread-safe rate limiters
 # ============================================================
@@ -485,7 +547,7 @@ def build_invalid_section_state(username: str, selected_year: int, empty_state_f
 
 
 # ============================================================
-# Community activity fetch
+# Community activity state
 # ============================================================
 def empty_community_state(selected_year: int):
     return {
@@ -500,36 +562,30 @@ def empty_community_state(selected_year: int):
     }
 
 
-def fetch_user_discussion_activity(
+# ============================================================
+# Generic recent-activity engine
+# ============================================================
+def collect_recent_activity(
+    *,
     username: str,
     selected_year: int,
-    mode: str,
+    activity_type: str,
+    processor,
+    build_session_fn,
+    rate_limiter_obj,
+    limit: int = RECENT_ACTIVITY_LIMIT,
+    max_pages: int = RECENT_ACTIVITY_MAX_PAGES,
+    max_stalled_pages: int = RECENT_ACTIVITY_MAX_STALLED_PAGES,
 ):
     username = (username or "").strip()
     selected_year = int(selected_year)
 
-    title_map = {
-        "pr_opened": "Pull Requests",
-        "pr_activity": "PR Activity",
-        "thread_opened": "Community Threads",
-        "discussion_activity": "Community Activity",
-    }
-    title = title_map.get(mode, "Community")
-
-    if not username:
-        return {
-            "title": title,
-            "fetched": True,
-            "dates": [],
-            "count": 0,
-            "logs": "No username provided.",
-        }
+    processor_state = processor.init_state()
+    year_start, year_end = get_year_bounds(selected_year)
+    session = build_session_fn("application/json")
 
-    dates = []
-    seen_ids = set()
-    skip = 0
-    limit = RECENT_ACTIVITY_LIMIT
     cursor = None
+    skip = 0
     pages_fetched = 0
 
     seen_target_year = False
@@ -540,93 +596,19 @@ def fetch_user_discussion_activity(
     seen_cursors = set()
     seen_page_signatures = set()
 
-    OPEN_EVENT_EPSILON_SECONDS = 90
     logs = []
 
-    year_start, year_end = get_year_bounds(selected_year)
-    session = build_session("application/json")
-
-    def extract_author_name(discussion: dict) -> str | None:
-        author = discussion.get("author")
-        if isinstance(author, str):
-            return author
-        if isinstance(author, dict):
-            for key in ["name", "username", "login"]:
-                value = author.get(key)
-                if isinstance(value, str):
-                    return value
-        return None
-
-    def parse_ts(raw_value):
-        if not raw_value:
-            return None
-        try:
-            return pd.to_datetime(raw_value, utc=True)
-        except Exception:
-            return None
-
-    def event_matches_mode(event: dict, discussion: dict) -> tuple[bool, str | None]:
-        is_pr = bool(discussion.get("isPullRequest", False))
-        author_name = extract_author_name(discussion)
-        repo_id = event.get("repoId", "unknown-repo")
-        discussion_num = discussion.get("num", "unknown-num")
-        event_id = event.get("eventId", "unknown-event")
-
-        if mode == "pr_opened":
-            if not is_pr:
-                return False, None
-            if not author_name or author_name.lower() != username.lower():
-                return False, None
-            return True, f"{repo_id}::pr::{discussion_num}"
-
-        if mode == "pr_activity":
-            if not is_pr:
-                return False, None
-            return True, f"event::{event_id}"
-
-        if mode == "thread_opened":
-            if is_pr:
-                return False, None
-            if not author_name or author_name.lower() != username.lower():
-                return False, None
-            return True, f"{repo_id}::thread::{discussion_num}"
-
-        if mode == "discussion_activity":
-            if is_pr:
-                return False, None
-            return True, f"event::{event_id}"
-
-        return False, None
-
-    def extract_timestamp_for_mode(event: dict, discussion: dict):
-        event_ts = parse_ts(event.get("time"))
-        created_ts = parse_ts(discussion.get("createdAt"))
-
-        if mode in {"pr_opened", "thread_opened"}:
-            return created_ts
-
-        if mode in {"pr_activity", "discussion_activity"}:
-            if event_ts is None:
-                return None
-            if created_ts is not None:
-                delta_seconds = abs((event_ts - created_ts).total_seconds())
-                if delta_seconds <= OPEN_EVENT_EPSILON_SECONDS:
-                    return None
-            return event_ts
-
-        return None
-
     try:
         while True:
-            if pages_fetched >= RECENT_ACTIVITY_MAX_PAGES:
+            if pages_fetched >= max_pages:
                 logs.append("Stopped due to safety max pages.")
                 break
 
-            recent_activity_rate_limiter.wait()
+            rate_limiter_obj.wait()
 
             params = {
                 "limit": limit,
-                "activityType": "discussion",
+                "activityType": activity_type,
                 "feedType": "user",
                 "entity": username,
             }
@@ -669,313 +651,330 @@ def fetch_user_discussion_activity(
                 if not isinstance(event, dict):
                     continue
 
-                event_ts = parse_ts(event.get("time"))
-                if event_ts is not None:
-                    if oldest_event_ts_seen is None or event_ts < oldest_event_ts_seen:
-                        oldest_event_ts_seen = event_ts
-                    if page_oldest_event_ts is None or event_ts < page_oldest_event_ts:
-                        page_oldest_event_ts = event_ts
-
-                    if year_start <= event_ts <= year_end:
-                        seen_target_year = True
-                    elif event_ts < year_start:
-                        crossed_before_year_start = True
-
-                discussion = event.get("discussionData") or {}
-                if not isinstance(discussion, dict):
-                    continue
-
-                keep, unique_id = event_matches_mode(event, discussion)
-                if not keep or not unique_id:
-                    continue
-
-                dt = extract_timestamp_for_mode(event, discussion)
-                if dt is None:
-                    continue
-
-                if year_start <= dt <= year_end:
-                    seen_target_year = True
-                elif dt < year_start:
-                    continue
-                elif dt > year_end:
-                    continue
-
-                if unique_id in seen_ids:
-                    continue
-
-                seen_ids.add(unique_id)
-                dates.append(dt.tz_localize(None).date())
-
-            if should_stop_after_crossing_year_start(crossed_before_year_start):
+                event_ts = parse_recent_activity_event_ts(event)
+
+                (
+                    oldest_event_ts_seen,
+                    page_oldest_event_ts,
+                    seen_target_year,
+                    crossed_before_year_start,
+                ) = update_recent_activity_progress(
+                    event_ts,
+                    year_start=year_start,
+                    year_end=year_end,
+                    oldest_event_ts_seen=oldest_event_ts_seen,
+                    page_oldest_event_ts=page_oldest_event_ts,
+                    seen_target_year=seen_target_year,
+                    crossed_before_year_start=crossed_before_year_start,
+                )
+
+                processor.process_event(
+                    event=event,
+                    event_ts=event_ts,
+                    year_start=year_start,
+                    year_end=year_end,
+                    state=processor_state,
+                )
+
+            if crossed_before_year_start:
                 logs.append("Stopped after crossing before selected year.")
                 break
 
-            if previous_oldest_event_ts_seen is not None and page_oldest_event_ts is not None:
-                if page_oldest_event_ts >= previous_oldest_event_ts_seen:
-                    stalled_oldest_count += 1
-                else:
-                    stalled_oldest_count = 0
+            previous_oldest_event_ts_seen, stalled_oldest_count = update_stalled_pagination_state(
+                page_oldest_event_ts=page_oldest_event_ts,
+                previous_oldest_event_ts_seen=previous_oldest_event_ts_seen,
+                stalled_oldest_count=stalled_oldest_count,
+            )
 
-            if stalled_oldest_count >= RECENT_ACTIVITY_MAX_STALLED_PAGES:
+            if stalled_oldest_count >= max_stalled_pages:
                 logs.append("Stopped because pagination no longer progressed toward older events.")
                 break
 
-            if page_oldest_event_ts is not None:
-                previous_oldest_event_ts_seen = page_oldest_event_ts
-
             if next_cursor:
                 cursor = next_cursor
             else:
                 skip += limit
 
-        logs.append(f"Fetched {pages_fetched} page(s) from recent-activity.")
-        if oldest_event_ts_seen is not None:
-            logs.append(f"Oldest event reached: {oldest_event_ts_seen.date()}.")
-        logs.append(f"Seen target year: {seen_target_year}.")
-        logs.append(f"Crossed before year start: {crossed_before_year_start}.")
-        logs.append(f"Found {len(dates)} unique {mode} event(s) in {selected_year}.")
-
-        return {
-            "title": title,
-            "fetched": True,
-            "dates": dates,
-            "count": len(dates),
-            "logs": "\n".join(logs),
-        }
+        return processor.build_result(
+            username=username,
+            selected_year=selected_year,
+            state=processor_state,
+            logs=logs,
+            pages_fetched=pages_fetched,
+            oldest_event_ts_seen=oldest_event_ts_seen,
+            seen_target_year=seen_target_year,
+            crossed_before_year_start=crossed_before_year_start,
+        )
 
     except Exception as e:
-        return {
-            "title": title,
-            "fetched": True,
-            "dates": [],
-            "count": 0,
-            "logs": f"{title} fetch failed: {str(e)}",
-        }
-
-
-def fetch_all_community_activity(username: str, selected_year: int):
-    username = (username or "").strip()
-    selected_year = int(selected_year)
-
-    cache_key = (username.lower(), selected_year, "community")
-    cached = get_cache(cache_key)
-    if cached:
-        return cached
-
-    if not username:
-        state = empty_community_state(selected_year)
-        state["fetched"] = True
-        state["logs"] = "No username provided."
-        set_cache(cache_key, state)
-        return state
-
-    pr_opened = fetch_user_discussion_activity(username, selected_year, "pr_opened")
-    pr_activity = fetch_user_discussion_activity(username, selected_year, "pr_activity")
-    thread_opened = fetch_user_discussion_activity(username, selected_year, "thread_opened")
-    discussion_activity = fetch_user_discussion_activity(username, selected_year, "discussion_activity")
-
-    combined_logs = "\n\n".join(
-        [
-            "[Pull Requests]\n" + pr_opened["logs"],
-            "[PR Activity]\n" + pr_activity["logs"],
-            "[Community Threads]\n" + thread_opened["logs"],
-            "[Community Activity]\n" + discussion_activity["logs"],
-        ]
-    )
-
-    result = {
-        "username": username,
-        "selected_year": selected_year,
-        "fetched": True,
-        "pr_opened": pr_opened,
-        "pr_activity": pr_activity,
-        "thread_opened": thread_opened,
-        "discussion_activity": discussion_activity,
-        "logs": combined_logs,
-    }
-
-    set_cache(cache_key, result)
-    return result
+        return processor.build_error_result(
+            username=username,
+            selected_year=selected_year,
+            error=e,
+            logs=logs,
+        )
 
 
 # ============================================================
-# Bucketed recent-activity fetch
+# Recent-activity processors
 # ============================================================
-def fetch_bucketed_recent_activity(
-    username: str,
-    selected_year: int,
-    cache_scope: str,
-    activity_type: str,
-    bucket_titles: dict[str, str],
-    bucket_log_labels: dict[str, str],
-    empty_logs_text: str,
-    classifier,
-):
-    username = (username or "").strip()
-    selected_year = int(selected_year)
-
-    cache_key = (username.lower(), selected_year, cache_scope)
-    cached = get_cache(cache_key)
-    if cached:
-        return cached
-
-    if not username:
-        state = empty_bucketed_activity_state(selected_year, empty_logs_text, bucket_titles)
-        state["fetched"] = True
-        state["logs"] = "No username provided."
-        set_cache(cache_key, state)
-        return state
-
-    bucket_dates = {key: [] for key in bucket_titles}
-    bucket_seen_ids = {key: set() for key in bucket_titles}
-
-    cursor = None
-    skip = 0
-    limit = RECENT_ACTIVITY_LIMIT
-    pages_fetched = 0
-
-    seen_target_year = False
-    crossed_before_year_start = False
-    oldest_event_ts_seen = None
-    previous_oldest_event_ts_seen = None
-    stalled_oldest_count = 0
-    seen_cursors = set()
-    seen_page_signatures = set()
-
-    logs = []
-
-    year_start, year_end = get_year_bounds(selected_year)
-    session = build_session("application/json")
-
-    try:
-        while True:
-            if pages_fetched >= RECENT_ACTIVITY_MAX_PAGES:
-                logs.append("Stopped due to safety max pages.")
-                break
-
-            recent_activity_rate_limiter.wait()
-
-            params = {
-                "limit": limit,
-                "activityType": activity_type,
-                "feedType": "user",
-                "entity": username,
-            }
-            if cursor:
-                params["cursor"] = cursor
-            else:
-                params["skip"] = skip
-
-            response = session.get(
-                "https://huggingface.co/api/recent-activity",
-                params=params,
-                timeout=20,
-            )
-            response.raise_for_status()
-
-            payload = response.json()
-            pages_fetched += 1
+class CommunityActivityProcessor:
+    TITLE_MAP = {
+        "pr_opened": "Pull Requests",
+        "pr_activity": "PR Activity",
+        "thread_opened": "Community Threads",
+        "discussion_activity": "Community Activity",
+    }
+    LOG_LABEL_MAP = {
+        "pr_opened": "unique pr_opened event(s)",
+        "pr_activity": "unique pr_activity event(s)",
+        "thread_opened": "unique thread_opened event(s)",
+        "discussion_activity": "unique discussion_activity event(s)",
+    }
+    OPEN_EVENT_EPSILON_SECONDS = 90
 
-            events, next_cursor = normalize_recent_activity_payload(payload)
+    def __init__(self, username: str):
+        self.username = (username or "").strip()
 
-            if not events:
-                logs.append("No more events.")
-                break
+    def init_state(self):
+        return {
+            "dates": {
+                "pr_opened": [],
+                "pr_activity": [],
+                "thread_opened": [],
+                "discussion_activity": [],
+            },
+            "seen_ids": {
+                "pr_opened": set(),
+                "pr_activity": set(),
+                "thread_opened": set(),
+                "discussion_activity": set(),
+            },
+        }
 
-            page_signature = build_events_page_signature(events)
-            if page_signature in seen_page_signatures:
-                logs.append("Stopped because recent-activity returned a repeated page.")
-                break
-            seen_page_signatures.add(page_signature)
+    def extract_author_name(self, discussion: dict) -> str | None:
+        author = discussion.get("author")
+        if isinstance(author, str):
+            return author
+        if isinstance(author, dict):
+            for key in ["name", "username", "login"]:
+                value = author.get(key)
+                if isinstance(value, str):
+                    return value
+        return None
 
-            if next_cursor:
-                if next_cursor in seen_cursors:
-                    logs.append("Stopped because recent-activity returned a repeated cursor.")
-                    break
-                seen_cursors.add(next_cursor)
+    def parse_ts(self, raw_value):
+        if not raw_value:
+            return None
+        try:
+            return pd.to_datetime(raw_value, utc=True)
+        except Exception:
+            return None
 
-            page_oldest_event_ts = None
+    def classify_event(self, event: dict, discussion: dict) -> tuple[str | None, str | None]:
+        is_pr = bool(discussion.get("isPullRequest", False))
+        author_name = self.extract_author_name(discussion)
+        repo_id = event.get("repoId", "unknown-repo")
+        discussion_num = discussion.get("num", "unknown-num")
+        event_id = event.get("eventId", "unknown-event")
 
-            for event in events:
-                if not isinstance(event, dict):
-                    continue
+        if is_pr:
+            if author_name and author_name.lower() == self.username.lower():
+                return "pr_opened", f"{repo_id}::pr::{discussion_num}"
+            return "pr_activity", f"event::{event_id}"
 
-                raw_time = event.get("time")
-                if not raw_time:
-                    continue
+        if author_name and author_name.lower() == self.username.lower():
+            return "thread_opened", f"{repo_id}::thread::{discussion_num}"
+        return "discussion_activity", f"event::{event_id}"
 
-                try:
-                    event_ts = pd.to_datetime(raw_time, utc=True)
-                except Exception:
-                    continue
+    def extract_business_timestamp(self, bucket_key: str, event: dict, discussion: dict):
+        event_ts = self.parse_ts(event.get("time"))
+        created_ts = self.parse_ts(discussion.get("createdAt"))
 
-                if oldest_event_ts_seen is None or event_ts < oldest_event_ts_seen:
-                    oldest_event_ts_seen = event_ts
-                if page_oldest_event_ts is None or event_ts < page_oldest_event_ts:
-                    page_oldest_event_ts = event_ts
+        if bucket_key in {"pr_opened", "thread_opened"}:
+            return created_ts
 
-                if year_start <= event_ts <= year_end:
-                    seen_target_year = True
-                elif event_ts < year_start:
-                    crossed_before_year_start = True
+        if bucket_key in {"pr_activity", "discussion_activity"}:
+            if event_ts is None:
+                return None
+            if created_ts is not None:
+                delta_seconds = abs((event_ts - created_ts).total_seconds())
+                if delta_seconds <= self.OPEN_EVENT_EPSILON_SECONDS:
+                    return None
+            return event_ts
 
-                if event_ts > year_end:
-                    continue
+        return None
 
-                if event_ts < year_start:
-                    continue
+    def process_event(
+        self,
+        *,
+        event: dict,
+        event_ts,
+        year_start,
+        year_end,
+        state: dict,
+    ):
+        _ = event_ts
 
-                bucket_key, unique_id = classifier(event)
+        discussion = event.get("discussionData") or {}
+        if not isinstance(discussion, dict):
+            return
 
-                if not bucket_key or not unique_id:
-                    continue
+        bucket_key, unique_id = self.classify_event(event, discussion)
+        if not bucket_key or not unique_id:
+            return
 
-                if bucket_key not in bucket_dates:
-                    continue
+        business_ts = self.extract_business_timestamp(bucket_key, event, discussion)
+        if business_ts is None:
+            return
 
-                if unique_id in bucket_seen_ids[bucket_key]:
-                    continue
+        if not (year_start <= business_ts <= year_end):
+            return
 
-                bucket_seen_ids[bucket_key].add(unique_id)
-                bucket_dates[bucket_key].append(event_ts.tz_localize(None).date())
+        if unique_id in state["seen_ids"][bucket_key]:
+            return
 
-            if should_stop_after_crossing_year_start(crossed_before_year_start):
-                logs.append("Stopped after crossing before selected year.")
-                break
+        state["seen_ids"][bucket_key].add(unique_id)
+        state["dates"][bucket_key].append(business_ts.tz_localize(None).date())
 
-            if previous_oldest_event_ts_seen is not None and page_oldest_event_ts is not None:
-                if page_oldest_event_ts >= previous_oldest_event_ts_seen:
-                    stalled_oldest_count += 1
-                else:
-                    stalled_oldest_count = 0
+    def build_bucket(self, key: str, dates: list, selected_year: int):
+        return {
+            "title": self.TITLE_MAP[key],
+            "fetched": True,
+            "dates": dates,
+            "count": len(dates),
+            "logs": f"Found {len(dates)} {self.LOG_LABEL_MAP[key]} in {selected_year}.",
+        }
 
-            if stalled_oldest_count >= RECENT_ACTIVITY_MAX_STALLED_PAGES:
-                logs.append("Stopped because pagination no longer progressed toward older events.")
-                break
+    def build_result(
+        self,
+        *,
+        username: str,
+        selected_year: int,
+        state: dict,
+        logs: list[str],
+        pages_fetched: int,
+        oldest_event_ts_seen,
+        seen_target_year: bool,
+        crossed_before_year_start: bool,
+    ):
+        logs = list(logs)
+        logs.append(f"Fetched {pages_fetched} page(s) from recent-activity.")
+        if oldest_event_ts_seen is not None:
+            logs.append(f"Oldest event reached: {oldest_event_ts_seen.date()}.")
+        logs.append(f"Seen target year: {seen_target_year}.")
+        logs.append(f"Crossed before year start: {crossed_before_year_start}.")
+        logs.append(
+            " | ".join(
+                f"{self.TITLE_MAP[key]}: {len(state['dates'][key])}"
+                for key in ["pr_opened", "pr_activity", "thread_opened", "discussion_activity"]
+            )
+        )
 
-            if page_oldest_event_ts is not None:
-                previous_oldest_event_ts_seen = page_oldest_event_ts
+        combined_logs = "\n\n".join(
+            [
+                "[Pull Requests]\n" + self.build_bucket("pr_opened", state["dates"]["pr_opened"], selected_year)["logs"],
+                "[PR Activity]\n" + self.build_bucket("pr_activity", state["dates"]["pr_activity"], selected_year)["logs"],
+                "[Community Threads]\n" + self.build_bucket("thread_opened", state["dates"]["thread_opened"], selected_year)["logs"],
+                "[Community Activity]\n" + self.build_bucket("discussion_activity", state["dates"]["discussion_activity"], selected_year)["logs"],
+                "[Fetch]\n" + "\n".join(logs),
+            ]
+        )
 
-            if next_cursor:
-                cursor = next_cursor
-            else:
-                skip += limit
+        return {
+            "username": username,
+            "selected_year": selected_year,
+            "fetched": True,
+            "pr_opened": self.build_bucket("pr_opened", state["dates"]["pr_opened"], selected_year),
+            "pr_activity": self.build_bucket("pr_activity", state["dates"]["pr_activity"], selected_year),
+            "thread_opened": self.build_bucket("thread_opened", state["dates"]["thread_opened"], selected_year),
+            "discussion_activity": self.build_bucket("discussion_activity", state["dates"]["discussion_activity"], selected_year),
+            "logs": combined_logs,
+        }
 
-        result = {
+    def build_error_result(
+        self,
+        *,
+        username: str,
+        selected_year: int,
+        error,
+        logs: list[str],
+    ):
+        _ = logs
+        return {
             "username": username,
             "selected_year": selected_year,
             "fetched": True,
-            "logs": "",
+            "pr_opened": build_failed_bucket("Pull Requests"),
+            "pr_activity": build_failed_bucket("PR Activity"),
+            "thread_opened": build_failed_bucket("Community Threads"),
+            "discussion_activity": build_failed_bucket("Community Activity"),
+            "logs": f"Community fetch failed: {str(error)}",
         }
 
-        for key, title in bucket_titles.items():
-            result[key] = build_fetched_bucket(
-                title=title,
-                dates=bucket_dates[key],
-                selected_year=selected_year,
-                label=bucket_log_labels[key],
-            )
 
+class BucketedActivityProcessor:
+    def __init__(
+        self,
+        *,
+        bucket_titles: dict[str, str],
+        bucket_log_labels: dict[str, str],
+        classifier,
+        error_label: str,
+    ):
+        self.bucket_titles = bucket_titles
+        self.bucket_log_labels = bucket_log_labels
+        self.classifier = classifier
+        self.error_label = error_label
+
+    def init_state(self):
+        return {
+            "bucket_dates": {key: [] for key in self.bucket_titles},
+            "bucket_seen_ids": {key: set() for key in self.bucket_titles},
+        }
+
+    def process_event(
+        self,
+        *,
+        event: dict,
+        event_ts,
+        year_start,
+        year_end,
+        state: dict,
+    ):
+        if event_ts is None:
+            return
+
+        if not (year_start <= event_ts <= year_end):
+            return
+
+        bucket_key, unique_id = self.classifier(event)
+
+        if not bucket_key or not unique_id:
+            return
+
+        if bucket_key not in state["bucket_dates"]:
+            return
+
+        if unique_id in state["bucket_seen_ids"][bucket_key]:
+            return
+
+        state["bucket_seen_ids"][bucket_key].add(unique_id)
+        state["bucket_dates"][bucket_key].append(event_ts.tz_localize(None).date())
+
+    def build_result(
+        self,
+        *,
+        username: str,
+        selected_year: int,
+        state: dict,
+        logs: list[str],
+        pages_fetched: int,
+        oldest_event_ts_seen,
+        seen_target_year: bool,
+        crossed_before_year_start: bool,
+    ):
+        logs = list(logs)
         logs.append(f"Fetched {pages_fetched} page(s) from recent-activity.")
         if oldest_event_ts_seen is not None:
             logs.append(f"Oldest event reached: {oldest_event_ts_seen.date()}.")
@@ -983,30 +982,54 @@ def fetch_bucketed_recent_activity(
         logs.append(f"Crossed before year start: {crossed_before_year_start}.")
         logs.append(
             " | ".join(
-                f"{bucket_titles[key]}: {len(bucket_dates[key])}"
-                for key in bucket_titles
+                f"{self.bucket_titles[key]}: {len(state['bucket_dates'][key])}"
+                for key in self.bucket_titles
             )
         )
 
-        result["logs"] = "\n".join(logs)
+        result = {
+            "username": username,
+            "selected_year": selected_year,
+            "fetched": True,
+            "logs": "\n".join(logs),
+        }
+
+        for key, title in self.bucket_titles.items():
+            dates = state["bucket_dates"][key]
+            result[key] = build_fetched_bucket(
+                title=title,
+                dates=dates,
+                selected_year=selected_year,
+                label=self.bucket_log_labels[key],
+            )
 
-        set_cache(cache_key, result)
         return result
 
-    except Exception as e:
-        failed = {
+    def build_error_result(
+        self,
+        *,
+        username: str,
+        selected_year: int,
+        error,
+        logs: list[str],
+    ):
+        _ = logs
+        result = {
             "username": username,
             "selected_year": selected_year,
             "fetched": True,
-            "logs": f"{cache_scope.capitalize()} fetch failed: {str(e)}",
+            "logs": f"{self.error_label} fetch failed: {str(error)}",
         }
 
-        for key, title in bucket_titles.items():
-            failed[key] = build_failed_bucket(title)
+        for key, title in self.bucket_titles.items():
+            result[key] = build_failed_bucket(title)
 
-        return failed
+        return result
 
 
+# ============================================================
+# Activity classifiers + config
+# ============================================================
 def classify_upvote_event(event: dict):
     target_type = event.get("targetType")
     target = event.get("target") or {}
@@ -1157,19 +1180,76 @@ def empty_single_section_state(section_key: str, selected_year: int):
     )
 
 
-def fetch_single_section_activity(section_key: str, username: str, selected_year: int):
-    cfg = SINGLE_SECTION_CONFIG[section_key]
-    return fetch_bucketed_recent_activity(
+# ============================================================
+# Public activity fetch functions
+# ============================================================
+def fetch_all_community_activity(username: str, selected_year: int):
+    username = (username or "").strip()
+    selected_year = int(selected_year)
+
+    cache_key = (username.lower(), selected_year, "community")
+    cached = get_cache(cache_key)
+    if cached:
+        return cached
+
+    if not username:
+        state = empty_community_state(selected_year)
+        state["fetched"] = True
+        state["logs"] = "No username provided."
+        set_cache(cache_key, state)
+        return state
+
+    processor = CommunityActivityProcessor(username=username)
+
+    result = collect_recent_activity(
         username=username,
         selected_year=selected_year,
-        cache_scope=cfg["cache_scope"],
-        activity_type=cfg["activity_type"],
+        activity_type="discussion",
+        processor=processor,
+        build_session_fn=build_session,
+        rate_limiter_obj=recent_activity_rate_limiter,
+    )
+
+    set_cache(cache_key, result)
+    return result
+
+
+def fetch_single_section_activity(section_key: str, username: str, selected_year: int):
+    username = (username or "").strip()
+    selected_year = int(selected_year)
+    cfg = SINGLE_SECTION_CONFIG[section_key]
+
+    cache_key = (username.lower(), selected_year, cfg["cache_scope"])
+    cached = get_cache(cache_key)
+    if cached:
+        return cached
+
+    if not username:
+        state = empty_single_section_state(section_key, selected_year)
+        state["fetched"] = True
+        state["logs"] = "No username provided."
+        set_cache(cache_key, state)
+        return state
+
+    processor = BucketedActivityProcessor(
         bucket_titles=cfg["bucket_titles"],
         bucket_log_labels=cfg["bucket_log_labels"],
-        empty_logs_text=cfg["empty_logs_text"],
         classifier=cfg["classifier"],
+        error_label=cfg["cache_scope"].capitalize(),
     )
 
+    result = collect_recent_activity(
+        username=username,
+        selected_year=selected_year,
+        activity_type=cfg["activity_type"],
+        processor=processor,
+        build_session_fn=build_session,
+        rate_limiter_obj=recent_activity_rate_limiter,
+    )
+
+    set_cache(cache_key, result)
+    return result
+
 
 # ============================================================
 # Contribution grid helpers