server.py

#!/usr/bin/env python3
"""Local, dependency-free editor for the legacy macOS Launchpad database."""

from __future__ import annotations

import argparse
import base64
import hashlib
import json
import math
import os
import plistlib
import shutil
import sqlite3
import subprocess
import sys
import tempfile
import threading
import time
import uuid
from datetime import datetime
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, quote, urlparse


BASE_DIR = Path(__file__).resolve().parent
APP_SUPPORT_DIR = Path(
    os.environ.get(
        "MAC_LAUNCHPAD_DATA_DIR",
        Path.home() / "Library/Application Support/Mac Launchpad Editor",
    )
)
BACKUP_DIR = APP_SUPPORT_DIR / "Backups"
CACHE_DIR = Path(
    os.environ.get(
        "MAC_LAUNCHPAD_CACHE_DIR",
        Path.home() / "Library/Caches/Mac Launchpad Editor",
    )
)
SETTINGS_PATH = APP_SUPPORT_DIR / "settings.json"
MAX_BODY = 8 * 1024 * 1024
LEGACY_ROWS = 5
LEGACY_COLUMNS = 7
POST_TOKEN = os.environ.get("MAC_LAUNCHPAD_EDITOR_TOKEN") or uuid.uuid4().hex
CATALOG_LOCK = threading.Lock()
ICON_LOCK = threading.Lock()
MUTATION_LOCK = threading.Lock()
CATALOG: dict[str, dict] | None = None


def migrate_legacy_data() -> None:
    """Copy pre-app backups/settings into Application Support once."""
    APP_SUPPORT_DIR.mkdir(parents=True, exist_ok=True)
    BACKUP_DIR.mkdir(parents=True, exist_ok=True)
    legacy_backup_dir = BASE_DIR / "备份"
    if legacy_backup_dir.is_dir() and legacy_backup_dir.resolve() != BACKUP_DIR.resolve():
        for source in legacy_backup_dir.iterdir():
            if source.is_file() and source.suffix in {".sqlite", ".json"}:
                destination = BACKUP_DIR / source.name
                if not destination.exists():
                    shutil.copy2(source, destination)
    legacy_settings = BASE_DIR / "editor-settings.json"
    if legacy_settings.is_file() and not SETTINGS_PATH.exists():
        shutil.copy2(legacy_settings, SETTINGS_PATH)


def discover_database() -> Path:
    # The per-user Darwin directory is the authoritative location. Returning it
    # immediately avoids accidentally selecting another user's or an abandoned
    # Launchpad database merely because it has a newer modification time.
    try:
        darwin_dir = subprocess.check_output(
            ["getconf", "DARWIN_USER_DIR"], text=True, stderr=subprocess.DEVNULL
        ).strip()
        if darwin_dir:
            current_user_database = Path(darwin_dir) / "com.apple.dock.launchpad/db/db"
            if current_user_database.is_file():
                return current_user_database
    except Exception:
        pass

    # Fallback for unusual environments where getconf is unavailable or its
    # expected database no longer exists. Prefer only readable regular files.
    existing = [
        path
        for path in Path("/private/var/folders").glob("*/*/0/com.apple.dock.launchpad/db/db")
        if path.is_file() and os.access(path, os.R_OK) and path.stat().st_uid == os.getuid()
    ]
    if not existing:
        raise FileNotFoundError("未找到 macOS 遗留 Launchpad 数据库")
    return max(existing, key=lambda p: p.stat().st_mtime)


def connect(path: Path, readonly: bool = False) -> sqlite3.Connection:
    if readonly:
        wal_path = Path(str(path) + "-wal")
        immutable = not wal_path.exists() or wal_path.stat().st_size == 0
        uri = f"file:{quote(str(path))}?mode=ro" + ("&immutable=1" if immutable else "")
        db = sqlite3.connect(uri, uri=True, timeout=5)
    else:
        db = sqlite3.connect(path, timeout=10)
    db.row_factory = sqlite3.Row
    return db


def load_editor_settings() -> dict:
    defaults = {
        "rows": LEGACY_ROWS,
        "columns": LEGACY_COLUMNS,
        "iconScale": 1,
        "paddingScale": 1,
    }
    try:
        saved = json.loads(SETTINGS_PATH.read_text("utf-8"))
        for key in ("iconScale", "paddingScale"):
            if key in saved:
                value = float(saved[key])
                if math.isfinite(value) and 0.5 <= value <= 2:
                    defaults[key] = value
    except Exception:
        pass
    return defaults


def save_editor_settings(layout: dict) -> None:
    payload = {
        "rows": LEGACY_ROWS,
        "columns": LEGACY_COLUMNS,
        "iconScale": layout["iconScale"],
        "paddingScale": layout["paddingScale"],
    }
    SETTINGS_PATH.write_text(json.dumps(payload, ensure_ascii=False, indent=2), "utf-8")


def app_record(db: sqlite3.Connection, item_id: int) -> dict:
    row = db.execute(
        "SELECT title,bundleid,storeid,category_id,moddate,bookmark FROM apps WHERE item_id=?",
        (item_id,),
    ).fetchone()
    if not row:
        return {"kind": "app", "bundleID": f"missing.item.{item_id}", "title": "缺失应用"}
    return {
        "kind": "app",
        "bundleID": row["bundleid"] or f"missing.item.{item_id}",
        "title": row["title"] or row["bundleid"] or "未知应用",
    }


