data/Models/Text2Persona_Style/persona_style.py

import argparse
import re
import json
from pathlib import Path
from collections import Counter


PERSONA_STYLE_PATH = (
    Path(__file__).resolve().parents[2]
    / "persona_style.json"
)

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

POLITE_PATTERNS = [
    "です", "ます", "でした", "ました", "ください", "でしょう", "ですね"
]

CASUAL_PATTERNS = [
    "じゃん", "だよ", "だね", "やん", "かも", "やろ", "やで",
    "やねん", "ちゃう", "めっちゃ", "そやな", "そやね",
    "やからな", "やからね", "やってん"
]

CASUAL_ENDING_PATTERN = re.compile(
    r"(じゃん|だよ|だね|やん|かも|やろ|やで|"
    r"やな|やねん|やった|やってん|んや|ちゃう|"
    r"(?<!です)ね|(?<!ます)よ)"
    r"[。！？!?…\s]*$"
)

LAUGH_PATTERN = re.compile(
    r"(笑+|草+|ｗ+|w{2,}|(?<![A-Za-z])w(?![A-Za-z]))"
)

TRAILING_LAUGH_PATTERN = re.compile(
    r"(笑+|草+|ｗ+|w{2,}|(?<![A-Za-z])w(?![A-Za-z]))+$"
)

JAPANESE_CHAR_PATTERN = re.compile(r"[\u3040-\u30ff\u3400-\u9fff]")

CONVERSATIONAL_ENDING_PATTERN = re.compile(
    r"("
    r"おけおけ|おっけ(?:ー|え)?|おけ|りょかい(?:した)?|了解|"
    r"ないす|ありがと(?:う)?|さんきゅ|お願いします|ください|"
    r"でした|ました|でしょう|ですね|です|ます|"
    r"じゃん|だよ|だね|やんな|やろ|やで|"
    r"やってん|やねん|やった|やからねー?|やな|やね|やん|"
    r"んや|ちゃう|"
    r"かも|ねー+|ね|よ|な|や"
    r")$"
)

COMMON_PHRASE_CANDIDATES = [
    "まじ", "まじで", "たしかに", "いいね", "なるほど",
    "やばい", "ほんま", "そうなん", "それな", "うける",
    "えぐい", "かわいい", "すごい", "ごめん", "ありがとう",
    "がち", "めっちゃ", "あね", "そやな", "そやね"
]

EMOJI_PATTERN = re.compile(
    "["
    "\U0001F300-\U0001FAFF"
    "\U00002700-\U000027BF"
    "]+",
    flags=re.UNICODE
)


def load_json(path, default):
    if not path.exists():
        return default
    return json.loads(path.read_text(encoding="utf-8"))


