157 lines
5.5 KiB
Python
157 lines
5.5 KiB
Python
"""
|
||
fission_package_exporter.py — 裂变笔记包下载
|
||
|
||
把 FissionNote 表中的 N 套裂变结果打成 zip：
|
||
note_01/文案.txt + 生成图 + 发布清单 + 说明。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
import shutil
|
||
import zipfile
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from app.core.config import get_settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def _safe_name(value: str | None, fallback: str = "img") -> str:
|
||
raw = (value or fallback).strip() or fallback
|
||
return "".join(c if c.isalnum() or c in ("-", "_") else "_" for c in raw)[:40]
|
||
|
||
|
||
def _resolve_upload_path(url: str, upload_root: Path) -> Path:
|
||
if url.startswith("/uploads/"):
|
||
return upload_root / url.removeprefix("/uploads/")
|
||
if url.startswith("uploads/"):
|
||
return upload_root / url.removeprefix("uploads/")
|
||
return Path(url)
|
||
|
||
|
||
def _loads_json(value: str | None, fallback: Any) -> Any:
|
||
try:
|
||
return json.loads(value) if value else fallback
|
||
except (TypeError, ValueError):
|
||
return fallback
|
||
|
||
|
||
def _write_note_text(note_dir: Path, note: dict[str, Any]) -> None:
|
||
tags = note.get("tags") or []
|
||
if not isinstance(tags, list):
|
||
tags = []
|
||
image_plan = note.get("imagePlan") or []
|
||
lines = [
|
||
f"【标题】{note.get('title', '')}",
|
||
"",
|
||
"【封面钩子】",
|
||
str(note.get("coverTitle") or ""),
|
||
"",
|
||
"【正文】",
|
||
str(note.get("content") or ""),
|
||
]
|
||
if tags:
|
||
lines += ["", "【标签】", " ".join(str(t) for t in tags)]
|
||
if image_plan:
|
||
lines += ["", "【图片排版说明】"]
|
||
for idx, plan in enumerate(image_plan, start=1):
|
||
if not isinstance(plan, dict):
|
||
continue
|
||
parts = [
|
||
f"{idx}. {plan.get('role') or f'图{idx}'}",
|
||
f"标题：{plan.get('title') or ''}",
|
||
f"叠字：{plan.get('overlayText') or ''}",
|
||
f"画面：{plan.get('text') or ''}",
|
||
]
|
||
lines.append(" / ".join(parts))
|
||
(note_dir / "文案.txt").write_text("\n".join(lines), encoding="utf-8")
|
||
|
||
|
||
def build_fission_package(ft, rows: list) -> str:
|
||
"""生成裂变下载 zip，返回 zip 本地路径。"""
|
||
upload_root = Path(get_settings().UPLOAD_ABS_ROOT).resolve()
|
||
package_dir = upload_root / "packages" / "fission" / str(ft.workspace_id) / str(ft.id)
|
||
if package_dir.exists():
|
||
shutil.rmtree(package_dir)
|
||
package_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
note_dirs: list[Path] = []
|
||
checklist: list[str] = [
|
||
"📋 裂变笔记发布清单",
|
||
f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}",
|
||
f"裂变ID：{ft.id}",
|
||
f"参考程度：{ft.reference_level}",
|
||
"",
|
||
]
|
||
missing_images: list[str] = []
|
||
|
||
for idx, row in enumerate(rows, start=1):
|
||
note = _loads_json(row.note_json, {})
|
||
images = _loads_json(row.images_json, [])
|
||
if not isinstance(note, dict):
|
||
note = {}
|
||
if not isinstance(images, list):
|
||
images = []
|
||
|
||
note_dir = package_dir / f"note_{idx:02d}"
|
||
note_dir.mkdir(parents=True, exist_ok=True)
|
||
note_dirs.append(note_dir)
|
||
_write_note_text(note_dir, note)
|
||
|
||
copied = 0
|
||
expected = 0
|
||
for img_idx, img in enumerate(images, start=1):
|
||
if not isinstance(img, dict) or img.get("error"):
|
||
continue
|
||
url = img.get("url") or img.get("image_url")
|
||
if not url:
|
||
continue
|
||
expected += 1
|
||
src = _resolve_upload_path(str(url), upload_root)
|
||
if not src.is_file():
|
||
missing_images.append(f"note_{idx:02d}: {url}")
|
||
continue
|
||
role = _safe_name(str(img.get("role") or f"img_{img_idx}"))
|
||
seq = img.get("seq") or img_idx
|
||
dst = note_dir / f"{int(seq):02d}_{role}.jpg"
|
||
shutil.copyfile(src, dst)
|
||
copied += 1
|
||
|
||
if expected and copied == 0:
|
||
raise ValueError(f"第 {idx} 套裂变图片读取失败，拒绝生成无图下载包")
|
||
|
||
title = str(note.get("title") or f"第{idx}套")
|
||
checklist.append(
|
||
f"note_{idx:02d} 标题：{title[:32]} 图片数：{copied} 分数：{row.score}"
|
||
)
|
||
|
||
if not note_dirs:
|
||
raise ValueError("裂变结果为空，无法下载")
|
||
|
||
if missing_images:
|
||
checklist += ["", "图片读取提示："]
|
||
checklist += [f"- {item}" for item in missing_images[:20]]
|
||
if len(missing_images) > 20:
|
||
checklist.append(f"- 另有 {len(missing_images) - 20} 张图片缺失")
|
||
checklist += [
|
||
"",
|
||
"使用说明：",
|
||
"- 每个 note_xx 文件夹是一套可独立发布的裂变笔记。",
|
||
"- 文案.txt 内含标题、封面钩子、正文、标签和图片排版说明。",
|
||
"- 图片按 01/02/03 顺序上传，避免传错序。",
|
||
]
|
||
(package_dir / "📋裂变发布清单.txt").write_text("\n".join(checklist), encoding="utf-8")
|
||
|
||
zip_path = package_dir / "fission_package.zip"
|
||
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
|
||
for note_dir in note_dirs:
|
||
for fpath in sorted(note_dir.iterdir()):
|
||
zf.write(fpath, arcname=f"{note_dir.name}/{fpath.name}")
|
||
zf.write(package_dir / "📋裂变发布清单.txt", arcname="📋裂变发布清单.txt")
|
||
|
||
logger.info("fission package built: %s notes=%d", zip_path, len(note_dirs))
|
||
return str(zip_path)
|