导出与视觉 QA 管线 深度解析

26from export_pptx import (
27    HOST_TEMPLATE,
28    BrowserSession,
29    ExportError,
30    build_payload,
31    ensure_agent_browser,
32    find_download,
33    find_manifest,
34    log,
35    ref_by_name,
36    serve,
37    wait_for_export_dialog,
38)

132def find_manifest(source: Path) -> Path:
133    source = source.expanduser().resolve()
134    if source.is_file():
135        if source.suffix.lower() != ".pptd":
136            raise ExportError(f"input must be a .pptd file or project directory: {source}")
137        return source
138    if not source.is_dir():
139        raise ExportError(f"input does not exist: {source}")
140    manifests = sorted(source.rglob("*.pptd"))
141    if not manifests:
142        raise ExportError(f"no .pptd manifest found under: {source}")
143    if len(manifests) > 1:
144        choices = "\n  ".join(str(path) for path in manifests[:20])
145        raise ExportError(
146            "multiple .pptd manifests found; pass one manifest explicitly:\n  " + choices
147        )
148    return manifests[0]
149
150
151def read_yaml_mapping(path: Path) -> Tuple[str, Dict[str, Any]]:
152    text = path.read_text(encoding="utf-8")
153    try:
154        value = yaml.safe_load(text)
155    except yaml.YAMLError as exc:
156        raise ExportError(f"invalid YAML in {path}: {exc}") from exc
157    if not isinstance(value, dict):
158        raise ExportError(f"expected a YAML mapping in {path}")
159    return text, value
160
161
162def safe_project_path(root: Path, relative: str) -> Path:
163    if not isinstance(relative, str) or not relative.strip():
164        raise ExportError("page path must be a non-empty string")
165    candidate = (root / relative).resolve()
166    try:
167        candidate.relative_to(root)
168    except ValueError as exc:
169        raise ExportError(f"project path escapes the PPTD directory: {relative}") from exc
170    return candidate
171
172
173def build_image_map(root: Path) -> Dict[str, str]:
174    image_map: Dict[str, str] = {}
175    total = 0
176    for path in sorted(root.rglob("*")):
177        if not path.is_file() or path.suffix.lower() not in IMAGE_MIME:
178            continue
179        size = path.stat().st_size
180        if size > MAX_IMAGE_BYTES:
181            log(f"skip local image over 20 MiB: {path.relative_to(root)}")
182            continue
183        if total + size > MAX_EMBEDDED_MEDIA_BYTES:
184            raise ExportError(
185                "local image payload exceeds 200 MiB; reduce media size or use remote URLs"
186            )
187        data = base64.b64encode(path.read_bytes()).decode("ascii")
188        rel = path.relative_to(root).as_posix()
189        image_map[rel] = f"data:{IMAGE_MIME[path.suffix.lower()]};base64,{data}"
190        total += size
191    if image_map:
192        log(f"prepared {len(image_map)} local image resource(s), {total} bytes")
193    return image_map
194
195
196def build_payload(manifest: Path) -> Dict[str, Any]:
197    manifest_text, manifest_data = read_yaml_mapping(manifest)
198    if manifest_data.get("version") != "v2":
199        raise ExportError("local PPTX export currently requires PPTD version: v2")
200    page_paths = manifest_data.get("pages")
201    if not isinstance(page_paths, list) or not page_paths:
202        raise ExportError("PPTD manifest must contain a non-empty pages list")
203
204    root = manifest.parent.resolve()
205    pages: List[Dict[str, str]] = []
206    for entry in page_paths:
207        page_path = safe_project_path(root, entry)
208        if not page_path.is_file():
209            raise ExportError(f"missing page file: {entry}")
210        page_text, page_data = read_yaml_mapping(page_path)
211        if not isinstance(page_data.get("elements"), list):
212            raise ExportError(f"page elements must be an array: {entry}")
213        pages.append({"path": str(entry), "content": page_text})
214
215    title = str(manifest_data.get("title") or manifest.stem)
216    return {
217        "id": f"local-export-{uuid.uuid4().hex}",
218        "title": title,
219        "manifestPath": manifest.name,
220        "manifestContent": manifest_text,
221        "pages": pages,
222        "imageMap": build_image_map(root),
223    }

