large_browser.py

"""Serve the large-directory SQLite index through a loopback-only browser UI."""

from __future__ import annotations

import argparse
import json
import sqlite3
from contextlib import closing
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, quote, urlparse

from large_path_export import (
    MAX_PAGE_NODES,
    MAX_PATH_NODES,
    PathExportError,
    build_path_svg,
)


PROJECT_DIR = Path(__file__).resolve().parent
DEFAULT_DATABASE_PATH = PROJECT_DIR / "outputs" / "folder-index.db"
DEFAULT_HTML_PATH = PROJECT_DIR / "large_browser.html"
LOOPBACK_HOST = "127.0.0.1"
DEFAULT_PORT = 8765
PAGE_SIZE = 300

REQUIRED_NODE_COLUMNS = {
    "id",
    "parent_id",
    "name",
    "normalized_name",
    "path",
    "node_type",
    "extension",
    "size",
    "depth",
    "order_index",
    "child_count",
}


class RequestError(ValueError):
    """An invalid browser API request."""


class FolderIndex:
    """Read-only access to the large-directory SQLite index."""

    def __init__(self, database_path: Path) -> None:
        self.database_path = database_path.resolve()
        if not self.database_path.is_file():
            raise FileNotFoundError(f"SQLite 索引不存在：{self.database_path}")
        self._validate_schema()

    def _connect(self) -> sqlite3.Connection:
        connection = sqlite3.connect(
            f"{self.database_path.as_uri()}?mode=ro",
            uri=True,
            timeout=10,
        )
        connection.row_factory = sqlite3.Row
        connection.execute("PRAGMA query_only = ON")
        return connection

    def _validate_schema(self) -> None:
        with closing(self._connect()) as connection:
            columns = {
                row["name"]
                for row in connection.execute("PRAGMA table_info('nodes')")
            }
            missing = REQUIRED_NODE_COLUMNS - columns
            if missing:
                names = ", ".join(sorted(missing))
                raise ValueError(f"SQLite nodes 表缺少字段：{names}")

            root_count = connection.execute(
                "SELECT COUNT(*) FROM nodes WHERE parent_id IS NULL"
            ).fetchone()[0]
            if root_count != 1:
                raise ValueError(
                    f"SQLite 索引必须且只能包含一个根节点，当前为 {root_count}"
                )

    def summary(self) -> dict[str, Any]:
        with closing(self._connect()) as connection:
            root = connection.execute(
                """
                SELECT id, name, path, size, depth, child_count
                FROM nodes
                WHERE parent_id IS NULL
                """
            ).fetchone()
            counts = connection.execute(
                """
                SELECT
                    COUNT(*) AS node_count,
                    SUM(CASE WHEN node_type = 'file' THEN 1 ELSE 0 END)
                        AS file_count,
                    SUM(CASE WHEN node_type = 'directory' THEN 1 ELSE 0 END)
                        AS directory_count
                FROM nodes
                """
            ).fetchone()

        if root is None:
            raise ValueError("SQLite 索引中没有根节点")

        return {
            "root": dict(root),
            "node_count": counts["node_count"],
            "file_count": counts["file_count"],
            "directory_count": counts["directory_count"],
            "total_size": root["size"],
            "page_size": PAGE_SIZE,
        }

    def children(self, parent_id: int, page: int) -> dict[str, Any]:
        if parent_id <= 0:
            raise RequestError("parent_id 必须是正整数")
        if page <= 0:
            raise RequestError("page 必须是正整数")

        with closing(self._connect()) as connection:
            parent = connection.execute(
                """
                SELECT id, parent_id, name, path, node_type, size, depth,
                       child_count
                FROM nodes
                WHERE id = ?
                """,
                (parent_id,),
            ).fetchone()
            if parent is None:
                raise RequestError(f"节点不存在：{parent_id}")
            if parent["node_type"] != "directory":
                raise RequestError("只能查询目录的直接子节点")

            total = connection.execute(
                "SELECT COUNT(*) FROM nodes WHERE parent_id = ?",
                (parent_id,),
            ).fetchone()[0]
            total_pages = max(1, (total + PAGE_SIZE - 1) // PAGE_SIZE)
            if page > total_pages:
                raise RequestError(
                    f"页码超出范围：{page}，该目录共 {total_pages} 页"
                )

            offset = (page - 1) * PAGE_SIZE
            rows = connection.execute(
                """
                SELECT id, parent_id, name, path, node_type, extension, size,
                       depth, order_index, child_count
                FROM nodes
                WHERE parent_id = ?
                ORDER BY order_index
                LIMIT ? OFFSET ?
                """,
                (parent_id, PAGE_SIZE, offset),
            ).fetchall()

        start = offset + 1 if total else 0
        end = offset + len(rows)
        return {
            "parent": dict(parent),
            "items": [dict(row) for row in rows],
            "pagination": {
                "page": page,
                "page_size": PAGE_SIZE,
                "total": total,
                "total_pages": total_pages,
                "start": start,
                "end": end,
                "has_previous": page > 1,
                "has_next": page < total_pages,
            },
        }

    def path_snapshot(self, parent_id: int, page: int) -> dict[str, Any]:
        """Return one validated browser path and its current children page."""

        page_data = self.children(parent_id, page)
        if len(page_data["items"]) > PAGE_SIZE or PAGE_SIZE != MAX_PAGE_NODES:
            raise RequestError("当前页节点数量超过导出安全上限")

        with closing(self._connect()) as connection:
            ancestor_rows = connection.execute(
                """
                WITH RECURSIVE ancestors(
                    id, parent_id, name, path, node_type, size, depth,
                    child_count
                ) AS (
                    SELECT id, parent_id, name, path, node_type, size, depth,
                           child_count
                    FROM nodes
                    WHERE id = ?

                    UNION

                    SELECT parent.id, parent.parent_id, parent.name,
                           parent.path, parent.node_type, parent.size,
                           parent.depth, parent.child_count
                    FROM nodes AS parent
                    JOIN ancestors AS child ON parent.id = child.parent_id
                )
                SELECT id, parent_id, name, path, node_type, size, depth,
                       child_count
                FROM ancestors
                """,
                (parent_id,),
            ).fetchall()

        if not ancestor_rows:
            raise RequestError(f"节点不存在：{parent_id}")
        if len(ancestor_rows) > MAX_PATH_NODES:
            raise RequestError(
                f"当前路径超过安全上限 {MAX_PATH_NODES:,} 个节点"
            )

        rows_by_id = {row["id"]: row for row in ancestor_rows}
        reversed_path: list[dict[str, Any]] = []
        seen_ids: set[int] = set()
        current_id: int | None = parent_id
        while current_id is not None:
            if current_id in seen_ids:
                raise RequestError("SQLite 路径包含循环父子关系")
            seen_ids.add(current_id)
            row = rows_by_id.get(current_id)
            if row is None:
                raise RequestError("SQLite 路径存在缺失的父目录")
            if row["node_type"] != "directory":
                raise RequestError("当前路径只能包含目录节点")
            reversed_path.append(dict(row))
            if len(reversed_path) > MAX_PATH_NODES:
                raise RequestError(
                    f"当前路径超过安全上限 {MAX_PATH_NODES:,} 个节点"
                )
            current_id = row["parent_id"]

        path = list(reversed(reversed_path))
        if path[0]["parent_id"] is not None or path[0]["depth"] != 0:
            raise RequestError("当前路径没有连接到有效根目录")
        for index in range(1, len(path)):
            parent = path[index - 1]
            child = path[index]
            if child["parent_id"] != parent["id"]:
                raise RequestError("当前路径的父子关系不连续")
            if child["depth"] != parent["depth"] + 1:
                raise RequestError("当前路径的目录深度不连续")

        current_directory = page_data["parent"]
        if current_directory["id"] != path[-1]["id"]:
            raise RequestError("当前目录与路径末端不一致")
        if any(
            item["parent_id"] != current_directory["id"]
            for item in page_data["items"]
        ):
            raise RequestError("当前页包含不属于当前目录的节点")

        return {
            "path": path,
            "current_directory": current_directory,
            "items": page_data["items"],
            "pagination": page_data["pagination"],
        }


class LargeBrowserServer(ThreadingHTTPServer):
    """Threaded local server with prompt shutdown behavior."""

    daemon_threads = True
    allow_reuse_address = True


class LargeBrowserHandler(BaseHTTPRequestHandler):
    """Serve the static browser and its read-only JSON API."""

    server_version = "FolderVisualizer/0.2.0"
    folder_index: FolderIndex
    html_path: Path

    def do_GET(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
        parsed = urlparse(self.path)
        try:
            if parsed.path in ("/", "/large_browser.html"):
                self._serve_html()
                return
            if parsed.path == "/api/summary":
                self._send_json(self.folder_index.summary())
                return
            if parsed.path == "/api/children":
                parameters = parse_qs(parsed.query, keep_blank_values=True)
                parent_id = self._single_positive_int(parameters, "parent_id")
                page = self._single_positive_int(parameters, "page", default=1)
                self._send_json(self.folder_index.children(parent_id, page))
                return
            if parsed.path == "/api/export-path":
                parameters = parse_qs(parsed.query, keep_blank_values=True)
                parent_id = self._single_positive_int(parameters, "parent_id")
                page = self._single_positive_int(parameters, "page", default=1)
                snapshot = self.folder_index.path_snapshot(parent_id, page)
                export = build_path_svg(snapshot)
                self._send_svg(export.content, export.filename)
                return
            if parsed.path == "/favicon.ico":
                self.send_error(HTTPStatus.NOT_FOUND)
                return
            self._send_json(
                {"error": "接口不存在"}, status=HTTPStatus.NOT_FOUND
            )
        except RequestError as error:
            self._send_json(
                {"error": str(error)}, status=HTTPStatus.BAD_REQUEST
            )
        except (OSError, ValueError, sqlite3.Error, PathExportError) as error:
            self._send_json(
                {"error": f"无法读取文件夹索引：{error}"},
                status=HTTPStatus.INTERNAL_SERVER_ERROR,
            )

    def do_POST(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
        self._method_not_allowed()

    def do_PUT(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
        self._method_not_allowed()

    def do_DELETE(self) -> None:  # noqa: N802 - BaseHTTPRequestHandler API
        self._method_not_allowed()

    def _method_not_allowed(self) -> None:
        self._send_json(
            {"error": "本地服务仅提供只读 GET 请求"},
            status=HTTPStatus.METHOD_NOT_ALLOWED,
        )

    def _serve_html(self) -> None:
        content = self.html_path.read_bytes()
        self.send_response(HTTPStatus.OK)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(content)))
        self.send_header("Cache-Control", "no-store")
        self._security_headers()
        self.end_headers()
        self.wfile.write(content)

    def _send_json(
        self, payload: dict[str, Any], status: HTTPStatus = HTTPStatus.OK
    ) -> None:
        content = 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(content)))
        self.send_header("Cache-Control", "no-store")
        self._security_headers()
        self.end_headers()
        self.wfile.write(content)

    def _send_svg(self, content: bytes, filename: str) -> None:
        encoded_filename = quote(filename, safe="")
        self.send_response(HTTPStatus.OK)
        self.send_header("Content-Type", "image/svg+xml; charset=utf-8")
        self.send_header("Content-Length", str(len(content)))
        self.send_header("Cache-Control", "no-store")
        self.send_header(
            "Content-Disposition",
            "attachment; filename=\"folder-path.svg\"; "
            f"filename*=UTF-8''{encoded_filename}",
        )
        self.send_header(
            "X-FolderVisualizer-Filename", encoded_filename
        )
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header(
            "Content-Security-Policy",
            "default-src 'none'; script-src 'none'; style-src 'none'; "
            "img-src 'none'; object-src 'none'; base-uri 'none'; "
            "frame-ancestors 'none'; sandbox",
        )
        self.end_headers()
        self.wfile.write(content)

    def _security_headers(self) -> None:
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header(
            "Content-Security-Policy",
            "default-src 'self'; script-src 'self' 'unsafe-inline'; "
            "style-src 'self' 'unsafe-inline'; connect-src 'self'; "
            "img-src 'none'; object-src 'none'; base-uri 'none'; "
            "frame-ancestors 'none'",
        )

    @staticmethod
    def _single_positive_int(
        parameters: dict[str, list[str]], key: str, default: int | None = None
    ) -> int:
        values = parameters.get(key)
        if values is None:
            if default is not None:
                return default
            raise RequestError(f"缺少参数：{key}")
        if len(values) != 1:
            raise RequestError(f"参数只能出现一次：{key}")
        try:
            value = int(values[0])
        except ValueError as error:
            raise RequestError(f"参数必须是正整数：{key}") from error
        if value <= 0:
            raise RequestError(f"参数必须是正整数：{key}")
        return value

    def log_message(self, format_string: str, *args: Any) -> None:
        print(
            f"{self.address_string()} - "
            f"[{self.log_date_time_string()}] {format_string % args}"
        )