def read_native_layout(path: Path | None = None) -> dict:
    path = path or discover_database()
    settings = load_editor_settings()
    with connect(path, readonly=True) as db:
        root_row = db.execute("SELECT value FROM dbinfo WHERE key='launchpad_root'").fetchone()
        root_id = int(root_row[0]) if root_row else 1
        page_rows = db.execute(
            "SELECT rowid FROM items WHERE type=3 AND parent_id=? AND uuid!='HOLDINGPAGE' "
            "ORDER BY ordering,rowid",
            (root_id,),
        ).fetchall()
        pages: list[list[dict]] = []
        warnings: list[str] = []
        for page_row in page_rows:
            page: list[dict] = []
            children = db.execute(
                "SELECT rowid,type FROM items WHERE parent_id=? ORDER BY ordering,rowid",
                (page_row["rowid"],),
            ).fetchall()
            for child in children:
                if child["type"] in (4, 5):
                    page.append(app_record(db, child["rowid"]))
                elif child["type"] == 2:
                    group = db.execute(
                        "SELECT title FROM groups WHERE item_id=?", (child["rowid"],)
                    ).fetchone()
                    folder_pages: list[list[dict]] = []
                    inner_pages = db.execute(
                        "SELECT rowid FROM items WHERE type=3 AND parent_id=? ORDER BY ordering,rowid",
                        (child["rowid"],),
                    ).fetchall()
                    for inner_page in inner_pages:
                        folder_children = db.execute(
                            "SELECT rowid,type FROM items WHERE parent_id=? ORDER BY ordering,rowid",
                            (inner_page["rowid"],),
                        ).fetchall()
                        folder_apps = []
                        for folder_child in folder_children:
                            if folder_child["type"] in (4, 5):
                                folder_apps.append(app_record(db, folder_child["rowid"]))
                            else:
                                warnings.append(
                                    "文件夹中存在未识别节点："
                                    f"item {folder_child['rowid']} / type {folder_child['type']}"
                                )
                        folder_pages.append(folder_apps)
                    page.append(
                        {
                            "kind": "folder",
                            "name": (group["title"] if group else None) or "未命名文件夹",
                            "pages": folder_pages or [[]],
                        }
                    )
                else:
                    warnings.append(f"页面中存在未识别节点：item {child['rowid']} / type {child['type']}")
            pages.append(page)
        integrity = db.execute("PRAGMA integrity_check").fetchone()[0]
        count = db.execute("SELECT COUNT(*) FROM apps").fetchone()[0]
    return {
        **settings,
        "pages": pages or [[]],
        "source": "macOS 原生遗留数据库",
        "databasePath": str(path),
        "databaseAppCount": count,
        "integrity": integrity,
        "warnings": warnings,
    }


def _preference(path: str) -> tuple[int, int, int]:
    p = path.lower()
    return (
        0 if p.startswith("/applications/") else 1,
        0 if "/contents/" not in p else 1,
        len(path),
    )


def scan_catalog(force: bool = False) -> dict[str, dict]:
    global CATALOG
    with CATALOG_LOCK:
        if CATALOG is not None and not force:
            return CATALOG
        paths: set[str] = set()
        try:
            output = subprocess.check_output(
                ["mdfind", "kMDItemContentType == 'com.apple.application-bundle'"],
                text=True,
                stderr=subprocess.DEVNULL,
                timeout=30,
            )
            paths.update(line for line in output.splitlines() if line.endswith(".app"))
        except Exception:
            pass
        application_roots = (
            Path("/Applications"),
            Path("/System/Applications"),
            Path.home() / "Applications",
        )
        for root in application_roots:
            if not root.exists():
                continue
            try:
                for current, directories, _files in os.walk(root):
                    app_directories = [name for name in directories if name.endswith(".app")]
                    for name in app_directories:
                        paths.add(str(Path(current) / name))
                    # Application bundles are packages. Do not expose their
                    # internal helpers as independently installed apps.
                    directories[:] = [
                        name for name in directories
                        if not name.endswith(".app") and not name.startswith(".")
                    ]
            except OSError:
                pass
        catalog: dict[str, dict] = {}
        for app_path in sorted(paths, key=_preference):
            info_path = Path(app_path) / "Contents/Info.plist"
            try:
                with info_path.open("rb") as handle:
                    info = plistlib.load(handle)
            except Exception:
                continue
            bundle_id = info.get("CFBundleIdentifier")
            if not isinstance(bundle_id, str) or not bundle_id:
                continue
            title = (
                info.get("CFBundleDisplayName")
                or info.get("CFBundleName")
                or Path(app_path).stem
            )
            version = info.get("CFBundleShortVersionString") or info.get("CFBundleVersion") or ""
            icon_name = info.get("CFBundleIconFile") or ""
            icon_path = None
            if icon_name:
                if not Path(icon_name).suffix:
                    icon_name += ".icns"
                candidate = Path(app_path) / "Contents/Resources" / icon_name
                if candidate.exists():
                    icon_path = str(candidate)
            app_path_object = Path(app_path)
            component_containers = {".bundle", ".framework", ".plugin", ".appex", ".xpc"}
            is_component = any(
                parent.suffix.lower() in component_containers
                for parent in app_path_object.parents
            )
            source_key = (
                "applications"
                if app_path.startswith("/Applications/")
                else "systemApps"
                if app_path.startswith("/System/Applications/")
                else "userApps"
                if app_path.startswith(str(Path.home() / "Applications") + "/")
                else "otherLocation"
            )
            record = {
                "bundleID": bundle_id,
                "title": str(title),
                "version": str(version),
                "path": app_path,
                "iconPath": icon_path,
                "nested": "/Contents/" in app_path,
                "component": is_component or bool(info.get("LSBackgroundOnly")),
                "background": bool(info.get("LSUIElement") or info.get("LSBackgroundOnly")),
                "sourceKey": source_key,
                "source": (
                    "Applications"
                    if app_path.startswith("/Applications/")
                    else "系统应用"
                    if app_path.startswith("/System/Applications/")
                    else "用户应用"
                    if app_path.startswith(str(Path.home() / "Applications") + "/")
                    else "其他位置"
                ),
            }
            old = catalog.get(bundle_id)
            if old is None or _preference(app_path) < _preference(old["path"]):
                catalog[bundle_id] = record
        CATALOG = catalog
        return catalog


