reader_server.py

from __future__ import annotations

import argparse
import html
import json
import mimetypes
import os
import re
import sys
import threading
import webbrowser
from dataclasses import dataclass
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, quote, urlparse

try:
    import markdown as markdown_lib
except ImportError as exc:  # pragma: no cover - startup guard
    print(
        "Missing dependency: Markdown. Run `pip install -r requirements.txt` first.",
        file=sys.stderr,
    )
    raise SystemExit(1) from exc


SUPPORTED_MARKDOWN_EXTENSIONS = {".md", ".markdown", ".mdx"}
SKIP_DIRECTORIES = {".git", ".hg", ".svn", "__pycache__", "node_modules", ".venv", "venv", "env"}
MARKDOWN_EXTENSIONS = ["extra", "admonition", "sane_lists", "toc", "nl2br"]
HTTP_ADDRESS = "127.0.0.1"

ASSET_ATTRIBUTE_PATTERN = re.compile(r'(<img\b[^>]*?\bsrc=")([^"]+)(")', re.IGNORECASE)
LINK_ATTRIBUTE_PATTERN = re.compile(r'(<a\b[^>]*?\bhref=")([^"]+)(")', re.IGNORECASE)
TAG_PATTERN = re.compile(r"<[^>]+>")
SPLIT_DIGITS_PATTERN = re.compile(r"(\d+)")
TITLE_PATTERN = re.compile(r"^\s*#\s+(.+?)\s*$", re.MULTILINE)
LATIN_WORD_PATTERN = re.compile(r"[A-Za-z0-9_]+(?:[-'][A-Za-z0-9_]+)?")
CJK_PATTERN = re.compile(r"[\u3400-\u9fff]")


@dataclass
class AppConfig:
    default_root: Path
    scan_interval_seconds: int
    initial_speed: int
    open_browser: bool
    host: str
    port: int


APP_CONFIG: AppConfig
APP_ROOT = Path(__file__).resolve().parent
STATIC_ROOT = APP_ROOT / "static"


def natural_key(value: str) -> list[object]:
    return [int(part) if part.isdigit() else part.lower() for part in SPLIT_DIGITS_PATTERN.split(value)]


def read_text_file(path: Path) -> str:
    for encoding in ("utf-8-sig", "utf-8", "gb18030"):
        try:
            return path.read_text(encoding=encoding)
        except UnicodeDecodeError:
            continue
    return path.read_text(encoding="utf-8", errors="replace")


def extract_front_matter(source: str) -> tuple[dict[str, str], str]:
    if not source.startswith("---"):
        return {}, source

    lines = source.splitlines()
    if not lines or lines[0].strip() != "---":
        return {}, source

    end_index = -1
    for index in range(1, min(len(lines), 120)):
        if lines[index].strip() == "---":
            end_index = index
            break

    if end_index == -1:
        return {}, source

    metadata: dict[str, str] = {}
    for line in lines[1:end_index]:
        if ":" not in line:
            continue
        key, value = line.split(":", 1)
        metadata[key.strip().lower()] = value.strip().strip("'\"")

    body = "\n".join(lines[end_index + 1 :])
    return metadata, body


def strip_html_tags(source: str) -> str:
    return TAG_PATTERN.sub(" ", source)


def count_reading_units(source: str) -> int:
    return max(len(CJK_PATTERN.findall(source)) + len(LATIN_WORD_PATTERN.findall(source)), 1)


def extract_title(body: str, metadata: dict[str, str], fallback: str) -> str:
    if metadata.get("title"):
        return metadata["title"]

    match = TITLE_PATTERN.search(body)
    if match:
        return match.group(1).strip()

    return fallback


def sanitize_user_path(raw_value: str | None, base_path: Path | None = None) -> Path:
    value = (raw_value or "").strip().strip("\"'")
    if not value:
        raise ValueError("Path is empty.")

    candidate = Path(value)
    if not candidate.is_absolute():
        root_base = base_path or APP_CONFIG.default_root
        candidate = (root_base / candidate).resolve()
    else:
        candidate = candidate.resolve()

    return candidate


def display_relative_path(path: Path, root: Path) -> str:
    try:
        return path.relative_to(root).as_posix()
    except ValueError:
        return path.name


def iter_markdown_files(root: Path) -> list[Path]:
    root = root.resolve()
    files: list[Path] = []
    for current_root, dirnames, filenames in os.walk(root):
        current_root_path = Path(current_root)
        dirnames[:] = [
            directory
            for directory in dirnames
            if directory not in SKIP_DIRECTORIES
            and not directory.startswith(".")
            and not (
                root != APP_ROOT
                and (current_root_path / directory).resolve() == APP_ROOT
            )
        ]
        for filename in filenames:
            suffix = Path(filename).suffix.lower()
            if suffix in SUPPORTED_MARKDOWN_EXTENSIONS:
                files.append(Path(current_root) / filename)
    files.sort(key=lambda path: natural_key(display_relative_path(path, root)))
    return files


