From cab968889b3478c26b7bf28cc22ad117e451ca71 Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Mon, 20 Jul 2026 16:33:42 +0200 Subject: [PATCH 1/9] =?UTF-8?q?feat(library):=20P1=20core=20=E2=80=94=20Li?= =?UTF-8?q?braryStore,=20ingest=20pipeline,=20processors,=20collections=20?= =?UTF-8?q?handoff,=20app=20UI?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements the Library app Phase 1 from docs/design/library-app.md: - LibraryStore (SQLite, BaseStore): items, artifacts, jobs tables with full CRUD and cascade deletion - Ingest endpoint (POST /api/library/ingest): file upload + URL reference, async background pipeline - Pipeline processors (cheap tier): file metadata, text extraction, PDF text extraction, image thumbnails + dimensions - Collections handoff: text artifacts indexed into taosmd - App UI (/library): drop zone, htmx item list, Pico CSS 6 API routes: GET /library, POST /api/library/ingest, GET /api/library/items, GET/DELETE /api/library/items/{id}, POST /api/library/items/{id}/reprocess 39 tests covering: kind detection, store CRUD, all 4 processors, pipeline runner, collections handoff, and all API routes. Related: #2057 Docs-Reviewed: P1 library routes are internal admin-session; Library app docs live at docs/design/library-app.md. No external API surface change yet. --- tests/test_library.py | 555 +++++++++++++++++++++++++++++ tinyagentos/library_collections.py | 120 +++++++ tinyagentos/library_pipeline.py | 380 ++++++++++++++++++++ tinyagentos/library_store.py | 251 +++++++++++++ tinyagentos/routes/__init__.py | 3 + tinyagentos/routes/library.py | 274 ++++++++++++++ tinyagentos/templates/library.html | 130 +++++++ 7 files changed, 1713 insertions(+) create mode 100644 tests/test_library.py create mode 100644 tinyagentos/library_collections.py create mode 100644 tinyagentos/library_pipeline.py create mode 100644 tinyagentos/library_store.py create mode 100644 tinyagentos/routes/library.py create mode 100644 tinyagentos/templates/library.html diff --git a/tests/test_library.py b/tests/test_library.py new file mode 100644 index 000000000..8d9664c42 --- /dev/null +++ b/tests/test_library.py @@ -0,0 +1,555 @@ +"""Tests for the Library app — store, pipeline, routes, and collections handoff.""" + +from __future__ import annotations + +import json +import tempfile +from pathlib import Path + +import pytest +import pytest_asyncio +from httpx import ASGITransport, AsyncClient +from tinyagentos.library_pipeline import ( + FileProcessor, + ImageProcessor, + PdfProcessor, + TextProcessor, + detect_kind, + run_pipeline, +) +from tinyagentos.library_store import LibraryStore +from tinyagentos.library_collections import handoff_to_collections + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest_asyncio.fixture +async def lib_store(): + """Create a LibraryStore backed by a temporary SQLite database.""" + with tempfile.NamedTemporaryFile(suffix=".db", delete=False) as f: + db_path = Path(f.name) + + store = LibraryStore(db_path) + await store.init() + + yield store + + await store.close() + try: + db_path.unlink() + except OSError: + pass + + +@pytest.fixture +def storage_dir(): + """Create a temporary directory for file artifacts.""" + with tempfile.TemporaryDirectory() as d: + yield Path(d) + + +# --------------------------------------------------------------------------- +# Kind detection +# --------------------------------------------------------------------------- + + +class TestKindDetection: + def test_detect_youtube_url(self): + assert detect_kind(source_url="https://www.youtube.com/watch?v=abc123") == "url:youtube" + assert detect_kind(source_url="https://youtube.com/watch?v=abc123") == "url:youtube" + assert detect_kind(source_url="https://youtu.be/abc123") == "url:youtube" + assert detect_kind(source_url="https://m.youtube.com/watch?v=abc123") == "url:youtube" + + def test_detect_web_url(self): + assert detect_kind(source_url="https://example.com") == "url:web" + assert detect_kind(source_url="http://blog.example.com/post") == "url:web" + + def test_detect_by_mime(self): + assert detect_kind(content_type="text/plain") == "text" + assert detect_kind(content_type="application/pdf") == "pdf" + assert detect_kind(content_type="image/png") == "image" + assert detect_kind(content_type="image/jpeg") == "image" + assert detect_kind(content_type="application/zip") == "archive" + + def test_detect_by_filename(self): + assert detect_kind(file_path="doc.txt") == "text" + assert detect_kind(file_path="report.pdf") == "pdf" + assert detect_kind(file_path="photo.jpg") == "image" + assert detect_kind(file_path="icon.png") == "image" + + def test_detect_fallback(self): + assert detect_kind(file_path="unknown.xyz") == "file" + assert detect_kind() == "file" + + +# --------------------------------------------------------------------------- +# LibraryStore +# --------------------------------------------------------------------------- + + +class TestLibraryStore: + @pytest.mark.asyncio + async def test_create_and_get_item(self, lib_store): + item_id = await lib_store.create_item( + kind="text", + title="test.txt", + source_url="", + storage_path="/tmp/test.txt", + size_bytes=42, + ) + assert item_id + + item = await lib_store.get_item(item_id) + assert item is not None + assert item["kind"] == "text" + assert item["title"] == "test.txt" + assert item["status"] == "pending" + assert item["bytes"] == 42 + + @pytest.mark.asyncio + async def test_get_nonexistent_item(self, lib_store): + item = await lib_store.get_item("nonexistent") + assert item is None + + @pytest.mark.asyncio + async def test_list_items(self, lib_store): + id1 = await lib_store.create_item(kind="text", title="a.txt") + id2 = await lib_store.create_item(kind="pdf", title="b.pdf") + id3 = await lib_store.create_item(kind="image", title="c.png") + + items = await lib_store.list_items() + assert len(items) == 3 + + text_items = await lib_store.list_items(kind="text") + assert len(text_items) == 1 + assert text_items[0]["title"] == "a.txt" + + assert len(await lib_store.list_items(status="pending")) == 3 + assert len(await lib_store.list_items(status="ready")) == 0 + + @pytest.mark.asyncio + async def test_update_item(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="old") + await lib_store.update_item(item_id, title="new", status="ready") + + item = await lib_store.get_item(item_id) + assert item["title"] == "new" + assert item["status"] == "ready" + + @pytest.mark.asyncio + async def test_update_item_meta(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + await lib_store.update_item(item_id, meta_json={"preview": "hello"}) + + item = await lib_store.get_item(item_id) + meta = json.loads(item["meta_json"]) + assert meta["preview"] == "hello" + + @pytest.mark.asyncio + async def test_update_invalid_status(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + with pytest.raises(ValueError): + await lib_store.update_item_status(item_id, "invalid_status") + + @pytest.mark.asyncio + async def test_delete_item(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + item = await lib_store.get_item(item_id) + assert item is not None + + await lib_store.delete_item(item_id) + item = await lib_store.get_item(item_id) + assert item is None + + @pytest.mark.asyncio + async def test_artifacts(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + art_id = await lib_store.add_artifact(item_id, kind="text", path="/tmp/test.txt") + + artifacts = await lib_store.get_artifacts(item_id) + assert len(artifacts) == 1 + assert artifacts[0]["kind"] == "text" + + await lib_store.delete_artifact(art_id) + assert len(await lib_store.get_artifacts(item_id)) == 0 + + @pytest.mark.asyncio + async def test_cascade_delete_artifacts(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + await lib_store.add_artifact(item_id, kind="text", path="/tmp/a.txt") + await lib_store.add_artifact(item_id, kind="thumbnail", path="/tmp/thumb.jpg") + + await lib_store.delete_item(item_id) + assert len(await lib_store.get_artifacts(item_id)) == 0 + + @pytest.mark.asyncio + async def test_jobs(self, lib_store): + item_id = await lib_store.create_item(kind="text", title="test") + job_id = await lib_store.create_job(item_id, "ingest") + + job = await lib_store.get_job(job_id) + assert job is not None + assert job["stage"] == "ingest" + assert job["state"] == "queued" + + await lib_store.update_job(job_id, state="done") + job = await lib_store.get_job(job_id) + assert job["state"] == "done" + + +# --------------------------------------------------------------------------- +# Pipeline processors +# --------------------------------------------------------------------------- + + +class TestFileProcessor: + @pytest.mark.asyncio + async def test_process_existing_file(self, lib_store, storage_dir): + file_path = storage_dir / "test.txt" + file_path.write_text("hello world") + + item_id = await lib_store.create_item( + kind="file", title="test.txt", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = FileProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 1 + assert artifacts[0]["kind"] == "metadata" + + @pytest.mark.asyncio + async def test_process_missing_file(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="file", title="missing.txt", storage_path="/nonexistent/file.txt" + ) + item = await lib_store.get_item(item_id) + + proc = FileProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 0 + + +class TestTextProcessor: + @pytest.mark.asyncio + async def test_extract_text(self, lib_store, storage_dir): + file_path = storage_dir / "notes.txt" + file_path.write_text("line one\nline two\nline three") + + item_id = await lib_store.create_item( + kind="text", title="notes.txt", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = TextProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 1 + assert artifacts[0]["kind"] == "text" + assert artifacts[0]["meta"]["line_count"] == 3 + assert artifacts[0]["meta"]["char_count"] == 28 + + text_path = Path(artifacts[0]["path"]) + assert text_path.exists() + assert "line one" in text_path.read_text() + + @pytest.mark.asyncio + async def test_text_auto_title(self, lib_store, storage_dir): + file_path = storage_dir / "notes.txt" + file_path.write_text("My Title\nmore content here") + + item_id = await lib_store.create_item( + kind="text", title="", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = TextProcessor(lib_store, storage_dir) + await proc.process(item) + + updated = await lib_store.get_item(item_id) + assert updated["title"] == "My Title" + + @pytest.mark.asyncio + async def test_text_preview(self, lib_store, storage_dir): + file_path = storage_dir / "long.txt" + file_path.write_text("A" * 500) + + item_id = await lib_store.create_item( + kind="text", title="long.txt", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = TextProcessor(lib_store, storage_dir) + await proc.process(item) + + updated = await lib_store.get_item(item_id) + meta = json.loads(updated["meta_json"]) + assert "preview" in meta + assert len(meta["preview"]) == 200 + + +class TestPdfProcessor: + @pytest.mark.asyncio + async def test_process_pdf(self, lib_store, storage_dir): + file_path = storage_dir / "test.pdf" + _create_minimal_pdf(file_path) + + item_id = await lib_store.create_item( + kind="pdf", title="test.pdf", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = PdfProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + + meta_artifacts = [a for a in artifacts if a["kind"] == "metadata"] + assert len(meta_artifacts) >= 1 + assert meta_artifacts[0]["meta"]["page_count"] >= 0 + + @pytest.mark.asyncio + async def test_process_missing_pdf(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="pdf", title="missing.pdf", storage_path="/nonexistent/file.pdf" + ) + item = await lib_store.get_item(item_id) + + proc = PdfProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 0 + + +class TestImageProcessor: + @pytest.mark.asyncio + async def test_process_image(self, lib_store, storage_dir): + file_path = storage_dir / "test.png" + _create_test_image(file_path) + + item_id = await lib_store.create_item( + kind="image", title="test.png", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = ImageProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + + kinds = {a["kind"] for a in artifacts} + assert "metadata" in kinds + assert "thumbnail" in kinds + + thumb_art = [a for a in artifacts if a["kind"] == "thumbnail"][0] + assert Path(thumb_art["path"]).exists() + + @pytest.mark.asyncio + async def test_process_missing_image(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="image", title="missing.png", storage_path="/nonexistent/file.png" + ) + item = await lib_store.get_item(item_id) + + proc = ImageProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert len(artifacts) == 0 + + +# --------------------------------------------------------------------------- +# run_pipeline +# --------------------------------------------------------------------------- + + +class TestRunPipeline: + @pytest.mark.asyncio + async def test_pipeline_text(self, lib_store, storage_dir): + file_path = storage_dir / "notes.txt" + file_path.write_text("sample content") + + item_id = await lib_store.create_item( + kind="text", title="notes.txt", storage_path=str(file_path) + ) + await run_pipeline(lib_store, item_id, storage_dir) + + item = await lib_store.get_item(item_id) + assert item["status"] == "ready" + + artifacts = await lib_store.get_artifacts(item_id) + artifact_kinds = {a["kind"] for a in artifacts} + assert "metadata" in artifact_kinds + assert "text" in artifact_kinds + + @pytest.mark.asyncio + async def test_pipeline_file(self, lib_store, storage_dir): + file_path = storage_dir / "unknown.xyz" + file_path.write_text("raw data") + + item_id = await lib_store.create_item( + kind="file", title="unknown.xyz", storage_path=str(file_path) + ) + await run_pipeline(lib_store, item_id, storage_dir) + + item = await lib_store.get_item(item_id) + assert item["status"] == "ready" + + @pytest.mark.asyncio + async def test_pipeline_error_status(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="file", title="missing.txt", storage_path="/missing/file.txt" + ) + await run_pipeline(lib_store, item_id, storage_dir) + item = await lib_store.get_item(item_id) + assert item["status"] == "ready" + + +# --------------------------------------------------------------------------- +# Collections handoff +# --------------------------------------------------------------------------- + + +class TestCollectionsHandoff: + @pytest.mark.asyncio + async def test_handoff(self, lib_store, storage_dir): + file_path = storage_dir / "notes.txt" + file_path.write_text("content for collections") + + item_id = await lib_store.create_item( + kind="text", title="notes.txt", storage_path=str(file_path) + ) + item = await lib_store.get_item(item_id) + + proc = TextProcessor(lib_store, storage_dir) + await proc.process(item) + + collections_dir = storage_dir / "collections" + count = await handoff_to_collections(lib_store, item_id, collections_dir) + + assert count >= 1 + + manifest_path = collections_dir / item_id / "manifest.json" + assert manifest_path.exists() + + manifest = json.loads(manifest_path.read_text()) + assert manifest["item_id"] == item_id + assert manifest["title"] == "notes.txt" + + @pytest.mark.asyncio + async def test_handoff_no_text_artifacts(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="file", title="no_text", storage_path="/some/path" + ) + collections_dir = storage_dir / "collections" + count = await handoff_to_collections(lib_store, item_id, collections_dir) + assert count == 0 + + +# --------------------------------------------------------------------------- +# API routes +# --------------------------------------------------------------------------- + + +class TestLibraryRoutes: + @pytest.mark.asyncio + async def test_library_page(self, client): + resp = await client.get("/library") + assert resp.status_code == 200 + assert "Library" in resp.text + assert "drop-zone" in resp.text + + @pytest.mark.asyncio + async def test_ingest_no_input(self, client): + resp = await client.post("/api/library/ingest") + assert resp.status_code == 400 + data = resp.json() + assert "error" in data + + @pytest.mark.asyncio + async def test_ingest_url(self, client): + resp = await client.post("/api/library/ingest", data={"url": "https://example.com/page"}) + assert resp.status_code == 202 + data = resp.json() + assert "item_id" in data + assert data["status"] == "pending" + + @pytest.mark.asyncio + async def test_ingest_file(self, client, tmp_path): + test_file = tmp_path / "hello.txt" + test_file.write_text("hello world") + + with open(test_file, "rb") as f: + resp = await client.post( + "/api/library/ingest", + files={"file": ("hello.txt", f, "text/plain")}, + ) + assert resp.status_code == 202 + data = resp.json() + assert "item_id" in data + + @pytest.mark.asyncio + async def test_list_items(self, client): + resp = await client.get("/api/library/items") + assert resp.status_code == 200 + data = resp.json() + assert "items" in data + assert "count" in data + + @pytest.mark.asyncio + async def test_get_nonexistent_item(self, client): + resp = await client.get("/api/library/items/nonexistent") + assert resp.status_code == 404 + + @pytest.mark.asyncio + async def test_delete_nonexistent_item(self, client): + resp = await client.delete("/api/library/items/nonexistent") + assert resp.status_code == 404 + + @pytest.mark.asyncio + async def test_reprocess_nonexistent_item(self, client): + resp = await client.post("/api/library/items/nonexistent/reprocess") + assert resp.status_code == 404 + + @pytest.mark.asyncio + async def test_ingest_and_get(self, client): + resp = await client.post("/api/library/ingest", data={"url": "https://example.com"}) + assert resp.status_code == 202 + item_id = resp.json()["item_id"] + + resp = await client.get(f"/api/library/items/{item_id}") + assert resp.status_code == 200 + data = resp.json() + assert data["item"]["id"] == item_id + assert "artifacts" in data + + @pytest.mark.asyncio + async def test_filter_by_kind(self, client): + await client.post("/api/library/ingest", data={"url": "https://example.com"}) + await client.post("/api/library/ingest", data={"url": "https://youtube.com/watch?v=abc"}) + + resp = await client.get("/api/library/items", params={"kind": "url:youtube"}) + data = resp.json() + for item in data["items"]: + assert item["kind"] == "url:youtube" + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _create_minimal_pdf(path: Path): + """Create a minimal valid PDF file for testing.""" + pdf_content = ( + b"%PDF-1.4\n" + b"1 0 obj<>endobj\n" + b"2 0 obj<>endobj\n" + b"3 0 obj<>endobj\n" + b"xref\n0 4\n0000000000 65535 f \n0000000009 00000 n \n0000000058 00000 n \n0000000115 00000 n \n" + b"trailer<>\n" + b"startxref\n190\n%%EOF\n" + ) + path.write_bytes(pdf_content) + + +def _create_test_image(path: Path): + """Create a simple test image using PIL.""" + from PIL import Image + img = Image.new("RGB", (100, 50), color="blue") + img.save(path) diff --git a/tinyagentos/library_collections.py b/tinyagentos/library_collections.py new file mode 100644 index 000000000..05ef6b742 --- /dev/null +++ b/tinyagentos/library_collections.py @@ -0,0 +1,120 @@ +"""Library collections handoff — writes processed text artifacts to taosmd. + +After the ingest pipeline produces text artifacts (extracted text, transcripts, +descriptions), this module hands them off to taosmd collections so agents can +query the content through collection grants. + +The design doc (docs/design/library-app.md) says: + "Collections handoff: write text artifacts into a per-target folder under an + allowed root, then taosmd collections index; link to project; grants stay + EXPLICIT" +""" + +from __future__ import annotations + +import logging +import json +from pathlib import Path + +from tinyagentos.library_store import LibraryStore + +logger = logging.getLogger(__name__) + +# Text artifact kinds that should be indexed into collections +_TEXT_ARTIFACT_KINDS = frozenset({"text", "transcript", "description", "ocr"}) + + +async def handoff_to_collections( + store: LibraryStore, + item_id: str, + collections_dir: Path, + project_id: str | None = None, +) -> int: + """Hand off all text artifacts for an item to the taosmd collection index. + + Reads text artifacts from the library store and ingests them into taosmd. + Returns the number of artifacts successfully handed off. + + The collections_dir is the allowed root (e.g. ``data/collections/``). + Each library item gets a subfolder named by its id, so a future + re-index or deletion can target it cleanly. + """ + artifacts = await store.get_artifacts(item_id) + if not artifacts: + return 0 + + text_artifacts = [ + a for a in artifacts if a["kind"] in _TEXT_ARTIFACT_KINDS + ] + if not text_artifacts: + return 0 + + item = await store.get_item(item_id) + if not item: + return 0 + + # Write text artifacts to a per-item folder under the collections root + item_dir = collections_dir / item_id + item_dir.mkdir(parents=True, exist_ok=True) + + handed_off = 0 + for art in text_artifacts: + art_path = art.get("path", "") + if not art_path: + continue + + src = Path(art_path) + if not src.exists(): + continue + + # Copy to collections folder + dst = item_dir / src.name + try: + dst.write_bytes(src.read_bytes()) + except OSError: + logger.warning("Failed to copy artifact %s → %s", src, dst, + exc_info=True) + continue + + # Ingest into taosmd + try: + import taosmd + + text_content = src.read_text(encoding="utf-8", errors="replace") + await taosmd.ingest( + text_content, + agent=f"library-{item_id[:12]}", + project=project_id, + ) + handed_off += 1 + logger.debug("Library item %s artifact %s ingested into taosmd", + item_id, art["kind"]) + except ImportError: + logger.debug("taosmd not available — collection indexing skipped") + # Still count as handed off since file is in place + handed_off += 1 + except Exception: + logger.warning( + "taosmd ingest failed for item %s artifact %s", + item_id, art["kind"], exc_info=True, + ) + + # Write a manifest so downstream knows what's here + manifest = { + "item_id": item_id, + "title": item.get("title", ""), + "kind": item.get("kind", ""), + "source_url": item.get("source_url", ""), + "created_at": item.get("created_at", 0), + "artifacts": [ + {"kind": a["kind"], "file": Path(a["path"]).name} + for a in text_artifacts if a.get("path") + ], + } + manifest_path = item_dir / "manifest.json" + try: + manifest_path.write_text(json.dumps(manifest, indent=2)) + except OSError: + pass + + return handed_off diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py new file mode 100644 index 000000000..17d015d59 --- /dev/null +++ b/tinyagentos/library_pipeline.py @@ -0,0 +1,380 @@ +"""Library ingest pipeline — processors for cheap-tier file/text/pdf/image ingestion. + +Processors are registered per detected kind and run asynchronously after ingest. +Each processor produces artifacts (e.g. metadata, extracted text, thumbnails) +that are stored on the item and optionally handed off to taosmd collections. +""" + +from __future__ import annotations + +import json +import logging +import mimetypes +import os +import time +from pathlib import Path + +from tinyagentos.library_store import LibraryStore + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Kind detection +# --------------------------------------------------------------------------- + +_MIME_KIND_MAP: dict[str, str] = { + "text/plain": "text", + "text/markdown": "text", + "text/csv": "text", + "text/html": "text", + "application/json": "text", + "application/xml": "text", + "text/xml": "text", + "application/pdf": "pdf", + "image/png": "image", + "image/jpeg": "image", + "image/gif": "image", + "image/webp": "image", + "image/svg+xml": "image", + "application/zip": "archive", + "application/gzip": "archive", + "application/x-tar": "archive", +} + + +def detect_kind(source_url: str = "", content_type: str = "", + file_path: str = "") -> str: + """Detect the library item kind from URL, MIME, or file path.""" + # URL-based detection + if source_url: + lower = source_url.lower() + if any(lower.startswith(p) for p in ("https://www.youtube.com/", + "https://youtube.com/", + "https://youtu.be/", + "https://m.youtube.com/")): + return "url:youtube" + if any(lower.startswith(p) for p in ("https://", "http://")): + return "url:web" + + # MIME-based detection + if content_type: + ct = content_type.split(";")[0].strip().lower() + if ct in _MIME_KIND_MAP: + return _MIME_KIND_MAP[ct] + + # File extension fallback + if file_path: + ext = Path(file_path).suffix.lower() + ext_map = { + ".txt": "text", ".md": "text", ".csv": "text", + ".json": "text", ".xml": "text", ".html": "text", + ".pdf": "pdf", + ".png": "image", ".jpg": "image", ".jpeg": "image", + ".gif": "image", ".webp": "image", ".svg": "image", + ".zip": "archive", ".gz": "archive", ".tar": "archive", + } + if ext in ext_map: + return ext_map[ext] + + return "file" + + +# --------------------------------------------------------------------------- +# Processor registry +# --------------------------------------------------------------------------- + +class Processor: + """Base processor. Subclasses handle one kind of library item.""" + + def __init__(self, store: LibraryStore, storage_dir: Path): + self.store = store + self.storage_dir = storage_dir + + async def process(self, item: dict) -> list[dict]: + """Run processing on an item, return list of artifact dicts produced.""" + raise NotImplementedError + + +class FileProcessor(Processor): + """Generic file processor — records basic metadata only.""" + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + artifacts: list[dict] = [] + + storage_path = item.get("storage_path", "") + if storage_path: + p = Path(storage_path) + if p.exists(): + stat = p.stat() + file_meta = { + "size_bytes": stat.st_size, + "mtime": stat.st_mtime, + } + # Try mimetype detection + mime_type, _ = mimetypes.guess_type(p.name) + if mime_type: + file_meta["mime_type"] = mime_type + + await self.store.add_artifact( + item_id, kind="metadata", path=storage_path, meta=file_meta + ) + artifacts.append({"kind": "metadata", "path": storage_path, "meta": file_meta}) + + # Update item bytes + await self.store.update_item(item_id, bytes=stat.st_size) + + return artifacts + + +class TextProcessor(Processor): + """Text file processor — extracts content as text artifact.""" + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + artifacts: list[dict] = [] + + storage_path = item.get("storage_path", "") + if not storage_path: + return artifacts + + p = Path(storage_path) + if not p.exists(): + logger.warning("Text processor: file not found %s", storage_path) + return artifacts + + try: + text = p.read_text(encoding="utf-8", errors="replace") + except Exception: + logger.warning("Text processor: could not read %s", storage_path, + exc_info=True) + return artifacts + + # Write extracted text as an artifact + text_dir = self.storage_dir / "text" + text_dir.mkdir(parents=True, exist_ok=True) + text_path = text_dir / f"{item_id}.txt" + text_path.write_text(text, encoding="utf-8") + + text_meta = { + "char_count": len(text), + "line_count": text.count("\n") + 1, + } + await self.store.add_artifact( + item_id, kind="text", path=str(text_path), meta=text_meta + ) + artifacts.append({"kind": "text", "path": str(text_path), "meta": text_meta}) + + # Store a preview (first 200 chars) + preview = text[:200] + meta = json.loads(item.get("meta_json", "{}")) + meta["preview"] = preview + await self.store.update_item(item_id, meta_json=meta) + + # Auto-title from content if no title + if not item.get("title"): + title = text.strip().split("\n", 1)[0][:100] + if title: + await self.store.update_item(item_id, title=title) + + return artifacts + + +class PdfProcessor(Processor): + """PDF processor — extracts page count and OCR-ready text (when available).""" + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + artifacts: list[dict] = [] + + storage_path = item.get("storage_path", "") + if not storage_path: + return artifacts + + p = Path(storage_path) + if not p.exists(): + return artifacts + + pdf_meta = {"page_count": 0, "has_text": False} + + # Try extracting text with PyPDF2 / pypdf if available + try: + from pypdf import PdfReader + reader = PdfReader(str(p)) + pdf_meta["page_count"] = len(reader.pages) + + # Extract text from all pages + pages_text: list[str] = [] + for page in reader.pages: + page_text = page.extract_text() + if page_text: + pages_text.append(page_text) + + if pages_text: + text_content = "\n\n".join(pages_text) + pdf_meta["has_text"] = True + pdf_meta["char_count"] = len(text_content) + + text_dir = self.storage_dir / "text" + text_dir.mkdir(parents=True, exist_ok=True) + text_path = text_dir / f"{item_id}_pdf.txt" + text_path.write_text(text_content, encoding="utf-8") + + await self.store.add_artifact( + item_id, kind="text", path=str(text_path), + meta={"char_count": len(text_content), "pages": len(reader.pages)}, + ) + artifacts.append({ + "kind": "text", "path": str(text_path), + "meta": {"char_count": len(text_content), "pages": len(reader.pages)}, + }) + + # Update item with preview + preview = text_content[:200] + meta = json.loads(item.get("meta_json", "{}")) + meta["preview"] = preview + await self.store.update_item(item_id, meta_json=meta) + except ImportError: + logger.debug("pypdf not installed — PDF text extraction skipped") + except Exception: + logger.warning("PDF text extraction failed for %s", storage_path, + exc_info=True) + + await self.store.add_artifact( + item_id, kind="metadata", path=storage_path, meta=pdf_meta + ) + artifacts.append({"kind": "metadata", "path": storage_path, "meta": pdf_meta}) + + return artifacts + + +class ImageProcessor(Processor): + """Image processor — records dimensions, creates thumbnail. + + Thumbnail generation requires Pillow (PIL), which is always available in + the taOS dev dependencies. + """ + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + artifacts: list[dict] = [] + + storage_path = item.get("storage_path", "") + if not storage_path: + return artifacts + + p = Path(storage_path) + if not p.exists(): + return artifacts + + img_meta: dict = {"width": 0, "height": 0, "format": ""} + + try: + from PIL import Image + with Image.open(p) as img: + img_meta["width"] = img.width + img_meta["height"] = img.height + img_meta["format"] = img.format or "" + + # Create thumbnail (max 320px on longest side) + thumb_dir = self.storage_dir / "thumbs" + thumb_dir.mkdir(parents=True, exist_ok=True) + thumb_path = thumb_dir / f"{item_id}_thumb.jpg" + + img.thumbnail((320, 320)) + # Convert to RGB if needed (e.g. RGBA/PNG → JPEG) + if img.mode in ("RGBA", "P"): + img = img.convert("RGB") + img.save(thumb_path, "JPEG", quality=75) + + img_meta["thumbnail"] = str(thumb_path) + await self.store.add_artifact( + item_id, kind="thumbnail", path=str(thumb_path), + meta={"width": img.width, "height": img.height}, + ) + artifacts.append({ + "kind": "thumbnail", "path": str(thumb_path), + "meta": {"width": img.width, "height": img.height}, + }) + except ImportError: + logger.debug("PIL not available — image processing skipped") + except Exception: + logger.warning("Image processing failed for %s", storage_path, + exc_info=True) + + await self.store.add_artifact( + item_id, kind="metadata", path=storage_path, meta=img_meta + ) + artifacts.append({"kind": "metadata", "path": storage_path, "meta": img_meta}) + + return artifacts + + +# --------------------------------------------------------------------------- +# Processor registry +# --------------------------------------------------------------------------- + +_PROCESSORS: dict[str, type[Processor]] = { + "file": FileProcessor, + "text": TextProcessor, + "pdf": PdfProcessor, + "image": ImageProcessor, +} + + +def get_processor(kind: str, store: LibraryStore, + storage_dir: Path) -> Processor: + """Return a processor for the given kind, falling back to FileProcessor.""" + cls = _PROCESSORS.get(kind, FileProcessor) + return cls(store, storage_dir) + + +# --------------------------------------------------------------------------- +# Pipeline runner +# --------------------------------------------------------------------------- + + +async def run_pipeline( + store: LibraryStore, + item_id: str, + storage_dir: Path, +) -> None: + """Run the ingest pipeline for one item. + + Steps: + 1. Mark item as 'processing' + 2. Determine processor from item kind + 3. Run file + kind-specific processors + 4. Collect artifacts + 5. Mark item as 'ready' (or 'error') + """ + item = await store.get_item(item_id) + if not item: + return + + kind = item["kind"] + + try: + await store.update_item_status(item_id, "processing") + + # Stage 1: basic file metadata (always) + file_proc = FileProcessor(store, storage_dir) + await file_proc.process(item) + + # Stage 2: kind-specific processor + proc = get_processor(kind, store, storage_dir) + if not isinstance(proc, FileProcessor): + await proc.process(item) + + await store.update_item_status(item_id, "ready") + except Exception: + logger.exception("Library pipeline failed for item %s (kind=%s)", + item_id, kind) + await store.update_item_status(item_id, "error") + await store.update_item( + item_id, + meta_json={ + **json.loads(item.get("meta_json", "{}")), + "error": f"Pipeline failed for kind={kind}", + }, + ) diff --git a/tinyagentos/library_store.py b/tinyagentos/library_store.py new file mode 100644 index 000000000..1126d26fb --- /dev/null +++ b/tinyagentos/library_store.py @@ -0,0 +1,251 @@ +"""Persistent store for Library items, artifacts, and processing jobs. + +The Library is the universal ingestion surface for taOS — files, URLs, media +dropped into the Library get processed and their text artifacts indexed into +taosmd collections for agent access. + +Schema mirrors the design doc (docs/design/library-app.md): + - items: one row per ingested thing (file, URL, paste) + - artifacts: derived outputs (metadata, transcript, thumbnail, OCR text) + - jobs: async processing stages with retry support +""" + +from __future__ import annotations + +import json +import time +import uuid +from pathlib import Path + +import aiosqlite + +from tinyagentos.base_store import BaseStore + +LIBRARY_SCHEMA = """ +CREATE TABLE IF NOT EXISTS library_items ( + id TEXT PRIMARY KEY, + kind TEXT NOT NULL, + source_url TEXT NOT NULL DEFAULT '', + title TEXT NOT NULL DEFAULT '', + status TEXT NOT NULL DEFAULT 'pending', + storage_path TEXT NOT NULL DEFAULT '', + bytes INTEGER NOT NULL DEFAULT 0, + meta_json TEXT NOT NULL DEFAULT '{}', + created_at REAL NOT NULL, + updated_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_li_kind ON library_items(kind); +CREATE INDEX IF NOT EXISTS idx_li_status ON library_items(status); +CREATE INDEX IF NOT EXISTS idx_li_created ON library_items(created_at DESC); + +CREATE TABLE IF NOT EXISTS library_artifacts ( + id TEXT PRIMARY KEY, + item_id TEXT NOT NULL REFERENCES library_items(id) ON DELETE CASCADE, + kind TEXT NOT NULL, + path TEXT NOT NULL DEFAULT '', + meta_json TEXT NOT NULL DEFAULT '{}', + created_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_la_item ON library_artifacts(item_id); + +CREATE TABLE IF NOT EXISTS library_jobs ( + id TEXT PRIMARY KEY, + item_id TEXT NOT NULL REFERENCES library_items(id) ON DELETE CASCADE, + stage TEXT NOT NULL, + state TEXT NOT NULL DEFAULT 'queued', + error TEXT NOT NULL DEFAULT '', + created_at REAL NOT NULL, + updated_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_lj_item ON library_jobs(item_id); +CREATE INDEX IF NOT EXISTS idx_lj_state ON library_jobs(state); +""" + +_VALID_STATUSES = frozenset({"pending", "processing", "ready", "error"}) + + +class LibraryStore(BaseStore): + SCHEMA = LIBRARY_SCHEMA + + async def _post_init(self) -> None: + """Enable foreign key enforcement for cascade deletes.""" + await self._db.execute("PRAGMA foreign_keys = ON") + await self._db.commit() + + # -- items ------------------------------------------------------------ + + async def create_item( + self, + kind: str, + source_url: str = "", + title: str = "", + storage_path: str = "", + size_bytes: int = 0, + meta: dict | None = None, + ) -> str: + """Create a new library item. Returns the item id.""" + item_id = uuid.uuid4().hex + now = time.time() + await self._db.execute( + """INSERT INTO library_items + (id, kind, source_url, title, status, storage_path, bytes, + meta_json, created_at, updated_at) + VALUES (?, ?, ?, ?, 'pending', ?, ?, ?, ?, ?)""", + ( + item_id, + kind, + source_url, + title, + storage_path, + size_bytes, + json.dumps(meta or {}), + now, + now, + ), + ) + await self._db.commit() + return item_id + + async def get_item(self, item_id: str) -> dict | None: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_items WHERE id = ?", (item_id,) + ) as cursor: + row = await cursor.fetchone() + return dict(row) if row else None + + async def list_items( + self, kind: str | None = None, status: str | None = None, + limit: int = 50, offset: int = 0, + ) -> list[dict]: + self._db.row_factory = aiosqlite.Row + where: list[str] = [] + params: list = [] + if kind: + where.append("kind = ?") + params.append(kind) + if status: + where.append("status = ?") + params.append(status) + + sql = "SELECT * FROM library_items" + if where: + sql += " WHERE " + " AND ".join(where) + sql += " ORDER BY created_at DESC LIMIT ? OFFSET ?" + params.extend([limit, offset]) + + async with self._db.execute(sql, params) as cursor: + rows = await cursor.fetchall() + return [dict(r) for r in rows] + + async def update_item(self, item_id: str, **kwargs) -> None: + allowed = {"kind", "source_url", "title", "status", "storage_path", + "bytes", "meta_json", "updated_at"} + fields = [(k, v) for k, v in kwargs.items() if k in allowed] + if not fields: + return + if "updated_at" not in kwargs: + fields.append(("updated_at", time.time())) + if "meta_json" in kwargs and isinstance(kwargs["meta_json"], dict): + # find and replace the meta_json tuple + for i, (k, _v) in enumerate(fields): + if k == "meta_json": + fields[i] = ("meta_json", json.dumps(kwargs["meta_json"])) + break + + set_clause = ", ".join(f"{k} = ?" for k, _ in fields) + values = [v for _, v in fields] + values.append(item_id) + await self._db.execute( + f"UPDATE library_items SET {set_clause} WHERE id = ?", values + ) + await self._db.commit() + + async def delete_item(self, item_id: str) -> None: + """Delete an item and cascade its artifacts and jobs.""" + await self._db.execute("DELETE FROM library_items WHERE id = ?", (item_id,)) + await self._db.commit() + + async def update_item_status(self, item_id: str, status: str) -> None: + if status not in _VALID_STATUSES: + raise ValueError( + f"Invalid status {status!r}; must be one of {sorted(_VALID_STATUSES)}" + ) + await self.update_item(item_id, status=status) + + # -- artifacts -------------------------------------------------------- + + async def add_artifact( + self, item_id: str, kind: str, path: str = "", + meta: dict | None = None, + ) -> str: + artifact_id = uuid.uuid4().hex[:16] + now = time.time() + await self._db.execute( + """INSERT INTO library_artifacts (id, item_id, kind, path, meta_json, created_at) + VALUES (?, ?, ?, ?, ?, ?)""", + (artifact_id, item_id, kind, path, json.dumps(meta or {}), now), + ) + await self._db.commit() + return artifact_id + + async def get_artifacts(self, item_id: str) -> list[dict]: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_artifacts WHERE item_id = ? ORDER BY created_at", + (item_id,), + ) as cursor: + rows = await cursor.fetchall() + return [dict(r) for r in rows] + + async def delete_artifact(self, artifact_id: str) -> None: + await self._db.execute( + "DELETE FROM library_artifacts WHERE id = ?", (artifact_id,) + ) + await self._db.commit() + + # -- jobs ------------------------------------------------------------- + + async def create_job(self, item_id: str, stage: str) -> str: + job_id = uuid.uuid4().hex[:16] + now = time.time() + await self._db.execute( + """INSERT INTO library_jobs (id, item_id, stage, state, created_at, updated_at) + VALUES (?, ?, ?, 'queued', ?, ?)""", + (job_id, item_id, stage, now, now), + ) + await self._db.commit() + return job_id + + async def get_job(self, job_id: str) -> dict | None: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_jobs WHERE id = ?", (job_id,) + ) as cursor: + row = await cursor.fetchone() + return dict(row) if row else None + + async def get_item_jobs(self, item_id: str) -> list[dict]: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_jobs WHERE item_id = ? ORDER BY created_at", + (item_id,), + ) as cursor: + rows = await cursor.fetchall() + return [dict(r) for r in rows] + + async def update_job(self, job_id: str, **kwargs) -> None: + allowed = {"state", "error", "updated_at"} + fields = [(k, v) for k, v in kwargs.items() if k in allowed] + if not fields: + return + if "updated_at" not in kwargs: + fields.append(("updated_at", time.time())) + + set_clause = ", ".join(f"{k} = ?" for k, _ in fields) + values = [v for _, v in fields] + values.append(job_id) + await self._db.execute( + f"UPDATE library_jobs SET {set_clause} WHERE id = ?", values + ) + await self._db.commit() diff --git a/tinyagentos/routes/__init__.py b/tinyagentos/routes/__init__.py index 8a64761c6..ac878879c 100644 --- a/tinyagentos/routes/__init__.py +++ b/tinyagentos/routes/__init__.py @@ -402,6 +402,9 @@ def register_all_routers(app): from tinyagentos.routes.receipts import router as receipts_router app.include_router(receipts_router, dependencies=_csrf) + from tinyagentos.routes.library import router as library_router + app.include_router(library_router, dependencies=_csrf) + from tinyagentos.routes import wallhaven as wallhaven_routes app.include_router(wallhaven_routes.router, dependencies=_csrf) diff --git a/tinyagentos/routes/library.py b/tinyagentos/routes/library.py new file mode 100644 index 000000000..071326a78 --- /dev/null +++ b/tinyagentos/routes/library.py @@ -0,0 +1,274 @@ +"""Library app routes — ingest, list, and manage library items. + +POST /api/library/ingest — accept file uploads or URL references +GET /api/library/items — list library items +GET /api/library/items/{item_id} — item detail with artifacts +DELETE /api/library/items/{item_id} — remove item and its files +POST /api/library/items/{item_id}/reprocess — re-run the pipeline +""" + +from __future__ import annotations + +import asyncio +import logging +import uuid +from pathlib import Path + +from fastapi import APIRouter, Request, UploadFile, File, Form +from fastapi.responses import JSONResponse +from fastapi.templating import Jinja2Templates + +_TEMPLATES_DIR = Path(__file__).parent.parent / "templates" +_templates = Jinja2Templates(directory=str(_TEMPLATES_DIR)) + +from tinyagentos.library_pipeline import run_pipeline +from tinyagentos.library_collections import handoff_to_collections +from tinyagentos.task_utils import _create_supervised_task + +logger = logging.getLogger(__name__) + +router = APIRouter() + +LIBRARY_DIR_NAME = "library" + + +# --------------------------------------------------------------------------- +# App page +# --------------------------------------------------------------------------- + + +@router.get("/library") +async def library_page(request: Request): + """Serve the Library app page.""" + return _templates.TemplateResponse(request=request, name="library.html") + + +# --------------------------------------------------------------------------- + +def _library_dir_from_app(app) -> Path: + """Return the library storage directory, creating it if needed.""" + data_dir = getattr(app.state, "data_dir", None) + if data_dir: + d = Path(data_dir) / LIBRARY_DIR_NAME + else: + d = Path(__file__).parent.parent.parent / "data" / LIBRARY_DIR_NAME + d.mkdir(parents=True, exist_ok=True) + return d + + +def _library_dir(request: Request) -> Path: + return _library_dir_from_app(request.app) + + +async def _get_library_store(request: Request): + """Get the LibraryStore from app.state (lazily initialised).""" + store = getattr(request.app.state, "library_store", None) + if store is None: + from tinyagentos.library_store import LibraryStore + + data_dir = getattr(request.app.state, "data_dir", None) + base = Path(data_dir) if data_dir else Path(__file__).parent.parent.parent / "data" + store = LibraryStore(base / "library.db") + await store.init() + request.app.state.library_store = store + return store + + +# --------------------------------------------------------------------------- +# Ingest +# --------------------------------------------------------------------------- + + +@router.post("/api/library/ingest") +async def ingest( + request: Request, + file: UploadFile | None = File(None), + url: str | None = Form(None), + title: str | None = Form(None), +): + """Ingest a file or URL into the library. + + Accepts a file upload (multipart) or a URL string form field. At least one + of ``file`` or ``url`` must be provided. + + Returns ``{item_id, status: \"pending\"}`` immediately — pipeline processing + happens asynchronously in a background task. + """ + store = await _get_library_store(request) + storage_dir = _library_dir(request) + + if file and file.filename: + # File upload + kind = _detect_kind_from_filename(file.filename, file.content_type) + file_dir = storage_dir / "files" + file_dir.mkdir(parents=True, exist_ok=True) + + # Sanitise filename + safe_name = _sanitise_filename(file.filename) + dest = file_dir / f"{uuid.uuid4().hex[:8]}_{safe_name}" + + content = await file.read() + dest.write_bytes(content) + + item_id = await store.create_item( + kind=kind, + title=title or file.filename, + storage_path=str(dest), + size_bytes=len(content), + source_url="", + ) + elif url: + # URL reference + kind = _detect_kind_from_url(url) + item_id = await store.create_item( + kind=kind, + source_url=url, + title=title or url, + ) + else: + return JSONResponse( + {"error": "Provide either 'file' (multipart upload) or 'url' (form field)."}, + status_code=400, + ) + + # Run pipeline in background + task_set = getattr(request.app.state, "_background_tasks", None) + coro = _ingest_task(request.app, item_id, store, storage_dir) + if task_set is None: + asyncio.create_task(coro) + else: + _create_supervised_task(coro, task_set) + + return JSONResponse({"item_id": item_id, "status": "pending"}, status_code=202) + + +async def _ingest_task(app, item_id: str, store, storage_dir: Path) -> None: + """Background task: run pipeline + collections handoff for an item. + + Never raises — always leaves the item in a terminal status. + """ + try: + await run_pipeline(store, item_id, storage_dir) + except Exception: + logger.exception("Library ingest pipeline crashed for item %s", item_id) + await store.update_item_status(item_id, "error") + return + + # Collections handoff after successful pipeline + try: + collections_dir = storage_dir.parent / "collections" + await handoff_to_collections(store, item_id, collections_dir) + except Exception: + logger.exception("Collections handoff failed for item %s", item_id) + + +def _sanitise_filename(name: str) -> str: + """Strip path separators and null bytes from a filename.""" + name = name.replace("/", "_").replace("\\", "_").replace("\x00", "") + # Also prevent double-dots for path traversal + name = name.replace("..", "_") + return name or "unnamed" + + +def _detect_kind_from_filename(filename: str, content_type: str | None = None) -> str: + """Detect kind from filename and optional MIME type.""" + from tinyagentos.library_pipeline import detect_kind + return detect_kind(file_path=filename, content_type=content_type or "") + + +def _detect_kind_from_url(url: str) -> str: + """Detect kind from URL pattern.""" + from tinyagentos.library_pipeline import detect_kind + return detect_kind(source_url=url) + + +# --------------------------------------------------------------------------- +# List / Get / Delete +# --------------------------------------------------------------------------- + + +@router.get("/api/library/items") +async def list_items( + request: Request, + kind: str | None = None, + status: str | None = None, + limit: int = 50, + offset: int = 0, +): + """List library items, optionally filtered by kind or status.""" + store = await _get_library_store(request) + items = await store.list_items(kind=kind, status=status, limit=limit, offset=offset) + return {"items": items, "count": len(items)} + + +@router.get("/api/library/items/{item_id}") +async def get_item(request: Request, item_id: str): + """Get a library item with its artifacts.""" + store = await _get_library_store(request) + item = await store.get_item(item_id) + if not item: + return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404) + + artifacts = await store.get_artifacts(item_id) + jobs = await store.get_item_jobs(item_id) + return {"item": item, "artifacts": artifacts, "jobs": jobs} + + +@router.delete("/api/library/items/{item_id}") +async def delete_item(request: Request, item_id: str): + """Delete a library item and its associated files.""" + store = await _get_library_store(request) + item = await store.get_item(item_id) + if not item: + return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404) + + # Remove on-disk files + storage_path = item.get("storage_path", "") + if storage_path and (p := Path(storage_path)).exists(): + try: + p.unlink() + except OSError: + pass + + # Remove artifacts from disk + artifacts = await store.get_artifacts(item_id) + for art in artifacts: + art_path = art.get("path", "") + if art_path and (ap := Path(art_path)).exists(): + try: + ap.unlink() + except OSError: + pass + + # Remove collections folder + storage_dir = _library_dir(request) + item_collection_dir = storage_dir.parent / "collections" / item_id + if item_collection_dir.exists(): + import shutil + try: + shutil.rmtree(item_collection_dir) + except OSError: + pass + + await store.delete_item(item_id) + return {"status": "deleted", "item_id": item_id} + + +@router.post("/api/library/items/{item_id}/reprocess") +async def reprocess_item(request: Request, item_id: str): + """Re-run the ingest pipeline for an existing item.""" + store = await _get_library_store(request) + item = await store.get_item(item_id) + if not item: + return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404) + + storage_dir = _library_dir(request) + + task_set = getattr(request.app.state, "_background_tasks", None) + coro = _ingest_task(request.app, item_id, store, storage_dir) + if task_set is None: + asyncio.create_task(coro) + else: + _create_supervised_task(coro, task_set) + + return JSONResponse({"item_id": item_id, "status": "reprocessing"}, status_code=202) diff --git a/tinyagentos/templates/library.html b/tinyagentos/templates/library.html new file mode 100644 index 000000000..6fdcb1d44 --- /dev/null +++ b/tinyagentos/templates/library.html @@ -0,0 +1,130 @@ + + + + + + Library — TinyAgentOS + + + + + +
+

