diff --git a/backend/local_rag.py b/backend/local_rag.py index 336c214..a7cadd3 100644 --- a/backend/local_rag.py +++ b/backend/local_rag.py @@ -25,6 +25,7 @@ from .app_settings import ( get_ollama_api_url, ) from .paths import library_root +from .websearch import fetch_website_snapshot router = APIRouter(tags=["local-rag"]) @@ -55,6 +56,21 @@ class RegisterPathsRequest(BaseModel): paths: List[str] +class CreateTextRequest(BaseModel): + title: str + content: str + + +class UpdateTextRequest(BaseModel): + title: str + content: str + + +class CreateWebsiteRequest(BaseModel): + url: str + title: Optional[str] = None + + class RemoveFileRequest(BaseModel): rel: str @@ -118,6 +134,12 @@ def stage_dir(slug: str) -> Path: return path +def sources_dir(slug: str) -> Path: + path = lib_dir(slug) / "sources" + path.mkdir(parents=True, exist_ok=True) + return path + + def indexes_dir(slug: str) -> Path: path = lib_dir(slug) / "indexes" path.mkdir(parents=True, exist_ok=True) @@ -190,6 +212,10 @@ def _source_signature(files: List[Dict[str, Any]]) -> Optional[str]: "path": entry.get("path") or "", "rel": entry.get("rel") or "", "size": int(entry.get("size") or 0), + "item_id": entry.get("item_id") or "", + "kind": entry.get("kind") or "file", + "title": entry.get("title") or entry.get("name") or "", + "url": entry.get("url") or "", } digest.update(json.dumps(payload, sort_keys=True).encode("utf-8")) digest.update(b"\n") @@ -303,6 +329,9 @@ def _build_file_sync_payload( for entry in files: file_entry = dict(entry) + file_entry["item_id"] = str(file_entry.get("item_id") or file_entry.get("sha256") or file_entry.get("rel") or "") + file_entry["kind"] = str(file_entry.get("kind") or "file") + file_entry["title"] = str(file_entry.get("title") or file_entry.get("name") or "") file_entry["enrich_enabled"] = _file_enrich_enabled(file_entry) stored_status = str(file_entry.get("sync_status") or "pending") sync_status = stored_status @@ -411,6 +440,96 @@ def _sha256_file(path: Path) -> str: return digest.hexdigest() +def _write_text_atomic(path: Path, content: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(content, encoding="utf-8") + tmp.replace(path) + + +def _managed_source_path(slug: str, item_id: str, suffix: str = ".md") -> Path: + return sources_dir(slug) / f"{item_id}{suffix}" + + +def _managed_stage_rel(item_id: str, suffix: str = ".md") -> str: + return f"managed-{item_id}{suffix}" + + +def _ensure_stage_link(slug: str, source_path: Path, rel: str) -> None: + staged = stage_dir(slug) / rel + if staged.is_symlink() or staged.exists(): + staged.unlink() + staged.symlink_to(source_path) + + +def _find_item(data: Dict[str, Any], item_id: str) -> Optional[Dict[str, Any]]: + return next( + (entry for entry in data.get("files", []) if str(entry.get("item_id") or "") == item_id), + None, + ) + + +def _mark_entry_pending(entry: Dict[str, Any]) -> None: + entry["sync_status"] = "pending" + entry.pop("sync_error", None) + entry.pop("synced_at", None) + + +async def _save_library_change(slug: str, data: Dict[str, Any]) -> Optional[str]: + signature = _prepare_signature(data.get("files", [])) + _set_pending_prepare_signature(data, signature) + write_library(slug, data) + if signature: + return await _ensure_prepare_job(slug) + _cleanup_generated_artifacts(slug) + return None + + +def _apply_source_metadata(corpus_path: Path, files: List[Dict[str, Any]]) -> None: + if not corpus_path.exists(): + return + + by_path: Dict[str, Dict[str, Any]] = {} + for entry in files: + raw_path = str(entry.get("path") or "").strip() + if not raw_path: + continue + try: + by_path[str(Path(raw_path).expanduser().resolve())] = entry + except Exception: + by_path[raw_path] = entry + + rewritten: List[str] = [] + changed = False + with corpus_path.open("r", encoding="utf-8") as handle: + for line in handle: + if not line.strip(): + continue + rec = json.loads(line) + raw_source = str(rec.get("source_path") or "") + try: + source_key = str(Path(raw_source).expanduser().resolve()) + except Exception: + source_key = raw_source + entry = by_path.get(source_key) + if entry and entry.get("item_id"): + item_id = str(entry["item_id"]) + rec["id"] = item_id + rec["parent_id"] = None + rec["title"] = entry.get("title") or entry.get("name") or rec.get("title") + rec["url"] = entry.get("url") or rec.get("url") + rec["record_type"] = entry.get("kind") or rec.get("record_type") + meta = rec.get("meta") if isinstance(rec.get("meta"), dict) else {} + rec["meta"] = {**meta, "item_id": item_id, "source_kind": entry.get("kind") or "file"} + changed = True + rewritten.append(json.dumps(rec, ensure_ascii=False) + "\n") + + if changed: + tmp = corpus_path.with_suffix(".metadata.tmp") + tmp.write_text("".join(rewritten), encoding="utf-8") + tmp.replace(corpus_path) + + def _stage_name(sha: str, path: Path) -> str: safe_name = re.sub(r"[^A-Za-z0-9._-]+", "_", path.name).strip("._") or "file" return f"{sha}--{safe_name}" @@ -810,6 +929,7 @@ def _run_prepare_pipeline(slug: str, on_progress=None, **opts): emit="per-file", lang_detect=False, ) + _apply_source_metadata(paths["corpus"], files) build_errors = list((results.get("build") or {}).get("errors") or []) if not paths["corpus"].exists() or _line_count(paths["corpus"]) == 0: if build_errors: @@ -960,8 +1080,11 @@ def _build_local_context(prompt: str, results: Dict[str, Any], top_k: int = 5) - if len(snippet) > 1400: snippet = snippet[:1400].rstrip() + "..." raw_path = source.get("url") or source.get("doc_id") or "" - if raw_path and os.path.isabs(raw_path): - file_sources.append(_file_uri(raw_path)) + if raw_path: + if os.path.isabs(raw_path): + file_sources.append(_file_uri(raw_path)) + elif str(raw_path).startswith(("http://", "https://")): + file_sources.append(str(raw_path)) blocks.append(f"[L{idx}] {title}\n{snippet}") blocks.append("") blocks.append( @@ -1086,6 +1209,8 @@ async def register_paths(slug: str, req: RegisterPathsRequest): symlink_path.unlink() symlink_path.symlink_to(file_path) entry = { + "item_id": uuid.uuid4().hex, + "kind": "file", "sha256": sha, "path": str(file_path), "rel": stage_name, @@ -1113,6 +1238,183 @@ async def register_paths(slug: str, req: RegisterPathsRequest): } +@router.get("/libraries/{slug}/items/{item_id}") +def get_library_item(slug: str, item_id: str): + data = read_library(slug) + entry = _find_item(data, item_id) + if not entry: + raise HTTPException(status_code=404, detail="Content item not found") + + payload = dict(entry) + if entry.get("kind") == "text": + path = Path(str(entry.get("path") or "")) + payload["content"] = path.read_text(encoding="utf-8") if path.exists() else "" + return payload + + +@router.post("/libraries/{slug}/texts") +async def create_library_text(slug: str, req: CreateTextRequest): + data = read_library(slug) + title = str(req.title or "").strip() + content = str(req.content or "").strip() + if not title: + raise HTTPException(status_code=400, detail="Text title is required.") + if not content: + raise HTTPException(status_code=400, detail="Text content is required.") + + item_id = uuid.uuid4().hex + source_path = _managed_source_path(slug, item_id) + rel = _managed_stage_rel(item_id) + _write_text_atomic(source_path, content) + _ensure_stage_link(slug, source_path, rel) + entry = { + "item_id": item_id, + "kind": "text", + "title": title[:240], + "name": title[:240], + "path": str(source_path), + "rel": rel, + "sha256": _sha256_file(source_path), + "size": source_path.stat().st_size, + "added_at": now_iso(), + "updated_at": now_iso(), + "sync_status": "pending", + "enrich_enabled": False, + "managed": True, + } + data.setdefault("files", []).append(entry) + job_id = await _save_library_change(slug, data) + return {"item": entry, "job_id": job_id, "library": library_payload(data)} + + +@router.patch("/libraries/{slug}/texts/{item_id}") +async def update_library_text(slug: str, item_id: str, req: UpdateTextRequest): + data = read_library(slug) + entry = _find_item(data, item_id) + if not entry or entry.get("kind") != "text": + raise HTTPException(status_code=404, detail="Text item not found") + + title = str(req.title or "").strip() + content = str(req.content or "").strip() + if not title or not content: + raise HTTPException(status_code=400, detail="Text title and content are required.") + + source_path = Path(str(entry.get("path") or "")) + _write_text_atomic(source_path, content) + _ensure_stage_link(slug, source_path, str(entry.get("rel") or _managed_stage_rel(item_id))) + entry.update({ + "title": title[:240], + "name": title[:240], + "sha256": _sha256_file(source_path), + "size": source_path.stat().st_size, + "updated_at": now_iso(), + }) + _mark_entry_pending(entry) + job_id = await _save_library_change(slug, data) + return {"item": entry, "job_id": job_id, "library": library_payload(data)} + + +@router.post("/libraries/{slug}/websites") +async def create_library_website(slug: str, req: CreateWebsiteRequest): + data = read_library(slug) + try: + snapshot = await fetch_website_snapshot(req.url) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except httpx.HTTPStatusError as exc: + raise HTTPException(status_code=400, detail=f"Website returned HTTP {exc.response.status_code}.") from exc + except Exception as exc: + raise HTTPException(status_code=400, detail=f"Could not fetch website: {exc}") from exc + + canonical_url = str(snapshot["url"]) + duplicate = next( + (entry for entry in data.get("files", []) if entry.get("kind") == "website" and entry.get("url") == canonical_url), + None, + ) + if duplicate: + raise HTTPException(status_code=409, detail="This website is already saved in the database.") + + item_id = uuid.uuid4().hex + source_path = _managed_source_path(slug, item_id) + rel = _managed_stage_rel(item_id) + _write_text_atomic(source_path, str(snapshot["text"])) + _ensure_stage_link(slug, source_path, rel) + custom_title = str(req.title or "").strip() + title = custom_title or str(snapshot.get("title") or canonical_url) + entry = { + "item_id": item_id, + "kind": "website", + "title": title[:240], + "title_locked": bool(custom_title), + "name": title[:240], + "url": canonical_url, + "requested_url": str(snapshot.get("requested_url") or req.url), + "final_url": str(snapshot.get("final_url") or canonical_url), + "content_type": snapshot.get("content_type"), + "etag": snapshot.get("etag"), + "last_modified": snapshot.get("last_modified"), + "fetched_at": now_iso(), + "path": str(source_path), + "rel": rel, + "sha256": _sha256_file(source_path), + "size": source_path.stat().st_size, + "added_at": now_iso(), + "updated_at": now_iso(), + "sync_status": "pending", + "enrich_enabled": False, + "managed": True, + } + data.setdefault("files", []).append(entry) + job_id = await _save_library_change(slug, data) + return {"item": entry, "job_id": job_id, "library": library_payload(data)} + + +@router.post("/libraries/{slug}/websites/{item_id}/refresh") +async def refresh_library_website(slug: str, item_id: str): + data = read_library(slug) + entry = _find_item(data, item_id) + if not entry or entry.get("kind") != "website": + raise HTTPException(status_code=404, detail="Website item not found") + + try: + snapshot = await fetch_website_snapshot(str(entry.get("url") or entry.get("requested_url") or "")) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except httpx.HTTPStatusError as exc: + raise HTTPException(status_code=400, detail=f"Website returned HTTP {exc.response.status_code}.") from exc + except Exception as exc: + raise HTTPException(status_code=400, detail=f"Could not refresh website: {exc}") from exc + + source_path = Path(str(entry.get("path") or "")) + previous_hash = str(entry.get("sha256") or "") + _write_text_atomic(source_path, str(snapshot["text"])) + next_hash = _sha256_file(source_path) + entry.update({ + "url": str(snapshot.get("url") or entry.get("url") or ""), + "final_url": str(snapshot.get("final_url") or entry.get("final_url") or ""), + "content_type": snapshot.get("content_type"), + "etag": snapshot.get("etag"), + "last_modified": snapshot.get("last_modified"), + "fetched_at": now_iso(), + "size": source_path.stat().st_size, + "sha256": next_hash, + "updated_at": now_iso(), + }) + if not entry.get("title_locked"): + title = str(snapshot.get("title") or entry.get("title") or entry.get("url") or "")[:240] + entry["title"] = title + entry["name"] = title + + job_id = None + changed = next_hash != previous_hash + if changed: + _mark_entry_pending(entry) + job_id = await _save_library_change(slug, data) + else: + write_library(slug, data) + return {"changed": changed, "item": entry, "job_id": job_id, "library": library_payload(data)} + + @router.delete("/libraries/{slug}/files") async def remove_file(slug: str, req: RemoveFileRequest): data = read_library(slug) @@ -1123,8 +1425,12 @@ async def remove_file(slug: str, req: RemoveFileRequest): data["files"] = [entry for entry in files if entry.get("rel") != req.rel] symlink_path = stage_dir(slug) / req.rel - if symlink_path.exists(): + if symlink_path.is_symlink() or symlink_path.exists(): symlink_path.unlink() + if removed.get("managed"): + managed_path = Path(str(removed.get("path") or "")) + if managed_path.exists() and managed_path.is_file(): + managed_path.unlink() prepare_signature = _prepare_signature(data.get("files", [])) _set_pending_prepare_signature(data, prepare_signature) diff --git a/backend/websearch.py b/backend/websearch.py index e6798d0..eeb7fcc 100644 --- a/backend/websearch.py +++ b/backend/websearch.py @@ -58,7 +58,7 @@ def render_recent_context(messages: Optional[List[Dict[str, str]]], char_limit: # ----- SearXNG search & fetching ------------------------------------------------- -from urllib.parse import urlparse +from urllib.parse import urljoin, urlparse import itertools # HTTP client tuning (keep-alive pool + HTTP/2) @@ -199,6 +199,83 @@ async def fetch_page(client: httpx.AsyncClient, url: str) -> str: print(f"[web] fetch_page error url={url}: {repr(e)}") return "" + +async def fetch_website_snapshot(url: str) -> Dict[str, Any]: + """Fetch one user-supplied web page and return a durable text snapshot.""" + requested_url = str(url or "").strip() + parsed = urlparse(requested_url) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ValueError("Enter a valid http:// or https:// website URL.") + if not _is_probably_html_url(requested_url): + raise ValueError("This URL does not appear to point to a web page.") + + async with httpx.AsyncClient( + headers=DEFAULT_HEADERS, + follow_redirects=True, + http2=True, + limits=HTTP_LIMITS, + timeout=HTTP_TIMEOUT, + ) as client: + async with client.stream("GET", requested_url) as response: + response.raise_for_status() + content_type = (response.headers.get("content-type") or "").lower() + if content_type and not ( + "text/html" in content_type + or "application/xhtml+xml" in content_type + or content_type.startswith("text/") + ): + raise ValueError(f"Unsupported website content type: {content_type.split(';', 1)[0]}") + + payload = bytearray() + async for chunk in response.aiter_bytes(): + if not chunk: + continue + remaining = MAX_BYTES_PER_PAGE - len(payload) + if remaining <= 0: + break + payload.extend(chunk[:remaining]) + + encoding = response.encoding or "utf-8" + html = bytes(payload).decode(encoding, errors="ignore") + final_url = str(response.url) + + try: + soup = BeautifulSoup(html, "lxml") + except Exception: + soup = BeautifulSoup(html, "html.parser") + + title = "" + if soup.title: + title = re.sub(r"\s+", " ", soup.title.get_text(" ", strip=True)).strip() + if not title: + heading = soup.find(["h1", "h2"]) + if heading: + title = re.sub(r"\s+", " ", heading.get_text(" ", strip=True)).strip() + + canonical_url = final_url + canonical = soup.find("link", rel=lambda value: value and "canonical" in str(value).lower()) + if canonical and canonical.get("href"): + candidate = urljoin(final_url, str(canonical.get("href")).strip()) + candidate_parsed = urlparse(candidate) + if candidate_parsed.scheme in {"http", "https"} and candidate_parsed.netloc: + canonical_url = candidate + + text = clean_text(html) + if len(text.strip()) < 80: + raise ValueError("The website did not contain enough readable text to save.") + + return { + "requested_url": requested_url, + "url": canonical_url, + "final_url": final_url, + "title": title or urlparse(final_url).netloc, + "text": text.strip(), + "html": html, + "content_type": content_type.split(";", 1)[0] or "text/html", + "etag": response.headers.get("etag"), + "last_modified": response.headers.get("last-modified"), + } + async def gather_pages( client: httpx.AsyncClient, urls: List[str],