From 3ce42927941991c4aeee11d44e09dbae5e7754ab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=B5=85=E5=BF=B5?= Date: Thu, 29 Oct 2026 12:05:57 +0800 Subject: [PATCH] Task 14: add CMS refined-data sync script (sync_cms_data.py) Pull hesupipe published products/news/cases from the CMS control-api public endpoints (the same refined data src/content/{products,news,cases}.ts fetch at build time; those TS files are fetchers, not literal arrays) and sync them into SiteForge entries via login + PATCH/POST: - match by route slug (product /product//, post keeps legacy WP route on update, slug-col truncation fallback), create when unmatched, never delete SiteForge-only entries - port wp-text.ts cleaning (Gutenberg comments, [caption] -> figure, trailing form blocks, H1 -> H2, wrapInlineInP, excerpt) so bodies land sanitized (bleach on the API side strips aligncenter classes/styles) - upload CMS media (/v1/media/... and local /images/ design fallbacks) through the SiteForge media API, rewriting body refs to media/; per-target temp-dir cache dedupes uploads across runs - product fields: color/standard/category + specs dict flattened into scalar keys (trade-pro Specifications table renders scalars only) + gallery as semicolon list; post fields: publish_date/category/tags; case fields: results/location/industry/gallery - sync site name/tagline from site.ts SITE literal and set locale=en via PUT /api/i18n/sites/{id}/locale Verified against a local uvicorn on a temp snapshot of the container DB: 52 products (47 updated + 5 created), 129 posts updated, 0 failures, 181 media covers, idempotent re-run, theme registry tests unchanged (30 pass, 1 pre-existing failure). Co-Authored-By: Claude Code --- scripts/sync_cms_data.py | 755 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 755 insertions(+) create mode 100644 scripts/sync_cms_data.py diff --git a/scripts/sync_cms_data.py b/scripts/sync_cms_data.py new file mode 100644 index 0000000..9adf5e5 --- /dev/null +++ b/scripts/sync_cms_data.py @@ -0,0 +1,755 @@ +# -*- 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 列 70 字符截断），命中→PATCH 更新，未命中→POST 创建； + - SiteForge 有而 CMS 没有的条目一律保留（可能手动添加）； + - 图片：CMS 媒体库 URL（/v1/media/public/...）与本地设计图通过 SiteForge + 媒体 API 上传后改写为 media/ 相对 URL（跨次运行经临时目录缓存去重）； + - 正文复刻 wp-text.ts 的 cleanWpHtml/wrapInlineInP/htmlToExcerpt 清洗 + WordPress 残留（Gutenberg 注释、caption 短码、表单区块、H1 降级等）， + SiteForge 侧 bleach 白名单再剥离 aligncenter 等类名/内联样式； + - 站点配置：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"(?:
\s*)?]*>[^<]*\s*" + r"(?:]*>(?:\s| | |\[[^\]]*\]|)*
\s*)+" + r"(?:)?" + r"(?:\s*]*>" + blank + r"\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"]*>", 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"
{caption}
" if caption else "" + return f'
{img}{figcap}
' + + +def clean_wp_html(html: str) -> str: + html = re.sub(r"", "", 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'
]*>(?:\s| )*
', "", html) + html = re.sub(r"]*)?>", "

", html, flags=re.I) + html = re.sub(r"

", "", 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]*?(?:|/>)", 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"

{before.strip()}

") + parts.append(m.group(0)) + last = m.end() + tail = html[last:] + if tail.strip(): + parts.append(f"

{tail.strip()}

") + 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/ 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/。""" + 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"]+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")) + ] + + body_html = raw_desc + for sec in sections: + body_html += f"\n\n

{sec['title']}

\n{sec['content']}" + if features: + body_html += "\n\n

Features

\n" + for f in features: + body_html += f"\n

{f['title']}

\n

{f['desc']}

\n" + body_html = media.rewrite_body(body_html) + + 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 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"

{h}

\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 = 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 + + 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}") + 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())