server.py

# -*- coding: utf-8 -*-
"""Seiton (星頓) - local server. Formerly MyHub v2. Stdlib only (http.server + sqlite3).

Usage: python server.py [--port 8768] [--no-browser]
"""
import json
import os
import re
import secrets
import subprocess
import sys
import threading
import webbrowser
import urllib.parse
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

import store
import suggest
import seed_demo
import launcher

def _resource_base_dir():
    if getattr(sys, "frozen", False):
        return getattr(sys, "_MEIPASS", os.path.dirname(os.path.abspath(sys.executable)))
    return os.path.dirname(os.path.abspath(__file__))


BASE_DIR = _resource_base_dir()
STATIC_DIR = os.path.join(BASE_DIR, "static")
DEFAULT_PORT = 8768

MIME = {
    ".html": "text/html; charset=utf-8",
    ".js": "text/javascript; charset=utf-8",
    ".css": "text/css; charset=utf-8",
    ".svg": "image/svg+xml",
    ".png": "image/png",
    ".ico": "image/x-icon",
    ".woff2": "font/woff2",
}


class RawResponse:
    def __init__(self, data, content_type, filename=None, code=200):
        self.data = data
        self.content_type = content_type
        self.filename = filename
        self.code = code


def today_stamp():
    return datetime.now().strftime("%Y%m%d")


# ---------------------------------------------------------------- scan

def scan_directory(conn, root, recursive=False, include_files=True, max_items=500, scan_depth=4):
    """Import a real directory as items. Folders become 'folder' items,
    files become 'file' items, containment becomes part_of relations.
    Only paths under a registered permitted root are ever read."""
    root = os.path.normpath(root)
    permit = store.root_of_path(conn, root)
    if not permit:
        if not store.all_roots(conn):
            return {"error": "err.no_roots"}
        return {"error": "err.root_outside"}
    if not os.path.isdir(root):
        return {"error": "err.folder_missing: " + root}

    created, linked, skipped = 0, 0, 0
    new_ids = []

    def ensure_folder_item(path):
        nonlocal created
        found = store.find_item_by_path(conn, path)
        if found:
            return found["id"], False
        iid = store.create_item(conn, kind="folder",
                                title=os.path.basename(path) or path, path=path)
        created += 1
        new_ids.append(iid)
        return iid, True

    root_id, _ = ensure_folder_item(root)

    def walk(dir_path, dir_id, depth):
        nonlocal created, linked, skipped
        try:
            entries = sorted(os.scandir(dir_path), key=lambda e: e.name.lower())
        except OSError:
            return
        for e in entries:
            if created >= max_items:
                return
            if e.name.startswith(".") or e.name.startswith("~$"):
                continue
            path = os.path.normpath(e.path)
            if e.is_dir():
                fid, is_new = ensure_folder_item(path)
                if store.add_relation(conn, fid, dir_id, "part_of"):
                    linked += 1
                if recursive and depth < scan_depth:
                    walk(path, fid, depth + 1)
            elif include_files:
                found = store.find_item_by_path(conn, path)
                if found:
                    skipped += 1
                    fid = found["id"]
                else:
                    fid = store.create_item(conn, kind="file", title=e.name, path=path)
                    created += 1
                    new_ids.append(fid)
                if store.add_relation(conn, fid, dir_id, "part_of"):
                    linked += 1

    walk(root, root_id, 0)

    # 全自動モード: 取り込んだものにタグ・状態の提案を自動適用（関係は既に part_of で編み済み）
    auto_applied = 0
    if store.get_setting(conn, "auto_apply", "0") == "1":
        for iid in new_ids[:80]:
            sug = suggest.heuristics(conn, iid)
            sug["relations"] = []  # scan already weaves part_of; keep auto writes modest
            r = suggest.apply_suggestions(conn, iid, sug)
            auto_applied += r["tags"]
    return {"root_id": root_id, "created": created, "linked": linked,
            "skipped": skipped, "auto_tags": auto_applied}


# ---------------------------------------------------------------- handler

ROUTES = []


def route(method, pattern):
    rx = re.compile("^" + pattern + "$")

    def deco(fn):
        ROUTES.append((method, rx, fn))
        return fn
    return deco


