541 行
19 KiB
Python
541 行
19 KiB
Python
from __future__ import annotations
|
||
"""把前端 JSON 文件中的数据导入到后端数据库。
|
||
UPSERT 逻辑：已存在则更新所有字段，不存在则插入。
|
||
重新运行脚本即可将 JSON 的最新数据同步到数据库。
|
||
|
||
Usage: cd hulai-admin-api && python -m scripts.import_json_data
|
||
"""
|
||
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)
|
||
|
||
|
||
# ─── Blog ────────────────────────────────────────────────────────────────────
|
||
|
||
def import_blog(db):
|
||
data = load_json("blog.json")
|
||
if not data:
|
||
return
|
||
articles = data.get("articles", [])
|
||
added = updated = 0
|
||
for i, a in enumerate(articles):
|
||
slug = a.get("id") or a.get("slug")
|
||
if not slug:
|
||
print(f" ⚠️ 博文缺少 id/slug，跳过: {a.get('title', '')[:30]}")
|
||
continue
|
||
# 章节内容合并成 HTML
|
||
content_parts = []
|
||
for ch in a.get("sections", []) or a.get("chapters", []):
|
||
if isinstance(ch, dict):
|
||
if ch.get("title"):
|
||
content_parts.append(f"<h2>{ch['title']}</h2>")
|
||
if ch.get("content"):
|
||
content_parts.append(ch["content"])
|
||
if ch.get("body"):
|
||
content_parts.append(ch["body"])
|
||
content = "\n".join(content_parts) if content_parts else a.get("content", "")
|
||
|
||
fields = dict(
|
||
title=a["title"],
|
||
cover_image=a.get("coverImage") or a.get("cover_image"),
|
||
summary=a.get("summary") or a.get("subtitle"),
|
||
content=content,
|
||
author=a.get("author"),
|
||
category=a.get("category"),
|
||
tags=a.get("tags"),
|
||
published_at=a.get("date") or a.get("publishedAt"),
|
||
is_visible=True,
|
||
sort_order=i,
|
||
)
|
||
|
||
existing = db.query(Blog).filter_by(slug=slug).first()
|
||
if existing:
|
||
for k, v in fields.items():
|
||
setattr(existing, k, v)
|
||
updated += 1
|
||
else:
|
||
db.add(Blog(slug=slug, **fields))
|
||
added += 1
|
||
print(f" blog_posts: 新增 {added} 条，更新 {updated} 条（共 {len(articles)} 条）")
|
||
|
||
|
||
# ─── Stories ─────────────────────────────────────────────────────────────────
|
||
|
||
def import_stories(db):
|
||
data = load_json("stories.json")
|
||
if not data:
|
||
return
|
||
stories = data.get("stories", [])
|
||
added = updated = 0
|
||
for i, s in enumerate(stories):
|
||
slug = s.get("slug") or s.get("id")
|
||
title = s.get("title", "")
|
||
# 章节内容合并
|
||
content_parts = []
|
||
for ch in s.get("chapters", []):
|
||
if isinstance(ch, dict):
|
||
if ch.get("title"):
|
||
content_parts.append(f"<h2>{ch['title']}</h2>")
|
||
body = ch.get("content") or ch.get("body", "")
|
||
if body:
|
||
content_parts.append(body)
|
||
content = "\n".join(content_parts) if content_parts else s.get("content", "")
|
||
|
||
fields = dict(
|
||
title=title,
|
||
cover_image=s.get("coverImage") or s.get("cover_image"),
|
||
summary=s.get("summary") or s.get("subtitle"),
|
||
content=content,
|
||
customer_name=s.get("customerName") or s.get("familyType"),
|
||
avatar=s.get("avatar"),
|
||
travel_date=s.get("travelDate"),
|
||
product_name=s.get("product"),
|
||
scenes=s.get("tags"),
|
||
is_visible=True,
|
||
sort_order=i,
|
||
)
|
||
|
||
# 优先按 slug 匹配，无 slug 则按 title
|
||
existing = db.query(Story).filter_by(title=title).first() if title else None
|
||
|
||
if existing:
|
||
for k, v in fields.items():
|
||
setattr(existing, k, v)
|
||
updated += 1
|
||
else:
|
||
db.add(Story(**fields))
|
||
added += 1
|
||
print(f" stories: 新增 {added} 条，更新 {updated} 条（共 {len(stories)} 条）")
|
||
|
||
|
||
# ─── News ─────────────────────────────────────────────────────────────────────
|
||
|
||
def import_news(db):
|
||
data = load_json("news.json")
|
||
if not data:
|
||
return
|
||
articles = data.get("articles", [])
|
||
added = updated = 0
|
||
for i, a in enumerate(articles):
|
||
slug = a.get("slug") or a.get("id")
|
||
title = a.get("title", "")
|
||
|
||
fields = dict(
|
||
title=title,
|
||
cover_image=a.get("coverImage") or a.get("cover_image"),
|
||
summary=a.get("summary"),
|
||
content=a.get("content", ""),
|
||
source=a.get("source"),
|
||
source_url=a.get("sourceUrl") or a.get("url"),
|
||
published_at=a.get("date") or a.get("publishedAt"),
|
||
is_visible=True,
|
||
sort_order=i,
|
||
)
|
||
|
||
existing = db.query(News).filter_by(title=title).first() if title else None
|
||
|
||
if existing:
|
||
for k, v in fields.items():
|
||
setattr(existing, k, v)
|
||
updated += 1
|
||
else:
|
||
db.add(News(**fields))
|
||
added += 1
|
||
print(f" news: 新增 {added} 条，更新 {updated} 条（共 {len(articles)} 条）")
|
||
|
||
|
||
# ─── Pricing ─────────────────────────────────────────────────────────────────
|
||
|
||
def import_pricing(db):
|
||
data = load_json("pricing.json")
|
||
if not data:
|
||
return
|
||
products = data.get("products", [])
|
||
added = updated = 0
|
||
for i, p in enumerate(products):
|
||
product_id = p.get("id") or p.get("slug")
|
||
|
||
fields = dict(
|
||
product_name=p.get("name", ""),
|
||
price_from=None,
|
||
price_unit=None,
|
||
price_label=p.get("priceLabel"),
|
||
description=p.get("priceNote") or p.get("audience"),
|
||
features=p.get("highlights"),
|
||
notes=None,
|
||
is_visible=True,
|
||
sort_order=i,
|
||
)
|
||
|
||
existing = None
|
||
if product_id:
|
||
existing = db.query(PricingItem).filter_by(product_slug=product_id).first()
|
||
|
||
if existing:
|
||
for k, v in fields.items():
|
||
setattr(existing, k, v)
|
||
updated += 1
|
||
else:
|
||
db.add(PricingItem(product_slug=product_id, **fields))
|
||
added += 1
|
||
print(f" pricing_items: 新增 {added} 条，更新 {updated} 条（共 {len(products)} 条）")
|
||
|
||
|
||
# ─── Qualifications ──────────────────────────────────────────────────────────
|
||
|
||
def import_qualifications(db):
|
||
data = load_json("qualifications.json")
|
||
if not data:
|
||
return
|
||
added = updated = 0
|
||
sort = 0
|
||
|
||
def upsert_qual(name, **fields):
|
||
nonlocal added, updated, sort
|
||
existing = db.query(Qualification).filter_by(title=name).first()
|
||
if existing:
|
||
for k, v in fields.items():
|
||
setattr(existing, k, v)
|
||
updated += 1
|
||
else:
|
||
db.add(Qualification(title=name, sort_order=sort, **fields))
|
||
added += 1
|
||
sort += 1
|
||
|
||
# 旅游经营许可证
|
||
for lic in data.get("licenses", []):
|
||
title = lic.get("title", "旅游经营许可证")
|
||
upsert_qual(
|
||
title,
|
||
issuer=lic.get("issuer"),
|
||
year=None,
|
||
image=None,
|
||
description=(
|
||
f"证件号：{lic.get('licenseNo', '')}｜"
|
||
f"持证主体：{lic.get('holder', '')}｜"
|
||
f"法人：{lic.get('legalPerson', '')}｜"
|
||
f"经营范围：{lic.get('businessScope', '')}｜"
|
||
f"统一信用代码：{lic.get('creditCode', '')}｜"
|
||
f"{lic.get('note', '')}"
|
||
),
|
||
category="license",
|
||
)
|
||
|
||
# 旅游责任险
|
||
insurance = data.get("insurance")
|
||
if insurance:
|
||
policies = insurance.get("policies", [])
|
||
policy_desc = "；".join([
|
||
f"{p.get('insured','')}·{p.get('insurer','')}·保单号{p.get('policyNo','')}·保额{p.get('coverage','')}·有效期{p.get('period','')}"
|
||
for p in policies
|
||
])
|
||
upsert_qual(
|
||
insurance.get("title", "旅游责任险"),
|
||
issuer=policies[0].get("insurer") if policies else None,
|
||
year=None,
|
||
image=None,
|
||
description=f"{insurance.get('description', '')}｜{policy_desc}",
|
||
category="insurance",
|
||
)
|
||
|
||
# 荣誉奖项
|
||
for award in data.get("awards", []):
|
||
title = award.get("title", "")
|
||
if not title:
|
||
continue
|
||
upsert_qual(
|
||
title,
|
||
issuer=award.get("issuer") or award.get("platform"),
|
||
year=str(award.get("year", "")) if award.get("year") else None,
|
||
image=None,
|
||
description=award.get("description") or award.get("detail"),
|
||
category="award",
|
||
)
|
||
|
||
# 品牌传承
|
||
heritage = data.get("heritage")
|
||
if heritage:
|
||
milestones_text = "；".join([
|
||
f"{m.get('year')}年：{m.get('event', '')}"
|
||
for m in heritage.get("milestones", [])
|
||
])
|
||
upsert_qual(
|
||
f"品牌资历：{heritage.get('brandYears', '')}年深耕呼伦贝尔",
|
||
issuer=None,
|
||
year=str(heritage.get("brandFounded")) if heritage.get("brandFounded") else None,
|
||
image=None,
|
||
description=f"{heritage.get('note', '')}｜{milestones_text}",
|
||
category="heritage",
|
||
)
|
||
|
||
print(f" qualifications: 新增 {added} 条，更新 {updated} 条")
|
||
|
||
|
||
# ─── Partners ────────────────────────────────────────────────────────────────
|
||
|
||
def import_partners(db):
|
||
data = load_json("partners.json")
|
||
if not data:
|
||
return
|
||
added = updated = 0
|
||
sort = 0
|
||
|
||
cat_map = {
|
||
"govPartners": "政府合作",
|
||
"associations": "行业协会",
|
||
"academicCoops": "校企合作",
|
||
"scenicPartners": "景区合作",
|
||
"hotelPartners": "酒店合作",
|
||
"platformEndorsements": "平台认证",
|
||
"mediaReports": "媒体报道",
|
||
}
|
||
|
||
for cat_key, cat_label in cat_map.items():
|
||
for item in data.get(cat_key, []):
|
||
name = item.get("name") or item.get("platform") or item.get("outlet", "")
|
||
if not name or name.startswith("⚠️"):
|
||
continue
|
||
if not item.get("verified", True):
|
||
continue
|
||
# 构建描述
|
||
desc_parts = []
|
||
for field in ["type", "relationship", "description", "detail", "status", "summary"]:
|
||
val = item.get(field)
|
||
if val and not str(val).startswith("⚠️"):
|
||
desc_parts.append(str(val))
|
||
if item.get("followers"):
|
||
desc_parts.append(f"粉丝：{item['followers']}")
|
||
if item.get("account"):
|
||
desc_parts.append(f"账号：{item['account']}")
|
||
if item.get("since"):
|
||
desc_parts.append(f"合作始于：{item['since']}")
|
||
|
||
fields = dict(
|
||
logo=None,
|
||
website=item.get("url"),
|
||
description="｜".join(desc_parts),
|
||
category=cat_label,
|
||
sort_order=sort,
|
||
)
|
||
|
||
existing = db.query(Partner).filter_by(name=name).first()
|
||
if existing:
|
||
for k, v in fields.items():
|
||
setattr(existing, k, v)
|
||
updated += 1
|
||
else:
|
||
db.add(Partner(name=name, **fields))
|
||
added += 1
|
||
sort += 1
|
||
|
||
print(f" partners: 新增 {added} 条，更新 {updated} 条")
|
||
|
||
|
||
# ─── Gallery ─────────────────────────────────────────────────────────────────
|
||
|
||
def import_gallery(db):
|
||
data = load_json("gallery.json")
|
||
if not data:
|
||
return
|
||
works = data.get("works", [])
|
||
added = updated = 0
|
||
for i, w in enumerate(works):
|
||
title = w.get("title")
|
||
tags = []
|
||
if w.get("category"):
|
||
tags.append(w["category"])
|
||
if w.get("season"):
|
||
tags.append(w["season"])
|
||
if w.get("costumeType"):
|
||
tags.append(w["costumeType"])
|
||
|
||
fields = dict(
|
||
image=w.get("image", ""),
|
||
photographer=None,
|
||
location=w.get("location"),
|
||
description=w.get("description"),
|
||
tags=tags if tags else None,
|
||
is_visible=True,
|
||
sort_order=i,
|
||
)
|
||
|
||
existing = db.query(GalleryItem).filter_by(title=title).first() if title else None
|
||
if existing:
|
||
for k, v in fields.items():
|
||
setattr(existing, k, v)
|
||
updated += 1
|
||
else:
|
||
db.add(GalleryItem(title=title, **fields))
|
||
added += 1
|
||
print(f" gallery_items: 新增 {added} 条，更新 {updated} 条（共 {len(works)} 条）")
|
||
|
||
|
||
# ─── Destination Details ──────────────────────────────────────────────────────
|
||
|
||
def import_destination_details(db):
|
||
data = load_json("destinations-detail.json")
|
||
if not data:
|
||
return
|
||
destinations = data.get("destinations", [])
|
||
added = updated = 0
|
||
for i, d in enumerate(destinations):
|
||
slug = d.get("id", "")
|
||
if not slug:
|
||
continue
|
||
# highlights 可能是 string list 或 dict list
|
||
raw_highlights = d.get("highlights", [])
|
||
if raw_highlights and isinstance(raw_highlights[0], str):
|
||
highlights = [{"text": h} for h in raw_highlights]
|
||
else:
|
||
highlights = raw_highlights
|
||
tags = []
|
||
if d.get("tag"):
|
||
tags.append(d["tag"])
|
||
|
||
fields = dict(
|
||
name=d.get("name", ""),
|
||
subtitle=d.get("subtitle"),
|
||
cover_image=d.get("heroImage") or d.get("coverImage"),
|
||
description=d.get("description", ""),
|
||
location=None,
|
||
best_season=(
|
||
d["bestSeason"].get("primary") if isinstance(d.get("bestSeason"), dict)
|
||
else d.get("bestSeason")
|
||
),
|
||
duration=None,
|
||
highlights=highlights,
|
||
gallery=None,
|
||
tags=tags if tags else None,
|
||
is_visible=True,
|
||
sort_order=i,
|
||
)
|
||
|
||
existing = db.query(DestinationDetail).filter_by(slug=slug).first()
|
||
if existing:
|
||
for k, v in fields.items():
|
||
setattr(existing, k, v)
|
||
updated += 1
|
||
else:
|
||
db.add(DestinationDetail(slug=slug, **fields))
|
||
added += 1
|
||
print(f" destination_details: 新增 {added} 条，更新 {updated} 条（共 {len(destinations)} 条）")
|
||
|
||
|
||
# ─── Selector Config ─────────────────────────────────────────────────────────
|
||
|
||
def import_selector(db):
|
||
data = load_json("selector.json")
|
||
if not data:
|
||
return
|
||
rules = {
|
||
"scoring": data.get("scoring", {}),
|
||
"matchReasons": data.get("matchReasons", {}),
|
||
}
|
||
existing = db.query(SelectorConfig).first()
|
||
if existing:
|
||
existing.questions = data.get("questions", [])
|
||
existing.rules = rules
|
||
print(" selector_config: ✓ 已更新")
|
||
else:
|
||
db.add(SelectorConfig(
|
||
questions=data.get("questions", []),
|
||
rules=rules,
|
||
))
|
||
print(" selector_config: ✓ 新增 1 条")
|
||
|
||
|
||
# ─── Courses Config ───────────────────────────────────────────────────────────
|
||
|
||
def import_courses(db):
|
||
data = load_json("courses.json")
|
||
if not data:
|
||
return
|
||
existing = db.query(CoursesConfig).first()
|
||
if existing:
|
||
existing.modules = data.get("modules", [])
|
||
existing.age_groups = data.get("ageGroups", [])
|
||
existing.faqs = data.get("faq", []) or data.get("faqs", [])
|
||
print(" courses_config: ✓ 已更新")
|
||
else:
|
||
db.add(CoursesConfig(
|
||
modules=data.get("modules", []),
|
||
age_groups=data.get("ageGroups", []),
|
||
faqs=data.get("faq", []) or data.get("faqs", []),
|
||
))
|
||
print(" courses_config: ✓ 新增 1 条")
|
||
|
||
|
||
# ─── Main ─────────────────────────────────────────────────────────────────────
|
||
|
||
def main():
|
||
print("确保数据库表存在...")
|
||
Base.metadata.create_all(bind=engine)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
print("\n开始 UPSERT JSON 数据到数据库:")
|
||
print("-" * 50)
|
||
import_blog(db)
|
||
import_stories(db)
|
||
import_news(db)
|
||
import_pricing(db)
|
||
import_qualifications(db)
|
||
import_partners(db)
|
||
import_gallery(db)
|
||
import_destination_details(db)
|
||
import_selector(db)
|
||
import_courses(db)
|
||
|
||
db.commit()
|
||
print("-" * 50)
|
||
print("\n✓ 数据同步完成！")
|
||
|
||
# 验证
|
||
print("\n数据库记录数验证:")
|
||
from app.models import (Blog, Story, News, PricingItem, Qualification,
|
||
Partner, GalleryItem, DestinationDetail,
|
||
SelectorConfig, CoursesConfig)
|
||
checks = [
|
||
("blog_posts", Blog),
|
||
("stories", Story),
|
||
("news", News),
|
||
("pricing_items", PricingItem),
|
||
("qualifications", Qualification),
|
||
("partners", Partner),
|
||
("gallery_items", GalleryItem),
|
||
("destination_details", DestinationDetail),
|
||
("selector_config", SelectorConfig),
|
||
("courses_config", CoursesConfig),
|
||
]
|
||
for name, model in checks:
|
||
cnt = db.query(model).count()
|
||
status = "✓" if cnt > 0 else "⚠ 空"
|
||
print(f" {status} {name}: {cnt}")
|
||
|
||
except Exception as e:
|
||
db.rollback()
|
||
print(f"\n✗ 导入失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
raise
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|