playchat/backend/app/core/notes.py

"""ユーザーとAIが共同で編集する共有ノート。

システムプロンプトにはページの「タイトル一覧」だけを注入し、本文はモデルが
read_note ツールで必要なときだけ読み出す。書き込みは write_note ツールで行う。
"""
import logging
import re

from sqlalchemy.orm import Session as DBSession

from app.core.config import NOTES_TOOL_CONTENT_LIMIT
from app.db.models import NotePage

logger = logging.getLogger(__name__)


READ_NOTE_TOOL = {
    "type": "function",
    "function": {
        "name": "read_note",
        "description": "共有ノートの特定ページの本文を読み出す。システムプロンプトに提示されたタイトル一覧の中から、"
        "今の会話に関係しそうなページがあるときに使う。",
        "parameters": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "読み出すページのタイトル（タイトル一覧に載っているもの）"},
            },
            "required": ["title"],
        },
    },
}

WRITE_NOTE_TOOL = {
    "type": "function",
    "function": {
        "name": "write_note",
        "description": "共有ノートのページに書き込む。ユーザーからノートへの記録・追記・整理を頼まれたときに使う。"
        "存在しないタイトルを指定した場合は新しいページを作成する。"
        "既存ページを replace で上書きすると元の内容は失われるので、追記でよい場合は必ず append を使うこと。",
        "parameters": {
            "type": "object",
            "properties": {
                "title": {"type": "string", "description": "書き込む対象のページタイトル"},
                "content": {"type": "string", "description": "書き込む本文（Markdown可）"},
                "mode": {
                    "type": "string",
                    "enum": ["append", "replace"],
                    "description": "append=末尾に追記（既定） / replace=ページ全体を置き換え",
                },
            },
            "required": ["title", "content"],
        },
    },
}

NOTE_TOOLS = [READ_NOTE_TOOL, WRITE_NOTE_TOOL]


def _find_page(db: DBSession, title: str) -> NotePage | None:
    """タイトルでページを探す。完全一致 → 大文字小文字・前後空白を無視した一致 の順に試す。"""
    pages = db.query(NotePage).order_by(NotePage.position, NotePage.id).all()
    for page in pages:
        if page.title == title:
            return page
    normalized = title.strip().casefold()
    for page in pages:
        if page.title.strip().casefold() == normalized:
            return page
    return None


def next_position(db: DBSession) -> int:
    """新規ページを末尾に置くための position を返す。"""
    last = db.query(NotePage).order_by(NotePage.position.desc()).first()
    return (last.position + 1) if last else 0


def set_section(db: DBSession, page_id: int, section: str | None) -> NotePage | None:
    """ページの所属セクションを変更し、同じセクションのページ群の直後に連続配置し直す。

    タブバーはposition順に描画され、隣接して同じsectionが続く区間を1つのグループとして
    まとめて表示する（app.api.notes / フロントのグルーピング表示を参照）。そのため
    セクションを変更したページは、常に同じセクションの最後のページの直後に移動する。
    """
    page = db.get(NotePage, page_id)
    if not page:
        return None

    normalized = (section or "").strip() or None
    ordered = [p for p in db.query(NotePage).order_by(NotePage.position, NotePage.id).all() if p.id != page_id]

    if normalized is None:
        insert_at = len(ordered)
    else:
        matching = [i for i, p in enumerate(ordered) if p.section == normalized]
        insert_at = (matching[-1] + 1) if matching else len(ordered)

    ordered.insert(insert_at, page)
    page.section = normalized
    for i, p in enumerate(ordered):
        p.position = i
    db.commit()
    db.refresh(page)
    return page


def rename_section(db: DBSession, old_name: str, new_name: str | None) -> int:
    """あるセクションに属する全ページのセクション名を一括で変更する。new_name が空ならグループ解除になる。

    戻り値: 変更したページ数。
    """
    normalized_new = (new_name or "").strip() or None
    pages = db.query(NotePage).filter(NotePage.section == old_name).order_by(NotePage.position, NotePage.id).all()
    for page in pages:
        page.section = normalized_new
    db.commit()
    return len(pages)


