Engagement Metrics Collector

.github/workflows/dev.yml

@@ -99,4 +99,4 @@ jobs:
         with:
           path: cutepets.log
           retention-days: 14
-          archive: false
+          archive: false

abstractions.py

@@ -139,3 +139,35 @@ def format_post(self, pet: AdoptablePet) -> Post:
                 pet.breed.lower().replace(" ", ""),
             ],
         )
+
+
+# =============================================================================
+# Engagement Metric Collector Interface
+# =============================================================================
+
+
+@dataclass
+class PostMetrics:
+    """Point-in-time engagement counts for a published post."""
+
+    collected_at: str
+    likes: int | None = None
+    reposts: int | None = None
+    comments: int | None = None
+
+
+class MetricCollector(ABC):
+    """Interface for collecting engagement metrics from a social platform."""
+
+    @property
+    @abstractmethod
+    def platform_name(self) -> str:
+        """Return the platform name used in persisted post records."""
+        ...
+
+    @abstractmethod
+    def fetch_metrics(
+        self, post_id: str, post_url: str | None = None
+    ) -> PostMetrics | None:
+        """Return a metric snapshot, or None when the post cannot be read."""
+        ...

main.py

@@ -1,39 +1,44 @@
 import argparse
+from dataclasses import asdict
+from datetime import datetime, timedelta, timezone
 import json
+import logging
 import os
+from pathlib import Path
+import pprint
 import random
 import sys
 import traceback
-import logging
-import pprint
-from datetime import datetime, timedelta, timezone
-from pathlib import Path
 
 import requests
 
 from adoption_sources import SourceManual, SourceRescueGroups
+from metric_collectors.bluesky import CollectorBluesky
+from metric_collectors.instagram import CollectorInstagram
+from metric_collectors.mastodon import CollectorMastodon
 from social_posters.bluesky import PosterBluesky
 from social_posters.debug import PosterDebug
 from social_posters.instagram import PosterInstagram
 from social_posters.mastodon import PosterMastodon
 
+
 file_handler = logging.FileHandler("cutepets.log")
 file_handler.setLevel(logging.DEBUG)
 console_handler = logging.StreamHandler(sys.stdout)
 console_handler.setLevel(logging.INFO)
 
 logging.basicConfig(
     level=logging.DEBUG,
-    format='%(asctime)s [%(levelname)s] %(name)s - %(message)s',
-    datefmt='%Y-%m-%d %H:%M:%S',
+    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
+    datefmt="%Y-%m-%d %H:%M:%S",
     handlers=[file_handler, console_handler],
 )
 
 logger = logging.getLogger(__name__)
 
+
 def main():
-
-    logger.info('Log started')
+    logger.info("Log started")
     parser = argparse.ArgumentParser()
     parser.add_argument("--debugsources", action="store_true") # this defaults to False
     parser.add_argument("--debugposters", action="store_true") # this defaults to False
@@ -43,8 +48,9 @@ def main():
     try:
         sources = create_sources(debug=args.debugsources)
         posters = create_posters(debug=args.debugposters)
+        collectors = create_collectors(debug=args.debugposters)
 
-        run(sources, posters)
+        run(sources, posters, collectors)
     except Exception:
         notify_slack_of_exception(traceback.format_exc())
         raise
@@ -54,31 +60,30 @@ def create_posters(debug=False):
     if debug:
         return [PosterDebug()]
 
-    posters = []
-    posters.append(PosterMastodon())
-    posters.append(PosterBluesky())
-    posters.append(PosterInstagram())
-    return posters
+    return [PosterMastodon(), PosterBluesky(), PosterInstagram()]
 
 
+def create_collectors(debug=False):
+    if debug:
+        return []
+
+    return [CollectorBluesky(), CollectorMastodon(), CollectorInstagram()]
 
 
 def create_sources(debug=False):
     if debug:
         cat_fixture_path = Path(__file__).parent / "tests" / "fixtures" / "sample_cats.json"
