Gan, Jimmy
a91a1132c9
Deploy Dashboard (Dev) / deploy-dev (push) Successful in 2m10s
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>
46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
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
|