WonderQ-Admin/app/seed.py

from __future__ import annotations

import argparse
import json
import re
from pathlib import Path
from urllib.parse import quote
from sqlalchemy import delete, select
from sqlalchemy.orm import Session
from .auth import hash_password
from .content import ALIASES, CAMPAIGNS, CTAS, DESTINATIONS, HERO_SLIDES, THEMES
from .database import Base, SessionLocal, engine
from .models import (
    AdminUser,
    Campaign,
    CampaignProduct,
    CtaBanner,
    Destination,
    DestinationAlias,
    HeroSlide,
    MediaAsset,
    Product,
    ProductImage,
    SiteVersion,
    ThemeCard,
    utc_now,
)


ROOT_DIR = Path(__file__).resolve().parent.parent


def slugify(value: str) -> str:
    return quote(value, safe="").replace("%", "").lower()


def create_media(db: Session, url: str | None, group: str, name: str | None = None) -> None:
    if not url:
        return
    media = db.scalar(select(MediaAsset).where(MediaAsset.url == url))
    if media:
        media.group = group
        media.name = name
    else:
        db.add(MediaAsset(url=url, group=group, name=name))


def default_detail_sections(summary: str, location: str) -> list[dict]:
    return [
        {
            "key": "overview",
            "label": "行程概述",
            "title": "小包团专属概览",
            "blocks": [{"type": "text", "text": f"{summary}。万趣会按同行人、预算、酒店偏好和体力强度重排细节，保留小车小团、错峰入园与在地向导服务。"}],
        },
        {
            "key": "itinerary",
            "label": "每日行程",
            "title": f"{location} 弹性安排",
            "blocks": [{"type": "text", "text": "默认按抵达接站、核心景点游览、特色体验、酒店休整和返程送站安排每日节奏；具体天数、停留时长和餐食可在行前由服务管家二次确认。"}],
        },
        {
            "key": "service",
            "label": "包含/不含服务",
            "title": "费用边界清晰",
            "blocks": [{"type": "text", "text": "通常包含当地用车、行程内住宿、列明门票/体验、必要讲解和服务管家跟进；大交通、个人消费、未列明餐食和自选项目以最终方案为准。"}],
        },
        {
            "key": "notice",
            "label": "出行须知",
            "title": "贵州山地旅行提示",
            "blocks": [{"type": "text", "text": "贵州多山多雨，建议准备防滑鞋、轻便雨具和薄外套；溶洞、漂流、徒步等体验会按天气和同行人体力调整。"}],
        },
        {
            "key": "price",
            "label": "价格区间",
            "title": "按人数、酒店和季节报价",
            "blocks": [{"type": "text", "text": "页面价格为参考起价，节假日、旺季房态、用车车型和体验资源会影响最终报价；提交需求后由服务管家给出可执行方案。"}],
        },
        {
            "key": "manager",
            "label": "服务管家",
            "title": "直接添加服务管家",
            "blocks": [{"type": "text", "text": "点击底部“服务管家”或拨打 18786174929，可直接添加服务管家沟通出行人数、日期、酒店偏好和预算。"}],
        },
    ]


def product_subtitle(title: str) -> str:
    return re.sub(r"^【.*?】\s*", "", title).split("·")[0]


def load_products() -> list[dict]:
    with (ROOT_DIR / "data" / "generated-products.json").open(encoding="utf-8") as handle:
        return json.load(handle)


