Read

#!/usr/bin/env python3
"""Fetch Feishu/Lark document as Markdown via Feishu Open API.

Special thanks to joeseesun for the excellent qiaomu-markdown-proxy project,
which inspired the Feishu API integration and document parsing approach here.
https://github.com/joeseesun/qiaomu-markdown-proxy

Requirements:
    pip install requests

Setup:
    export FEISHU_APP_ID=your_app_id
    export FEISHU_APP_SECRET=your_app_secret
    App needs: docx:document:readonly, wiki:wiki:readonly

Usage:
    python3 fetch_feishu.py <feishu_url>
    python3 fetch_feishu.py <feishu_url> --json
"""

import sys
import json
import os
import re
import urllib.parse

try:
    import requests
except ImportError:
    print("Error: requests not installed. Run: pip install requests", file=sys.stderr)
    sys.exit(1)

API = "https://open.feishu.cn/open-apis"
TIMEOUT = 20


def yaml_string(value):
    return json.dumps("" if value is None else str(value), ensure_ascii=False)


def get_token():
    app_id = os.environ.get("FEISHU_APP_ID")
    app_secret = os.environ.get("FEISHU_APP_SECRET")
    if not app_id or not app_secret:
        return None, "FEISHU_APP_ID or FEISHU_APP_SECRET not set"
    resp = requests.post(f"{API}/auth/v3/tenant_access_token/internal",
                         json={"app_id": app_id, "app_secret": app_secret},
                         timeout=TIMEOUT)
    d = resp.json()
    if d.get("code") != 0:
        return None, f"Auth failed: {d.get('msg', resp.text)}"
    return d["tenant_access_token"], None


def parse_url(url):
    patterns = [
        (r"feishu\.cn/docx/([A-Za-z0-9]+)", "docx"),
        (r"feishu\.cn/docs/([A-Za-z0-9]+)", "legacy_doc"),
        (r"feishu\.cn/wiki/([A-Za-z0-9]+)", "wiki"),
        (r"larksuite\.com/docx/([A-Za-z0-9]+)", "docx"),
        (r"larksuite\.com/docs/([A-Za-z0-9]+)", "legacy_doc"),
        (r"larksuite\.com/wiki/([A-Za-z0-9]+)", "wiki"),
    ]
    for pattern, doc_type in patterns:
        m = re.search(pattern, url)
        if m:
            return m.group(1), doc_type
    return url, "docx"


def resolve_wiki(token, wiki_token):
    resp = requests.get(f"{API}/wiki/v2/spaces/get_node",
                        headers={"Authorization": f"Bearer {token}"},
                        params={"token": wiki_token},
                        timeout=TIMEOUT)
    d = resp.json()
    if d.get("code") == 0:
        node = d["data"]["node"]
        return node.get("obj_token"), node.get("obj_type")
    return None, None


def get_blocks(token, doc_id):
    blocks, page_token = [], None
    while True:
        params = {"page_size": 500}
        if page_token:
            params["page_token"] = page_token
        resp = requests.get(f"{API}/docx/v1/documents/{doc_id}/blocks",
                            headers={"Authorization": f"Bearer {token}"},
                            params=params,
                            timeout=TIMEOUT)
        d = resp.json()
        if d.get("code") != 0:
            return None, f"Blocks fetch failed: {d.get('msg', resp.text)}"
        blocks.extend(d["data"].get("items", []))
        if not d["data"].get("has_more"):
            break
        page_token = d["data"].get("page_token")
    return blocks, None


def extract_text(elements):
    if not elements:
        return ""
    parts = []
    for el in elements:
        if "text_run" in el:
            tr = el["text_run"]
            text = tr.get("content", "")
            s = tr.get("text_element_style", {})
            if s.get("bold"):        text = f"**{text}**"
            if s.get("italic"):      text = f"*{text}*"
            if s.get("inline_code"): text = f"`{text}`"
            if s.get("link", {}).get("url"):
                text = f"[{text}]({urllib.parse.unquote(s['link']['url'])})"
            parts.append(text)
        elif "mention_user" in el:
            parts.append(f"@{el['mention_user'].get('user_id', 'user')}")
        elif "equation" in el:
            parts.append(f"${el['equation'].get('content', '')}$")
    return "".join(parts)


LANG_MAP = {
    7: "bash", 8: "c", 9: "csharp", 10: "cpp", 14: "css", 19: "dockerfile",
    25: "go", 29: "html", 31: "java", 32: "javascript", 33: "json",
    35: "kotlin", 40: "markdown", 46: "php", 50: "python", 52: "ruby",
    53: "rust", 58: "sql", 59: "swift", 62: "typescript", 68: "xml", 69: "yaml",
}


