浅念
f5dc2f7624
User report: product detail page was wrong — section titles are
per-product custom data (CMS wppop_sections migration: Packing &
Delivery / Our Exhibition / Our Certifications / Our Company / FAQ),
not fixed headings; they must drive an anchor tab system like the
CMS [locale]/products/[slug].astro detailTabs.
Data side (both layers, task option 1 + 2):
- theme_pro.product_detail_sections: fields.sections
[{title, content}] JSON is the structured source; legacy sync
output that appended sections as trailing body H2 blocks is
recognized by known/data titles and split out (overview keeps the
rest; same-title blocks dedupe to the structured version; Features
body block drops when fields.features renders it)
- sync_cms_data.build_product: store sections/features as structured
fields JSON, stop appending H2 blocks to the body; stats now count
entries with a sections field
- Specifications table no longer leaks structural JSON field keys
(sections/features/specs/gallery/categories) — scalar spec keys
only
Template side (trade-pro product.html, replaces 12a panel
switching):
- tab sequence is fully data-driven and matches the CMS: Product
Details (always) -> custom sections (data titles) ->
Specifications / Features (shown only when data exists)
- tabs are #anchor links, not panel toggles: all sections render
flat with visible titles (readable and crawlable without JS; native
anchor jump), JS adds smooth scrolling + scrollspy highlight +
tab centering in the horizontal scroll row
- section ids are slug-derived (sec-packing-delivery), duplicate
titles get an index suffix; features grid ports the CMS
detail-feature-row
Tests: test_trade_pro_product_layout_structure rewritten for the
anchor-tab contract — sections via fields, legacy body-H2 fallback,
spec/features conditional tabs, no-sections product showing only
Product Details + Specifications, spec table free of JSON key leaks,
updated pro.css needles. 30 passed; the 1 failure
(test_trade_pro_blocks_visual_bridge) fails identically on the
previous commit (pre-existing, unrelated).
Co-Authored-By: Claude Code <noreply@example.com>
761 lines
29 KiB
Python
761 lines
29 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Task 14 — CMS 精整理数据 → SiteForge 条目同步。
|
||
|
||
数据源：CMS control-api 公开端点（hesupipe 站 published 修订）——即
|
||
runtimes/astro-site-runtime/src/content/{products,news,cases}.ts 在构建期拉取并
|
||
精整理的同一份事实数据（TS 文件本身是 fetcher + 字段映射，不内含条目字面量；
|
||
本脚本按同一映射规则在 Python 侧复刻）。站点名称/tagline 取自 site.ts 的
|
||
SITE 常量（字面量，直接文本解析）。
|
||
|
||
同步策略：
|
||
- 产品/新闻/案例按 slug 匹配现有条目（产品按 route 尾段 /product/<slug>/，
|
||
文章额外容忍 slug 列 70 字符截断），命中→PATCH 更新，未命中→POST 创建；
|
||
- SiteForge 有而 CMS 没有的条目一律保留（可能手动添加）；
|
||
- 图片：CMS 媒体库 URL（/v1/media/public/...）与本地设计图通过 SiteForge
|
||
媒体 API 上传后改写为 media/<file> 相对 URL（跨次运行经临时目录缓存去重）；
|
||
- 正文复刻 wp-text.ts 的 cleanWpHtml/wrapInlineInP/htmlToExcerpt 清洗
|
||
WordPress 残留（Gutenberg 注释、caption 短码、表单区块、H1 降级等），
|
||
SiteForge 侧 bleach 白名单再剥离 aligncenter 等类名/内联样式；
|
||
- 产品详情区块（CMS wppop_sections 迁移，每产品自定义标题）以
|
||
[{title, content}] JSON 存 fields.sections，不追加进 body——主题层渲染
|
||
为锚点 Tab 区块（theme_pro.product_detail_sections；旧同步产物追加进
|
||
body 的 H2 段由主题层识别拆出）；
|
||
- 站点配置：site.ts 的 name/tagline + locale=en（PUT /api/i18n/.../locale）。
|
||
|
||
用法：
|
||
PYTHONUTF8=1 .venv/Scripts/python scripts/sync_cms_data.py --site 483295c29ea0 \
|
||
--api http://127.0.0.1:8787 --cms-api http://127.0.0.1:3210/v1 \
|
||
[--password $SITEFORGE_ADMIN_PASSWORD] [--dry-run] [--types product,post,case]
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
import tempfile
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse
|
||
|
||
import httpx
|
||
|
||
# --------------------------------------------------------------------------
|
||
# WordPress HTML 清洗（wp-text.ts 逐条移植；SiteForge 端 bleach 再做属性剥离）
|
||
# --------------------------------------------------------------------------
|
||
|
||
def strip_trailing_form_block(html: str) -> str:
|
||
blank = r"(?:\s| | )*"
|
||
tail = (
|
||
blank + r"(?:<div>\s*)?<h[1-4][^>]*>[^<]*</h[1-4]>\s*"
|
||
r"(?:<div[^>]*>(?:\s| | |\[[^\]]*\]|<br\s*/?>)*</div>\s*)+"
|
||
r"(?:</div>)?"
|
||
r"(?:\s*<div[^>]*>" + blank + r"</div>\s*)*"
|
||
+ blank + r"$"
|
||
)
|
||
return re.sub(tail, "", html, flags=re.I)
|
||
|
||
|
||
def _caption_repl(m: "re.Match[str]") -> str:
|
||
attrs, inner = m.group(1), m.group(2)
|
||
img_m = re.search(r"<img[^>]*>", inner)
|
||
img = img_m.group(0) if img_m else ""
|
||
caption = inner.replace(img, "", 1).strip()
|
||
align_m = re.search(r'align="([^"]*)"', attrs)
|
||
align = align_m.group(1) if align_m else "aligncenter"
|
||
width_m = re.search(r'width="(\d+)"', attrs)
|
||
style = f' style="max-width:{int(width_m.group(1)) + 12}px"' if width_m else ""
|
||
figcap = f"<figcaption>{caption}</figcaption>" if caption else ""
|
||
return f'<figure class="wp-caption {align}"{style}>{img}{figcap}</figure>'
|
||
|
||
|
||
def clean_wp_html(html: str) -> str:
|
||
html = re.sub(r"<!--\s*/?wp:[^>]*-->", "", html) # Gutenberg 块注释
|
||
html = re.sub(r"\[caption([^\]]*)\](.*?)\[/caption\]", _caption_repl, html, flags=re.S)
|
||
html = re.sub(
|
||
r"\[/?((?:contact-form-7|wpforms|formidable|gravityform|caldera_form)[^\]]*)\]",
|
||
"",
|
||
html,
|
||
flags=re.I,
|
||
)
|
||
html = strip_trailing_form_block(html)
|
||
html = re.sub(r'<div class="richTextDocs-code-line"[^>]*>(?:\s| )*</div>', "", html)
|
||
html = re.sub(r"<h1(\s[^>]*)?>", "<h2>", html, flags=re.I)
|
||
html = re.sub(r"</h1>", "</h2>", html, flags=re.I)
|
||
html = re.sub(r"\n{3,}", "\n\n", html)
|
||
return html.strip()
|
||
|
||
|
||
_BLOCK_RE = re.compile(
|
||
r"<(p|h[1-6]|ul|ol|table|figure|blockquote|pre|hr)\b[\s\S]*?(?:</\1>|/>)", re.I
|
||
)
|
||
|
||
|
||
def wrap_inline_in_p(html: str) -> str:
|
||
parts: list[str] = []
|
||
last = 0
|
||
for m in _BLOCK_RE.finditer(html):
|
||
before = html[last : m.start()]
|
||
if before.strip():
|
||
parts.append(f"<p>{before.strip()}</p>")
|
||
parts.append(m.group(0))
|
||
last = m.end()
|
||
tail = html[last:]
|
||
if tail.strip():
|
||
parts.append(f"<p>{tail.strip()}</p>")
|
||
return "\n".join(parts)
|
||
|
||
|
||
def html_to_excerpt(html: str, max_len: int = 140) -> str:
|
||
text = clean_wp_html(html)
|
||
text = re.sub(r"<[^>]+>", " ", text)
|
||
text = text.replace(" ", " ").replace("&", "&")
|
||
text = re.sub(r"\s+", " ", text).strip()
|
||
if len(text) > max_len:
|
||
return re.sub(r"\s+\S*$", "", text[:max_len]) + "…"
|
||
return text
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# site.ts 字面量解析（站点 name/tagline）
|
||
# --------------------------------------------------------------------------
|
||
|
||
def parse_site_ts(path: Path) -> dict:
|
||
out = {}
|
||
try:
|
||
src = path.read_text(encoding="utf-8")
|
||
except OSError as exc:
|
||
print(f"[warn] 无法读取 site.ts（{exc}），站点配置跳过")
|
||
return out
|
||
m = re.search(r"export const SITE = \{([\s\S]*?)\} as const;", src)
|
||
if not m:
|
||
return out
|
||
body = m.group(1)
|
||
for key in ("name", "fullName", "tagline"):
|
||
km = re.search(rf'{key}:\s*"([^"]*)"', body)
|
||
if km:
|
||
out[key] = km.group(1)
|
||
return out
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 图片处理：CMS 媒体 URL / 本地图 → SiteForge 媒体库
|
||
# --------------------------------------------------------------------------
|
||
|
||
MEDIA_URL_RE = re.compile(
|
||
r"/v1/media/public/([0-9a-f-]+)/([0-9a-f-]+)/(thumb|medium|original)"
|
||
)
|
||
|
||
|
||
def sniff_image(data: bytes) -> tuple[str, str] | None:
|
||
"""返回 (扩展名, mime)；不识别返回 None。"""
|
||
if data.startswith(b"\x89PNG"):
|
||
return ".png", "image/png"
|
||
if data.startswith(b"\xff\xd8"):
|
||
return ".jpg", "image/jpeg"
|
||
if data.startswith(b"GIF8"):
|
||
return ".gif", "image/gif"
|
||
if len(data) > 12 and data[8:12] == b"WEBP":
|
||
return ".webp", "image/webp"
|
||
return None
|
||
|
||
|
||
class MediaSync:
|
||
"""上传图片到 SiteForge 媒体库；同源 URL 进程内与跨次运行（临时缓存）去重。"""
|
||
|
||
def __init__(self, client: httpx.Client, api: str, site_id: str, csrf: str,
|
||
cms_origin: str, cms_public_dir: Path, cache_path: Path,
|
||
max_uploads: int = 800, dry_run: bool = False):
|
||
self.client = client
|
||
self.api = api
|
||
self.site_id = site_id
|
||
self.csrf = csrf
|
||
self.cms_origin = cms_origin
|
||
self.cms_public_dir = cms_public_dir
|
||
self.cache_path = cache_path
|
||
self.max_uploads = max_uploads
|
||
self.dry_run = dry_run
|
||
self.hits = 0
|
||
self.cache: dict[str, str] = {}
|
||
self.uploads = 0
|
||
self.failures: list[str] = []
|
||
if cache_path.exists():
|
||
try:
|
||
self.cache = json.loads(cache_path.read_text(encoding="utf-8"))
|
||
except (OSError, ValueError):
|
||
self.cache = {}
|
||
|
||
def _save_cache(self) -> None:
|
||
try:
|
||
self.cache_path.write_text(
|
||
json.dumps(self.cache, ensure_ascii=False, indent=0), encoding="utf-8"
|
||
)
|
||
except OSError as exc:
|
||
print(f"[warn] 媒体缓存写入失败：{exc}")
|
||
|
||
def _fetch_bytes(self, src: str) -> bytes | None:
|
||
if src.startswith("/v1/media/"):
|
||
try:
|
||
r = self.client.get(self.cms_origin + src, timeout=30)
|
||
r.raise_for_status()
|
||
return r.content
|
||
except httpx.HTTPError as exc:
|
||
self.failures.append(f"{src}: 拉取失败 {exc}")
|
||
return None
|
||
if src.startswith("/images/"):
|
||
p = self.cms_public_dir / src.lstrip("/")
|
||
try:
|
||
return p.read_bytes()
|
||
except OSError:
|
||
self.failures.append(f"{src}: 本地文件缺失 {p}")
|
||
return None
|
||
return None
|
||
|
||
def url_for(self, src: str) -> str:
|
||
"""CMS 图片引用 → SiteForge media/<file> URL；失败返回空串。"""
|
||
src = (src or "").strip()
|
||
if not src:
|
||
return ""
|
||
if not src.startswith(("/v1/media/", "/images/")):
|
||
return src # http(s) 或已同步过的引用原样返回
|
||
if self.dry_run:
|
||
return src # 演练模式不落任何写入，保留原引用
|
||
if src in self.cache:
|
||
self.hits += 1
|
||
return self.cache[src]
|
||
if self.uploads >= self.max_uploads:
|
||
self.failures.append(f"{src}: 超过单次上传上限 {self.max_uploads}")
|
||
return ""
|
||
data = self._fetch_bytes(src)
|
||
if not data:
|
||
return ""
|
||
sniff = sniff_image(data)
|
||
if not sniff:
|
||
self.failures.append(f"{src}: 无法识别图片格式")
|
||
return ""
|
||
ext, mime = sniff
|
||
try:
|
||
r = self.client.post(
|
||
f"{self.api}/api/sites/{self.site_id}/media",
|
||
files={"file": (f"cms{ext}", data, mime)},
|
||
headers={"x-siteforge-csrf": self.csrf},
|
||
timeout=60,
|
||
)
|
||
r.raise_for_status()
|
||
sf_url = r.json().get("url", "")
|
||
except (httpx.HTTPError, ValueError) as exc:
|
||
self.failures.append(f"{src}: 上传失败 {exc}")
|
||
return ""
|
||
if not sf_url:
|
||
self.failures.append(f"{src}: 上传响应无 url")
|
||
return ""
|
||
self.uploads += 1
|
||
self.cache[src] = sf_url
|
||
return sf_url
|
||
|
||
def rewrite_body(self, html: str) -> str:
|
||
"""正文中出现的全部 CMS 媒体 URL 原位改写为 media/<file>。"""
|
||
def repl(m: "re.Match[str]") -> str:
|
||
return self.url_for(m.group(0)) or m.group(0)
|
||
return MEDIA_URL_RE.sub(repl, html)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# CMS 数据 → SiteForge payload
|
||
# --------------------------------------------------------------------------
|
||
|
||
PRODUCT_DEFAULT_IMAGES = {
|
||
"hdpe-water-supply-pipe": "/images/pipe-gas-2.jpg",
|
||
"hdpe-gas-pipe": "/images/pipe-gas-1.jpg",
|
||
"hdpe-drainage-pipe": "/images/pipe-gas-3.jpg",
|
||
"hdpe-industrial-pipe": "/images/pipe-gas-4.jpg",
|
||
"hdpe-agricultural-irrigation-pipe": "/images/design/agriculture.webp",
|
||
"hdpe-pipe-coils": "/images/design/fittings.webp",
|
||
}
|
||
PRODUCT_DEFAULT_IMAGE = "/images/design/hdpe.webp"
|
||
CASE_DEFAULT_IMAGE = "/images/design/agriculture.webp"
|
||
NEWS_DEFAULT_IMAGES = (
|
||
("/images/design/factory.webp", ("exhib",)),
|
||
("/images/design/agriculture.webp", ("industr",)),
|
||
)
|
||
NEWS_DEFAULT_IMAGE = "/images/design/hero-pipes.webp"
|
||
|
||
# 主题 Specifications 表只渲染标量字段，且排除这些键（theme_pro.make_context）。
|
||
SPEC_EXCLUDED_KEYS = {
|
||
"brand", "category", "model", "description", "image", "cover", "url", "gallery",
|
||
}
|
||
|
||
|
||
def _s(value) -> str:
|
||
return str(value) if value is not None else ""
|
||
|
||
|
||
def _clean_sections(items) -> list[dict]:
|
||
out = []
|
||
for it in items or []:
|
||
if not isinstance(it, dict):
|
||
continue
|
||
title = _s(it.get("title")).strip()
|
||
content = clean_wp_html(_s(it.get("content")))
|
||
if title and content:
|
||
out.append({"title": title, "content": content})
|
||
return out
|
||
|
||
|
||
def _map_gallery_urls(items) -> list[str]:
|
||
urls = []
|
||
for g in items or []:
|
||
if isinstance(g, dict):
|
||
u = _s(g.get("url")).strip()
|
||
if u:
|
||
urls.append(u)
|
||
return urls
|
||
|
||
|
||
def build_product(slug: str, entry: dict, media: MediaSync) -> dict:
|
||
data = entry.get("data") or {}
|
||
name = _s(data.get("name")) or slug
|
||
raw_desc = clean_wp_html(_s(data.get("description")))
|
||
summary_raw = _s(data.get("summary"))
|
||
|
||
gallery_src = _map_gallery_urls(data.get("images"))
|
||
cover_raw = _s(data.get("coverImage")).strip()
|
||
body_first = None
|
||
m = re.search(r"<img[^>]+src=[\"']([^\"']+)[\"']", raw_desc, re.I)
|
||
if m and m.group(1).startswith(("/images/", "/v1/media/")):
|
||
body_first = m.group(1)
|
||
img_src = (
|
||
gallery_src[0] if gallery_src
|
||
else cover_raw if cover_raw.startswith(("/images/", "/v1/media/"))
|
||
else body_first or PRODUCT_DEFAULT_IMAGES.get(slug, PRODUCT_DEFAULT_IMAGE)
|
||
)
|
||
|
||
sections = _clean_sections(data.get("sections"))
|
||
features = [
|
||
{"title": _s(f.get("title")), "desc": _s(f.get("desc"))}
|
||
for f in (data.get("features") or []) if isinstance(f, dict) and _s(f.get("title"))
|
||
]
|
||
specs = {
|
||
_s(k): _s(v)
|
||
for k, v in (data.get("specifications") or {}).items()
|
||
if _s(v)
|
||
}
|
||
categories = [
|
||
{"slug": _s(c.get("slug")), "name": _s(c.get("name"))}
|
||
for c in (data.get("productCategories") or [])
|
||
if isinstance(c, dict) and _s(c.get("slug"))
|
||
]
|
||
|
||
# 详情区块/特性存结构化 fields（Task 15）：正文 body 只放 description，
|
||
# sections/features 以 [{title, content|desc}] JSON 进 fields——主题层
|
||
# （theme_pro.product_detail_sections）据此渲染每产品自定义锚点 Tab 区块，
|
||
# 不再把区块标题追加进 body（旧产物由主题层按 body H2 段识别拆出）。
|
||
body_html = media.rewrite_body(raw_desc)
|
||
|
||
fields: dict = {}
|
||
if _s(data.get("brand")):
|
||
fields["brand"] = _s(data["brand"])
|
||
if _s(data.get("color")):
|
||
fields["color"] = _s(data["color"])
|
||
if _s(data.get("standard")):
|
||
fields["standard"] = _s(data["standard"])
|
||
if sections:
|
||
fields["sections"] = json.dumps(sections, ensure_ascii=False)
|
||
if categories:
|
||
fields["category"] = categories[0]["name"]
|
||
fields["categories"] = json.dumps(categories, ensure_ascii=False)
|
||
elif _s(data.get("category")):
|
||
fields["category"] = _s(data["category"])
|
||
if specs:
|
||
fields["specs"] = json.dumps(specs, ensure_ascii=False)
|
||
for k, v in specs.items(): # 展平为标量 → 主题 Specifications 表逐键成行
|
||
if k not in SPEC_EXCLUDED_KEYS and k not in fields:
|
||
fields[k] = v
|
||
if data.get("applications"):
|
||
fields["applications"] = ";".join(_s(a) for a in data["applications"])
|
||
if features:
|
||
fields["features"] = json.dumps(features, ensure_ascii=False)
|
||
|
||
gallery_urls = []
|
||
for u in gallery_src + ([cover_raw] if cover_raw else []):
|
||
sf = media.url_for(u)
|
||
if sf and sf not in gallery_urls:
|
||
gallery_urls.append(sf)
|
||
if gallery_urls:
|
||
fields["gallery"] = ";".join(gallery_urls)
|
||
|
||
desc_text = (
|
||
html_to_excerpt(raw_desc)
|
||
or html_to_excerpt(summary_raw)
|
||
or f"{name} — manufactured by HESU under ISO 9001 certified quality management."
|
||
)
|
||
payload = {
|
||
"type": "product",
|
||
"title": name,
|
||
"route": f"/product/{slug}/",
|
||
"summary": desc_text,
|
||
"body": body_html,
|
||
"cover": media.url_for(img_src),
|
||
"meta": {"fields": fields},
|
||
}
|
||
if _s(data.get("seoTitle")):
|
||
payload["seo_title"] = _s(data["seoTitle"])
|
||
if _s(data.get("seoDescription")):
|
||
payload["seo_description"] = _s(data["seoDescription"])
|
||
return payload
|
||
|
||
|
||
def build_post(slug: str, entry: dict, media: MediaSync) -> dict:
|
||
data = entry.get("data") or {}
|
||
title = _s(data.get("title")) or slug
|
||
body = wrap_inline_in_p(clean_wp_html(_s(data.get("body"))))
|
||
|
||
cats = [
|
||
{"slug": _s(c.get("slug")), "name": _s(c.get("name"))}
|
||
for c in (data.get("categories") or [])
|
||
if isinstance(c, dict) and _s(c.get("slug"))
|
||
]
|
||
tag = _s(data.get("tag")) or (cats[0]["name"] if cats else "News")
|
||
|
||
cover_raw = _s(data.get("coverImage")).strip()
|
||
if not cover_raw.startswith(("/images/", "/v1/media/")):
|
||
low = tag.lower()
|
||
cover_raw = next(
|
||
(u for u, keys in NEWS_DEFAULT_IMAGES if any(k in low for k in keys)),
|
||
NEWS_DEFAULT_IMAGE,
|
||
)
|
||
|
||
date = ""
|
||
for raw in (_s(data.get("publishedAt")), _s(entry.get("publishedAt"))):
|
||
dm = re.match(r"(\d{4}-\d{2}-\d{2})", raw)
|
||
if dm:
|
||
date = dm.group(1)
|
||
break
|
||
|
||
fields: dict = {}
|
||
if date:
|
||
fields["publish_date"] = date
|
||
if tag:
|
||
fields["category"] = tag
|
||
if len(cats) > 1:
|
||
fields["tags"] = ";".join(c["name"] for c in cats)
|
||
|
||
payload = {
|
||
"type": "post",
|
||
"title": title,
|
||
"route": f"/post/{slug}/",
|
||
"summary": html_to_excerpt(_s(data.get("summary"))) or html_to_excerpt(body) or title,
|
||
"body": media.rewrite_body(body),
|
||
"cover": media.url_for(cover_raw),
|
||
"meta": {"fields": fields},
|
||
}
|
||
if _s(data.get("seoTitle")):
|
||
payload["seo_title"] = _s(data["seoTitle"])
|
||
if _s(data.get("seoDescription")):
|
||
payload["seo_description"] = _s(data["seoDescription"])
|
||
return payload
|
||
|
||
|
||
def build_case(slug: str, entry: dict, media: MediaSync) -> dict:
|
||
data = entry.get("data") or {}
|
||
title = _s(data.get("title")) or slug
|
||
background = clean_wp_html(_s(data.get("customerBackground")))
|
||
challenge = clean_wp_html(_s(data.get("challenge")))
|
||
solution = clean_wp_html(_s(data.get("solution")))
|
||
results = [
|
||
_s((r.get("value") if isinstance(r, dict) else r)).strip()
|
||
for r in (data.get("results") or [])
|
||
]
|
||
results = [r for r in results if r]
|
||
|
||
cover_raw = _s(data.get("coverImage")).strip()
|
||
if not cover_raw.startswith(("/images/", "/v1/media/")):
|
||
cover_raw = CASE_DEFAULT_IMAGE
|
||
|
||
body = ""
|
||
for h, html in (("Customer Background", background), ("The Challenge", challenge),
|
||
("Our Solution", solution)):
|
||
if html:
|
||
body += f"<h2>{h}</h2>\n{html}\n\n"
|
||
body = media.rewrite_body(body.strip())
|
||
|
||
fields: dict = {}
|
||
if results:
|
||
fields["results"] = ";".join(results)
|
||
if _s(data.get("location")):
|
||
fields["location"] = _s(data["location"])
|
||
if _s(data.get("industry")):
|
||
fields["industry"] = _s(data["industry"])
|
||
gallery_src = _map_gallery_urls(data.get("images"))
|
||
gallery_urls = []
|
||
for u in gallery_src + ([cover_raw] if cover_raw else []):
|
||
sf = media.url_for(u)
|
||
if sf and sf not in gallery_urls:
|
||
gallery_urls.append(sf)
|
||
if gallery_urls:
|
||
fields["gallery"] = ";".join(gallery_urls)
|
||
|
||
payload = {
|
||
"type": "case",
|
||
"title": title,
|
||
"route": f"/case/{slug}/",
|
||
"summary": (challenge.splitlines()[0] if challenge else "") or \
|
||
(solution.splitlines()[0] if solution else "") or title,
|
||
"body": body,
|
||
"cover": media.url_for(cover_raw),
|
||
"meta": {"fields": fields},
|
||
}
|
||
if _s(data.get("seoTitle")):
|
||
payload["seo_title"] = _s(data["seoTitle"])
|
||
if _s(data.get("seoDescription")):
|
||
payload["seo_description"] = _s(data["seoDescription"])
|
||
return payload
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# SiteForge 客户端
|
||
# --------------------------------------------------------------------------
|
||
|
||
class SiteForge:
|
||
def __init__(self, api: str, username: str, password: str):
|
||
self.api = api.rstrip("/")
|
||
self.client = httpx.Client(base_url=self.api, timeout=60)
|
||
r = self.client.post(
|
||
"/api/auth/login", json={"username": username, "password": password}
|
||
)
|
||
if r.status_code != 200:
|
||
raise SystemExit(f"[fatal] 登录失败 {r.status_code}: {r.text[:200]}")
|
||
me = self.client.get("/api/auth/me").json()
|
||
self.csrf = me.get("csrf") or ""
|
||
|
||
def entries(self, site_id: str) -> list[dict]:
|
||
r = self.client.get(f"/api/sites/{site_id}/entries")
|
||
r.raise_for_status()
|
||
return r.json()
|
||
|
||
def save(self, site_id: str, payload: dict, entry_id: str | None) -> tuple[str, dict]:
|
||
headers = {"x-siteforge-csrf": self.csrf}
|
||
if entry_id:
|
||
r = self.client.patch(
|
||
f"/api/sites/{site_id}/entries/{entry_id}",
|
||
json=payload, headers=headers,
|
||
)
|
||
action = "updated"
|
||
else:
|
||
r = self.client.post(
|
||
f"/api/sites/{site_id}/entries", json=payload, headers=headers
|
||
)
|
||
action = "created"
|
||
if r.status_code not in (200, 201):
|
||
raise RuntimeError(f"HTTP {r.status_code}: {r.text[:300]}")
|
||
return action, r.json()
|
||
|
||
def patch_site(self, site_id: str, patch: dict) -> None:
|
||
r = self.client.patch(
|
||
f"/api/sites/{site_id}", json=patch, headers={"x-siteforge-csrf": self.csrf}
|
||
)
|
||
r.raise_for_status()
|
||
|
||
def set_locale(self, site_id: str, locale: str) -> None:
|
||
r = self.client.put(
|
||
f"/api/i18n/sites/{site_id}/locale",
|
||
json={"locale": locale}, headers={"x-siteforge-csrf": self.csrf},
|
||
)
|
||
r.raise_for_status()
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 主流程
|
||
# --------------------------------------------------------------------------
|
||
|
||
def fetch_cms(cms_api: str, cms_site: str) -> dict[str, list[dict]]:
|
||
endpoints = {
|
||
"product": f"/content/public/entries?type=product&locale=en&site={cms_site}",
|
||
"post": f"/content/public/news?locale=en&site={cms_site}",
|
||
"case": f"/content/public/cases?locale=en&site={cms_site}",
|
||
}
|
||
out: dict[str, list[dict]] = {}
|
||
with httpx.Client(base_url=cms_api.rstrip("/"), timeout=60) as c:
|
||
for typ, path in endpoints.items():
|
||
r = c.get(path)
|
||
if r.status_code == 404:
|
||
out[typ] = []
|
||
continue
|
||
r.raise_for_status()
|
||
out[typ] = r.json()
|
||
return out
|
||
|
||
|
||
def route_slug(route: str) -> str:
|
||
return (route or "").strip("/").split("/")[-1] if (route or "").strip("/") else ""
|
||
|
||
|
||
def build_match_index(existing: list[dict]) -> dict[str, dict[str, dict]]:
|
||
idx: dict[str, dict[str, dict]] = {"route": {}, "slug": {}}
|
||
seen_slug: dict[str, int] = {}
|
||
for e in existing:
|
||
rs = route_slug(e.get("route", ""))
|
||
if rs and e["type"] != "page":
|
||
key = (e["type"], rs)
|
||
idx["route"].setdefault(key, e)
|
||
sl = e.get("slug") or ""
|
||
if sl:
|
||
key = (e["type"], sl)
|
||
seen_slug[key] = seen_slug.get(key, 0) + 1
|
||
idx["slug"].setdefault(key, e)
|
||
for key, n in seen_slug.items():
|
||
if n > 1:
|
||
idx["slug"].pop(key, None) # 撞名则不可靠，弃用
|
||
return idx
|
||
|
||
|
||
def match_entry(typ: str, cms_slug: str, idx: dict) -> dict | None:
|
||
hit = idx["route"].get((typ, cms_slug))
|
||
if hit:
|
||
return hit
|
||
hit = idx["slug"].get((typ, cms_slug))
|
||
if hit:
|
||
return hit
|
||
if len(cms_slug) > 70: # core.slug_of 截断为 70 字符
|
||
hit = idx["slug"].get((typ, cms_slug[:70]))
|
||
if hit:
|
||
return hit
|
||
return None
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="CMS 精整理数据 → SiteForge 同步")
|
||
ap.add_argument("--site", required=True, help="SiteForge site id")
|
||
ap.add_argument("--api", default="http://127.0.0.1:8787", help="SiteForge API origin")
|
||
ap.add_argument("--username", default="admin")
|
||
ap.add_argument("--password", default="", help="缺省读 SITEFORGE_ADMIN_PASSWORD")
|
||
ap.add_argument("--cms-api", default="http://localhost:3000/v1",
|
||
help="CMS control-api（公开端点）")
|
||
ap.add_argument("--cms-site", default="hesupipe")
|
||
ap.add_argument("--cms-public-dir",
|
||
default="D:/project/cms/runtimes/astro-site-runtime/public",
|
||
help="CMS 静态资源目录（/images/ 设计图兜底）")
|
||
ap.add_argument("--site-ts",
|
||
default="D:/project/cms/runtimes/astro-site-runtime/src/content/site.ts")
|
||
ap.add_argument("--locale", default="en")
|
||
ap.add_argument("--types", default="product,post,case")
|
||
ap.add_argument("--max-media", type=int, default=800)
|
||
ap.add_argument("--dry-run", action="store_true")
|
||
ap.add_argument("--skip-site-config", action="store_true")
|
||
args = ap.parse_args()
|
||
|
||
import os
|
||
password = args.password or os.environ.get("SITEFORGE_ADMIN_PASSWORD", "")
|
||
if not password:
|
||
print("[fatal] 缺少管理员密码：--password 或 SITEFORGE_ADMIN_PASSWORD")
|
||
return 2
|
||
|
||
types = [t.strip() for t in args.types.split(",") if t.strip()]
|
||
cms = fetch_cms(args.cms_api, args.cms_site)
|
||
print(f"[cms] 拉取完成："
|
||
+ " ".join(f"{t}={len(cms.get(t, []))}" for t in types))
|
||
|
||
sf = SiteForge(args.api, args.username, password)
|
||
existing = sf.entries(args.site)
|
||
idx = build_match_index(existing)
|
||
site_ts = parse_site_ts(Path(args.site_ts))
|
||
|
||
api_key = re.sub(r"[^a-z0-9]+", "-", urlparse(args.api).netloc.lower()).strip("-")
|
||
cache_path = (
|
||
Path(tempfile.gettempdir())
|
||
/ f"siteforge-cms-sync-media-{api_key}-{args.site}.json"
|
||
)
|
||
media = MediaSync(
|
||
sf.client, args.api.rstrip("/"), args.site, sf.csrf,
|
||
cms_origin=args.cms_api.rstrip("/").removesuffix("/v1"),
|
||
cms_public_dir=Path(args.cms_public_dir),
|
||
cache_path=cache_path, max_uploads=args.max_media, dry_run=args.dry_run,
|
||
)
|
||
|
||
builders = {"product": build_product, "post": build_post, "case": build_case}
|
||
stats = {t: {"created": 0, "updated": 0, "failed": 0} for t in types}
|
||
failures: list[str] = []
|
||
|
||
for typ in types:
|
||
for entry in cms.get(typ, []):
|
||
slug = entry.get("slug") or ""
|
||
if not slug:
|
||
continue
|
||
try:
|
||
payload = builders[typ](slug, entry, media)
|
||
except Exception as exc: # noqa: BLE001 —— 单条失败不阻断整体
|
||
stats[typ]["failed"] += 1
|
||
failures.append(f"{typ}/{slug}: 构建失败 {exc}")
|
||
continue
|
||
if args.dry_run:
|
||
hit = match_entry(typ, slug, idx)
|
||
stats[typ]["updated" if hit else "created"] += 1
|
||
continue
|
||
hit = match_entry(typ, slug, idx)
|
||
body = dict(payload)
|
||
if hit:
|
||
body.pop("route", None) # 保留既有 URL（WP 迁移 SEO 连续性）
|
||
merged_meta = dict(hit.get("meta") or {})
|
||
merged_meta.update(body.get("meta") or {})
|
||
body["meta"] = merged_meta
|
||
try:
|
||
action, _ = sf.save(args.site, body, hit["id"] if hit else None)
|
||
stats[typ][action] += 1
|
||
except Exception as exc: # noqa: BLE001
|
||
stats[typ]["failed"] += 1
|
||
failures.append(f"{typ}/{slug}: {exc}")
|
||
|
||
if not args.dry_run:
|
||
media._save_cache()
|
||
if not args.skip_site_config:
|
||
patch: dict = {}
|
||
if site_ts.get("name"):
|
||
patch["name"] = site_ts["name"]
|
||
if site_ts.get("tagline"):
|
||
patch["tagline"] = site_ts["tagline"]
|
||
if patch:
|
||
sf.patch_site(args.site, patch)
|
||
print(f"[site] 站点配置已同步：{patch}")
|
||
try:
|
||
sf.set_locale(args.site, args.locale)
|
||
print(f"[site] locale={args.locale}")
|
||
except Exception as exc: # noqa: BLE001
|
||
failures.append(f"locale: {exc}")
|
||
|
||
final = sf.entries(args.site)
|
||
per_type: dict[str, int] = {}
|
||
cover_media = gallery_fields = spec_fields = sections_fields = 0
|
||
for e in final:
|
||
per_type[e["type"]] = per_type.get(e["type"], 0) + 1
|
||
f = (e.get("meta") or {}).get("fields") or {}
|
||
if str(e.get("cover", "")).startswith("media/"):
|
||
cover_media += 1
|
||
if f.get("gallery"):
|
||
gallery_fields += 1
|
||
if f.get("specs"):
|
||
spec_fields += 1
|
||
if f.get("sections"):
|
||
sections_fields += 1
|
||
|
||
print("\n===== 同步统计 =====")
|
||
for t in types:
|
||
s = stats[t]
|
||
print(f"{t}: cms={len(cms.get(t, []))} created={s['created']} "
|
||
f"updated={s['updated']} failed={s['failed']} site_total={per_type.get(t, 0)}")
|
||
print(f"媒体上传：{media.uploads} 张（缓存命中 {media.hits}）")
|
||
print(f"全站 media 封面：{cover_media} | 含 gallery 字段：{gallery_fields} "
|
||
f"| 含 specs 字段：{spec_fields} | 含 sections 字段：{sections_fields}")
|
||
if media.failures:
|
||
print(f"媒体失败 {len(media.failures)} 项（前 10）：")
|
||
for f in media.failures[:10]:
|
||
print(f" - {f}")
|
||
if failures:
|
||
print(f"条目失败 {len(failures)} 项（前 10）：")
|
||
for f in failures[:10]:
|
||
print(f" - {f}")
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|