feat: add info engine MVP pipeline and dashboard page

Introduce a standalone scheduled info-engine worker with SQLite persistence and expose read-only dashboard APIs/UI so curated intelligence items can be collected and viewed with minimal architecture changes.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

info-engine/Dockerfile

@@ -0,0 +1,6 @@
+FROM python:3.12-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY . .
+CMD ["python", "-u", "main.py"]

info-engine/collector.py

@@ -0,0 +1,151 @@
+import logging
+import os
+from datetime import datetime
+from email.utils import parsedate_to_datetime
+from urllib.parse import urlparse
+import xml.etree.ElementTree as ET
+
+import httpx
+
+log = logging.getLogger(__name__)
+
+
+def _source_name(url: str) -> str:
+    host = urlparse(url).netloc
+    return host or url
+
+
+def _first_text(parent, paths: list[str]) -> str:
+    for path in paths:
+        node = parent.find(path)
+        if node is not None and node.text:
+            value = node.text.strip()
+            if value:
+                return value
+    return ""
+
+
+def _atom_link(entry) -> str:
+    for link in entry.findall("{http://www.w3.org/2005/Atom}link"):
+        href = (link.attrib.get("href") or "").strip()
+        rel = (link.attrib.get("rel") or "alternate").strip()
+        if href and rel == "alternate":
+            return href
+    links = entry.findall("{http://www.w3.org/2005/Atom}link")
+    if links:
+        return (links[0].attrib.get("href") or "").strip()
+    return ""
+
+
+def _parse_datetime(value: str | None) -> str | None:
+    if not value:
+        return None
+    value = value.strip()
+    if not value:
+        return None
+
+    try:
+        return parsedate_to_datetime(value).isoformat()
+    except Exception:
+        pass
+
+    try:
+        iso = value.replace("Z", "+00:00")
+        return datetime.fromisoformat(iso).isoformat()
+    except Exception:
+        return None
+
+
+def _parse_feed(feed_url: str, xml_text: str) -> list[dict]:
+    root = ET.fromstring(xml_text)
+    source = _source_name(feed_url)
+
+    items = []
+
+    channel = root.find("channel")
+    if channel is not None:
+        for item in channel.findall("item"):
+            title = _first_text(item, ["title"])
+            link = _first_text(item, ["link"])
+            summary = _first_text(item, ["description"])
+            pub = _first_text(item, ["pubDate", "date"])
+            tags = [n.text.strip() for n in item.findall("category") if n.text and n.text.strip()]
+            if title and link:
+                items.append(
+                    {
+                        "source": source,
+                        "title": title,
+                        "summary": summary,
+                        "url": link,
+                        "tags": tags,
+                        "published_at": _parse_datetime(pub),
+                        "content_text": summary,
+                        "content_html": summary,
+                    }
+                )
+        return items
+
+    for entry in root.findall("{http://www.w3.org/2005/Atom}entry"):
+        title = _first_text(entry, ["{http://www.w3.org/2005/Atom}title"])
+        link = _atom_link(entry)
+        summary = _first_text(
+            entry,
+            [
+                "{http://www.w3.org/2005/Atom}summary",
+                "{http://www.w3.org/2005/Atom}content",
+            ],
+        )
+        pub = _first_text(
+            entry,
+            [
+                "{http://www.w3.org/2005/Atom}published",
+                "{http://www.w3.org/2005/Atom}updated",
+            ],
+        )
+        tags = [
+            (cat.attrib.get("term") or "").strip()
+            for cat in entry.findall("{http://www.w3.org/2005/Atom}category")
+            if (cat.attrib.get("term") or "").strip()
+        ]
+        if title and link:
+            items.append(
+                {
+                    "source": source,
+                    "title": title,
+                    "summary": summary,
+                    "url": link,
+                    "tags": tags,
+                    "published_at": _parse_datetime(pub),
+                    "content_text": summary,
+                    "content_html": summary,
+                }
+            )
+
+    return items
+
+
+async def collect_sources() -> list[dict]:
+    sources_env = os.environ.get(
+        "INFO_ENGINE_SOURCES",
+        "https://hnrss.org/frontpage,https://lobste.rs/rss",
+    )
+    sources = [s.strip() for s in sources_env.split(",") if s.strip()]
+    if not sources:
+        log.warning("No INFO_ENGINE_SOURCES configured")
+        return []
+
+    timeout = httpx.Timeout(15.0)
+    all_items = []
+
+    async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
+        for source_url in sources:
+            try:
+                response = await client.get(source_url)
+                response.raise_for_status()
+                parsed = _parse_feed(source_url, response.text)
+                all_items.extend(parsed)
+                log.info("Collected %d items from %s", len(parsed), source_url)
+            except Exception as exc:
+                log.error("Failed to collect from %s: %s", source_url, exc)
+
+    return all_items

info-engine/db.py