def blocks_to_md(blocks):
    lines = []
    counters = {}
    for block in blocks:
        bt = block.get("block_type")
        pid = block.get("parent_id", "")

        if bt == 2:
            text = extract_text(block.get("text", {}).get("elements", []))
            lines.append(text if text.strip() else "")
        elif bt in range(3, 10):
            level = bt - 2
            key = f"heading{level}"
            data = block.get(key) or block.get("heading", {})
            text = extract_text(data.get("elements", []))
            lines.append(f"{'#' * min(level, 6)} {text}")
        elif bt == 10:
            text = extract_text(block.get("bullet", {}).get("elements", []))
            lines.append(f"- {text}")
        elif bt == 11:
            text = extract_text(block.get("ordered", {}).get("elements", []))
            n = counters.get(pid, 0) + 1
            counters[pid] = n
            lines.append(f"{n}. {text}")
        elif bt == 12:
            code_data = block.get("code", {})
            text = extract_text(code_data.get("elements", []))
            lang = LANG_MAP.get(code_data.get("style", {}).get("language", 0), "")
            lines.extend([f"```{lang}", text, "```"])
        elif bt == 13:
            text = extract_text(block.get("quote", {}).get("elements", []))
            lines.append(f"> {text}")
        elif bt == 15:
            todo_data = block.get("todo", {})
            text = extract_text(todo_data.get("elements", []))
            done = todo_data.get("style", {}).get("done", False)
            lines.append(f"- {'[x]' if done else '[ ]'} {text}")
        elif bt == 16:
            lines.append("---")
        elif bt == 17:
            tok = block.get("image", {}).get("token", "")
            lines.append(f"![image](feishu-image://{tok})")
        elif bt == 1:
            pass
        else:
            for key, val in block.items():
                if isinstance(val, dict) and "elements" in val:
                    text = extract_text(val["elements"])
                    if text.strip():
                        lines.append(text)
                    break

    return "\n\n".join(lines)


def fetch_feishu(url):
    doc_id, doc_type = parse_url(url)

    if doc_type == "legacy_doc":
        return {
            "error": (
                "Legacy Feishu /docs/ pages are not supported by this script. "
                "Convert the document to docx first, or use a public-page fallback if the page is accessible without the API."
            )
        }

    token, err = get_token()
    if err:
        return {"error": err}

    if doc_type == "wiki":
        real_id, real_type = resolve_wiki(token, doc_id)
        if not real_id:
            return {"error": f"Cannot resolve wiki node: {doc_id}"}
        doc_id, doc_type = real_id, real_type or "docx"

    info_resp = requests.get(f"{API}/docx/v1/documents/{doc_id}",
                             headers={"Authorization": f"Bearer {token}"},
                             timeout=TIMEOUT)
    doc_info = (info_resp.json().get("data") or {}).get("document") or {}
    title = doc_info.get("title", "")

    blocks, err = get_blocks(token, doc_id)
    if err:
        return {"error": err}

    return {"title": title, "document_id": doc_id, "url": url, "content": blocks_to_md(blocks)}


def to_markdown(r):
    if "error" in r:
        return f"Error: {r['error']}"
    parts = [
        "---",
        f"title: {yaml_string(r.get('title', ''))}",
        f"document_id: {yaml_string(r.get('document_id', ''))}",
        f"url: {yaml_string(r.get('url', ''))}",
        "---",
        "",
        f"# {r['title']}" if r.get("title") else "",
        "",
        r.get("content", ""),
    ]
    return "\n".join(parts)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: fetch_feishu.py <feishu_url> [--json]", file=sys.stderr)
        print("  Requires: FEISHU_APP_ID, FEISHU_APP_SECRET", file=sys.stderr)
        sys.exit(1)

    result = fetch_feishu(sys.argv[1])
    if "--json" in sys.argv:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(to_markdown(result))
    if "error" in result:
        sys.exit(1)