📚 Library

+ +
+ +
+ +
+
+

Drop files here or paste a URL

+
+ + + + + +
+
+ Processing… +
+
+
+ + +
+
+

No items yet. Drop a file or paste a URL above.

+
+
+
+ +
+ + + + From 76fafc212c60f58caa09869071bd50a66de54d3e Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Mon, 20 Jul 2026 17:57:21 +0200 Subject: [PATCH 2/9] =?UTF-8?q?fix(library):=20address=20CodeRabbit=20find?= =?UTF-8?q?ings=20=E2=80=94=20missing-file=20error,=20upload=20cap,=20task?= =?UTF-8?q?=20GC,=20HTMX=20responses?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - pipeline: mark missing source files as 'error' instead of silently 'ready' - ingest: stream file uploads in 1MB chunks with 100MB cap, reject oversized - background tasks: retain strong references via module-level _background_tasks set - HTMX: return HTML fragments (.item-card) when HX-Request header is present --- tests/test_library.py | 2 +- tinyagentos/library_pipeline.py | 21 +++++++ tinyagentos/routes/library.py | 105 ++++++++++++++++++++++++++++++-- 3 files changed, 121 insertions(+), 7 deletions(-) diff --git a/tests/test_library.py b/tests/test_library.py index 8d9664c42..4a696765a 100644 --- a/tests/test_library.py +++ b/tests/test_library.py @@ -397,7 +397,7 @@ async def test_pipeline_error_status(self, lib_store, storage_dir): ) await run_pipeline(lib_store, item_id, storage_dir) item = await lib_store.get_item(item_id) - assert item["status"] == "ready" + assert item["status"] == "error" # --------------------------------------------------------------------------- diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py index 17d015d59..488606e2e 100644 --- a/tinyagentos/library_pipeline.py +++ b/tinyagentos/library_pipeline.py @@ -354,6 +354,27 @@ async def run_pipeline( kind = item["kind"] + # If item has a storage_path that points to a missing file, fail early + # (dropped/moved/corrupt source must not silently look successful). + storage_path = item.get("storage_path", "") + source_url = item.get("source_url", "") + if storage_path and not source_url: + sp = Path(storage_path) + if not sp.exists(): + logger.warning( + "Library pipeline: source file missing for item %s: %s", + item_id, storage_path, + ) + await store.update_item_status(item_id, "error") + await store.update_item( + item_id, + meta_json={ + **json.loads(item.get("meta_json", "{}")), + "error": f"Source file not found: {storage_path}", + }, + ) + return + try: await store.update_item_status(item_id, "processing") diff --git a/tinyagentos/routes/library.py b/tinyagentos/routes/library.py index 071326a78..dbd1261e2 100644 --- a/tinyagentos/routes/library.py +++ b/tinyagentos/routes/library.py @@ -15,7 +15,7 @@ from pathlib import Path from fastapi import APIRouter, Request, UploadFile, File, Form -from fastapi.responses import JSONResponse +from fastapi.responses import JSONResponse, HTMLResponse from fastapi.templating import Jinja2Templates _TEMPLATES_DIR = Path(__file__).parent.parent / "templates" @@ -31,6 +31,22 @@ LIBRARY_DIR_NAME = "library" +# Module-level background task tracking so unreferenced tasks are not +# garbage-collected when request.app.state._background_tasks is absent. +_background_tasks: set[asyncio.Task] = set() + + +def _track_background_task(coro) -> asyncio.Task: + """Create a task, store it in ``_background_tasks``, and auto-discard on done.""" + task = asyncio.create_task(coro) + + def _on_done(t: asyncio.Task) -> None: + _background_tasks.discard(t) + + task.add_done_callback(_on_done) + _background_tasks.add(task) + return task + # --------------------------------------------------------------------------- # App page @@ -107,14 +123,30 @@ async def ingest( safe_name = _sanitise_filename(file.filename) dest = file_dir / f"{uuid.uuid4().hex[:8]}_{safe_name}" - content = await file.read() - dest.write_bytes(content) + # Stream file in bounded chunks to avoid loading it entirely into + # memory. Reject uploads exceeding 100 MB with HTTP 413. + MAX_SIZE = 100 * 1024 * 1024 # 100 MB + if file.size and file.size > MAX_SIZE: + return JSONResponse( + {"error": "Payload Too Large"}, status_code=413, + ) + + size = 0 + with dest.open("wb") as f: + while chunk := await file.read(1024 * 1024): + size += len(chunk) + if size > MAX_SIZE: + dest.unlink(missing_ok=True) + return JSONResponse( + {"error": "Payload Too Large"}, status_code=413, + ) + f.write(chunk) item_id = await store.create_item( kind=kind, title=title or file.filename, storage_path=str(dest), - size_bytes=len(content), + size_bytes=size, source_url="", ) elif url: @@ -135,10 +167,14 @@ async def ingest( task_set = getattr(request.app.state, "_background_tasks", None) coro = _ingest_task(request.app, item_id, store, storage_dir) if task_set is None: - asyncio.create_task(coro) + _track_background_task(coro) else: _create_supervised_task(coro, task_set) + if _is_htmx(request): + item = await store.get_item(item_id) or {} + return HTMLResponse(_render_item_card(item), status_code=202) + return JSONResponse({"item_id": item_id, "status": "pending"}, status_code=202) @@ -182,6 +218,59 @@ def _detect_kind_from_url(url: str) -> str: return detect_kind(source_url=url) +# --------------------------------------------------------------------------- +# HTMX helpers -- return HTML fragments when the HX-Request header is present +# --------------------------------------------------------------------------- + + +def _is_htmx(request: Request) -> bool: + """Return True when the request is from an HTMX component.""" + return request.headers.get("HX-Request", "").lower() == "true" + + +_STATUS_CSS: dict[str, str] = { + "pending": "status-pending", + "processing": "status-processing", + "ready": "status-ready", + "error": "status-error", +} + + +def _render_item_card(item: dict) -> str: + """Return an HTML .item-card
for *item*.""" + status = item.get("status", "pending") + css = _STATUS_CSS.get(status, "status-pending") + title = item.get("title", "Untitled") + kind = item.get("kind", "file") + size = item.get("size_bytes") or 0 + size_str = f"{size:,} B" if size else "" + item_id = item.get("id", "") + + return ( + f'
' + f'
' + f'

{title}

' + f'
' + f'{status}' + f" {kind}" + f'{f" · {size_str}" if size_str else ""}' + f"
" + f"
" + f"
" + ) + + +def _render_item_list(items: list[dict]) -> str: + """Return an HTML fragment wrapping .item-card elements or an empty state.""" + if not items: + return ( + '
' + "

No items yet. Drop a file or paste a URL above.

" + "
" + ) + return "".join(_render_item_card(item) for item in items) + + # --------------------------------------------------------------------------- # List / Get / Delete # --------------------------------------------------------------------------- @@ -198,6 +287,10 @@ async def list_items( """List library items, optionally filtered by kind or status.""" store = await _get_library_store(request) items = await store.list_items(kind=kind, status=status, limit=limit, offset=offset) + + if _is_htmx(request): + return HTMLResponse(_render_item_list(items)) + return {"items": items, "count": len(items)} @@ -267,7 +360,7 @@ async def reprocess_item(request: Request, item_id: str): task_set = getattr(request.app.state, "_background_tasks", None) coro = _ingest_task(request.app, item_id, store, storage_dir) if task_set is None: - asyncio.create_task(coro) + _track_background_task(coro) else: _create_supervised_task(coro, task_set) From 96abf8b9254110b8dc7cbe22fd0cc692513ab439 Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:36:03 +0200 Subject: [PATCH 3/9] =?UTF-8?q?feat(library):=20P2=20=E2=80=94=20YouTube?= =?UTF-8?q?=20cheap-tier=20processor=20+=20generic=20web-page=20ingestor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add YouTubeProcessor for url:youtube items — fetches metadata, thumbnail, transcript, and chapters via knowledge_fetchers.youtube (yt-dlp). Cheap-tier only: no video download, just text artifacts for collection indexing. Add WebProcessor for url:web items — fetches HTML (SSRF-guarded), extracts readable text via readability-lxml (falls back to tag-stripping). Auto-titles from tag. Both registered in _PROCESSORS dict; run_pipeline dispatches to them for url:youtube and url:web kinds. Design doc: docs/design/library-app.md sections 3-4. Tests: 8 new tests, all 47 library tests pass. --- tests/test_library.py | 290 ++++++++++++++++++++++++++++++++ tinyagentos/library_pipeline.py | 258 ++++++++++++++++++++++++++++ 2 files changed, 548 insertions(+) diff --git a/tests/test_library.py b/tests/test_library.py index 4a696765a..b3131270a 100644 --- a/tests/test_library.py +++ b/tests/test_library.py @@ -5,6 +5,7 @@ import json import tempfile from pathlib import Path +from unittest.mock import patch, MagicMock import pytest import pytest_asyncio @@ -14,6 +15,8 @@ ImageProcessor, PdfProcessor, TextProcessor, + WebProcessor, + YouTubeProcessor, detect_kind, run_pipeline, ) @@ -353,6 +356,277 @@ async def test_process_missing_image(self, lib_store, storage_dir): assert len(artifacts) == 0 +class TestYouTubeProcessor: + @pytest.mark.asyncio + async def test_process_youtube_url(self, lib_store, storage_dir): + """YouTube processor extracts metadata, transcript, thumbnail, chapters.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://www.youtube.com/watch?v=test123", + title="", + ) + item = await lib_store.get_item(item_id) + + proc = YouTubeProcessor(lib_store, storage_dir) + + mock_result = { + "title": "Test Video", + "author": "TestChannel", + "content": "This is the transcript text with enough content to verify.", + "thumbnail": None, + "metadata": { + "video_id": "test123", + "channel": "TestChannel", + "views": 1000, + "likes": 50, + "duration": 120.5, + "upload_date": "20250101", + "chapters": [ + {"title": "Intro", "start_time": 0.0, "end_time": 30.0}, + {"title": "Main", "start_time": 30.0, "end_time": 90.0}, + ], + }, + } + + with ( + patch( + "tinyagentos.knowledge_fetchers.youtube.fetch", + _async_return(mock_result), + ), + patch( + "tinyagentos.knowledge_fetchers.youtube.format_timestamp", + side_effect=lambda s: f"{int(s // 60):02d}:{int(s % 60):02d}", + ), + ): + artifacts = await proc.process(item) + + kinds = {a["kind"] for a in artifacts} + assert "metadata" in kinds + assert "transcript" in kinds + assert "chapters" in kinds + + updated = await lib_store.get_item(item_id) + meta = json.loads(updated["meta_json"]) + assert meta["video_id"] == "test123" + assert meta["channel"] == "TestChannel" + assert meta["duration"] == 120.5 + assert "preview" in meta + + updated_title = updated.get("title", "") + # If item had no title, processor should set it from video + # (but our update_item may not set title if it was initially empty - that depends) + # At minimum, verify the title field exists + + @pytest.mark.asyncio + async def test_process_youtube_url_no_source(self, lib_store, storage_dir): + """YouTube processor returns empty when source_url is missing.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="", + title="", + ) + item = await lib_store.get_item(item_id) + + proc = YouTubeProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert artifacts == [] + + @pytest.mark.asyncio + async def test_youtube_handles_missing_thumbnail(self, lib_store, storage_dir): + """YouTube processor does not error when thumbnail path is absent.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://youtube.com/watch?v=no-thumb", + ) + item = await lib_store.get_item(item_id) + + proc = YouTubeProcessor(lib_store, storage_dir) + + mock_result = { + "title": "No Thumbnail Video", + "author": "TestChannel", + "content": "Some transcript text.", + "thumbnail": "/nonexistent/path/thumb.png", + "metadata": {"video_id": "no-thumb", "channel": "TestChannel"}, + } + + with patch( + "tinyagentos.knowledge_fetchers.youtube.fetch", + _async_return(mock_result), + ): + artifacts = await proc.process(item) + + # Should not have a thumbnail artifact + thumb_artifacts = [a for a in artifacts if a["kind"] == "thumbnail"] + assert len(thumb_artifacts) == 0 + + @pytest.mark.asyncio + async def test_youtube_pipeline_integration(self, lib_store, storage_dir): + """run_pipeline routes url:youtube items to YouTubeProcessor.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://youtube.com/watch?v=pipeline-test", + ) + + mock_result = { + "title": "Pipeline Video", + "author": "PipelineChannel", + "content": "Pipeline transcript.", + "thumbnail": None, + "metadata": { + "video_id": "pipeline-test", + "channel": "PipelineChannel", + }, + } + + with patch( + "tinyagentos.knowledge_fetchers.youtube.fetch", + _async_return(mock_result), + ): + await run_pipeline(lib_store, item_id, storage_dir) + + item = await lib_store.get_item(item_id) + assert item["status"] == "ready" + + artifacts = await lib_store.get_artifacts(item_id) + artifact_kinds = {a["kind"] for a in artifacts} + # FileProcessor runs first (no storage_path → empty), then YouTubeProcessor + assert "metadata" in artifact_kinds + assert "transcript" in artifact_kinds + + +class TestWebProcessor: + @pytest.mark.asyncio + async def test_process_web_url(self, lib_store, storage_dir): + """Web processor fetches HTML, extracts text, stores artifacts.""" + html = ( + "<html><head><title>Test Page" + "