def read_settings(conn):
    ai_mode = store.get_setting(conn, "ai_mode", "cli")
    if ai_mode not in ("cli", "manual", "off"):
        ai_mode = "cli"
    return {
        "auto_apply": store.get_setting(conn, "auto_apply", "0"),
        "reduce_motion": store.get_setting(conn, "reduce_motion", "0"),
        "ai_mode": ai_mode,
        "ai_command": store.get_setting(conn, "ai_command", suggest.DEFAULT_AI_COMMAND),
        "demo_mode": store.get_setting(conn, "demo_mode", "0"),
        "onboarding_started": store.get_setting(conn, "onboarding_started", "0"),
        "onboarding_ai_done": store.get_setting(conn, "onboarding_ai_done", "0"),
        "scan_max_items": store.get_setting(conn, "scan_max_items", "500"),
        "scan_depth": store.get_setting(conn, "scan_depth", "4"),
        "language": store.get_setting(conn, "language", ""),
    }


@route("GET", r"/api/boot")
def api_boot(conn, m, body, qs):
    settings = read_settings(conn)
    return {"tags": store.all_tags(conn),
            "relation_types": store.relation_types(conn),
            "counts": store.counts(conn),
            "roots": store.all_roots(conn),
            "settings": settings,
            "backup": store.backup_info(),
            "path_health": store.cached_path_health(conn),
            "undo": store.last_undo(conn)}


@route("GET", r"/api/roots")
def api_roots(conn, m, body, qs):
    return {"roots": store.all_roots(conn)}


@route("POST", r"/api/roots")
def api_roots_post(conn, m, body, qs):
    r = store.add_root(conn, body.get("path", ""), body.get("mode", "names"))
    if r.get("error"):
        return r
    return {"ok": True, "roots": store.all_roots(conn)}


@route("DELETE", r"/api/roots/(\d+)")
def api_roots_delete(conn, m, body, qs):
    store.delete_root(conn, int(m.group(1)))
    return {"ok": True, "roots": store.all_roots(conn)}


@route("POST", r"/api/settings")
def api_settings_post(conn, m, body, qs):
    for k, v in (body or {}).items():
        if k in ("auto_apply", "reduce_motion"):
            store.set_setting(conn, k, "1" if str(v) in ("1", "true", "True") else "0")
        elif k == "ai_mode":
            mode = str(v)
            if mode in ("cli", "manual", "off"):
                store.set_setting(conn, k, mode)
        elif k == "ai_command":
            store.set_setting(conn, k, str(v).strip())
        elif k in ("scan_max_items", "scan_depth"):
            try:
                n = int(v)
            except (TypeError, ValueError):
                continue
            if k == "scan_max_items":
                n = max(1, min(5000, n))
            else:
                n = max(0, min(12, n))
            store.set_setting(conn, k, str(n))
        elif k == "language":
            lang = str(v or "").strip().lower()
            store.set_setting(conn, k, "ja" if lang.startswith("ja") else "en")
    return {"settings": read_settings(conn)}


@route("POST", r"/api/onboarding")
def api_onboarding(conn, m, body, qs):
    if body.get("started"):
        store.set_setting(conn, "onboarding_started", "1")
    ai = body.get("ai")
    if ai == "cli":
        store.set_setting(conn, "ai_mode", "cli")
        store.set_setting(conn, "onboarding_ai_done", "1")
    elif ai == "manual":
        store.set_setting(conn, "ai_mode", "manual")
        store.set_setting(conn, "onboarding_ai_done", "1")
    elif ai == "skip":
        store.set_setting(conn, "ai_mode", "off")
        store.set_setting(conn, "onboarding_ai_done", "1")
    return {"settings": read_settings(conn)}


@route("POST", r"/api/newfolder")
def api_newfolder(conn, m, body, qs):
    """Create a NEW folder (never touches existing ones), only under a permitted root."""
    parent = os.path.normpath(body.get("parent", "").strip())
    name = re.sub(r'[<>:"/\\|?*]', "", body.get("name", "").strip())[:80]
    if not name:
        return {"error": "err.folder_name_empty"}
    if not store.root_of_path(conn, parent):
        return {"error": "err.create_outside_root"}
    if not os.path.isdir(parent):
        return {"error": "err.parent_missing"}
    target = os.path.join(parent, name)
    if os.path.exists(target):
        return {"error": "err.folder_exists"}
    try:
        os.mkdir(target)
    except OSError as e:
        return {"error": f"err.folder_create_failed: {e}"}
    iid = store.create_item(conn, kind="folder", title=name, path=target,
                            tags=body.get("tags") or [])
    parent_item = store.find_item_by_path(conn, parent)
    if parent_item:
        store.add_relation(conn, iid, parent_item["id"], "part_of")
    return {"ok": True, "id": iid}