def resolve_local_reference(raw_target: str, source_path: Path) -> Path | None:
    target = html.unescape(raw_target).strip()
    if not target:
        return None

    lowered = target.lower()
    if target.startswith("#") or lowered.startswith(
        ("http://", "https://", "mailto:", "tel:", "data:", "javascript:", "//")
    ):
        return None

    core_target = target.split("#", 1)[0].split("?", 1)[0].strip()
    if not core_target:
        return None

    if re.match(r"^[A-Za-z]:[\\/]", core_target):
        candidate = Path(core_target)
    else:
        candidate = Path(core_target)
        if not candidate.is_absolute():
            candidate = source_path.parent / candidate

    try:
        return candidate.resolve()
    except OSError:
        return None


def rewrite_rendered_html(rendered_html: str, source_path: Path) -> str:
    def replace_asset(match: re.Match[str]) -> str:
        resolved = resolve_local_reference(match.group(2), source_path)
        if not resolved or not resolved.exists():
            return match.group(0)
        asset_url = f"/api/asset?path={quote(str(resolved))}"
        return f"{match.group(1)}{asset_url}{match.group(3)}"

    def replace_link(match: re.Match[str]) -> str:
        href = match.group(2)
        lowered = href.lower()
        if lowered.startswith(("http://", "https://")):
            return f'{match.group(1)}{href}{match.group(3)} target="_blank" rel="noreferrer"'

        resolved = resolve_local_reference(href, source_path)
        if not resolved or not resolved.exists():
            return match.group(0)

        if resolved.suffix.lower() in SUPPORTED_MARKDOWN_EXTENSIONS:
            escaped_path = html.escape(str(resolved), quote=True)
            return f'{match.group(1)}#{match.group(3)} data-local-md-path="{escaped_path}"'

        asset_url = f"/api/asset?path={quote(str(resolved))}"
        return f"{match.group(1)}{asset_url}{match.group(3)}"

    rendered_html = ASSET_ATTRIBUTE_PATTERN.sub(replace_asset, rendered_html)
    rendered_html = LINK_ATTRIBUTE_PATTERN.sub(replace_link, rendered_html)
    return rendered_html


def load_markdown_payload(file_path: Path, root: Path) -> dict[str, object]:
    source = read_text_file(file_path)
    metadata, body = extract_front_matter(source)

    renderer = markdown_lib.Markdown(extensions=MARKDOWN_EXTENSIONS, output_format="html5")
    rendered_html = renderer.convert(body)
    rendered_html = rewrite_rendered_html(rendered_html, file_path)

    plain_text = strip_html_tags(rendered_html)
    reading_units = count_reading_units(plain_text)

    return {
        "path": str(file_path),
        "name": file_path.name,
        "relative_path": display_relative_path(file_path, root),
        "title": extract_title(body, metadata, file_path.stem),
        "html": rendered_html,
        "reading_units": reading_units,
        "estimated_seconds_at_default_speed": round(reading_units / APP_CONFIG.initial_speed * 60, 1),
        "modified_timestamp": file_path.stat().st_mtime,
        "metadata": metadata,
    }


def scan_directory_payload(root: Path) -> dict[str, object]:
    if not root.exists():
        raise FileNotFoundError(f"Directory does not exist: {root}")
    if not root.is_dir():
        raise NotADirectoryError(f"Not a directory: {root}")

    files = iter_markdown_files(root)
    items = [
        {
            "path": str(path),
            "name": path.name,
            "relative_path": display_relative_path(path, root),
            "modified_timestamp": path.stat().st_mtime,
            "size": path.stat().st_size,
        }
        for path in files
    ]
    return {
        "root": str(root),
        "count": len(items),
        "items": items,
    }


