Create ingest.py

rag/ingest.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+from typing import Dict, List, Any
+import os, json, glob, re
+from .util_text import clean_text
+
+_TXT_EXT = {".txt", ".md", ".markdown"}
+_JSON_EXT = {".json", ".jsonl"}
+
+def _read_text(path: str) -> str:
+    with open(path, "r", encoding="utf-8", errors="ignore") as f:
+        return f.read()
+
+def _title_from_text(t: str) -> str:
+    for line in t.splitlines():
+        s = clean_text(line)
+        if s:
+            return s[:80]
+    return "Untitled"
+
+def _meta_from_text(t: str) -> dict:
+    meta = {}
+    # "lat: 35.6", "lon: 139.7" 的な記述を素朴に拾う
+    mlat = re.search(r"\blat(?:itude)?\s*[:=]\s*([-+]?\d+(?:\.\d+)?)", t, re.I)
+    mlon = re.search(r"\blon(?:gitude)?\s*[:=]\s*([-+]?\d+(?:\.\d+)?)", t, re.I)
+    if mlat and mlon:
+        meta["lat"] = float(mlat.group(1))
+        meta["lon"] = float(mlon.group(1))
+    # タグ行: "tags: a,b,c"
+    mtag = re.search(r"\btags?\s*[:=]\s*([^\n]+)", t, re.I)
+    if mtag:
+        tags = [s.strip() for s in mtag.group(1).split(",") if s.strip()]
+        meta["tags"] = tags
+    return meta
+
+def _doc_from_json(obj: dict) -> dict:
+    text = clean_text(str(obj.get("text") or obj.get("body") or ""))
+    title = obj.get("title") or _title_from_text(text)
+    meta = obj.get("meta") or {}
+    for k in ("lat", "lon", "tags", "url", "city", "type"):
+        if k in obj and k not in meta:
+            meta[k] = obj[k]
+    meta.setdefault("title", title)
+    return {"id": str(obj.get("id") or title), "text": text, "meta": meta}
+
+def build_corpus(dir_map: Dict[str, str]) -> List[dict]:
+    """
+    dir_map 例: {"blogs": "data/blogs", "reviews": "data/reviews", "events": "data/events"}
+    各ディレクトリ配下の .txt/.md/.json/.jsonl を収集して Doc 配列を返す。
+    Doc = {"id","text","meta{title,lat,lon,tags,...}"}
+    """
+    docs: List[dict] = []
+    for _, d in (dir_map or {}).items():
+        if not d or not os.path.isdir(d):
+            continue
+        paths = []
+        for ext in list(_TXT_EXT | _JSON_EXT):
+            paths += glob.glob(os.path.join(d, f"**/*{ext}"), recursive=True)
+
+        for p in paths:
+            _, ext = os.path.splitext(p)
+            try:
+                if ext.lower() in _TXT_EXT:
+                    t = clean_text(_read_text(p))
+                    if not t: 
+                        continue
+                    m = _meta_from_text(t)
+                    title = _title_from_text(t)
+                    m.setdefault("title", title)
+                    docs.append({"id": p, "text": t, "meta": m})
+                else:  # JSON / JSONL
+                    raw = _read_text(p)
+                    if ext.lower() == ".jsonl":
+                        for line in raw.splitlines():
+                            line = line.strip()
+                            if not line:
+                                continue
+                            try:
+                                obj = json.loads(line)
+                                docs.append(_doc_from_json(obj))
+                            except Exception:
+                                continue
+                    else:
+                        obj = json.loads(raw)
+                        if isinstance(obj, list):
+                            for o in obj:
+                                if isinstance(o, dict):
+                                    docs.append(_doc_from_json(o))
+                        elif isinstance(obj, dict):
+                            docs.append(_doc_from_json(obj))
+            except Exception:
+                continue
+    return docs