Qiaomu Markdown Proxy

MIT License

Copyright (c) 2026 joeseesun

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

#!/usr/bin/env bash
# Extract PDF to Markdown with multiple fallback methods.
# Usage: extract_pdf.sh <pdf_path>
set -euo pipefail

PDF_PATH="${1:?Usage: extract_pdf.sh <pdf_path>}"

if [ ! -f "$PDF_PATH" ]; then
  echo "ERROR: File not found: $PDF_PATH" >&2
  exit 1
fi

# Method 1: marker-pdf (best quality for papers, tables, complex layouts)
if command -v marker_single &>/dev/null; then
  OUTPUT_DIR="${2:-$HOME/Downloads}"
  marker_single "$PDF_PATH" --output_dir "$OUTPUT_DIR"
  exit 0
fi

# Method 2: pdftotext (fast, good for text-heavy PDFs)
if command -v pdftotext &>/dev/null; then
  pdftotext -layout "$PDF_PATH" - | sed 's/\f/\n---\n/g'
  exit 0
fi

# Method 3: pypdf (no-dependency fallback)
python3 -c "
import sys
try:
    import pypdf
except ImportError:
    print('ERROR: pypdf not installed. Run: pip install pypdf', file=sys.stderr)
    sys.exit(1)

reader = pypdf.PdfReader(sys.argv[1])
print('\n\n'.join(page.extract_text() for page in reader.pages))
" "$PDF_PATH"

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

import sys
import json
import os
import re
import requests

FEISHU_API_BASE = "https://open.feishu.cn/open-apis"