class ReaderRequestHandler(BaseHTTPRequestHandler):
    server_version = "MarkdownAutoReader/1.0"

    def log_message(self, format_string: str, *args: object) -> None:
        return

    def do_GET(self) -> None:  # noqa: N802
        parsed = urlparse(self.path)
        if parsed.path == "/api/config":
            self.handle_config()
            return
        if parsed.path == "/api/files":
            self.handle_files(parsed.query)
            return
        if parsed.path == "/api/file":
            self.handle_file(parsed.query)
            return
        if parsed.path == "/api/asset":
            self.handle_asset(parsed.query)
            return
        self.handle_static(parsed.path)

    def handle_config(self) -> None:
        payload = {
            "default_root": str(APP_CONFIG.default_root),
            "scan_interval_seconds": APP_CONFIG.scan_interval_seconds,
            "initial_speed": APP_CONFIG.initial_speed,
            "host": APP_CONFIG.host,
            "port": APP_CONFIG.port,
        }
        self.send_json(HTTPStatus.OK, payload)

    def handle_files(self, query_string: str) -> None:
        query = parse_qs(query_string)
        raw_root = query.get("root", [str(APP_CONFIG.default_root)])[0]
        try:
            root = sanitize_user_path(raw_root)
            payload = scan_directory_payload(root)
        except (FileNotFoundError, NotADirectoryError, ValueError) as exc:
            self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
            return

        self.send_json(HTTPStatus.OK, payload)

    def handle_file(self, query_string: str) -> None:
        query = parse_qs(query_string)
        raw_root = query.get("root", [str(APP_CONFIG.default_root)])[0]
        raw_path = query.get("path", [""])[0]
        try:
            root = sanitize_user_path(raw_root)
            file_path = sanitize_user_path(raw_path)
        except ValueError as exc:
            self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
            return

        if not file_path.exists():
            self.send_error_json(HTTPStatus.NOT_FOUND, f"Markdown file not found: {file_path}")
            return
        if file_path.suffix.lower() not in SUPPORTED_MARKDOWN_EXTENSIONS:
            self.send_error_json(HTTPStatus.BAD_REQUEST, "File is not a supported markdown file.")
            return

        try:
            payload = load_markdown_payload(file_path, root)
        except ValueError as exc:
            self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
            return

        self.send_json(HTTPStatus.OK, payload)

    def handle_asset(self, query_string: str) -> None:
        query = parse_qs(query_string)
        raw_path = query.get("path", [""])[0]
        try:
            asset_path = sanitize_user_path(raw_path)
        except ValueError as exc:
            self.send_error_json(HTTPStatus.BAD_REQUEST, str(exc))
            return

        if not asset_path.exists() or not asset_path.is_file():
            self.send_error_json(HTTPStatus.NOT_FOUND, f"Asset not found: {asset_path}")
            return

        content_type, _ = mimetypes.guess_type(asset_path.name)
        if not content_type:
            content_type = "application/octet-stream"

        data = asset_path.read_bytes()
        self.send_response(HTTPStatus.OK)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def handle_static(self, request_path: str) -> None:
        relative_path = "index.html" if request_path in ("", "/") else request_path.lstrip("/")
        static_path = (STATIC_ROOT / relative_path).resolve()
        if not static_path.exists() or not static_path.is_file() or STATIC_ROOT not in static_path.parents:
            self.send_error_json(HTTPStatus.NOT_FOUND, f"Static file not found: {relative_path}")
            return

        content_type, _ = mimetypes.guess_type(static_path.name)
        if not content_type:
            content_type = "application/octet-stream"
        data = static_path.read_bytes()

        self.send_response(HTTPStatus.OK)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def send_json(self, status: HTTPStatus, payload: dict[str, object]) -> None:
        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(data)

    def send_error_json(self, status: HTTPStatus, message: str) -> None:
        self.send_json(status, {"error": message})


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Local markdown auto reader.")
    parser.add_argument("--root", default=str(Path.cwd()), help="Directory to scan on startup.")
    parser.add_argument("--host", default=HTTP_ADDRESS, help="Host to bind.")
    parser.add_argument("--port", type=int, default=8765, help="Port to bind.")
    parser.add_argument("--speed", type=int, default=12000, help="Default reading speed units per minute.")
    parser.add_argument(
        "--scan-interval",
        type=int,
        default=10,
        help="Directory rescan interval in seconds.",
    )
    parser.add_argument(
        "--no-browser",
        action="store_true",
        help="Do not open the browser automatically.",
    )
    return parser.parse_args()


def main() -> None:
    global APP_CONFIG

    args = parse_args()
    default_root = sanitize_user_path(args.root, base_path=Path.cwd())
    APP_CONFIG = AppConfig(
        default_root=default_root,
        scan_interval_seconds=max(args.scan_interval, 3),
        initial_speed=max(args.speed, 60),
        open_browser=not args.no_browser,
        host=args.host,
        port=args.port,
    )

    server = ThreadingHTTPServer((APP_CONFIG.host, APP_CONFIG.port), ReaderRequestHandler)
    url = f"http://{APP_CONFIG.host}:{APP_CONFIG.port}/"

    print(f"Markdown Auto Reader is running at {url}")
    print(f"Startup directory: {APP_CONFIG.default_root}")

    if APP_CONFIG.open_browser:
        threading.Timer(0.7, lambda: webbrowser.open(url)).start()

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\nStopping Markdown Auto Reader...")
    finally:
        server.server_close()


if __name__ == "__main__":
    main()