def reveal_catalog_app(bundle_id: str) -> dict:
    if not isinstance(bundle_id, str) or not bundle_id:
        raise ValueError("缺少应用 Bundle ID")
    record = scan_catalog().get(bundle_id)
    if not record or not Path(record["path"]).is_dir():
        record = scan_catalog(force=True).get(bundle_id)
    if not record:
        raise ValueError("这个应用已不在已安装应用列表中")
    app_path = Path(record["path"])
    if not app_path.is_dir() or app_path.suffix.lower() != ".app":
        raise ValueError("应用路径已经失效")
    subprocess.run(
        ["open", "-R", str(app_path)],
        check=True,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        timeout=10,
    )
    return {"ok": True, "path": str(app_path), "title": record["title"]}


def validate_layout(raw: dict) -> tuple[dict, list[str]]:
    if not isinstance(raw, dict) or not isinstance(raw.get("pages"), list):
        raise ValueError("布局缺少 pages 数组")
    rows = LEGACY_ROWS
    columns = LEGACY_COLUMNS
    if not (1 <= len(raw["pages"]) <= 50):
        raise ValueError("页面数量必须为 1–50")
    seen: set[str] = set()
    warnings: list[str] = []
    clean_pages: list[list[dict]] = []
    app_count = 0

    def clean_app(app: dict) -> dict:
        nonlocal app_count
        bundle_id = app.get("bundleID")
        if not isinstance(bundle_id, str) or not bundle_id.strip():
            raise ValueError("发现缺少 Bundle ID 的应用")
        bundle_id = bundle_id.strip()
        if len(bundle_id) > 500:
            raise ValueError("Bundle ID 过长")
        if bundle_id in seen:
            raise ValueError(f"Bundle ID 重复：{bundle_id}")
        seen.add(bundle_id)
        app_count += 1
        if app_count > 3000:
            raise ValueError("应用数量超过安全上限")
        return {
            "kind": "app",
            "bundleID": bundle_id,
            "title": str(app.get("title") or bundle_id)[:500],
            "version": str(app.get("version") or "")[:100],
        }

    for page_index, page in enumerate(raw["pages"]):
        if not isinstance(page, list):
            raise ValueError(f"第 {page_index + 1} 页不是数组")
        clean_page: list[dict] = []
        for item in page:
            if not isinstance(item, dict):
                raise ValueError("布局节点格式错误")
            if item.get("kind") == "app":
                clean_page.append(clean_app(item))
            elif item.get("kind") == "folder":
                folder_pages = item.get("pages") or [[]]
                if not isinstance(folder_pages, list):
                    raise ValueError("文件夹 pages 格式错误")
                clean_folder_pages: list[list[dict]] = []
                for inner in folder_pages:
                    if not isinstance(inner, list):
                        raise ValueError("文件夹内页格式错误")
                    clean_folder_pages.append([clean_app(app) for app in inner])
                total = sum(len(p) for p in clean_folder_pages)
                if total == 0:
                    warnings.append(f"空文件夹“{item.get('name') or '未命名文件夹'}”可能被读取器忽略")
                clean_page.append(
                    {
                        "kind": "folder",
                        "name": str(item.get("name") or "未命名文件夹")[:500],
                        "pages": clean_folder_pages or [[]],
                    }
                )
            else:
                raise ValueError("发现未知布局节点")
        if len(clean_page) > rows * columns:
            raise ValueError(
                f"第 {page_index + 1} 页有 {len(clean_page)} 项；"
                f"每页最多只能写入 {rows * columns} 个网格位置"
            )
        clean_pages.append(clean_page)
    try:
        icon_scale = float(raw.get("iconScale", 1))
        padding_scale = float(raw.get("paddingScale", 1))
    except (TypeError, ValueError):
        raise ValueError("显示比例格式错误") from None
    if not math.isfinite(icon_scale) or not math.isfinite(padding_scale):
        raise ValueError("显示比例必须是有限数值")
    if not (0.5 <= icon_scale <= 2 and 0.5 <= padding_scale <= 2):
        raise ValueError("显示比例必须在 0.5–2 之间")
    clean = {
        "rows": rows,
        "columns": columns,
        "iconScale": icon_scale,
        "paddingScale": padding_scale,
        "pages": clean_pages,
    }
    return clean, warnings


def database_backup(db_path: Path, reason: str) -> Path:
    BACKUP_DIR.mkdir(parents=True, exist_ok=True)
    stamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
    safe_reason = "".join(c for c in reason if c.isalnum() or c in "-_ ")[:40].strip() or "backup"
    destination = BACKUP_DIR / f"Launchpad-{stamp}-{safe_reason}.sqlite"
    # A normal source connection lets SQLite create the shared-memory sidecar
    # when the source is a checkpointed WAL database copied for testing.
    with connect(db_path) as source, sqlite3.connect(destination) as target:
        source.backup(target)
    metadata = {
        "createdAt": datetime.now().isoformat(timespec="seconds"),
        "reason": reason,
        "source": str(db_path),
        "size": destination.stat().st_size,
    }
    destination.with_suffix(".json").write_text(
        json.dumps(metadata, ensure_ascii=False, indent=2), "utf-8"
    )
    return destination


