nomadweb/backend/seed_pocketbase.py

"""
Seed PocketBase collections with NomadFlow data.
Run after PocketBase is started: python seed_pocketbase.py
"""
import asyncio
import json
import httpx
from app.config import settings
from app.data.mock_data import DESTINATIONS, VISAS, FAQS, TESTIMONIALS, TOOLS, BLOG_POSTS

COLLECTIONS = {
  "destinations": {
    "schema": [
      {"name": "slug", "type": "text", "required": True},
      {"name": "name", "type": "text", "required": True},
      {"name": "country", "type": "text"},
      {"name": "emoji", "type": "text"},
      {"name": "tag", "type": "text"},
      {"name": "description", "type": "text"},
      {"name": "region", "type": "text"},
      {"name": "cost", "type": "number"},
      {"name": "speed", "type": "number"},
      {"name": "temperature", "type": "number"},
      {"name": "rating", "type": "number"},
      {"name": "hue", "type": "number"},
      {"name": "nomads_count", "type": "text"},
      {"name": "highlights", "type": "json"},
      {"name": "map_x", "type": "number"},
      {"name": "map_y", "type": "number"},
    ],
    "data": DESTINATIONS,
    "key": "slug",
  },
  "visas": {
    "schema": [
      {"name": "country", "type": "text"}, {"name": "flag", "type": "text"},
      {"name": "name", "type": "text"}, {"name": "badge", "type": "text"},
      {"name": "badge_type", "type": "text"}, {"name": "duration", "type": "text"},
      {"name": "income_req", "type": "text"}, {"name": "approval_time", "type": "text"},
      {"name": "extra", "type": "text"}, {"name": "difficulty", "type": "number"},
      {"name": "difficulty_label", "type": "text"},
    ],
    "data": VISAS,
    "key": "name",
  },
  "faqs": {
    "schema": [
      {"name": "question", "type": "text"}, {"name": "answer", "type": "text"},
      {"name": "order", "type": "number"},
    ],
    "data": FAQS,
    "key": "question",
  },
  "testimonials": {
    "schema": [
      {"name": "avatar", "type": "text"}, {"name": "content", "type": "text"},
      {"name": "author", "type": "text"}, {"name": "role", "type": "text"},
      {"name": "rating", "type": "number"},
    ],
    "data": TESTIMONIALS,
    "key": "author",
  },
  "tools": {
    "schema": [
      {"name": "emoji", "type": "text"}, {"name": "name", "type": "text"},
      {"name": "description", "type": "text"}, {"name": "tags", "type": "json"},
      {"name": "category", "type": "text"},
    ],
    "data": TOOLS,
    "key": "name",
  },
  "blog_posts": {
    "schema": [
      {"name": "slug", "type": "text"}, {"name": "title", "type": "text"},
      {"name": "excerpt", "type": "text"}, {"name": "content", "type": "text"},
      {"name": "emoji", "type": "text"}, {"name": "author", "type": "text"},
      {"name": "published_at", "type": "text"}, {"name": "read_time", "type": "number"},
      {"name": "tags", "type": "json"},
    ],
    "data": [{**p, "content": mock_data.BLOG_CONTENT.get(p["slug"], "")} for p in BLOG_POSTS],
    "key": "slug",
  },
  "subscriptions": {
    "schema": [{"name": "email", "type": "email", "required": True}],
    "data": [],
    "key": "email",
  },
}


async def main():
  base = settings.pocketbase_url.rstrip("/")
  async with httpx.AsyncClient(timeout=15.0) as client:
    # Auth
    auth = await client.post(
      f"{base}/api/admins/auth-with-password",
      json={"identity": settings.pocketbase_admin_email, "password": settings.pocketbase_admin_password},
    )
    if auth.status_code != 200:
      print("❌ PocketBase 认证失败，请先在 Admin UI 创建管理员账号")
      print(f"   访问 {base}/_/ 创建账号: {settings.pocketbase_admin_email}")
      return

    token = auth.json()["token"]
    headers = {"Authorization": token}

    for coll_name, config in COLLECTIONS.items():
      # Check if collection exists
      check = await client.get(f"{base}/api/collections/{coll_name}", headers=headers)
      if check.status_code == 404:
        body = {
          "name": coll_name,
          "type": "base",
          "schema": config["schema"],
        }
        create = await client.post(f"{base}/api/collections", json=body, headers=headers)
        if create.status_code in (200, 201):
          print(f"✅ 创建集合: {coll_name}")
        else:
          print(f"⚠️  集合 {coll_name} 创建失败: {create.text}")
          continue
      else:
        print(f"📦 集合已存在: {coll_name}")

      # Seed records
      for item in config["data"]:
        payload = {k: v for k, v in item.items() if k != "id"}
        r = await client.post(
          f"{base}/api/collections/{coll_name}/records",
          json=payload,
          headers=headers,
        )
        if r.status_code in (200, 201):
          print(f"  + {item.get(config['key'], '')}")
        elif "already" in r.text.lower() or r.status_code == 400:
          print(f"  ~ {item.get(config['key'], '')} (已存在)")

    print("\n🎉 数据种子完成！")


if __name__ == "__main__":
  asyncio.run(main())