每月为 10,000 个页面配上真实照片——只需 500 次 API 调用

curl -sG "https://api.pexafy.com/api/v1/search/photos" \
  -H "X-Api-Key: $PEXAFY_API_KEY" \
  --data-urlencode "q=a woman signing paperwork with an insurance agent at a kitchen table in her home" \
  --data-urlencode "per_page=100" \
  --data-urlencode "orientation=landscape" \
  --data-urlencode "score_threshold=0.5" \
  --data-urlencode "fields=photo_id,urls,width,height,blur_hash,alt_description,attribution"

# → { "success": true, "data": [ …最多 100 张照片… ],
#     "pagination": { "per_page": 100, "has_more": true, "next_cursor": "eyJyc…" },
#     "meta": { "took_ms": 122.33, "request_id": "7ff1fb9b-…" } }

                     ┌─────────── 每周运行，约 500 次请求 ───────────┐
  主题集群       ──▶ POOL   每个集群搜一次，per_page=100
                     │       └─▶ 每个集群存 100 行照片
                     └──────────────────┬───────────────────────────────┘
                                        ▼
                              image_pool 表
                        (cluster, photo_id, urls, blur_hash,
                         alt, credit, used_by_page, used_at)
                                        │
                     ┌──────────────────┴──── 每次发布运行，0 次请求 ───┐
  文章       ─────────▶ ASSIGN  为该集群挑出最佳未使用行，
                     │        标记为已用，渲染
                     └────────────────────────────────────────────────────────┘

import asyncio, os, time, httpx

SEARCH = "https://api.pexafy.com/api/v1/search/photos"
KEY    = os.environ["PEXAFY_API_KEY"]
FIELDS = "photo_id,urls,width,height,blur_hash,alt_description,attribution"

# Free = 20 req/min，保持低一档。节流器是全局的：把 sleep 放在
# 并发闸门内部会让 N 个 worker 以 N 倍速率发请求。
PER_MIN = 19
gate    = asyncio.Semaphore(4)         # 在途连接数
_lock   = asyncio.Lock()
_slot   = 0.0                          # 共享时间轴上下一个空闲时刻

async def pace():
    """全局范围内，每 60/PER_MIN 秒发放一个请求名额。"""
    global _slot
    async with _lock:
        now  = time.monotonic()
        _slot = max(now, _slot) + 60 / PER_MIN
        wait  = _slot - now
    if wait > 0:
        await asyncio.sleep(wait)

class QuotaExhausted(Exception): ...   # 月度配额：重试无济于事

async def fill_pool(client, cluster) -> list[dict]:
    """一次请求 → 为一个主题集群取回最多 100 张带署名照片。"""
    for attempt in range(4):
        await pace()
        async with gate:
            r = await client.get(
                SEARCH,
                headers={"X-Api-Key": KEY},
                params={
                    "q": cluster["camera_brief"],   # 一个场景，不是关键词
                    "per_page": 100,
                    "orientation": "landscape",
                    "score_threshold": 0.5,
                    "fields": FIELDS,
                },
                timeout=15,
            )
        if r.status_code == 429:
            retry_after = r.headers.get("Retry-After")
            if retry_after is None:            # 无 Retry-After ⇒ 月度配额，
                raise QuotaExhausted(cluster["id"])   # 停止整轮运行
            await asyncio.sleep(int(retry_after))   # 每分钟限流：会恢复
            continue
        r.raise_for_status()
        return r.json()["data"]
    return []                                  # 记录日志；池保留旧数据行

async def refill(clusters):
    async with httpx.AsyncClient(http2=True) as client:
        pools = await asyncio.gather(*(fill_pool(client, c) for c in clusters))
    for cluster, photos in zip(clusters, pools):
        db.upsert_pool(cluster["id"], photos)     # photo_id 上 ON CONFLICT DO NOTHING

# 唯一性由表结构强制，而不是靠查询写得小心。
# 一张照片可以出现在多个集群池中；但只能被发布一次。
# CREATE UNIQUE INDEX one_use_per_photo ON image_pool (photo_id)
#        WHERE used_by_page IS NOT NULL;

def assign_hero(page_id: str, cluster_id: str) -> dict | None:
    for _ in range(5):                # 竞争失败就取下一张照片
        try:
            return db.query_one("""
                UPDATE image_pool p SET used_by_page = %s, used_at = NOW()
                WHERE p.id = (
                    SELECT c.id FROM image_pool c
                    WHERE c.cluster_id = %s AND c.used_by_page IS NULL
                      AND NOT EXISTS (              -- 被任何其他集群用过？
                          SELECT 1 FROM image_pool u
                          WHERE u.photo_id = c.photo_id
                            AND u.used_by_page IS NOT NULL
                      )
                    ORDER BY c.rank ASC
                    LIMIT 1 FOR UPDATE SKIP LOCKED  -- 并行 worker 下安全
                )
                RETURNING photo_id, urls, width, height, blur_hash, alt, credit
            """, [page_id, cluster_id])
        except UniqueViolation:      # 两个集群在同一毫秒内争抢
            continue
    return None                     # 池已空 → 放宽简报，重新填充

# 渲染：每个属性都来自数据行——没有布局偏移，不用猜。
# <img src="{urls[regular]}" width="{width}" height="{height}"
#      alt="{alt}" loading="lazy" style="background:{blur_placeholder}">
# <figcaption>{credit}</figcaption>

# Claude Code / CI runner
claude mcp add --transport http pexafy https://mcp.pexafy.com/mcp \
  --header "Authorization: Bearer $PEXAFY_API_KEY"

# 或者提交 .mcp.json，让机群里的每个 worker 都继承它
{
  "mcpServers": {
    "pexafy": {
      "type": "http",
      "url": "https://mcp.pexafy.com/mcp",
      "headers": { "Authorization": "Bearer YOUR_API_KEY" }
    }
  }
}

You  这是内容日历里的 12 个主题集群。为每个集群写一份相机简报，
     拉取 100 张横构图照片，并把池写入 Postgres。标记出任何在
     0.5 以上结果少于 40 条的集群——那些简报需要重写。

Agent  → search_photos(q="a woman signing paperwork with an insurance
           agent at a kitchen table in her home", orientation="landscape")
      ← 100 张照片 · 122 ms · 87 张高于阈值
      … 还有 11 个集群 …

      ✓ 已写入 11 个池（1,043 张照片，全部带署名）
      ⚠ “index fund rebalancing” → 12 条结果。简报过于抽象；
        建议改为："a person at a kitchen table checking figures on a
        laptop with a notebook and coffee beside them"