def create_bookmarks(paths: list[str]) -> dict[str, bytes]:
    if not paths:
        return {}
    bundled_helper = BASE_DIR / "bookmark-helper"
    helper_source = BASE_DIR / "bookmark-helper.swift"
    helper_binary = bundled_helper if bundled_helper.is_file() else CACHE_DIR / "bookmark-helper"
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    try:
        if helper_binary == CACHE_DIR / "bookmark-helper" and (
            not helper_binary.exists() or helper_binary.stat().st_mtime < helper_source.stat().st_mtime
        ):
            module_cache = CACHE_DIR / "swift-module-cache"
            module_cache.mkdir(exist_ok=True)
            subprocess.run(
                [
                    "xcrun", "swiftc", str(helper_source),
                    "-module-cache-path", str(module_cache),
                    "-o", str(helper_binary),
                ],
                check=True,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                timeout=60,
            )
        result = subprocess.run(
            [str(helper_binary), *paths],
            check=True,
            text=True,
            capture_output=True,
            timeout=30,
        )
    except Exception:
        return {}
    bookmarks: dict[str, bytes] = {}
    for line in result.stdout.splitlines():
        try:
            path, encoded = line.split("\t", 1)
            bookmarks[path] = base64.b64decode(encoded)
        except Exception:
            continue
    return bookmarks


def apply_layout(db_path: Path, raw_layout: dict, create_backup: bool = True) -> dict:
    current_warnings = read_native_layout(db_path).get("warnings", [])
    if current_warnings:
        raise ValueError(
            "原生数据库包含当前版本无法安全保留的节点，已阻止写入："
            + "；".join(current_warnings[:5])
        )
    layout, warnings = validate_layout(raw_layout)
    backup_path = database_backup(db_path, "Before applying layout") if create_backup else None
    catalog = scan_catalog()
    try:
        db = connect(db_path)
        existing_rows = db.execute(
            "SELECT i.uuid,i.flags,a.*,ic.size_big AS cache_size_big,"
            "ic.size_mini AS cache_size_mini,ic.image_data AS cache_image_data,"
            "ic.image_data_mini AS cache_image_data_mini "
            "FROM apps a JOIN items i ON i.rowid=a.item_id "
            "LEFT JOIN image_cache ic ON ic.item_id=a.item_id "
            "ORDER BY (a.bookmark IS NOT NULL) DESC,a.item_id"
        ).fetchall()
        existing: dict[str, dict] = {}
        for row in existing_rows:
            if row["bundleid"] and row["bundleid"] not in existing:
                existing[row["bundleid"]] = dict(row)
        needed_paths = []
        for page in layout["pages"]:
            for item in page:
                apps = [item] if item["kind"] == "app" else [a for p in item["pages"] for a in p]
                for app in apps:
                    old = existing.get(app["bundleID"])
                    current = catalog.get(app["bundleID"])
                    if (not old or old.get("bookmark") is None) and current:
                        needed_paths.append(current["path"])
        bookmarks = create_bookmarks(sorted(set(needed_paths)))

        db.execute("BEGIN IMMEDIATE")
        db.execute("UPDATE dbinfo SET value=1 WHERE key='ignore_items_update_triggers'")
        keep_uuids = ("ROOTPAGE", "HOLDINGPAGE", "ROOTPAGE_VERS", "HOLDINGPAGE_VERS", "HOLDINGPAGE_DB")
        placeholders = ",".join("?" for _ in keep_uuids)
        keep_ids = [
            row[0]
            for row in db.execute(
                f"SELECT rowid FROM items WHERE uuid IN ({placeholders})", keep_uuids
            ).fetchall()
        ]
        db.execute("DELETE FROM apps")
        db.execute("DELETE FROM groups")
        db.execute("DELETE FROM downloading_apps")
        db.execute("DELETE FROM image_cache")
        if keep_ids:
            marks = ",".join("?" for _ in keep_ids)
            db.execute(f"DELETE FROM items WHERE rowid NOT IN ({marks})", keep_ids)
        else:
            raise RuntimeError("原生数据库缺少根节点，已中止")

        root_row = db.execute("SELECT value FROM dbinfo WHERE key='launchpad_root'").fetchone()
        root_id = int(root_row[0]) if root_row else 1

        def add_item(item_type: int, parent: int, ordering: int, flags: int = 0) -> int:
            cur = db.execute(
                "INSERT INTO items(uuid,flags,type,parent_id,ordering) VALUES(?,?,?,?,?)",
                (str(uuid.uuid4()).upper(), flags, item_type, parent, ordering),
            )
            return int(cur.lastrowid)

        def add_app(app: dict, parent: int, ordering: int) -> None:
            bundle_id = app["bundleID"]
            old = existing.get(bundle_id, {})
            current = catalog.get(bundle_id, {})
            item_id = add_item(4, parent, ordering, int(old.get("flags") or 0))
            title = current.get("title") or app.get("title") or old.get("title") or bundle_id
            path = current.get("path")
            bookmark = old.get("bookmark") or (bookmarks.get(path) if path else None)
            if bookmark is None:
                warnings.append(f"{title} 没有可用的文件书签；只保留 Bundle ID")
            db.execute(
                "INSERT INTO apps(item_id,title,bundleid,storeid,category_id,moddate,bookmark) "
                "VALUES(?,?,?,?,?,?,?)",
                (
                    item_id,
                    title,
                    bundle_id,
                    old.get("storeid"),
                    old.get("category_id"),
                    old.get("moddate") or (time.time() - 978307200),
                    bookmark,
                ),
            )
            if old.get("cache_image_data") is not None or old.get("cache_image_data_mini") is not None:
                db.execute(
                    "INSERT INTO image_cache(item_id,size_big,size_mini,image_data,image_data_mini) "
                    "VALUES(?,?,?,?,?)",
                    (
                        item_id,
                        old.get("cache_size_big") or 0,
                        old.get("cache_size_mini") or 0,
                        old.get("cache_image_data"),
                        old.get("cache_image_data_mini"),
                    ),
                )

        for page_index, page in enumerate(layout["pages"]):
            page_id = add_item(3, root_id, page_index + 1, 2)
            for item_index, item in enumerate(page):
                if item["kind"] == "app":
                    add_app(item, page_id, item_index)
                else:
                    folder_id = add_item(2, page_id, item_index, 0)
                    db.execute(
                        "INSERT INTO groups(item_id,category_id,title) VALUES(?,?,?)",
                        (folder_id, None, item["name"]),
                    )
                    for inner_index, inner_page in enumerate(item["pages"] or [[]]):
                        inner_id = add_item(3, folder_id, inner_index, 2)
                        for app_index, app in enumerate(inner_page):
                            add_app(app, inner_id, app_index)
        db.execute("UPDATE dbinfo SET value=0 WHERE key='ignore_items_update_triggers'")
        integrity = db.execute("PRAGMA integrity_check").fetchone()[0]
        if integrity != "ok":
            raise RuntimeError(f"数据库完整性检查失败：{integrity}")
        db.commit()
        try:
            db.execute("PRAGMA wal_checkpoint(TRUNCATE)")
        except sqlite3.Error:
            pass
        db.close()
        try:
            save_editor_settings(layout)
        except OSError as exc:
            warnings.append(f"布局已写入，但编辑器设置保存失败：{exc}")
    except Exception:
        try:
            db.rollback()
            db.close()
        except Exception:
            pass
        raise
    return {
        "ok": True,
        "backup": backup_path.name if backup_path else None,
        "warnings": sorted(set(warnings)),
        "layout": read_native_layout(db_path),
    }


