Release standalone PPT template workflow 0.5.0
This commit is contained in:
@@ -0,0 +1,150 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PPT Master - Source Converter Batch Helpers
|
||||
|
||||
Share explicit multi-file and directory expansion logic across source_to_md
|
||||
backend converters.
|
||||
|
||||
Usage:
|
||||
Imported by scripts/source_to_md/*_to_md.py
|
||||
|
||||
Examples:
|
||||
run_path_batch(["docs"], {".docx"}, None, convert_one)
|
||||
|
||||
Dependencies:
|
||||
None
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ConvertOne = Callable[[Path, Path], bool]
|
||||
IsSupportedFile = Callable[[Path], bool]
|
||||
|
||||
|
||||
def _print_status(message: str) -> None:
|
||||
print(message, file=sys.stderr)
|
||||
|
||||
|
||||
def expand_directory_inputs(
|
||||
inputs: list[str],
|
||||
is_supported_file: IsSupportedFile,
|
||||
is_external_ref: Callable[[str], bool] | None = None,
|
||||
) -> tuple[list[str], list[str], bool]:
|
||||
"""Expand non-recursive directory inputs while preserving other items."""
|
||||
expanded: list[str] = []
|
||||
errors: list[str] = []
|
||||
saw_directory = False
|
||||
is_external = is_external_ref or (lambda _item: False)
|
||||
|
||||
for item in inputs:
|
||||
if is_external(item):
|
||||
expanded.append(item)
|
||||
continue
|
||||
path = Path(item)
|
||||
if path.is_dir():
|
||||
saw_directory = True
|
||||
matches = sorted(
|
||||
child for child in path.iterdir()
|
||||
if child.is_file() and is_supported_file(child)
|
||||
)
|
||||
if matches:
|
||||
expanded.extend(str(match) for match in matches)
|
||||
else:
|
||||
errors.append(f"{item}: no supported files found")
|
||||
continue
|
||||
expanded.append(item)
|
||||
|
||||
return expanded, errors, saw_directory
|
||||
|
||||
|
||||
def _output_key(path: Path) -> Path:
|
||||
return path.resolve(strict=False)
|
||||
|
||||
|
||||
def unique_output_path(output_dir: Path, stem: str, used_outputs: set[Path]) -> Path:
|
||||
"""Return an in-run unique Markdown path without consulting the filesystem."""
|
||||
base = stem or "output"
|
||||
candidate = output_dir / f"{base}.md"
|
||||
suffix = 2
|
||||
while _output_key(candidate) in used_outputs:
|
||||
candidate = output_dir / f"{base}_{suffix}.md"
|
||||
suffix += 1
|
||||
used_outputs.add(_output_key(candidate))
|
||||
return candidate
|
||||
|
||||
|
||||
def _output_for(
|
||||
source: Path,
|
||||
output_arg: str | None,
|
||||
batch_mode: bool,
|
||||
used_outputs: set[Path],
|
||||
) -> Path:
|
||||
if output_arg and batch_mode:
|
||||
return unique_output_path(Path(output_arg), source.stem, used_outputs)
|
||||
if output_arg:
|
||||
return Path(output_arg)
|
||||
return source.with_suffix(".md")
|
||||
|
||||
|
||||
def run_path_batch(
|
||||
inputs: list[str],
|
||||
supported_suffixes: set[str],
|
||||
output_arg: str | None,
|
||||
convert_one: ConvertOne,
|
||||
) -> int:
|
||||
"""Run one converter across explicit files and non-recursive directories."""
|
||||
expanded, expansion_errors, saw_directory = expand_directory_inputs(
|
||||
inputs,
|
||||
lambda path: path.suffix.lower() in supported_suffixes,
|
||||
)
|
||||
sources = [Path(item) for item in expanded]
|
||||
batch_mode = saw_directory or len(sources) > 1
|
||||
|
||||
if output_arg and batch_mode:
|
||||
output_dir = Path(output_arg)
|
||||
if output_dir.exists() and not output_dir.is_dir():
|
||||
_print_status(f"[ERROR] Batch output path is not a directory: {output_arg}")
|
||||
return 1
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
success_count = 0
|
||||
failed: list[str] = []
|
||||
skipped: list[str] = list(expansion_errors)
|
||||
used_outputs: set[Path] = set()
|
||||
|
||||
for source in sources:
|
||||
output = _output_for(source, output_arg, batch_mode, used_outputs)
|
||||
if batch_mode:
|
||||
_print_status(f"\n==> {source}")
|
||||
try:
|
||||
converted = convert_one(source, output)
|
||||
sys.stdout.flush()
|
||||
if converted:
|
||||
success_count += 1
|
||||
else:
|
||||
failed.append(str(source))
|
||||
except Exception as exc:
|
||||
sys.stdout.flush()
|
||||
failed.append(f"{source}: {exc}")
|
||||
_print_status(f"[ERROR] {source}: {exc}")
|
||||
|
||||
if batch_mode:
|
||||
sys.stdout.flush()
|
||||
_print_status(f"\n[Done] Success: {success_count}/{len(sources)}, Failed: {len(failed)}")
|
||||
if skipped:
|
||||
_print_status("\n[Skipped directories]:")
|
||||
for item in skipped:
|
||||
_print_status(f" - {item}")
|
||||
if failed:
|
||||
_print_status("\n[Failed inputs]:")
|
||||
for item in failed:
|
||||
_print_status(f" - {item}")
|
||||
|
||||
if not sources:
|
||||
return 1
|
||||
return 0 if not failed and not skipped else 1
|
||||
@@ -0,0 +1,235 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PPT Master - Markdown Conversion Profile Helpers
|
||||
|
||||
Write lightweight sidecar metadata for source_to_md conversion outputs.
|
||||
|
||||
Usage:
|
||||
Imported by scripts/source_to_md/*.py
|
||||
|
||||
Examples:
|
||||
write_conversion_profile(input_path="demo.pdf", markdown_path="demo.md", ...)
|
||||
|
||||
Dependencies:
|
||||
None
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
IMAGE_MANIFEST_NAME = "image_manifest.json"
|
||||
PROFILE_SCHEMA = "ppt-master.source_to_md.profile.v1"
|
||||
PROFILE_SUFFIX = ".conversion_profile.json"
|
||||
|
||||
|
||||
def default_asset_dir(markdown_path: Path) -> Path:
|
||||
"""Return the conventional companion asset directory for one Markdown output."""
|
||||
return markdown_path.parent / f"{markdown_path.stem}_files"
|
||||
|
||||
|
||||
def profile_path_for(markdown_path: Path) -> Path:
|
||||
"""Return the sidecar profile path for one Markdown output."""
|
||||
return markdown_path.with_name(f"{markdown_path.stem}{PROFILE_SUFFIX}")
|
||||
|
||||
|
||||
def _display_path(path: Path | None, root: Path) -> str:
|
||||
if path is None:
|
||||
return ""
|
||||
try:
|
||||
return path.resolve().relative_to(root.resolve()).as_posix()
|
||||
except ValueError:
|
||||
return path.resolve().as_posix()
|
||||
|
||||
|
||||
def _count_tables(lines: list[str]) -> int:
|
||||
count = 0
|
||||
in_table = False
|
||||
separator_re = re.compile(
|
||||
r"^\|?\s*:?-{3,}:?\s*(\|\s*:?-{3,}:?\s*)+\|?$"
|
||||
)
|
||||
for line in lines:
|
||||
stripped = line.strip()
|
||||
is_table_line = stripped.startswith("|") and stripped.endswith("|")
|
||||
has_separator = bool(separator_re.match(stripped))
|
||||
if is_table_line or has_separator:
|
||||
if not in_table:
|
||||
count += 1
|
||||
in_table = True
|
||||
continue
|
||||
in_table = False
|
||||
return count
|
||||
|
||||
|
||||
def markdown_stats(markdown_path: Path) -> dict[str, int]:
|
||||
"""Return low-cost Markdown structure counts for inspection/debugging."""
|
||||
if not markdown_path.is_file():
|
||||
return {
|
||||
"line_count": 0,
|
||||
"char_count": 0,
|
||||
"heading_count": 0,
|
||||
"table_count": 0,
|
||||
"image_ref_count": 0,
|
||||
"link_count": 0,
|
||||
}
|
||||
|
||||
text = markdown_path.read_text(encoding="utf-8", errors="replace")
|
||||
lines = text.splitlines()
|
||||
return {
|
||||
"line_count": len(lines),
|
||||
"char_count": len(text),
|
||||
"heading_count": sum(1 for line in lines if re.match(r"^#{1,6}\s+", line)),
|
||||
"table_count": _count_tables(lines),
|
||||
"image_ref_count": len(re.findall(r"!\[[^\]]*\]\([^)]+\)", text)),
|
||||
"link_count": len(re.findall(r"(?<!!)\[[^\]]+\]\([^)]+\)", text)),
|
||||
}
|
||||
|
||||
|
||||
def _read_json(path: Path) -> Any:
|
||||
try:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return None
|
||||
|
||||
|
||||
def _image_count_from_manifest(path: Path) -> int:
|
||||
payload = _read_json(path)
|
||||
if isinstance(payload, list):
|
||||
return len(payload)
|
||||
if isinstance(payload, dict):
|
||||
items = payload.get("items")
|
||||
if isinstance(items, list):
|
||||
return len(items)
|
||||
return 0
|
||||
|
||||
|
||||
def build_conversion_profile(
|
||||
*,
|
||||
input_path: str,
|
||||
markdown_path: str | Path,
|
||||
converter: str,
|
||||
conversion_type: str,
|
||||
asset_dir: str | Path | None = None,
|
||||
warnings: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Build a sidecar profile without changing the Markdown conversion result."""
|
||||
markdown = Path(markdown_path)
|
||||
root = markdown.parent
|
||||
is_url = input_path.startswith(("http://", "https://"))
|
||||
source = None if is_url else Path(input_path)
|
||||
assets = Path(asset_dir) if asset_dir else default_asset_dir(markdown)
|
||||
image_manifest = assets / IMAGE_MANIFEST_NAME
|
||||
source_exists = bool(source and source.exists())
|
||||
|
||||
return {
|
||||
"schema": PROFILE_SCHEMA,
|
||||
"converter": converter,
|
||||
"conversion_type": conversion_type,
|
||||
"source": {
|
||||
"path": input_path if is_url else _display_path(source, root),
|
||||
"name": input_path if is_url else (source.name if source else input_path),
|
||||
"suffix": "" if is_url else (source.suffix.lower() if source else ""),
|
||||
"kind": "url" if is_url else "file",
|
||||
"exists": source_exists,
|
||||
"size_bytes": (
|
||||
source.stat().st_size
|
||||
if source_exists and source is not None and source.is_file()
|
||||
else None
|
||||
),
|
||||
},
|
||||
"outputs": {
|
||||
"markdown": _display_path(markdown, root),
|
||||
"asset_dir": _display_path(assets, root) if assets.is_dir() else "",
|
||||
"image_manifest": (
|
||||
_display_path(image_manifest, root) if image_manifest.is_file() else ""
|
||||
),
|
||||
"image_count": _image_count_from_manifest(image_manifest),
|
||||
},
|
||||
"markdown": markdown_stats(markdown),
|
||||
"warnings": warnings or [],
|
||||
}
|
||||
|
||||
|
||||
def write_conversion_profile(
|
||||
*,
|
||||
input_path: str,
|
||||
markdown_path: str | Path,
|
||||
converter: str,
|
||||
conversion_type: str,
|
||||
asset_dir: str | Path | None = None,
|
||||
warnings: list[str] | None = None,
|
||||
) -> Path:
|
||||
"""Write `<stem>.conversion_profile.json` beside one Markdown output."""
|
||||
markdown = Path(markdown_path)
|
||||
profile_path = profile_path_for(markdown)
|
||||
profile = build_conversion_profile(
|
||||
input_path=input_path,
|
||||
markdown_path=markdown,
|
||||
converter=converter,
|
||||
conversion_type=conversion_type,
|
||||
asset_dir=asset_dir,
|
||||
warnings=warnings,
|
||||
)
|
||||
profile_path.write_text(
|
||||
json.dumps(profile, ensure_ascii=False, indent=2) + "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
return profile_path
|
||||
|
||||
|
||||
def write_conversion_profile_best_effort(
|
||||
*,
|
||||
input_path: str,
|
||||
markdown_path: str | Path,
|
||||
converter: str,
|
||||
conversion_type: str,
|
||||
asset_dir: str | Path | None = None,
|
||||
warnings: list[str] | None = None,
|
||||
) -> Path | None:
|
||||
"""Write a profile sidecar, warning without changing converter success."""
|
||||
try:
|
||||
return write_conversion_profile(
|
||||
input_path=input_path,
|
||||
markdown_path=markdown_path,
|
||||
converter=converter,
|
||||
conversion_type=conversion_type,
|
||||
asset_dir=asset_dir,
|
||||
warnings=warnings,
|
||||
)
|
||||
except OSError as exc:
|
||||
print(f"[WARN] Could not write conversion profile: {exc}", file=sys.stderr)
|
||||
return None
|
||||
|
||||
|
||||
def build_result_payload(
|
||||
*,
|
||||
input_path: str,
|
||||
markdown_path: str | Path,
|
||||
converter: str,
|
||||
conversion_type: str,
|
||||
asset_dir: str | Path | None = None,
|
||||
profile_path: str | Path | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Return a compact JSON payload for CLI consumers."""
|
||||
markdown = Path(markdown_path)
|
||||
assets = Path(asset_dir) if asset_dir else default_asset_dir(markdown)
|
||||
image_manifest = assets / IMAGE_MANIFEST_NAME
|
||||
profile = Path(profile_path) if profile_path else profile_path_for(markdown)
|
||||
return {
|
||||
"input": (
|
||||
input_path
|
||||
if input_path.startswith(("http://", "https://"))
|
||||
else str(Path(input_path).resolve())
|
||||
),
|
||||
"markdown": str(markdown.resolve()),
|
||||
"asset_dir": str(assets.resolve()) if assets.is_dir() else "",
|
||||
"image_manifest": str(image_manifest.resolve()) if image_manifest.is_file() else "",
|
||||
"conversion_profile": str(profile.resolve()) if profile.is_file() else "",
|
||||
"converter": converter,
|
||||
"conversion_type": conversion_type,
|
||||
}
|
||||
@@ -0,0 +1,187 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
PPT Master - Source to Markdown Dispatcher
|
||||
|
||||
Shared routing and backend command construction for source-to-Markdown tools.
|
||||
|
||||
Usage:
|
||||
Imported by scripts/source_to_md.py and scripts/project_manager.py
|
||||
|
||||
Examples:
|
||||
build_conversion_command("report.pdf", "report.md")
|
||||
|
||||
Dependencies:
|
||||
None
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
|
||||
|
||||
SOURCE_TO_MD_DIR = Path(__file__).resolve().parent
|
||||
WECHAT_HOST_KEYWORDS = ("mp.weixin.qq.com", "weixin.qq.com")
|
||||
|
||||
DOC_SUFFIXES = {
|
||||
".docx", ".doc", ".odt", ".rtf", # Office documents
|
||||
".epub", # eBooks
|
||||
".html", ".htm", # Web pages
|
||||
".tex", ".latex", ".rst", ".org", # Academic / technical
|
||||
".ipynb", ".typ", # Notebooks / Typst
|
||||
}
|
||||
EXCEL_SUFFIXES = {".xlsx", ".xlsm"}
|
||||
LEGACY_EXCEL_SUFFIXES = {".xls"}
|
||||
MARKDOWN_SUFFIXES = {".md", ".markdown"}
|
||||
PDF_SUFFIXES = {".pdf"}
|
||||
PRESENTATION_SUFFIXES = {".pptx", ".pptm", ".ppsx", ".ppsm", ".potx", ".potm"}
|
||||
TEXT_SUFFIXES = {".txt", ".text"}
|
||||
|
||||
|
||||
@dataclass
|
||||
class ConversionCommand:
|
||||
"""Backend command plus metadata for one conversion route."""
|
||||
|
||||
command: list[str]
|
||||
script_name: str
|
||||
conversion_type: str
|
||||
output_path: Path | None
|
||||
|
||||
|
||||
def is_url(value: str) -> bool:
|
||||
"""Return whether a string looks like an HTTP(S) URL."""
|
||||
parsed = urlparse(value)
|
||||
return parsed.scheme in {"http", "https"} and bool(parsed.netloc)
|
||||
|
||||
|
||||
def detect_source_type(input_arg: str) -> str:
|
||||
"""Detect a conversion type from a URL, file, or directory."""
|
||||
if is_url(input_arg):
|
||||
return "web"
|
||||
|
||||
path = Path(input_arg)
|
||||
if not path.exists():
|
||||
return "unknown"
|
||||
if path.is_dir():
|
||||
return "directory"
|
||||
|
||||
suffix = path.suffix.lower()
|
||||
if suffix in PDF_SUFFIXES:
|
||||
return "pdf"
|
||||
if suffix in DOC_SUFFIXES:
|
||||
return "doc"
|
||||
if suffix in EXCEL_SUFFIXES or suffix in LEGACY_EXCEL_SUFFIXES:
|
||||
return "excel"
|
||||
if suffix in PRESENTATION_SUFFIXES:
|
||||
return "pptx"
|
||||
if suffix in MARKDOWN_SUFFIXES:
|
||||
return "markdown"
|
||||
if suffix in TEXT_SUFFIXES:
|
||||
return "text"
|
||||
return "unknown"
|
||||
|
||||
|
||||
def default_markdown_path(input_arg: str) -> Path:
|
||||
"""Return the conventional Markdown output path for a local input."""
|
||||
path = Path(input_arg)
|
||||
return path.parent / f"{path.stem}.md"
|
||||
|
||||
|
||||
def _curl_cffi_available() -> bool:
|
||||
"""Return whether curl_cffi is importable."""
|
||||
try:
|
||||
import curl_cffi # noqa: F401
|
||||
return True
|
||||
except ImportError:
|
||||
return False
|
||||
|
||||
|
||||
def _web_script_command(
|
||||
url: str,
|
||||
output_path: Path | None,
|
||||
python_executable: str,
|
||||
allow_node_fallback: bool,
|
||||
) -> tuple[str, list[str]]:
|
||||
"""Return the web backend script name and base command."""
|
||||
host = urlparse(url).netloc.lower()
|
||||
is_tls_sensitive = any(keyword in host for keyword in WECHAT_HOST_KEYWORDS)
|
||||
node_backend = SOURCE_TO_MD_DIR / "web_to_md.cjs"
|
||||
|
||||
if (
|
||||
allow_node_fallback
|
||||
and is_tls_sensitive
|
||||
and not _curl_cffi_available()
|
||||
and node_backend.is_file()
|
||||
and shutil.which("node")
|
||||
):
|
||||
command = ["node", str(node_backend), url]
|
||||
script_name = "web_to_md.cjs"
|
||||
else:
|
||||
command = [python_executable, str(SOURCE_TO_MD_DIR / "web_to_md.py"), url]
|
||||
script_name = "web_to_md.py"
|
||||
|
||||
if output_path is not None:
|
||||
command.extend(["-o", str(output_path)])
|
||||
return script_name, command
|
||||
|
||||
|
||||
def build_conversion_command(
|
||||
input_arg: str,
|
||||
output_path: str | Path | None,
|
||||
*,
|
||||
forced_type: str | None = None,
|
||||
extra_args: list[str] | None = None,
|
||||
pdf_image_mode: str | None = None,
|
||||
render_vector_figures: bool = False,
|
||||
python_executable: str | None = None,
|
||||
allow_node_web_fallback: bool = True,
|
||||
) -> ConversionCommand:
|
||||
"""Build the backend CLI command for one source-to-Markdown conversion."""
|
||||
conversion_type = forced_type or detect_source_type(input_arg)
|
||||
output = Path(output_path) if output_path is not None else None
|
||||
extra = extra_args or []
|
||||
python = python_executable or sys.executable
|
||||
|
||||
if conversion_type == "web":
|
||||
if not is_url(input_arg):
|
||||
raise ValueError("web conversion requires an http:// or https:// URL")
|
||||
script_name, command = _web_script_command(
|
||||
input_arg,
|
||||
output,
|
||||
python,
|
||||
allow_node_web_fallback,
|
||||
)
|
||||
command.extend(extra)
|
||||
return ConversionCommand(command, script_name, conversion_type, output)
|
||||
|
||||
if conversion_type in {"markdown", "text", "directory", "unknown"}:
|
||||
raise ValueError(f"conversion type {conversion_type!r} has no backend command")
|
||||
|
||||
script_by_type = {
|
||||
"pdf": "pdf_to_md.py",
|
||||
"doc": "doc_to_md.py",
|
||||
"excel": "excel_to_md.py",
|
||||
"pptx": "ppt_to_md.py",
|
||||
}
|
||||
script_name = script_by_type.get(conversion_type)
|
||||
if script_name is None:
|
||||
raise ValueError(f"unsupported conversion type: {conversion_type}")
|
||||
if output is None:
|
||||
output = default_markdown_path(input_arg)
|
||||
|
||||
command = [
|
||||
python,
|
||||
str(SOURCE_TO_MD_DIR / script_name),
|
||||
input_arg,
|
||||
"-o",
|
||||
str(output),
|
||||
]
|
||||
if conversion_type == "pdf" and pdf_image_mode:
|
||||
command.extend(["--images", pdf_image_mode])
|
||||
if conversion_type == "pdf" and render_vector_figures:
|
||||
command.append("--render-vector-figures")
|
||||
command.extend(extra)
|
||||
return ConversionCommand(command, script_name, conversion_type, output)
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,410 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Excel to Markdown Converter
|
||||
|
||||
Supported formats:
|
||||
.xlsx Excel workbook
|
||||
.xlsm Excel macro-enabled workbook
|
||||
|
||||
Unsupported by default:
|
||||
.xls Legacy binary Excel format; resave as .xlsx first
|
||||
|
||||
All paths produce the same output convention:
|
||||
<input>.md Markdown file
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sys
|
||||
from datetime import date, datetime, time
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
_SCRIPTS_DIR = Path(__file__).resolve().parents[1]
|
||||
if str(_SCRIPTS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(_SCRIPTS_DIR))
|
||||
|
||||
from console_encoding import configure_utf8_stdio # noqa: E402
|
||||
from _batch import run_path_batch # noqa: E402
|
||||
from _conversion_profile import write_conversion_profile_best_effort # noqa: E402
|
||||
|
||||
configure_utf8_stdio()
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# Format registry
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
EXCEL_FORMATS = {".xlsx", ".xlsm"}
|
||||
LEGACY_EXCEL_FORMATS = {".xls"}
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# Shared helpers
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _format_size(size: int) -> str:
|
||||
for unit in ("B", "KB", "MB"):
|
||||
if size < 1024:
|
||||
return f"{size:.0f} {unit}"
|
||||
size /= 1024
|
||||
return f"{size:.1f} GB"
|
||||
|
||||
|
||||
def _report_result(out_file: Path) -> None:
|
||||
size = out_file.stat().st_size
|
||||
print(f"[OK] Saved Markdown to: {out_file} ({_format_size(size)})")
|
||||
|
||||
|
||||
def _is_empty(value: Any) -> bool:
|
||||
return value is None or (isinstance(value, str) and value.strip() == "")
|
||||
|
||||
|
||||
def _markdown_escape(value: str) -> str:
|
||||
value = value.replace("\\", "\\\\")
|
||||
value = value.replace("|", "\\|")
|
||||
value = value.replace("\r\n", "\n").replace("\r", "\n")
|
||||
return re.sub(r"\s*\n\s*", "<br>", value).strip()
|
||||
|
||||
|
||||
def _format_cell_value(value: Any) -> str:
|
||||
if _is_empty(value):
|
||||
return ""
|
||||
if isinstance(value, bool):
|
||||
return "TRUE" if value else "FALSE"
|
||||
if isinstance(value, datetime):
|
||||
return value.isoformat(sep=" ", timespec="seconds")
|
||||
if isinstance(value, date):
|
||||
return value.isoformat()
|
||||
if isinstance(value, time):
|
||||
return value.isoformat(timespec="seconds")
|
||||
if isinstance(value, float):
|
||||
return _markdown_escape(f"{value:g}")
|
||||
return _markdown_escape(str(value))
|
||||
|
||||
|
||||
def _is_numeric_value(value: Any) -> bool:
|
||||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||||
|
||||
|
||||
def _sheet_state_label(sheet_state: str) -> str:
|
||||
if sheet_state == "visible":
|
||||
return "visible"
|
||||
return sheet_state or "unknown"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# Worksheet extraction
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _merged_value_map(worksheet) -> dict[tuple[int, int], Any]:
|
||||
"""Return propagated values for merged cells, keyed by (row, column).
|
||||
|
||||
Merged regions whose top-left cell is empty are intentionally skipped.
|
||||
These are typically formatting-only ranges that carry no textual content.
|
||||
"""
|
||||
merged_values: dict[tuple[int, int], Any] = {}
|
||||
for merged_range in worksheet.merged_cells.ranges:
|
||||
value = worksheet.cell(merged_range.min_row, merged_range.min_col).value
|
||||
if _is_empty(value):
|
||||
continue
|
||||
for row in range(merged_range.min_row, merged_range.max_row + 1):
|
||||
for col in range(merged_range.min_col, merged_range.max_col + 1):
|
||||
merged_values[(row, col)] = value
|
||||
return merged_values
|
||||
|
||||
|
||||
def _cell_value(worksheet, row: int, col: int, merged_values: dict[tuple[int, int], Any]) -> Any:
|
||||
value = worksheet.cell(row, col).value
|
||||
if _is_empty(value):
|
||||
return merged_values.get((row, col), value)
|
||||
return value
|
||||
|
||||
|
||||
def _content_bounds(worksheet, merged_values: dict[tuple[int, int], Any]) -> tuple[int, int, int, int] | None:
|
||||
min_row = min_col = None
|
||||
max_row = max_col = None
|
||||
|
||||
for row in worksheet.iter_rows():
|
||||
for cell in row:
|
||||
if _is_empty(cell.value):
|
||||
continue
|
||||
min_row = cell.row if min_row is None else min(min_row, cell.row)
|
||||
max_row = cell.row if max_row is None else max(max_row, cell.row)
|
||||
min_col = cell.column if min_col is None else min(min_col, cell.column)
|
||||
max_col = cell.column if max_col is None else max(max_col, cell.column)
|
||||
|
||||
for row, col in merged_values:
|
||||
min_row = row if min_row is None else min(min_row, row)
|
||||
max_row = row if max_row is None else max(max_row, row)
|
||||
min_col = col if min_col is None else min(min_col, col)
|
||||
max_col = col if max_col is None else max(max_col, col)
|
||||
|
||||
if min_row is None or min_col is None or max_row is None or max_col is None:
|
||||
return None
|
||||
return min_row, min_col, max_row, max_col
|
||||
|
||||
|
||||
def _trim_trailing_empty_cells(row: list[Any]) -> list[Any]:
|
||||
trimmed = list(row)
|
||||
while trimmed and _is_empty(trimmed[-1]):
|
||||
trimmed.pop()
|
||||
return trimmed
|
||||
|
||||
|
||||
def _extract_rows(
|
||||
worksheet,
|
||||
bounds: tuple[int, int, int, int],
|
||||
merged_values: dict[tuple[int, int], Any],
|
||||
max_rows: int,
|
||||
max_cols: int,
|
||||
) -> tuple[list[list[Any]], bool, bool]:
|
||||
min_row, min_col, max_row, max_col = bounds
|
||||
|
||||
row_limit = max_row
|
||||
col_limit = max_col
|
||||
rows_truncated = False
|
||||
cols_truncated = False
|
||||
|
||||
if max_rows > 0 and (max_row - min_row + 1) > max_rows:
|
||||
row_limit = min_row + max_rows - 1
|
||||
rows_truncated = True
|
||||
if max_cols > 0 and (max_col - min_col + 1) > max_cols:
|
||||
col_limit = min_col + max_cols - 1
|
||||
cols_truncated = True
|
||||
|
||||
rows: list[list[Any]] = []
|
||||
width = 0
|
||||
for row_index in range(min_row, row_limit + 1):
|
||||
row = [
|
||||
_cell_value(worksheet, row_index, col_index, merged_values)
|
||||
for col_index in range(min_col, col_limit + 1)
|
||||
]
|
||||
row = _trim_trailing_empty_cells(row)
|
||||
width = max(width, len(row))
|
||||
rows.append(row)
|
||||
|
||||
if width == 0:
|
||||
return [], rows_truncated, cols_truncated
|
||||
|
||||
normalized_rows = [row + [""] * (width - len(row)) for row in rows]
|
||||
return normalized_rows, rows_truncated, cols_truncated
|
||||
|
||||
|
||||
def _column_alignments(rows: list[list[Any]]) -> list[str]:
|
||||
if not rows:
|
||||
return []
|
||||
|
||||
width = len(rows[0])
|
||||
alignments: list[str] = []
|
||||
data_rows = rows[1:] if len(rows) > 1 else rows
|
||||
for col_index in range(width):
|
||||
values = [row[col_index] for row in data_rows if not _is_empty(row[col_index])]
|
||||
if values and all(_is_numeric_value(value) for value in values):
|
||||
alignments.append("---:")
|
||||
else:
|
||||
alignments.append("---")
|
||||
return alignments
|
||||
|
||||
|
||||
def _rows_to_markdown_table(rows: list[list[Any]]) -> str:
|
||||
if not rows:
|
||||
return "_No tabular content found._"
|
||||
|
||||
formatted_rows = [[_format_cell_value(value) for value in row] for row in rows]
|
||||
width = len(formatted_rows[0])
|
||||
separator = _column_alignments(rows)
|
||||
lines = [
|
||||
"| " + " | ".join(formatted_rows[0]) + " |",
|
||||
"| " + " | ".join(separator or ["---"] * width) + " |",
|
||||
]
|
||||
|
||||
for row in formatted_rows[1:]:
|
||||
lines.append("| " + " | ".join(row) + " |")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# Excel → Markdown
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _convert_excel(input_file: Path, out_file: Path, max_rows: int, max_cols: int) -> str:
|
||||
try:
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.utils import get_column_letter
|
||||
except ImportError:
|
||||
print("[ERROR] openpyxl not installed. Run: pip install openpyxl")
|
||||
return ""
|
||||
|
||||
workbook = load_workbook(input_file, data_only=True, read_only=False)
|
||||
visible_sheets = [sheet for sheet in workbook.worksheets if sheet.sheet_state == "visible"]
|
||||
|
||||
lines: list[str] = [
|
||||
f"# Spreadsheet Source: {input_file.name}",
|
||||
"",
|
||||
"## Workbook Summary",
|
||||
"",
|
||||
f"- Sheets: {len(workbook.worksheets)}",
|
||||
f"- Visible sheets: {', '.join(sheet.title for sheet in visible_sheets) or 'None'}",
|
||||
"",
|
||||
"> Note: Formula cells are exported as cached values. This converter does not recalculate formulas.",
|
||||
"",
|
||||
]
|
||||
|
||||
if not visible_sheets:
|
||||
lines.extend(["_No visible sheets found._", ""])
|
||||
|
||||
for worksheet in visible_sheets:
|
||||
merged_values = _merged_value_map(worksheet)
|
||||
bounds = _content_bounds(worksheet, merged_values)
|
||||
|
||||
lines.extend([
|
||||
f"## Sheet: {worksheet.title}",
|
||||
"",
|
||||
f"- State: {_sheet_state_label(worksheet.sheet_state)}",
|
||||
])
|
||||
|
||||
if bounds is None:
|
||||
lines.extend(["", "_No content found._", ""])
|
||||
continue
|
||||
|
||||
min_row, min_col, max_row, max_col = bounds
|
||||
used_range = (
|
||||
f"{get_column_letter(min_col)}{min_row}:"
|
||||
f"{get_column_letter(max_col)}{max_row}"
|
||||
)
|
||||
rows, rows_truncated, cols_truncated = _extract_rows(
|
||||
worksheet,
|
||||
bounds,
|
||||
merged_values,
|
||||
max_rows=max_rows,
|
||||
max_cols=max_cols,
|
||||
)
|
||||
|
||||
lines.extend([
|
||||
f"- Used range: {used_range}",
|
||||
f"- Rows: {max_row - min_row + 1}",
|
||||
f"- Columns: {max_col - min_col + 1}",
|
||||
"",
|
||||
])
|
||||
|
||||
if rows_truncated or cols_truncated:
|
||||
limit_notes = []
|
||||
if rows_truncated:
|
||||
limit_notes.append(f"rows limited to {max_rows}")
|
||||
if cols_truncated:
|
||||
limit_notes.append(f"columns limited to {max_cols}")
|
||||
lines.extend([f"> Truncated: {', '.join(limit_notes)}.", ""])
|
||||
|
||||
lines.extend([_rows_to_markdown_table(rows), ""])
|
||||
|
||||
markdown = "\n".join(lines).rstrip() + "\n"
|
||||
out_file.write_text(markdown, encoding="utf-8")
|
||||
_report_result(out_file)
|
||||
return markdown
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# Dispatcher
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
def convert_to_markdown(
|
||||
input_path: str,
|
||||
output_path: str | None = None,
|
||||
max_rows: int = 0,
|
||||
max_cols: int = 0,
|
||||
) -> str:
|
||||
input_file = Path(input_path)
|
||||
if not input_file.exists():
|
||||
print(f"[ERROR] File not found: {input_path}")
|
||||
return ""
|
||||
|
||||
suffix = input_file.suffix.lower()
|
||||
if suffix in LEGACY_EXCEL_FORMATS:
|
||||
print("[ERROR] Unsupported legacy Excel format: .xls")
|
||||
print(" Please resave the workbook as .xlsx and run this converter again.")
|
||||
return ""
|
||||
if suffix not in EXCEL_FORMATS:
|
||||
supported = ", ".join(sorted(EXCEL_FORMATS))
|
||||
print(f"[ERROR] Unsupported format: {suffix}")
|
||||
print(f" Supported: {supported}")
|
||||
return ""
|
||||
|
||||
if max_rows < 0 or max_cols < 0:
|
||||
print("[ERROR] --max-rows and --max-cols must be zero or positive integers")
|
||||
return ""
|
||||
|
||||
out_file = Path(output_path) if output_path else input_file.with_suffix(".md")
|
||||
out_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
print(f"[INFO] Converting Excel workbook: {input_file.name}")
|
||||
markdown = _convert_excel(input_file, out_file, max_rows=max_rows, max_cols=max_cols)
|
||||
if markdown:
|
||||
profile_path = write_conversion_profile_best_effort(
|
||||
input_path=str(input_file),
|
||||
markdown_path=out_file,
|
||||
converter="excel_to_md.py",
|
||||
conversion_type=suffix.lstrip("."),
|
||||
)
|
||||
if profile_path:
|
||||
print(f" Wrote conversion profile -> {profile_path}")
|
||||
return markdown
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Convert Excel workbooks to Markdown",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
Examples:
|
||||
python excel_to_md.py report.xlsx
|
||||
python excel_to_md.py report.xlsx budget.xlsm
|
||||
python excel_to_md.py ./workbooks -o ./markdown
|
||||
python excel_to_md.py report.xlsx -o output.md
|
||||
python excel_to_md.py report.xlsm --max-rows 200 --max-cols 40
|
||||
|
||||
Supported formats:
|
||||
.xlsx .xlsm
|
||||
|
||||
Unsupported by default:
|
||||
.xls Resave as .xlsx first
|
||||
""",
|
||||
)
|
||||
parser.add_argument("inputs", nargs="+", help="Input Excel workbook(s) or directories")
|
||||
parser.add_argument(
|
||||
"-o",
|
||||
"--output",
|
||||
help="Output Markdown file for one input, or output directory for multiple inputs/directories",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-rows",
|
||||
type=int,
|
||||
default=0,
|
||||
help="Maximum rows per sheet to export (0 = no limit)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-cols",
|
||||
type=int,
|
||||
default=0,
|
||||
help="Maximum columns per sheet to export (0 = no limit)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
return run_path_batch(
|
||||
args.inputs,
|
||||
EXCEL_FORMATS | LEGACY_EXCEL_FORMATS,
|
||||
args.output,
|
||||
lambda source, output: bool(
|
||||
convert_to_markdown(
|
||||
str(source),
|
||||
str(output),
|
||||
max_rows=args.max_rows,
|
||||
max_cols=args.max_cols,
|
||||
)
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,986 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
web_to_md.py - Web Page to Markdown Converter (Python Version)
|
||||
|
||||
Usage:
|
||||
python scripts/source_to_md/web_to_md.py <url>
|
||||
python scripts/source_to_md/web_to_md.py <url1> <url2> ...
|
||||
python scripts/source_to_md/web_to_md.py -f urls.txt
|
||||
python scripts/source_to_md/web_to_md.py <url> -o output.md
|
||||
|
||||
Dependencies:
|
||||
pip install requests beautifulsoup4
|
||||
|
||||
TLS fingerprint handling:
|
||||
Some sites (e.g., WeChat mp.weixin.qq.com) block Python's default 'requests'
|
||||
library based on TLS fingerprints (JA3). If 'curl_cffi' is installed, this script
|
||||
uses it to impersonate a modern Chrome fingerprint and bypass such blocks. If
|
||||
'curl_cffi' is unavailable, it silently falls back to plain 'requests' — so
|
||||
non-blocking sites still work without the extra dependency.
|
||||
|
||||
Install for WeChat / Chinese-portal coverage:
|
||||
pip install curl_cffi
|
||||
|
||||
If curl_cffi is unavailable on your platform, the Node.js counterpart
|
||||
(scripts/source_to_md/web_to_md.cjs) remains available as a fallback.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import codecs
|
||||
import datetime
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
from urllib.parse import urljoin, urlparse
|
||||
|
||||
_SCRIPTS_DIR = Path(__file__).resolve().parents[1]
|
||||
if str(_SCRIPTS_DIR) not in sys.path:
|
||||
sys.path.insert(0, str(_SCRIPTS_DIR))
|
||||
|
||||
from console_encoding import configure_utf8_stdio # noqa: E402
|
||||
from _conversion_profile import ( # noqa: E402
|
||||
profile_path_for,
|
||||
write_conversion_profile_best_effort,
|
||||
)
|
||||
|
||||
configure_utf8_stdio()
|
||||
|
||||
try:
|
||||
import requests
|
||||
from bs4 import BeautifulSoup, NavigableString, Tag
|
||||
except ImportError:
|
||||
print("Error: This script requires 'requests' and 'beautifulsoup4'.")
|
||||
print("Please run: pip install requests beautifulsoup4")
|
||||
sys.exit(1)
|
||||
|
||||
# Prefer curl_cffi for TLS-fingerprint impersonation (bypasses JA3 blocking on
|
||||
# sites like WeChat). Fall back to plain requests when it's not installed.
|
||||
try:
|
||||
from curl_cffi import requests as curl_requests # type: ignore
|
||||
_CURL_IMPERSONATE = "chrome120"
|
||||
except ImportError:
|
||||
curl_requests = None
|
||||
_CURL_IMPERSONATE = None
|
||||
|
||||
|
||||
def _http_get(url: str, *, headers: dict | None = None, timeout: int | None = None,
|
||||
verify: bool = False, stream: bool = False):
|
||||
"""HTTP GET with curl_cffi preferred, requests fallback.
|
||||
|
||||
Using curl_cffi lets this script fetch sites that reject Python's default
|
||||
TLS fingerprint (notably mp.weixin.qq.com). Signature mirrors the subset of
|
||||
requests.get() this script actually uses.
|
||||
"""
|
||||
if curl_requests is not None:
|
||||
return curl_requests.get(
|
||||
url, headers=headers, timeout=timeout,
|
||||
verify=verify, impersonate=_CURL_IMPERSONATE, stream=stream,
|
||||
)
|
||||
return requests.get(url, headers=headers, timeout=timeout,
|
||||
verify=verify, stream=stream)
|
||||
|
||||
|
||||
def _normalize_charset(charset: str | None) -> str:
|
||||
"""Return a Python codec name when the declared charset is usable."""
|
||||
if not charset:
|
||||
return ""
|
||||
charset = charset.strip().strip('"').strip("'").lower()
|
||||
if not charset:
|
||||
return ""
|
||||
try:
|
||||
return codecs.lookup(charset).name
|
||||
except LookupError:
|
||||
return ""
|
||||
|
||||
|
||||
def _charset_from_headers(headers: dict) -> str:
|
||||
content_type = headers.get("Content-Type") or headers.get("content-type") or ""
|
||||
match = re.search(r"charset\s*=\s*([^;\s]+)", content_type, re.I)
|
||||
return _normalize_charset(match.group(1)) if match else ""
|
||||
|
||||
|
||||
def _charset_from_html(raw: bytes) -> str:
|
||||
"""Extract a charset declaration from the first chunk of HTML bytes."""
|
||||
head = raw[:8192]
|
||||
patterns = [
|
||||
rb"<meta[^>]+charset=[\"']?\s*([a-zA-Z0-9_\-]+)",
|
||||
rb"<meta[^>]+content=[\"'][^\"']*charset=\s*([a-zA-Z0-9_\-]+)",
|
||||
]
|
||||
for pattern in patterns:
|
||||
match = re.search(pattern, head, re.I)
|
||||
if match:
|
||||
return _normalize_charset(match.group(1).decode("ascii", "ignore"))
|
||||
return ""
|
||||
|
||||
|
||||
def _decode_quality_score(text: str) -> int:
|
||||
"""Score obvious decode artifacts; lower is better."""
|
||||
mojibake_markers = [
|
||||
"�", "锟", "Ã", "Â", "â€", "’", "“", "â€\x9d",
|
||||
"琚", "佸", "鍦", "涓", "鏄", "寤", "骞", "鏈", "鏃", "鈥",
|
||||
]
|
||||
marker_hits = sum(text.count(marker) for marker in mojibake_markers)
|
||||
control_hits = sum(1 for ch in text if ord(ch) < 32 and ch not in "\t\n\r")
|
||||
return marker_hits * 20 + control_hits * 10 + text.count("\ufffd") * 50
|
||||
|
||||
|
||||
def _decode_response_text(response) -> str:
|
||||
"""Decode HTTP response bytes without letting guessed encodings override declarations."""
|
||||
raw = response.content
|
||||
declared = [
|
||||
_charset_from_headers(response.headers),
|
||||
_charset_from_html(raw),
|
||||
]
|
||||
if raw.startswith(codecs.BOM_UTF8):
|
||||
declared.insert(0, "utf-8-sig")
|
||||
|
||||
seen = set()
|
||||
declared = [enc for enc in declared if enc and not (enc in seen or seen.add(enc))]
|
||||
for enc in declared:
|
||||
try:
|
||||
return raw.decode(enc)
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
|
||||
candidates = []
|
||||
for enc in [
|
||||
getattr(response, "encoding", None),
|
||||
getattr(response, "apparent_encoding", None),
|
||||
"utf-8",
|
||||
"gb18030",
|
||||
"big5",
|
||||
]:
|
||||
enc = _normalize_charset(enc)
|
||||
if enc and enc not in candidates:
|
||||
candidates.append(enc)
|
||||
|
||||
decoded = []
|
||||
for enc in candidates:
|
||||
try:
|
||||
text = raw.decode(enc)
|
||||
except UnicodeDecodeError:
|
||||
continue
|
||||
decoded.append((_decode_quality_score(text), enc, text))
|
||||
|
||||
if decoded:
|
||||
decoded.sort(key=lambda item: item[0])
|
||||
return decoded[0][2]
|
||||
|
||||
return raw.decode("utf-8", errors="replace")
|
||||
|
||||
try:
|
||||
from PIL import Image
|
||||
PILLOW_AVAILABLE = True
|
||||
except ImportError:
|
||||
PILLOW_AVAILABLE = False
|
||||
print("[WARN] Pillow not installed. WebP images will not be converted to PNG.")
|
||||
print(" Run: pip install Pillow")
|
||||
|
||||
# ============ Config ============
|
||||
CONFIG = {
|
||||
"output_dir": "./projects",
|
||||
"timeout": 30,
|
||||
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||||
# Specific content identifiers often found in Chinese CMS (Gov/News)
|
||||
"content_selectors": [
|
||||
{"class_": re.compile(r"tys-main-zt-show", re.I)},
|
||||
{"class_": re.compile(r"tys-main", re.I)},
|
||||
{"class_": "TRS_Editor"},
|
||||
{"class_": "TRS_UEDITOR"},
|
||||
{"class_": "ucontent"},
|
||||
{"class_": "article-content"},
|
||||
{"class_": "news-content"},
|
||||
{"class_": "detail-content"},
|
||||
{"class_": "content-text"},
|
||||
{"class_": "pages_content"},
|
||||
{"class_": "zwgk_content"},
|
||||
{"class_": "content_detail"},
|
||||
{"class_": "text_content"},
|
||||
{"class_": "main-content"},
|
||||
{"class_": "main_content"},
|
||||
{"class_": "view-content"},
|
||||
{"class_": "info-content"},
|
||||
{"id": "Zoom"},
|
||||
{"id": "content"},
|
||||
{"id": "article"},
|
||||
{"class_": "content"},
|
||||
{"name": "article"}, # tag name
|
||||
{"name": "main"}, # tag name
|
||||
]
|
||||
}
|
||||
|
||||
|
||||
def fetch_url(url: str) -> str:
|
||||
"""Fetch a web page with explicit headers and encoding detection.
|
||||
|
||||
Args:
|
||||
url: Target URL.
|
||||
|
||||
Returns:
|
||||
The response body as text.
|
||||
"""
|
||||
headers = {
|
||||
"User-Agent": CONFIG["user_agent"],
|
||||
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
||||
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8"
|
||||
}
|
||||
|
||||
try:
|
||||
response = _http_get(url, headers=headers,
|
||||
timeout=CONFIG["timeout"], verify=False)
|
||||
response.raise_for_status()
|
||||
|
||||
return _decode_response_text(response)
|
||||
except Exception as e:
|
||||
raise Exception(f"Failed to fetch {url}: {str(e)}")
|
||||
|
||||
|
||||
def clean_title(title: str) -> str:
|
||||
"""Remove common site suffixes from a title."""
|
||||
if not title:
|
||||
return ""
|
||||
# Remove site name suffixes often found in Chinese titles
|
||||
clean = re.sub(r"[-_|].*?(政府|门户|网站|委员会).*$", "", title)
|
||||
return clean.strip()
|
||||
|
||||
|
||||
def sanitize_filename(name: str) -> str:
|
||||
"""Sanitize a string for filesystem-safe filenames."""
|
||||
# Replace whitespace with underscore first
|
||||
clean = re.sub(r'\s+', '_', name)
|
||||
# Remove all except Chinese, English, Numbers, Underscore
|
||||
clean = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9_]', '', clean)
|
||||
# Collapse repeating underscores
|
||||
clean = re.sub(r'_+', '_', clean)
|
||||
return clean[:80] # Truncate
|
||||
|
||||
|
||||
def derive_base_name(title: str, url: str) -> str:
|
||||
"""Derive a safe, non-empty basename from a title or URL."""
|
||||
base = sanitize_filename(title or "")
|
||||
if base:
|
||||
return base
|
||||
|
||||
parsed = urlparse(url)
|
||||
path = parsed.path.strip('/')
|
||||
if path:
|
||||
candidate = f"{parsed.netloc}_{path}"
|
||||
else:
|
||||
candidate = parsed.netloc or "untitled"
|
||||
base = sanitize_filename(candidate)
|
||||
if base:
|
||||
return base
|
||||
|
||||
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
|
||||
return f"untitled_{ts}"
|
||||
|
||||
|
||||
def build_image_filename(abs_url: str, seq: int, content_type: str | None = None) -> str:
|
||||
"""Build a safe image filename from URL metadata."""
|
||||
parsed = urlparse(abs_url)
|
||||
basename = os.path.basename(parsed.path).split('?')[0]
|
||||
stem, ext = os.path.splitext(basename)
|
||||
if not ext or len(ext) > 5 or '/' in ext:
|
||||
ext = ""
|
||||
if not ext and content_type:
|
||||
ctype = content_type.split(';')[0].lower()
|
||||
ext_map = {
|
||||
"image/jpeg": ".jpg",
|
||||
"image/jpg": ".jpg",
|
||||
"image/png": ".png",
|
||||
"image/gif": ".gif",
|
||||
"image/webp": ".webp",
|
||||
}
|
||||
ext = ext_map.get(ctype, "")
|
||||
if not ext:
|
||||
ext = ".jpg"
|
||||
stem = sanitize_filename(stem) if stem else f"image_{seq}"
|
||||
return f"{stem}{ext}"
|
||||
|
||||
|
||||
def download_and_rewrite_images(
|
||||
content_element: Tag | None,
|
||||
page_url: str,
|
||||
image_dir: str,
|
||||
rel_prefix: str,
|
||||
) -> int:
|
||||
"""Download images under the main content node and rewrite `src` paths."""
|
||||
if content_element is None:
|
||||
return 0
|
||||
images = list(content_element.find_all("img"))
|
||||
if not images:
|
||||
return 0
|
||||
|
||||
os.makedirs(image_dir, exist_ok=True)
|
||||
downloaded = {}
|
||||
manifest_by_filename: dict[str, dict[str, object]] = {}
|
||||
saved = 0
|
||||
|
||||
for idx, img in enumerate(images):
|
||||
# Prefer lazy-load attributes — WeChat, Zhihu, and many CMSes keep the
|
||||
# real image URL in data-src / data-original / data-lazy-src, with
|
||||
# `src` pointing at a 1x1 placeholder or a template literal.
|
||||
candidates = [
|
||||
img.get("data-src"),
|
||||
img.get("data-original"),
|
||||
img.get("data-lazy-src"),
|
||||
img.get("data-actualsrc"),
|
||||
img.get("src"),
|
||||
]
|
||||
src = next((s for s in candidates
|
||||
if s and not s.startswith("data:")
|
||||
and s.startswith(("http://", "https://", "//", "/"))), None)
|
||||
if not src:
|
||||
continue
|
||||
|
||||
# Promote the chosen URL into the element's src so downstream rewrite
|
||||
# (which matches on src) can retarget it to the local file.
|
||||
img["src"] = src
|
||||
|
||||
abs_url = urljoin(page_url, src)
|
||||
content_type = ""
|
||||
converted_from = ""
|
||||
if abs_url in downloaded:
|
||||
saved_name = downloaded[abs_url]
|
||||
else:
|
||||
try:
|
||||
resp = _http_get(
|
||||
abs_url,
|
||||
headers={"User-Agent": CONFIG["user_agent"]},
|
||||
timeout=CONFIG["timeout"],
|
||||
verify=False,
|
||||
)
|
||||
resp.raise_for_status()
|
||||
filename = build_image_filename(
|
||||
abs_url, idx, resp.headers.get("Content-Type"))
|
||||
|
||||
# Check if image is webp and convert to png
|
||||
stem, ext = os.path.splitext(filename)
|
||||
content_type = resp.headers.get("Content-Type", "").lower()
|
||||
is_webp = ext.lower() == ".webp" or "webp" in content_type
|
||||
|
||||
if is_webp and PILLOW_AVAILABLE:
|
||||
# Convert webp to png (optimized)
|
||||
try:
|
||||
img_data = io.BytesIO(resp.content)
|
||||
pil_image = Image.open(img_data)
|
||||
|
||||
# Update filename to .png
|
||||
converted_from = filename
|
||||
filename = f"{stem}.png"
|
||||
local_path = os.path.join(image_dir, filename)
|
||||
|
||||
# Avoid accidental overwrites if filenames collide
|
||||
counter = 1
|
||||
while os.path.exists(local_path):
|
||||
local_path = os.path.join(
|
||||
image_dir, f"{stem}_{counter}.png")
|
||||
filename = os.path.basename(local_path)
|
||||
counter += 1
|
||||
|
||||
# Save as PNG directly (Pillow auto-converts, no need for explicit mode conversion)
|
||||
pil_image.save(local_path, 'PNG', optimize=False)
|
||||
pil_image.close()
|
||||
print(f" [INFO] Converted webp to png: {filename}")
|
||||
except Exception as convert_err:
|
||||
print(
|
||||
f" [WARN] Failed to convert webp: {convert_err}, saving as-is")
|
||||
local_path = os.path.join(image_dir, filename)
|
||||
counter = 1
|
||||
stem, ext = os.path.splitext(filename)
|
||||
while os.path.exists(local_path):
|
||||
local_path = os.path.join(
|
||||
image_dir, f"{stem}_{counter}{ext}")
|
||||
filename = os.path.basename(local_path)
|
||||
counter += 1
|
||||
with open(local_path, "wb") as f:
|
||||
f.write(resp.content)
|
||||
else:
|
||||
local_path = os.path.join(image_dir, filename)
|
||||
|
||||
# Avoid accidental overwrites if filenames collide
|
||||
counter = 1
|
||||
stem, ext = os.path.splitext(filename)
|
||||
while os.path.exists(local_path):
|
||||
local_path = os.path.join(
|
||||
image_dir, f"{stem}_{counter}{ext}")
|
||||
filename = os.path.basename(local_path)
|
||||
counter += 1
|
||||
|
||||
with open(local_path, "wb") as f:
|
||||
f.write(resp.content)
|
||||
downloaded[abs_url] = filename
|
||||
saved_name = filename
|
||||
manifest_by_filename[saved_name] = {
|
||||
"index": len(manifest_by_filename) + 1,
|
||||
"filename": saved_name,
|
||||
"original_filename": converted_from or saved_name,
|
||||
"asset_kind": "bitmap",
|
||||
"svg_renderable": True,
|
||||
"pptx_native_supported": True,
|
||||
"source_kind": "web_image",
|
||||
"source_url": abs_url,
|
||||
"source_page_url": page_url,
|
||||
"content_type": content_type.split(";")[0] if content_type else "",
|
||||
"occurrences": [],
|
||||
}
|
||||
saved += 1
|
||||
except Exception as e:
|
||||
print(f" [WARN] Skip image {abs_url}: {e}")
|
||||
continue
|
||||
|
||||
rel_path = os.path.join(
|
||||
rel_prefix, saved_name) if rel_prefix else saved_name
|
||||
img["src"] = rel_path
|
||||
manifest_item = manifest_by_filename.get(saved_name)
|
||||
if manifest_item is not None:
|
||||
occurrences = manifest_item.setdefault("occurrences", [])
|
||||
if isinstance(occurrences, list):
|
||||
occurrences.append({
|
||||
"occurrence_index": idx + 1,
|
||||
"source_url": abs_url,
|
||||
"alt_text": img.get("alt", ""),
|
||||
})
|
||||
manifest_item["usage_count"] = len(occurrences)
|
||||
|
||||
if manifest_by_filename:
|
||||
manifest_path = os.path.join(image_dir, "image_manifest.json")
|
||||
with open(manifest_path, "w", encoding="utf-8") as f:
|
||||
json.dump(
|
||||
list(manifest_by_filename.values()),
|
||||
f,
|
||||
ensure_ascii=False,
|
||||
indent=2,
|
||||
)
|
||||
f.write("\n")
|
||||
|
||||
return saved
|
||||
|
||||
|
||||
def extract_metadata(soup: BeautifulSoup, url: str) -> dict[str, str]:
|
||||
"""Extract page metadata such as title, date, description, and author."""
|
||||
|
||||
# 1. Title
|
||||
title_tag = soup.title
|
||||
title = clean_title(title_tag.string if title_tag else "")
|
||||
|
||||
# 2. Meta tags
|
||||
metas = {}
|
||||
for meta in soup.find_all("meta"):
|
||||
name = meta.get("name") or meta.get("property")
|
||||
content = meta.get("content")
|
||||
if name and content:
|
||||
metas[name.lower()] = content.strip()
|
||||
|
||||
# 3. Date Extraction Strategies
|
||||
date = (
|
||||
metas.get("article:published_time") or
|
||||
metas.get("og:published_time") or
|
||||
metas.get("pubdate") or
|
||||
metas.get("publishdate") or
|
||||
metas.get("date")
|
||||
)
|
||||
|
||||
if not date:
|
||||
# Try matching date patterns in the text
|
||||
text_content = soup.get_text()
|
||||
date_patterns = [
|
||||
r"发布[时日]间[：:]\s*(\d{4}[-\/年]\d{1,2}[-\/月]\d{1,2}[日]?)",
|
||||
r"日期[：:]\s*(\d{4}[-\/年]\d{1,2}[-\/月]\d{1,2}[日]?)",
|
||||
r"(\d{4}[-\/年]\d{1,2}[-\/月]\d{1,2}[日]?)\s*(?:发布|来源)",
|
||||
r"时间[：:]\s*(\d{4}[-\/]\d{1,2}[-\/]\d{1,2})"
|
||||
]
|
||||
for pattern in date_patterns:
|
||||
match = re.search(pattern, text_content)
|
||||
if match:
|
||||
date = match.group(1).replace(
|
||||
"年", "-").replace("月", "-").replace("日", "")
|
||||
break
|
||||
|
||||
if not date:
|
||||
# Try URL matching
|
||||
match = re.search(r"(\d{4})(\d{2})[\/_](?:t\d+_)?", url)
|
||||
if match:
|
||||
date = f"{match.group(1)}-{match.group(2)}"
|
||||
else:
|
||||
match = re.search(r"(\d{4})[-\/](\d{2})[-\/](\d{2})", url)
|
||||
if match:
|
||||
date = f"{match.group(1)}-{match.group(2)}-{match.group(3)}"
|
||||
|
||||
# 4. Description
|
||||
description = (
|
||||
metas.get("description") or
|
||||
metas.get("og:description") or
|
||||
metas.get("twitter:description") or
|
||||
""
|
||||
)
|
||||
|
||||
# 5. Author/Source
|
||||
author = metas.get("author") or metas.get("article:author")
|
||||
if not author:
|
||||
# Try common patterns
|
||||
source_patterns = [
|
||||
r"来源[：:]\s*([^\s<]+)",
|
||||
r"发布(?:单位|机构)[：:]\s*([^\s<]+)"
|
||||
]
|
||||
for pattern in source_patterns:
|
||||
match = re.search(pattern, soup.get_text())
|
||||
if match:
|
||||
author = match.group(1)
|
||||
break
|
||||
|
||||
return {
|
||||
"title": title or metas.get("og:title") or "Untitled",
|
||||
"date": date or "",
|
||||
"description": description,
|
||||
"author": author or "",
|
||||
"source_url": url
|
||||
}
|
||||
|
||||
|
||||
def find_main_content(soup: BeautifulSoup) -> Tag | None:
|
||||
"""Find the most likely main content container in a page."""
|
||||
# 1. Clean up first (remove known clutter)
|
||||
for tag in soup(["script", "style", "nav", "header", "footer", "aside", "noscript", "iframe"]):
|
||||
tag.decompose()
|
||||
|
||||
best_element = None
|
||||
max_score = 0
|
||||
|
||||
# 2. Strategy A: Check specific classes/ids
|
||||
for selector in CONFIG["content_selectors"]:
|
||||
if "name" in selector:
|
||||
# Tag name match (article, main)
|
||||
elements = soup.find_all(selector["name"])
|
||||
else:
|
||||
# Class or ID match
|
||||
elements = soup.find_all(attrs=selector)
|
||||
|
||||
for el in elements:
|
||||
# Score based on text length and chinese character count
|
||||
text = el.get_text(strip=True)
|
||||
length = len(text)
|
||||
if length < 100:
|
||||
continue
|
||||
|
||||
chinese_count = len(re.findall(r'[\u4e00-\u9fa5]', text))
|
||||
score = length + (chinese_count * 2)
|
||||
|
||||
if score > max_score:
|
||||
max_score = score
|
||||
best_element = el
|
||||
|
||||
# 3. Strategy B: If no specific container found, look for dense text areas with paragraphs
|
||||
if not best_element or max_score < 200:
|
||||
for div in soup.find_all("div"):
|
||||
p_count = len(div.find_all("p", recursive=False))
|
||||
# recursive=False ensures we don't just pick the top-level body by accident
|
||||
# but sometimes content is nested deep
|
||||
if p_count == 0:
|
||||
# Check if it has lots of text even without p tags (br tags?)
|
||||
pass
|
||||
|
||||
text = div.get_text(strip=True)
|
||||
if len(text) > 200 and p_count >= 1:
|
||||
# Recalculate deep score
|
||||
chinese_count = len(re.findall(r'[\u4e00-\u9fa5]', text))
|
||||
score = len(text) + (chinese_count * 2) + (p_count * 50)
|
||||
if score > max_score:
|
||||
max_score = score
|
||||
best_element = div
|
||||
|
||||
# Fallback to body
|
||||
return best_element if best_element else soup.body
|
||||
|
||||
|
||||
def element_to_markdown(element: Tag | NavigableString | None) -> str:
|
||||
"""Recursively convert a BeautifulSoup node to Markdown."""
|
||||
if element is None:
|
||||
return ""
|
||||
|
||||
if isinstance(element, NavigableString):
|
||||
text = str(element).strip()
|
||||
return text if text else ""
|
||||
|
||||
tag_name = element.name.lower()
|
||||
|
||||
# Skip hidden/unwanted tags
|
||||
if tag_name in ['script', 'style', 'meta', 'link', 'input', 'button', 'select']:
|
||||
return ""
|
||||
|
||||
content = ""
|
||||
for child in element.children:
|
||||
content += element_to_markdown(child)
|
||||
# Add spacing logic here if needed, but usually block elements handle it
|
||||
|
||||
# Block handlers
|
||||
if tag_name in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6']:
|
||||
level = int(tag_name[1])
|
||||
return f"\n{'#' * level} {content}\n\n"
|
||||
|
||||
elif tag_name == 'p':
|
||||
# Clean up internal whitespace
|
||||
content = re.sub(r'\s+', ' ', content).strip()
|
||||
return f"\n{content}\n\n" if content else ""
|
||||
|
||||
elif tag_name == 'br':
|
||||
return " \n"
|
||||
|
||||
elif tag_name == 'hr':
|
||||
return "\n---\n"
|
||||
|
||||
elif tag_name == 'div':
|
||||
return f"\n{content}\n"
|
||||
|
||||
elif tag_name == 'blockquote':
|
||||
lines = content.strip().split('\n')
|
||||
quoted = '\n'.join([f"> {line}" for line in lines if line.strip()])
|
||||
return f"\n{quoted}\n\n"
|
||||
|
||||
elif tag_name in ['ul', 'ol']:
|
||||
# This is tricky without "state" (knowing we are in a list)
|
||||
# For simplicity in this recursive version, we rely on LI handling
|
||||
return f"\n{content}\n"
|
||||
|
||||
elif tag_name == 'li':
|
||||
# Simple list handling
|
||||
clean_content = content.strip()
|
||||
return f"- {clean_content}\n"
|
||||
|
||||
elif tag_name == 'pre':
|
||||
return f"\n```\n{content}\n```\n\n"
|
||||
|
||||
elif tag_name == 'code':
|
||||
# If parent is pre, handle in pre. If inline:
|
||||
parent = element.parent
|
||||
if parent and parent.name == 'pre':
|
||||
return content
|
||||
return f"`{content}`"
|
||||
|
||||
elif tag_name == 'a':
|
||||
href = element.get('href', '')
|
||||
if href and not href.startswith('javascript:'):
|
||||
return f"[{content}]({href})"
|
||||
return content
|
||||
|
||||
elif tag_name == 'img':
|
||||
src = element.get('src', '')
|
||||
alt = element.get('alt', '')
|
||||
if src:
|
||||
return f""
|
||||
return ""
|
||||
|
||||
elif tag_name == 'table':
|
||||
# Basic table text extraction, full markdown table support is complex
|
||||
# Leaving as raw text or simplistic conversion for now
|
||||
# Ideally, we'd parse TRs and TDs
|
||||
return f"\n{content}\n"
|
||||
|
||||
elif tag_name == 'tr':
|
||||
return f"{content}|\n"
|
||||
|
||||
elif tag_name in ['td', 'th']:
|
||||
return f"| {content.strip()} "
|
||||
|
||||
# Style formatting
|
||||
elif tag_name in ['strong', 'b']:
|
||||
return f"**{content}**"
|
||||
elif tag_name in ['em', 'i']:
|
||||
return f"*{content}*"
|
||||
elif tag_name in ['del', 's', 'strike']:
|
||||
return f"~~{content}~~"
|
||||
|
||||
# Default for span, section, etc.
|
||||
return f"{content} "
|
||||
|
||||
|
||||
def simple_html_to_markdown_traversal(soup: Tag | BeautifulSoup | None) -> str:
|
||||
"""Convert HTML content to Markdown using BeautifulSoup traversal."""
|
||||
lines = []
|
||||
|
||||
def traverse(node: Tag | NavigableString) -> str:
|
||||
if isinstance(node, NavigableString):
|
||||
text = str(node)
|
||||
# Normalize whitespace but keep single spaces
|
||||
text = re.sub(r'\s+', ' ', text)
|
||||
if text.strip():
|
||||
return text
|
||||
return ""
|
||||
|
||||
if node.name in ['script', 'style', 'comment', 'meta', 'link']:
|
||||
return ""
|
||||
|
||||
# Handle Block Elements
|
||||
is_block = node.name in ['p', 'div', 'h1', 'h2', 'h3', 'h4',
|
||||
'h5', 'h6', 'li', 'blockquote', 'pre', 'hr', 'table', 'tr']
|
||||
|
||||
# Pre-processing
|
||||
prefix = ""
|
||||
suffix = ""
|
||||
|
||||
if node.name in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6']:
|
||||
level = int(node.name[1])
|
||||
prefix = f"\n\n{'#' * level} "
|
||||
suffix = "\n\n"
|
||||
elif node.name == 'p':
|
||||
prefix = "\n\n"
|
||||
suffix = "\n\n"
|
||||
elif node.name == 'li':
|
||||
prefix = "\n- "
|
||||
elif node.name == 'blockquote':
|
||||
prefix = "\n> "
|
||||
suffix = "\n"
|
||||
elif node.name == 'hr':
|
||||
return "\n\n---\n\n"
|
||||
elif node.name == 'br':
|
||||
return " \n"
|
||||
elif node.name == 'pre':
|
||||
# Extract raw text from pre to preserve formatting
|
||||
return f"\n\n```\n{node.get_text()}\n```\n\n"
|
||||
|
||||
# Inline formatting
|
||||
if node.name in ['strong', 'b']:
|
||||
prefix, suffix = "**", "**"
|
||||
elif node.name in ['em', 'i']:
|
||||
prefix, suffix = "*", "*"
|
||||
elif node.name == 'code' and node.parent.name != 'pre':
|
||||
prefix, suffix = "`", "`"
|
||||
elif node.name == 'a':
|
||||
href = node.get('href')
|
||||
if href and not href.startswith('javascript:'):
|
||||
prefix = "["
|
||||
suffix = f"]({href})"
|
||||
else:
|
||||
prefix, suffix = "", ""
|
||||
elif node.name == 'img':
|
||||
src = node.get('src')
|
||||
alt = node.get('alt', '')
|
||||
if src:
|
||||
return f""
|
||||
return ""
|
||||
|
||||
# Recurse
|
||||
inner_text = ""
|
||||
for child in node.children:
|
||||
res = traverse(child)
|
||||
if res:
|
||||
inner_text += res
|
||||
|
||||
# Post-processing for tables (simplified)
|
||||
if node.name == 'tr':
|
||||
# count tds
|
||||
cells = [c.get_text(strip=True) for c in node.find_all(
|
||||
['td', 'th'], recursive=False)]
|
||||
return f"| {' | '.join(cells)} |\n"
|
||||
if node.name == 'table':
|
||||
# Try to add a separator line after first row if it looks like a header
|
||||
rows = inner_text.strip().split('\n')
|
||||
if rows:
|
||||
cols_count = rows[0].count('|') - 1
|
||||
if cols_count > 0:
|
||||
# rough approx
|
||||
sep = "| " + " | ".join(["---"] * int(cols_count/2)) + " |"
|
||||
# Actually, the traverse of TR returns newline terminated strings.
|
||||
# Let's just return what we gathered.
|
||||
pass
|
||||
return f"\n\n{inner_text}\n\n"
|
||||
|
||||
return f"{prefix}{inner_text}{suffix}"
|
||||
|
||||
# Actually, a simpler approach for this script is "just get string" but with markers?
|
||||
# Let's use a simplified approach: use get_text but with 'separator' logic?
|
||||
# text = soup.get_text(separator='\n\n')
|
||||
# But that loses links and boldness.
|
||||
|
||||
# Recommendation: Let's stick to the traversal above which constructs a string.
|
||||
md = traverse(soup)
|
||||
|
||||
# Cleanup Markdown
|
||||
if md:
|
||||
# Remove excessive newlines
|
||||
md = re.sub(r'\n{3,}', '\n\n', md)
|
||||
md = md.strip()
|
||||
return md or ""
|
||||
|
||||
|
||||
def process_url(url: str, output_file: str | None = None) -> tuple[bool, str, str | None, str | None]:
|
||||
"""Fetch, convert, and save one web page as Markdown.
|
||||
|
||||
Returns (success, url, error, output_path). output_path is the actual saved
|
||||
Markdown path (derived from the article title when no output_file is given),
|
||||
so a caller can locate a title-named file it did not choose upfront.
|
||||
"""
|
||||
print(f"\n[Fetching] {url}")
|
||||
try:
|
||||
html = fetch_url(url)
|
||||
soup = BeautifulSoup(html, 'html.parser')
|
||||
|
||||
# Extract Metadata
|
||||
metadata = extract_metadata(soup, url)
|
||||
print(f" [OK] Title: {metadata['title']}")
|
||||
if metadata['date']:
|
||||
print(f" [OK] Date: {metadata['date']}")
|
||||
|
||||
# Determine output path and image directory upfront
|
||||
if output_file:
|
||||
output_path = output_file
|
||||
else:
|
||||
base_name = derive_base_name(metadata['title'], url)
|
||||
filename = f"{base_name}.md"
|
||||
output_path = os.path.join(CONFIG["output_dir"], filename)
|
||||
|
||||
output_dirname = os.path.dirname(output_path) or "."
|
||||
os.makedirs(output_dirname, exist_ok=True)
|
||||
base_name = os.path.splitext(os.path.basename(output_path))[0]
|
||||
image_dir = os.path.join(output_dirname, f"{base_name}_files")
|
||||
rel_image_prefix = os.path.relpath(image_dir, output_dirname)
|
||||
|
||||
# Extract Content
|
||||
content_div = find_main_content(soup)
|
||||
|
||||
# Download images and rewrite src before markdown conversion
|
||||
image_count = download_and_rewrite_images(
|
||||
content_div, url, image_dir, rel_image_prefix)
|
||||
if image_count:
|
||||
print(f" [OK] Images: {image_count} saved to {image_dir}")
|
||||
|
||||
# Convert to MD
|
||||
# Note: We pass the element to our traversal function
|
||||
markdown_text = simple_html_to_markdown_traversal(content_div)
|
||||
print(f" [OK] Content: {len(markdown_text)} chars")
|
||||
|
||||
# Construct content
|
||||
final_output = []
|
||||
final_output.append("<!--")
|
||||
final_output.append(f" Source: {url}")
|
||||
final_output.append(
|
||||
f" Crawled: {datetime.datetime.now().isoformat()}")
|
||||
if metadata['date']:
|
||||
final_output.append(f" Published: {metadata['date']}")
|
||||
if metadata['author']:
|
||||
final_output.append(f" Author: {metadata['author']}")
|
||||
final_output.append("-->\n")
|
||||
|
||||
if metadata['title']:
|
||||
final_output.append(f"# {metadata['title']}\n")
|
||||
|
||||
if metadata['description']:
|
||||
final_output.append(f"> {metadata['description']}\n")
|
||||
|
||||
final_output.append(markdown_text)
|
||||
|
||||
full_content = "\n".join(final_output)
|
||||
|
||||
with open(output_path, 'w', encoding='utf-8') as f:
|
||||
f.write(full_content)
|
||||
profile_path = write_conversion_profile_best_effort(
|
||||
input_path=url,
|
||||
markdown_path=output_path,
|
||||
converter="web_to_md.py",
|
||||
conversion_type="web",
|
||||
asset_dir=image_dir,
|
||||
)
|
||||
|
||||
print(f" [OK] Saved: {output_path}")
|
||||
if profile_path:
|
||||
print(f" [OK] Conversion profile: {profile_path}")
|
||||
return True, url, None, output_path
|
||||
|
||||
except Exception as e:
|
||||
print(f" [ERROR] {str(e)}")
|
||||
return False, url, str(e), None
|
||||
|
||||
|
||||
def _write_emit_result(result_file: str, url: str, markdown_path: str) -> None:
|
||||
"""Write the actual saved path as JSON so a caller can locate the output."""
|
||||
md = Path(markdown_path).resolve()
|
||||
profile = profile_path_for(md)
|
||||
payload = {
|
||||
"input": url,
|
||||
"markdown": str(md),
|
||||
"conversion_profile": str(profile) if profile.is_file() else "",
|
||||
}
|
||||
try:
|
||||
Path(result_file).write_text(
|
||||
json.dumps(payload, ensure_ascii=False), encoding="utf-8")
|
||||
except OSError as exc:
|
||||
print(f" [WARN] Could not write --emit-result: {exc}")
|
||||
|
||||
|
||||
def main(argv: list[str] | None = None) -> int:
|
||||
"""Run the CLI entry point."""
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Web to Markdown Converter (Python)")
|
||||
parser.add_argument("urls", nargs="*", help="URLs to process")
|
||||
parser.add_argument(
|
||||
"-f", "--file", help="File containing URLs (one per line)")
|
||||
parser.add_argument("-o", "--output", help="Output file (single URL only)")
|
||||
parser.add_argument("-d", "--dir", help="Output directory")
|
||||
parser.add_argument(
|
||||
"--emit-result",
|
||||
help="On success, write the saved output path as JSON to this file "
|
||||
"(single-URL dispatcher use, so a title-named file can be located)")
|
||||
|
||||
args = parser.parse_args(argv)
|
||||
|
||||
if args.dir:
|
||||
CONFIG["output_dir"] = args.dir
|
||||
|
||||
targets = []
|
||||
if args.urls:
|
||||
targets.extend(args.urls)
|
||||
|
||||
if args.file:
|
||||
if os.path.exists(args.file):
|
||||
with open(args.file, 'r', encoding='utf-8') as f:
|
||||
lines = [l.strip() for l in f if l.strip()
|
||||
and not l.strip().startswith("#")]
|
||||
targets.extend(lines)
|
||||
else:
|
||||
print(f"Error: File {args.file} not found", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
if not targets:
|
||||
parser.print_usage(sys.stderr)
|
||||
print(
|
||||
"web_to_md.py: error: at least one URL or --file is required",
|
||||
file=sys.stderr,
|
||||
)
|
||||
return 2
|
||||
|
||||
results = []
|
||||
for i, url in enumerate(targets):
|
||||
# Allow specific output file only if 1 URL
|
||||
out = args.output if (len(targets) == 1 and args.output) else None
|
||||
success, url, err, out_path = process_url(url, out)
|
||||
results.append((success, url, err))
|
||||
if args.emit_result and success and out_path:
|
||||
_write_emit_result(args.emit_result, url, out_path)
|
||||
|
||||
# Summary
|
||||
success_count = sum(1 for r in results if r[0])
|
||||
fail_count = len(results) - success_count
|
||||
|
||||
print("\n" + "="*50)
|
||||
print(
|
||||
f"[Done] Success: {success_count}/{len(results)}, Failed: {fail_count}")
|
||||
|
||||
if fail_count > 0:
|
||||
print("\n[Failed URLs]:")
|
||||
for r in results:
|
||||
if not r[0]:
|
||||
print(f" - {r[1]}: {r[2]}")
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Disable warnings for verify=False if needed, though often useful to see
|
||||
import urllib3
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user