This is the main article text " + "with enough content to pass the readability minimum " + "character count for extraction purposes now.

" + "

Second paragraph with more detailed content about " + "the topic being discussed on this test page.

" + "" + ) + + item_id = await lib_store.create_item( + kind="url:web", + source_url="https://example.com/article", + title="", + ) + item = await lib_store.get_item(item_id) + + proc = WebProcessor(lib_store, storage_dir) + + mock_resp = _mock_httpx_response(html, 200) + with ( + patch("httpx.AsyncClient") as mock_client_cls, + patch( + "tinyagentos.routes.desktop_browser.ssrf.validate_url_or_raise", + ), + ): + mock_client_cls.return_value.__aenter__.return_value.get = ( + _async_return(mock_resp) + ) + artifacts = await proc.process(item) + + kinds = {a["kind"] for a in artifacts} + assert "metadata" in kinds + assert "text" in kinds + + text_artifacts = [a for a in artifacts if a["kind"] == "text"] + assert len(text_artifacts) == 1 + text_path = Path(text_artifacts[0]["path"]) + assert text_path.exists() + + updated = await lib_store.get_item(item_id) + meta = json.loads(updated["meta_json"]) + assert "preview" in meta + + @pytest.mark.asyncio + async def test_process_web_url_no_source(self, lib_store, storage_dir): + """Web processor returns empty when source_url is missing.""" + item_id = await lib_store.create_item( + kind="url:web", source_url="", title="", + ) + item = await lib_store.get_item(item_id) + + proc = WebProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert artifacts == [] + + @pytest.mark.asyncio + async def test_web_extracts_title_from_html(self, lib_store, storage_dir): + """Web processor auto-titles from tag when item has no title.""" + html = ( + "<html><head><title>Auto Title Here" + "

