| 1 | #!/usr/bin/env python3 |
| 2 | # -*- coding: utf-8 -*- |
| 3 | """ |
| 4 | web_to_md.py - Web Page to Markdown Converter (Python Version) |
| 5 | |
| 6 | Usage: |
| 7 | python scripts/source_to_md/web_to_md.py <url> |
| 8 | python scripts/source_to_md/web_to_md.py <url1> <url2> ... |
| 9 | python scripts/source_to_md/web_to_md.py -f urls.txt |
| 10 | python scripts/source_to_md/web_to_md.py <url> -o output.md |
| 11 | |
| 12 | Dependencies: |
| 13 | pip install requests beautifulsoup4 |
| 14 | |
| 15 | TLS fingerprint handling: |
| 16 | Some sites (e.g., WeChat mp.weixin.qq.com) block Python's default 'requests' |
| 17 | library based on TLS fingerprints (JA3). If 'curl_cffi' is installed, this script |
| 18 | uses it to impersonate a modern Chrome fingerprint and bypass such blocks. If |
| 19 | 'curl_cffi' is unavailable, it silently falls back to plain 'requests' — so |
| 20 | non-blocking sites still work without the extra dependency. |
| 21 | |
| 22 | Install for WeChat / Chinese-portal coverage: |
| 23 | pip install curl_cffi |
| 24 | |
| 25 | If curl_cffi is unavailable on your platform, the Node.js counterpart |
| 26 | (scripts/source_to_md/web_to_md.cjs) remains available as a fallback. |
| 27 | """ |
| 28 | |
| 29 | import argparse |
| 30 | import codecs |
| 31 | import datetime |
| 32 | import io |
| 33 | import json |
| 34 | import os |
| 35 | import re |
| 36 | import sys |
| 37 | import time |
| 38 | from pathlib import Path |
| 39 | from urllib.parse import urljoin, urlparse |
| 40 | |
| 41 | _SCRIPTS_DIR = Path(__file__).resolve().parents[1] |
| 42 | if str(_SCRIPTS_DIR) not in sys.path: |
| 43 | sys.path.insert(0, str(_SCRIPTS_DIR)) |
| 44 | |
| 45 | from console_encoding import configure_utf8_stdio # noqa: E402 |
| 46 | from _conversion_profile import ( # noqa: E402 |
| 47 | profile_path_for, |
| 48 | write_conversion_profile_best_effort, |
| 49 | ) |
| 50 | |
| 51 | configure_utf8_stdio() |
| 52 | |
| 53 | try: |
| 54 | import requests |
| 55 | from bs4 import BeautifulSoup, NavigableString, Tag |
| 56 | except ImportError: |
| 57 | print("Error: This script requires 'requests' and 'beautifulsoup4'.") |
| 58 | print("Please run: pip install requests beautifulsoup4") |
| 59 | sys.exit(1) |
| 60 | |
| 61 | # Prefer curl_cffi for TLS-fingerprint impersonation (bypasses JA3 blocking on |
| 62 | # sites like WeChat). Fall back to plain requests when it's not installed. |
| 63 | try: |
| 64 | from curl_cffi import requests as curl_requests # type: ignore |
| 65 | _CURL_IMPERSONATE = "chrome120" |
| 66 | except ImportError: |
| 67 | curl_requests = None |
| 68 | _CURL_IMPERSONATE = None |
| 69 | |
| 70 | |
| 71 | def _http_get(url: str, *, headers: dict | None = None, timeout: int | None = None, |
| 72 | verify: bool = False, stream: bool = False): |
| 73 | """HTTP GET with curl_cffi preferred, requests fallback. |
| 74 | |
| 75 | Using curl_cffi lets this script fetch sites that reject Python's default |
| 76 | TLS fingerprint (notably mp.weixin.qq.com). Signature mirrors the subset of |
| 77 | requests.get() this script actually uses. |
| 78 | """ |
| 79 | if curl_requests is not None: |
| 80 | return curl_requests.get( |
| 81 | url, headers=headers, timeout=timeout, |
| 82 | verify=verify, impersonate=_CURL_IMPERSONATE, stream=stream, |
| 83 | ) |
| 84 | return requests.get(url, headers=headers, timeout=timeout, |
| 85 | verify=verify, stream=stream) |
| 86 | |
| 87 | |
| 88 | def _normalize_charset(charset: str | None) -> str: |
| 89 | """Return a Python codec name when the declared charset is usable.""" |
| 90 | if not charset: |
| 91 | return "" |
| 92 | charset = charset.strip().strip('"').strip("'").lower() |
| 93 | if not charset: |
| 94 | return "" |
| 95 | try: |
| 96 | return codecs.lookup(charset).name |
| 97 | except LookupError: |
| 98 | return "" |
| 99 | |
| 100 | |
| 101 | def _charset_from_headers(headers: dict) -> str: |
| 102 | content_type = headers.get("Content-Type") or headers.get("content-type") or "" |
| 103 | match = re.search(r"charset\s*=\s*([^;\s]+)", content_type, re.I) |
| 104 | return _normalize_charset(match.group(1)) if match else "" |
| 105 | |
| 106 | |
| 107 | def _charset_from_html(raw: bytes) -> str: |
| 108 | """Extract a charset declaration from the first chunk of HTML bytes.""" |
| 109 | head = raw[:8192] |
| 110 | patterns = [ |
| 111 | rb"<meta[^>]+charset=[\"']?\s*([a-zA-Z0-9_\-]+)", |
| 112 | rb"<meta[^>]+content=[\"'][^\"']*charset=\s*([a-zA-Z0-9_\-]+)", |
| 113 | ] |
| 114 | for pattern in patterns: |
| 115 | match = re.search(pattern, head, re.I) |
| 116 | if match: |
| 117 | return _normalize_charset(match.group(1).decode("ascii", "ignore")) |
| 118 | return "" |
| 119 | |
| 120 | |
| 121 | def _decode_quality_score(text: str) -> int: |
| 122 | """Score obvious decode artifacts; lower is better.""" |
| 123 | mojibake_markers = [ |
| 124 | "�", "锟", "Ã", "Â", "â€", "’", "“", "â€\x9d", |
| 125 | "琚", "佸", "鍦", "涓", "鏄", "寤", "骞", "鏈", "鏃", "鈥", |
| 126 | ] |
| 127 | marker_hits = sum(text.count(marker) for marker in mojibake_markers) |
| 128 | control_hits = sum(1 for ch in text if ord(ch) < 32 and ch not in "\t\n\r") |
| 129 | return marker_hits * 20 + control_hits * 10 + text.count("\ufffd") * 50 |
| 130 | |
| 131 | |
| 132 | def _decode_response_text(response) -> str: |
| 133 | """Decode HTTP response bytes without letting guessed encodings override declarations.""" |
| 134 | raw = response.content |
| 135 | declared = [ |
| 136 | _charset_from_headers(response.headers), |
| 137 | _charset_from_html(raw), |
| 138 | ] |
| 139 | if raw.startswith(codecs.BOM_UTF8): |
| 140 | declared.insert(0, "utf-8-sig") |
| 141 | |
| 142 | seen = set() |
| 143 | declared = [enc for enc in declared if enc and not (enc in seen or seen.add(enc))] |
| 144 | for enc in declared: |
| 145 | try: |
| 146 | return raw.decode(enc) |
| 147 | except UnicodeDecodeError: |
| 148 | continue |
| 149 | |
| 150 | candidates = [] |
| 151 | for enc in [ |
| 152 | getattr(response, "encoding", None), |
| 153 | getattr(response, "apparent_encoding", None), |
| 154 | "utf-8", |
| 155 | "gb18030", |
| 156 | "big5", |
| 157 | ]: |
| 158 | enc = _normalize_charset(enc) |
| 159 | if enc and enc not in candidates: |
| 160 | candidates.append(enc) |
| 161 | |
| 162 | decoded = [] |
| 163 | for enc in candidates: |
| 164 | try: |
| 165 | text = raw.decode(enc) |
| 166 | except UnicodeDecodeError: |
| 167 | continue |
| 168 | decoded.append((_decode_quality_score(text), enc, text)) |
| 169 | |
| 170 | if decoded: |
| 171 | decoded.sort(key=lambda item: item[0]) |
| 172 | return decoded[0][2] |
| 173 | |
| 174 | return raw.decode("utf-8", errors="replace") |
| 175 | |
| 176 | try: |
| 177 | from PIL import Image |
| 178 | PILLOW_AVAILABLE = True |
| 179 | except ImportError: |
| 180 | PILLOW_AVAILABLE = False |
| 181 | print("[WARN] Pillow not installed. WebP images will not be converted to PNG.") |
| 182 | print(" Run: pip install Pillow") |
| 183 | |
| 184 | # ============ Config ============ |
| 185 | CONFIG = { |
| 186 | "output_dir": "./projects", |
| 187 | "timeout": 30, |
| 188 | "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", |
| 189 | # Specific content identifiers often found in Chinese CMS (Gov/News) |
| 190 | "content_selectors": [ |
| 191 | {"class_": re.compile(r"tys-main-zt-show", re.I)}, |
| 192 | {"class_": re.compile(r"tys-main", re.I)}, |
| 193 | {"class_": "TRS_Editor"}, |
| 194 | {"class_": "TRS_UEDITOR"}, |
| 195 | {"class_": "ucontent"}, |
| 196 | {"class_": "article-content"}, |
| 197 | {"class_": "news-content"}, |
| 198 | {"class_": "detail-content"}, |
| 199 | {"class_": "content-text"}, |
| 200 | {"class_": "pages_content"}, |
| 201 | {"class_": "zwgk_content"}, |
| 202 | {"class_": "content_detail"}, |
| 203 | {"class_": "text_content"}, |
| 204 | {"class_": "main-content"}, |
| 205 | {"class_": "main_content"}, |
| 206 | {"class_": "view-content"}, |
| 207 | {"class_": "info-content"}, |
| 208 | {"id": "Zoom"}, |
| 209 | {"id": "content"}, |
| 210 | {"id": "article"}, |
| 211 | {"class_": "content"}, |
| 212 | {"name": "article"}, # tag name |
| 213 | {"name": "main"}, # tag name |
| 214 | ] |
| 215 | } |
| 216 | |
| 217 | |
| 218 | def fetch_url(url: str) -> str: |
| 219 | """Fetch a web page with explicit headers and encoding detection. |
| 220 | |
| 221 | Args: |
| 222 | url: Target URL. |
| 223 | |
| 224 | Returns: |
| 225 | The response body as text. |
| 226 | """ |
| 227 | headers = { |
| 228 | "User-Agent": CONFIG["user_agent"], |
| 229 | "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", |
| 230 | "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8" |
| 231 | } |
| 232 | |
| 233 | try: |
| 234 | response = _http_get(url, headers=headers, |
| 235 | timeout=CONFIG["timeout"], verify=False) |
| 236 | response.raise_for_status() |
| 237 | |
| 238 | return _decode_response_text(response) |
| 239 | except Exception as e: |
| 240 | raise Exception(f"Failed to fetch {url}: {str(e)}") |
| 241 | |
| 242 | |
| 243 | def clean_title(title: str) -> str: |
| 244 | """Remove common site suffixes from a title.""" |
| 245 | if not title: |
| 246 | return "" |
| 247 | # Remove site name suffixes often found in Chinese titles |
| 248 | clean = re.sub(r"[-_|].*?(政府|门户|网站|委员会).*$", "", title) |
| 249 | return clean.strip() |
| 250 | |
| 251 | |
| 252 | def sanitize_filename(name: str) -> str: |
| 253 | """Sanitize a string for filesystem-safe filenames.""" |
| 254 | # Replace whitespace with underscore first |
| 255 | clean = re.sub(r'\s+', '_', name) |
| 256 | # Remove all except Chinese, English, Numbers, Underscore |
| 257 | clean = re.sub(r'[^\u4e00-\u9fa5a-zA-Z0-9_]', '', clean) |
| 258 | # Collapse repeating underscores |
| 259 | clean = re.sub(r'_+', '_', clean) |
| 260 | return clean[:80] # Truncate |
| 261 | |
| 262 | |
| 263 | def derive_base_name(title: str, url: str) -> str: |
| 264 | """Derive a safe, non-empty basename from a title or URL.""" |
| 265 | base = sanitize_filename(title or "") |
| 266 | if base: |
| 267 | return base |
| 268 | |
| 269 | parsed = urlparse(url) |
| 270 | path = parsed.path.strip('/') |
| 271 | if path: |
| 272 | candidate = f"{parsed.netloc}_{path}" |
| 273 | else: |
| 274 | candidate = parsed.netloc or "untitled" |
| 275 | base = sanitize_filename(candidate) |
| 276 | if base: |
| 277 | return base |
| 278 | |
| 279 | ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") |
| 280 | return f"untitled_{ts}" |
| 281 | |
| 282 | |
| 283 | def build_image_filename(abs_url: str, seq: int, content_type: str | None = None) -> str: |
| 284 | """Build a safe image filename from URL metadata.""" |
| 285 | parsed = urlparse(abs_url) |
| 286 | basename = os.path.basename(parsed.path).split('?')[0] |
| 287 | stem, ext = os.path.splitext(basename) |
| 288 | if not ext or len(ext) > 5 or '/' in ext: |
| 289 | ext = "" |
| 290 | if not ext and content_type: |
| 291 | ctype = content_type.split(';')[0].lower() |
| 292 | ext_map = { |
| 293 | "image/jpeg": ".jpg", |
| 294 | "image/jpg": ".jpg", |
| 295 | "image/png": ".png", |
| 296 | "image/gif": ".gif", |
| 297 | "image/webp": ".webp", |
| 298 | } |
| 299 | ext = ext_map.get(ctype, "") |
| 300 | if not ext: |
| 301 | ext = ".jpg" |
| 302 | stem = sanitize_filename(stem) if stem else f"image_{seq}" |
| 303 | return f"{stem}{ext}" |
| 304 | |
| 305 | |
| 306 | def resolve_content_image_url(img: Tag, page_url: str) -> str | None: |
| 307 | """Resolve one content image, preferring real lazy-load URLs.""" |
| 308 | candidates = [ |
| 309 | img.get("data-src"), |
| 310 | img.get("data-original"), |
| 311 | img.get("data-lazy-src"), |
| 312 | img.get("data-actualsrc"), |
| 313 | img.get("src"), |
| 314 | ] |
| 315 | for value in candidates: |
| 316 | if not isinstance(value, str): |
| 317 | continue |
| 318 | src = value.strip() |
| 319 | if not src or src.startswith(("data:", "javascript:", "blob:", "#")): |
| 320 | continue |
| 321 | resolved = urljoin(page_url, src) |
| 322 | parsed = urlparse(resolved) |
| 323 | if parsed.scheme in {"http", "https"} and parsed.netloc: |
| 324 | img["src"] = resolved |
| 325 | return resolved |
| 326 | return None |
| 327 | |
| 328 | |
| 329 | def rewrite_images_to_remote_urls(content_element: Tag | None, page_url: str) -> int: |
| 330 | """Retain remote image links without downloading image bytes.""" |
| 331 | if content_element is None: |
| 332 | return 0 |
| 333 | return sum( |
| 334 | resolve_content_image_url(img, page_url) is not None |
| 335 | for img in content_element.find_all("img") |
| 336 | ) |
| 337 | |
| 338 | |
| 339 | def download_and_rewrite_images( |
| 340 | content_element: Tag | None, |
| 341 | page_url: str, |
| 342 | image_dir: str, |
| 343 | rel_prefix: str, |
| 344 | ) -> int: |
| 345 | """Download images under the main content node and rewrite `src` paths.""" |
| 346 | if content_element is None: |
| 347 | return 0 |
| 348 | images = list(content_element.find_all("img")) |
| 349 | if not images: |
| 350 | return 0 |
| 351 | |
| 352 | os.makedirs(image_dir, exist_ok=True) |
| 353 | downloaded = {} |
| 354 | manifest_by_filename: dict[str, dict[str, object]] = {} |
| 355 | saved = 0 |
| 356 | |
| 357 | for idx, img in enumerate(images): |
| 358 | abs_url = resolve_content_image_url(img, page_url) |
| 359 | if abs_url is None: |
| 360 | continue |
| 361 | content_type = "" |
| 362 | converted_from = "" |
| 363 | if abs_url in downloaded: |
| 364 | saved_name = downloaded[abs_url] |
| 365 | else: |
| 366 | try: |
| 367 | resp = _http_get( |
| 368 | abs_url, |
| 369 | headers={"User-Agent": CONFIG["user_agent"]}, |
| 370 | timeout=CONFIG["timeout"], |
| 371 | verify=False, |
| 372 | ) |
| 373 | resp.raise_for_status() |
| 374 | filename = build_image_filename( |
| 375 | abs_url, idx, resp.headers.get("Content-Type")) |
| 376 | |
| 377 | # Check if image is webp and convert to png |
| 378 | stem, ext = os.path.splitext(filename) |
| 379 | content_type = resp.headers.get("Content-Type", "").lower() |
| 380 | is_webp = ext.lower() == ".webp" or "webp" in content_type |
| 381 | |
| 382 | if is_webp and PILLOW_AVAILABLE: |
| 383 | # Convert webp to png (optimized) |
| 384 | try: |
| 385 | img_data = io.BytesIO(resp.content) |
| 386 | pil_image = Image.open(img_data) |
| 387 | |
| 388 | # Update filename to .png |
| 389 | converted_from = filename |
| 390 | filename = f"{stem}.png" |
| 391 | local_path = os.path.join(image_dir, filename) |
| 392 | |
| 393 | # Avoid accidental overwrites if filenames collide |
| 394 | counter = 1 |
| 395 | while os.path.exists(local_path): |
| 396 | local_path = os.path.join( |
| 397 | image_dir, f"{stem}_{counter}.png") |
| 398 | filename = os.path.basename(local_path) |
| 399 | counter += 1 |
| 400 | |
| 401 | # Save as PNG directly (Pillow auto-converts, no need for explicit mode conversion) |
| 402 | pil_image.save(local_path, 'PNG', optimize=False) |
| 403 | pil_image.close() |
| 404 | print(f" [INFO] Converted webp to png: {filename}") |
| 405 | except Exception as convert_err: |
| 406 | print( |
| 407 | f" [WARN] Failed to convert webp: {convert_err}, saving as-is") |
| 408 | local_path = os.path.join(image_dir, filename) |
| 409 | counter = 1 |
| 410 | stem, ext = os.path.splitext(filename) |
| 411 | while os.path.exists(local_path): |
| 412 | local_path = os.path.join( |
| 413 | image_dir, f"{stem}_{counter}{ext}") |
| 414 | filename = os.path.basename(local_path) |
| 415 | counter += 1 |
| 416 | with open(local_path, "wb") as f: |
| 417 | f.write(resp.content) |
| 418 | else: |
| 419 | local_path = os.path.join(image_dir, filename) |
| 420 | |
| 421 | # Avoid accidental overwrites if filenames collide |
| 422 | counter = 1 |
| 423 | stem, ext = os.path.splitext(filename) |
| 424 | while os.path.exists(local_path): |
| 425 | local_path = os.path.join( |
| 426 | image_dir, f"{stem}_{counter}{ext}") |
| 427 | filename = os.path.basename(local_path) |
| 428 | counter += 1 |
| 429 | |
| 430 | with open(local_path, "wb") as f: |
| 431 | f.write(resp.content) |
| 432 | downloaded[abs_url] = filename |
| 433 | saved_name = filename |
| 434 | manifest_by_filename[saved_name] = { |
| 435 | "index": len(manifest_by_filename) + 1, |
| 436 | "filename": saved_name, |
| 437 | "original_filename": converted_from or saved_name, |
| 438 | "asset_kind": "bitmap", |
| 439 | "svg_renderable": True, |
| 440 | "pptx_native_supported": True, |
| 441 | "source_kind": "web_image", |
| 442 | "source_url": abs_url, |
| 443 | "source_page_url": page_url, |
| 444 | "content_type": content_type.split(";")[0] if content_type else "", |
| 445 | "occurrences": [], |
| 446 | } |
| 447 | saved += 1 |
| 448 | except Exception as e: |
| 449 | print(f" [WARN] Skip image {abs_url}: {e}") |
| 450 | continue |
| 451 | |
| 452 | rel_path = os.path.join( |
| 453 | rel_prefix, saved_name) if rel_prefix else saved_name |
| 454 | img["src"] = rel_path |
| 455 | manifest_item = manifest_by_filename.get(saved_name) |
| 456 | if manifest_item is not None: |
| 457 | occurrences = manifest_item.setdefault("occurrences", []) |
| 458 | if isinstance(occurrences, list): |
| 459 | occurrences.append({ |
| 460 | "occurrence_index": idx + 1, |
| 461 | "source_url": abs_url, |
| 462 | "alt_text": img.get("alt", ""), |
| 463 | }) |
| 464 | manifest_item["usage_count"] = len(occurrences) |
| 465 | |
| 466 | if manifest_by_filename: |
| 467 | manifest_path = os.path.join(image_dir, "image_manifest.json") |
| 468 | with open(manifest_path, "w", encoding="utf-8") as f: |
| 469 | json.dump( |
| 470 | list(manifest_by_filename.values()), |
| 471 | f, |
| 472 | ensure_ascii=False, |
| 473 | indent=2, |
| 474 | ) |
| 475 | f.write("\n") |
| 476 | |
| 477 | return saved |
| 478 | |
| 479 | |
| 480 | def extract_metadata(soup: BeautifulSoup, url: str) -> dict[str, str]: |
| 481 | """Extract page metadata such as title, date, description, and author.""" |
| 482 | |
| 483 | # 1. Title |
| 484 | title_tag = soup.title |
| 485 | title = clean_title(title_tag.string if title_tag else "") |
| 486 | |
| 487 | # 2. Meta tags |
| 488 | metas = {} |
| 489 | for meta in soup.find_all("meta"): |
| 490 | name = meta.get("name") or meta.get("property") |
| 491 | content = meta.get("content") |
| 492 | if name and content: |
| 493 | metas[name.lower()] = content.strip() |
| 494 | |
| 495 | # 3. Date Extraction Strategies |
| 496 | date = ( |
| 497 | metas.get("article:published_time") or |
| 498 | metas.get("og:published_time") or |
| 499 | metas.get("pubdate") or |
| 500 | metas.get("publishdate") or |
| 501 | metas.get("date") |
| 502 | ) |
| 503 | |
| 504 | if not date: |
| 505 | # Try matching date patterns in the text |
| 506 | text_content = soup.get_text() |
| 507 | date_patterns = [ |
| 508 | r"发布[时日]间[：:]\s*(\d{4}[-\/年]\d{1,2}[-\/月]\d{1,2}[日]?)", |
| 509 | r"日期[：:]\s*(\d{4}[-\/年]\d{1,2}[-\/月]\d{1,2}[日]?)", |
| 510 | r"(\d{4}[-\/年]\d{1,2}[-\/月]\d{1,2}[日]?)\s*(?:发布|来源)", |
| 511 | r"时间[：:]\s*(\d{4}[-\/]\d{1,2}[-\/]\d{1,2})" |
| 512 | ] |
| 513 | for pattern in date_patterns: |
| 514 | match = re.search(pattern, text_content) |
| 515 | if match: |
| 516 | date = match.group(1).replace( |
| 517 | "年", "-").replace("月", "-").replace("日", "") |
| 518 | break |
| 519 | |
| 520 | if not date: |
| 521 | # Try URL matching |
| 522 | match = re.search(r"(\d{4})(\d{2})[\/_](?:t\d+_)?", url) |
| 523 | if match: |
| 524 | date = f"{match.group(1)}-{match.group(2)}" |
| 525 | else: |
| 526 | match = re.search(r"(\d{4})[-\/](\d{2})[-\/](\d{2})", url) |
| 527 | if match: |
| 528 | date = f"{match.group(1)}-{match.group(2)}-{match.group(3)}" |
| 529 | |
| 530 | # 4. Description |
| 531 | description = ( |
| 532 | metas.get("description") or |
| 533 | metas.get("og:description") or |
| 534 | metas.get("twitter:description") or |
| 535 | "" |
| 536 | ) |
| 537 | |
| 538 | # 5. Author/Source |
| 539 | author = metas.get("author") or metas.get("article:author") |
| 540 | if not author: |
| 541 | # Try common patterns |
| 542 | source_patterns = [ |
| 543 | r"来源[：:]\s*([^\s<]+)", |
| 544 | r"发布(?:单位|机构)[：:]\s*([^\s<]+)" |
| 545 | ] |
| 546 | for pattern in source_patterns: |
| 547 | match = re.search(pattern, soup.get_text()) |
| 548 | if match: |
| 549 | author = match.group(1) |
| 550 | break |
| 551 | |
| 552 | return { |
| 553 | "title": title or metas.get("og:title") or "Untitled", |
| 554 | "date": date or "", |
| 555 | "description": description, |
| 556 | "author": author or "", |
| 557 | "source_url": url |
| 558 | } |
| 559 | |
| 560 | |
| 561 | def find_main_content(soup: BeautifulSoup) -> Tag | None: |
| 562 | """Find the most likely main content container in a page.""" |
| 563 | # 1. Clean up first (remove known clutter) |
| 564 | for tag in soup(["script", "style", "nav", "header", "footer", "aside", "noscript", "iframe"]): |
| 565 | tag.decompose() |
| 566 | |
| 567 | best_element = None |
| 568 | max_score = 0 |
| 569 | |
| 570 | # 2. Strategy A: Check specific classes/ids |
| 571 | for selector in CONFIG["content_selectors"]: |
| 572 | if "name" in selector: |
| 573 | # Tag name match (article, main) |
| 574 | elements = soup.find_all(selector["name"]) |
| 575 | else: |
| 576 | # Class or ID match |
| 577 | elements = soup.find_all(attrs=selector) |
| 578 | |
| 579 | for el in elements: |
| 580 | # Score based on text length and chinese character count |
| 581 | text = el.get_text(strip=True) |
| 582 | length = len(text) |
| 583 | if length < 100: |
| 584 | continue |
| 585 | |
| 586 | chinese_count = len(re.findall(r'[\u4e00-\u9fa5]', text)) |
| 587 | score = length + (chinese_count * 2) |
| 588 | |
| 589 | if score > max_score: |
| 590 | max_score = score |
| 591 | best_element = el |
| 592 | |
| 593 | # 3. Strategy B: If no specific container found, look for dense text areas with paragraphs |
| 594 | if not best_element or max_score < 200: |
| 595 | for div in soup.find_all("div"): |
| 596 | p_count = len(div.find_all("p", recursive=False)) |
| 597 | # recursive=False ensures we don't just pick the top-level body by accident |
| 598 | # but sometimes content is nested deep |
| 599 | if p_count == 0: |
| 600 | # Check if it has lots of text even without p tags (br tags?) |
| 601 | pass |
| 602 | |
| 603 | text = div.get_text(strip=True) |
| 604 | if len(text) > 200 and p_count >= 1: |
| 605 | # Recalculate deep score |
| 606 | chinese_count = len(re.findall(r'[\u4e00-\u9fa5]', text)) |
| 607 | score = len(text) + (chinese_count * 2) + (p_count * 50) |
| 608 | if score > max_score: |
| 609 | max_score = score |
| 610 | best_element = div |
| 611 | |
| 612 | # Fallback to body |
| 613 | return best_element if best_element else soup.body |
| 614 | |
| 615 | |
| 616 | def element_to_markdown(element: Tag | NavigableString | None) -> str: |
| 617 | """Recursively convert a BeautifulSoup node to Markdown.""" |
| 618 | if element is None: |
| 619 | return "" |
| 620 | |
| 621 | if isinstance(element, NavigableString): |
| 622 | text = str(element).strip() |
| 623 | return text if text else "" |
| 624 | |
| 625 | tag_name = element.name.lower() |
| 626 | |
| 627 | # Skip hidden/unwanted tags |
| 628 | if tag_name in ['script', 'style', 'meta', 'link', 'input', 'button', 'select']: |
| 629 | return "" |
| 630 | |
| 631 | content = "" |
| 632 | for child in element.children: |
| 633 | content += element_to_markdown(child) |
| 634 | # Add spacing logic here if needed, but usually block elements handle it |
| 635 | |
| 636 | # Block handlers |
| 637 | if tag_name in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6']: |
| 638 | level = int(tag_name[1]) |
| 639 | return f"\n{'#' * level} {content}\n\n" |
| 640 | |
| 641 | elif tag_name == 'p': |
| 642 | # Clean up internal whitespace |
| 643 | content = re.sub(r'\s+', ' ', content).strip() |
| 644 | return f"\n{content}\n\n" if content else "" |
| 645 | |
| 646 | elif tag_name == 'br': |
| 647 | return " \n" |
| 648 | |
| 649 | elif tag_name == 'hr': |
| 650 | return "\n---\n" |
| 651 | |
| 652 | elif tag_name == 'div': |
| 653 | return f"\n{content}\n" |
| 654 | |
| 655 | elif tag_name == 'blockquote': |
| 656 | lines = content.strip().split('\n') |
| 657 | quoted = '\n'.join([f"> {line}" for line in lines if line.strip()]) |
| 658 | return f"\n{quoted}\n\n" |
| 659 | |
| 660 | elif tag_name in ['ul', 'ol']: |
| 661 | # This is tricky without "state" (knowing we are in a list) |
| 662 | # For simplicity in this recursive version, we rely on LI handling |
| 663 | return f"\n{content}\n" |
| 664 | |
| 665 | elif tag_name == 'li': |
| 666 | # Simple list handling |
| 667 | clean_content = content.strip() |
| 668 | return f"- {clean_content}\n" |
| 669 | |
| 670 | elif tag_name == 'pre': |
| 671 | return f"\n```\n{content}\n```\n\n" |
| 672 | |
| 673 | elif tag_name == 'code': |
| 674 | # If parent is pre, handle in pre. If inline: |
| 675 | parent = element.parent |
| 676 | if parent and parent.name == 'pre': |
| 677 | return content |
| 678 | return f"`{content}`" |
| 679 | |
| 680 | elif tag_name == 'a': |
| 681 | href = element.get('href', '') |
| 682 | if href and not href.startswith('javascript:'): |
| 683 | return f"[{content}]({href})" |
| 684 | return content |
| 685 | |
| 686 | elif tag_name == 'img': |
| 687 | src = element.get('src', '') |
| 688 | alt = element.get('alt', '') |
| 689 | if src: |
| 690 | return f"" |
| 691 | return "" |
| 692 | |
| 693 | elif tag_name == 'table': |
| 694 | # Basic table text extraction, full markdown table support is complex |
| 695 | # Leaving as raw text or simplistic conversion for now |
| 696 | # Ideally, we'd parse TRs and TDs |
| 697 | return f"\n{content}\n" |
| 698 | |
| 699 | elif tag_name == 'tr': |
| 700 | return f"{content}|\n" |
| 701 | |
| 702 | elif tag_name in ['td', 'th']: |
| 703 | return f"| {content.strip()} " |
| 704 | |
| 705 | # Style formatting |
| 706 | elif tag_name in ['strong', 'b']: |
| 707 | return f"**{content}**" |
| 708 | elif tag_name in ['em', 'i']: |
| 709 | return f"*{content}*" |
| 710 | elif tag_name in ['del', 's', 'strike']: |
| 711 | return f"~~{content}~~" |
| 712 | |
| 713 | # Default for span, section, etc. |
| 714 | return f"{content} " |
| 715 | |
| 716 | |
| 717 | def simple_html_to_markdown_traversal(soup: Tag | BeautifulSoup | None) -> str: |
| 718 | """Convert HTML content to Markdown using BeautifulSoup traversal.""" |
| 719 | lines = [] |
| 720 | |
| 721 | def traverse(node: Tag | NavigableString) -> str: |
| 722 | if isinstance(node, NavigableString): |
| 723 | text = str(node) |
| 724 | # Normalize whitespace but keep single spaces |
| 725 | text = re.sub(r'\s+', ' ', text) |
| 726 | if text.strip(): |
| 727 | return text |
| 728 | return "" |
| 729 | |
| 730 | if node.name in ['script', 'style', 'comment', 'meta', 'link']: |
| 731 | return "" |
| 732 | |
| 733 | # Handle Block Elements |
| 734 | is_block = node.name in ['p', 'div', 'h1', 'h2', 'h3', 'h4', |
| 735 | 'h5', 'h6', 'li', 'blockquote', 'pre', 'hr', 'table', 'tr'] |
| 736 | |
| 737 | # Pre-processing |
| 738 | prefix = "" |
| 739 | suffix = "" |
| 740 | |
| 741 | if node.name in ['h1', 'h2', 'h3', 'h4', 'h5', 'h6']: |
| 742 | level = int(node.name[1]) |
| 743 | prefix = f"\n\n{'#' * level} " |
| 744 | suffix = "\n\n" |
| 745 | elif node.name == 'p': |
| 746 | prefix = "\n\n" |
| 747 | suffix = "\n\n" |
| 748 | elif node.name == 'li': |
| 749 | prefix = "\n- " |
| 750 | elif node.name == 'blockquote': |
| 751 | prefix = "\n> " |
| 752 | suffix = "\n" |
| 753 | elif node.name == 'hr': |
| 754 | return "\n\n---\n\n" |
| 755 | elif node.name == 'br': |
| 756 | return " \n" |
| 757 | elif node.name == 'pre': |
| 758 | # Extract raw text from pre to preserve formatting |
| 759 | return f"\n\n```\n{node.get_text()}\n```\n\n" |
| 760 | |
| 761 | # Inline formatting |
| 762 | if node.name in ['strong', 'b']: |
| 763 | prefix, suffix = "**", "**" |
| 764 | elif node.name in ['em', 'i']: |
| 765 | prefix, suffix = "*", "*" |
| 766 | elif node.name == 'code' and node.parent.name != 'pre': |
| 767 | prefix, suffix = "`", "`" |
| 768 | elif node.name == 'a': |
| 769 | href = node.get('href') |
| 770 | if href and not href.startswith('javascript:'): |
| 771 | prefix = "[" |
| 772 | suffix = f"]({href})" |
| 773 | else: |
| 774 | prefix, suffix = "", "" |
| 775 | elif node.name == 'img': |
| 776 | src = node.get('src') |
| 777 | alt = node.get('alt', '') |
| 778 | if src: |
| 779 | return f"" |
| 780 | return "" |
| 781 | |
| 782 | # Recurse |
| 783 | inner_text = "" |
| 784 | for child in node.children: |
| 785 | res = traverse(child) |
| 786 | if res: |
| 787 | inner_text += res |
| 788 | |
| 789 | # Post-processing for tables (simplified) |
| 790 | if node.name == 'tr': |
| 791 | # count tds |
| 792 | cells = [c.get_text(strip=True) for c in node.find_all( |
| 793 | ['td', 'th'], recursive=False)] |
| 794 | return f"| {' | '.join(cells)} |\n" |
| 795 | if node.name == 'table': |
| 796 | # Try to add a separator line after first row if it looks like a header |
| 797 | rows = inner_text.strip().split('\n') |
| 798 | if rows: |
| 799 | cols_count = rows[0].count('|') - 1 |
| 800 | if cols_count > 0: |
| 801 | # rough approx |
| 802 | sep = "| " + " | ".join(["---"] * int(cols_count/2)) + " |" |
| 803 | # Actually, the traverse of TR returns newline terminated strings. |
| 804 | # Let's just return what we gathered. |
| 805 | pass |
| 806 | return f"\n\n{inner_text}\n\n" |
| 807 | |
| 808 | return f"{prefix}{inner_text}{suffix}" |
| 809 | |
| 810 | # Actually, a simpler approach for this script is "just get string" but with markers? |
| 811 | # Let's use a simplified approach: use get_text but with 'separator' logic? |
| 812 | # text = soup.get_text(separator='\n\n') |
| 813 | # But that loses links and boldness. |
| 814 | |
| 815 | # Recommendation: Let's stick to the traversal above which constructs a string. |
| 816 | md = traverse(soup) |
| 817 | |
| 818 | # Cleanup Markdown |
| 819 | if md: |
| 820 | # Remove excessive newlines |
| 821 | md = re.sub(r'\n{3,}', '\n\n', md) |
| 822 | md = md.strip() |
| 823 | return md or "" |
| 824 | |
| 825 | |
| 826 | def process_url( |
| 827 | url: str, |
| 828 | output_file: str | None = None, |
| 829 | *, |
| 830 | download_images: bool = True, |
| 831 | ) -> tuple[bool, str, str | None, str | None]: |
| 832 | """Fetch, convert, and save one web page as Markdown. |
| 833 | |
| 834 | Returns (success, url, error, output_path). output_path is the actual saved |
| 835 | Markdown path (derived from the article title when no output_file is given), |
| 836 | so a caller can locate a title-named file it did not choose upfront. |
| 837 | """ |
| 838 | print(f"\n[Fetching] {url}") |
| 839 | try: |
| 840 | html = fetch_url(url) |
| 841 | soup = BeautifulSoup(html, 'html.parser') |
| 842 | |
| 843 | # Extract Metadata |
| 844 | metadata = extract_metadata(soup, url) |
| 845 | print(f" [OK] Title: {metadata['title']}") |
| 846 | if metadata['date']: |
| 847 | print(f" [OK] Date: {metadata['date']}") |
| 848 | |
| 849 | # Determine output path and image directory upfront |
| 850 | if output_file: |
| 851 | output_path = output_file |
| 852 | else: |
| 853 | base_name = derive_base_name(metadata['title'], url) |
| 854 | filename = f"{base_name}.md" |
| 855 | output_path = os.path.join(CONFIG["output_dir"], filename) |
| 856 | |
| 857 | output_dirname = os.path.dirname(output_path) or "." |
| 858 | os.makedirs(output_dirname, exist_ok=True) |
| 859 | base_name = os.path.splitext(os.path.basename(output_path))[0] |
| 860 | image_dir = os.path.join(output_dirname, f"{base_name}_files") |
| 861 | rel_image_prefix = os.path.relpath(image_dir, output_dirname) |
| 862 | |
| 863 | # Extract Content |
| 864 | content_div = find_main_content(soup) |
| 865 | |
| 866 | # Download images and rewrite src before markdown conversion |
| 867 | image_count = 0 |
| 868 | if download_images: |
| 869 | image_count = download_and_rewrite_images( |
| 870 | content_div, url, image_dir, rel_image_prefix) |
| 871 | else: |
| 872 | rewrite_images_to_remote_urls(content_div, url) |
| 873 | if image_count: |
| 874 | print(f" [OK] Images: {image_count} saved to {image_dir}") |
| 875 | |
| 876 | # Convert to MD |
| 877 | # Note: We pass the element to our traversal function |
| 878 | markdown_text = simple_html_to_markdown_traversal(content_div) |
| 879 | print(f" [OK] Content: {len(markdown_text)} chars") |
| 880 | |
| 881 | # Construct content |
| 882 | final_output = [] |
| 883 | final_output.append("<!--") |
| 884 | final_output.append(f" Source: {url}") |
| 885 | final_output.append( |
| 886 | f" Crawled: {datetime.datetime.now().isoformat()}") |
| 887 | if metadata['date']: |
| 888 | final_output.append(f" Published: {metadata['date']}") |
| 889 | if metadata['author']: |
| 890 | final_output.append(f" Author: {metadata['author']}") |
| 891 | final_output.append("-->\n") |
| 892 | |
| 893 | if metadata['title']: |
| 894 | final_output.append(f"# {metadata['title']}\n") |
| 895 | |
| 896 | if metadata['description']: |
| 897 | final_output.append(f"> {metadata['description']}\n") |
| 898 | |
| 899 | final_output.append(markdown_text) |
| 900 | |
| 901 | full_content = "\n".join(final_output) |
| 902 | |
| 903 | with open(output_path, 'w', encoding='utf-8') as f: |
| 904 | f.write(full_content) |
| 905 | profile_path = write_conversion_profile_best_effort( |
| 906 | input_path=url, |
| 907 | markdown_path=output_path, |
| 908 | converter="web_to_md.py", |
| 909 | conversion_type="web", |
| 910 | asset_dir=image_dir if image_count else None, |
| 911 | ) |
| 912 | |
| 913 | print(f" [OK] Saved: {output_path}") |
| 914 | if profile_path: |
| 915 | print(f" [OK] Conversion profile: {profile_path}") |
| 916 | return True, url, None, output_path |
| 917 | |
| 918 | except Exception as e: |
| 919 | print(f" [ERROR] {str(e)}") |
| 920 | return False, url, str(e), None |
| 921 | |
| 922 | |
| 923 | def _write_emit_result(result_file: str, url: str, markdown_path: str) -> None: |
| 924 | """Write the actual saved path as JSON so a caller can locate the output.""" |
| 925 | md = Path(markdown_path).resolve() |
| 926 | profile = profile_path_for(md) |
| 927 | payload = { |
| 928 | "input": url, |
| 929 | "markdown": str(md), |
| 930 | "conversion_profile": str(profile) if profile.is_file() else "", |
| 931 | } |
| 932 | try: |
| 933 | Path(result_file).write_text( |
| 934 | json.dumps(payload, ensure_ascii=False), encoding="utf-8") |
| 935 | except OSError as exc: |
| 936 | print(f" [WARN] Could not write --emit-result: {exc}") |
| 937 | |
| 938 | |
| 939 | def main(argv: list[str] | None = None) -> int: |
| 940 | """Run the CLI entry point.""" |
| 941 | parser = argparse.ArgumentParser( |
| 942 | description="Web to Markdown Converter (Python)") |
| 943 | parser.add_argument("urls", nargs="*", help="URLs to process") |
| 944 | parser.add_argument( |
| 945 | "-f", "--file", help="File containing URLs (one per line)") |
| 946 | parser.add_argument("-o", "--output", help="Output file (single URL only)") |
| 947 | parser.add_argument("-d", "--dir", help="Output directory") |
| 948 | parser.add_argument( |
| 949 | "--emit-result", |
| 950 | help="On success, write the saved output path as JSON to this file " |
| 951 | "(single-URL dispatcher use, so a title-named file can be located)") |
| 952 | parser.add_argument( |
| 953 | "--no-images", |
| 954 | action="store_true", |
| 955 | help="Keep remote image links without downloading image files", |
| 956 | ) |
| 957 | |
| 958 | args = parser.parse_args(argv) |
| 959 | |
| 960 | if args.dir: |
| 961 | CONFIG["output_dir"] = args.dir |
| 962 | |
| 963 | targets = [] |
| 964 | if args.urls: |
| 965 | targets.extend(args.urls) |
| 966 | |
| 967 | if args.file: |
| 968 | if os.path.exists(args.file): |
| 969 | with open(args.file, 'r', encoding='utf-8') as f: |
| 970 | lines = [l.strip() for l in f if l.strip() |
| 971 | and not l.strip().startswith("#")] |
| 972 | targets.extend(lines) |
| 973 | else: |
| 974 | print(f"Error: File {args.file} not found", file=sys.stderr) |
| 975 | return 1 |
| 976 | |
| 977 | if not targets: |
| 978 | parser.print_usage(sys.stderr) |
| 979 | print( |
| 980 | "web_to_md.py: error: at least one URL or --file is required", |
| 981 | file=sys.stderr, |
| 982 | ) |
| 983 | return 2 |
| 984 | |
| 985 | results = [] |
| 986 | for i, url in enumerate(targets): |
| 987 | # Allow specific output file only if 1 URL |
| 988 | out = args.output if (len(targets) == 1 and args.output) else None |
| 989 | success, url, err, out_path = process_url( |
| 990 | url, |
| 991 | out, |
| 992 | download_images=not args.no_images, |
| 993 | ) |
| 994 | results.append((success, url, err)) |
| 995 | if args.emit_result and success and out_path: |
| 996 | _write_emit_result(args.emit_result, url, out_path) |
| 997 | |
| 998 | # Summary |
| 999 | success_count = sum(1 for r in results if r[0]) |
| 1000 | fail_count = len(results) - success_count |
| 1001 | |
| 1002 | print("\n" + "="*50) |
| 1003 | print( |
| 1004 | f"[Done] Success: {success_count}/{len(results)}, Failed: {fail_count}") |
| 1005 | |
| 1006 | if fail_count > 0: |
| 1007 | print("\n[Failed URLs]:") |
| 1008 | for r in results: |
| 1009 | if not r[0]: |
| 1010 | print(f" - {r[1]}: {r[2]}") |
| 1011 | return 1 |
| 1012 | return 0 |
| 1013 | |
| 1014 | |
| 1015 | if __name__ == "__main__": |
| 1016 | # Disable warnings for verify=False if needed, though often useful to see |
| 1017 | import urllib3 |
| 1018 | urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) |
| 1019 | raise SystemExit(main()) |
| 1020 |