core/prompt_builder.py

"""初期/system prompt を組み立てる。

このモジュールの主な責務は、user_memory / history_summary / persona_style から
LLM に渡す system プロンプト文字列と messages 配列を作ること。

直近の conversation_history は app.py や memory.py 側で管理し、このファイルでは
呼び出し側から明示的に渡された場合だけメッセージ列に並べる。
"""
from __future__ import annotations

import json
import re
from pathlib import Path
from typing import Any

import session_paths as sp
from persona import (
    SAFETY_RULES,
    summarize_relationship,
    summarize_traits,
)

ROOT = sp.ROOT

DEFAULT_HISTORY_TURNS = 3

__all__ = [
    "build_system_blocks",
]

LINE_MESSAGE_RE = re.compile(r"^(\d{1,2}:\d{2})\s+(.+)$")
MEDIA_MARKERS = {"画像", "動画", "スタンプ", "写真", "ボイスメッセージ", "音声", "アルバム", "連絡先"}
CALL_MARKERS = {"応答なし", "不在着信", "キャンセル"}
CALL_DURATION_RE = re.compile(r"^\d{1,2}:\d{2}$")
URL_RE = re.compile(r"https?://|www\.")
DELETED_MESSAGE_RE = re.compile(r"メッセージの送信を取り消しました")


def _load_json(path: Path, default: Any) -> Any:
    if not path.exists():
        return default
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError:
        return default

def _summary_to_text(summary: Any) -> str:
    if isinstance(summary, dict):
        return str(summary.get("recent_summary") or summary.get("summary") or "")
    return str(summary or "")


def _profile_line(profile: dict[str, Any]) -> str:
    parts: list[str] = []
    if profile.get("user_name"):
        parts.append(f"名前={profile['user_name']}")
    nickname = profile.get("preferred_nickname") or profile.get("nickname")
    if nickname:
        parts.append(f"呼称={nickname}")
    if profile.get("user_persona"):
        parts.append(f"人柄={profile['user_persona']}")
    if profile.get("age"):
        parts.append(f"年齢={profile['age']}")
    if profile.get("goals"):
        parts.append(f"目的={profile['goals']}")
    return " / ".join(parts) or "(未設定)"


def _style_line(style: Any) -> str:
    if not isinstance(style, dict):
        return str(style or "")

    feats = style.get("style_features", {})
    desc = style.get("style_description", "")
    endings = feats.get("ending_patterns", [])[:5]
    phrases = feats.get("favorite_phrases", [])[:5]

    end_str = "、".join(e[0] for e in endings if isinstance(e, list) and e)
    ph_str = "、".join(p[0] for p in phrases if isinstance(p, list) and p)

    extras: list[str] = []
    if end_str:
        extras.append(f"語尾={end_str}")
    if ph_str:
        extras.append(f"頻出表現={ph_str}")

    if desc and extras:
        return " / ".join([desc, *extras])
    return desc or " / ".join(extras)


def _resolve_txt_file(txt_path: Path | str | None) -> Path | None:
    if txt_path is None:
        return None

    path = Path(txt_path)
    if path.is_file():
        return path
    if not path.is_dir():
        return None

    txt_files = sorted(
        path.glob("*.txt"),
        key=lambda p: p.stat().st_mtime,
        reverse=True,
    )
    return txt_files[0] if txt_files else None


def _other_name_from_line_filename(path: Path) -> str | None:
    stem = path.stem
    prefix = "[LINE]"
    if stem.upper().startswith(prefix):
        name = stem[len(prefix):].strip()
        return name or None
    return None


def _should_skip_line_text(text: str) -> bool:
    if text.startswith("[PayPay]"):
        return True
    if text in MEDIA_MARKERS or text in CALL_MARKERS:
        return True
    if CALL_DURATION_RE.fullmatch(text):
        return True
    if URL_RE.search(text):
        return True
    if DELETED_MESSAGE_RE.search(text):
        return True
    return False


def _split_speaker_and_text(
    rest: str,
    speaker_names: list[str],
) -> tuple[str, str] | None:
    for speaker in speaker_names:
        if not speaker:
            continue
        if rest == speaker:
            return speaker, ""
        for sep in (" ", "\t"):
            prefix = f"{speaker}{sep}"
            if rest.startswith(prefix):
                return speaker, rest[len(prefix):].strip()

    # Fallback for LINE exports where the other person's name cannot be
    # inferred. This only works for one-token display names.
    parts = rest.split(maxsplit=1)
    if len(parts) == 2:
        return parts[0], parts[1].strip()
    return None