-        with open(cat_fixture_path) as f:
-            cat_animals = json.load(f)
+        with cat_fixture_path.open() as fixture_file:
+            cat_animals = json.load(fixture_file)
         return [
             SourceManual(species="dog"),
             SourceManual(species="cat", animals=cat_animals),
         ]
 
-    sources = []
-    sources.append(SourceRescueGroups())
-    return sources
+    return [SourceRescueGroups()]
 
 
-def run(sources, posters):
+def run(sources, posters, collectors=None, database_path="database.json"):
     pets = []
     for source in sources:
         try:
@@ -87,65 +92,167 @@ def run(sources, posters):
             raise SystemExit(str(exc)) from exc
 
     logger.info("Fetched %d records", len(pets))
-    pet = pick_pet(pets)
+    pet = pick_pet(pets, database_path=database_path)
+    results = []
+
     if not pet:
         logger.error("No pets available to post.")
-        return []
     else:
-        pet_format = pprint.pformat(pet)
-        logger.info("Picked pet %s", pet_format)
+        logger.info("Picked pet %s", pprint.pformat(pet))
+        results, published_results = publish_posts(pet, posters)
+        record_publish_results(pet, published_results, database_path=database_path)
+
+    collect_metrics(collectors or [], database_path=database_path)
+    return results
 
-    if not posters:
-        logger.error("No social media credentials set; skipping post.")
-        return []
 
+def publish_posts(pet, posters):
     results = []
-    for poster in posters:
-        post = poster.format_post(pet)
-        result = poster.publish(post)
-        results.append(result)
+    published_results = []
+
+    if not posters:
+        logger.error("No social media credentials set; skipping post.")
+    else:
+        for poster in posters:
+            post = poster.format_post(pet)
+            result = poster.publish(post)
+            results.append(result)
+            published_results.append((poster, result))
+            if not result.success:
+                logger.error(
+                    "%s post failed: %s",
+                    poster.platform_name,
+                    result.error_message,
+                )
+            else:
+                logger.info("%s post published.", poster.platform_name)
+
+    return results, published_results
+
+
+def pick_pet(pets, database_path="database.json"):
+    data = _read_database(database_path)
+    posted_pet_ids = {
+        posted_pet["pet_id"] for posted_pet in data.get("posted_pets", [])
+    }
+    eligible = [
+        pet
+        for pet in pets
+        if pet.image_url
+        and pet.adoption_url
+        and pet.pet_id not in posted_pet_ids
+    ]
+    if not eligible:
+        raise ValueError("No eligible pet found")
+
+    return random.choice(eligible)
+
+
+def record_publish_results(pet, results, database_path="database.json"):
+    data = _read_database(database_path)
+    posted_pets = data.setdefault("posted_pets", [])
+    posts = data.setdefault("posts", [])
+    posted_at = datetime.now(timezone.utc).isoformat()
+
+    posted_pets.append(
+        {"name": pet.name, "pet_id": pet.pet_id, "posted_at": posted_at}
+    )
+    for poster, result in results:
         if not result.success:
