Compare commits
1 Commits
main
...
codex/clip
| Author | SHA1 | Date |
|---|---|---|
 wantsong
|
4d37c93d87 |
|
|
@ -22,26 +22,26 @@ PowerShell:
|
|||
|
||||
```powershell
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill bundle-zip
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill clip-web-article
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill lifecycle-status-guard-scan
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill repair-markdown-citations
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill regression-validation-gate-runner
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill review-bundle-audit
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill review-context-builder
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill routing-behavior-diff-audit
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill voice-generation
|
||||
```
|
||||
|
||||
Use `-Force` to back up and replace an existing installed copy:
|
||||
|
||||
```powershell
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill bundle-zip -Force
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill clip-web-article -Force
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill lifecycle-status-guard-scan -Force
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill repair-markdown-citations -Force
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill regression-validation-gate-runner -Force
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill review-bundle-audit -Force
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill review-context-builder -Force
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill routing-behavior-diff-audit -Force
|
||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill voice-generation -Force
|
||||
```
|
||||
|
||||
`-All` installs only registry rows with an installable status and `default` install policy. Optional Skills must be named explicitly.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,90 @@
|
|||
# Clip Web Article Implementation Plan
|
||||
|
||||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
|
||||
|
||||
**Goal:** Build a lightweight `clip-web-article` Skill that saves one visible web article URL as one local Markdown file plus sibling image files.
|
||||
|
||||
**Architecture:** Use OpenCLI as the browser/login/article extraction substrate, then run a small Python normalizer to flatten OpenCLI output into the user's `knowledge-vault` article-reference shape. Keep deterministic naming, image rewrites, and safety checks in the script; keep platform policy and fallback routing in `SKILL.md`.
|
||||
|
||||
**Tech Stack:** OpenCLI `@jackwener/opencli`, Python 3.11 stdlib, `unittest`, existing `scripts/quick_validate.py`.
|
||||
|
||||
## Global Constraints
|
||||
|
||||
- Skill source lives under `C:\Users\wangq\Documents\Codex\skills-vault\skills\clip-web-article`.
|
||||
- Default output root is `C:\Users\wangq\Documents\Codex\knowledge-vault\sources\clipped-articles\文章参考`.
|
||||
- Output article folder contains only Markdown and local image files by default.
|
||||
- Do not create output logs, metadata JSON, raw HTML archives, screenshots, or platform subfolders by default.
|
||||
- Do not read, save, or ask the user to paste cookies, localStorage, passwords, session stores, or tokens.
|
||||
- Preserve existing outputs by creating a non-colliding folder.
|
||||
- Run tests with `conda run -n skills-vault python -B -m unittest discover -s skills\clip-web-article\tests -v`.
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Deterministic Normalizer
|
||||
|
||||
**Files:**
|
||||
- Create: `skills/clip-web-article/scripts/clip_web_article.py`
|
||||
- Create: `skills/clip-web-article/tests/test_clip_web_article.py`
|
||||
- Create: `skills/clip-web-article/fixtures/simple-opencli-output/`
|
||||
|
||||
**Interfaces:**
|
||||
- Produces: `normalize_opencli_article(opencli_dir: Path, output_root: Path, capture_date: date) -> ClipResult`
|
||||
- Produces: `save_twitter_rows(rows: list[dict], output_root: Path, url: str, capture_date: date) -> ClipResult`
|
||||
- Produces: CLI `python clip_web_article.py <url> --output-root <path>`
|
||||
|
||||
- [x] **Step 1: Write failing unit tests**
|
||||
- [x] **Step 2: Run tests and verify they fail because implementation is missing**
|
||||
- [x] **Step 3: Implement the normalizer and OpenCLI command routing**
|
||||
- [x] **Step 4: Run unit tests and verify they pass**
|
||||
|
||||
### Task 2: Skill Surface
|
||||
|
||||
**Files:**
|
||||
- Create: `skills/clip-web-article/SKILL.md`
|
||||
- Create: `skills/clip-web-article/README.md`
|
||||
- Create: `skills/clip-web-article/agents/openai.yaml`
|
||||
- Modify: `registry/skills-index.md`
|
||||
- Modify: `docs/install-sync.md`
|
||||
|
||||
**Interfaces:**
|
||||
- Consumes: script CLI from Task 1.
|
||||
- Produces: installed agent-facing Skill instructions and registry row.
|
||||
|
||||
- [x] **Step 1: Write concise Skill instructions**
|
||||
- [x] **Step 2: Add registry/install documentation**
|
||||
- [x] **Step 3: Validate with `scripts/quick_validate.py`**
|
||||
|
||||
### Task 3: Smoke And Install
|
||||
|
||||
**Files:**
|
||||
- Installed copy target: `C:\Users\wangq\.agents\skills\clip-web-article`
|
||||
|
||||
**Interfaces:**
|
||||
- Consumes: source Skill directory and install script.
|
||||
- Produces: installed runtime Skill.
|
||||
|
||||
- [x] **Step 1: Run OpenCLI help checks for required commands**
|
||||
- [x] **Step 2: Run unit tests and skill validation**
|
||||
- [x] **Step 3: Install Skill to `.agents\skills`**
|
||||
- [x] **Step 4: Attempt real URL smoke tests without saving cookies**
|
||||
|
||||
Smoke note: after installing the OpenCLI Chrome extension, `opencli doctor` reported daemon and extension connected. Real URL smoke tests passed for WeChat, Zhihu answer, Zhihu pin, and X/Twitter without manual cookies. Later refinement smoke tests verified batch clipping, custom `--output-root`, short names, no extra JSON/log/raw files, no Zhihu sibling answers, and no X replies.
|
||||
|
||||
### Task 4: User Feedback Refinements
|
||||
|
||||
**Files:**
|
||||
- Modify: `skills/clip-web-article/scripts/clip_web_article.py`
|
||||
- Modify: `skills/clip-web-article/tests/test_clip_web_article.py`
|
||||
- Modify: `skills/clip-web-article/SKILL.md`
|
||||
- Modify: `skills/clip-web-article/README.md`
|
||||
|
||||
**Interfaces:**
|
||||
- Produces: multi-URL CLI `python clip_web_article.py <url1> <url2> ... --output-root <path>`.
|
||||
- Produces: OpenCLI browser-backed commands with `--keep-tab false`.
|
||||
|
||||
- [x] **Step 1: Route Zhihu answer URLs through `opencli zhihu answer-detail`**
|
||||
- [x] **Step 2: Save only the target X status row, not replies**
|
||||
- [x] **Step 3: Trim Zhihu pin page chrome and comment UI**
|
||||
- [x] **Step 4: Shorten folder and Markdown file names**
|
||||
- [x] **Step 5: Add batch URL support and custom output-root coverage**
|
||||
- [x] **Step 6: Install and validate source plus installed Skill copies**
|
||||
|
|
@ -14,6 +14,7 @@ Existing CCPE content is not migrated here.
|
|||
| Skill | Purpose | Source | Status | Install Policy | Installed Path | CCPE Registration |
|
||||
| --- | --- | --- | --- | --- | --- | --- |
|
||||
| `bundle-zip` | Create zip archives from explicit file lists while preserving source-relative paths and validating entries by readback. | `C:\Users\wangq\Documents\Codex\skills-vault\skills\bundle-zip` | `installed` | `default` | `C:\Users\wangq\.agents\skills\bundle-zip` | none |
|
||||
| `clip-web-article` | Save one or more article URLs as lightweight Markdown plus local sibling images using OpenCLI browser/login state and no cookie-file workflow. | `C:\Users\wangq\Documents\Codex\skills-vault\skills\clip-web-article` | `installed` | `default` | `C:\Users\wangq\.agents\skills\clip-web-article` | none |
|
||||
| `fix-title` | Agentic semantic repair for copied GPT/LLM Markdown heading hierarchy across one or more discussion or artifact files, with fixed copies, heading plans, and reports. | `C:\Users\wangq\Documents\Codex\skills-vault\skills\fix-title` | `installed` | `default` | `C:\Users\wangq\.agents\skills\fix-title` | none |
|
||||
| `lifecycle-status-guard-scan` | Scan configured Markdown, JSON, YAML, and text files for lifecycle/status overclaim candidates and missing local evidence markers. | `C:\Users\wangq\Documents\Codex\skills-vault\skills\lifecycle-status-guard-scan` | `installed` | `default` | `C:\Users\wangq\.agents\skills\lifecycle-status-guard-scan` | none |
|
||||
| `repair-markdown-citations` | Repair ChatGPT/Deep Research Markdown citation tokens into standard footnotes and a deduplicated reference section using exact turn-ID metadata mapping. | `C:\Users\wangq\Documents\Codex\skills-vault\skills\repair-markdown-citations` | `installed` | `default` | `C:\Users\wangq\.agents\skills\repair-markdown-citations` | none |
|
||||
|
|
@ -21,7 +22,7 @@ Existing CCPE content is not migrated here.
|
|||
| `review-bundle-audit` | Preflight review, handoff, release, CCRA, Web upload, or code-review bundle directories for required files, manifests, sidecars, reports, zip readability, warnings, and blockers. | `C:\Users\wangq\Documents\Codex\skills-vault\skills\review-bundle-audit` | `installed` | `default` | `C:\Users\wangq\.agents\skills\review-bundle-audit` | none |
|
||||
| `review-context-builder` | Build file-first review, audit, planning, release, handoff, PR, or Agent invocation context indexes and manifests from configured local source roots, patterns, and metadata. | `C:\Users\wangq\Documents\Codex\skills-vault\skills\review-context-builder` | `installed` | `default` | `C:\Users\wangq\.agents\skills\review-context-builder` | none |
|
||||
| `routing-behavior-diff-audit` | Compare before/after selector, classifier, or routing outputs for targeted changes, collateral non-target changes, no-call changes, missing cases, and expected-route mismatch changes. | `C:\Users\wangq\Documents\Codex\skills-vault\skills\routing-behavior-diff-audit` | `installed` | `default` | `C:\Users\wangq\.agents\skills\routing-behavior-diff-audit` | none |
|
||||
| `voice-generation` | Generate spoken audio from Markdown scripts using the local `voice-gen` CLI and MiniMax `mmx`, including custom voice clone registration and voice registry management. | `C:\Users\wangq\Documents\Codex\skills-vault\skills\voice-generation` | `installed` | `conditional` | `C:\Users\wangq\.agents\skills\voice-generation` | none |
|
||||
| `voice-generation` | Generate spoken audio from Markdown scripts using the local `voice-gen` CLI and MiniMax `mmx`, including custom voice clone registration and voice registry management. | `D:\AI\ClaudeCode\voice-ge` | `installed` | `conditional` | `C:\Users\wangq\.agents\skills\voice-generation` | none |
|
||||
|
||||
## Status Values
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,20 @@
|
|||
# clip-web-article
|
||||
|
||||
Lightweight article clipping Skill source.
|
||||
|
||||
This Skill uses OpenCLI as the browser/login/extraction substrate, then normalizes output to short-named Markdown plus sibling `img_N.ext` files under the `knowledge-vault` article reference folder. Callers can pass `--output-root` to write to another destination.
|
||||
|
||||
It intentionally avoids logs, metadata JSON, raw HTML archives, screenshots, cookie files, and platform subfolders in the final clipped output.
|
||||
|
||||
Discussion-like URLs are clipped at the target boundary: a Zhihu answer URL saves only that answer, and an X/Twitter status URL saves only that status, not replies or comments. English sources keep the original Markdown and require a sibling Chinese translation from the calling agent.
|
||||
|
||||
The script accepts one or more URLs in a single call. Browser-backed OpenCLI commands use `--keep-tab false` so the tab lease opened for clipping is released after each command; the script does not close the user's Chrome process.
|
||||
|
||||
## Test
|
||||
|
||||
```powershell
|
||||
conda run -n skills-vault python -B -m unittest discover -s skills\clip-web-article\tests -v
|
||||
conda run -n skills-vault python -B scripts\quick_validate.py skills\clip-web-article
|
||||
```
|
||||
|
||||
Use `conda run --no-capture-output` for real URL clipping commands on Windows so OpenCLI/browser output is not re-encoded by conda.
|
||||
|
|
@ -0,0 +1,95 @@
|
|||
---
|
||||
name: clip-web-article
|
||||
description: Use when the user wants to download or clip a single web article URL into lightweight local Markdown with local sibling images, especially requests like "下载文章 URL" for Zhihu answers/articles/pins, WeChat public account articles, X/Twitter posts or threads, and ordinary web articles. Uses OpenCLI browser/login state and never asks the user to paste cookies.
|
||||
---
|
||||
|
||||
# Clip Web Article
|
||||
|
||||
Save one or more user-provided article URLs as discussion material. Each URL becomes one folder containing one Markdown file plus local image files.
|
||||
|
||||
## Default Output
|
||||
|
||||
Save to:
|
||||
|
||||
```text
|
||||
C:\Users\wangq\Documents\Codex\knowledge-vault\sources\clipped-articles\文章参考
|
||||
```
|
||||
|
||||
The default final shape is short enough for repeated daily use:
|
||||
|
||||
```text
|
||||
YYYYMMDD-短标题/
|
||||
短标题.md
|
||||
img_1.jpg
|
||||
img_2.png
|
||||
```
|
||||
|
||||
Do not add platform folders, metadata JSON, logs, raw HTML, screenshots, YAML frontmatter, or original URL lines by default.
|
||||
|
||||
Callers may override the destination:
|
||||
|
||||
```powershell
|
||||
conda run --no-capture-output -n skills-vault python -B .\scripts\clip_web_article.py "URL" --output-root "D:\target\folder"
|
||||
```
|
||||
|
||||
## Procedure
|
||||
|
||||
1. Confirm the input is one or more article-like URLs.
|
||||
2. Run the bundled script from this Skill directory:
|
||||
|
||||
```powershell
|
||||
conda run --no-capture-output -n skills-vault python -B .\scripts\clip_web_article.py "URL"
|
||||
```
|
||||
|
||||
3. For batch clipping, pass multiple URLs in one call:
|
||||
|
||||
```powershell
|
||||
conda run --no-capture-output -n skills-vault python -B .\scripts\clip_web_article.py "URL1" "URL2" "URL3"
|
||||
```
|
||||
|
||||
4. If running from another working directory, use the absolute script path:
|
||||
|
||||
```powershell
|
||||
conda run --no-capture-output -n skills-vault python -B "C:\Users\wangq\.agents\skills\clip-web-article\scripts\clip_web_article.py" "URL"
|
||||
```
|
||||
|
||||
5. If the script reports `Language: English`, keep the generated English Markdown as the original and create a sibling Chinese translation Markdown in the same folder. Preserve image links and code blocks. Do not replace the English original.
|
||||
6. Report only the saved folder, Markdown file name(s), image count, and any short failure/partial-success note.
|
||||
|
||||
## Routing
|
||||
|
||||
The script uses OpenCLI as the browser and extraction substrate:
|
||||
|
||||
- `mp.weixin.qq.com`: `opencli weixin download`, then flatten output.
|
||||
- `zhuanlan.zhihu.com`: `opencli zhihu download`, then flatten output.
|
||||
- `www.zhihu.com/question/.../answer/...`: `opencli zhihu answer-detail <answer-id>`, so only the target answer is saved.
|
||||
- `www.zhihu.com/pin/...`: `opencli web read`, then remove Zhihu page chrome, author cards, edit/comment UI, and keep the pin body plus body images.
|
||||
- `x.com` / `twitter.com`: `opencli twitter thread`, then save only the target status row and localize its media.
|
||||
- Other URLs: `opencli web read`, then flatten output.
|
||||
|
||||
All OpenCLI browser-backed commands pass `--keep-tab false` so the tab lease opened for clipping is released after the command. Do not close the user's Chrome process.
|
||||
|
||||
## Boundaries
|
||||
|
||||
- Use the user's connected OpenCLI Browser Bridge / logged-in browser state when needed.
|
||||
- Do not read or save raw cookies, localStorage, passwords, session stores, or tokens.
|
||||
- Do not ask the user to paste cookies.
|
||||
- Do not bypass CAPTCHA, verification gates, paywalls, DRM, anti-bot restrictions, or login walls.
|
||||
- Do not perform side-effect actions such as posting, liking, following, commenting, collecting, or sending messages.
|
||||
- Do not batch crawl feeds or accounts.
|
||||
- Save only the original article/post/answer body. Do not include comments, reply threads, other Zhihu answers, recommendation cards, profile cards, or page operation UI.
|
||||
- For discussion-like URLs, treat the URL target as the boundary: a Zhihu answer URL means that answer only; an X status URL means that status only.
|
||||
- If OpenCLI reports auth, verification, bridge, timeout, or blocked access, stop and ask the user to open the page in a connected browser or provide the content manually.
|
||||
|
||||
## Translation
|
||||
|
||||
For any English article/post from any source, preserve the English original Markdown and create a separate Chinese translation Markdown in the same folder. Do not append the translation into the original file and do not overwrite the original.
|
||||
|
||||
## Validation
|
||||
|
||||
For source changes, run:
|
||||
|
||||
```powershell
|
||||
conda run -n skills-vault python -B -m unittest discover -s skills\clip-web-article\tests -v
|
||||
conda run -n skills-vault python -B scripts\quick_validate.py skills\clip-web-article
|
||||
```
|
||||
|
|
@ -0,0 +1,6 @@
|
|||
interface:
|
||||
display_name: "Clip Web Article"
|
||||
short_description: "Save one article URL as Markdown plus images"
|
||||
default_prompt: "Use $clip-web-article to download this article URL into lightweight local Markdown with images."
|
||||
policy:
|
||||
allow_implicit_invocation: true
|
||||
|
|
@ -0,0 +1,671 @@
|
|||
#!/usr/bin/env python3
|
||||
"""Clip one visible web article into a lightweight Markdown folder."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, NamedTuple
|
||||
|
||||
|
||||
DEFAULT_OUTPUT_ROOT = Path(
|
||||
r"C:\Users\wangq\Documents\Codex\knowledge-vault\sources\clipped-articles\文章参考"
|
||||
)
|
||||
DEFAULT_WORK_ROOT = Path(r"C:\tmp\clip-web-article")
|
||||
|
||||
INVALID_WINDOWS_CHARS_RE = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
|
||||
IMAGE_RE = re.compile(r"!\[([^\]]*)\]\(([^)]+)\)")
|
||||
METADATA_RE = re.compile(r"^>\s*(作者|公众号|发布时间|原文链接|author|published|publish time|source url)\s*[:：]", re.I)
|
||||
DATE_RE = re.compile(r"(\d{4})[-/年.](\d{1,2})[-/月.](\d{1,2})")
|
||||
|
||||
|
||||
class ClipError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class ClipResult(NamedTuple):
|
||||
title: str
|
||||
output_dir: Path
|
||||
markdown_path: Path
|
||||
image_count: int
|
||||
language: str
|
||||
|
||||
|
||||
class ClipBatchItem(NamedTuple):
|
||||
url: str
|
||||
result: ClipResult | None
|
||||
error: str
|
||||
|
||||
|
||||
class OpenCliPlan(NamedTuple):
|
||||
kind: str
|
||||
command: list[str]
|
||||
output_dir: Path
|
||||
|
||||
|
||||
def opencli_binary() -> str:
|
||||
return "opencli.cmd" if os.name == "nt" else "opencli"
|
||||
|
||||
|
||||
def release_tab_args() -> list[str]:
|
||||
return ["--keep-tab", "false"]
|
||||
|
||||
|
||||
def sanitize_windows_name(text: str, max_len: int = 80) -> str:
|
||||
value = html.unescape(text or "").replace("\ufeff", "")
|
||||
value = INVALID_WINDOWS_CHARS_RE.sub("", value)
|
||||
value = re.sub(r"\s+", " ", value).strip().rstrip(". ")
|
||||
if len(value) > max_len:
|
||||
value = value[:max_len].rstrip(". ")
|
||||
return value or "untitled"
|
||||
|
||||
|
||||
def compact_title(text: str, max_len: int = 24) -> str:
|
||||
safe = sanitize_windows_name(text, max_len=max_len)
|
||||
safe = re.sub(r"\s+", "", safe)
|
||||
return safe[:max_len] or "untitled"
|
||||
|
||||
|
||||
def short_file_stem(text: str, max_len: int = 32) -> str:
|
||||
return sanitize_windows_name(text, max_len=max_len)
|
||||
|
||||
|
||||
def short_folder_name(title: str, article_date: date, max_len: int = 30) -> str:
|
||||
title_part = compact_title(title, max_len=12)
|
||||
folder = f"{article_date:%Y%m%d}-{title_part}"
|
||||
return sanitize_windows_name(folder, max_len=max_len)
|
||||
|
||||
|
||||
def unique_dir(parent: Path, name: str) -> Path:
|
||||
candidate = parent / name
|
||||
if not candidate.exists():
|
||||
return candidate
|
||||
index = 2
|
||||
while True:
|
||||
candidate = parent / f"{name}-{index}"
|
||||
if not candidate.exists():
|
||||
return candidate
|
||||
index += 1
|
||||
|
||||
|
||||
def parse_date_from_text(text: str) -> date | None:
|
||||
match = DATE_RE.search(text or "")
|
||||
if not match:
|
||||
return None
|
||||
year, month, day = (int(part) for part in match.groups())
|
||||
try:
|
||||
return date(year, month, day)
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def extract_title(markdown: str, fallback: str = "untitled") -> str:
|
||||
for line in markdown.splitlines():
|
||||
stripped = line.strip()
|
||||
if stripped.startswith("# "):
|
||||
return stripped[2:].strip() or fallback
|
||||
return fallback
|
||||
|
||||
|
||||
def extract_header_date(markdown: str) -> date | None:
|
||||
for line in markdown.splitlines()[:20]:
|
||||
if line.lstrip().startswith(">"):
|
||||
parsed = parse_date_from_text(line)
|
||||
if parsed:
|
||||
return parsed
|
||||
return None
|
||||
|
||||
|
||||
def tidy_markdown(markdown: str) -> str:
|
||||
text = markdown.replace("\u200b", "").replace("\ufeff", "")
|
||||
lines = [line.rstrip() for line in text.splitlines()]
|
||||
text = "\n".join(lines).strip()
|
||||
text = re.sub(r"\n{3,}", "\n\n", text)
|
||||
return text.strip() + ("\n" if text.strip() else "")
|
||||
|
||||
|
||||
def strip_zhihu_pin_page_chrome(title: str, body: str) -> str:
|
||||
lines = body.splitlines()
|
||||
title_key = re.split(r"[|｜]", title, maxsplit=1)[0].strip()
|
||||
|
||||
start = 0
|
||||
if title_key:
|
||||
for index, line in enumerate(lines):
|
||||
stripped = line.strip()
|
||||
if stripped and title_key in stripped and not stripped.startswith(("[", "!", ">")):
|
||||
start = index + 1
|
||||
break
|
||||
|
||||
kept: list[str] = []
|
||||
for line in lines[start:]:
|
||||
stripped = line.strip()
|
||||
if "编辑于" in stripped and "zhihu.com/pin/" in stripped:
|
||||
break
|
||||
if "IP 属地" in stripped and "编辑于" in stripped:
|
||||
break
|
||||
if re.fullmatch(r"\+\d+", stripped):
|
||||
continue
|
||||
if stripped in {"默认", "最新", "理性发言，友善互动"}:
|
||||
continue
|
||||
kept.append(line)
|
||||
|
||||
return tidy_markdown("\n".join(kept))
|
||||
|
||||
|
||||
def clean_opencli_markdown(markdown: str, source_kind: str = "generic") -> str:
|
||||
lines = markdown.replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
||||
if not lines:
|
||||
return ""
|
||||
|
||||
title_line = ""
|
||||
body_start = 0
|
||||
for index, line in enumerate(lines):
|
||||
if line.strip().startswith("# "):
|
||||
title_line = line.strip()
|
||||
body_start = index + 1
|
||||
break
|
||||
if not title_line:
|
||||
title_line = "# untitled"
|
||||
|
||||
while body_start < len(lines):
|
||||
stripped = lines[body_start].strip()
|
||||
if not stripped or stripped == "---" or METADATA_RE.match(stripped):
|
||||
body_start += 1
|
||||
continue
|
||||
break
|
||||
|
||||
body = "\n".join(lines[body_start:]).strip()
|
||||
title = title_line[2:].strip() if title_line.startswith("# ") else title_line.lstrip("# ").strip()
|
||||
if source_kind == "zhihu-pin":
|
||||
body = strip_zhihu_pin_page_chrome(title, body).strip()
|
||||
else:
|
||||
body = tidy_markdown(body).strip()
|
||||
return f"{title_line}\n\n{body}\n" if body else f"{title_line}\n"
|
||||
|
||||
|
||||
def image_extension_from_url(url: str, default: str = ".jpg") -> str:
|
||||
parsed = urllib.parse.urlparse(url)
|
||||
suffix = Path(parsed.path).suffix.lower()
|
||||
if suffix in {".jpg", ".jpeg", ".png", ".gif", ".webp", ".bmp", ".avif"}:
|
||||
return suffix
|
||||
query = urllib.parse.parse_qs(parsed.query)
|
||||
fmt = (query.get("wx_fmt") or query.get("format") or [""])[0].lower()
|
||||
if fmt in {"jpg", "jpeg", "png", "gif", "webp", "bmp", "avif"}:
|
||||
return f".{fmt}"
|
||||
return default
|
||||
|
||||
|
||||
def download_remote_image(url: str, target: Path, referer: str = "") -> bool:
|
||||
headers = {"User-Agent": "Mozilla/5.0"}
|
||||
if referer:
|
||||
headers["Referer"] = referer
|
||||
try:
|
||||
request = urllib.request.Request(url, headers=headers)
|
||||
with urllib.request.urlopen(request, timeout=20) as response:
|
||||
target.write_bytes(response.read())
|
||||
return target.exists() and target.stat().st_size > 0
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
def localize_markdown_images(markdown: str, source_base: Path, output_dir: Path, referer: str = "") -> tuple[str, int]:
|
||||
replacements: dict[str, str] = {}
|
||||
image_count = 0
|
||||
|
||||
for _alt, raw_url in IMAGE_RE.findall(markdown):
|
||||
if raw_url in replacements:
|
||||
continue
|
||||
if raw_url.startswith("data:"):
|
||||
continue
|
||||
|
||||
parsed = urllib.parse.urlparse(raw_url)
|
||||
local_source: Path | None = None
|
||||
ext = ".jpg"
|
||||
|
||||
if parsed.scheme in {"http", "https"}:
|
||||
ext = image_extension_from_url(raw_url)
|
||||
elif parsed.scheme == "file":
|
||||
local_source = Path(urllib.request.url2pathname(parsed.path))
|
||||
ext = local_source.suffix or ".jpg"
|
||||
else:
|
||||
local_source = (source_base / urllib.parse.unquote(raw_url)).resolve()
|
||||
ext = local_source.suffix or ".jpg"
|
||||
|
||||
target_name = f"img_{image_count + 1}{ext}"
|
||||
target = output_dir / target_name
|
||||
copied = False
|
||||
|
||||
if local_source is not None:
|
||||
try:
|
||||
if local_source.exists() and local_source.is_file():
|
||||
shutil.copy2(local_source, target)
|
||||
copied = True
|
||||
except Exception:
|
||||
copied = False
|
||||
else:
|
||||
copied = download_remote_image(raw_url, target, referer=referer)
|
||||
|
||||
if copied:
|
||||
image_count += 1
|
||||
replacements[raw_url] = target_name
|
||||
|
||||
def replace(match: re.Match[str]) -> str:
|
||||
alt = match.group(1)
|
||||
raw = match.group(2)
|
||||
return f"" if raw in replacements else match.group(0)
|
||||
|
||||
return IMAGE_RE.sub(replace, markdown), image_count
|
||||
|
||||
|
||||
def find_primary_markdown(opencli_dir: Path) -> Path:
|
||||
candidates = [path for path in opencli_dir.rglob("*.md") if path.is_file()]
|
||||
if not candidates:
|
||||
raise ClipError(f"No Markdown file found under OpenCLI output: {opencli_dir}")
|
||||
return max(candidates, key=lambda p: p.stat().st_size)
|
||||
|
||||
|
||||
def normalize_opencli_article(
|
||||
opencli_dir: Path,
|
||||
output_root: Path,
|
||||
capture_date: date | None = None,
|
||||
source_kind: str = "generic",
|
||||
) -> ClipResult:
|
||||
capture_date = capture_date or date.today()
|
||||
markdown_path = find_primary_markdown(Path(opencli_dir))
|
||||
raw_markdown = markdown_path.read_text(encoding="utf-8")
|
||||
title = extract_title(raw_markdown, fallback=markdown_path.stem)
|
||||
article_date = extract_header_date(raw_markdown)
|
||||
if article_date is None and source_kind == "zhihu-pin":
|
||||
article_date = parse_date_from_text(raw_markdown)
|
||||
article_date = article_date or capture_date
|
||||
folder_name = short_folder_name(title, article_date)
|
||||
output_dir = unique_dir(Path(output_root), folder_name)
|
||||
output_dir.mkdir(parents=True, exist_ok=False)
|
||||
|
||||
cleaned = clean_opencli_markdown(raw_markdown, source_kind=source_kind)
|
||||
localized, image_count = localize_markdown_images(cleaned, markdown_path.parent, output_dir)
|
||||
localized = tidy_markdown(localized)
|
||||
output_md = output_dir / f"{short_file_stem(title)}.md"
|
||||
output_md.write_text(localized, encoding="utf-8")
|
||||
return ClipResult(title, output_dir, output_md, image_count, detect_language(localized))
|
||||
|
||||
|
||||
def normalize_media_urls(value: Any) -> list[str]:
|
||||
if not value:
|
||||
return []
|
||||
if isinstance(value, list):
|
||||
return [str(item) for item in value if str(item).strip()]
|
||||
if isinstance(value, str):
|
||||
parts = re.split(r"[\n,]+", value)
|
||||
return [part.strip() for part in parts if part.strip().startswith(("http://", "https://"))]
|
||||
return []
|
||||
|
||||
|
||||
def detect_language(text: str) -> str:
|
||||
clean = re.sub(r"!\[[^\]]*\]\([^)]+\)|https?://\S+|[#*>`~_\-|]", "", text)
|
||||
cjk = len(re.findall(r"[\u4e00-\u9fff]", clean))
|
||||
latin = len(re.findall(r"[A-Za-z]", clean))
|
||||
if cjk and latin:
|
||||
return "mixed"
|
||||
if cjk:
|
||||
return "chinese"
|
||||
if latin:
|
||||
return "english"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def title_from_text(text: str, fallback: str = "tweet") -> str:
|
||||
normalized = re.sub(r"\s+", " ", text or "").strip()
|
||||
return normalized[:40].rstrip(".,;:!? ") or fallback
|
||||
|
||||
|
||||
def extract_zhihu_answer_id(url: str) -> str:
|
||||
match = re.search(r"/answer/(\d+)", url)
|
||||
return match.group(1) if match else ""
|
||||
|
||||
|
||||
def extract_twitter_status_id(url: str) -> str:
|
||||
match = re.search(r"/status/(\d+)", url)
|
||||
if match:
|
||||
return match.group(1)
|
||||
return url if re.fullmatch(r"\d+", url) else ""
|
||||
|
||||
|
||||
def parse_iso_date(value: str) -> date | None:
|
||||
if not value:
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(value.replace("Z", "+00:00")).date()
|
||||
except ValueError:
|
||||
pass
|
||||
try:
|
||||
return datetime.strptime(value, "%a %b %d %H:%M:%S %z %Y").date()
|
||||
except ValueError:
|
||||
return parse_date_from_text(value)
|
||||
|
||||
|
||||
def save_zhihu_answer_rows(rows: list[dict[str, Any]], output_root: Path, capture_date: date | None = None) -> ClipResult:
|
||||
capture_date = capture_date or date.today()
|
||||
if not rows:
|
||||
raise ClipError("Zhihu did not return the requested answer")
|
||||
row = rows[0]
|
||||
content = tidy_markdown(str(row.get("content", "")).strip())
|
||||
if not content:
|
||||
raise ClipError("Zhihu answer-detail returned no readable content")
|
||||
|
||||
title = str(row.get("question_title") or "zhihu-answer").strip()
|
||||
article_date = parse_iso_date(str(row.get("created_at", ""))) or capture_date
|
||||
output_dir = unique_dir(Path(output_root), short_folder_name(title, article_date))
|
||||
output_dir.mkdir(parents=True, exist_ok=False)
|
||||
|
||||
markdown = tidy_markdown(f"# {title}\n\n{content}")
|
||||
output_md = output_dir / f"{short_file_stem(title)}.md"
|
||||
output_md.write_text(markdown, encoding="utf-8")
|
||||
return ClipResult(title, output_dir, output_md, 0, detect_language(markdown))
|
||||
|
||||
|
||||
def save_twitter_rows(rows: list[dict[str, Any]], output_root: Path, url: str, capture_date: date | None = None) -> ClipResult:
|
||||
capture_date = capture_date or date.today()
|
||||
target_id = extract_twitter_status_id(url)
|
||||
useful_rows = [row for row in rows if str(row.get("text", "")).strip()]
|
||||
if target_id:
|
||||
useful_rows = [row for row in useful_rows if str(row.get("id", "")) == target_id]
|
||||
elif useful_rows:
|
||||
useful_rows = [useful_rows[0]]
|
||||
if not useful_rows:
|
||||
raise ClipError("Twitter/X did not return readable tweet text")
|
||||
|
||||
title = title_from_text(str(useful_rows[0].get("text", "")), fallback="tweet")
|
||||
article_date = parse_iso_date(str(useful_rows[0].get("created_at", ""))) or capture_date
|
||||
output_dir = unique_dir(Path(output_root), short_folder_name(title, article_date))
|
||||
output_dir.mkdir(parents=True, exist_ok=False)
|
||||
|
||||
image_count = 0
|
||||
parts = [f"# {title}", ""]
|
||||
for row in useful_rows:
|
||||
text = str(row.get("text", "")).strip()
|
||||
parts.append(text)
|
||||
for media_url in normalize_media_urls(row.get("media_urls")):
|
||||
ext = image_extension_from_url(media_url)
|
||||
target_name = f"img_{image_count + 1}{ext}"
|
||||
if download_remote_image(media_url, output_dir / target_name, referer=url):
|
||||
image_count += 1
|
||||
parts.append("")
|
||||
parts.append(f"")
|
||||
else:
|
||||
parts.append("")
|
||||
parts.append(f"")
|
||||
parts.append("")
|
||||
|
||||
markdown = tidy_markdown("\n".join(parts))
|
||||
output_md = output_dir / f"{short_file_stem(title)}.md"
|
||||
output_md.write_text(markdown, encoding="utf-8")
|
||||
return ClipResult(title, output_dir, output_md, image_count, detect_language(markdown))
|
||||
|
||||
|
||||
def plan_opencli_invocation(url: str, work_dir: Path) -> OpenCliPlan:
|
||||
parsed = urllib.parse.urlparse(url)
|
||||
host = parsed.hostname.lower() if parsed.hostname else ""
|
||||
output_dir = Path(work_dir) / "opencli-output"
|
||||
base = opencli_binary()
|
||||
|
||||
if host in {"www.zhihu.com", "zhihu.com"} or host.endswith(".zhihu.com"):
|
||||
answer_id = extract_zhihu_answer_id(url)
|
||||
if answer_id:
|
||||
return OpenCliPlan(
|
||||
"opencli-zhihu-answer-detail",
|
||||
[base, "zhihu", "answer-detail", answer_id, *release_tab_args(), "-f", "json"],
|
||||
output_dir,
|
||||
)
|
||||
if parsed.path.startswith("/pin/"):
|
||||
return OpenCliPlan(
|
||||
"opencli-zhihu-pin-web-read",
|
||||
[
|
||||
base,
|
||||
"web",
|
||||
"read",
|
||||
"--url",
|
||||
url,
|
||||
"--output",
|
||||
str(output_dir),
|
||||
"--download-images",
|
||||
"true",
|
||||
"--wait",
|
||||
"5",
|
||||
*release_tab_args(),
|
||||
"-f",
|
||||
"json",
|
||||
],
|
||||
output_dir,
|
||||
)
|
||||
|
||||
if host == "mp.weixin.qq.com":
|
||||
return OpenCliPlan(
|
||||
"opencli-download",
|
||||
[
|
||||
base,
|
||||
"weixin",
|
||||
"download",
|
||||
"--url",
|
||||
url,
|
||||
"--output",
|
||||
str(output_dir),
|
||||
"--download-images",
|
||||
"true",
|
||||
*release_tab_args(),
|
||||
"-f",
|
||||
"json",
|
||||
],
|
||||
output_dir,
|
||||
)
|
||||
if host in {"x.com", "twitter.com"} or host.endswith(".x.com") or host.endswith(".twitter.com"):
|
||||
return OpenCliPlan(
|
||||
"opencli-twitter-thread",
|
||||
[base, "twitter", "thread", url, "--limit", "50", *release_tab_args(), "-f", "json"],
|
||||
output_dir,
|
||||
)
|
||||
if host == "zhuanlan.zhihu.com":
|
||||
return OpenCliPlan(
|
||||
"opencli-download",
|
||||
[
|
||||
base,
|
||||
"zhihu",
|
||||
"download",
|
||||
"--url",
|
||||
url,
|
||||
"--output",
|
||||
str(output_dir),
|
||||
"--download-images",
|
||||
"true",
|
||||
*release_tab_args(),
|
||||
"-f",
|
||||
"json",
|
||||
],
|
||||
output_dir,
|
||||
)
|
||||
|
||||
return OpenCliPlan(
|
||||
"opencli-web-read",
|
||||
[
|
||||
base,
|
||||
"web",
|
||||
"read",
|
||||
"--url",
|
||||
url,
|
||||
"--output",
|
||||
str(output_dir),
|
||||
"--download-images",
|
||||
"true",
|
||||
"--wait",
|
||||
"5",
|
||||
*release_tab_args(),
|
||||
"-f",
|
||||
"json",
|
||||
],
|
||||
output_dir,
|
||||
)
|
||||
|
||||
|
||||
def parse_opencli_json(stdout: str) -> Any:
|
||||
text = stdout.strip()
|
||||
if not text:
|
||||
return None
|
||||
try:
|
||||
return json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
starts = [pos for pos in (text.find("["), text.find("{")) if pos >= 0]
|
||||
if not starts:
|
||||
return None
|
||||
start = min(starts)
|
||||
end = max(text.rfind("]"), text.rfind("}"))
|
||||
if end <= start:
|
||||
return None
|
||||
return json.loads(text[start : end + 1])
|
||||
|
||||
|
||||
def rows_from_json(value: Any) -> list[dict[str, Any]]:
|
||||
if value is None:
|
||||
return []
|
||||
if isinstance(value, list):
|
||||
return [row for row in value if isinstance(row, dict)]
|
||||
if isinstance(value, dict):
|
||||
for key in ("rows", "data", "result", "items"):
|
||||
nested = value.get(key)
|
||||
if isinstance(nested, list):
|
||||
return [row for row in nested if isinstance(row, dict)]
|
||||
return []
|
||||
|
||||
|
||||
def format_opencli_failure(returncode: int, stderr: str, stdout: str) -> str:
|
||||
combined = "\n".join(part for part in [stderr, stdout] if part).strip()
|
||||
lowered = combined.lower()
|
||||
if "exitcode: 69" in lowered or "browser bridge" in lowered:
|
||||
return "OpenCLI Browser Bridge extension not connected. Run opencli doctor, connect the extension, then retry."
|
||||
if "exitcode: 77" in lowered or "auth required" in lowered or "not logged" in lowered:
|
||||
return "OpenCLI needs a logged-in browser session for this page. Open the page in the connected browser, then retry."
|
||||
if "verification" in lowered or "captcha" in lowered:
|
||||
return "the page requires verification. Complete it in the connected browser, then retry."
|
||||
lines = combined.splitlines()
|
||||
return lines[-1].strip() if lines else f"exit {returncode}"
|
||||
|
||||
|
||||
def run_opencli(plan: OpenCliPlan, timeout: int = 240) -> Any:
|
||||
completed = subprocess.run(
|
||||
plan.command,
|
||||
text=True,
|
||||
encoding="utf-8",
|
||||
errors="replace",
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
timeout=timeout,
|
||||
)
|
||||
if completed.returncode != 0:
|
||||
raise ClipError(f"OpenCLI failed: {format_opencli_failure(completed.returncode, completed.stderr, completed.stdout)}")
|
||||
return parse_opencli_json(completed.stdout)
|
||||
|
||||
|
||||
def clip_url(url: str, output_root: Path, work_dir: Path, capture_date: date | None = None) -> ClipResult:
|
||||
plan = plan_opencli_invocation(url, work_dir)
|
||||
work_dir.mkdir(parents=True, exist_ok=True)
|
||||
data = run_opencli(plan)
|
||||
if plan.kind == "opencli-zhihu-answer-detail":
|
||||
rows = rows_from_json(data)
|
||||
return save_zhihu_answer_rows(rows, output_root, capture_date=capture_date)
|
||||
if plan.kind == "opencli-twitter-thread":
|
||||
rows = rows_from_json(data)
|
||||
return save_twitter_rows(rows, output_root, url, capture_date=capture_date)
|
||||
source_kind = "zhihu-pin" if plan.kind == "opencli-zhihu-pin-web-read" else "generic"
|
||||
return normalize_opencli_article(plan.output_dir, output_root, capture_date=capture_date, source_kind=source_kind)
|
||||
|
||||
|
||||
def batch_work_dir(work_root: Path, index: int, total: int) -> Path:
|
||||
return Path(work_root) if total == 1 else Path(work_root) / f"{index:03d}"
|
||||
|
||||
|
||||
def clip_urls(
|
||||
urls: list[str],
|
||||
output_root: Path,
|
||||
work_root: Path,
|
||||
keep_work: bool = False,
|
||||
capture_date: date | None = None,
|
||||
) -> list[ClipBatchItem]:
|
||||
items: list[ClipBatchItem] = []
|
||||
total = len(urls)
|
||||
for index, url in enumerate(urls, start=1):
|
||||
work_dir = batch_work_dir(work_root, index, total)
|
||||
try:
|
||||
result = clip_url(url, output_root, work_dir, capture_date=capture_date)
|
||||
items.append(ClipBatchItem(url, result, ""))
|
||||
except ClipError as exc:
|
||||
items.append(ClipBatchItem(url, None, str(exc)))
|
||||
finally:
|
||||
if not keep_work and work_dir.exists():
|
||||
shutil.rmtree(work_dir, ignore_errors=True)
|
||||
if not keep_work and total > 1 and Path(work_root).exists():
|
||||
try:
|
||||
Path(work_root).rmdir()
|
||||
except OSError:
|
||||
pass
|
||||
return items
|
||||
|
||||
|
||||
def make_work_dir(root: Path) -> Path:
|
||||
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
|
||||
return root / f"{stamp}-{os.getpid()}"
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
parser = argparse.ArgumentParser(description="Clip visible web article URL(s) to Markdown plus local images.")
|
||||
parser.add_argument("urls", nargs="+", help="One or more article, Zhihu, WeChat, X/Twitter, or ordinary web URLs")
|
||||
parser.add_argument("--output-root", default=str(DEFAULT_OUTPUT_ROOT), help="Final clipped article output root")
|
||||
parser.add_argument("--work-dir", default="", help="Temporary OpenCLI output directory")
|
||||
parser.add_argument("--keep-work", action="store_true", help="Keep temporary OpenCLI output for debugging")
|
||||
return parser
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
args = build_parser().parse_args(argv)
|
||||
output_root = Path(args.output_root)
|
||||
work_root = Path(args.work_dir) if args.work_dir else make_work_dir(DEFAULT_WORK_ROOT)
|
||||
items = clip_urls(args.urls, output_root, work_root, keep_work=args.keep_work)
|
||||
failures = 0
|
||||
|
||||
for index, item in enumerate(items, start=1):
|
||||
if len(items) > 1:
|
||||
print(f"[{index}/{len(items)}] {item.url}")
|
||||
if item.error or item.result is None:
|
||||
failures += 1
|
||||
print(f"未能抓取正文。原因: {item.error}")
|
||||
print("建议: 在已登录浏览器中打开该页面后重试，或手动提供正文。")
|
||||
print()
|
||||
continue
|
||||
|
||||
result = item.result
|
||||
print("已保存到:")
|
||||
print(result.output_dir)
|
||||
print()
|
||||
print(f"Markdown: {result.markdown_path.name}")
|
||||
print(f"Images: {result.image_count}")
|
||||
if result.language == "english":
|
||||
print("Language: English (按 Skill 规则保留英文原文，并生成同目录中文译文)")
|
||||
elif result.language != "unknown":
|
||||
print(f"Language: {result.language}")
|
||||
print()
|
||||
|
||||
return 1 if failures else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -0,0 +1,281 @@
|
|||
import importlib.util
|
||||
import shutil
|
||||
import tempfile
|
||||
import threading
|
||||
import unittest
|
||||
from datetime import date
|
||||
from http.server import SimpleHTTPRequestHandler
|
||||
from pathlib import Path
|
||||
from socketserver import TCPServer
|
||||
|
||||
|
||||
SCRIPT_PATH = Path(__file__).resolve().parents[1] / "scripts" / "clip_web_article.py"
|
||||
spec = importlib.util.spec_from_file_location("clip_web_article", SCRIPT_PATH)
|
||||
clip = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(clip)
|
||||
|
||||
|
||||
class ClipWebArticleTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.tmp = Path(tempfile.mkdtemp(prefix="clip-web-article-test-", dir=Path.cwd() / "tmp"))
|
||||
|
||||
def tearDown(self):
|
||||
shutil.rmtree(self.tmp)
|
||||
|
||||
def test_normalize_opencli_article_flattens_images_and_removes_added_metadata(self):
|
||||
source = self.tmp / "opencli"
|
||||
article_dir = source / "BadTitle"
|
||||
images_dir = article_dir / "images"
|
||||
images_dir.mkdir(parents=True)
|
||||
(images_dir / "img_001.jpg").write_bytes(b"image-one")
|
||||
(article_dir / "BadTitle.md").write_text(
|
||||
"\n".join(
|
||||
[
|
||||
"# Bad: Title?/ Example",
|
||||
"",
|
||||
"> 作者: Alice",
|
||||
"> 发布时间: 2026-06-26 10:00:00",
|
||||
"> 原文链接: https://example.test/a",
|
||||
"",
|
||||
"---",
|
||||
"",
|
||||
"First paragraph.",
|
||||
"",
|
||||
"",
|
||||
"",
|
||||
"",
|
||||
"",
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = clip.normalize_opencli_article(source, self.tmp / "out", date(2026, 6, 27))
|
||||
|
||||
self.assertEqual(result.image_count, 1)
|
||||
self.assertTrue(result.output_dir.name.startswith("20260626-"))
|
||||
self.assertNotRegex(result.output_dir.name, r'[<>:"/\\|?*]')
|
||||
markdown_files = list(result.output_dir.glob("*.md"))
|
||||
self.assertEqual(len(markdown_files), 1)
|
||||
md_text = markdown_files[0].read_text(encoding="utf-8")
|
||||
self.assertTrue(md_text.startswith("# Bad: Title?/ Example\n\n"))
|
||||
self.assertNotIn("原文链接", md_text)
|
||||
self.assertNotIn("> 作者:", md_text)
|
||||
self.assertNotIn("> 发布时间:", md_text)
|
||||
self.assertNotIn("\n---\n", md_text)
|
||||
self.assertIn("", md_text)
|
||||
self.assertIn("", md_text)
|
||||
self.assertEqual((result.output_dir / "img_1.jpg").read_bytes(), b"image-one")
|
||||
|
||||
def test_normalize_opencli_article_never_overwrites_existing_folder(self):
|
||||
source = self.tmp / "opencli"
|
||||
article_dir = source / "Title"
|
||||
article_dir.mkdir(parents=True)
|
||||
(article_dir / "Title.md").write_text("# Same Title\n\nBody\n", encoding="utf-8")
|
||||
|
||||
first = clip.normalize_opencli_article(source, self.tmp / "out", date(2026, 6, 27))
|
||||
second = clip.normalize_opencli_article(source, self.tmp / "out", date(2026, 6, 27))
|
||||
|
||||
self.assertNotEqual(first.output_dir, second.output_dir)
|
||||
self.assertTrue(second.output_dir.name.endswith("-2"))
|
||||
self.assertTrue((first.output_dir / "Same Title.md").exists())
|
||||
self.assertTrue((second.output_dir / "Same Title.md").exists())
|
||||
|
||||
def test_save_twitter_rows_keeps_only_target_status_and_downloads_images(self):
|
||||
image_dir = self.tmp / "server"
|
||||
image_dir.mkdir()
|
||||
(image_dir / "photo.jpg").write_bytes(b"tweet-image")
|
||||
|
||||
class QuietHandler(SimpleHTTPRequestHandler):
|
||||
def log_message(self, format, *args):
|
||||
return
|
||||
|
||||
old_cwd = Path.cwd()
|
||||
server = TCPServer(("127.0.0.1", 0), QuietHandler)
|
||||
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
||||
try:
|
||||
import os
|
||||
|
||||
os.chdir(image_dir)
|
||||
thread.start()
|
||||
image_url = f"http://127.0.0.1:{server.server_address[1]}/photo.jpg"
|
||||
rows = [
|
||||
{
|
||||
"id": "1",
|
||||
"author": "alice",
|
||||
"text": "AI agents should preserve original English text.",
|
||||
"created_at": "Wed Jun 24 04:02:06 +0000 2026",
|
||||
"media_urls": [image_url],
|
||||
},
|
||||
{
|
||||
"id": "2",
|
||||
"author": "bob",
|
||||
"text": "@alice This reply should not be saved.",
|
||||
"in_reply_to": "1",
|
||||
"media_urls": [image_url],
|
||||
},
|
||||
]
|
||||
|
||||
result = clip.save_twitter_rows(
|
||||
rows,
|
||||
self.tmp / "out",
|
||||
"https://x.com/alice/status/1",
|
||||
date(2026, 6, 27),
|
||||
)
|
||||
finally:
|
||||
server.shutdown()
|
||||
server.server_close()
|
||||
import os
|
||||
|
||||
os.chdir(old_cwd)
|
||||
|
||||
md_text = result.markdown_path.read_text(encoding="utf-8")
|
||||
self.assertIn("AI agents should preserve original English text.", md_text)
|
||||
self.assertNotIn("This reply should not be saved", md_text)
|
||||
self.assertNotIn("**@alice**", md_text)
|
||||
self.assertNotIn("**@bob**", md_text)
|
||||
self.assertIn("", md_text)
|
||||
self.assertNotIn("https://x.com/alice/status/1", md_text)
|
||||
self.assertEqual((result.output_dir / "img_1.jpg").read_bytes(), b"tweet-image")
|
||||
self.assertEqual(result.image_count, 1)
|
||||
self.assertEqual(result.language, "english")
|
||||
self.assertTrue(result.output_dir.name.startswith("20260624-"))
|
||||
self.assertLessEqual(len(result.output_dir.name), 30)
|
||||
self.assertLessEqual(len(result.markdown_path.stem), 32)
|
||||
|
||||
def test_normalize_zhihu_pin_web_read_keeps_article_body_and_drops_page_chrome(self):
|
||||
source = self.tmp / "opencli"
|
||||
article_dir = source / "Pin"
|
||||
images_dir = article_dir / "images"
|
||||
images_dir.mkdir(parents=True)
|
||||
(images_dir / "img_003.jpg").write_bytes(b"cover")
|
||||
(article_dir / "Pin.md").write_text(
|
||||
"\n".join(
|
||||
[
|
||||
"# OpenAI内部是怎么使用Codex？ | 最近OpenAI公布了其内部是如何使用Codex。为了方便大家阅读，我基于OpenA…",
|
||||
"> 原文链接: https://www.zhihu.com/pin/2042210955053065858",
|
||||
"",
|
||||
"---",
|
||||
"",
|
||||
"[](https://www.zhihu.com/people/yang-chao-62)",
|
||||
"AI生命克劳德",
|
||||
"65 人赞同了该想法",
|
||||
"",
|
||||
"OpenAI内部是怎么使用Codex？ |",
|
||||
"",
|
||||
"最近OpenAI公布了其内部是如何使用Codex。",
|
||||
"",
|
||||
"",
|
||||
"",
|
||||
"+5",
|
||||
"",
|
||||
"[编辑于2026-05-25 20:36](https://www.zhihu.com/pin/2042210955053065858)・IP 属地上海",
|
||||
"2 条评论",
|
||||
"默认",
|
||||
"最新评论不应该保存",
|
||||
]
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
result = clip.normalize_opencli_article(source, self.tmp / "out", date(2026, 6, 27), source_kind="zhihu-pin")
|
||||
|
||||
md_text = result.markdown_path.read_text(encoding="utf-8")
|
||||
self.assertTrue(result.output_dir.name.startswith("20260525-"))
|
||||
self.assertIn("最近OpenAI公布了其内部是如何使用Codex。", md_text)
|
||||
self.assertIn("(img_1.jpg)", md_text)
|
||||
self.assertNotIn("AI生命克劳德", md_text)
|
||||
self.assertNotIn("最新评论不应该保存", md_text)
|
||||
self.assertNotIn("原文链接", md_text)
|
||||
self.assertLessEqual(len(result.output_dir.name), 30)
|
||||
self.assertLessEqual(len(result.markdown_path.stem), 32)
|
||||
|
||||
def test_route_prefers_opencli_without_using_cookie_files(self):
|
||||
zhihu_answer = clip.plan_opencli_invocation(
|
||||
"https://www.zhihu.com/question/1998112125915267964/answer/2052741315209831340",
|
||||
self.tmp / "work",
|
||||
)
|
||||
zhihu_pin = clip.plan_opencli_invocation("https://www.zhihu.com/pin/2042210955053065858", self.tmp / "work")
|
||||
weixin = clip.plan_opencli_invocation("https://mp.weixin.qq.com/s/TApHplXcBONvmVXgDKUUOg", self.tmp / "work")
|
||||
x_url = clip.plan_opencli_invocation("https://x.com/dotey/status/2069632132431929651", self.tmp / "work")
|
||||
|
||||
self.assertEqual(zhihu_answer.kind, "opencli-zhihu-answer-detail")
|
||||
self.assertIn("zhihu", zhihu_answer.command)
|
||||
self.assertIn("answer-detail", zhihu_answer.command)
|
||||
self.assertIn("2052741315209831340", zhihu_answer.command)
|
||||
self.assertEqual(zhihu_pin.kind, "opencli-zhihu-pin-web-read")
|
||||
self.assertEqual(weixin.kind, "opencli-download")
|
||||
self.assertIn("weixin", weixin.command)
|
||||
self.assertEqual(x_url.kind, "opencli-twitter-thread")
|
||||
self.assertIn("twitter", x_url.command)
|
||||
joined = " ".join(zhihu_answer.command + zhihu_pin.command + weixin.command + x_url.command)
|
||||
self.assertNotIn("cookie", joined.lower())
|
||||
|
||||
def test_opencli_plans_release_browser_tabs(self):
|
||||
urls = [
|
||||
"https://www.zhihu.com/question/1998112125915267964/answer/2052741315209831340",
|
||||
"https://www.zhihu.com/pin/2042210955053065858",
|
||||
"https://mp.weixin.qq.com/s/TApHplXcBONvmVXgDKUUOg",
|
||||
"https://x.com/dotey/status/2069632132431929651",
|
||||
"https://example.com/article",
|
||||
]
|
||||
|
||||
for url in urls:
|
||||
with self.subTest(url=url):
|
||||
command = clip.plan_opencli_invocation(url, self.tmp / "work").command
|
||||
self.assertIn("--keep-tab", command)
|
||||
keep_tab_index = command.index("--keep-tab")
|
||||
self.assertEqual(command[keep_tab_index + 1], "false")
|
||||
|
||||
def test_parser_accepts_batch_urls_and_custom_output_root(self):
|
||||
args = clip.build_parser().parse_args(
|
||||
[
|
||||
"https://example.test/one",
|
||||
"https://example.test/two",
|
||||
"--output-root",
|
||||
str(self.tmp / "custom-out"),
|
||||
]
|
||||
)
|
||||
|
||||
self.assertEqual(args.urls, ["https://example.test/one", "https://example.test/two"])
|
||||
self.assertEqual(Path(args.output_root), self.tmp / "custom-out")
|
||||
|
||||
def test_clip_urls_processes_batch_with_distinct_temporary_work_dirs(self):
|
||||
calls = []
|
||||
old_clip_url = clip.clip_url
|
||||
try:
|
||||
def fake_clip_url(url, output_root, work_dir, capture_date=None):
|
||||
calls.append((url, output_root, work_dir))
|
||||
work_dir.mkdir(parents=True, exist_ok=True)
|
||||
md_path = output_root / f"{url[-3:]}.md"
|
||||
return clip.ClipResult(url, output_root / url[-3:], md_path, 0, "english")
|
||||
|
||||
clip.clip_url = fake_clip_url
|
||||
items = clip.clip_urls(
|
||||
["https://example.test/one", "https://example.test/two"],
|
||||
self.tmp / "out",
|
||||
self.tmp / "work",
|
||||
keep_work=False,
|
||||
capture_date=date(2026, 6, 27),
|
||||
)
|
||||
finally:
|
||||
clip.clip_url = old_clip_url
|
||||
|
||||
self.assertEqual([item.url for item in items], ["https://example.test/one", "https://example.test/two"])
|
||||
self.assertTrue(all(item.result is not None and not item.error for item in items))
|
||||
self.assertEqual(len({work_dir for _url, _output_root, work_dir in calls}), 2)
|
||||
self.assertFalse(any(work_dir.exists() for _url, _output_root, work_dir in calls))
|
||||
|
||||
def test_format_opencli_failure_explains_browser_bridge_exit_code(self):
|
||||
reason = clip.format_opencli_failure(
|
||||
69,
|
||||
"Error: Browser Bridge extension not connected\n exitCode: 69\n",
|
||||
"",
|
||||
)
|
||||
|
||||
self.assertIn("Browser Bridge", reason)
|
||||
self.assertIn("not connected", reason)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -38,9 +38,8 @@ C:\Users\wangq\.conda\envs\skills-vault\python.exe -m pip install -e .\skills\vo
|
|||
Install and authenticate MiniMax `mmx`:
|
||||
|
||||
```powershell
|
||||
npm.cmd install -g mmx-cli
|
||||
cmd /c mmx auth login
|
||||
cmd /c mmx auth status
|
||||
npm install -g mmx-cli
|
||||
mmx auth login
|
||||
```
|
||||
|
||||
## Install Skill Wrapper
|
||||
|
|
@ -61,27 +60,18 @@ powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill voic
|
|||
|
||||
Testing is intentionally separate from migration.
|
||||
|
||||
Run source tests from the repository root. Disabling pytest's cache provider avoids writes under a read-only installed Skill directory:
|
||||
Run unit tests only after installation is ready:
|
||||
|
||||
```powershell
|
||||
New-Item -ItemType Directory -Force -Path .\tmp | Out-Null
|
||||
conda run -n skills-vault python -B -m pytest .\skills\voice-generation\tests -p no:cacheprovider --basetemp .\tmp\pytest-voice-generation
|
||||
python -m pytest .\skills\voice-generation\tests --basetemp .\tmp\pytest-voice-generation
|
||||
```
|
||||
|
||||
Validate the Skill structure and installed copy separately:
|
||||
Or without activating the environment:
|
||||
|
||||
```powershell
|
||||
conda run -n skills-vault python -B .\scripts\quick_validate.py .\skills\voice-generation
|
||||
conda run -n skills-vault python -B .\scripts\quick_validate.py C:\Users\wangq\.agents\skills\voice-generation
|
||||
C:\Users\wangq\.conda\envs\skills-vault\python.exe -B -m pytest C:\Users\wangq\.agents\skills\voice-generation\tests -p no:cacheprovider --basetemp .\tmp\pytest-voice-generation-installed
|
||||
New-Item -ItemType Directory -Force -Path .\tmp | Out-Null
|
||||
C:\Users\wangq\.conda\envs\skills-vault\python.exe -m pytest .\skills\voice-generation\tests --basetemp .\tmp\pytest-voice-generation
|
||||
```
|
||||
|
||||
Smoke tests may call real MiniMax services and consume quota. Do not run them unless explicitly requested.
|
||||
|
||||
For a real long-form test, create an isolated project under the repository `tmp` directory, populate `voices.json` and `scripts\`, then run from that project root:
|
||||
|
||||
```powershell
|
||||
conda run -n skills-vault voice-gen gen --voice <voice-name> scripts --out output --format mp3
|
||||
```
|
||||
|
||||
The command sends the extracted narration text to MiniMax. Confirm that external transmission is intended before running it.
|
||||
|
|
|
|||
|
|
@ -69,25 +69,3 @@ conda environment: shared skills-vault
|
|||
editable package installed in shared env: yes
|
||||
old dedicated conda env removed: C:\Users\wangq\miniconda3\envs\voice-gen
|
||||
```
|
||||
|
||||
## Post-Migration Windows Hardening
|
||||
|
||||
Validated on 2026-07-10:
|
||||
|
||||
- Resolve the npm `mmx.CMD` shim and execute it through `cmd /c` on Windows.
|
||||
- Pass narration through a temporary UTF-8 file with `--text-file`; do not pass long text through `--text` or use mmx-cli's Windows-incompatible `--text-file -` path.
|
||||
- Remove the temporary text file after success or failure.
|
||||
- Escape non-ASCII path characters in CLI status lines as `\uXXXX` so `conda run` remains safe on GBK consoles; generated files retain their Unicode names.
|
||||
- Force repository tests to import the source tree instead of an older `.agents` installation.
|
||||
|
||||
Verification evidence:
|
||||
|
||||
```text
|
||||
source tests: 40 passed, 1 skipped
|
||||
installed-copy tests: 40 passed, 1 skipped
|
||||
real long-form voice: BroTsong-2026-06-10
|
||||
real input: 2602 extracted characters / 7030 UTF-8 bytes
|
||||
real output: tmp\voice-generation-e2e-20260710-01\output\01.从想法到作品口播文案.mp3
|
||||
audio: 505.152 seconds, 32000 Hz, 128000 bps, 14032 MPEG frames
|
||||
temporary text files after completion: 0
|
||||
```
|
||||
|
|
|
|||
|
|
@ -28,8 +28,8 @@ powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill voic
|
|||
External requirements:
|
||||
|
||||
- Conda
|
||||
- `mmx` CLI v1.0.16+ (`npm.cmd install -g mmx-cli` on Windows)
|
||||
- `cmd /c mmx auth login` once on Windows
|
||||
- `mmx` CLI v1.0.16+ (`npm install -g mmx-cli`)
|
||||
- `mmx auth login` once
|
||||
|
||||
## Quick Start
|
||||
|
||||
|
|
@ -70,9 +70,7 @@ The CLI wraps MiniMax voice operations:
|
|||
- Uploaded files are registered through the MiniMax voice clone API.
|
||||
- Synthesis uses `mmx speech synthesize`.
|
||||
|
||||
On Windows, `mmx` is a `.cmd` shim from npm, so subprocess calls are routed through `cmd /c` where needed. Synthesis text is written to a short-lived UTF-8 file next to the requested output and passed with `--text-file`; the file is removed after each call. This avoids Windows command-line length and quoting limits. It also avoids mmx-cli's current `--text-file -` stdin implementation, which resolves `/dev/stdin` as a drive-local path on Windows.
|
||||
|
||||
CLI status lines escape non-ASCII path characters as ASCII `\uXXXX` sequences so `conda run` remains safe on GBK Windows consoles. The files themselves retain their original Unicode names.
|
||||
On Windows, `mmx` is a `.cmd` shim from npm, so subprocess calls are routed through `cmd /c` where needed.
|
||||
|
||||
## Project Layout
|
||||
|
||||
|
|
@ -80,6 +78,7 @@ CLI status lines escape non-ASCII path characters as ASCII `\uXXXX` sequences so
|
|||
skills/voice-generation/
|
||||
SKILL.md
|
||||
INSTALL.md
|
||||
environment.yml
|
||||
pyproject.toml
|
||||
src/voice_gen/
|
||||
tests/
|
||||
|
|
@ -97,15 +96,13 @@ output/
|
|||
## Development
|
||||
|
||||
```bash
|
||||
# From the repository root
|
||||
conda run -n skills-vault python -B -m pytest skills/voice-generation/tests -p no:cacheprovider --basetemp tmp/pytest-voice-generation
|
||||
# Inside the skills-vault Conda env
|
||||
python -m pytest
|
||||
|
||||
# Smoke tests may call real MiniMax services and consume quota.
|
||||
conda run -n skills-vault python -B -m pytest skills/voice-generation/tests --run-smoke -p no:cacheprovider --basetemp tmp/pytest-voice-generation-smoke
|
||||
python -m pytest --run-smoke
|
||||
```
|
||||
|
||||
The test configuration forces imports from this repository's `src` tree so a separately installed `.agents/skills` copy cannot make source tests pass accidentally.
|
||||
|
||||
## As An Agentic Skill
|
||||
|
||||
The root `SKILL.md` exposes this tool to Agentic systems. The public installed Skill path is:
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ This Skill operates the local `voice-gen` Python CLI. The CLI wraps MiniMax `mmx
|
|||
|
||||
- Activate the configured Conda environment before running the CLI.
|
||||
- Ensure the `voice-gen` package has been installed in editable mode from the repository source.
|
||||
- Ensure the `mmx` CLI is installed and authenticated. On Windows, verify with `cmd /c mmx auth status` to avoid PowerShell shim policy issues.
|
||||
- Ensure the `mmx` CLI is installed and authenticated with `mmx auth login`.
|
||||
- Run CLI commands from the user's project root unless the user specifies another working directory.
|
||||
|
||||
## Workflow
|
||||
|
|
@ -61,8 +61,6 @@ Use `voice-gen voices pull --force` only when overwriting existing local entries
|
|||
|
||||
- Input scripts are Markdown files matching `*.md`.
|
||||
- Frontmatter, headings, code fences, list markers, bold, italic, and inline code markup are stripped before synthesis.
|
||||
- Synthesis text is passed to mmx through a temporary UTF-8 file, not a command-line `--text` value or `--text-file -`; the temporary file is removed after the call.
|
||||
- Status output escapes non-ASCII path characters as `\uXXXX` sequences for Windows console safety; generated files retain their original Unicode names.
|
||||
- Output files are named `<script-stem>.<format>`.
|
||||
- The local voice registry is `voices.json` in the current working directory.
|
||||
|
||||
|
|
@ -70,7 +68,7 @@ Use `voice-gen voices pull --force` only when overwriting existing local entries
|
|||
|
||||
- If `voice-gen voices list` is empty, ask the user to register or pull voices first.
|
||||
- If a voice is not found, list available voices and ask the user to choose one.
|
||||
- If `mmx` authentication fails on Windows, ask the user to run `cmd /c mmx auth status` and `cmd /c mmx auth login`.
|
||||
- If `mmx` authentication fails, ask the user to run `mmx auth status` and `mmx auth login`.
|
||||
- If quota fails, surface the quota error and suggest waiting for the next quota window or upgrading.
|
||||
- If the request is for music, sound effects, video, image, or non-MiniMax TTS, do not use this Skill.
|
||||
|
||||
|
|
|
|||
|
|
@ -16,11 +16,6 @@ _MMX_CONFIG_PATH = Path.home() / ".mmx" / "config.json"
|
|||
_oauth_token_cache: str | None = None
|
||||
|
||||
|
||||
def _ascii_safe(value: object) -> str:
|
||||
"""Return deterministic ASCII text for status output on Windows consoles."""
|
||||
return str(value).encode("ascii", errors="backslashreplace").decode("ascii")
|
||||
|
||||
|
||||
def _load_oauth_token() -> str:
|
||||
"""Load the OAuth access token from ~/.mmx/config.json (cached)."""
|
||||
global _oauth_token_cache
|
||||
|
|
@ -446,13 +441,10 @@ def cmd_gen(args: argparse.Namespace) -> int:
|
|||
out_path=out_path,
|
||||
format=args.format,
|
||||
)
|
||||
print(f"OK {_ascii_safe(md.name)} -> {_ascii_safe(out_path)}")
|
||||
print(f"OK {md.name} -> {out_path}")
|
||||
successes += 1
|
||||
except Exception as e: # noqa: BLE001 - log and continue
|
||||
print(
|
||||
f"ERROR {_ascii_safe(md.name)}: {_ascii_safe(e)}",
|
||||
file=sys.stderr,
|
||||
)
|
||||
print(f"ERROR {md.name}: {e}", file=sys.stderr)
|
||||
failures += 1
|
||||
print(f"done: {successes} ok, {failures} failed")
|
||||
return 0 if failures == 0 else 1
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@ from __future__ import annotations
|
|||
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from typing import Callable, Optional
|
||||
|
||||
|
|
@ -12,7 +11,7 @@ from .errors import MmxError
|
|||
|
||||
|
||||
def _build_cmd(
|
||||
text_file: Path,
|
||||
text: str,
|
||||
voice_id: int | str,
|
||||
out_path: Path,
|
||||
format: str = "mp3",
|
||||
|
|
@ -26,7 +25,7 @@ def _build_cmd(
|
|||
"synthesize",
|
||||
"--non-interactive",
|
||||
"--quiet",
|
||||
"--text-file", str(text_file),
|
||||
"--text", text,
|
||||
"--voice", str(voice_id),
|
||||
"--format", format,
|
||||
"--out", str(out_path),
|
||||
|
|
@ -80,43 +79,28 @@ def synthesize(
|
|||
"""
|
||||
if run is None:
|
||||
run = _default_run
|
||||
cmd = _build_cmd(
|
||||
text=text,
|
||||
voice_id=voice_id,
|
||||
out_path=out_path,
|
||||
format=format,
|
||||
speed=speed,
|
||||
volume=volume,
|
||||
pitch=pitch,
|
||||
)
|
||||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w",
|
||||
encoding="utf-8",
|
||||
newline="\n",
|
||||
suffix=".txt",
|
||||
prefix=".voice-gen-",
|
||||
dir=out_path.parent,
|
||||
delete=False,
|
||||
) as text_handle:
|
||||
text_handle.write(text)
|
||||
text_path = Path(text_handle.name)
|
||||
|
||||
try:
|
||||
cmd = _build_cmd(
|
||||
text_file=text_path,
|
||||
voice_id=voice_id,
|
||||
out_path=out_path,
|
||||
format=format,
|
||||
speed=speed,
|
||||
volume=volume,
|
||||
pitch=pitch,
|
||||
result = run(cmd)
|
||||
if result.returncode != 0:
|
||||
raise MmxError(
|
||||
exit_code=result.returncode,
|
||||
stderr=result.stderr or "",
|
||||
command=" ".join(cmd),
|
||||
)
|
||||
if not out_path.exists():
|
||||
# mmx sometimes writes hex-decoded content; treat as failure if file absent
|
||||
raise MmxError(
|
||||
exit_code=1,
|
||||
stderr=f"mmx exited 0 but {out_path} was not created",
|
||||
command=" ".join(cmd),
|
||||
)
|
||||
result = run(cmd)
|
||||
if result.returncode != 0:
|
||||
raise MmxError(
|
||||
exit_code=result.returncode,
|
||||
stderr=result.stderr or "",
|
||||
command=" ".join(cmd),
|
||||
)
|
||||
if not out_path.exists():
|
||||
# mmx sometimes writes hex-decoded content; treat as failure if file absent
|
||||
raise MmxError(
|
||||
exit_code=1,
|
||||
stderr=f"mmx exited 0 but {out_path} was not created",
|
||||
command=" ".join(cmd),
|
||||
)
|
||||
finally:
|
||||
text_path.unlink(missing_ok=True)
|
||||
return out_path
|
||||
|
|
|
|||
|
|
@ -1,14 +1,7 @@
|
|||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
SOURCE_ROOT = Path(__file__).resolve().parents[1] / "src"
|
||||
sys.path.insert(0, str(SOURCE_ROOT))
|
||||
|
||||
|
||||
def pytest_addoption(parser):
|
||||
parser.addoption("--run-smoke", action="store_true", default=False,
|
||||
help="Run real-mmx smoke tests (consumes quota)")
|
||||
|
|
|
|||
|
|
@ -164,41 +164,6 @@ def test_gen_batch_end_to_end(tmp_path, monkeypatch):
|
|||
assert cmd[i + 1] == "anchor"
|
||||
|
||||
|
||||
def test_gen_status_output_is_ascii_safe(tmp_path, monkeypatch):
|
||||
import io
|
||||
import sys
|
||||
|
||||
monkeypatch.chdir(tmp_path)
|
||||
assert main(["init"]) == 0
|
||||
(tmp_path / "scripts" / "中文口播.md").write_text("测试。", encoding="utf-8")
|
||||
(tmp_path / "voices.json").write_text(json.dumps({
|
||||
"voices": {
|
||||
"anchor": {
|
||||
"file_id": 42,
|
||||
"filename": "a.mp3",
|
||||
"created_at": "2026-07-10T00:00:00Z",
|
||||
}
|
||||
}
|
||||
}), encoding="utf-8")
|
||||
|
||||
from voice_gen import synthesizer
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
out = Path(cmd[cmd.index("--out") + 1])
|
||||
out.write_bytes(b"audio")
|
||||
return type("R", (), {"returncode": 0, "stderr": ""})()
|
||||
|
||||
monkeypatch.setattr(synthesizer, "_default_run", fake_run)
|
||||
buffer = io.BytesIO()
|
||||
ascii_stdout = io.TextIOWrapper(buffer, encoding="ascii", errors="strict")
|
||||
monkeypatch.setattr(sys, "stdout", ascii_stdout)
|
||||
|
||||
assert main(["gen", "--voice", "anchor", "scripts"]) == 0
|
||||
ascii_stdout.flush()
|
||||
output = buffer.getvalue().decode("ascii")
|
||||
assert "\\u4e2d\\u6587\\u53e3\\u64ad.md" in output
|
||||
|
||||
|
||||
def test_voices_add_rejects_duplicate_local_name(tmp_path, monkeypatch):
|
||||
"""Re-adding a file whose stem matches an existing local entry errors out."""
|
||||
monkeypatch.chdir(tmp_path)
|
||||
|
|
|
|||
|
|
@ -1,13 +0,0 @@
|
|||
from pathlib import Path
|
||||
|
||||
import voice_gen
|
||||
|
||||
|
||||
def test_tests_import_repository_source():
|
||||
expected = (
|
||||
Path(__file__).resolve().parents[1]
|
||||
/ "src"
|
||||
/ "voice_gen"
|
||||
/ "__init__.py"
|
||||
)
|
||||
assert Path(voice_gen.__file__).resolve() == expected
|
||||
|
|
@ -8,14 +8,13 @@ from voice_gen.errors import MmxError
|
|||
|
||||
def test_build_cmd_minimum():
|
||||
cmd = _build_cmd(
|
||||
text_file=Path("input.txt"),
|
||||
text="hi",
|
||||
voice_id=123,
|
||||
out_path=Path("out.mp3"),
|
||||
format="mp3",
|
||||
)
|
||||
assert cmd[:5] == ["mmx", "speech", "synthesize", "--non-interactive", "--quiet"]
|
||||
assert "--text-file" in cmd and "input.txt" in cmd
|
||||
assert "--text" not in cmd
|
||||
assert "--text" in cmd and "hi" in cmd
|
||||
assert "--voice" in cmd and "123" in cmd
|
||||
assert "--out" in cmd and "out.mp3" in cmd
|
||||
assert "--format" in cmd and "mp3" in cmd
|
||||
|
|
@ -46,32 +45,6 @@ def test_synthesize_invokes_mmx_and_returns_path(tmp_path):
|
|||
assert captured["cmd"][captured["cmd"].index("--voice") + 1] == "999"
|
||||
|
||||
|
||||
def test_synthesize_passes_long_utf8_text_via_temporary_file(tmp_path):
|
||||
text = ("这是一段用于验证 Windows 长命令行安全性的中文口播稿。\n" * 500)[:9000]
|
||||
captured = {}
|
||||
|
||||
def fake_run(cmd, **kwargs):
|
||||
captured["cmd"] = cmd
|
||||
assert "--text" not in cmd
|
||||
text_path = Path(cmd[cmd.index("--text-file") + 1])
|
||||
captured["text_path"] = text_path
|
||||
captured["text"] = text_path.read_text(encoding="utf-8")
|
||||
out = Path(cmd[cmd.index("--out") + 1])
|
||||
out.write_bytes(b"fake-audio")
|
||||
return type("R", (), {"returncode": 0, "stderr": ""})()
|
||||
|
||||
out = synthesize(
|
||||
text=text,
|
||||
voice_id="test-voice",
|
||||
out_path=tmp_path / "long.mp3",
|
||||
run=fake_run,
|
||||
)
|
||||
|
||||
assert out.exists()
|
||||
assert captured["text"] == text
|
||||
assert not captured["text_path"].exists()
|
||||
|
||||
|
||||
def test_synthesize_raises_mmxerror_on_nonzero(tmp_path):
|
||||
def fake_run(cmd, **kwargs):
|
||||
return type("R", (), {"returncode": 3, "stderr": "auth failed"})()
|
||||
|
|
|
|||
Loading…
Reference in New Issue