@route("GET", r"/api/home")
def api_home(conn, m, body, qs):
    trails = store.home_trails(conn)
    trail_ids = set()
    for t in trails:
        trail_ids.add(t["current"]["id"])
    return {"trails": trails,
            "recent": store.recent_items(conn, 10, exclude=trail_ids),
            "inbox": store.inbox_items(conn, 8),
            "counts": store.counts(conn)}


@route("POST", r"/api/demo")
def api_demo(conn, m, body, qs):
    existing = store.counts(conn)["items"]
    if existing and not body.get("force"):
        return {"error": "err.demo_not_empty"}
    if existing:
        store.clear_catalog(conn)
    seed_demo.seed(conn)
    store.set_setting(conn, "demo_mode", "1")
    store.set_setting(conn, "onboarding_started", "1")
    return {"ok": True, "counts": store.counts(conn), "settings": read_settings(conn)}


@route("POST", r"/api/demo/reset")
def api_demo_reset(conn, m, body, qs):
    if store.get_setting(conn, "demo_mode", "0") != "1" and not body.get("force"):
        return {"error": "err.not_demo_mode"}
    store.clear_catalog(conn)
    store.set_setting(conn, "onboarding_started", "1")
    return {"ok": True, "counts": store.counts(conn), "settings": read_settings(conn)}


@route("GET", r"/api/inbox")
def api_inbox(conn, m, body, qs):
    return {"items": store.inbox_items(conn, 50)}


@route("GET", r"/api/overview")
def api_overview(conn, m, body, qs):
    return store.overview(conn, blackhole_mode=qs.get("blackhole") in ("1", "true"))


@route("GET", r"/api/export")
def api_export(conn, m, body, qs):
    data = json.dumps(store.export_json(conn), ensure_ascii=False, indent=2).encode("utf-8")
    return RawResponse(data, "application/json; charset=utf-8",
                       f"seiton-export-{today_stamp()}.json")


@route("POST", r"/api/export/markdown")
def api_export_markdown(conn, m, body, qs):
    return store.export_markdown(conn, body.get("dir", ""))


@route("GET", r"/api/export/zip")
def api_export_zip(conn, m, body, qs):
    return RawResponse(store.export_zip(conn), "application/zip",
                       f"seiton-export-{today_stamp()}.zip")


@route("GET", r"/api/export/md\.zip")
def api_export_md_zip(conn, m, body, qs):
    return RawResponse(store.export_markdown_zip(conn), "application/zip",
                       f"seiton-md-{today_stamp()}.zip")


@route("GET", r"/api/health/paths")
def api_health_paths(conn, m, body, qs):
    return store.check_paths(conn)


@route("POST", r"/api/undo")
def api_undo(conn, m, body, qs):
    return store.undo_last(conn)


@route("GET", r"/api/items")
def api_items(conn, m, body, qs):
    return {"items": store.search_items(conn, qs.get("q", ""),
                                        tag=qs.get("tag") or None,
                                        kind=qs.get("kind") or None,
                                        status=qs.get("status") or None)}


@route("POST", r"/api/items")
def api_items_post(conn, m, body, qs):
    title = (body.get("title") or "").strip()
    if not title:
        return {"error": "err.title_required"}
    iid = store.create_item(conn, kind=body.get("kind", "note"), title=title,
                            body=body.get("body", ""), url=body.get("url", ""),
                            path=body.get("path", ""), tags=body.get("tags"))
    auto = None
    if store.get_setting(conn, "auto_apply", "0") == "1":
        store.record_undo_snapshot(conn, "undo.auto_apply", [iid])
        auto = suggest.apply_suggestions(conn, iid, suggest.heuristics(conn, iid))
    return {"id": iid, "item": store.get_item(conn, iid), "auto": auto}