def get_notes_context(db: DBSession) -> str | None:
    """共有ノートのタイトル一覧をシステムプロンプト用のテキストとして返す。セクションごとに見出しを付ける。"""
    pages = db.query(NotePage).order_by(NotePage.position, NotePage.id).all()

    lines = [
        "【共有ノート】",
        "ユーザーとあなたが共同で編集できるノートです。以下のページがあります"
        "（セクションはページをグループ化する見出しで、あれば併記します）。",
        "",
    ]
    if pages:
        current_section = "__unset__"  # どのページのsectionとも一致しない番兵
        for page in pages:
            if page.section != current_section:
                current_section = page.section
                if current_section:
                    lines.append(f"## {current_section}")
            preview = " ".join(page.content.split())[:40]
            suffix = f" — {preview}…" if preview else " — (空)"
            lines.append(f"- {page.title}{suffix}")
    else:
        lines.append("- (まだページがありません)")
    lines += [
        "",
        "本文はここには含まれていません。内容が必要なときは read_note ツールで読み出してください。",
        "ノートへの記録・追記を頼まれたときは write_note ツールで書き込んでください"
        "（新しいタイトルを指定すればページが新規作成されます）。",
        "ユーザーが明示的に頼んでいないのに勝手に書き込まないこと。",
    ]
    return "\n".join(lines)


def read_note(db: DBSession, title: str) -> str:
    """ページ本文をツール結果用のテキストとして返す。"""
    page = _find_page(db, title)
    if not page:
        available = ", ".join(p.title for p in db.query(NotePage).order_by(NotePage.position, NotePage.id).all())
        return f"「{title}」というページはありません。現在のページ: {available or '(なし)'}"
    if not page.content.strip():
        return f"ページ「{page.title}」は空です。"
    content = page.content
    if len(content) > NOTES_TOOL_CONTENT_LIMIT:
        content = content[:NOTES_TOOL_CONTENT_LIMIT] + "\n…(以下省略)"
    return f"ページ「{page.title}」の本文:\n\n{content}"


_LIST_ITEM = re.compile(r"^\s*([-*+]|\d+[.)])\s")


def _append(existing: str, addition: str) -> str:
    """既存本文に追記する。箇条書き同士をつなぐ場合だけ空行を挟まない。"""
    if not existing.strip():
        return addition
    body = existing.rstrip()
    last_line = body.rsplit("\n", 1)[-1]
    first_line = addition.lstrip("\n").split("\n", 1)[0]
    separator = "\n" if _LIST_ITEM.match(last_line) and _LIST_ITEM.match(first_line) else "\n\n"
    return f"{body}{separator}{addition}"


def write_note(db: DBSession, title: str, content: str, mode: str = "append") -> str:
    """ページに書き込む。存在しなければ新規作成する。"""
    if not title.strip():
        return "タイトルが空のため書き込めませんでした。"

    page = _find_page(db, title)
    if page is None:
        page = NotePage(title=title.strip(), content=content, position=next_position(db))
        db.add(page)
        db.commit()
        return f"新しいページ「{page.title}」を作成し、本文を書き込みました。"

    if mode == "replace":
        page.content = content
        result = f"ページ「{page.title}」の内容を置き換えました。"
    else:
        page.content = _append(page.content, content)
        result = f"ページ「{page.title}」に追記しました。"
    db.commit()
    return result


def execute_note_tool(db: DBSession, name: str, args: dict) -> tuple[str, bool]:
    """ノート系ツールを実行する。戻り値: (ツール結果テキスト, ノートを変更したか)。"""
    try:
        if name == "read_note":
            return read_note(db, args.get("title", "")), False
        if name == "write_note":
            result = write_note(
                db,
                args.get("title", ""),
                args.get("content", ""),
                args.get("mode", "append"),
            )
            return result, True
    except Exception as e:
        db.rollback()
        logger.exception("Note tool %s failed", name)
        return f"ノート操作に失敗しました: {e}", False
    return f"未知のツールです: {name}", False