Compare commits
1 Commits
main
...
codex/clip
| Author | SHA1 | Date |
|---|---|---|
|
|
4d37c93d87 |
|
|
@ -22,6 +22,7 @@ PowerShell:
|
||||||
|
|
||||||
```powershell
|
```powershell
|
||||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill bundle-zip
|
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 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 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 regression-validation-gate-runner
|
||||||
|
|
@ -34,6 +35,7 @@ Use `-Force` to back up and replace an existing installed copy:
|
||||||
|
|
||||||
```powershell
|
```powershell
|
||||||
powershell -ExecutionPolicy Bypass -File .\scripts\install-skill.ps1 -Skill bundle-zip -Force
|
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 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 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 regression-validation-gate-runner -Force
|
||||||
|
|
|
||||||
|
|
@ -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 |
|
| 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 |
|
| `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 |
|
| `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 |
|
| `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 |
|
| `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 |
|
||||||
|
|
|
||||||
|
|
@ -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()
|
||||||
Loading…
Reference in New Issue