capture-ai/memory.py

import json
import re
import math
import hashlib
from pathlib import Path
from collections import Counter


def normalize_chat_data(raw):
    if isinstance(raw, list):
        return {
            "summary": "",
            "messages": raw,
            "code_context": {},
            "memory_chunks": []
        }

    if not isinstance(raw, dict):
        raw = {}

    raw.setdefault("summary", "")
    raw.setdefault("messages", [])
    raw.setdefault("code_context", {})
    raw.setdefault("memory_chunks", [])

    return raw


def load_chat_data(chat_file: Path) -> dict:
    try:
        raw = json.loads(chat_file.read_text(encoding="utf-8")) or []
    except Exception:
        raw = []

    return normalize_chat_data(raw)


def save_chat_data(chat_file: Path, chat_data: dict):
    chat_file.write_text(
        json.dumps(chat_data, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )


def get_chat_messages(chat_data: dict) -> list:
    return chat_data.get("messages", [])


def add_memory_chunk(chat_data: dict, text: str, source: str = "chat"):
    text = str(text or "").strip()
    if not text:
        return

    cid = hashlib.sha256((source + text).encode("utf-8")).hexdigest()[:16]

    for item in chat_data.get("memory_chunks", []):
        if item.get("id") == cid:
            return

    chat_data.setdefault("memory_chunks", []).append({
        "id": cid,
        "source": source,
        "text": text[:4000]
    })


def tokenize(text: str):
    return re.findall(r"\w+", str(text).lower())


def cosine_similarity(a: str, b: str) -> float:
    va = Counter(tokenize(a))
    vb = Counter(tokenize(b))

    if not va or not vb:
        return 0.0

    common = set(va) & set(vb)
    dot = sum(va[w] * vb[w] for w in common)

    na = math.sqrt(sum(v * v for v in va.values()))
    nb = math.sqrt(sum(v * v for v in vb.values()))

    if na == 0 or nb == 0:
        return 0.0

    return dot / (na * nb)


def retrieve_relevant_chunks(chat_data: dict, query: str, limit: int = 5) -> list:
    scored = []

    for chunk in chat_data.get("memory_chunks", []):
        text = chunk.get("text", "")
        score = cosine_similarity(query, text)
        if score > 0:
            scored.append((score, chunk))

    scored.sort(key=lambda x: x[0], reverse=True)
    return [chunk for score, chunk in scored[:limit]]


def summarize_code_basic(file_path: Path) -> dict:
    try:
        code = file_path.read_text(encoding="utf-8", errors="replace")
    except Exception:
        return {}

    classes = []
    functions = []

    for line in code.splitlines():
        s = line.strip()
        if s.startswith("class "):
            classes.append(s)
        elif s.startswith("def "):
            functions.append(s)

    return {
        "path": str(file_path),
        "name": file_path.name,
        "summary": f"{file_path.name}: {len(classes)} class, {len(functions)} function bulundu.",
        "classes": classes[:30],
        "functions": functions[:80]
    }


def update_code_context_from_message(chat_data: dict, msg: dict):
    files = msg.get("files") or []
    if not isinstance(files, list):
        return

    for f in files:
        if not isinstance(f, dict):
            continue

        p_raw = str(f.get("path") or "").strip()
        if not p_raw:
            continue

        p = Path(p_raw)
        if not p.exists() or not p.is_file():
            continue

        if p.suffix.lower() not in {
            ".py", ".js", ".ts", ".tsx", ".jsx", ".json", ".html",
            ".css", ".cpp", ".c", ".h", ".hpp", ".java", ".go",
            ".rs", ".sh", ".md", ".txt", ".yaml", ".yml"
        }:
            continue

        ctx = summarize_code_basic(p)
        if not ctx:
            continue

        chat_data.setdefault("code_context", {})[str(p)] = ctx

        add_memory_chunk(
            chat_data,
            ctx["summary"] + "\nFunctions:\n" + "\n".join(ctx.get("functions", [])[:30]),
            source=f"code:{p.name}"
        )


def build_memory_context(chat_data: dict, user_message: str) -> str:
    parts = []

    summary = str(chat_data.get("summary") or "").strip()
    if summary:
        parts.append("Conversation summary:\n" + summary)

    relevant = retrieve_relevant_chunks(chat_data, user_message, limit=5)
    if relevant:
        txt = "Relevant old context:\n"
        for ch in relevant:
            txt += f"- [{ch.get('source')}] {ch.get('text')}\n"
        parts.append(txt)

    code_context = chat_data.get("code_context") or {}
    if isinstance(code_context, dict) and code_context:
        txt = "Known code context:\n"
        for path, ctx in code_context.items():
            txt += f"\nFile: {ctx.get('name') or path}\n"
            txt += f"{ctx.get('summary', '')}\n"

            funcs = ctx.get("functions") or []
            if funcs:
                txt += "Functions:\n" + "\n".join(funcs[:30]) + "\n"

            classes = ctx.get("classes") or []
            if classes:
                txt += "Classes:\n" + "\n".join(classes[:20]) + "\n"

        parts.append(txt)

    return "\n\n".join(parts).strip()


def should_update_summary(chat_data: dict, every: int = 20) -> bool:
    messages = chat_data.get("messages", [])
    real = [m for m in messages if isinstance(m, dict) and m.get("role") in ("user", "bot", "assistant")]
    return len(real) > 0 and len(real) % every == 0

def make_simple_summary(chat_data: dict) -> str:
    messages = chat_data.get("messages", [])

    real_messages = [
        m for m in messages
        if isinstance(m, dict) and m.get("role") in ("user", "bot", "assistant")
    ]

    last_items = real_messages[-20:]

    parts = []

    for m in last_items:
        role = str(m.get("role") or "").strip()
        text = str(m.get("content") or "").strip()

        if not text:
            continue

        text = text.replace("\n", " ")
        if len(text) > 300:
            text = text[:300] + "..."

        parts.append(f"{role}: {text}")

    code_context = chat_data.get("code_context") or {}
    known_files = []

    if isinstance(code_context, dict):
        for _path, ctx in code_context.items():
            if not isinstance(ctx, dict):
                continue

            name = str(ctx.get("name") or "").strip()
            summary = str(ctx.get("summary") or "").strip()

            if name:
                known_files.append(name)

            if summary:
                parts.append(f"code_context: {summary}")

    summary = "Recent conversation summary:\n"
    summary += "\n".join(parts)

    if known_files:
        summary += "\n\nKnown code files: " + ", ".join(known_files)

    return summary[:6000]