#!/usr/bin/env python3
"""Local URL → Markdown extractor. Privacy-preserving tier 1 for fetch.sh.

Two paths, picked at runtime:

1. **Best path**: `readability-lxml` + `html2text` installed. Extract main
   content via readability scoring, convert to clean Markdown. Quality close
   to defuddle.md / r.jina.ai for static pages.

2. **Fallback path**: stdlib only. Strip HTML tags, collapse whitespace, drop
   <script>/<style>/<nav>/<footer>. Quality is poor for JS-heavy pages and
   complex layouts, but works for simple article-style HTML without any
   third-party dependency.

JS-rendered pages (X/Twitter, paywalled news, SPA) are out of reach for both
paths. Use `fetch.sh --use-proxy` for those.

Exit codes:
  0  success, markdown on stdout
  1  fetch or extraction failed; reason on stderr
  2  invocation error (missing URL)
"""

from __future__ import annotations

import argparse
import re
import sys
import urllib.error
import urllib.request
from html.parser import HTMLParser


USER_AGENT = (
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 12_0) "
    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120 Safari/537.36"
)
FETCH_TIMEOUT_SECS = 20


def fetch_html(url: str) -> str:
    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT})
    with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT_SECS) as resp:
        raw = resp.read()
    # Detect charset from Content-Type header; fall back to utf-8 with replace.
    charset = "utf-8"
    ctype = resp.headers.get("Content-Type", "")
    m = re.search(r"charset=([\w\-]+)", ctype, re.IGNORECASE)
    if m:
        charset = m.group(1).lower()
    try:
        return raw.decode(charset, errors="replace")
    except LookupError:
        return raw.decode("utf-8", errors="replace")


def extract_with_readability(html: str, url: str) -> str | None:
    """Best path: readability-lxml + html2text. Returns Markdown or None if
    deps missing. Raises only on genuine extraction failure."""
    try:
        from readability import Document  # type: ignore
        import html2text  # type: ignore
    except ImportError:
        return None
    doc = Document(html)
    cleaned_html = doc.summary(html_partial=True)
    title = (doc.short_title() or "").strip()
    converter = html2text.HTML2Text()
    converter.body_width = 0
    converter.unicode_snob = True
    converter.ignore_links = False
    converter.ignore_images = False
    body = converter.handle(cleaned_html).strip()
    if not body:
        return None
    header = ""
    if title:
        header = f"# {title}\n\n"
    return f"{header}> Source: {url}\n\n{body}\n"


class _StdlibStripper(HTMLParser):
    """Fallback HTML → text converter using only stdlib. Quality is poor but
    deterministic; intended as a last resort when readability isn't installed.
    Drops common non-content blocks (script/style/nav/footer/aside)."""

    DROP_TAGS = {"script", "style", "nav", "footer", "aside", "noscript", "form"}
    BLOCK_TAGS = {
        "p", "div", "section", "article", "li", "ul", "ol",
        "h1", "h2", "h3", "h4", "h5", "h6", "br", "tr",
    }

    def __init__(self) -> None:
        super().__init__()
        self._buf: list[str] = []
        self._drop_depth = 0
        self._title = ""
        self._in_title = False

    def handle_starttag(self, tag, attrs):
        if tag in self.DROP_TAGS:
            self._drop_depth += 1
        elif tag == "title":
            self._in_title = True
        elif tag in self.BLOCK_TAGS:
            self._buf.append("\n")

    def handle_endtag(self, tag):
        if tag in self.DROP_TAGS:
            self._drop_depth = max(0, self._drop_depth - 1)
        elif tag == "title":
            self._in_title = False
        elif tag in self.BLOCK_TAGS:
            self._buf.append("\n")

    def handle_data(self, data):
        if self._drop_depth:
            return
        if self._in_title:
            self._title += data
            return
        self._buf.append(data)

    def text(self) -> str:
        raw = "".join(self._buf)
        lines = [line.strip() for line in raw.splitlines()]
        # Drop empty lines but keep paragraph breaks.
        out: list[str] = []
        prev_blank = False
        for line in lines:
            if not line:
                if not prev_blank:
                    out.append("")
                prev_blank = True
            else:
                out.append(line)
                prev_blank = False
        return "\n".join(out).strip()