def get_tenant_access_token():
    """获取 tenant_access_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 或 FEISHU_APP_SECRET 未设置"

    url = f"{FEISHU_API_BASE}/auth/v3/tenant_access_token/internal"
    resp = requests.post(url, json={"app_id": app_id, "app_secret": app_secret})
    data = resp.json()
    if data.get("code") != 0:
        return None, f"获取 token 失败: {data.get('msg', resp.text)}"
    return data["tenant_access_token"], None


def parse_feishu_url(url):
    """从飞书 URL 解析 document_id 和文档类型"""
    patterns = [
        (r"feishu\.cn/docx/([A-Za-z0-9]+)", "docx"),
        (r"feishu\.cn/docs/([A-Za-z0-9]+)", "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]+)", "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 None, None


def get_document_info(token, doc_id):
    """获取文档元信息（标题等）"""
    url = f"{FEISHU_API_BASE}/docx/v1/documents/{doc_id}"
    headers = {"Authorization": f"Bearer {token}"}
    resp = requests.get(url, headers=headers)
    data = resp.json()
    if data.get("code") == 0:
        return data.get("data", {}).get("document", {})
    return {}


def get_document_blocks(token, doc_id):
    """获取文档所有 blocks"""
    url = f"{FEISHU_API_BASE}/docx/v1/documents/{doc_id}/blocks"
    headers = {"Authorization": f"Bearer {token}"}
    all_blocks = []
    page_token = None

    while True:
        params = {"page_size": 500}
        if page_token:
            params["page_token"] = page_token
        resp = requests.get(url, headers=headers, params=params)
        data = resp.json()
        if data.get("code") != 0:
            return None, f"获取 blocks 失败: {data.get('msg', resp.text)}"

        items = data.get("data", {}).get("items", [])
        all_blocks.extend(items)

        if not data.get("data", {}).get("has_more", False):
            break
        page_token = data["data"].get("page_token")

    return all_blocks, None


def get_wiki_node(token, wiki_token):
    """获取知识库节点信息，返回实际的 obj_token 和 obj_type"""
    url = f"{FEISHU_API_BASE}/wiki/v2/spaces/get_node"
    headers = {"Authorization": f"Bearer {token}"}
    resp = requests.get(url, headers=headers, params={"token": wiki_token})
    data = resp.json()
    if data.get("code") == 0:
        node = data.get("data", {}).get("node", {})
        return node.get("obj_token"), node.get("obj_type")
    return None, None


def extract_text_from_elements(elements):
    """从 text_run / mention_user 等元素中提取文本"""
    if not elements:
        return ""
    parts = []
    for el in elements:
        if "text_run" in el:
            tr = el["text_run"]
            text = tr.get("content", "")
            style = tr.get("text_element_style", {})
            if style.get("bold"):
                text = f"**{text}**"
            if style.get("italic"):
                text = f"*{text}*"
            if style.get("strikethrough"):
                text = f"~~{text}~~"
            if style.get("inline_code"):
                text = f"`{text}`"
            if style.get("link", {}).get("url"):
                import urllib.parse
                link_url = urllib.parse.unquote(style["link"]["url"])
                text = f"[{text}]({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)


def blocks_to_markdown(blocks):
    """将飞书 blocks 转为 Markdown"""
    lines = []
    ordered_list_counter = {}  # parent_id -> counter

    for block in blocks:
        block_type = block.get("block_type")
        parent_id = block.get("parent_id", "")

        # 1 = Page, 2 = Text, 3 = Heading1, 4 = Heading2, ..., 9 = Heading7+
        # 10 = BulletList, 11 = OrderedList, 12 = Code, 13 = Quote
        # 14 = Equation, 15 = Todo, 16 = Divider
        # 17 = Image, 18 = TableCell, 19 = Table
        # 22 = Callout, 23 = ChatCard, 27 = Grid, 28 = GridColumn

        if block_type == 2:  # Text
            text_data = block.get("text", {})
            text = extract_text_from_elements(text_data.get("elements", []))
            if text.strip():
                lines.append(text)
            else:
                lines.append("")

        elif block_type in (3, 4, 5, 6, 7, 8, 9):  # Heading 1-7
            level = block_type - 2
            heading_data = block.get("heading" + str(level), {}) or block.get("heading", {})
            # Try multiple key formats
            for key in [f"heading{level}", "heading"]:
                if key in block:
                    heading_data = block[key]
                    break
            text = extract_text_from_elements(heading_data.get("elements", []))
            lines.append(f"{'#' * level} {text}")

        elif block_type == 10:  # Bullet list
            text_data = block.get("bullet", {})
            text = extract_text_from_elements(text_data.get("elements", []))
            lines.append(f"- {text}")

        elif block_type == 11:  # Ordered list
            text_data = block.get("ordered", {})
            text = extract_text_from_elements(text_data.get("elements", []))
            counter = ordered_list_counter.get(parent_id, 0) + 1
            ordered_list_counter[parent_id] = counter
            lines.append(f"{counter}. {text}")

        elif block_type == 12:  # Code block
            code_data = block.get("code", {})
            text = extract_text_from_elements(code_data.get("elements", []))
            lang = code_data.get("style", {}).get("language", "")
            # Map language codes
            lang_map = {1: "plaintext", 2: "abap", 3: "ada", 4: "apache", 5: "apex",
                        6: "assembly", 7: "bash", 8: "c", 9: "csharp", 10: "cpp",
                        11: "clojure", 12: "cmake", 13: "coffeescript", 14: "css",
                        15: "d", 16: "dart", 17: "delphi", 18: "django", 19: "dockerfile",
                        20: "elixir", 21: "elm", 22: "erlang", 23: "fortran",
                        24: "fsharp", 25: "go", 26: "graphql", 27: "groovy", 28: "haskell",
                        29: "html", 30: "http", 31: "java", 32: "javascript",
                        33: "json", 34: "julia", 35: "kotlin", 36: "latex", 37: "lisp",
                        38: "lua", 39: "makefile", 40: "markdown", 41: "matlab",
                        42: "nginx", 43: "objectivec", 44: "ocaml", 45: "perl",
                        46: "php", 47: "powershell", 48: "properties", 49: "protobuf",
                        50: "python", 51: "r", 52: "ruby", 53: "rust", 54: "scala",
                        55: "scheme", 56: "scss", 57: "shell", 58: "sql", 59: "swift",
                        60: "thrift", 61: "toml", 62: "typescript", 63: "vbnet",
                        64: "verilog", 65: "vhdl", 66: "visual_basic", 67: "vue",
                        68: "xml", 69: "yaml"}
            lang_str = lang_map.get(lang, "") if isinstance(lang, int) else str(lang)
            lines.append(f"```{lang_str}")
            lines.append(text)
            lines.append("```")

        elif block_type == 13:  # Quote
            text_data = block.get("quote", {})
            text = extract_text_from_elements(text_data.get("elements", []))
            lines.append(f"> {text}")

        elif block_type == 14:  # Equation block
            eq_data = block.get("equation", {})
            text = extract_text_from_elements(eq_data.get("elements", []))
            lines.append(f"$$\n{text}\n$$")

        elif block_type == 15:  # Todo
            todo_data = block.get("todo", {})
            text = extract_text_from_elements(todo_data.get("elements", []))
            done = todo_data.get("style", {}).get("done", False)
            checkbox = "[x]" if done else "[ ]"
            lines.append(f"- {checkbox} {text}")

        elif block_type == 16:  # Divider
            lines.append("---")

        elif block_type == 17:  # Image
            image_data = block.get("image", {})
            token_val = image_data.get("token", "")
            lines.append(f"![image](feishu-image://{token_val})")

        elif block_type == 22:  # Callout
            callout_data = block.get("callout", {})
            # Callout is a container, children will be processed separately
            emoji = callout_data.get("emoji_id", "")
            if emoji:
                lines.append(f"> {emoji}")

        elif block_type == 1:  # Page (root), skip
            pass

        else:
            # Unknown block type, try to extract any text
            for key in block:
                if isinstance(block[key], dict) and "elements" in block[key]:
                    text = extract_text_from_elements(block[key]["elements"])
                    if text.strip():
                        lines.append(text)
                    break

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


def fetch_feishu_doc(url_or_id):
    """主函数：获取飞书文档并转为 Markdown"""
    # 解析 URL
    doc_id, doc_type = parse_feishu_url(url_or_id)
    if not doc_id:
        # 可能直接传了 doc_token
        doc_id = url_or_id
        doc_type = "docx"

    # 获取 token
    token, err = get_tenant_access_token()
    if err:
        return {"error": err}

    # Wiki 需要先获取实际文档 ID
    if doc_type == "wiki":
        real_id, real_type = get_wiki_node(token, doc_id)
        if real_id:
            doc_id = real_id
            doc_type = real_type or "docx"
        else:
            return {"error": f"无法获取知识库节点信息: {doc_id}"}

    # 获取文档信息
    doc_info = get_document_info(token, doc_id)
    title = doc_info.get("title", "")

    # 获取 blocks
    blocks, err = get_document_blocks(token, doc_id)
    if err:
        return {"error": err}

    # 转换为 Markdown
    content = blocks_to_markdown(blocks)

    return {
        "title": title,
        "document_id": doc_id,
        "url": url_or_id,
        "content": content,
    }


def format_as_markdown(result):
    """格式化为 Markdown 文档"""
    if "error" in result:
        return f"Error: {result['error']}"

    parts = ["---"]
    if result.get("title"):
        parts.append(f'title: "{result["title"]}"')
    parts.append(f'document_id: "{result["document_id"]}"')
    if result.get("url"):
        parts.append(f'url: "{result["url"]}"')
    parts.append("---")
    parts.append("")
    if result.get("title"):
        parts.append(f"# {result['title']}")
        parts.append("")
    parts.append(result.get("content", ""))
    return "\n".join(parts)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: fetch_feishu.py <feishu_url_or_doc_token> [--json]", file=sys.stderr)
        print("  需要环境变量: FEISHU_APP_ID, FEISHU_APP_SECRET", file=sys.stderr)
        sys.exit(1)

    url = sys.argv[1]
    use_json = "--json" in sys.argv

    result = fetch_feishu_doc(url)

    if use_json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(format_as_markdown(result))

#!/usr/bin/env python3
"""Fetch WeChat (公众号) article as Markdown. Standalone script using Playwright + BeautifulSoup."""

import sys
import json
import asyncio
import re

async def fetch_weixin_article(url: str) -> dict:
    """Fetch and parse a WeChat article, return dict with title, author, publish_time, content."""
    try:
        from playwright.async_api import async_playwright
    except ImportError:
        return {"error": "playwright not installed. Run: pip install playwright && playwright install chromium"}

    try:
        from bs4 import BeautifulSoup
    except ImportError:
        return {"error": "beautifulsoup4 not installed. Run: pip install beautifulsoup4 lxml"}

    html = None
    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"Failed to load page: {e}"}
        await browser.close()

    if not html:
        return {"error": "No HTML content retrieved"}

    soup = BeautifulSoup(html, "lxml")

    # Extract title
    title_el = soup.select_one("#activity-name")
    title = title_el.get_text(strip=True) if title_el else ""

    # Extract author
    author_el = soup.select_one("#js_author_name") or soup.select_one(".rich_media_meta_text")
    author = author_el.get_text(strip=True) if author_el else ""

    # Extract publish time
    time_el = soup.select_one("#publish_time")
    publish_time = time_el.get_text(strip=True) if time_el else ""

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

    # Convert to markdown-like text
    # Remove scripts and styles
    for tag in content_el.find_all(["script", "style"]):
        tag.decompose()

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

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

    content = "\n\n".join(lines)

    # If structured extraction got nothing, fall back to plain text
    if not content.strip():
        content = content_el.get_text("\n", strip=True)

    return {
        "title": title,
        "author": author,
        "publish_time": publish_time,
        "content": content,
        "url": url,
    }


def format_as_markdown(result: dict) -> str:
    """Format result dict as a Markdown document."""
    if "error" in result:
        return f"Error: {result['error']}"

    parts = ["---"]
    if result.get("title"):
        parts.append(f"title: \"{result['title']}\"")
    if result.get("author"):
        parts.append(f"author: \"{result['author']}\"")
    if result.get("publish_time"):
        parts.append(f"date: \"{result['publish_time']}\"")
    parts.append(f"url: \"{result['url']}\"")
    parts.append("---")
    parts.append("")
    if result.get("title"):
        parts.append(f"# {result['title']}")
        parts.append("")
    parts.append(result.get("content", ""))
    return "\n".join(parts)


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

    url = sys.argv[1]
    use_json = "--json" in sys.argv

    result = asyncio.run(fetch_weixin_article(url))

    if use_json:
        print(json.dumps(result, ensure_ascii=False, indent=2))
    else:
        print(format_as_markdown(result))

#!/usr/bin/env bash
# Fetch a URL as Markdown via proxy cascade with auto-fallback.
# Usage: fetch.sh <url> [proxy_url]
# Example: fetch.sh https://example.com http://127.0.0.1:7890
set -euo pipefail

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

_curl() {
  if [ -n "$PROXY" ]; then
    https_proxy="$PROXY" http_proxy="$PROXY" curl -sL "$@"
  else
    curl -sL "$@"
  fi
}

_has_content() {
  local content="$1"
  local line_count=$(echo "$content" | wc -l | tr -d ' ')

  # Must have more than 5 lines
  [ "$line_count" -gt 5 ] || return 1

  # Filter out common error pages
  echo "$content" | grep -qv "Don't miss what's happening" || return 1
  echo "$content" | grep -qv "Access Denied" || return 1
  echo "$content" | grep -qv "404 Not Found" || return 1

  return 0
}

# 1. r.jina.ai - wide coverage, preserves image links
OUT=$(_curl "https://r.jina.ai/$URL" 2>/dev/null || true)
if _has_content "$OUT"; then
  echo "$OUT"
  exit 0
fi

# 2. defuddle.md - cleaner output with YAML frontmatter
OUT=$(_curl "https://defuddle.md/$URL" 2>/dev/null || true)
if _has_content "$OUT"; then
  echo "$OUT"
  exit 0
fi

# 3. agent-fetch - last resort local tool
if command -v npx &>/dev/null; then
  OUT=$(npx --yes agent-fetch "$URL" --json 2>/dev/null || true)
  if [ -n "$OUT" ]; then
    echo "$OUT"
    exit 0
  fi
fi

echo "ERROR: All fetch methods failed for: $URL" >&2
exit 1