def list_backups() -> list[dict]:
    BACKUP_DIR.mkdir(parents=True, exist_ok=True)
    results = []
    for path in BACKUP_DIR.glob("*.sqlite"):
        metadata = {}
        try:
            metadata = json.loads(path.with_suffix(".json").read_text("utf-8"))
        except Exception:
            pass
        created_at = metadata.get("createdAt") or datetime.fromtimestamp(path.stat().st_mtime).isoformat()
        try:
            sort_time = datetime.fromisoformat(created_at.replace("Z", "+00:00")).timestamp()
        except (AttributeError, TypeError, ValueError):
            sort_time = path.stat().st_mtime
        results.append(
            {
                "name": path.name,
                "createdAt": created_at,
                "reason": metadata.get("reason") or "Manual backup",
                "size": path.stat().st_size,
                "_sortTime": sort_time,
            }
        )
    results.sort(key=lambda item: item["_sortTime"], reverse=True)
    for item in results:
        item.pop("_sortTime", None)
    return results


def backup_path(name: str) -> Path:
    if not isinstance(name, str) or not name:
        raise ValueError("备份文件无效")
    candidate = (BACKUP_DIR / name).resolve()
    if candidate.parent != BACKUP_DIR.resolve() or candidate.suffix != ".sqlite" or not candidate.is_file():
        raise ValueError("备份文件无效")
    return candidate


def delete_backup(name: str) -> None:
    candidate = backup_path(name)
    metadata = candidate.with_suffix(".json")
    candidate.unlink()
    try:
        metadata.unlink()
    except FileNotFoundError:
        pass


def rename_backup(name: str, new_name: str) -> dict:
    source = backup_path(name)
    if not isinstance(new_name, str):
        raise ValueError("请输入新的备份名称")
    label = new_name.strip()
    if label.lower().endswith(".sqlite"):
        label = label[:-7].strip()
    if not label or len(label) > 100:
        raise ValueError("备份名称必须为 1–100 个字符")
    if any(ord(character) < 32 or character in "/\\:" for character in label):
        raise ValueError("备份名称不能包含 /、\\、: 或控制字符")
    safe_name = "".join(
        character for character in label
        if character.isalnum() or character in " -_().[]"
    ).strip(" .")
    if not safe_name:
        raise ValueError("备份名称无效")
    destination = BACKUP_DIR / f"{safe_name}.sqlite"
    destination_metadata = destination.with_suffix(".json")
    if destination != source and (destination.exists() or destination_metadata.exists()):
        raise ValueError("已经存在同名备份")

    source_metadata = source.with_suffix(".json")
    metadata = {}
    try:
        metadata = json.loads(source_metadata.read_text("utf-8"))
    except Exception:
        metadata = {
            "createdAt": datetime.fromtimestamp(source.stat().st_mtime).isoformat(timespec="seconds"),
            "source": "",
            "size": source.stat().st_size,
        }
    metadata["reason"] = label
    metadata["size"] = source.stat().st_size
    source_metadata.write_text(json.dumps(metadata, ensure_ascii=False, indent=2), "utf-8")

    if destination != source:
        os.replace(source, destination)
        try:
            os.replace(source_metadata, destination_metadata)
        except Exception:
            os.replace(destination, source)
            raise
    return {"ok": True, "name": destination.name, "reason": label}


def reveal_backup(name: str) -> dict:
    candidate = backup_path(name)
    subprocess.run(
        ["open", "-R", str(candidate)],
        check=True,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        timeout=10,
    )
    return {"ok": True, "path": str(candidate), "name": candidate.name}


