Add files via upload

memory.py

@@ -0,0 +1,252 @@
+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]