@route("GET", r"/api/items/(\d+)")
def api_item_get(conn, m, body, qs):
    iid = int(m.group(1))
    item = store.get_item(conn, iid)
    if not item:
        return {"error": "err.not_found"}
    item["tags"] = store.item_tags_map(conn, [iid]).get(iid, [])
    item["missing"] = iid in store.missing_ids(conn)
    rels = []
    for r in store.relations_of(conn, iid):
        other_id = r["dst"] if r["src"] == iid else r["src"]
        other = store.get_item(conn, other_id)
        if not other:
            continue
        rels.append({"rel_id": r["id"], "type": r["type"], "type_label": r["type_label"],
                     "zone": r["zone"], "outgoing": r["src"] == iid,
                     "other": {"id": other["id"], "title": other["title"],
                               "kind": other["kind"], "status": other["status"]}})
    item["relations"] = rels
    return {"item": item}


@route("PATCH", r"/api/items/(\d+)")
def api_item_patch(conn, m, body, qs):
    iid = int(m.group(1))
    store.record_undo_snapshot(conn, "undo.update", [iid])
    store.update_item(conn, iid, body or {})
    return {"ok": True}


@route("DELETE", r"/api/items/(\d+)")
def api_item_delete(conn, m, body, qs):
    iid = int(m.group(1))
    store.record_undo_snapshot(conn, "undo.delete", [iid])
    store.delete_item(conn, iid)
    return {"ok": True}


@route("POST", r"/api/blackhole")
def api_blackhole(conn, m, body, qs):
    iid = int(body.get("id") or 0)
    if not iid or not store.get_item(conn, iid):
        return {"error": "err.not_found"}
    ids = store.blackhole_targets(conn, iid, include_children=bool(body.get("include_children")))
    store.record_undo_snapshot(conn, "undo.blackhole", ids)
    count = store.send_to_blackhole(conn, ids)
    return {"ok": True, "count": count, "ids": ids}


@route("POST", r"/api/items/(\d+)/touch")
def api_item_touch(conn, m, body, qs):
    store.touch_item(conn, int(m.group(1)))
    return {"ok": True}


@route("POST", r"/api/items/(\d+)/tags")
def api_item_tags(conn, m, body, qs):
    iid = int(m.group(1))
    store.record_undo_snapshot(conn, "undo.tags", [iid])
    store.assign_tags(conn, iid,
                      add=body.get("add"), remove=body.get("remove"))
    return {"ok": True, "tags": store.item_tags_map(conn, [iid]).get(iid, [])}


@route("POST", r"/api/relations")
def api_rel_post(conn, m, body, qs):
    src, dst = int(body["src"]), int(body["dst"])
    rtype = body.get("type", "related")
    replace_parent = bool(body.get("replace_parent")) and rtype == "part_of"
    created_ids = {int(i) for i in (body.get("created_ids") or []) if i}
    snap_ids = {src, dst} - created_ids
    old_parent_rels = []
    if replace_parent:
        old_parent_rels = conn.execute(
            "SELECT r.id,r.dst FROM relations r "
            "JOIN relation_types rt ON rt.name=r.type "
            "WHERE rt.zone='hierarchy' AND r.src=?", (src,)).fetchall()
        snap_ids.update(r["dst"] for r in old_parent_rels)
    store.record_undo_snapshot(conn, "undo.relation_parent" if replace_parent else "undo.relation_create",
                               snap_ids, created_ids=created_ids)
    for r in old_parent_rels:
        store.delete_relation(conn, r["id"])
    rid = store.add_relation(conn, src, dst, rtype, body.get("note", ""))
    return {"id": rid}


@route("DELETE", r"/api/relations/(\d+)")
def api_rel_delete(conn, m, body, qs):
    rel_id = int(m.group(1))
    row = conn.execute("SELECT src,dst FROM relations WHERE id=?", (rel_id,)).fetchone()
    if row:
        store.record_undo_snapshot(conn, "undo.relation_delete", [row["src"], row["dst"]])
    store.delete_relation(conn, rel_id)
    return {"ok": True}


@route("POST", r"/api/reltypes")
def api_reltype_post(conn, m, body, qs):
    name = store.ensure_relation_type(conn, body.get("name", ""),
                                      label=body.get("label"),
                                      zone=body.get("zone", "orbit"))
    if not name:
        return {"error": "invalid name"}
    return {"ok": True, "relation_types": store.relation_types(conn)}


@route("GET", r"/api/graph/(\d+)")
def api_graph(conn, m, body, qs):
    g = store.graph_of(conn, int(m.group(1)))
    return g if g else {"error": "not found"}


