#!/usr/bin/env python3 """First runnable version of the Xiaohongshu content approval flow. The image-generation and publishing providers are intentionally adapters. The demo provider reuses the approved local assets so the ingestion -> review -> notification path can be exercised without inventing product facts or touching a Xiaohongshu account. """ from __future__ import annotations import argparse import json import os from pathlib import Path import shutil import socket import subprocess import time import uuid from typing import Optional import db try: import qrcode except ImportError: qrcode = None ROOT = Path(__file__).resolve().parent TASKS_DIR = ROOT / "outputs" / "tasks" ASSETS_DIR = ROOT / "assets" / "raw" def local_ip() -> str: sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) try: sock.connect(("8.8.8.8", 80)) return sock.getsockname()[0] except OSError: return "127.0.0.1" finally: sock.close() def review_url(task_id: str) -> str: base = os.environ.get("REVIEW_PUBLIC_URL", "https://xhs.opennj.cn/review.html") return f"{base}{'&' if '?' in base else '?'}task={task_id}" def write_qr(url: str, path: Path) -> bool: if qrcode is None: return False path.parent.mkdir(parents=True, exist_ok=True) qrcode.make(url).save(path) return True def infer_product_name(photo: Path) -> str: stem = photo.stem.replace("_", " ").replace("-", " ").strip() return stem or "待识别商品" def build_content(product_name: str, feedback: str = "") -> dict: title = f"{product_name}怎么穿不显俗？这套搭配很加分" body = ( f"最近很喜欢这件{product_name}。\n\n" "版型和比例比较利落，日常搭配不会太用力；我会用低饱和配饰留一点留白，让单品成为重点。\n\n" "这版内容会把商品全貌、细节和上身效果按顺序展示，方便直接判断是否适合发布。\n\n" "你们会把它穿去什么场合？" ) if feedback: title = f"{title}（已按修改意见优化）" body += f"\n\n本版修改记录：{feedback}" return { "title": title, "cover_copy": f"{product_name}也可以很高级", "body": body, "topics": "#日常穿搭 #新品分享 #穿搭灵感 #夏日穿搭", } def demo_images(task_dir: Path, source: Path) -> list[dict]: """Use the uploaded photo plus existing generated assets for a deterministic demo.""" task_dir.mkdir(parents=True, exist_ok=True) source_copy = task_dir / f"source{source.suffix.lower() or '.jpg'}" shutil.copy2(source, source_copy) assets = [ (source_copy, "2 / 商品全貌：本次上传照片"), (ASSETS_DIR / "leopard-fabric.jpg", "3 / 面料细节：真实纹理"), (ASSETS_DIR / "leopard-dress-walking.png", "4 / 场景上身：行走动态"), (ASSETS_DIR / "leopard-dress-sideback.png", "5 / 背面版型：肩带和裙摆"), (ASSETS_DIR / "leopard-dress-detail.png", "6 / 细节特写：领口、肩带和腰线"), ] images = [{"url": "/assets/raw/leopard-ai-model.png", "caption": "1 / 封面：AI 模特正面上身"}] for path, caption in assets: if path == source_copy: url = f"/outputs/tasks/{task_dir.name}/{path.name}" else: url = f"/assets/raw/{path.name}" images.append({"url": url, "caption": caption}) return images def notify(task: dict, user_id: str, identity: str, send: bool) -> None: if not user_id: raise ValueError("notify requires --user-id (the Feishu recipient)") message = ( f"## 小红书内容待审核\n\n" f"商品：{task['product_name']}\n\n" f"版本：{task['version']}\n\n" f"[打开审核页面]({task['review_url']})\n\n" "通过后才会进入发布队列；驳回时请在页面填写修改意见。" ) command = ["lark-cli", "im", "+messages-send", "--as", identity, "--user-id", user_id, "--markdown", message] if not send: command.append("--dry-run") result = subprocess.run(command, cwd=ROOT, capture_output=True, text=True) if result.returncode: raise RuntimeError(result.stderr or result.stdout or "Feishu notification failed") print(result.stdout.strip()) def create_task(photo: Path, product_name: Optional[str], feedback: str = "", version: str = "v1", price: Optional[float] = None, product_url: str = "", selling_points: str = "") -> dict: if not photo.exists() or not photo.is_file(): raise FileNotFoundError(photo) task_id = f"xhs-{time.strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}" task_dir = TASKS_DIR / task_id product_name = product_name or infer_product_name(photo) content = build_content(product_name, feedback) task = { "task_id": task_id, "product_name": product_name, "version": version, "status": "待审核", "generation_mode": "demo-assets", "source_photo": str(photo), "price": price, "product_url": product_url, "selling_points": selling_points, **content, } task["images"] = demo_images(task_dir, photo) task["review_url"] = review_url(task_id) task["qr_url"] = task["review_url"] TASKS_DIR.mkdir(parents=True, exist_ok=True) (TASKS_DIR / f"{task_id}.json").write_text(json.dumps(task, ensure_ascii=False, indent=2), encoding="utf-8") write_qr(task["review_url"], task_dir / "qr.png") return task def cmd_ingest(args: argparse.Namespace) -> None: db.init_schema(args.database_url) task = create_task(Path(args.photo).resolve(), args.product_name, price=args.price, product_url=args.product_url, selling_points=args.selling_points) task["product_id"] = task["task_id"] db.save_task(task, args.database_url) (TASKS_DIR / f"{task['task_id']}.json").write_text(json.dumps(task, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({"task_id": task["task_id"], "review_url": task["review_url"]}, ensure_ascii=False, indent=2)) if args.user_id: notify(task, args.user_id, args.identity, args.send) def cmd_rework(args: argparse.Namespace) -> None: old_path = TASKS_DIR / f"{args.task_id}.json" old = json.loads(old_path.read_text(encoding="utf-8")) source = Path(old["source_photo"]) revision = int(old.get("version", "v1").lstrip("v")) + 1 db.init_schema(args.database_url) task = create_task(source, old["product_name"], args.feedback, f"v{revision}", old.get("price"), old.get("product_url", ""), old.get("selling_points", "")) task["parent_task_id"] = args.task_id task["product_id"] = old.get("product_id") or old.get("task_id") db.save_task(task, args.database_url) (TASKS_DIR / f"{task['task_id']}.json").write_text(json.dumps(task, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({"task_id": task["task_id"], "review_url": task["review_url"]}, ensure_ascii=False, indent=2)) if args.user_id: notify(task, args.user_id, args.identity, args.send) def cmd_import_json(args: argparse.Namespace) -> None: db.init_schema(args.database_url) source = Path(args.path) files = sorted(source.glob("*.json")) if source.is_dir() else [source] imported = 0 for path in files: task = json.loads(path.read_text(encoding="utf-8")) db.save_task(task, args.database_url) imported += 1 print(f"Imported {imported} task(s) into PostgreSQL") def parser() -> argparse.ArgumentParser: root = argparse.ArgumentParser(prog="xhs-flow", description="小红书图文内容审核流程") sub = root.add_subparsers(dest="command", required=True) init = sub.add_parser("init-db", help="初始化本地 PostgreSQL 数据库表") init.add_argument("--database-url", default="") init.set_defaults(func=lambda args: (db.init_schema(args.database_url), print(f"Database initialized: {db.database_url(args.database_url)}"))) ingest = sub.add_parser("ingest", help="导入一张商品照片并生成审核任务") ingest.add_argument("photo") ingest.add_argument("--product-name") ingest.add_argument("--price", type=float) ingest.add_argument("--product-url", default="") ingest.add_argument("--selling-points", default="") ingest.add_argument("--database-url", default="") ingest.add_argument("--user-id", help="飞书审核消息接收人的 open_id") ingest.add_argument("--identity", choices=["user", "bot"], default="user") ingest.add_argument("--send", action="store_true", help="实际发送飞书消息；默认只 dry-run") ingest.set_defaults(func=cmd_ingest) rework = sub.add_parser("rework", help="根据驳回意见创建下一版审核任务") rework.add_argument("task_id") rework.add_argument("--feedback", required=True) rework.add_argument("--database-url", default="") rework.add_argument("--user-id") rework.add_argument("--identity", choices=["user", "bot"], default="user") rework.add_argument("--send", action="store_true") rework.set_defaults(func=cmd_rework) imp = sub.add_parser("import-json", help="将现有任务 JSON 导入本地 PostgreSQL") imp.add_argument("path", help="任务 JSON 文件或目录") imp.add_argument("--database-url", default="") imp.set_defaults(func=cmd_import_json) return root if __name__ == "__main__": args = parser().parse_args() args.func(args)