def reset_guizhou_content(db: Session) -> dict:
    products = load_products()
    for model in [SiteVersion, CampaignProduct, Campaign, ProductImage, Product, CtaBanner, ThemeCard, HeroSlide, DestinationAlias, Destination, MediaAsset]:
        db.execute(delete(model))
    db.flush()

    for index, slide in enumerate(HERO_SLIDES):
        create_media(db, slide["image"], "hero", slide["title"])
        db.add(
            HeroSlide(
                title=slide["title"],
                kicker=slide["kicker"],
                actionLabel=slide["action"],
                image=slide["image"],
                targetType="campaign",
                targetValue=slide["targetValue"],
                sortOrder=index,
            )
        )

    destination_map: dict[str, str] = {}
    for index, (name, image) in enumerate(DESTINATIONS):
        create_media(db, image, "destination", name)
        destination = Destination(name=name, slug=slugify(name), image=image, isHot=index < 8, sortOrder=index)
        db.add(destination)
        db.flush()
        destination_map[name] = destination.id
        for alias in ALIASES.get(name, []):
            db.add(DestinationAlias(destinationId=destination.id, alias=alias))

    for index, (label, image) in enumerate(THEMES):
        create_media(db, image, "theme", label)
        db.add(ThemeCard(label=label, image=image, targetType="search", targetValue=label, sortOrder=index))

    for index, (alt, image, target_type, target_value) in enumerate(CTAS):
        create_media(db, image, "cta", alt)
        db.add(CtaBanner(alt=alt, image=image, targetType=target_type, targetValue=target_value, sortOrder=index))

    for product in products:
        create_media(db, product.get("image"), "product", product["title"])
        matched_destination = product.get("destinationName") if product.get("destinationName") in destination_map else None
        if not matched_destination:
            matched_destination = next(
                (name for name in destination_map if name in product["title"] or name in product.get("tags", [])),
                None,
            )
        summary = product.get("summary") or " · ".join(product.get("tags", []))
        created = Product(
            sourceId=product["id"],
            title=product["title"],
            subtitle=product_subtitle(product["title"]),
            destinationId=destination_map.get(matched_destination) if matched_destination else None,
            priceAmount=int(product["price"]),
            tags=product.get("tags", []),
            coverImage=product.get("image"),
            summary=summary,
            detailSections=default_detail_sections(summary, matched_destination or "贵州省内定制"),
            status="published",
            sortWeight=product["id"],
            publishedAt=utc_now(),
        )
        db.add(created)
        db.flush()
        db.add(ProductImage(productId=created.id, url=product["image"], alt=product["title"]))

    for seed in CAMPAIGNS:
        campaign = Campaign(
            slug=seed["slug"],
            title=seed["title"],
            description="万趣贵州小包团活动专题。",
            coverImage=seed["coverImage"],
            status="published",
        )
        db.add(campaign)
        db.flush()
        linked_products = db.scalars(
            select(Product)
            .where(Product.sourceId >= seed["start"] + 1, Product.sourceId <= seed["end"])
            .order_by(Product.sourceId.asc())
        ).all()
        for index, product in enumerate(linked_products):
            db.add(CampaignProduct(campaignId=campaign.id, productId=product.id, sortOrder=index))

    snapshot = SiteVersion(
        title="guizhou-content-reset",
        status="published",
        publishedAt=utc_now(),
        snapshot={
            "heroSlides": len(HERO_SLIDES),
            "destinations": len(DESTINATIONS),
            "themeCards": len(THEMES),
            "products": len(products),
        },
    )
    db.add(snapshot)
    db.flush()
    return {
        "heroSlides": len(HERO_SLIDES),
        "destinations": len(DESTINATIONS),
        "themes": len(THEMES),
        "ctaBanners": len(CTAS),
        "products": len(products),
        "siteVersionId": snapshot.id,
    }


def seed_database(reset: bool) -> dict:
    Base.metadata.create_all(bind=engine)
    with SessionLocal() as db:
        user = db.scalar(select(AdminUser).where(AdminUser.email == "admin@example.com"))
        if user:
            user.passwordHash = hash_password("ChangeMe123!")
            user.isActive = True
            user.name = "后台管理员"
            user.role = "super_admin"
        else:
            db.add(
                AdminUser(
                    email="admin@example.com",
                    name="后台管理员",
                    passwordHash=hash_password("ChangeMe123!"),
                    role="super_admin",
                )
            )
        result = reset_guizhou_content(db) if reset else {"reset": False}
        db.commit()
        return result


def main() -> None:
    parser = argparse.ArgumentParser(description="Seed WonderQ Admin database")
    parser.add_argument("--no-reset", action="store_true", help="Only ensure admin user exists")
    args = parser.parse_args()
    result = seed_database(reset=not args.no_reset)
    print(f"Seed complete. Admin login: admin@example.com / ChangeMe123!")
    print(json.dumps(result, ensure_ascii=False))


if __name__ == "__main__":
    main()