240class BrowserSession:
241    def __init__(self, executable: str, session: str, cwd: Path, download_dir: Path):
242        self.executable = executable
243        self.session = session
244        self.cwd = cwd
245        self.download_dir = download_dir
246        self.env = os.environ.copy()
247        self.env.setdefault("AGENT_BROWSER_DEFAULT_TIMEOUT", "60000")
248        self.env.setdefault("AGENT_BROWSER_IDLE_TIMEOUT_MS", "180000")
249
250    def run(
251        self,
252        args: Sequence[str],
253        *,
254        timeout: int = 90,
255        check: bool = True,
256    ) -> subprocess.CompletedProcess[str]:
257        command = [self.executable, "--session", self.session, *args]
258        process = subprocess.run(
259            command,
260            cwd=self.cwd,
261            env=self.env,
262            text=True,
263            stdout=subprocess.PIPE,
264            stderr=subprocess.STDOUT,
265            timeout=timeout,
266        )
267        if check and process.returncode != 0:
268            raise ExportError(
269                f"agent-browser command failed ({process.returncode}): "
270                f"{' '.join(args)}\n{process.stdout[-4000:]}"
271            )
272        return process
273
274    def open(self, url: str) -> None:
275        self.run(
276            ["--download-path", str(self.download_dir), "open", url], timeout=90
277        )
278
279    def snapshot(self) -> Dict[str, Any]:
280        process = self.run(["snapshot", "-i", "-C", "--json"])
281        return json_result(process.stdout)
282
283    def close(self) -> None:
284        self.run(["close"], timeout=20, check=False)

376def replace_transition(slide_xml: bytes, transition: str) -> bytes:
377    text = slide_xml.decode("utf-8")
378    pattern = re.compile(
379        r"<p:transition\b[^>]*(?:/>|>.*?</p:transition>)", re.DOTALL
380    )
381    text = pattern.sub("", text)
382    if transition == "none":
383        return text.encode("utf-8")
384
385    # CT_Slide requires transition as a direct child after cSld/clrMapOvr and
386    # before timing/extLst. Searching for the first p:extLst is incorrect:
387    # shapes may contain their own nested extLst inside cSld, causing Office to
388    # ignore a transition inserted there.
389    color_map = re.search(
390        r"<p:clrMapOvr\b[^>]*(?:/>|>.*?</p:clrMapOvr>)", text, re.DOTALL
391    )
392    common_slide = re.search(
393        r"<p:cSld\b[^>]*(?:/>|>.*?</p:cSld>)", text, re.DOTALL
394    )
395    anchor = color_map or common_slide
396    if anchor is None:
397        raise ExportError("slide XML has no cSld/clrMapOvr insertion anchor")
398    position = anchor.end()
399    return (text[:position] + FADE_TRANSITION_XML + text[position:]).encode("utf-8")

508def export_pptx(
509    source: Path,
510    output: Path,
511    transition: str,
512    embed_fonts: bool,
513    keep_download: bool = False,
514    force: bool = False,
515) -> Dict[str, Any]:
516    manifest = find_manifest(source)
517    payload = build_payload(manifest)
518    output = output.expanduser().resolve()
519    output.parent.mkdir(parents=True, exist_ok=True)
520    if output.exists() and not force:
521        raise ExportError(f"output already exists (pass --force to replace it): {output}")
522    agent_browser = ensure_agent_browser()
523
524    log(f"manifest: {manifest}")
525    log(
526        f"defaults: transition={transition}, embed_fonts={'on' if embed_fonts else 'off'}"
527    )
528
529    with tempfile.TemporaryDirectory(prefix="open-kimi-ppt-export-") as temp_name:
530        temp_dir = Path(temp_name)
531        download_dir = temp_dir / "downloads"
532        download_dir.mkdir()
533        shutil.copy2(HOST_TEMPLATE, temp_dir / HOST_TEMPLATE.name)
534        (temp_dir / "payload.json").write_text(
535            json.dumps(payload, ensure_ascii=False), encoding="utf-8"
536        )
537        server, thread, url = serve(temp_dir)
538        session = f"open-kimi-ppt-export-{os.getpid()}-{uuid.uuid4().hex[:8]}"
539        browser = BrowserSession(agent_browser, session, temp_dir, download_dir)
540        try:
541            log("opening the public Kimi slide editor")
542            browser.open(url)
543            browser.run(
544                [
545                    "wait",
546                    "--fn",
547                    'document.documentElement.dataset.deckStatus === "ready"',
548                ],
549                timeout=120,
550            )
551            browser.run(["set", "viewport", "1280", "720"])
552            snapshot = browser.snapshot()
553            export_ref = ref_by_name(snapshot, "导出", "button")
554            browser.run(["click", f"@{export_ref}"])
555            dialog = wait_for_export_dialog(browser)
556
557            state = switch_state(dialog)
558            if state is not None:
559                switch_ref, checked, disabled = state
560                if disabled and checked != embed_fonts:
561                    log("warning: the official font switch is disabled for this deck")
562                elif checked != embed_fonts:
563                    browser.run(["click", f"@{switch_ref}"])
564                    dialog = wait_for_export_dialog(browser)
565            elif embed_fonts:
566                log("warning: the official export dialog exposed no font switch")
567
568            download_ref = ref_by_name(dialog, "下载", "button")
569            log("generating PPTX in the browser")
570            result = browser.run(
571                ["download", f"@{download_ref}", str(temp_dir / "browser-output.pptx")],
572                timeout=180,
573                check=False,
574            )
575            if result.returncode != 0:
576                log("download capture reported a timeout; checking browser output files")
577            downloaded = find_download((download_dir, temp_dir), timeout=90)
578            shutil.copy2(downloaded, output)
579            if keep_download:
580                debug_copy = output.with_name(f"{output.stem}.browser-raw.pptx")
581                if debug_copy.exists() and not force:
582                    raise ExportError(
583                        f"raw debug output already exists (pass --force): {debug_copy}"
584                    )
585                shutil.copy2(downloaded, debug_copy)
586        finally:
587            browser.close()
588            server.shutdown()
589            server.server_close()
590            thread.join(timeout=2)
591
592    slide_count = patch_transitions(output, transition)
593    summary = verify_output(output, transition, embed_fonts)
594    summary["transitionPatchedSlides"] = slide_count
595    summary["output"] = str(output)
596    return summary