def validate_backup_database(path: Path) -> None:
    with connect(path, readonly=True) as db:
        integrity = db.execute("PRAGMA integrity_check").fetchone()[0]
        if integrity != "ok":
            raise ValueError(f"备份数据库完整性检查失败：{integrity}")
        tables = {
            row[0]
            for row in db.execute(
                "SELECT name FROM sqlite_master WHERE type='table'"
            ).fetchall()
        }
        required = {"dbinfo", "items", "apps", "groups"}
        if not required.issubset(tables):
            raise ValueError("备份文件不是兼容的 Launchpad 数据库")


def replace_database_file(db_path: Path, candidate: Path, safety: Path) -> None:
    target_stat = db_path.stat()
    temp_path: Path | None = None
    replaced = False
    try:
        # Flush the current database before replacing its inode, so stale WAL
        # pages cannot be paired with the restored database file.
        with connect(db_path) as target:
            target.execute("PRAGMA wal_checkpoint(TRUNCATE)")

        descriptor, temp_name = tempfile.mkstemp(
            prefix=".launchpad-restore-", suffix=".sqlite", dir=db_path.parent
        )
        os.close(descriptor)
        temp_path = Path(temp_name)
        shutil.copyfile(candidate, temp_path)

        # The bytes come from the selected backup; filesystem identity follows
        # the existing target so third-party readers see the same access model.
        shutil.copystat(db_path, temp_path, follow_symlinks=False)
        try:
            os.chown(temp_path, target_stat.st_uid, target_stat.st_gid)
        except PermissionError:
            pass
        with temp_path.open("r+b") as handle:
            handle.flush()
            os.fsync(handle.fileno())
        validate_backup_database(temp_path)

        os.replace(temp_path, db_path)
        temp_path = None
        replaced = True
        for suffix in ("-wal", "-shm"):
            try:
                Path(str(db_path) + suffix).unlink()
            except FileNotFoundError:
                pass
        validate_backup_database(db_path)
    except Exception:
        if replaced:
            rollback = db_path.parent / f".launchpad-rollback-{uuid.uuid4().hex}.sqlite"
            try:
                shutil.copyfile(safety, rollback)
                shutil.copystat(db_path, rollback, follow_symlinks=False)
                os.replace(rollback, db_path)
                for suffix in ("-wal", "-shm"):
                    try:
                        Path(str(db_path) + suffix).unlink()
                    except FileNotFoundError:
                        pass
            finally:
                try:
                    rollback.unlink()
                except FileNotFoundError:
                    pass
        raise
    finally:
        if temp_path is not None:
            try:
                temp_path.unlink()
            except FileNotFoundError:
                pass


def restore_backup(
    db_path: Path,
    name: str,
    mode: str = "write",
    create_safety_backup: bool = True,
) -> dict:
    candidate = backup_path(name)
    if mode not in {"write", "replace"}:
        raise ValueError("未知的恢复模式")
    validate_backup_database(candidate)
    safety = database_backup(db_path, "Before restoring backup")
    try:
        if mode == "replace":
            replace_database_file(db_path, candidate, safety)
        else:
            try:
                with connect(candidate, readonly=True) as source, connect(db_path) as target:
                    source.backup(target)
                    target.execute("PRAGMA wal_checkpoint(TRUNCATE)")
            except Exception:
                with connect(safety, readonly=True) as source, connect(db_path) as target:
                    source.backup(target)
                    target.execute("PRAGMA wal_checkpoint(TRUNCATE)")
                raise
    except Exception:
        # Keep the safety copy when recovery itself fails so the user still has
        # a known-good database snapshot available for manual recovery.
        raise
    cleanup_warning = None
    if not create_safety_backup:
        try:
            delete_backup(safety.name)
        except OSError as exc:
            cleanup_warning = f"恢复已完成，但临时安全备份无法删除：{exc}"
    target_stat = db_path.stat()
    return {
        "ok": True,
        "mode": mode,
        "databasePath": str(db_path),
        "permissions": oct(target_stat.st_mode & 0o777),
        "owner": target_stat.st_uid,
        "group": target_stat.st_gid,
        "safetyBackup": safety.name if create_safety_backup or cleanup_warning else None,
        "warnings": [cleanup_warning] if cleanup_warning else [],
        "layout": read_native_layout(db_path),
    }


def icon_png(bundle_id: str) -> Path | None:
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    destination = CACHE_DIR / (hashlib.sha256(bundle_id.encode()).hexdigest() + ".png")
    if destination.exists():
        return destination
    try:
        with connect(discover_database(), readonly=True) as db:
            row = db.execute(
                "SELECT ic.image_data_mini,ic.image_data FROM image_cache ic "
                "JOIN apps a ON a.item_id=ic.item_id WHERE a.bundleid=? "
                "ORDER BY (ic.image_data_mini IS NOT NULL) DESC LIMIT 1",
                (bundle_id,),
            ).fetchone()
            if row:
                image = row["image_data_mini"] or row["image_data"]
                if image and bytes(image).startswith(b"\x89PNG"):
                    destination.write_bytes(image)
                    return destination
    except Exception:
        pass
    record = scan_catalog().get(bundle_id)
    if not record:
        return None
    if record.get("iconPath"):
        try:
            subprocess.run(
                ["sips", "-s", "format", "png", record["iconPath"], "--out", str(destination)],
                check=True,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                timeout=15,
            )
            if destination.exists():
                return destination
        except Exception:
            pass
    # Apps that store icons in Assets.car do not expose an .icns file. Ask
    # NSWorkspace for the same icon Finder uses.
    with ICON_LOCK:
        if destination.exists():
            return destination
        bundled_helper = BASE_DIR / "icon-helper"
        helper_source = BASE_DIR / "icon-helper.swift"
        helper_binary = bundled_helper if bundled_helper.is_file() else CACHE_DIR / "icon-helper"
        module_cache = CACHE_DIR / "swift-module-cache"
        module_cache.mkdir(exist_ok=True)
        try:
            if helper_binary == CACHE_DIR / "icon-helper" and (
                not helper_binary.exists() or helper_binary.stat().st_mtime < helper_source.stat().st_mtime
            ):
                subprocess.run(
                    [
                        "xcrun", "swiftc", str(helper_source),
                        "-module-cache-path", str(module_cache),
                        "-o", str(helper_binary),
                    ],
                    check=True,
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    timeout=60,
                )
            subprocess.run(
                [str(helper_binary), record["path"], str(destination)],
                check=True,
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                timeout=15,
            )
            return destination if destination.exists() else None
        except Exception:
            return None