def _recent_line_conversation(
    txt_path: Path | str | None,
    self_name: str,
    *,
    limit: int = 6,
) -> list[dict[str, str]]:
    resolved = _resolve_txt_file(txt_path)
    if resolved is None:
        return []

    speaker_names = [self_name]
    other_name = _other_name_from_line_filename(resolved)
    if other_name and other_name not in speaker_names:
        speaker_names.append(other_name)
    speaker_names = sorted(speaker_names, key=len, reverse=True)

    turns: list[dict[str, str]] = []
    for raw_line in resolved.read_text(encoding="utf-8", errors="replace").splitlines():
        match = LINE_MESSAGE_RE.match(raw_line.strip())
        if not match:
            continue

        _, rest = match.groups()
        speaker_and_text = _split_speaker_and_text(rest, speaker_names)
        if not speaker_and_text:
            continue

        speaker, text = speaker_and_text
        if not text or _should_skip_line_text(text):
            continue

        role = "user" if speaker == self_name else "assistant"
        turns.append({"role": role, "speaker": speaker, "text": text})

    return turns[-limit:]


def _format_recent_conversation(
    turns: list[dict[str, str]],
    *,
    persona_name: str | None,
) -> str:
    lines: list[str] = []
    for turn in turns:
        text = turn.get("text", "")
        if not text:
            continue
        if turn.get("role") == "user":
            label = "ユーザー"
        else:
            label = persona_name or turn.get("speaker") or "相手"
        lines.append(f"{label}: {text}")
    return "\n".join(lines)


def build_system_blocks(
    user_memory: dict[str, Any],
    txt_path: Path | str | None,
    self_name: str,
) -> list[dict[str, Any]]:
    """Anthropic Messages API 用の system blocks を返す。

    安定部分(キャラ/関係性/文体/返答ルールなど)を1ブロック目に置いて
    `cache_control: ephemeral` を付与し、可変部分(history_summary)を
    2ブロック目に置く。毎ターン同じリッチなコンテキストをLLMに渡す。
    """
    resolved_txt_path = _resolve_txt_file(txt_path)
    profile = user_memory.get("user_profile", {})
    ai = user_memory.get("ai_persona", {})
    persona_name = ai.get("name")
    history_summary = _load_json(
        sp.summary_path(),
        user_memory.get("history_summary", ""),
    )
    persona_style = _load_json(
        sp.persona_style_path(),
        ai.get("persona_style_description", {}),
    )
    goal = profile.get("goals") or profile.get("goal")

    static_lines: list[str] = [
        f"あなたは{persona_name}として会話する。",
    ]
    if goal:
        static_lines.append(
            f"ユーザーのゴールである「{goal}」を達成する相手として、現実にいそうな人間として自然なLINE風の会話をする。"
        )
    else:
        static_lines.append(
            "現実にいそうな人間として自然なLINE風の会話をする。"
        )
    if ai.get("persona"):
        static_lines.append(f"[人物像] {ai['persona']}")
    static_lines.append(f"[ユーザー] {_profile_line(profile)}")
    if user_memory.get("relationship_prompt"):
        static_lines.append(f"[距離感] {user_memory['relationship_prompt']}")
    personality_text = ai.get("personality_text") or summarize_traits(ai.get("traits", {}))
    if personality_text:
        static_lines.append(f"[性格傾向] {personality_text}")
    if ai.get("persona_style_description"):
        static_lines.append(f"[口調] {ai['persona_style_description']}")
    style_line = _style_line(persona_style)
    if style_line:
        static_lines.append(f"[文体] {style_line}")
    recent_conversation = _recent_line_conversation(
        resolved_txt_path,
        self_name,
        limit=6,
    )
    recent_conversation_text = _format_recent_conversation(
        recent_conversation,
        persona_name=persona_name,
    )
    if recent_conversation_text:
        static_lines.append(f"[過去のLINE抜粋]\n{recent_conversation_text}")
    static_lines.extend(
        [
            "[返答ルール]",
            "日本語で返す。",
            "1〜3文で、LINE風に短めに返す。",
            "説明口調、箇条書き、営業口調にしない。",
            "AIであることを言わない。",
            "不自然に優しすぎたり、相手の気持ちを決めつけたりしない。",
            "質問は1返信につき1つまでにする。",
            "まず相手の発言に自然に反応してから、必要なら話題を広げる。",
            SAFETY_RULES,
        ]
    )

    blocks: list[dict[str, Any]] = [
        {
            "type": "text",
            "text": "\n".join(static_lines),
            "cache_control": {"type": "ephemeral"},
        }
    ]

    summary_text = _summary_to_text(history_summary)
    if summary_text:
        blocks.append(
            {
                "type": "text",
                "text": f"[これまでの流れ]\n{summary_text}",
            }
        )
    return blocks