def save_json(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(
        json.dumps(data, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )


def extract_partner_messages(messages, partner_role="partner"):
    return [
        m["text"]
        for m in messages
        if m.get("role") == partner_role and m.get("text", "").strip()
    ]


def partner_name_from_line_filename(path):
    stem = Path(path).stem
    prefix = "[LINE]"
    if stem.upper().startswith(prefix):
        name = stem[len(prefix):].strip()
        return name or None
    return None


def infer_partner_name_from_messages(parsed_messages, user_names):
    user_name_set = set(user_names or [])
    candidates = []

    for message in parsed_messages:
        speaker = message["speaker"]
        if speaker in user_name_set:
            continue
        if speaker not in candidates:
            candidates.append(speaker)

    if len(candidates) == 1:
        return candidates[0]
    if not candidates:
        return None

    raise ValueError(
        "partner_nameを推定できませんでした。候補が複数あります: "
        + ", ".join(candidates)
    )


def should_skip_text_message(text):
    text = text.strip()
    if not text:
        return True
    if text in MEDIA_MARKERS:
        return True
    if text in CALL_MARKERS:
        return True
    if URL_RE.search(text):
        return True
    if CALL_DURATION_RE.match(text):
        return True
    if DELETED_MESSAGE_RE.search(text):
        return True
    return False


def split_line_message(line, speaker_names):
    match = LINE_MESSAGE_RE.match(line)
    if not match:
        return None

    content = match.group(2).strip()
    for speaker_name in speaker_names:
        prefix = f"{speaker_name} "
        if content == speaker_name:
            return {"speaker": speaker_name, "text": ""}
        if content.startswith(prefix):
            return {
                "speaker": speaker_name,
                "text": content[len(prefix):].strip()
            }

    parts = content.split(maxsplit=1)
    if len(parts) == 1:
        return {"speaker": parts[0], "text": ""}
    return {"speaker": parts[0], "text": parts[1].strip()}


def extract_messages_from_txt(
    txt_path,
    partner_name=None,
    user_names=None,
    tail_lines=None,
    encoding="utf-8"
):
    path = Path(txt_path)
    resolved_partner_name = partner_name or partner_name_from_line_filename(path)
    lines = path.read_text(encoding=encoding).splitlines()
    if tail_lines is not None:
        lines = lines[-tail_lines:]

    user_names = user_names or []
    speaker_names = sorted(
        {name for name in [resolved_partner_name, *user_names] if name},
        key=len,
        reverse=True
    )

    parsed_messages = []
    for line in lines:
        if LINE_DATE_RE.match(line):
            continue

        parsed = split_line_message(line, speaker_names)
        if parsed:
            parsed_messages.append(parsed)
        elif parsed_messages and line.strip():
            parsed_messages[-1]["text"] += "\n" + line.strip()

    if resolved_partner_name is None:
        resolved_partner_name = infer_partner_name_from_messages(
            parsed_messages,
            user_names
        )
    if resolved_partner_name is None:
        raise ValueError(
            "partner_nameを推定できませんでした。"
            "ファイル名を[LINE]相手名.txtにするか、partner_nameを指定してください。"
        )

    messages = []
    for parsed in parsed_messages:
        text = parsed["text"].strip()
        if should_skip_text_message(text):
            continue

        role = "partner" if parsed["speaker"] == resolved_partner_name else "user"
        messages.append({
            "role": role,
            "text": text
        })

    return messages


def normalize_for_ending(text):
    cleaned = text.strip()
    cleaned = re.sub(r"[。！？!?…\s]+$", "", cleaned)
    cleaned = TRAILING_LAUGH_PATTERN.sub("", cleaned)
    cleaned = re.sub(r"[。！？!?…\s]+$", "", cleaned)
    return cleaned


def extract_endings(texts):
    endings = []

    for text in texts:
        cleaned = normalize_for_ending(text)

        if not JAPANESE_CHAR_PATTERN.search(cleaned):
            continue

        match = CONVERSATIONAL_ENDING_PATTERN.search(cleaned)
        if match:
            endings.append(match.group(1))

    return Counter(endings).most_common(10)


def has_laugh_expression(text):
    return bool(LAUGH_PATTERN.search(text))


def has_casual_expression(text):
    if any(p in text for p in CASUAL_PATTERNS):
        return True
    return bool(CASUAL_ENDING_PATTERN.search(text.strip()))


def estimate_reliability(sample_count):
    if sample_count >= 30:
        return {
            "level": "high",
            "description": "会話数が十分あり、比較的安定した推定"
        }
    if sample_count >= 10:
        return {
            "level": "medium",
            "description": "ある程度の会話数があり、参考にしやすい推定"
        }
    if sample_count > 0:
        return {
            "level": "low",
            "description": "会話数が少ないため、まだ仮の推定"
        }
    return {
        "level": "none",
        "description": "会話データがないため推定不可"
    }


def extract_favorite_phrases(texts):
    joined = "\n".join(texts)
    counts = Counter()

    for phrase in COMMON_PHRASE_CANDIDATES:
        c = joined.count(phrase)
        if c > 0:
            counts[phrase] = c

    return counts.most_common(10)


def extract_style_features(messages, partner_role="partner"):
    texts = extract_partner_messages(messages, partner_role=partner_role)

    if not texts:
        return {
            "sample_count": 0,
            "avg_length": 0,
            "short_reply_rate": 0,
            "question_rate": 0,
            "emoji_rate": 0,
            "laugh_rate": 0,
            "polite_rate": 0,
            "casual_rate": 0,
            "reliability": estimate_reliability(0),
            "ending_patterns": [],
            "favorite_phrases": []
        }

    total = len(texts)
    lengths = [len(t.strip()) for t in texts]

    short_reply_count = sum(length <= 8 for length in lengths)
    question_count = sum(("?" in t or "？" in t) for t in texts)
    emoji_count = sum(bool(EMOJI_PATTERN.search(t)) for t in texts)
    laugh_count = sum(has_laugh_expression(t) for t in texts)
    polite_count = sum(any(p in t for p in POLITE_PATTERNS) for t in texts)
    casual_count = sum(has_casual_expression(t) for t in texts)

    return {
        "sample_count": total,
        "avg_length": round(sum(lengths) / total, 2),
        "short_reply_rate": round(short_reply_count / total, 3),
        "question_rate": round(question_count / total, 3),
        "emoji_rate": round(emoji_count / total, 3),
        "laugh_rate": round(laugh_count / total, 3),
        "polite_rate": round(polite_count / total, 3),
        "casual_rate": round(casual_count / total, 3),
        "reliability": estimate_reliability(total),
        "ending_patterns": extract_endings(texts),
        "favorite_phrases": extract_favorite_phrases(texts)
    }

def describe_style(features):
    descriptions = []

    avg_length = features["avg_length"]
    short_rate = features["short_reply_rate"]
    polite_rate = features["polite_rate"]
    casual_rate = features["casual_rate"]
    laugh_rate = features["laugh_rate"]
    emoji_rate = features["emoji_rate"]
    question_rate = features["question_rate"]
    reliability = features["reliability"]

    if reliability["level"] in ["low", "none"]:
        descriptions.append(reliability["description"])

    if avg_length <= 10 or short_rate >= 0.6:
        descriptions.append("短文で返す傾向が強い")
    elif avg_length >= 30:
        descriptions.append("比較的長めに話す傾向がある")
    else:
        descriptions.append("文の長さは中程度")

    if polite_rate >= 0.5:
        descriptions.append("丁寧語が多い")
    elif casual_rate >= 0.3:
        descriptions.append("カジュアルなタメ口が多い")
    else:
        descriptions.append("硬すぎない自然な口調")

    if laugh_rate >= 0.25:
        descriptions.append("「笑」や「w」などの笑い表現をよく使う")
    elif laugh_rate >= 0.08:
        descriptions.append("笑い表現をたまに使う")
    else:
        descriptions.append("笑い表現は控えめ")

    if emoji_rate >= 0.2:
        descriptions.append("絵文字を比較的よく使う")
    else:
        descriptions.append("絵文字は少なめ")

    if question_rate >= 0.25:
        descriptions.append("質問で会話を続けることが多い")

    endings = [e for e, _ in features.get("ending_patterns", [])[:5]]
    if endings:
        descriptions.append(f"よく出る語尾は「{'」「'.join(endings)}」")

    phrases = [p for p, _ in features.get("favorite_phrases", [])[:5]]
    if phrases:
        descriptions.append(f"よく使う表現は「{'」「'.join(phrases)}」")

    return "。".join(descriptions) + "。"


def update_persona_style(messages, partner_role="partner"):
    features = extract_style_features(messages, partner_role=partner_role)
    style_description = describe_style(features)

    data = {
        "style_features": features,
        "style_description": style_description
    }

    save_json(PERSONA_STYLE_PATH, data)
    return data


def update_persona_style_from_txt(
    txt_path,
    user_names=None,
    tail_lines=200,
    encoding="utf-8"
):
    """LINE .txtから相手の文体を分析してpersona_style.jsonへ保存する."""
    messages = extract_messages_from_txt(
        txt_path=txt_path,
        user_names=user_names,
        tail_lines=tail_lines,
        encoding=encoding
    )
    return update_persona_style(messages, partner_role="partner")


def main():
    parser = argparse.ArgumentParser(
        description="Analyze persona style from a LINE chat export."
    )
    parser.add_argument(
        "txt_path",
        type=Path,
        help="LINE export .txt file. If named [LINE]name.txt, name is used as the partner.",
    )
    parser.add_argument(
        "--user-name",
        dest="user_names",
        action="append",
        default=None,
        help="Your LINE display name. Can be passed multiple times. Defaults to Rayta.",
    )
    parser.add_argument(
        "--tail-lines",
        type=int,
        default=200,
        help="Most recent N lines to analyze (0 = all).",
    )
    parser.add_argument("--encoding", default="utf-8")
    args = parser.parse_args()

    data = update_persona_style_from_txt(
        txt_path=args.txt_path,
        user_names=args.user_names or ["Rayta"],
        tail_lines=None if args.tail_lines == 0 else args.tail_lines,
        encoding=args.encoding,
    )
    print(data["style_description"])
    print("saved to:", PERSONA_STYLE_PATH)


if __name__ == "__main__":
    main()