""" Task #20 — Sample 50 posts from data/saved_posts.json for the benchmark fixture. Sampling strategy: - 30 food candidates (keyword-matched), diversified by creator and hashtag set - 20 non-food candidates (keyword-matched) - Edge cases included: thin captions (<60 chars), no hashtags, ambiguous posts - Deterministic via fixed random seed Run from the project root: python3 tests/create_fixture.py """ import json import random import sys from pathlib import Path # Allow importing from the project root sys.path.insert(0, str(Path(__file__).parent.parent)) from pipeline.extract import load_posts SEED = 42 FOOD_KEYWORDS = { "food", "eats", "eating", "restaurant", "cafe", "bakery", "brunch", "dinner", "lunch", "foodie", "ramen", "sushi", "pizza", "burger", "taco", "coffee", "matcha", "boba", "tea", "bistro", "bar", "grill", "kitchen", "eatery", "diner", "noodle", "bbq", "dessert", "pastry", "croissant", "bread", "pho", "dim", "curry", "korean", "japanese", "chinese", "italian", "mexican", "thai", "vietnamese", "indian", "mediterranean", "omakase", "izakaya", "tapas", "taproom", "brewery", "brunchspot", "foodstagram", } NON_FOOD_KEYWORDS = { "travel", "trip", "hike", "hiking", "outdoors", "fitness", "gym", "workout", "fashion", "style", "outfit", "ootd", "skincare", "beauty", "nature", "landscape", "architecture", "museum", "art", "photography", "recipe", "cook", "homecook", "science", "biology", "meme", "sunscreen", "flipkart", "scramble", "vertical", } def _is_food_candidate(post: dict) -> bool: tags = {t.lower() for t in post["hashtags"]} cap = post["caption"].lower() return bool(tags & FOOD_KEYWORDS or any(k in cap for k in FOOD_KEYWORDS)) def _is_nonfood_candidate(post: dict) -> bool: tags = {t.lower() for t in post["hashtags"]} cap = post["caption"].lower() return bool(tags & NON_FOOD_KEYWORDS or any(k in cap for k in NON_FOOD_KEYWORDS)) def _is_edge_case(post: dict) -> bool: """Thin caption or no hashtags — intentionally tricky posts.""" return len(post["caption"]) < 60 or len(post["hashtags"]) == 0 def _score_food_diversity(post: dict, chosen: list[dict]) -> float: """Prefer posts from creators not already selected, and novel hashtags.""" chosen_creators = {p["creator"] for p in chosen} chosen_tags = {t for p in chosen for t in p["hashtags"]} creator_bonus = 0.0 if post["creator"] in chosen_creators else 1.0 new_tags = len(set(post["hashtags"]) - chosen_tags) return creator_bonus + new_tags * 0.1 def sample_posts( all_posts: list[dict], n_food: int = 30, n_nonfood: int = 20, seed: int = SEED, ) -> list[dict]: rng = random.Random(seed) food = [p for p in all_posts if _is_food_candidate(p) and not _is_nonfood_candidate(p)] nonfood = [p for p in all_posts if _is_nonfood_candidate(p) and not _is_food_candidate(p)] ambiguous = [p for p in all_posts if not _is_food_candidate(p) and not _is_nonfood_candidate(p)] # Guaranteed edge-case slots: 4 food edge cases, 2 non-food edge cases food_edges = [p for p in food if _is_edge_case(p)] nonfood_edges = [p for p in nonfood if _is_edge_case(p)] ambiguous_edges = [p for p in ambiguous if _is_edge_case(p)] rng.shuffle(food_edges) rng.shuffle(nonfood_edges) chosen_food_edges = food_edges[:4] chosen_nonfood_edges = nonfood_edges[:2] # Fill remaining food slots with diversity-scored greedy selection remaining_food = [p for p in food if p not in chosen_food_edges] rng.shuffle(remaining_food) chosen_food = list(chosen_food_edges) for post in remaining_food: if len(chosen_food) >= n_food: break chosen_food.append(post) # Fill remaining non-food slots: mix keyword-matched and some ambiguous remaining_nonfood = [p for p in nonfood if p not in chosen_nonfood_edges] rng.shuffle(remaining_nonfood) chosen_nonfood = list(chosen_nonfood_edges) for post in remaining_nonfood: if len(chosen_nonfood) >= n_nonfood: break chosen_nonfood.append(post) # If non-food pool was too small, pad from ambiguous (labelled later by Opus) if len(chosen_nonfood) < n_nonfood: rng.shuffle(ambiguous_edges) rng.shuffle(ambiguous) extras = ambiguous_edges + [p for p in ambiguous if p not in ambiguous_edges] for post in extras: if len(chosen_nonfood) >= n_nonfood: break if post not in chosen_nonfood: chosen_nonfood.append(post) sampled = chosen_food + chosen_nonfood rng.shuffle(sampled) print(f"Sampled {len(chosen_food)} food candidates + {len(chosen_nonfood)} non-food candidates") print(f" Food edge cases included: {sum(1 for p in chosen_food if _is_edge_case(p))}") print(f" Non-food edge cases included: {sum(1 for p in chosen_nonfood if _is_edge_case(p))}") print(f" Unique creators: {len({p['creator'] for p in sampled})}") return sampled def main() -> None: root = Path(__file__).parent.parent json_path = root / "data" / "saved_posts.json" out_path = root / "tests" / "fixtures" / "sampled_posts.json" all_posts = load_posts(str(json_path)) print(f"Loaded {len(all_posts)} posts from {json_path.name}") sampled = sample_posts(all_posts) # Write only the raw fields — no ground-truth labels yet records = [ { "url": p["url"], "caption": p["caption"], "hashtags": p["hashtags"], "creator": p["creator"], "saved_at": p["saved_at"], } for p in sampled ] out_path.parent.mkdir(parents=True, exist_ok=True) with open(out_path, "w", encoding="utf-8") as f: json.dump(records, f, indent=2, ensure_ascii=False) print(f"\nWrote {len(records)} posts → {out_path.relative_to(root)}") if __name__ == "__main__": main()