@route("GET", r"/api/suggest/(\d+)")
def api_suggest(conn, m, body, qs):
    return suggest.heuristics(conn, int(m.group(1)))


@route("POST", r"/api/ai/test")
def api_ai_test(conn, m, body, qs):
    command = body.get("ai_command")
    if command is None:
        command = store.get_setting(conn, "ai_command", suggest.DEFAULT_AI_COMMAND)
    return suggest.test_ai_command(str(command))


@route("GET", r"/api/ai/(\d+)/prompt")
def api_ai_prompt(conn, m, body, qs):
    prompt = suggest.build_ai_prompt(conn, int(m.group(1)))
    if prompt is None:
        return {"error": "item not found"}
    return {"prompt": prompt}


@route("POST", r"/api/ai/(\d+)/manual")
def api_ai_manual(conn, m, body, qs):
    return suggest.parse_ai_text(conn, int(m.group(1)), body.get("text", ""))


@route("POST", r"/api/ai/(\d+)")
def api_ai(conn, m, body, qs):
    return suggest.deep_suggest(conn, int(m.group(1)))


@route("POST", r"/api/scan")
def api_scan(conn, m, body, qs):
    def int_setting(name, default, lo, hi):
        raw = body.get(name)
        if raw is None:
            raw = store.get_setting(conn, name, str(default))
        try:
            n = int(raw)
        except (TypeError, ValueError):
            n = default
        return max(lo, min(hi, n))

    return scan_directory(conn, body.get("path", ""),
                          recursive=bool(body.get("recursive")),
                          include_files=body.get("include_files", True),
                          max_items=int_setting("scan_max_items", 500, 1, 5000),
                          scan_depth=int_setting("scan_depth", 4, 0, 12))


@route("POST", r"/api/open-path")
def api_open_path(conn, m, body, qs):
    """Open a path written inside an item's body.

    Notes often *mention* a file ("the report is at D:\\...\\report.html") without
    the item itself being that file. Before this, the only way to open it was to
    select the path by hand and paste it into a file manager — which defeats the
    point of having the note. The client picks paths out of the body text and
    posts one back here.
    """
    raw = (body.get("path") or "").strip().strip('"').strip("'")
    if not raw:
        return {"error": "err.open_target_missing"}
    p = os.path.abspath(os.path.expanduser(raw))
    if not os.path.exists(p):
        return {"error": "err.not_found"}
    try:
        if sys.platform == "win32":
            os.startfile(p)  # noqa: S606 - local personal tool
        elif sys.platform == "darwin":
            subprocess.run(["open", p], check=False)
        else:
            subprocess.run(["xdg-open", p], check=False)
        return {"ok": True, "path": p}
    except OSError as e:
        return {"error": str(e)}


@route("POST", r"/api/open/(\d+)")
def api_open(conn, m, body, qs):
    item = store.get_item(conn, int(m.group(1)))
    if not item:
        return {"error": "err.not_found"}
    target = item["path"] or item["url"]
    if not target:
        return {"error": "err.open_target_missing"}
    try:
        if item["path"]:
            if sys.platform == "win32":
                os.startfile(item["path"])  # noqa: S606 - local personal tool
            elif sys.platform == "darwin":
                subprocess.run(["open", item["path"]], check=False)
            else:
                subprocess.run(["xdg-open", item["path"]], check=False)
        else:
            webbrowser.open(item["url"])
        store.touch_item(conn, item["id"])
        return {"ok": True}
    except OSError as e:
        return {"error": str(e)}


# ---- quick-open launcher (logic in launcher.py) ----

@route("GET", r"/api/launcher")
def api_launcher(conn, m, body, qs):
    return launcher.state(conn)


@route("POST", r"/api/launcher/config")
def api_launcher_config(conn, m, body, qs):
    return launcher.save_config(conn, body)


@route("POST", r"/api/launcher/open")
def api_launcher_open(conn, m, body, qs):
    return launcher.open_target(body)


@route("GET", r"/api/launcher/locate")
def api_launcher_locate(conn, m, body, qs):
    return launcher.locate(conn, qs.get("path", ""))