class Handler(BaseHTTPRequestHandler):
    server_version = "LaunchpadEditor/1.0"

    def log_message(self, fmt: str, *args) -> None:
        sys.stdout.write("[%s] %s\n" % (self.log_date_time_string(), fmt % args))

    def send_security_headers(self) -> None:
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header("Cross-Origin-Resource-Policy", "same-origin")
        self.send_header(
            "Content-Security-Policy",
            "default-src 'self'; img-src 'self' data:; style-src 'self'; "
            "script-src 'self'; connect-src 'self'; base-uri 'none'; frame-ancestors 'none'",
        )

    def reject_untrusted_origin(self) -> bool:
        port = self.server.server_address[1]
        allowed_hosts = {f"127.0.0.1:{port}", f"localhost:{port}"}
        host = self.headers.get("Host", "")
        origin = self.headers.get("Origin")
        allowed_origins = {f"http://127.0.0.1:{port}", f"http://localhost:{port}"}
        if host not in allowed_hosts or (origin is not None and origin not in allowed_origins):
            self.send_error_json("请求来源无效", 403)
            return True
        return False

    def send_json(self, payload: object, status: int = 200) -> 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.send_security_headers()
        self.end_headers()
        self.wfile.write(data)

    def send_error_json(self, message: str, status: int = 400) -> None:
        self.send_json({"ok": False, "error": message}, status)

    def do_GET(self) -> None:
        if self.reject_untrusted_origin():
            return
        parsed = urlparse(self.path)
        try:
            if parsed.path == "/api/layout":
                self.send_json(read_native_layout())
                return
            if parsed.path == "/api/catalog":
                force = parse_qs(parsed.query).get("refresh", ["0"])[0] == "1"
                data = [dict(item) for item in scan_catalog(force=force).values()]
                for item in data:
                    item = item.pop("iconPath", None)
                self.send_json({"apps": data})
                return
            if parsed.path == "/api/backups":
                self.send_json({"backups": list_backups()})
                return
            if parsed.path == "/api/icon":
                bundle_id = parse_qs(parsed.query).get("bundleID", [""])[0]
                icon = icon_png(bundle_id)
                if not icon:
                    self.send_response(HTTPStatus.NOT_FOUND)
                    self.end_headers()
                    return
                data = icon.read_bytes()
                self.send_response(200)
                self.send_header("Content-Type", "image/png")
                self.send_header("Content-Length", str(len(data)))
                self.send_header("Cache-Control", "public, max-age=86400")
                self.send_security_headers()
                self.end_headers()
                self.wfile.write(data)
                return
            self.send_error_json("接口不存在", 404)
        except FileNotFoundError as exc:
            self.send_error_json(str(exc), 404)
        except Exception as exc:
            self.send_error_json(str(exc), 500)

    def read_body(self) -> dict:
        length = int(self.headers.get("Content-Length", "0"))
        if length <= 0 or length > MAX_BODY:
            raise ValueError("请求内容大小不正确")
        raw = self.rfile.read(length)
        return json.loads(raw.decode("utf-8"))

    def do_POST(self) -> None:
        if self.reject_untrusted_origin():
            return
        if self.headers.get("X-Editor-Token") != POST_TOKEN:
            self.send_error_json("安全令牌无效", 403)
            return
        parsed = urlparse(self.path)
        mutating = parsed.path in {
            "/api/apply", "/api/backup", "/api/delete-backup", "/api/rename-backup", "/api/restore"
        }
        lock_acquired = False
        if mutating:
            lock_acquired = MUTATION_LOCK.acquire(blocking=False)
            if not lock_acquired:
                self.send_error_json("另一个数据库操作正在进行，请稍后重试", 409)
                return
        try:
            body = self.read_body()
            if parsed.path == "/api/validate":
                clean, warnings = validate_layout(body.get("layout"))
                self.send_json({"ok": True, "warnings": warnings, "layout": clean})
                return
            if parsed.path == "/api/apply":
                if body.get("confirmToken") != "APPLY_NATIVE_LAYOUT":
                    raise ValueError("缺少写入确认")
                self.send_json(
                    apply_layout(
                        discover_database(),
                        body.get("layout"),
                        body.get("createBackup", True) is not False,
                    )
                )
                return
            if parsed.path == "/api/backup":
                path = database_backup(discover_database(), "Manual backup")
                self.send_json({"ok": True, "backup": path.name})
                return
            if parsed.path == "/api/delete-backup":
                delete_backup(body.get("name", ""))
                self.send_json({"ok": True})
                return
            if parsed.path == "/api/rename-backup":
                self.send_json(rename_backup(body.get("name", ""), body.get("newName", "")))
                return
            if parsed.path == "/api/reveal-backup":
                self.send_json(reveal_backup(body.get("name", "")))
                return
            if parsed.path == "/api/reveal":
                self.send_json(reveal_catalog_app(body.get("bundleID", "")))
                return
            if parsed.path == "/api/restore":
                if body.get("confirmToken") != "RESTORE_NATIVE_LAYOUT":
                    raise ValueError("缺少恢复确认")
                self.send_json(
                    restore_backup(
                        discover_database(),
                        body.get("name", ""),
                        body.get("mode", "write"),
                        body.get("createSafetyBackup", True) is not False,
                    )
                )
                return
            self.send_error_json("接口不存在", 404)
        except (ValueError, json.JSONDecodeError) as exc:
            self.send_error_json(str(exc), 400)
        except sqlite3.Error as exc:
            self.send_error_json(f"数据库操作失败：{exc}", 500)
        except Exception as exc:
            self.send_error_json(str(exc), 500)
        finally:
            if lock_acquired:
                MUTATION_LOCK.release()


