143 lines
4.9 KiB
Python
143 lines
4.9 KiB
Python
"""
|
|
Seed PocketBase collections with nomadro 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())
|