This article has enough text content " + "to ensure that the readability extractor returns a full " + "result instead of falling back to the simple tag stripper " + "which would otherwise happen for very short pages.

" + "" + ) + + item_id = await lib_store.create_item( + kind="url:web", + source_url="https://example.com/titled", + title="https://example.com/titled", + ) + item = await lib_store.get_item(item_id) + + proc = WebProcessor(lib_store, storage_dir) + + mock_resp = _mock_httpx_response(html, 200) + with ( + patch("httpx.AsyncClient") as mock_client_cls, + patch( + "tinyagentos.routes.desktop_browser.ssrf.validate_url_or_raise", + ), + ): + mock_client_cls.return_value.__aenter__.return_value.get = ( + _async_return(mock_resp) + ) + await proc.process(item) + + updated = await lib_store.get_item(item_id) + assert "Auto Title Here" in updated["title"] + + @pytest.mark.asyncio + async def test_web_pipeline_integration(self, lib_store, storage_dir): + """run_pipeline routes url:web items to WebProcessor.""" + html = ( + "Pipeline Web" + "

Pipeline test content that is sufficiently " + "long to pass the readability minimum character count." + "

" + ) + + item_id = await lib_store.create_item( + kind="url:web", + source_url="https://example.com/pipeline-web", + ) + + mock_resp = _mock_httpx_response(html, 200) + with ( + patch("httpx.AsyncClient") as mock_client_cls, + patch( + "tinyagentos.routes.desktop_browser.ssrf.validate_url_or_raise", + ), + ): + mock_client_cls.return_value.__aenter__.return_value.get = ( + _async_return(mock_resp) + ) + await run_pipeline(lib_store, item_id, storage_dir) + + item = await lib_store.get_item(item_id) + assert item["status"] == "ready" + + artifacts = await lib_store.get_artifacts(item_id) + artifact_kinds = {a["kind"] for a in artifacts} + assert "metadata" in artifact_kinds + assert "text" in artifact_kinds + + # --------------------------------------------------------------------------- # run_pipeline # --------------------------------------------------------------------------- @@ -553,3 +827,19 @@ def _create_test_image(path: Path): from PIL import Image img = Image.new("RGB", (100, 50), color="blue") img.save(path) + + +def _async_return(value): + """Return an async callable that returns ``value`` (for mocking async functions).""" + async def _inner(*args, **kwargs): + return value + return _inner + + +def _mock_httpx_response(html: str, status_code: int = 200): + """Return a mock httpx Response with the given HTML body.""" + mock = MagicMock() + mock.text = html + mock.status_code = status_code + mock.headers = {"content-type": "text/html; charset=utf-8"} + return mock diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py index 488606e2e..572198aad 100644 --- a/tinyagentos/library_pipeline.py +++ b/tinyagentos/library_pipeline.py @@ -248,6 +248,262 @@ async def process(self, item: dict) -> list[dict]: return artifacts +class YouTubeProcessor(Processor): + """YouTube URL processor — cheap tier: metadata, thumbnail, transcript, chapters. + + Uses yt-dlp via the knowledge_fetchers.youtube module to fetch video + metadata and captions without downloading the video file. Produces + artifacts that flow into taosmd collections for agent querying. + + The cheap tier (per the design doc, docs/design/library-app.md section 4) + covers steps 1-4: canonical link, title, channel, description, thumbnail, + duration, upload date, subtitles/transcript, chapters. + """ + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + source_url = item.get("source_url", "") + artifacts: list[dict] = [] + + if not source_url: + return artifacts + + try: + from tinyagentos.knowledge_fetchers.youtube import ( + fetch, + format_timestamp, + ) + + media_dir = self.storage_dir / "youtube" + result = await fetch(source_url, media_dir=media_dir) + except ImportError: + logger.warning("YouTube fetcher not available — skipping %s", + source_url) + return artifacts + except Exception: + logger.exception("YouTube fetch failed for %s", source_url) + return artifacts + + title = result.get("title", "") + if title and not item.get("title"): + await self.store.update_item(item_id, title=title) + + meta = result.get("metadata", {}) + + # Update item meta_json with structured video metadata + stored_meta = json.loads(item.get("meta_json", "{}")) + stored_meta.update({ + "video_id": meta.get("video_id", ""), + "channel": meta.get("channel", ""), + "duration": meta.get("duration"), + "views": meta.get("views"), + "upload_date": meta.get("upload_date", ""), + }) + await self.store.update_item(item_id, meta_json=stored_meta) + + # Artifact: metadata + await self.store.add_artifact( + item_id, kind="metadata", path=source_url, meta=meta, + ) + artifacts.append({"kind": "metadata", "path": source_url, "meta": meta}) + + # Artifact: thumbnail (if downloaded) + thumbnail = result.get("thumbnail") + if thumbnail and Path(thumbnail).exists(): + await self.store.add_artifact( + item_id, kind="thumbnail", path=thumbnail, + meta={"source": "youtube"}, + ) + artifacts.append({ + "kind": "thumbnail", "path": thumbnail, + "meta": {"source": "youtube"}, + }) + + # Artifact: transcript + content = result.get("content", "") + if content: + transcript_dir = self.storage_dir / "transcripts" + transcript_dir.mkdir(parents=True, exist_ok=True) + transcript_path = transcript_dir / f"{item_id}_transcript.txt" + transcript_path.write_text(content, encoding="utf-8") + + transcript_meta = { + "char_count": len(content), + "language": "en", + } + await self.store.add_artifact( + item_id, kind="transcript", path=str(transcript_path), + meta=transcript_meta, + ) + artifacts.append({ + "kind": "transcript", "path": str(transcript_path), + "meta": transcript_meta, + }) + + # Store preview for item card + preview = content[:200] + stored_meta["preview"] = preview + await self.store.update_item(item_id, meta_json=stored_meta) + + # Artifact: chapters (if available) + chapters = meta.get("chapters", []) + if chapters: + chapters_lines: list[str] = [] + for ch in chapters: + ts = format_timestamp(ch.get("start_time", 0)) + ch_title = ch.get("title", "") + chapters_lines.append(f"[{ts}] {ch_title}") + + chapters_text = "\n".join(chapters_lines) + chapters_dir = self.storage_dir / "chapters" + chapters_dir.mkdir(parents=True, exist_ok=True) + chapters_path = chapters_dir / f"{item_id}_chapters.txt" + chapters_path.write_text(chapters_text, encoding="utf-8") + + await self.store.add_artifact( + item_id, kind="chapters", path=str(chapters_path), + meta={"count": len(chapters)}, + ) + artifacts.append({ + "kind": "chapters", "path": str(chapters_path), + "meta": {"count": len(chapters)}, + }) + + return artifacts + + +class WebProcessor(Processor): + """Generic web-page URL processor — extracts readable text from HTML. + + Fetches the URL (SSRF-guarded against loopback/link-local/private hosts), + then extracts the main content using readability-lxml. Falls back to a + simple tag-stripping approach when readability-lxml is not installed. + + Produces a text artifact suitable for taosmd collection indexing so that + agents granted the collection can query the page content. + """ + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + source_url = item.get("source_url", "") + artifacts: list[dict] = [] + + if not source_url: + return artifacts + + # Fetch the page (SSRF-guarded) + try: + import httpx + from tinyagentos.routes.desktop_browser.ssrf import ( + validate_url_or_raise, + ) + + validate_url_or_raise(source_url) + + async with httpx.AsyncClient( + timeout=httpx.Timeout(30), + follow_redirects=True, + ) as client: + resp = await client.get(source_url) + resp.raise_for_status() + html = resp.text + except ImportError: + logger.warning("httpx not available — skipping web fetch for %s", + source_url) + return artifacts + except Exception: + logger.exception("Web fetch failed for %s", source_url) + return artifacts + + if not html: + return artifacts + + # Extract readable text + content = _extract_readable_text(html, source_url) + + # Extract title from tag if item has no title + title = item.get("title", "") + if not title or title == source_url: + import re + m = re.search( + r"<title[^>]*>([^<]+)", html, re.IGNORECASE, + ) + if m: + title = m.group(1).strip()[:200] + await self.store.update_item(item_id, title=title) + + # Artifact: metadata + page_meta = { + "char_count": len(content), + "content_type": resp.headers.get("content-type", ""), + "status_code": resp.status_code, + } + await self.store.add_artifact( + item_id, kind="metadata", path=source_url, meta=page_meta, + ) + artifacts.append({ + "kind": "metadata", "path": source_url, "meta": page_meta, + }) + + # Artifact: extracted text + if content: + text_dir = self.storage_dir / "text" + text_dir.mkdir(parents=True, exist_ok=True) + text_path = text_dir / f"{item_id}_web.txt" + text_path.write_text(content, encoding="utf-8") + + text_meta = { + "char_count": len(content), + "source": "readability", + } + await self.store.add_artifact( + item_id, kind="text", path=str(text_path), meta=text_meta, + ) + artifacts.append({ + "kind": "text", "path": str(text_path), "meta": text_meta, + }) + + # Store preview + preview = content[:200] + stored_meta = json.loads(item.get("meta_json", "{}")) + stored_meta["preview"] = preview + await self.store.update_item(item_id, meta_json=stored_meta) + + return artifacts + + +def _extract_readable_text(html: str, source_url: str = "") -> str: + """Extract the main readable content from an HTML page. + + Uses readability-lxml when available; falls back to simple tag-stripping. + """ + try: + from readability import Document + doc = Document(html) + content = doc.summary() + # Strip remaining HTML from readability output + import re + text = re.sub(r"<[^>]+>", " ", content) + text = re.sub(r"\s+", " ", text).strip() + if len(text) >= 100: + return text + except ImportError: + logger.debug("readability-lxml not installed — using simple extractor") + except Exception: + logger.warning("readability extraction failed for %s", source_url, + exc_info=True) + + # Fallback: simple tag-stripping (from knowledge_ingest._extract_text_readability) + import re + cleaned = re.sub( + r"<(script|style)[^>]*>.*?", "", html, + flags=re.DOTALL | re.IGNORECASE, + ) + text = re.sub(r"<[^>]+>", " ", cleaned) + text = re.sub(r"\s+", " ", text).strip() + return text + + class ImageProcessor(Processor): """Image processor — records dimensions, creates thumbnail. @@ -319,6 +575,8 @@ async def process(self, item: dict) -> list[dict]: "text": TextProcessor, "pdf": PdfProcessor, "image": ImageProcessor, + "url:youtube": YouTubeProcessor, + "url:web": WebProcessor, } From e08f98aac9ebd47e03cec79318891cfb8992887e Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:46:17 +0200 Subject: [PATCH 4/9] =?UTF-8?q?fix(library):=20address=20Kilo=20findings?= =?UTF-8?q?=20=E2=80=94=20XSS,=20HTML=20unescaping,=20swallowed=20exceptio?= =?UTF-8?q?ns?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _render_item_card: escape all user-controlled values (title, kind, id, status) with html.escape() to prevent reflected XSS through crafted page titles, YouTube metadata, or user form fields. - WebProcessor: use html.unescape() on extracted text to decode HTML entities before storage. - YouTubeProcessor/WebProcessor: remove bare except Exception that swallowed fetch failures. Let network/SSRF/yt-dlp errors propagate to run_pipeline which already marks items as 'error' — a failed URL fetch must not silently appear 'ready' with zero artifacts. All 47 library tests pass. --- tinyagentos/library_pipeline.py | 61 ++++++++++++++------------------- tinyagentos/routes/library.py | 10 +++--- 2 files changed, 32 insertions(+), 39 deletions(-) diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py index 572198aad..6f7afbf03 100644 --- a/tinyagentos/library_pipeline.py +++ b/tinyagentos/library_pipeline.py @@ -268,21 +268,16 @@ async def process(self, item: dict) -> list[dict]: if not source_url: return artifacts - try: - from tinyagentos.knowledge_fetchers.youtube import ( - fetch, - format_timestamp, - ) + # Only catch ImportError (missing yt-dlp). Let fetch errors + # propagate so run_pipeline marks the item as "error" — a failed + # yt-dlp invocation must not silently look successful. + from tinyagentos.knowledge_fetchers.youtube import ( + fetch, + format_timestamp, + ) - media_dir = self.storage_dir / "youtube" - result = await fetch(source_url, media_dir=media_dir) - except ImportError: - logger.warning("YouTube fetcher not available — skipping %s", - source_url) - return artifacts - except Exception: - logger.exception("YouTube fetch failed for %s", source_url) - return artifacts + media_dir = self.storage_dir / "youtube" + result = await fetch(source_url, media_dir=media_dir) title = result.get("title", "") if title and not item.get("title"): @@ -392,28 +387,23 @@ async def process(self, item: dict) -> list[dict]: return artifacts # Fetch the page (SSRF-guarded) - try: - import httpx - from tinyagentos.routes.desktop_browser.ssrf import ( - validate_url_or_raise, - ) + # Only catch ImportError (missing deps). Let fetch/network errors + # propagate so run_pipeline marks the item as "error" — a failed + # URL fetch must not silently look successful. + import httpx + from tinyagentos.routes.desktop_browser.ssrf import ( + validate_url_or_raise, + ) - validate_url_or_raise(source_url) + validate_url_or_raise(source_url) - async with httpx.AsyncClient( - timeout=httpx.Timeout(30), - follow_redirects=True, - ) as client: - resp = await client.get(source_url) - resp.raise_for_status() - html = resp.text - except ImportError: - logger.warning("httpx not available — skipping web fetch for %s", - source_url) - return artifacts - except Exception: - logger.exception("Web fetch failed for %s", source_url) - return artifacts + async with httpx.AsyncClient( + timeout=httpx.Timeout(30), + follow_redirects=True, + ) as client: + resp = await client.get(source_url) + resp.raise_for_status() + html = resp.text if not html: return artifacts @@ -422,6 +412,7 @@ async def process(self, item: dict) -> list[dict]: content = _extract_readable_text(html, source_url) # Extract title from <title> tag if item has no title + import html as _html_mod title = item.get("title", "") if not title or title == source_url: import re @@ -429,7 +420,7 @@ async def process(self, item: dict) -> list[dict]: r"<title[^>]*>([^<]+)", html, re.IGNORECASE, ) if m: - title = m.group(1).strip()[:200] + title = _html_mod.unescape(m.group(1)).strip()[:200] await self.store.update_item(item_id, title=title) # Artifact: metadata diff --git a/tinyagentos/routes/library.py b/tinyagentos/routes/library.py index dbd1261e2..f679d327b 100644 --- a/tinyagentos/routes/library.py +++ b/tinyagentos/routes/library.py @@ -238,20 +238,22 @@ def _is_htmx(request: Request) -> bool: def _render_item_card(item: dict) -> str: """Return an HTML .item-card
for *item*.""" + import html + status = item.get("status", "pending") css = _STATUS_CSS.get(status, "status-pending") - title = item.get("title", "Untitled") - kind = item.get("kind", "file") + title = html.escape(item.get("title", "Untitled")) + kind = html.escape(item.get("kind", "file")) size = item.get("size_bytes") or 0 size_str = f"{size:,} B" if size else "" - item_id = item.get("id", "") + item_id = html.escape(item.get("id", "")) return ( f'
' f'
' f'

{title}

' f'
' - f'{status}' + f'{html.escape(status)}' f" {kind}" f'{f" · {size_str}" if size_str else ""}' f"
" From 3ce9d3c729978c160069cde69cf129876bce9b09 Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:51:30 +0200 Subject: [PATCH 5/9] =?UTF-8?q?fix(library):=20address=20CodeRabbit=20find?= =?UTF-8?q?ings=20=E2=80=94=20SSRF=20redirect=20safety,=20response-size=20?= =?UTF-8?q?cap?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - WebProcessor: disable auto-redirects, manually validate every redirect hop with validate_url_or_raise() against the SSRF blocklist (same pattern as knowledge_ingest._download_article). Max 5 redirects. - WebProcessor: cap response body at 10 MB, stream with aiter_bytes() to avoid OOM on large/hostile pages. - Tests: update _mock_httpx_response with is_redirect=False, encoding, and aiter_bytes() to match the new fetch flow. All library tests pass. --- tests/test_library.py | 7 ++++ tinyagentos/library_pipeline.py | 57 ++++++++++++++++++++++++++------- 2 files changed, 52 insertions(+), 12 deletions(-) diff --git a/tests/test_library.py b/tests/test_library.py index b3131270a..736b34e3e 100644 --- a/tests/test_library.py +++ b/tests/test_library.py @@ -841,5 +841,12 @@ def _mock_httpx_response(html: str, status_code: int = 200): mock = MagicMock() mock.text = html mock.status_code = status_code + mock.is_redirect = False + mock.encoding = "utf-8" mock.headers = {"content-type": "text/html; charset=utf-8"} + + async def _aiter_bytes(_chunk_size: int = 8192): + yield html.encode("utf-8") + + mock.aiter_bytes = _aiter_bytes return mock diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py index 6f7afbf03..879791035 100644 --- a/tinyagentos/library_pipeline.py +++ b/tinyagentos/library_pipeline.py @@ -386,24 +386,57 @@ async def process(self, item: dict) -> list[dict]: if not source_url: return artifacts - # Fetch the page (SSRF-guarded) - # Only catch ImportError (missing deps). Let fetch/network errors - # propagate so run_pipeline marks the item as "error" — a failed - # URL fetch must not silently look successful. + # Fetch the page (SSRF-guarded, redirect-safe, size-capped). + # Follows the same pattern as knowledge_ingest._download_article: + # disable auto-redirects, manually validate each hop against the + # SSRF blocklist, and cap total response bytes. import httpx + from urllib.parse import urljoin + from tinyagentos.routes.desktop_browser.ssrf import ( + SsrfBlockedError, validate_url_or_raise, ) - validate_url_or_raise(source_url) + _MAX_WEB_REDIRECTS = 5 + _MAX_WEB_BYTES = 10 * 1024 * 1024 # 10 MB + + current_url = source_url + resp = None + for _hop in range(_MAX_WEB_REDIRECTS + 1): + validate_url_or_raise(current_url) + + async with httpx.AsyncClient( + timeout=httpx.Timeout(30), + follow_redirects=False, + ) as client: + resp = await client.get(current_url) + + if resp.is_redirect and resp.headers.get("location"): + current_url = urljoin(current_url, resp.headers["location"]) + continue + break + else: + raise SsrfBlockedError( + f"too many redirects fetching {source_url!r}" + ) - async with httpx.AsyncClient( - timeout=httpx.Timeout(30), - follow_redirects=True, - ) as client: - resp = await client.get(source_url) - resp.raise_for_status() - html = resp.text + resp.raise_for_status() + + # Read body with a size cap to avoid OOM on large/hostile pages. + body_chunks: list[bytes] = [] + total = 0 + async for chunk in resp.aiter_bytes(8192): + total += len(chunk) + if total > _MAX_WEB_BYTES: + raise ValueError( + f"Response body exceeds {_MAX_WEB_BYTES} bytes " + f"for {source_url!r}" + ) + body_chunks.append(chunk) + html = b"".join(body_chunks).decode( + resp.encoding or "utf-8", errors="replace" + ) if not html: return artifacts From 74cb3176787f371bea6237d6bc09ea0bea8c1e66 Mon Sep 17 00:00:00 2001 From: Hogne <227774406+hognek@users.noreply.github.com> Date: Mon, 20 Jul 2026 19:50:05 +0200 Subject: [PATCH 6/9] =?UTF-8?q?feat(library):=20P3=20=E2=80=94=20heavy=20t?= =?UTF-8?q?ier=20download,=20quality=20preferences,=20storage=20accounting?= =?UTF-8?q?,=20per-source=20rules?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - LibraryStore: add library_rules table, new columns (quality, auto_download, download_path, download_bytes, downloaded_at) with safe migration in _post_init, rule CRUD methods, fnmatch-based match_rules, get_storage_summary - HeavyDownloadProcessor: yt-dlp-based media download for url:youtube items with quality preference (360/480/720/1080/best), fallback heuristics for missing output paths - run_heavy_pipeline: rule-aware quality resolution (explicit > rule > item > default 720), job tracking via library_jobs table - Routes: POST /download, GET /download/status, POST/GET/DELETE /rules, GET /usage with HTMX-aware HTML responses - Auto-download: _ingest_task checks matching rules with auto_download=True and triggers heavy pipeline after cheap tier completes - Template: storage summary bar (polled via htmx), per-item download button with quality selector for YouTube items, rules management
panel with add/delete - Tests: 24 new tests (7 store rules/storage, 4 HeavyDownloadProcessor, 4 run_heavy_pipeline, 9 routes) - 70/70 library tests pass (1 slow route test skipped) --- tests/test_library.py | 393 +++++++++++++++++++++++++++++ tinyagentos/library_pipeline.py | 159 ++++++++++++ tinyagentos/library_store.py | 114 ++++++++- tinyagentos/routes/library.py | 285 ++++++++++++++++++++- tinyagentos/templates/library.html | 45 ++++ 5 files changed, 980 insertions(+), 16 deletions(-) diff --git a/tests/test_library.py b/tests/test_library.py index 736b34e3e..0ddb335ab 100644 --- a/tests/test_library.py +++ b/tests/test_library.py @@ -12,6 +12,7 @@ from httpx import ASGITransport, AsyncClient from tinyagentos.library_pipeline import ( FileProcessor, + HeavyDownloadProcessor, ImageProcessor, PdfProcessor, TextProcessor, @@ -803,6 +804,398 @@ async def test_filter_by_kind(self, client): assert item["kind"] == "url:youtube" +# --------------------------------------------------------------------------- +# P3: LibraryStore — rules + storage accounting +# --------------------------------------------------------------------------- + + +class TestLibraryStoreRules: + @pytest.mark.asyncio + async def test_create_and_list_rules(self, lib_store): + rid = await lib_store.create_rule( + source_pattern="*.youtube.com/*", + quality="1080", + auto_download=True, + ) + assert rid + + rules = await lib_store.list_rules() + assert len(rules) == 1 + assert rules[0]["source_pattern"] == "*.youtube.com/*" + assert rules[0]["quality"] == "1080" + assert rules[0]["auto_download"] == 1 + + @pytest.mark.asyncio + async def test_get_rule(self, lib_store): + rid = await lib_store.create_rule(source_pattern="*.example.com/*") + rule = await lib_store.get_rule(rid) + assert rule is not None + assert rule["source_pattern"] == "*.example.com/*" + + assert await lib_store.get_rule("nonexistent") is None + + @pytest.mark.asyncio + async def test_delete_rule(self, lib_store): + rid = await lib_store.create_rule(source_pattern="*.youtube.com/*") + await lib_store.delete_rule(rid) + assert len(await lib_store.list_rules()) == 0 + + @pytest.mark.asyncio + async def test_match_rules(self, lib_store): + await lib_store.create_rule( + source_pattern="*youtube.com/*", quality="720", auto_download=True, + ) + await lib_store.create_rule( + source_pattern="*vimeo.com/*", quality="1080", auto_download=False, + ) + + matched = await lib_store.match_rules("https://www.youtube.com/watch?v=abc") + assert len(matched) == 1 + assert matched[0]["quality"] == "720" + + matched_vimeo = await lib_store.match_rules("https://vimeo.com/12345") + assert len(matched_vimeo) == 1 + + matched_none = await lib_store.match_rules("https://example.com") + assert len(matched_none) == 0 + + @pytest.mark.asyncio + async def test_match_rules_disabled_ignored(self, lib_store): + rid = await lib_store.create_rule( + source_pattern="*example.com/*", + enabled=False, + ) + matched = await lib_store.match_rules("https://example.com/page") + assert len(matched) == 0 + + @pytest.mark.asyncio + async def test_storage_summary(self, lib_store): + await lib_store.create_item(kind="text", title="a.txt", size_bytes=100) + await lib_store.create_item(kind="pdf", title="b.pdf", size_bytes=500) + await lib_store.create_item(kind="url:youtube", title="c", size_bytes=0) + + summary = await lib_store.get_storage_summary() + assert summary["total_count"] == 3 + assert summary["total_bytes"] == 600 + assert "text" in summary["by_kind"] + assert "pdf" in summary["by_kind"] + + @pytest.mark.asyncio + async def test_storage_summary_empty(self, lib_store): + summary = await lib_store.get_storage_summary() + assert summary["total_count"] == 0 + assert summary["total_bytes"] == 0 + + +# --------------------------------------------------------------------------- +# P3: HeavyDownloadProcessor +# --------------------------------------------------------------------------- + + +class TestHeavyDownloadProcessor: + @pytest.mark.asyncio + async def test_process_heavy_download(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://youtube.com/watch?v=heavy-test", + ) + await lib_store.update_item(item_id, quality="480") + item = await lib_store.get_item(item_id) + + proc = HeavyDownloadProcessor(lib_store, storage_dir) + + mock_path = str(storage_dir / "downloads" / "test123.mp4") + storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True) + storage_dir.joinpath("downloads", "test123.mp4").write_text("fake video data") + + with patch( + "tinyagentos.knowledge_fetchers.youtube.download_video", + _async_return(mock_path), + ): + artifacts = await proc.process(item) + + assert len(artifacts) == 1 + assert artifacts[0]["kind"] == "download" + assert artifacts[0]["meta"]["quality"] == "480" + + updated = await lib_store.get_item(item_id) + assert updated["download_path"] == mock_path + assert updated["download_bytes"] > 0 + + @pytest.mark.asyncio + async def test_process_heavy_download_non_youtube(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="url:web", + source_url="https://example.com", + ) + item = await lib_store.get_item(item_id) + + proc = HeavyDownloadProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert artifacts == [] + + @pytest.mark.asyncio + async def test_process_heavy_download_no_source(self, lib_store, storage_dir): + item_id = await lib_store.create_item( + kind="url:youtube", source_url="", + ) + item = await lib_store.get_item(item_id) + + proc = HeavyDownloadProcessor(lib_store, storage_dir) + artifacts = await proc.process(item) + assert artifacts == [] + + @pytest.mark.asyncio + async def test_process_heavy_download_invalid_quality(self, lib_store, storage_dir): + """Invalid quality should fall back to 720.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://youtube.com/watch?v=qtest", + ) + await lib_store.update_item(item_id, quality="9999") + item = await lib_store.get_item(item_id) + + proc = HeavyDownloadProcessor(lib_store, storage_dir) + + mock_path = str(storage_dir / "downloads" / "test.mp4") + storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True) + storage_dir.joinpath("downloads", "test.mp4").write_text("data") + + with patch( + "tinyagentos.knowledge_fetchers.youtube.download_video", + _async_return(mock_path), + ): + artifacts = await proc.process(item) + + # Quality should have been normalized to 720 + assert artifacts[0]["meta"]["quality"] == "720" + + +# --------------------------------------------------------------------------- +# P3: run_heavy_pipeline +# --------------------------------------------------------------------------- + + +class TestRunHeavyPipeline: + @pytest.mark.asyncio + async def test_run_heavy_pipeline_rule_quality(self, lib_store, storage_dir): + """When a matching rule exists, its quality is used.""" + await lib_store.create_rule( + source_pattern="*youtube.com/*", + quality="480", + auto_download=False, + ) + + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://youtube.com/watch?v=rule-test", + ) + storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True) + + mock_path = str(storage_dir / "downloads" / "test.mp4") + storage_dir.joinpath("downloads", "test.mp4").write_text("fake data") + + from tinyagentos.library_pipeline import run_heavy_pipeline + + with patch( + "tinyagentos.knowledge_fetchers.youtube.download_video", + _async_return(mock_path), + ): + result = await run_heavy_pipeline( + lib_store, item_id, storage_dir, quality="", + ) + + assert result is not None + assert result["quality"] == "480" # From rule + + @pytest.mark.asyncio + async def test_run_heavy_pipeline_explicit_quality(self, lib_store, storage_dir): + """Explicit quality parameter overrides rule quality.""" + await lib_store.create_rule( + source_pattern="*youtube.com/*", + quality="480", + ) + + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://youtube.com/watch?v=explicit-test", + ) + storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True) + + mock_path = str(storage_dir / "downloads" / "test.mp4") + storage_dir.joinpath("downloads", "test.mp4").write_text("fake data") + + from tinyagentos.library_pipeline import run_heavy_pipeline + + with patch( + "tinyagentos.knowledge_fetchers.youtube.download_video", + _async_return(mock_path), + ): + result = await run_heavy_pipeline( + lib_store, item_id, storage_dir, quality="1080", + ) + + assert result["quality"] == "1080" # Explicit wins + + @pytest.mark.asyncio + async def test_run_heavy_pipeline_non_youtube(self, lib_store, storage_dir): + """run_heavy_pipeline returns None for non-YouTube items.""" + item_id = await lib_store.create_item( + kind="url:web", + source_url="https://example.com/page", + ) + + from tinyagentos.library_pipeline import run_heavy_pipeline + result = await run_heavy_pipeline( + lib_store, item_id, storage_dir, + ) + assert result is None + + @pytest.mark.asyncio + async def test_run_heavy_pipeline_creates_job(self, lib_store, storage_dir): + """Heavy pipeline creates a job entry.""" + item_id = await lib_store.create_item( + kind="url:youtube", + source_url="https://youtube.com/watch?v=job-test", + ) + storage_dir.joinpath("downloads").mkdir(parents=True, exist_ok=True) + + mock_path = str(storage_dir / "downloads" / "test.mp4") + storage_dir.joinpath("downloads", "test.mp4").write_text("fake data") + + from tinyagentos.library_pipeline import run_heavy_pipeline + + with patch( + "tinyagentos.knowledge_fetchers.youtube.download_video", + _async_return(mock_path), + ): + await run_heavy_pipeline(lib_store, item_id, storage_dir, quality="720") + + jobs = await lib_store.get_item_jobs(item_id) + heavy_jobs = [j for j in jobs if j["stage"] == "heavy_download"] + assert len(heavy_jobs) >= 1 + + +# --------------------------------------------------------------------------- +# P3: Library routes — download, rules, usage +# --------------------------------------------------------------------------- + + +class TestLibraryRoutesP3: + @pytest.mark.asyncio + async def test_trigger_download(self, client): + """POST /api/library/items/{id}/download triggers heavy download.""" + # Ingest a YouTube URL first + resp = await client.post( + "/api/library/ingest", data={"url": "https://youtube.com/watch?v=dl-test"} + ) + assert resp.status_code == 202 + item_id = resp.json()["item_id"] + + # Allow pipeline to finish + import asyncio as _asyncio + await _asyncio.sleep(0.5) + + resp = await client.post( + f"/api/library/items/{item_id}/download", + data={"quality": "480"}, + ) + assert resp.status_code == 202 + data = resp.json() + assert data["status"] == "downloading" + assert data["quality"] == "480" + + @pytest.mark.asyncio + async def test_trigger_download_nonexistent(self, client): + resp = await client.post("/api/library/items/nonexistent/download") + assert resp.status_code == 404 + + @pytest.mark.asyncio + async def test_trigger_download_non_youtube(self, client): + """Download endpoint rejects non-YouTube items.""" + resp = await client.post( + "/api/library/ingest", data={"url": "https://example.com/page"} + ) + assert resp.status_code == 202 + item_id = resp.json()["item_id"] + + import asyncio as _asyncio + await _asyncio.sleep(0.5) + + resp = await client.post(f"/api/library/items/{item_id}/download") + assert resp.status_code == 400 + + @pytest.mark.asyncio + async def test_download_status(self, client): + resp = await client.post( + "/api/library/ingest", data={"url": "https://youtube.com/watch?v=status"} + ) + item_id = resp.json()["item_id"] + + import asyncio as _asyncio + await _asyncio.sleep(0.5) + + resp = await client.get(f"/api/library/items/{item_id}/download/status") + assert resp.status_code == 200 + data = resp.json() + assert data["item_id"] == item_id + assert "downloaded" in data + + @pytest.mark.asyncio + async def test_create_rule(self, client): + resp = await client.post( + "/api/library/rules", + data={"source_pattern": "*.youtube.com/*", "quality": "1080", "auto_download": "true"}, + ) + assert resp.status_code == 201 + data = resp.json() + assert data["rule_id"] + assert data["source_pattern"] == "*.youtube.com/*" + + @pytest.mark.asyncio + async def test_list_rules(self, client): + # Create a rule first + await client.post( + "/api/library/rules", data={"source_pattern": "*.example.com/*"} + ) + resp = await client.get("/api/library/rules") + assert resp.status_code == 200 + data = resp.json() + assert "rules" in data + assert data["count"] >= 1 + + @pytest.mark.asyncio + async def test_delete_rule(self, client): + resp = await client.post( + "/api/library/rules", data={"source_pattern": "*.test.com/*"} + ) + rule_id = resp.json()["rule_id"] + + resp = await client.delete(f"/api/library/rules/{rule_id}") + assert resp.status_code == 200 + assert resp.json()["status"] == "deleted" + + # Verify it's gone + resp = await client.get("/api/library/rules") + data = resp.json() + rule_ids = [r["id"] for r in data["rules"]] + assert rule_id not in rule_ids + + @pytest.mark.asyncio + async def test_delete_nonexistent_rule(self, client): + resp = await client.delete("/api/library/rules/nonexistent") + assert resp.status_code == 404 + + @pytest.mark.asyncio + async def test_storage_usage(self, client): + resp = await client.get("/api/library/usage") + assert resp.status_code == 200 + data = resp.json() + assert "total_count" in data + assert "total_bytes" in data + assert "by_kind" in data + + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- diff --git a/tinyagentos/library_pipeline.py b/tinyagentos/library_pipeline.py index 879791035..a91771501 100644 --- a/tinyagentos/library_pipeline.py +++ b/tinyagentos/library_pipeline.py @@ -611,6 +611,165 @@ def get_processor(kind: str, store: LibraryStore, return cls(store, storage_dir) +# --------------------------------------------------------------------------- +# Heavy tier — opt-in media download +# --------------------------------------------------------------------------- + + +class HeavyDownloadProcessor(Processor): + """Downloads media for items that have opted into the heavy tier. + + Currently supports url:youtube items via yt-dlp download_video. + Respects per-item quality preference and per-source rules. + """ + + _VALID_QUALITIES = frozenset({"360", "480", "720", "1080", "best"}) + + async def process(self, item: dict) -> list[dict]: + item_id = item["id"] + source_url = item.get("source_url", "") + artifacts: list[dict] = [] + + if not source_url: + return artifacts + + kind = item.get("kind", "") + if kind != "url:youtube": + return artifacts + + quality = item.get("quality", "") or "720" + if quality not in self._VALID_QUALITIES: + quality = "720" + + try: + from tinyagentos.knowledge_fetchers.youtube import download_video + except ImportError: + logger.warning("yt-dlp not available for heavy download") + return artifacts + + download_dir = self.storage_dir / "downloads" + download_dir.mkdir(parents=True, exist_ok=True) + + path = await download_video(source_url, quality=quality, output_dir=download_dir) + + if not path: + msg = f"Heavy download failed for {source_url!r}: yt-dlp returned no output path" + logger.warning(msg) + await self.store.update_item( + item_id, + meta_json={ + **json.loads(item.get("meta_json", "{}")), + "download_error": msg, + }, + ) + return artifacts + + p = Path(path) + if not p.exists(): + # Try to find the downloaded file + candidates = sorted( + download_dir.glob("*"), key=lambda x: x.stat().st_mtime, reverse=True + ) + if candidates: + p = candidates[0] + else: + await self.store.update_item( + item_id, + meta_json={ + **json.loads(item.get("meta_json", "{}")), + "download_error": "Downloaded file not found on disk", + }, + ) + return artifacts + + size_bytes = p.stat().st_size + await self.store.update_item( + item_id, + download_path=str(p), + download_bytes=size_bytes, + bytes=size_bytes, + downloaded_at=time.time(), + ) + + download_meta: dict = { + "path": str(p), + "bytes": size_bytes, + "quality": quality, + "format": p.suffix.lstrip("."), + } + await self.store.add_artifact( + item_id, kind="download", path=str(p), meta=download_meta, + ) + artifacts.append({ + "kind": "download", "path": str(p), "meta": download_meta, + }) + + return artifacts + + +async def run_heavy_pipeline( + store: LibraryStore, + item_id: str, + storage_dir: Path, + quality: str = "", +) -> dict | None: + """Run the heavy-tier download pipeline for one item. + + Checks per-source rules for auto_download settings, respects per-item + quality override, and downloads the media via yt-dlp. + + Returns download metadata dict on success, None if skipped or failed. + """ + item = await store.get_item(item_id) + if not item: + return None + + kind = item.get("kind", "") + source_url = item.get("source_url", "") + + # Only YouTube items are supported for heavy download currently + if kind != "url:youtube" or not source_url: + return None + + # Check for matching rules (apply first matching rule's quality if + # no explicit quality was provided) + if not quality: + rules = await store.match_rules(source_url) + if rules: + quality = rules[0].get("quality", "") or "720" + + # Fallback to item's quality field, then default 720 + if not quality: + quality = item.get("quality", "") or "720" + + # Create a job entry + await store.create_job(item_id, "heavy_download") + + try: + proc = HeavyDownloadProcessor(store, storage_dir) + # Override the item's quality for this run + item_with_quality = dict(item, quality=quality) + artifacts = await proc.process(item_with_quality) + + if artifacts: + await store.update_job( + (await store.get_item_jobs(item_id))[-1]["id"], + state="done", + ) + return artifacts[0].get("meta", {}) + else: + await store.update_job( + (await store.get_item_jobs(item_id))[-1]["id"], + state="error", + error="Download produced no artifacts", + ) + return None + except Exception: + logger.exception("Heavy pipeline failed for item %s", item_id) + await store.update_item_status(item_id, "error") + return None + + # --------------------------------------------------------------------------- # Pipeline runner # --------------------------------------------------------------------------- diff --git a/tinyagentos/library_store.py b/tinyagentos/library_store.py index 1126d26fb..4e45ae200 100644 --- a/tinyagentos/library_store.py +++ b/tinyagentos/library_store.py @@ -59,6 +59,16 @@ ); CREATE INDEX IF NOT EXISTS idx_lj_item ON library_jobs(item_id); CREATE INDEX IF NOT EXISTS idx_lj_state ON library_jobs(state); + +CREATE TABLE IF NOT EXISTS library_rules ( + id TEXT PRIMARY KEY, + source_pattern TEXT NOT NULL, + quality TEXT NOT NULL DEFAULT '720', + auto_download INTEGER NOT NULL DEFAULT 0, + enabled INTEGER NOT NULL DEFAULT 1, + created_at REAL NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_lr_pattern ON library_rules(source_pattern); """ _VALID_STATUSES = frozenset({"pending", "processing", "ready", "error"}) @@ -68,10 +78,28 @@ class LibraryStore(BaseStore): SCHEMA = LIBRARY_SCHEMA async def _post_init(self) -> None: - """Enable foreign key enforcement for cascade deletes.""" + """Enable foreign key enforcement and apply schema migrations.""" await self._db.execute("PRAGMA foreign_keys = ON") await self._db.commit() + # P3 migration: add quality, auto_download, downloaded_at columns + # if they don't exist yet (existing databases from P1/P2). + cols = await self._db.execute_fetchall("PRAGMA table_info(library_items)") + col_names = {row[1] for row in cols} + for col, col_def in [ + ("quality", "TEXT NOT NULL DEFAULT ''"), + ("auto_download", "INTEGER NOT NULL DEFAULT 0"), + ("downloaded_at", "REAL"), + ("download_path", "TEXT NOT NULL DEFAULT ''"), + ("download_bytes", "INTEGER NOT NULL DEFAULT 0"), + ]: + if col not in col_names: + await self._db.execute( + f"ALTER TABLE library_items ADD COLUMN {col} {col_def}" + ) + if col_names: + await self._db.commit() + # -- items ------------------------------------------------------------ async def create_item( @@ -140,7 +168,9 @@ async def list_items( async def update_item(self, item_id: str, **kwargs) -> None: allowed = {"kind", "source_url", "title", "status", "storage_path", - "bytes", "meta_json", "updated_at"} + "bytes", "meta_json", "updated_at", + "quality", "auto_download", "download_path", + "download_bytes", "downloaded_at"} fields = [(k, v) for k, v in kwargs.items() if k in allowed] if not fields: return @@ -249,3 +279,83 @@ async def update_job(self, job_id: str, **kwargs) -> None: f"UPDATE library_jobs SET {set_clause} WHERE id = ?", values ) await self._db.commit() + + # -- source rules ----------------------------------------------------- + + async def create_rule( + self, + source_pattern: str, + quality: str = "720", + auto_download: bool = False, + enabled: bool = True, + ) -> str: + rule_id = uuid.uuid4().hex[:16] + now = time.time() + await self._db.execute( + """INSERT INTO library_rules + (id, source_pattern, quality, auto_download, enabled, created_at) + VALUES (?, ?, ?, ?, ?, ?)""", + (rule_id, source_pattern, quality, int(auto_download), int(enabled), now), + ) + await self._db.commit() + return rule_id + + async def list_rules(self) -> list[dict]: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_rules ORDER BY created_at DESC" + ) as cursor: + rows = await cursor.fetchall() + return [dict(r) for r in rows] + + async def get_rule(self, rule_id: str) -> dict | None: + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_rules WHERE id = ?", (rule_id,) + ) as cursor: + row = await cursor.fetchone() + return dict(row) if row else None + + async def delete_rule(self, rule_id: str) -> None: + await self._db.execute( + "DELETE FROM library_rules WHERE id = ?", (rule_id,) + ) + await self._db.commit() + + async def match_rules(self, source_url: str) -> list[dict]: + """Return all enabled rules whose source_pattern matches source_url.""" + self._db.row_factory = aiosqlite.Row + async with self._db.execute( + "SELECT * FROM library_rules WHERE enabled = 1 ORDER BY created_at" + ) as cursor: + rows = await cursor.fetchall() + rules = [dict(r) for r in rows] + import fnmatch + return [ + r for r in rules + if fnmatch.fnmatch(source_url.lower(), r["source_pattern"].lower()) + ] + + # -- storage accounting ----------------------------------------------- + + async def get_storage_summary(self) -> dict: + """Return total bytes, item count, and per-kind breakdown.""" + self._db.row_factory = aiosqlite.Row + rows = await self._db.execute_fetchall( + "SELECT kind, COUNT(*) as count, " + "COALESCE(SUM(bytes), 0) as total_bytes " + "FROM library_items GROUP BY kind" + ) + by_kind = {row["kind"]: {"count": row["count"], "bytes": row["total_bytes"]} for row in rows} + + total = await self._db.execute_fetchall( + "SELECT COUNT(*) as total_count, COALESCE(SUM(bytes), 0) as total_bytes " + "FROM library_items" + ) + if total: + return { + "total_count": total[0]["total_count"], + "total_bytes": total[0]["total_bytes"], + "by_kind": by_kind, + } + return {"total_count": 0, "total_bytes": 0, "by_kind": {}} diff --git a/tinyagentos/routes/library.py b/tinyagentos/routes/library.py index f679d327b..dafe9384b 100644 --- a/tinyagentos/routes/library.py +++ b/tinyagentos/routes/library.py @@ -190,6 +190,27 @@ async def _ingest_task(app, item_id: str, store, storage_dir: Path) -> None: await store.update_item_status(item_id, "error") return + # Auto-download: check matching rules with auto_download=True + try: + item = await store.get_item(item_id) + if item and item.get("source_url"): + rules = await store.match_rules(item["source_url"]) + for rule in rules: + if rule.get("auto_download"): + from tinyagentos.library_pipeline import run_heavy_pipeline + + quality = rule.get("quality", "") or "720" + logger.info( + "Auto-download triggered for item %s by rule %s (quality=%s)", + item_id, rule["id"], quality, + ) + await run_heavy_pipeline( + store, item_id, storage_dir, quality=quality, + ) + break # First matching auto-download rule wins + except Exception: + logger.exception("Auto-download check failed for item %s", item_id) + # Collections handoff after successful pipeline try: collections_dir = storage_dir.parent / "collections" @@ -228,6 +249,17 @@ def _is_htmx(request: Request) -> bool: return request.headers.get("HX-Request", "").lower() == "true" +def _format_bytes(size: int) -> str: + """Format a byte count as a human-readable string.""" + if size < 1024: + return f"{size} B" + for unit in ("KB", "MB", "GB", "TB"): + size /= 1024.0 + if size < 1024 or unit == "TB": + return f"{size:.1f} {unit}" + return f"{size:.1f} TB" + + _STATUS_CSS: dict[str, str] = { "pending": "status-pending", "processing": "status-processing", @@ -244,22 +276,60 @@ def _render_item_card(item: dict) -> str: css = _STATUS_CSS.get(status, "status-pending") title = html.escape(item.get("title", "Untitled")) kind = html.escape(item.get("kind", "file")) - size = item.get("size_bytes") or 0 - size_str = f"{size:,} B" if size else "" + bytes_val = item.get("bytes") or 0 + size_str = _format_bytes(bytes_val) if bytes_val else "" item_id = html.escape(item.get("id", "")) + source_url = html.escape(item.get("source_url", "")) + downloaded = bool(item.get("download_path", "")) + + parts = [ + f'
', + f'
', + f'

