672 lines
23 KiB
Python
672 lines
23 KiB
Python
#!/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())
|