def _port(value: str) -> int:
    try:
        port = int(value)
    except ValueError as error:
        raise argparse.ArgumentTypeError("端口必须是整数") from error
    if not 1 <= port <= 65535:
        raise argparse.ArgumentTypeError("端口必须在 1–65535 之间")
    return port


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="启动 FolderVisualizer 大目录按需浏览服务"
    )
    parser.add_argument("--port", type=_port, default=DEFAULT_PORT)
    parser.add_argument(
        "--database", type=Path, default=DEFAULT_DATABASE_PATH
    )
    parser.add_argument("--html", type=Path, default=DEFAULT_HTML_PATH)
    return parser.parse_args()


def create_server(
    database_path: Path = DEFAULT_DATABASE_PATH,
    html_path: Path = DEFAULT_HTML_PATH,
    port: int = DEFAULT_PORT,
) -> LargeBrowserServer:
    """Create a server bound exclusively to the IPv4 loopback address."""

    folder_index = FolderIndex(database_path)
    resolved_html_path = html_path.resolve()
    if not resolved_html_path.is_file():
        raise FileNotFoundError(f"浏览页面不存在：{resolved_html_path}")

    handler_class = type(
        "ConfiguredLargeBrowserHandler",
        (LargeBrowserHandler,),
        {"folder_index": folder_index, "html_path": resolved_html_path},
    )
    return LargeBrowserServer((LOOPBACK_HOST, port), handler_class)


def main() -> int:
    args = _parse_args()
    try:
        server = create_server(args.database, args.html, args.port)
    except (OSError, ValueError, sqlite3.Error) as error:
        print(f"错误：无法启动大目录浏览服务：{error}")
        return 1

    url = f"http://{LOOPBACK_HOST}:{server.server_port}/"
    print("FolderVisualizer 大目录浏览服务已启动")
    print(f"地址：{url}")
    print(f"SQLite：{args.database.resolve()}")
    print("仅监听 127.0.0.1，不会上传文件或扫描数据")
    print("按 Ctrl+C 关闭服务")
    try:
        server.serve_forever(poll_interval=0.25)
    except KeyboardInterrupt:
        print("\n正在关闭本地服务……")
    finally:
        server.server_close()
    print("本地服务已关闭")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())