hulai-admin-api/app/services/export_service.py
刘涛 8c0a5f489d Initial commit: hulai admin API backend
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-22 22:01:23 +08:00

804 行
28 KiB
Python

from __future__ import annotations
import json
import os
from sqlalchemy.orm import Session
from app.config import settings
from app.models.brand import Brand, BrandTrustStat, BrandDifferentiator
from app.models.product import ProductConfig, ProductVersion, ProductVersionHighlight, SummerCamp, SelectionGuide
from app.models.faq import FaqCategory, FaqQuestion
from app.models.review import ReviewSummary, Review
from app.models.about import (
AboutStory, AboutSubsidiary, AboutCertification, AboutTrademark,
AboutCopyright, AboutGuarantee, AboutXiaohongshu, AboutTeam, AboutCulture,
AboutFounderDetail, AboutFounder, AboutMilestone,
AboutServicePrinciple, AboutDifferentiation,
AboutServiceJourney, AboutStats,
)
from app.models.customize_config import CustomizeConfig
from app.models.contact import ContactChannel, ContactConfig
from app.models.seo import SeoPage
from app.models.navigation import NavHeader, NavFooterGroup
from app.models.version import VersionConfig, VersionUpgrade, VersionHighlight, VersionCompare, VersionTimeline, VersionPhilosophy
from app.models.guide import GuideConfig, GuideSection
from app.models.site_image import SiteImage
from app.models.seasonal_product import SeasonalProductConfig, SeasonalProductVersion, SeasonalProductHighlight, SeasonalProductTimeline
from app.models.winter_camp import WinterCampConfig, WinterCampHotel, WinterCampItinerary, WinterCampFaq
from app.models.destination import DestinationConfig, DestinationItem, DestinationDimension, DestinationHonestItem
from app.models.blog import Blog
from app.models.story import Story
from app.models.news import News
from app.models.pricing import PricingItem
from app.models.qualification import Qualification
from app.models.partner import Partner
from app.models.gallery import GalleryItem
from app.models.destination_detail import DestinationDetail
from app.models.selector import SelectorConfig
from app.models.courses import CoursesConfig
def _write_json(filename: str, data):
path = os.path.join(settings.NUXT_DATA_PATH, filename)
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def export_brand(db: Session):
brand = db.query(Brand).first()
if not brand:
return
data = {
"name": brand.name,
"fullName": brand.full_name,
"domain": brand.domain,
"url": brand.url,
"slogan": {
"emotional": brand.slogan_emotional,
"functional": brand.slogan_functional,
},
"trustStats": [
{"value": s.value, "unit": s.unit, "label": s.label}
for s in brand.trust_stats
],
"differentiators": [
{"title": d.title, "description": d.description}
for d in brand.differentiators
],
"icpEntity": brand.icp_entity,
"icp": brand.icp,
"icpUrl": brand.icp_url,
"eContract": brand.e_contract,
}
if brand.cta_buttons:
data["ctaButtons"] = brand.cta_buttons
if brand.conversion_path:
data["conversionPath"] = brand.conversion_path
_write_json("brand.json", data)
def export_products(db: Session):
config = db.query(ProductConfig).first()
if not config:
return
versions = db.query(ProductVersion).filter_by(config_id=config.id).order_by(ProductVersion.sort_order).all()
data = {
"narrative": config.narrative,
"versions": [],
}
for v in versions:
data["versions"].append({
"id": v.version_id,
"name": v.name,
"days": v.days,
"nights": v.nights,
"audience": v.audience,
"description": v.description,
"highlights": [h.text for h in (v.highlights or [])],
"tag": v.tag,
})
# Summer camp
camp = db.query(SummerCamp).filter_by(config_id=config.id).first()
if camp:
data["summerCamp"] = {
"name": camp.name,
"positioning": camp.positioning,
"days": camp.days,
"nights": camp.nights,
"sessions": camp.sessions_json or {},
"principles": [p.text for p in (camp.principles or [])],
"coreActivities": [a.text for a in (camp.activities or [])],
"itinerary": [it.text for it in (camp.itinerary or [])],
"faq": [{"question": f.question, "answer": f.answer} for f in (camp.faq or [])],
"differenceFromV9": camp.difference_from_v9,
}
# Selection guide
guide = db.query(SelectionGuide).filter_by(config_id=config.id).first()
if guide:
data["selectionGuide"] = {
"byVacationLength": guide.by_vacation_length or [],
"byChildAge": guide.by_child_age or [],
"byPreference": guide.by_preference or [],
}
if config.pricing_philosophy:
data["pricingPhilosophy"] = config.pricing_philosophy
_write_json("products.json", data)
def export_faq(db: Session):
categories = db.query(FaqCategory).order_by(FaqCategory.sort_order).all()
data = {
"categories": [
{
"id": cat.category_id,
"name": cat.name,
"questions": [
{
"id": q.question_id,
"question": q.question,
"answer": q.answer,
"relatedLinks": q.related_links or [],
}
for q in cat.questions
],
}
for cat in categories
]
}
_write_json("faq.json", data)
def export_reviews(db: Session):
summary = db.query(ReviewSummary).first()
reviews = db.query(Review).filter(Review.is_visible == True).order_by(Review.sort_order, Review.id).all()
data = {
"summary": {
"totalCount": summary.total_count if summary else 0,
"approvalRate": summary.approval_rate if summary else "98%",
"keywords": summary.keywords if summary else [],
},
"items": [
{
"id": r.id,
"nickname": r.nickname,
"travelDate": r.travel_date,
"productVersion": r.product_version,
"screenshot": r.screenshot,
"content": r.content,
"scenes": r.scenes or [],
"concerns": r.concerns or [],
}
for r in reviews
],
}
_write_json("reviews.json", data)
def export_about(db: Session):
story = db.query(AboutStory).first()
subs = db.query(AboutSubsidiary).order_by(AboutSubsidiary.sort_order).all()
certs = db.query(AboutCertification).order_by(AboutCertification.sort_order).all()
tm = db.query(AboutTrademark).first()
copyrights = db.query(AboutCopyright).order_by(AboutCopyright.sort_order).all()
guarantees = db.query(AboutGuarantee).order_by(AboutGuarantee.sort_order).all()
xhs = db.query(AboutXiaohongshu).first()
team = db.query(AboutTeam).first()
culture = db.query(AboutCulture).first()
founder_detail = db.query(AboutFounderDetail).first()
founders = db.query(AboutFounder).order_by(AboutFounder.sort_order).all()
milestones = db.query(AboutMilestone).order_by(AboutMilestone.sort_order).all()
principles = db.query(AboutServicePrinciple).order_by(AboutServicePrinciple.sort_order).all()
differentiation = db.query(AboutDifferentiation).order_by(AboutDifferentiation.sort_order).all()
sj = db.query(AboutServiceJourney).first()
stats_row = db.query(AboutStats).first()
data = {}
if story:
paragraphs = [p.strip() for p in story.content.split("\n\n") if p.strip()] if story.content else []
story_data = {"title": story.title, "paragraphs": paragraphs}
if story.totem_description or story.totem_tagline:
story_data["totem"] = {"description": story.totem_description, "tagline": story.totem_tagline}
data["story"] = story_data
data["subsidiaries"] = [{"name": s.name, "role": s.role, "established": s.established} for s in subs]
data["certifications"] = [{"title": c.title, "detail": c.detail} for c in certs]
if tm:
data["trademark"] = {"name": tm.name, "scope": tm.scope, "holder": tm.holder, "description": tm.description}
data["copyrights"] = [
{"name": c.name, "regNo": c.reg_no, "category": c.category, "holder": c.holder, "date": c.date, "description": c.description}
for c in copyrights
]
data["guarantees"] = [{"title": g.title, "detail": g.detail} for g in guarantees]
if xhs:
data["xiaohongshu"] = {
"account": xhs.account, "verified": xhs.verified, "verifiedType": xhs.verified_type,
"followers": xhs.followers, "likes": xhs.likes, "awards": xhs.awards or [],
"tagline": xhs.tagline, "tags": xhs.tags or [], "description": xhs.description,
}
if team:
data["team"] = {"summary": team.summary}
if culture:
culture_data = {
"transparency": culture.transparency,
"values": [{"name": v.name, "expression": v.expression} for v in (culture.values or [])],
}
if culture.core:
culture_data["core"] = culture.core
if culture.trust:
culture_data["trust"] = culture.trust
if culture.grief_award_name:
culture_data["griefAward"] = {"name": culture.grief_award_name, "description": culture.grief_award_description}
data["culture"] = culture_data
if founder_detail:
data["founder"] = {
"name": founder_detail.name,
"title": founder_detail.title,
"brandFounded": founder_detail.brand_founded,
"yearsInHulunbuir": founder_detail.years_in_hulunbuir,
"background": founder_detail.background,
"expertise": founder_detail.expertise or [],
"mediaPresence": founder_detail.media_presence or [],
}
if founders:
data["founders"] = [{"name": f.name, "title": f.title, "story": f.story} for f in founders]
if milestones:
data["milestones"] = [{"year": m.year, "event": m.event} for m in milestones]
if principles:
data["servicePrinciples"] = [p.text for p in principles]
if differentiation:
data["differentiation"] = [{"name": d.name, "detail": d.detail} for d in differentiation]
if sj:
data["serviceJourney"] = {
"title": sj.title,
"subtitle": sj.subtitle,
"moments": [{"step": m.step, "name": m.name, "detail": m.detail} for m in (sj.moments or [])],
}
if stats_row and stats_row.data:
data["stats"] = stats_row.data
_write_json("about.json", data)
def export_customize(db: Session):
cfg = db.query(CustomizeConfig).first()
if not cfg:
return
data = {
"trustStats": cfg.trust_stats or [],
"durations": cfg.durations or [],
"activities": cfg.activities or [],
"budgets": cfg.budgets or [],
"process": cfg.process or [],
"contact": cfg.contact or {},
}
_write_json("customize.json", data)
def export_contact(db: Session):
channels = db.query(ContactChannel).order_by(ContactChannel.sort_order).all()
config = db.query(ContactConfig).first()
data = {
"channels": [
{
"type": c.type, "label": c.label, "value": c.value,
**({"qrImage": c.qr_image} if c.qr_image else {}),
**({"primary": True} if c.is_primary else {}),
"description": c.description,
}
for c in channels
],
"securityNotice": config.security_notice if config else "",
}
_write_json("contact.json", data)
def export_seo(db: Session):
pages = db.query(SeoPage).all()
data = {"pages": {}}
for p in pages:
page_data = {
"title": p.title,
"description": p.description,
"h1": p.h1,
"ogImage": p.og_image,
}
if p.keywords:
page_data["keywords"] = p.keywords
data["pages"][p.page_key] = page_data
_write_json("seo.json", data)
def export_navigation(db: Session):
headers = db.query(NavHeader).order_by(NavHeader.sort_order).all()
groups = db.query(NavFooterGroup).order_by(NavFooterGroup.sort_order).all()
header_list = []
for h in headers:
item = {"text": h.text, "to": h.to_path}
if h.children:
item["children"] = h.children
header_list.append(item)
data = {
"header": header_list,
"footer": [
{"title": g.title, "links": [{"text": l.text, "to": l.to_path} for l in (g.links or [])]}
for g in groups
],
}
_write_json("navigation.json", data)
def export_versions(db: Session):
config = db.query(VersionConfig).first()
if not config:
return
upgrades = db.query(VersionUpgrade).filter_by(config_id=config.id).order_by(VersionUpgrade.sort_order).all()
highlights = db.query(VersionHighlight).filter_by(config_id=config.id).order_by(VersionHighlight.sort_order).all()
compare = db.query(VersionCompare).filter_by(config_id=config.id).first()
timeline = db.query(VersionTimeline).filter_by(config_id=config.id).order_by(VersionTimeline.sort_order).all()
philosophy = db.query(VersionPhilosophy).filter_by(config_id=config.id).order_by(VersionPhilosophy.sort_order).all()
data = {
"stats": {
"iterations": config.stats_iterations,
"years": config.stats_years,
"guests": config.stats_guests,
},
"upgrades2026": [
{"tag": u.tag, "name": u.name, "description": u.description, "reason": u.reason}
for u in upgrades
],
"highlights": [{"label": h.label, "text": h.text} for h in highlights],
}
if compare:
data["compareV8V9"] = {"headers": compare.headers or [], "rows": compare.rows or []}
data["timeline"] = [
{
"version": t.version, "date": t.date, "title": t.title,
"changes": [{"type": ch.type, "text": ch.text} for ch in (t.changes or [])],
"reason": t.reason,
}
for t in timeline
]
if philosophy:
data["philosophy"] = [{"label": p.label, "text": p.text} for p in philosophy]
if config.quote_text:
data["quote"] = {"text": config.quote_text, "author": config.quote_author}
_write_json("versions.json", data)
def export_guides(db: Session):
config = db.query(GuideConfig).first()
if not config:
return
sections = db.query(GuideSection).filter_by(config_id=config.id).order_by(GuideSection.sort_order).all()
data = {
"pageIntro": config.page_intro,
"sections": [],
}
for s in sections:
section_data = {
"id": s.section_id,
"title": s.title,
"subtitle": s.subtitle,
"icon": s.icon,
"content": s.content,
}
# Merge the JSON data fields back into the section
if s.data and isinstance(s.data, dict):
section_data.update(s.data)
data["sections"].append(section_data)
_write_json("guides.json", data)
def export_images(db: Session):
images = db.query(SiteImage).order_by(SiteImage.group_name, SiteImage.image_key).all()
data = {}
for img in images:
if img.group_name not in data:
data[img.group_name] = {}
data[img.group_name][img.image_key] = img.image_path
_write_json("images.json", data)
def _export_seasonal(db: Session, season: str, filename: str):
config = db.query(SeasonalProductConfig).filter_by(season=season).first()
versions = db.query(SeasonalProductVersion).filter_by(season=season).order_by(SeasonalProductVersion.sort_order).all()
highlights = db.query(SeasonalProductHighlight).filter_by(season=season).order_by(SeasonalProductHighlight.sort_order).all()
timeline = db.query(SeasonalProductTimeline).filter_by(season=season).order_by(SeasonalProductTimeline.sort_order).all()
data = {}
if config:
if config.brand_name:
data["brand"] = config.brand_name
if config.version_label:
data["version"] = config.version_label
if config.season_label:
data["season"] = config.season_label
data["narrative"] = config.narrative
data["style"] = config.style
data["versions"] = [
{
"id": v.version_id, "name": v.name, "days": v.days, "nights": v.nights,
"tag": v.tag, "line": v.line, "route": v.route, "audience": v.audience,
"description": v.description, "highlights": v.highlights or [], "itinerary": v.itinerary or [],
}
for v in versions
]
shared = [h for h in highlights if h.category == "shared"]
south = [h for h in highlights if h.category == "south"]
north = [h for h in highlights if h.category == "north"]
data["highlights"] = [{"title": h.title, "description": h.description} for h in shared]
data["southHighlights"] = [{"title": h.title, "description": h.description} for h in south]
data["northHighlights"] = [{"title": h.title, "description": h.description} for h in north]
data["timeline"] = [
{
"version": t.version, "date": t.date, "title": t.title,
"changes": t.changes or [], "reason": t.reason,
}
for t in timeline
]
if config and config.selection_guide:
data["selectionGuide"] = config.selection_guide
_write_json(filename, data)
def export_autumn_products(db: Session):
_export_seasonal(db, "autumn", "autumn-products.json")
def export_winter_products(db: Session):
_export_seasonal(db, "winter", "winter-products.json")
def export_winter_camp(db: Session):
config = db.query(WinterCampConfig).first()
if not config:
return
hotels = db.query(WinterCampHotel).order_by(WinterCampHotel.sort_order).all()
itinerary = db.query(WinterCampItinerary).order_by(WinterCampItinerary.sort_order).all()
faq = db.query(WinterCampFaq).order_by(WinterCampFaq.sort_order).all()
data = {
"name": config.name,
"positioning": config.positioning,
"days": config.days,
"nights": config.nights,
"maxFamilies": config.max_families,
"totalSessions": config.total_sessions,
"ageRange": config.age_range,
"deposit": config.deposit,
"season": config.season,
"route": config.route,
"whyHulunbuir": config.why_hulunbuir,
"closingNote": config.closing_note,
"photographer": config.photographer or {},
"winterClothing": config.winter_clothing or {},
"campAdvantages": config.camp_advantages or [],
"serviceConfig": config.service_config or [],
"campEssentials": config.camp_essentials or [],
"hotels": [{"name": h.name, "star": h.star, "nights": h.nights, "description": h.description} for h in hotels],
"itinerary": [
{"day": d.day, "title": d.title, "summary": d.summary, "highlights": d.highlights or [], "hotel": d.hotel}
for d in itinerary
],
"faq": [{"q": f.question, "a": f.answer} for f in faq],
}
_write_json("winter-camp.json", data)
def export_destinations(db: Session):
config = db.query(DestinationConfig).first()
if not config:
return
items = db.query(DestinationItem).order_by(DestinationItem.sort_order).all()
dims = db.query(DestinationDimension).order_by(DestinationDimension.sort_order).all()
honest = db.query(DestinationHonestItem).order_by(DestinationHonestItem.sort_order).all()
data = {
"title": config.title,
"subtitle": config.subtitle,
"intro": config.intro,
"destinations": [
{**({"highlight": True} if d.highlight else {}), "id": d.dest_id, "name": d.name, "tag": d.tag}
for d in items
],
"dimensions": [{"label": d.label, "icon": d.icon, "values": d.values or []} for d in dims],
"honestNote": {
"title": config.honest_title,
"subtitle": config.honest_subtitle,
"items": [h.text for h in honest],
},
"closing": {"title": config.closing_title, "text": config.closing_text},
"dataSources": config.data_sources,
}
_write_json("destinations.json", data)
def export_blog(db: Session):
posts = db.query(Blog).filter(Blog.is_visible == True).order_by(Blog.sort_order, Blog.id).all()
data = {
"articles": [
{
"id": p.slug,
"title": p.title,
"subtitle": p.summary,
"author": p.author,
"date": p.published_at,
"category": p.category,
"summary": p.summary,
"coverImage": p.cover_image,
"tags": p.tags or [],
"content": p.content or "",
}
for p in posts
]
}
_write_json("blog.json", data)
def export_stories(db: Session):
stories = db.query(Story).filter(Story.is_visible == True).order_by(Story.sort_order, Story.id).all()
data = {
"stories": [
{
"id": f"story-{s.id}",
"title": s.title,
"subtitle": s.summary,
"coverImage": s.cover_image,
"familyType": s.customer_name,
"travelDate": s.travel_date,
"product": s.product_name,
"tags": s.scenes or [],
"summary": s.summary,
"content": s.content or "",
}
for s in stories
]
}
_write_json("stories.json", data)
def export_news(db: Session):
articles = db.query(News).filter(News.is_visible == True).order_by(News.sort_order, News.id).all()
data = {
"articles": [
{
"id": f"news-{n.id}",
"title": n.title,
"date": n.published_at,
"category": n.source,
"summary": n.summary,
"content": n.content or "",
}
for n in articles
]
}
_write_json("news.json", data)
def export_pricing(db: Session):
items = db.query(PricingItem).filter(PricingItem.is_visible == True).order_by(PricingItem.sort_order).all()
data = {
"products": [
{
"id": p.product_slug or f"product-{p.id}",
"name": p.product_name,
"priceLabel": p.price_label,
"priceNote": p.description,
"highlights": p.features or [],
"url": f"/products/{p.product_slug}" if p.product_slug else None,
}
for p in items
]
}
_write_json("pricing.json", data)
def export_qualifications(db: Session):
items = db.query(Qualification).order_by(Qualification.sort_order).all()
licenses = [
{"title": q.title, "issuer": q.issuer, "description": q.description}
for q in items if q.category == "license"
]
insurance_items = [
{"title": q.title, "issuer": q.issuer, "description": q.description}
for q in items if q.category == "insurance"
]
awards = [
{"title": q.title, "issuer": q.issuer, "year": int(q.year) if q.year and q.year.isdigit() else q.year,
"description": q.description}
for q in items if q.category == "award"
]
heritage_items = [
{"title": q.title, "description": q.description}
for q in items if q.category == "heritage"
]
data = {
"licenses": licenses,
"insurance": insurance_items[0] if insurance_items else {},
"awards": awards,
"heritage": heritage_items[0] if heritage_items else {},
}
_write_json("qualifications.json", data)
def export_partners(db: Session):
partners = db.query(Partner).order_by(Partner.sort_order).all()
cat_map = {
"政府合作": "govPartners",
"行业协会": "associations",
"校企合作": "academicCoops",
"景区合作": "scenicPartners",
"酒店合作": "hotelPartners",
"平台认证": "platformEndorsements",
"媒体报道": "mediaReports",
}
data = {v: [] for v in cat_map.values()}
for p in partners:
key = cat_map.get(p.category, "govPartners")
data[key].append({
"name": p.name,
"description": p.description,
"website": p.website,
"verified": True,
})
_write_json("partners.json", data)
def export_gallery(db: Session):
items = db.query(GalleryItem).filter(GalleryItem.is_visible == True).order_by(GalleryItem.sort_order, GalleryItem.id).all()
data = {
"works": [
{
"id": f"gallery-{str(g.id).zfill(3)}",
"title": g.title,
"category": (g.tags or [None])[0] if g.tags else None,
"location": g.location,
"description": g.description,
"image": g.image,
}
for g in items
]
}
_write_json("gallery.json", data)
def export_destinations_detail(db: Session):
items = db.query(DestinationDetail).filter(DestinationDetail.is_visible == True).order_by(DestinationDetail.sort_order).all()
data = {
"destinations": [
{
"id": d.slug,
"name": d.name,
"subtitle": d.subtitle,
"tag": (d.tags or [None])[0] if d.tags else None,
"description": d.description,
"highlights": [
h.get("text", h) if isinstance(h, dict) else h
for h in (d.highlights or [])
],
"bestSeason": {"primary": d.best_season} if d.best_season else {},
"heroImage": d.cover_image,
}
for d in items
]
}
_write_json("destinations-detail.json", data)
def export_selector(db: Session):
cfg = db.query(SelectorConfig).first()
if not cfg:
return
rules = cfg.rules or {}
data = {
"questions": cfg.questions or [],
"scoring": rules.get("scoring", {}),
"matchReasons": rules.get("matchReasons", {}),
}
_write_json("selector.json", data)
def export_courses(db: Session):
cfg = db.query(CoursesConfig).first()
if not cfg:
return
data = {
"modules": cfg.modules or [],
"ageGroups": cfg.age_groups or [],
"faq": cfg.faqs or [],
}
_write_json("courses.json", data)
# Module name -> export function mapping
EXPORT_FUNCTIONS = {
"brand": export_brand,
"products": export_products,
"faq": export_faq,
"reviews": export_reviews,
"about": export_about,
"contact": export_contact,
"seo": export_seo,
"navigation": export_navigation,
"versions": export_versions,
"guides": export_guides,
"images": export_images,
"autumn-products": export_autumn_products,
"winter-products": export_winter_products,
"winter-camp": export_winter_camp,
"destinations": export_destinations,
"customize": export_customize,
"blog": export_blog,
"stories": export_stories,
"news": export_news,
"pricing": export_pricing,
"qualifications": export_qualifications,
"partners": export_partners,
"gallery": export_gallery,
"destinations-detail": export_destinations_detail,
"selector": export_selector,
"courses": export_courses,
}
def export_modules(db: Session, modules: list[str]) -> dict:
if "all" in modules:
modules = list(EXPORT_FUNCTIONS.keys())
results = {}
for module in modules:
fn = EXPORT_FUNCTIONS.get(module)
if fn:
try:
fn(db)
results[module] = "success"
except Exception as e:
results[module] = f"failed: {str(e)}"
else:
results[module] = "unknown module"
return results