@@ -0,0 +1,64 @@
+import json
+import os
+from datetime import datetime, timezone
+
+import aiosqlite
+
+DB_PATH = os.environ.get("DB_PATH", "/app/data/info_engine.db")
+
+
+async def init_db():
+    async with aiosqlite.connect(DB_PATH) as db:
+        await db.execute(
+            """
+            CREATE TABLE IF NOT EXISTS info_items (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                source TEXT NOT NULL,
+                title TEXT NOT NULL,
+                summary TEXT NOT NULL,
+                url TEXT NOT NULL UNIQUE,
+                tags TEXT NOT NULL,
+                published_at DATETIME,
+                collected_at DATETIME NOT NULL,
+                content_text TEXT,
+                content_html TEXT
+            )
+            """
+        )
+        await db.execute("CREATE INDEX IF NOT EXISTS idx_info_items_collected_at ON info_items(collected_at DESC)")
+        await db.execute("CREATE INDEX IF NOT EXISTS idx_info_items_published_at ON info_items(published_at DESC)")
+        await db.execute("CREATE INDEX IF NOT EXISTS idx_info_items_source ON info_items(source)")
+        await db.commit()
+
+
+async def upsert_items(items: list[dict]) -> int:
+    if not items:
+        return 0
+
+    inserted = 0
+    now = datetime.now(timezone.utc).isoformat()
+
+    async with aiosqlite.connect(DB_PATH) as db:
+        for item in items:
+            cursor = await db.execute(
+                """
+                INSERT OR IGNORE INTO info_items
+                (source, title, summary, url, tags, published_at, collected_at, content_text, content_html)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    item["source"],
+                    item["title"],
+                    item.get("summary", ""),
+                    item["url"],
+                    json.dumps(item.get("tags", []), ensure_ascii=False),
+                    item.get("published_at"),
+                    now,
+                    item.get("content_text"),
+                    item.get("content_html"),
+                ),
+            )
+            inserted += cursor.rowcount
+        await db.commit()
+
+    return inserted

info-engine/docker-compose.yml

@@ -0,0 +1,16 @@
+services:
+  info-engine:
+    build: .
+    container_name: info-engine
+    restart: unless-stopped
+    volumes:
+      - /volume1/docker/info-engine/data:/app/data
+    environment:
+      - DB_PATH=/app/data/info_engine.db
+      - INFO_ENGINE_INTERVAL_SECONDS=${INFO_ENGINE_INTERVAL_SECONDS:-900}
+      - INFO_ENGINE_SOURCES=${INFO_ENGINE_SOURCES:-https://hnrss.org/frontpage,https://lobste.rs/rss}
+    logging:
+      driver: json-file
+      options:
+        max-size: "10m"
+        max-file: "3"

info-engine/main.py

@@ -0,0 +1,37 @@
+import asyncio
+import logging
+import os
+
+from collector import collect_sources
+from db import init_db, upsert_items
+from processor import process_items
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
+log = logging.getLogger(__name__)
+
+COLLECT_INTERVAL_SECONDS = int(os.environ.get("INFO_ENGINE_INTERVAL_SECONDS", "900"))
+
+
+async def run_once() -> int:
+    raw_items = await collect_sources()
+    processed = process_items(raw_items)
+    inserted = await upsert_items(processed)
+    return inserted
+
+
+async def main():
+    await init_db()
+    log.info("Info engine started (interval=%ss)", COLLECT_INTERVAL_SECONDS)
+
+    while True:
+        try:
+            inserted = await run_once()
+            log.info("Info engine cycle complete (inserted=%d)", inserted)
+        except Exception as exc:
+            log.error("Info engine cycle failed: %s", exc)
+
+        await asyncio.sleep(COLLECT_INTERVAL_SECONDS)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

info-engine/processor.py

@@ -0,0 +1,45 @@
+from urllib.parse import urldefrag, urlparse
+
+
+def _normalize_url(url: str) -> str:
+    clean, _ = urldefrag(url.strip())
+    parsed = urlparse(clean)
+    if parsed.scheme and parsed.netloc:
+        return clean
+    return ""
+
+
+def process_items(items: list[dict]) -> list[dict]:
+    seen = set()
+    processed = []
+
+    for item in items:
+        title = (item.get("title") or "").strip()
+        url = _normalize_url(item.get("url") or "")
+        if not title or not url:
+            continue
+
+        dedupe_key = url.lower()
+        if dedupe_key in seen:
+            continue
+        seen.add(dedupe_key)
+
+        summary = (item.get("summary") or "").strip()
+        tags = item.get("tags") or []
+        if not isinstance(tags, list):
+            tags = []
+
+        processed.append(
+            {
+                "source": (item.get("source") or "unknown").strip() or "unknown",
+                "title": title,
+                "summary": summary,
+                "url": url,
+                "tags": [str(t).strip() for t in tags if str(t).strip()],
+                "published_at": item.get("published_at"),
+                "content_text": item.get("content_text") or summary,
+                "content_html": item.get("content_html") or summary,
+            }
+        )
+
+    return processed

info-engine/requirements.txt

@@ -0,0 +1,2 @@
+aiosqlite
+httpx