86def unzip_images(archive_path: Path, pages_dir: Path) -> List[Path]:
87    pages_dir.mkdir(parents=True, exist_ok=True)
88    images: List[Path] = []
89    with zipfile.ZipFile(archive_path) as archive:
90        for info in archive.infolist():
91            if info.is_dir() or Path(info.filename).suffix.lower() not in IMAGE_SUFFIXES:
92                continue
93            name = Path(info.filename).name
94            if not name:
95                continue
96            target = pages_dir / name
97            with archive.open(info) as source, target.open("wb") as out:
98                shutil.copyfileobj(source, out)
99            images.append(target)
100    images.sort(key=page_sort_key)
101    if not images:
102        raise ExportError(f"no page images found in: {archive_path}")
103    return images
104
105
106def label_font(image_font: Any) -> Any:
107    try:
108        return image_font.load_default(size=18)
109    except TypeError:  # older Pillow without the size argument
110        return image_font.load_default()
111
112
113def stitch_overview(
114    images: Sequence[Path],
115    output: Path,
116    image_cls: Any,
117    draw_cls: Any,
118    image_font: Any,
119) -> Path:
120    thumbs: List[Tuple[str, Any]] = []
121    for index, path in enumerate(images, start=1):
122        with image_cls.open(path) as opened:
123            frame = opened.convert("RGB")
124            ratio = OVERVIEW_THUMB_WIDTH / frame.width
125            thumb = frame.resize(
126                (OVERVIEW_THUMB_WIDTH, max(1, round(frame.height * ratio)))
127            )
128        thumbs.append((f"P{index}", thumb))
129
130    columns = OVERVIEW_COLUMNS
131    rows = math.ceil(len(thumbs) / columns)
132    cell_height = OVERVIEW_LABEL_HEIGHT + max(thumb.height for _, thumb in thumbs)
133    width = columns * OVERVIEW_THUMB_WIDTH + (columns + 1) * OVERVIEW_GAP
134    height = rows * cell_height + (rows + 1) * OVERVIEW_GAP
135
136    overview = image_cls.new("RGB", (width, height), "#e5e7eb")
137    draw = draw_cls.Draw(overview)
138    font = label_font(image_font)
139    for position, (label, thumb) in enumerate(thumbs):
140        column = position % columns
141        row = position // columns
142        x = OVERVIEW_GAP + column * (OVERVIEW_THUMB_WIDTH + OVERVIEW_GAP)
143        y = OVERVIEW_GAP + row * (cell_height + OVERVIEW_GAP)
144        draw.rectangle(
145            (x, y, x + OVERVIEW_THUMB_WIDTH, y + OVERVIEW_LABEL_HEIGHT - 4),
146            fill="#111827",
147        )
148        draw.text((x + 8, y + 5), label, fill="#ffffff", font=font)
149        overview.paste(thumb, (x, y + OVERVIEW_LABEL_HEIGHT))
150
151    overview.save(output, "JPEG", quality=85)
152    return output