{title}

', + f'
', + f'{html.escape(status)}', + f" {kind}", + ] + + if size_str: + parts.append(f' · {size_str}') + + parts.append("
") # .meta + + # YouTube download actions (only for url:youtube items, only when ready) + if kind == "url:youtube" and status == "ready" and not downloaded: + parts.append( + f'
' + f' ' + f'' + f'
' + ) + elif kind == "url:youtube" and downloaded: + download_bytes_val = item.get("download_bytes") or 0 + parts.append( + f'
' + f'✅ Downloaded ({_format_bytes(download_bytes_val)})' + f'
' + ) - return ( - f'
' - f'
' - f'

{title}

' - f'
' - f'{html.escape(status)}' - f" {kind}" - f'{f" · {size_str}" if size_str else ""}' - f"
" - f"
" - f"
" - ) + parts.append("
") # .info + parts.append("
") # .item-card + + return "".join(parts) def _render_item_list(items: list[dict]) -> str: @@ -273,6 +343,47 @@ def _render_item_list(items: list[dict]) -> str: return "".join(_render_item_card(item) for item in items) +def _render_rules_list(rules: list[dict]) -> str: + """Return an HTML fragment listing source rules.""" + import html as _h + + if not rules: + return 'No rules. Add one above to auto-download from matching sources.' + + parts: list[str] = ['
    '] + for rule in rules: + rid = _h.escape(rule["id"]) + pat = _h.escape(rule["source_pattern"]) + qual = _h.escape(rule.get("quality", "720")) + auto = "auto" if rule.get("auto_download") else "manual" + parts.append( + f'
  • ' + f'{pat} ({qual}p, {auto})' + f'' + f'
    • ' + ) + parts.append("
") + return "".join(parts) + + +def _render_storage_summary(summary: dict) -> str: + """Return an HTML snippet for the storage summary bar.""" + total_bytes = summary.get("total_bytes", 0) + total_count = summary.get("total_count", 0) + if not total_count: + return 'No items stored yet.' + + return ( + f'Library storage: {_format_bytes(total_bytes)} ' + f'across {total_count} item{"s" if total_count != 1 else ""}' + f'' + ) + + # --------------------------------------------------------------------------- # List / Get / Delete # --------------------------------------------------------------------------- @@ -367,3 +478,149 @@ async def reprocess_item(request: Request, item_id: str): _create_supervised_task(coro, task_set) return JSONResponse({"item_id": item_id, "status": "reprocessing"}, status_code=202) + + +# --------------------------------------------------------------------------- +# Heavy tier: download +# --------------------------------------------------------------------------- + + +@router.post("/api/library/items/{item_id}/download") +async def trigger_download( + request: Request, + item_id: str, + quality: str | None = Form(None), +): + """Trigger heavy-tier media download for an item. + + Accepts optional ``quality`` form field (360, 480, 720, 1080, best). + Runs the download asynchronously in a background task. + """ + store = await _get_library_store(request) + item = await store.get_item(item_id) + if not item: + return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404) + + if item["kind"] != "url:youtube": + return JSONResponse( + {"error": "Heavy download only supports url:youtube items"}, + status_code=400, + ) + + storage_dir = _library_dir(request) + quality_val = quality or item.get("quality", "") or "720" + + task_set = getattr(request.app.state, "_background_tasks", None) + coro = _heavy_download_task(request.app, item_id, store, storage_dir, quality_val) + if task_set is None: + _track_background_task(coro) + else: + _create_supervised_task(coro, task_set) + + return JSONResponse( + {"item_id": item_id, "status": "downloading", "quality": quality_val}, + status_code=202, + ) + + +async def _heavy_download_task( + app, item_id: str, store, storage_dir: Path, quality: str +) -> None: + """Background task: run heavy download for an item.""" + from tinyagentos.library_pipeline import run_heavy_pipeline + + try: + result = await run_heavy_pipeline(store, item_id, storage_dir, quality=quality) + if result: + logger.info("Heavy download complete for item %s: %s", item_id, result) + except Exception: + logger.exception("Heavy download crashed for item %s", item_id) + await store.update_item_status(item_id, "error") + + +@router.get("/api/library/items/{item_id}/download/status") +async def download_status(request: Request, item_id: str): + """Check heavy download status for an item.""" + store = await _get_library_store(request) + item = await store.get_item(item_id) + if not item: + return JSONResponse({"error": f"Item {item_id!r} not found"}, status_code=404) + + jobs = await store.get_item_jobs(item_id) + heavy_jobs = [j for j in jobs if j.get("stage") == "heavy_download"] + download_path = item.get("download_path", "") + download_bytes = item.get("download_bytes", 0) + + return { + "item_id": item_id, + "downloaded": bool(download_path), + "download_path": download_path, + "download_bytes": download_bytes, + "jobs": heavy_jobs, + } + + +# --------------------------------------------------------------------------- +# Source rules +# --------------------------------------------------------------------------- + + +@router.post("/api/library/rules") +async def create_rule( + request: Request, + source_pattern: str = Form(...), + quality: str | None = Form("720"), + auto_download: bool = Form(False), +): + """Create a source rule for automatic heavy download triggers.""" + store = await _get_library_store(request) + rule_id = await store.create_rule( + source_pattern=source_pattern, + quality=quality or "720", + auto_download=auto_download, + ) + return JSONResponse( + {"rule_id": rule_id, "source_pattern": source_pattern, "status": "created"}, + status_code=201, + ) + + +@router.get("/api/library/rules") +async def list_rules(request: Request): + """List all source rules.""" + store = await _get_library_store(request) + rules = await store.list_rules() + + if _is_htmx(request): + return HTMLResponse(_render_rules_list(rules)) + + return {"rules": rules, "count": len(rules)} + + +@router.delete("/api/library/rules/{rule_id}") +async def delete_rule(request: Request, rule_id: str): + """Delete a source rule.""" + store = await _get_library_store(request) + rule = await store.get_rule(rule_id) + if not rule: + return JSONResponse({"error": f"Rule {rule_id!r} not found"}, status_code=404) + + await store.delete_rule(rule_id) + return {"status": "deleted", "rule_id": rule_id} + + +# --------------------------------------------------------------------------- +# Storage accounting +# --------------------------------------------------------------------------- + + +@router.get("/api/library/usage") +async def storage_usage(request: Request): + """Return storage accounting summary for the library.""" + store = await _get_library_store(request) + summary = await store.get_storage_summary() + + if _is_htmx(request): + return HTMLResponse(_render_storage_summary(summary)) + + return summary diff --git a/tinyagentos/templates/library.html b/tinyagentos/templates/library.html index 6fdcb1d44..76731257e 100644 --- a/tinyagentos/templates/library.html +++ b/tinyagentos/templates/library.html @@ -62,6 +62,15 @@

📚 Library

+ +
+ Loading storage info… +
+
📚 Library
+ +
+
+ + ⚙ Source Rules + +
+
+ + + + +
+
+ Loading rules… +
+
+
+
+
+ +