# ---- roadmap (optional) ----
# A roadmap file is the single source of truth: something outside Seiton (you,
# or an AI session) writes the JSON, and the screen only draws it — the view
# never owns the data, so a reload always shows the latest.
# The file is optional: without it the "全体図 / Roadmap" button never appears.
# Location: the `roadmap_path` setting if set, else data/roadmap.json.
# See roadmap.example.json for the format.

def _roadmap_path(conn):
    custom = (store.get_setting(conn, "roadmap_path", "") or "").strip()
    return custom or os.path.join(store.DATA_DIR, "roadmap.json")


@route("GET", r"/api/roadmap")
def api_roadmap(conn, m, body, qs):
    path = _roadmap_path(conn)
    # probe=1: the client asks "is there a roadmap at all?" on page load, and
    # only then adds the button. Keep it cheap — do not parse the file.
    if qs.get("probe"):
        return {"available": os.path.exists(path)}
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
    except FileNotFoundError:
        return {"error": "err.roadmap_missing", "path": path}
    except (OSError, json.JSONDecodeError) as ex:
        # 壊れたJSONを黙って空表示にすると「進捗ゼロ」に見えて事故になる
        return {"error": "err.roadmap_broken", "detail": str(ex)}
    if not isinstance(data, dict):
        return {"error": "err.roadmap_broken", "detail": "root must be an object"}
    try:
        data["mtime"] = os.path.getmtime(path)
    except OSError:
        pass
    return data