def extract_with_stdlib(html: str, url: str) -> str:
    p = _StdlibStripper()
    p.feed(html)
    body = p.text()
    if not body:
        body = "(no text content extracted)"
    header = ""
    title = (p._title or "").strip()
    if title:
        header = f"# {title}\n\n"
    return f"{header}> Source: {url}\n\n{body}\n"


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("url", help="URL to fetch")
    parser.add_argument(
        "--prefer",
        choices=("auto", "readability", "stdlib"),
        default="auto",
        help="Force a specific extractor (default: auto = readability if installed, else stdlib)",
    )
    args = parser.parse_args()

    try:
        html = fetch_html(args.url)
    except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError) as exc:
        print(
            f"[fetch] tier=local status=fail reason=\"fetch failed: {exc}\"",
            file=sys.stderr,
        )
        return 1

    used = ""
    body: str | None = None

    if args.prefer in ("auto", "readability"):
        body = extract_with_readability(html, args.url)
        if body is not None:
            used = "readability"
        elif args.prefer == "readability":
            print(
                "[fetch] tier=local status=fail "
                "reason=\"--prefer readability but readability-lxml or html2text not installed; "
                "install with: pip install --user readability-lxml html2text\"",
                file=sys.stderr,
            )
            return 1

    if body is None:
        body = extract_with_stdlib(html, args.url)
        used = "stdlib"

    # Sanity floor: if even the stdlib extractor returns essentially nothing,
    # treat as failure so the proxy fallback (if --use-proxy) gets a chance.
    body_lines = [l for l in body.splitlines() if l.strip()]
    if len(body_lines) < 4:
        print(
            f"[fetch] tier=local status=fail reason=\"extractor={used} produced <4 non-empty lines\"",
            file=sys.stderr,
        )
        return 1

    if used == "stdlib":
        print(
            "[fetch] tier=local status=ok extractor=stdlib "
            "hint=\"install readability-lxml + html2text for cleaner output\"",
            file=sys.stderr,
        )
    else:
        print(f"[fetch] tier=local status=ok extractor={used}", file=sys.stderr)

    sys.stdout.write(body)
    return 0


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env python3
"""Fetch WeChat public account article as Markdown using Playwright + BeautifulSoup.

Special thanks to joeseesun for the excellent qiaomu-markdown-proxy project,
which inspired the Playwright-based WeChat scraping approach in this script.
https://github.com/joeseesun/qiaomu-markdown-proxy

Requirements:
    pip install playwright beautifulsoup4 lxml
    playwright install chromium

Usage:
    python3 fetch_weixin.py <url>
    python3 fetch_weixin.py <url> --json
"""

import sys
import json
import asyncio


def yaml_string(value: str) -> str:
    return json.dumps("" if value is None else str(value), ensure_ascii=False)


async def fetch(url: str) -> dict:
    try:
        from playwright.async_api import async_playwright
        from bs4 import BeautifulSoup
    except ImportError as e:
        return {"error": str(e) + "\nRun: pip install playwright beautifulsoup4 lxml && playwright install chromium"}

    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page(
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
        )
        try:
            await page.goto(url, wait_until="domcontentloaded", timeout=30000)
            await page.wait_for_selector("#js_content", timeout=15000)
            html = await page.content()
        except Exception as e:
            await browser.close()
            return {"error": f"Page load failed: {e}"}
        await browser.close()

    soup = BeautifulSoup(html, "lxml")

    title = (soup.select_one("#activity-name") or soup.new_tag("x")).get_text(strip=True)
    author = (soup.select_one("#js_author_name") or soup.new_tag("x")).get_text(strip=True)
    date = (soup.select_one("#publish_time") or soup.new_tag("x")).get_text(strip=True)

    content_el = soup.select_one("#js_content")
    if not content_el:
        return {"error": "Could not find #js_content"}

    for tag in content_el.find_all(["script", "style"]):
        tag.decompose()

    for img in content_el.find_all("img"):
        src = img.get("data-src") or img.get("src") or ""
        img.replace_with(f"\n![image]({src})\n" if src else "")

    lines = []
    for el in content_el.find_all(["p", "h1", "h2", "h3", "h4", "section", "blockquote"]):
        text = el.get_text(strip=True)
        if not text:
            continue
        if el.name in ("h1", "h2", "h3", "h4"):
            lines.append(f"{'#' * int(el.name[1])} {text}")
        elif el.name == "blockquote":
            lines.append(f"> {text}")
        else:
            lines.append(text)

    content = "\n\n".join(lines) or content_el.get_text("\n", strip=True)
    return {"title": title, "author": author, "date": date, "url": url, "content": content}


def to_markdown(r: dict) -> str:
    if "error" in r:
        return f"Error: {r['error']}"
    parts = [
        "---",
        f"title: {yaml_string(r.get('title', ''))}",
        *([f"author: {yaml_string(r['author'])}"] if r.get("author") else []),
        *([f"date: {yaml_string(r['date'])}"] if r.get("date") else []),
        f"url: {yaml_string(r.get('url', ''))}",
        "---",
        "",
        f"# {r['title']}" if r.get("title") else "",
        "",
        r.get("content", ""),
    ]
    return "\n".join(parts)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: fetch_weixin.py <url> [--json]", file=sys.stderr)
        sys.exit(1)

    result = asyncio.run(fetch(sys.argv[1]))
    if "--json" in sys.argv:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(to_markdown(result))

