525 行
20 KiB
Python
525 行
20 KiB
Python
from __future__ import annotations
|
||
"""将现有 JSON 数据导入 MySQL
|
||
Usage: cd hulai-admin-api && python -m scripts.migrate_json_to_db
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from app.config import settings
|
||
from app.database import SessionLocal, engine, Base
|
||
from app.models import *
|
||
|
||
|
||
def load_json(filename):
|
||
path = os.path.join(settings.NUXT_DATA_PATH, filename)
|
||
if not os.path.exists(path):
|
||
print(f" ⚠️ 文件不存在: {path}，跳过")
|
||
return None
|
||
with open(path, "r", encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
def migrate_brand(db):
|
||
data = load_json("brand.json")
|
||
if not data:
|
||
return
|
||
if db.query(Brand).first():
|
||
print(" brand: 已存在，跳过")
|
||
return
|
||
|
||
brand = Brand(
|
||
name=data["name"], full_name=data.get("fullName"), domain=data.get("domain"),
|
||
url=data.get("url"), slogan_emotional=data.get("slogan", {}).get("emotional"),
|
||
slogan_functional=data.get("slogan", {}).get("functional"),
|
||
icp_entity=data.get("icpEntity"), icp=data.get("icp"),
|
||
icp_url=data.get("icpUrl"), e_contract=data.get("eContract"),
|
||
)
|
||
db.add(brand)
|
||
db.flush()
|
||
|
||
for i, s in enumerate(data.get("trustStats", [])):
|
||
db.add(BrandTrustStat(brand_id=brand.id, value=s["value"], unit=s.get("unit"), label=s["label"], sort_order=i))
|
||
for i, d in enumerate(data.get("differentiators", [])):
|
||
db.add(BrandDifferentiator(brand_id=brand.id, title=d["title"], description=d.get("description"), sort_order=i))
|
||
print(" brand: ✓")
|
||
|
||
|
||
def migrate_products(db):
|
||
data = load_json("products.json")
|
||
if not data:
|
||
return
|
||
if db.query(ProductConfig).first():
|
||
print(" products: 已存在，跳过")
|
||
return
|
||
|
||
config = ProductConfig(narrative=data.get("narrative"), pricing_philosophy=data.get("pricingPhilosophy"))
|
||
db.add(config)
|
||
db.flush()
|
||
|
||
for i, v in enumerate(data.get("versions", [])):
|
||
pv = ProductVersion(
|
||
config_id=config.id, version_id=v["id"], name=v["name"],
|
||
days=v["days"], nights=v["nights"], audience=v.get("audience"),
|
||
description=v.get("description"), tag=v.get("tag"), sort_order=i,
|
||
)
|
||
db.add(pv)
|
||
db.flush()
|
||
for j, h in enumerate(v.get("highlights", [])):
|
||
db.add(ProductVersionHighlight(version_id=pv.id, text=h, sort_order=j))
|
||
|
||
camp_data = data.get("summerCamp")
|
||
if camp_data:
|
||
camp = SummerCamp(
|
||
config_id=config.id, name=camp_data.get("name"),
|
||
positioning=camp_data.get("positioning"),
|
||
days=camp_data.get("days"), nights=camp_data.get("nights"),
|
||
sessions_json=camp_data.get("sessions"),
|
||
difference_from_v9=camp_data.get("differenceFromV9"),
|
||
)
|
||
db.add(camp)
|
||
db.flush()
|
||
for i, p in enumerate(camp_data.get("principles", [])):
|
||
db.add(SummerCampPrinciple(camp_id=camp.id, text=p, sort_order=i))
|
||
for i, a in enumerate(camp_data.get("activities", [])):
|
||
db.add(SummerCampActivity(camp_id=camp.id, text=a, sort_order=i))
|
||
for i, it in enumerate(camp_data.get("itinerary", [])):
|
||
db.add(SummerCampItinerary(camp_id=camp.id, text=it, sort_order=i))
|
||
for i, f in enumerate(camp_data.get("faq", [])):
|
||
db.add(SummerCampFaq(camp_id=camp.id, question=f.get("q") or f.get("question", ""), answer=f.get("a") or f.get("answer", ""), sort_order=i))
|
||
|
||
guide_data = data.get("selectionGuide")
|
||
if guide_data:
|
||
db.add(SelectionGuide(
|
||
config_id=config.id,
|
||
by_vacation_length=guide_data.get("byVacationLength"),
|
||
by_child_age=guide_data.get("byChildAge"),
|
||
by_preference=guide_data.get("byPreference"),
|
||
))
|
||
|
||
print(" products: ✓")
|
||
|
||
|
||
def migrate_faq(db):
|
||
data = load_json("faq.json")
|
||
if not data:
|
||
return
|
||
if db.query(FaqCategory).first():
|
||
print(" faq: 已存在，跳过")
|
||
return
|
||
|
||
for i, cat in enumerate(data.get("categories", [])):
|
||
c = FaqCategory(category_id=cat["id"], name=cat["name"], sort_order=i)
|
||
db.add(c)
|
||
db.flush()
|
||
for j, q in enumerate(cat.get("questions", [])):
|
||
db.add(FaqQuestion(
|
||
category_id=c.id, question_id=q["id"], question=q["question"],
|
||
answer=q["answer"], related_links=q.get("relatedLinks"), sort_order=j,
|
||
))
|
||
print(" faq: ✓")
|
||
|
||
|
||
def migrate_reviews(db):
|
||
data = load_json("reviews.json")
|
||
if not data:
|
||
return
|
||
if db.query(Review).first():
|
||
print(" reviews: 已存在，跳过")
|
||
return
|
||
|
||
summary = data.get("summary", {})
|
||
db.add(ReviewSummary(
|
||
total_count=summary.get("totalCount", 0),
|
||
approval_rate=summary.get("approvalRate"),
|
||
keywords=summary.get("keywords"),
|
||
))
|
||
|
||
for i, r in enumerate(data.get("items", [])):
|
||
db.add(Review(
|
||
nickname=r["nickname"], travel_date=r.get("travelDate"),
|
||
product_version=r.get("productVersion"), screenshot=r.get("screenshot"),
|
||
content=r["content"], scenes=r.get("scenes"), concerns=r.get("concerns"),
|
||
sort_order=i, is_visible=True,
|
||
))
|
||
print(" reviews: ✓")
|
||
|
||
|
||
def migrate_about(db):
|
||
data = load_json("about.json")
|
||
if not data:
|
||
return
|
||
if db.query(AboutStory).first():
|
||
print(" about: 已存在，跳过")
|
||
return
|
||
|
||
story = data.get("story", {})
|
||
db.add(AboutStory(title=story.get("title"), content=story.get("content"), founding_moment=story.get("foundingMoment")))
|
||
|
||
for i, s in enumerate(data.get("subsidiaries", [])):
|
||
db.add(AboutSubsidiary(name=s["name"], role=s.get("role"), established=s.get("established"), sort_order=i))
|
||
|
||
for i, c in enumerate(data.get("certifications", [])):
|
||
db.add(AboutCertification(title=c["title"], detail=c.get("detail"), sort_order=i))
|
||
|
||
tm = data.get("trademark", {})
|
||
if tm:
|
||
db.add(AboutTrademark(name=tm.get("name"), scope=tm.get("scope"), holder=tm.get("holder"), description=tm.get("description")))
|
||
|
||
for i, c in enumerate(data.get("copyrights", [])):
|
||
db.add(AboutCopyright(
|
||
name=c["name"], reg_no=c.get("regNo"), category=c.get("category"),
|
||
holder=c.get("holder"), date=c.get("date"), description=c.get("description"), sort_order=i,
|
||
))
|
||
|
||
for i, g in enumerate(data.get("guarantees", [])):
|
||
db.add(AboutGuarantee(title=g["title"], detail=g.get("detail"), sort_order=i))
|
||
|
||
xhs = data.get("xiaohongshu", {})
|
||
if xhs:
|
||
db.add(AboutXiaohongshu(
|
||
account=xhs.get("account"), verified=xhs.get("verified", False),
|
||
verified_type=xhs.get("verifiedType"), followers=xhs.get("followers"),
|
||
likes=xhs.get("likes"), awards=xhs.get("awards"), tagline=xhs.get("tagline"),
|
||
tags=xhs.get("tags"), description=xhs.get("description"),
|
||
))
|
||
|
||
team = data.get("team", {})
|
||
if team:
|
||
db.add(AboutTeam(summary=team.get("summary")))
|
||
|
||
culture = data.get("culture", {})
|
||
if culture:
|
||
c = AboutCulture(transparency=culture.get("transparency"))
|
||
db.add(c)
|
||
db.flush()
|
||
for i, v in enumerate(culture.get("values", [])):
|
||
db.add(AboutCultureValue(culture_id=c.id, name=v["name"], expression=v.get("expression"), sort_order=i))
|
||
|
||
print(" about: ✓")
|
||
|
||
|
||
def migrate_contact(db):
|
||
data = load_json("contact.json")
|
||
if not data:
|
||
return
|
||
if db.query(ContactChannel).first():
|
||
print(" contact: 已存在，跳过")
|
||
return
|
||
|
||
for i, ch in enumerate(data.get("channels", [])):
|
||
db.add(ContactChannel(
|
||
type=ch["type"], label=ch["label"], value=ch.get("value"),
|
||
qr_image=ch.get("qrImage"), is_primary=ch.get("primary", False),
|
||
description=ch.get("description"), sort_order=i,
|
||
))
|
||
|
||
notice = data.get("securityNotice")
|
||
if notice:
|
||
db.add(ContactConfig(security_notice=notice))
|
||
|
||
print(" contact: ✓")
|
||
|
||
|
||
def migrate_seo(db):
|
||
data = load_json("seo.json")
|
||
if not data:
|
||
return
|
||
if db.query(SeoPage).first():
|
||
print(" seo: 已存在，跳过")
|
||
return
|
||
|
||
for key, page in data.get("pages", {}).items():
|
||
db.add(SeoPage(
|
||
page_key=key, title=page.get("title"), description=page.get("description"),
|
||
h1=page.get("h1"), og_image=page.get("ogImage"),
|
||
))
|
||
print(" seo: ✓")
|
||
|
||
|
||
def migrate_navigation(db):
|
||
data = load_json("navigation.json")
|
||
if not data:
|
||
return
|
||
if db.query(NavHeader).first():
|
||
print(" navigation: 已存在，跳过")
|
||
return
|
||
|
||
for i, h in enumerate(data.get("header", [])):
|
||
db.add(NavHeader(text=h["text"], to_path=h["to"], sort_order=i))
|
||
|
||
for i, g in enumerate(data.get("footer", [])):
|
||
group = NavFooterGroup(title=g["title"], sort_order=i)
|
||
db.add(group)
|
||
db.flush()
|
||
for j, l in enumerate(g.get("links", [])):
|
||
db.add(NavFooterLink(group_id=group.id, text=l["text"], to_path=l["to"], sort_order=j))
|
||
|
||
print(" navigation: ✓")
|
||
|
||
|
||
def migrate_versions(db):
|
||
data = load_json("versions.json")
|
||
if not data:
|
||
return
|
||
if db.query(VersionConfig).first():
|
||
print(" versions: 已存在，跳过")
|
||
return
|
||
|
||
stats = data.get("stats", {})
|
||
quote = data.get("quote", {})
|
||
config = VersionConfig(
|
||
stats_iterations=stats.get("iterations"), stats_years=stats.get("years"),
|
||
stats_guests=stats.get("guests"),
|
||
quote_text=quote.get("text") if quote else None,
|
||
quote_author=quote.get("author") if quote else None,
|
||
)
|
||
db.add(config)
|
||
db.flush()
|
||
|
||
for i, u in enumerate(data.get("upgrades2026", [])):
|
||
db.add(VersionUpgrade(config_id=config.id, tag=u.get("tag"), name=u.get("name"), description=u.get("description"), reason=u.get("reason"), sort_order=i))
|
||
|
||
for i, h in enumerate(data.get("highlights", [])):
|
||
db.add(VersionHighlight(config_id=config.id, label=h.get("label"), text=h.get("text"), sort_order=i))
|
||
|
||
cmp = data.get("compareV8V9", {})
|
||
if cmp:
|
||
db.add(VersionCompare(config_id=config.id, headers=cmp.get("headers"), rows=cmp.get("rows")))
|
||
|
||
for i, t in enumerate(data.get("timeline", [])):
|
||
tl = VersionTimeline(
|
||
config_id=config.id, version=t["version"], date=t.get("date"),
|
||
title=t.get("title"), reason=t.get("reason"), sort_order=i,
|
||
)
|
||
db.add(tl)
|
||
db.flush()
|
||
for j, ch in enumerate(t.get("changes", [])):
|
||
db.add(VersionTimelineChange(timeline_id=tl.id, type=ch["type"], text=ch["text"], sort_order=j))
|
||
|
||
for i, p in enumerate(data.get("philosophy", [])):
|
||
db.add(VersionPhilosophy(config_id=config.id, label=p.get("label"), text=p.get("text"), sort_order=i))
|
||
|
||
print(" versions: ✓")
|
||
|
||
|
||
def migrate_guides(db):
|
||
data = load_json("guides.json")
|
||
if not data:
|
||
return
|
||
if db.query(GuideConfig).first():
|
||
print(" guides: 已存在，跳过")
|
||
return
|
||
|
||
config = GuideConfig(page_intro=data.get("pageIntro"))
|
||
db.add(config)
|
||
db.flush()
|
||
|
||
for i, s in enumerate(data.get("sections", [])):
|
||
# Extract common fields, store the rest as JSON data
|
||
common_keys = {"id", "title", "subtitle", "icon", "content"}
|
||
extra_data = {k: v for k, v in s.items() if k not in common_keys}
|
||
|
||
db.add(GuideSection(
|
||
config_id=config.id, section_id=s["id"], title=s.get("title"),
|
||
subtitle=s.get("subtitle"), icon=s.get("icon"), content=s.get("content"),
|
||
data=extra_data if extra_data else None, sort_order=i,
|
||
))
|
||
|
||
print(" guides: ✓")
|
||
|
||
|
||
def migrate_images(db):
|
||
data = load_json("images.json")
|
||
if not data:
|
||
return
|
||
if db.query(SiteImage).first():
|
||
print(" images: 已存在，跳过")
|
||
return
|
||
|
||
# 图片元数据：label 和建议尺寸
|
||
meta = {
|
||
("logo", "main"): ("品牌Logo", "200x200px（正方形透明PNG）", "网站头部和底部的品牌标识"),
|
||
("logo", "square"): ("方形Logo", "200x200px（正方形）", "社交媒体头像等场景"),
|
||
("hero", "background"): ("首页大图", "1920x1080px（16:9横图）", "首页顶部全屏背景图"),
|
||
("brandStory", "photo"): ("品牌故事配图", "800x600px（4:3横图）", "关于我们-品牌故事旁配图"),
|
||
("mascot", "front"): ("IP形象正面", "560x760px（竖版透明PNG）", "关于我们-IP展示正面形象"),
|
||
("mascot", "banner"): ("IP场景横幅", "1600x600px（宽横幅）", "关于我们-IP草原场景插画"),
|
||
("mascot", "stickerQr"): ("表情包二维码", "320x320px（正方形）", "微信表情包扫码下载"),
|
||
("mascot", "xiaomengma"): ("小蒙马IP", "400x400px（正方形透明PNG）", "夏令营相关页面的小蒙马形象"),
|
||
("team", "guideGroup"): ("领队合影", "800x600px（4:3横图）", "团队介绍-领队团队合影"),
|
||
("team", "routeSurvey"): ("踩线照片", "800x600px（4:3横图）", "团队介绍-线路踩线工作照"),
|
||
("team", "guideTraining"): ("培训照片", "800x600px（4:3横图）", "团队介绍-领队培训照片"),
|
||
("team", "ranchStore"): ("牧场团队", "800x600px（4:3横图）", "团队介绍-牧场商贸团队照"),
|
||
("xiaohongshu", "storefront"): ("小红书店铺", "1200x400px（3:1横幅）", "关于我们-小红书店铺展示横幅"),
|
||
("summerCamp", "river"): ("夏令营河景", "800x600px（4:3横图）", "夏令营页面-河景团照"),
|
||
("summerCamp", "grassland"): ("夏令营草原", "800x600px（4:3横图）", "夏令营页面-草原活动照"),
|
||
("summerCamp", "birch"): ("夏令营白桦林", "800x600px（4:3横图）", "夏令营页面-白桦林研学照"),
|
||
("summerCamp", "graduation"): ("夏令营结营", "800x600px（4:3横图）", "夏令营页面-结营合影"),
|
||
}
|
||
|
||
for group_name, items in data.items():
|
||
if isinstance(items, dict):
|
||
for key, path in items.items():
|
||
m = meta.get((group_name, key), (f"{group_name}.{key}", "", ""))
|
||
db.add(SiteImage(
|
||
group_name=group_name, image_key=key, image_path=path,
|
||
label=m[0], size_hint=m[1], description=m[2],
|
||
))
|
||
|
||
print(" images: ✓")
|
||
|
||
|
||
def _migrate_seasonal(db, season, filename):
|
||
data = load_json(filename)
|
||
if not data:
|
||
return
|
||
if db.query(SeasonalProductConfig).filter_by(season=season).first():
|
||
print(f" {filename}: 已存在，跳过")
|
||
return
|
||
|
||
config = SeasonalProductConfig(
|
||
season=season, narrative=data.get("narrative"), style=data.get("style"),
|
||
)
|
||
db.add(config)
|
||
db.flush()
|
||
|
||
for i, v in enumerate(data.get("versions", [])):
|
||
db.add(SeasonalProductVersion(
|
||
season=season, version_id=v["id"], name=v["name"],
|
||
days=v["days"], nights=v["nights"], tag=v.get("tag"),
|
||
line=v.get("line"), route=v.get("route"), audience=v.get("audience"),
|
||
description=v.get("description"), highlights=v.get("highlights", []),
|
||
itinerary=v.get("itinerary", []), sort_order=i,
|
||
))
|
||
|
||
for cat_key in ["highlights", "southHighlights", "northHighlights"]:
|
||
cat_map = {"highlights": "shared", "southHighlights": "south", "northHighlights": "north"}
|
||
cat = cat_map[cat_key]
|
||
for i, h in enumerate(data.get(cat_key, [])):
|
||
db.add(SeasonalProductHighlight(
|
||
season=season, category=cat, title=h["title"],
|
||
description=h.get("description"), sort_order=i,
|
||
))
|
||
|
||
for i, t in enumerate(data.get("timeline", [])):
|
||
db.add(SeasonalProductTimeline(
|
||
season=season, version=t.get("version"), date=t.get("date"),
|
||
title=t.get("title"), changes=t.get("changes", []),
|
||
reason=t.get("reason"), sort_order=i,
|
||
))
|
||
|
||
print(f" {filename}: ✓")
|
||
|
||
|
||
def migrate_autumn_products(db):
|
||
_migrate_seasonal(db, "autumn", "autumn-products.json")
|
||
|
||
|
||
def migrate_winter_products(db):
|
||
_migrate_seasonal(db, "winter", "winter-products.json")
|
||
|
||
|
||
def migrate_winter_camp(db):
|
||
data = load_json("winter-camp.json")
|
||
if not data:
|
||
return
|
||
if db.query(WinterCampConfig).first():
|
||
print(" winter-camp: 已存在，跳过")
|
||
return
|
||
|
||
config = WinterCampConfig(
|
||
name=data.get("name"), positioning=data.get("positioning"),
|
||
days=data.get("days"), nights=data.get("nights"),
|
||
max_families=data.get("maxFamilies"), total_sessions=data.get("totalSessions"),
|
||
age_range=data.get("ageRange"), deposit=data.get("deposit"),
|
||
season=data.get("season"), route=data.get("route"),
|
||
why_hulunbuir=data.get("whyHulunbuir"), closing_note=data.get("closingNote"),
|
||
photographer=data.get("photographer"), winter_clothing=data.get("winterClothing"),
|
||
camp_advantages=data.get("campAdvantages", []),
|
||
service_config=data.get("serviceConfig", []),
|
||
camp_essentials=data.get("campEssentials", []),
|
||
)
|
||
db.add(config)
|
||
db.flush()
|
||
|
||
for i, h in enumerate(data.get("hotels", [])):
|
||
db.add(WinterCampHotel(name=h["name"], star=h.get("star"), nights=h.get("nights"), description=h.get("description"), sort_order=i))
|
||
|
||
for i, d in enumerate(data.get("itinerary", [])):
|
||
db.add(WinterCampItinerary(day=d["day"], title=d.get("title"), summary=d.get("summary"), highlights=d.get("highlights", []), hotel=d.get("hotel"), sort_order=i))
|
||
|
||
for i, f in enumerate(data.get("faq", [])):
|
||
db.add(WinterCampFaq(question=f["q"], answer=f["a"], sort_order=i))
|
||
|
||
print(" winter-camp: ✓")
|
||
|
||
|
||
def migrate_destinations(db):
|
||
data = load_json("destinations.json")
|
||
if not data:
|
||
return
|
||
if db.query(DestinationConfig).first():
|
||
print(" destinations: 已存在，跳过")
|
||
return
|
||
|
||
closing = data.get("closing", {})
|
||
honest = data.get("honestNote", {})
|
||
config = DestinationConfig(
|
||
title=data.get("title"), subtitle=data.get("subtitle"), intro=data.get("intro"),
|
||
closing_title=closing.get("title"), closing_text=closing.get("text"),
|
||
data_sources=data.get("dataSources"),
|
||
honest_title=honest.get("title"), honest_subtitle=honest.get("subtitle"),
|
||
)
|
||
db.add(config)
|
||
db.flush()
|
||
|
||
for i, d in enumerate(data.get("destinations", [])):
|
||
db.add(DestinationItem(dest_id=d.get("id", f"dest-{i}"), name=d["name"], tag=d.get("tag"), highlight=d.get("highlight", False), sort_order=i))
|
||
|
||
for i, dim in enumerate(data.get("dimensions", [])):
|
||
db.add(DestinationDimension(label=dim["label"], icon=dim.get("icon"), values=dim.get("values", []), sort_order=i))
|
||
|
||
for i, h in enumerate(honest.get("items", [])):
|
||
db.add(DestinationHonestItem(text=h, sort_order=i))
|
||
|
||
print(" destinations: ✓")
|
||
|
||
|
||
def main():
|
||
print("创建数据库表...")
|
||
Base.metadata.create_all(bind=engine)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
print("\n开始迁移数据:")
|
||
migrate_brand(db)
|
||
migrate_products(db)
|
||
migrate_faq(db)
|
||
migrate_reviews(db)
|
||
migrate_about(db)
|
||
migrate_contact(db)
|
||
migrate_seo(db)
|
||
migrate_navigation(db)
|
||
migrate_versions(db)
|
||
migrate_guides(db)
|
||
migrate_images(db)
|
||
migrate_autumn_products(db)
|
||
migrate_winter_products(db)
|
||
migrate_winter_camp(db)
|
||
migrate_destinations(db)
|
||
|
||
db.commit()
|
||
print("\n✓ 数据迁移完成！")
|
||
except Exception as e:
|
||
db.rollback()
|
||
print(f"\n✗ 迁移失败: {e}")
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|