Skill 化中台：把 AI 办公、自媒体、短视频与电商通关串成一条流水线

from abc import ABC, abstractmethod
from pydantic import BaseModel
from typing import Any

class SkillInput(BaseModel): ...
class SkillOutput(BaseModel): ...

class Skill(ABC):
    name: str
    version: str
    @abstractmethod
    def run(self, data: SkillInput) -> SkillOutput: ...
    def can_retry(self, err: Exception) -> bool:
        return isinstance(err, (TimeoutError, ConnectionError))

REGISTRY: dict[str, Skill] = {}
def register(s: Skill):
    REGISTRY[f"{s.name}@{s.version}"] = s

from pydantic import Field

class SheetIn(SkillInput):
    rows: list[dict[str, Any]]
    ask: str

class ReportOut(SkillOutput):
    summary: str
    anomalies: list[dict] = Field(default_factory=list)

class OfficeReport(Skill):
    name, version = "office.report", "1.2.0"

    def __init__(self, llm): self.llm = llm

    def run(self, data: SheetIn) -> ReportOut:
        import json
        prompt = f"数据:{json.dumps(data.rows, ensure_ascii=False)}\n任务:{data.ask}\n" \
                 "只输出JSON: {summary, anomalies:[{row, reason}]}"
        raw = self.llm.complete(prompt, temperature=0)
        return ReportOut(**json.loads(raw))   # 校验失败即抛错，不静默放行

PLATFORM_SPEC = {
    "wechat":  {"max": 4000, "style": "长文，小标题分段"},
    "xhs":     {"max": 900,  "style": "口语化，emoji，首句钩子"},
    "twitter": {"max": 260,  "style": "短句，结论前置"},
}

class Repurpose(Skill):
    name, version = "media.repurpose", "2.0.0"
    def __init__(self, llm): self.llm = llm

    def run(self, data):
        out = {}
        for p, spec in PLATFORM_SPEC.items():
            text = self.llm.complete(
                f"改写为{spec['style']}，不超过{spec['max']}字，保留事实不改数字:\n{data.text}",
                temperature=0.4)
            out[p] = text[: spec["max"]]
        return out

import subprocess

class ShortVideo(Skill):
    name, version = "video.clip", "1.0.0"

    def run(self, data):
        clips = []
        for i, (start, dur) in enumerate(data.segments):
            dst = f"{data.out_dir}/clip_{i:02d}.mp4"
            subprocess.run([
                "ffmpeg", "-y", "-ss", str(start), "-i", data.src,
                "-t", str(dur), "-vf", "scale=1080:-2,subtitles=auto.srt",
                "-c:v", "libx264", "-preset", "fast", dst
            ], check=True, capture_output=True)
            clips.append(dst)
        return {"clips": clips}

BANNED = {"最", "第一", "国家级", "绝对"}
REQUIRED_ATTRS = {"electronics": {"voltage", "cert"}, "apparel": {"fabric", "size_chart"}}

class Listing(Skill):
    name, version = "ecom.listing", "3.1.0"

    def run(self, data):
        errors = []
        if REQUIRED_ATTRS.get(data.category, set()) - data.attrs.keys():
            errors.append(f"缺少必填属性: {REQUIRED_ATTRS[data.category] - data.attrs.keys()}")
        hit = BANNED & set(data.title.split())
        if hit:
            errors.append(f"标题含违禁词: {hit}")
        if errors:
            raise ValueError("; ".join(errors))     # 阻断上架，不进人工复核队列
        return {"status": "pass", "hs_code": self.classify(data)}

def retro(events: list[dict]) -> dict:
    agg = {}
    for e in events:
        k = (e["skill"], e["version"], e["platform"])
        s = agg.setdefault(k, {"n": 0, "conv": 0})
        s["n"] += 1; s["conv"] += e["converted"]
    return {f"{a}@{b}/{c}": round(v["conv"] / v["n"], 4) for (a, b, c), v in agg.items()}