-            logger.error(f"{poster.platform_name} post failed: {result.error_message}")
-        else:
-            logger.info(f"{poster.platform_name} post published.")
+            continue
+        posts.append(
+            {
+                "pet_id": pet.pet_id,
+                "platform": poster.platform_name,
+                "post_id": result.post_id,
+                "post_url": result.post_url,
+                "posted_at": posted_at,
+                "metrics": [],
+            }
+        )
+
+    cutoff = datetime.now(timezone.utc) - timedelta(weeks=12)
+    data["posted_pets"] = [
+        item
+        for item in posted_pets
+        if datetime.fromisoformat(item["posted_at"]) >= cutoff
+    ]
+    data["posts"] = [
+        item
+        for item in posts
+        if datetime.fromisoformat(item["posted_at"]) >= cutoff
+    ]
+    _write_database(database_path, data)
+
+
+def collect_metrics(collectors, database_path="database.json", window_days=14):
+    try:
+        data = _read_database(database_path)
+        posts = data.get("posts", [])
+        if not posts:
+            return
+
+        collectors_by_platform = {
+            collector.platform_name: collector for collector in collectors
+        }
+        cutoff = datetime.now(timezone.utc) - timedelta(days=window_days)
+        updated = False
+
+        for entry in posts:
+            try:
+                if datetime.fromisoformat(entry["posted_at"]) < cutoff:
+                    continue
+
+                collector = collectors_by_platform.get(entry.get("platform"))
+                if collector is None:
+                    continue
+
+                metrics = collector.fetch_metrics(
+                    entry["post_id"], entry.get("post_url")
+                )
+                if metrics is None:
+                    continue
+
+                snapshot = asdict(metrics)
+                snapshot["collected_at"] = datetime.now(timezone.utc).isoformat()
+                entry.setdefault("metrics", []).append(snapshot)
+                updated = True
+            except Exception as exc:
+                platform = entry.get("platform", "unknown platform")
+                post_id = entry.get("post_id", "unknown post")
+                logger.error(
+                    "%s metric collection failed for %s: %s",
+                    platform,
+                    post_id,
+                    exc,
+                )
+
+        if updated:
+            _write_database(database_path, data)
+    except Exception as exc:
+        logger.error("Metric collection failed: %s", exc)
+
+
+def _read_database(database_path):
+    path = Path(database_path)
+    if not path.exists() or path.stat().st_size == 0:
+        return {}
 
-    return results
+    try:
+        with path.open() as database_file:
+            return json.load(database_file)
+    except (json.JSONDecodeError, ValueError) as exc:
+        logger.error("%s:%s", type(exc).__name__, exc)
+        traceback.print_exc()
+        return {}
 
 
-def pick_pet(pets):
-    Path("database.json").touch(exist_ok=True)
-    # Open file
-    with open("database.json", "r+") as f:
-        # Load json
-        try:
-            data = json.load(f)
-        except (json.JSONDecodeError, ValueError) as e:
-            logger.error(f"{type(e).__name__}:{e}")
-            traceback.print_exc()
-            data = {}
-
-        if "posted_pets" in data:
-            posted_pet_ids =  {posted_pet["pet_id"] for posted_pet in data["posted_pets"]}
-        else:
-            posted_pet_ids = {}
-            data["posted_pets"] = []
-        # Check pet has an image, adoption url, and has not been posted
-        eligible = [pet for pet in pets if pet.image_url and pet.adoption_url and pet.pet_id not in posted_pet_ids]
-        if not eligible:
-            raise ValueError("No elligible pet found")
-
-        selected_pet = random.choice(eligible)
-        # Add pet ID to list of posted pets
-        data["posted_pets"].append({"name": selected_pet.name, "pet_id": selected_pet.pet_id, "posted_at": datetime.now(timezone.utc).isoformat()})
-        # Remove old pets
-        cutoff = datetime.now(timezone.utc) - timedelta(weeks=12)
-        recent_pets = [item for item in data["posted_pets"] if datetime.fromisoformat(item['posted_at']) > cutoff]
-        data["posted_pets"] = recent_pets
-        # Export json
-        f.seek(0)
-        json.dump(data, f, indent=4)
-        f.truncate()
-        return selected_pet
+def _write_database(database_path, data):
+    path = Path(database_path)
+    temporary_path = path.with_name(f"{path.name}.tmp")
+    with temporary_path.open("w") as database_file:
+        json.dump(data, database_file, indent=4)
+    temporary_path.replace(path)
 
 
 # Slack incoming-webhook messages have a ~40k-char limit; cap the traceback
@@ -183,7 +290,7 @@ def notify_slack_of_exception(traceback_text):
         response = requests.post(webhook_url, json={"text": text}, timeout=10)
         response.raise_for_status()
     except Exception as slack_exc:
-        logger.error(f"Failed to post Slack alert: {slack_exc}")
+        logger.error("Failed to post Slack alert: %s", slack_exc)
 
 
 if __name__ == "__main__":