class Handler(BaseHTTPRequestHandler):
    server_version = "Seiton/1.1.0"

    def log_message(self, fmt, *args):
        sys.stderr.write("[seiton] %s\n" % (fmt % args))

    def _send_json(self, obj, code=200):
        data = json.dumps(obj, ensure_ascii=False).encode("utf-8")
        self.send_response(code)
        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()
        if self.command != "HEAD":
            self.wfile.write(data)

    def _send_raw(self, raw):
        data = raw.data
        self.send_response(raw.code)
        self.send_header("Content-Type", raw.content_type)
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        if raw.filename:
            self.send_header("Content-Disposition", f'attachment; filename="{raw.filename}"')
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(data)

    def _allowed_hosts(self):
        port = self.server.server_port
        return {f"127.0.0.1:{port}", f"localhost:{port}", "127.0.0.1", "localhost"}

    def _api_allowed(self):
        host = (self.headers.get("Host") or "").strip().lower()
        if host not in self._allowed_hosts():
            return False
        origin = (self.headers.get("Origin") or "").strip().rstrip("/")
        if origin:
            allowed = {f"http://127.0.0.1:{self.server.server_port}",
                       f"http://localhost:{self.server.server_port}"}
            if origin not in allowed:
                return False
        token = self.headers.get("X-Seiton-Token") or ""
        return secrets.compare_digest(token, self.server.seiton_token)

    def _serve_static(self, path, head_only=False):
        if path == "/":
            path = "/index.html"
        if path == "/favicon.ico":
            path = "/favicon.ico"
        static_root = os.path.abspath(STATIC_DIR)
        fp = os.path.abspath(os.path.normpath(os.path.join(static_root, path.lstrip("/"))))
        if not (fp == static_root or fp.startswith(static_root + os.sep)) or not os.path.isfile(fp):
            self.send_error(404)
            return
        ext = os.path.splitext(fp)[1].lower()
        with open(fp, "rb") as f:
            data = f.read()
        if path == "/index.html":
            data = data.decode("utf-8").replace("{{TOKEN}}", self.server.seiton_token).encode("utf-8")
        self.send_response(200)
        self.send_header("Content-Type", MIME.get(ext, "application/octet-stream"))
        self.send_header("Content-Length", str(len(data)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        if not head_only:
            self.wfile.write(data)

    def _dispatch(self, method):
        parsed = urllib.parse.urlparse(self.path)
        path = urllib.parse.unquote(parsed.path)
        if not path.startswith("/api/"):
            if method == "GET":
                self._serve_static(path)
            else:
                self.send_error(405)
            return
        if not self._api_allowed():
            self._send_json({"error": "err.forbidden"}, 403)
            return
        qs = {k: v[0] for k, v in urllib.parse.parse_qs(parsed.query).items()}
        body = {}
        if method in ("POST", "PATCH", "DELETE"):
            length = int(self.headers.get("Content-Length") or 0)
            if length:
                try:
                    body = json.loads(self.rfile.read(length).decode("utf-8"))
                except (json.JSONDecodeError, UnicodeDecodeError):
                    self._send_json({"error": "err.bad_json"}, 400)
                    return
        for m_method, rx, fn in ROUTES:
            if m_method != method:
                continue
            m = rx.match(path)
            if m:
                conn = store.connect()
                try:
                    result = fn(conn, m, body, qs)
                except Exception as e:  # keep the personal server alive
                    import traceback
                    traceback.print_exc()
                    result = {"error": f"err.server: {e}"}
                finally:
                    conn.close()
                if isinstance(result, RawResponse):
                    self._send_raw(result)
                    return
                code = 400 if isinstance(result, dict) and result.get("error") else 200
                self._send_json(result, code)
                return
        self._send_json({"error": "err.no_endpoint"}, 404)

    def do_GET(self):
        self._dispatch("GET")

    def do_POST(self):
        self._dispatch("POST")

    def do_PATCH(self):
        self._dispatch("PATCH")

    def do_DELETE(self):
        self._dispatch("DELETE")

    def do_HEAD(self):
        parsed = urllib.parse.urlparse(self.path)
        path = urllib.parse.unquote(parsed.path)
        if path.startswith("/api/"):
            if not self._api_allowed():
                self._send_json({"error": "err.forbidden"}, 403)
                return
            self._send_json({"ok": True})
            return
        self._serve_static(path, head_only=True)


def _run_tray_mode(port):
    """--tray（Windows）: コンソール窓なしでトレイ常駐。ブラウザは自動で開かない。

    初期化→自動バックアップ→bind の順は通常経路と同一。二重起動時は既存を開いて静かに終了。
    仕様: afk20260713_specs/08_星頓v1.1_実装仕様.md §2.2
    """
    url = f"http://127.0.0.1:{port}/"
    store.init_db()
    store.backup_db()
    token = secrets.token_urlsafe(32)

    class Server(ThreadingHTTPServer):
        allow_reuse_address = False  # Windows: prevent silent double-bind

    try:
        httpd = Server(("127.0.0.1", port), Handler)
        httpd.seiton_token = token
    except OSError:
        webbrowser.open(url)  # 二重起動: 既存の星頓を開いて静かに終了
        return
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    import tray  # pystray はここで初めて import（通常経路は stdlib のみ）
    tray.run_tray(httpd, url)  # メインスレッドでブロック。「終了」で戻る
    httpd.shutdown()


def main():
    port = DEFAULT_PORT
    open_browser = True
    args = sys.argv[1:]
    if args and args[0] == "add":
        result = store.add_quick_item(" ".join(args[1:]))
        print(json.dumps(result, ensure_ascii=False))
        return
    if "--add" in args:
        idx = args.index("--add")
        result = store.add_quick_item(args[idx + 1] if idx + 1 < len(args) else "")
        print(json.dumps(result, ensure_ascii=False))
        return
    if "--import" in args:
        idx = args.index("--import")
        if idx + 1 >= len(args):
            print(json.dumps({"error": "--import requires a JSON file"}, ensure_ascii=False))
            return
        result = store.import_json_file(args[idx + 1], force="--force" in args)
        print(json.dumps(result, ensure_ascii=False))
        return
    if "--port" in args:
        port = int(args[args.index("--port") + 1])
    if "--no-browser" in args:
        open_browser = False

    if "--tray" in args and sys.platform == "win32":
        _run_tray_mode(port)
        return
    # 非Windowsの --tray は以降の通常経路（コンソール起動）へフォールバックする

    store.init_db()
    backup = store.backup_db()
    addr = ("127.0.0.1", port)
    url = f"http://127.0.0.1:{port}/"
    token = secrets.token_urlsafe(32)

    class Server(ThreadingHTTPServer):
        allow_reuse_address = False  # Windows: prevent silent double-bind

    try:
        httpd = Server(addr, Handler)
        httpd.seiton_token = token
    except OSError:
        print(f"* 星頓 (Seiton) は既に起動しています: {url}")
        if open_browser:
            webbrowser.open(url)
        return
    print(f"* Seiton (星頓) running at {url}  (Ctrl+C to stop)")
    if backup.get("ok"):
        print(f"* backup: {backup['path']}")
    if open_browser:
        threading.Timer(0.6, lambda: webbrowser.open(url)).start()
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\n* bye")


if __name__ == "__main__":
    main()