def self_test() -> int:
    global BACKUP_DIR, SETTINGS_PATH
    db_path = discover_database()
    layout = read_native_layout(db_path)
    clean, warnings = validate_layout(layout)
    overfull = json.loads(json.dumps(clean))
    while len(overfull["pages"][0]) <= LEGACY_ROWS * LEGACY_COLUMNS:
        number = len(overfull["pages"][0])
        overfull["pages"][0].append(
            {
                "kind": "app",
                "bundleID": f"com.leosu.selftest.overflow.{number}",
                "title": f"Overflow {number}",
            }
        )
    try:
        validate_layout(overfull)
        raise AssertionError("over-capacity page was accepted")
    except ValueError as exc:
        assert "最多只能写入" in str(exc)
    original_backup_dir = BACKUP_DIR
    original_settings_path = SETTINGS_PATH
    with tempfile.TemporaryDirectory() as tmp:
        temporary_root = Path(tmp)
        BACKUP_DIR = temporary_root / "backups"
        SETTINGS_PATH = temporary_root / "settings.json"
        try:
            test_db = temporary_root / "db"
            no_backup_db = temporary_root / "db-no-backup"
            with connect(db_path, readonly=True) as src, sqlite3.connect(test_db) as dst:
                src.backup(dst)
            shutil.copyfile(test_db, no_backup_db)
            with connect(test_db, readonly=True) as copied:
                original_image_cache_count = copied.execute("SELECT COUNT(*) FROM image_cache").fetchone()[0]
            result = apply_layout(test_db, clean)
            no_backup_result = apply_layout(no_backup_db, clean, create_backup=False)
            reread = result["layout"]
            assert len(reread["pages"]) == len(clean["pages"])
            expected = [
                item["bundleID"]
                for page in clean["pages"]
                for node in page
                for item in ([node] if node["kind"] == "app" else [a for p in node["pages"] for a in p])
            ]
            actual = [
                item["bundleID"]
                for page in reread["pages"]
                for node in page
                for item in ([node] if node["kind"] == "app" else [a for p in node["pages"] for a in p])
            ]
            assert expected == actual
            assert result["backup"]
            assert no_backup_result["backup"] is None
            assert result["layout"]["integrity"] == "ok"
            with connect(test_db, readonly=True) as rewritten:
                rewritten_image_cache_count = rewritten.execute("SELECT COUNT(*) FROM image_cache").fetchone()[0]
            if original_image_cache_count:
                assert rewritten_image_cache_count > 0

            renamed = rename_backup(result["backup"], "Renamed self-test backup")
            assert renamed["name"] == "Renamed self-test backup.sqlite"
            assert backup_path(renamed["name"]).is_file()
            assert json.loads(backup_path(renamed["name"]).with_suffix(".json").read_text("utf-8"))["reason"] == "Renamed self-test backup"

            restore_target = temporary_root / "restore-target"
            shutil.copyfile(test_db, restore_target)
            before_restore_count = len(list_backups())
            restored_without_backup = restore_backup(
                restore_target, renamed["name"], "write", create_safety_backup=False
            )
            assert restored_without_backup["safetyBackup"] is None
            assert len(list_backups()) == before_restore_count

            replace_target = temporary_root / "replace-target"
            shutil.copyfile(test_db, replace_target)
            restored_with_backup = restore_backup(
                replace_target, renamed["name"], "replace", create_safety_backup=True
            )
            assert restored_with_backup["safetyBackup"]
            assert len(list_backups()) == before_restore_count + 1
        finally:
            BACKUP_DIR = original_backup_dir
            SETTINGS_PATH = original_settings_path
    print(json.dumps({"ok": True, "pages": len(clean["pages"]), "warnings": warnings}, ensure_ascii=False))
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description="Mac Launchpad Editor")
    parser.add_argument("--port", type=int, default=8765)
    parser.add_argument("--app-mode", action="store_true")
    parser.add_argument("--self-test", action="store_true")
    args = parser.parse_args()
    if args.self_test:
        return self_test()
    migrate_legacy_data()
    BACKUP_DIR.mkdir(parents=True, exist_ok=True)
    CACHE_DIR.mkdir(parents=True, exist_ok=True)
    server = ThreadingHTTPServer(("127.0.0.1", args.port), Handler)
    actual_port = server.server_address[1]
    url = f"http://127.0.0.1:{actual_port}"
    print(f"READY {url}", flush=True)
    if not args.app_mode:
        print(f"Mac Launchpad Editor: {url}")
        print("Press Control-C to stop.")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
    return 0


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