#!/usr/bin/env bash
# Fetch a URL as Markdown.
#
# Privacy-first cascade:
#   Default (no --use-proxy): local extractor only. URL is never sent to a
#   third party. Best quality when readability-lxml + html2text are pip-
#   installed; degrades to a stdlib-only stripper otherwise.
#
#   With --use-proxy: tries local first, then defuddle.md, then r.jina.ai.
#   Use this for JS-heavy pages, X/Twitter, paywalls, or anything the local
#   extractor cannot reach. Be aware: the URL is sent to those third-party
#   services and may be cached or logged. Never feed sensitive URLs through
#   --use-proxy.
#
# Every tier writes a structured stderr line:
#   [fetch] tier=<local|defuddle|jina> status=<ok|fail|skip> reason="..."
#
# Special thanks to joeseesun for the qiaomu-markdown-proxy project, which
# inspired the proxy cascade design:
# https://github.com/joeseesun/qiaomu-markdown-proxy
#
# Usage:
#   fetch.sh <url> [proxy_url]
#   fetch.sh --use-proxy <url> [proxy_url]
set -euo pipefail

USE_PROXY=0
if [ "${1:-}" = "--use-proxy" ]; then
  USE_PROXY=1
  shift
fi

URL="${1:?Usage: fetch.sh [--use-proxy] <url> [proxy_url]}"
PROXY="${2:-}"

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"

LOCAL_ERR="$(mktemp)"
trap 'rm -f "$LOCAL_ERR"' EXIT

# shellcheck disable=SC2329,SC2317  # called indirectly via _with_retry / _try_once
_curl() {
  if [ -n "$PROXY" ]; then
    https_proxy="$PROXY" http_proxy="$PROXY" curl -sfL --connect-timeout 10 --max-time 30 "$@"
  else
    curl -sfL --connect-timeout 10 --max-time 30 "$@"
  fi
}

_has_content() {
  local content="$1"
  [ "$(printf '%s' "$content" | wc -l)" -gt 5 ] || return 1
  # Reject pages dominated by login walls, captchas, or bot challenges that
  # otherwise pass the line-count check. Add new markers here, not new branches.
  if printf '%s' "$content" | grep -qE "Don't miss what's happening|Sign in to continue|Please sign in|Log in to continue|请登录|登录后查看|机器人验证|人机验证|Just a moment\.\.\.|Checking your browser" 2>/dev/null; then
    return 1
  fi
  return 0
}

_try_once() {
  local out
  out=$("$@" 2>/dev/null || true)
  if _has_content "$out"; then echo "$out"; return 0; fi
  return 1
}

_with_retry() {
  _try_once "$@" && return 0
  sleep 2
  _try_once "$@" && return 0
  return 1
}

# Tier 1: local extractor. Always tried first.
if OUT=$(python3 "$SCRIPT_DIR/fetch_local.py" "$URL" 2>"$LOCAL_ERR"); then
  cat "$LOCAL_ERR" >&2 2>/dev/null || true
  echo "$OUT"
  exit 0
fi
cat "$LOCAL_ERR" >&2 2>/dev/null || true

# Without --use-proxy, stop here. URL never leaves the machine.
if [ "$USE_PROXY" -eq 0 ]; then
  echo "[fetch] status=fail reason=\"local extractor failed; rerun with --use-proxy to try defuddle.md and r.jina.ai (URL will be sent to those services)\"" >&2
  exit 1
fi

# Tier 2: defuddle.md (third party; user opted in via --use-proxy).
if OUT=$(_with_retry _curl "https://defuddle.md/$URL"); then
  echo "[fetch] tier=defuddle status=ok" >&2
  echo "$OUT"
  exit 0
fi
echo "[fetch] tier=defuddle status=fail reason=\"empty or paywall-like response\"" >&2

# Tier 3: r.jina.ai (third party; user opted in via --use-proxy).
if OUT=$(_with_retry _curl "https://r.jina.ai/$URL"); then
  echo "[fetch] tier=jina status=ok" >&2
  echo "$OUT"
  exit 0
fi
echo "[fetch] tier=jina status=fail reason=\"empty or paywall-like response\"" >&2

echo "[fetch] status=fail reason=\"all tiers (local, defuddle, jina) failed for $URL\"" >&2
exit 1