Return user-scoped wiki nodes and edges for the desktop graph view.
(user_id: str, query: str = "", limit: int = 24, daily_scope: str = "latest", daily_month: str = "")
| 3716 | for part in [ |
| 3717 | f"Daily Note topic: {entry.get('topic') or '未分类'}", |
| 3718 | f"Daily Note summary: {summary}" if summary else "", |
| 3719 | "Deep Reading report:", |
| 3720 | report_content or str(entry.get("body") or ""), |
| 3721 | ] |
| 3722 | if part |
| 3723 | ) |
| 3724 | key = str(entry.get("report_path") or title) |
| 3725 | return { |
| 3726 | "node_id": f"daily-reading:{hashlib.sha1(key.encode('utf-8')).hexdigest()[:16]}", |
| 3727 | "node_type": "paper", |
| 3728 | "title": title, |
| 3729 | "body": body, |
| 3730 | "keywords": " ".join([str(entry.get("topic") or ""), title, "daily note deep reading"]).strip(), |
| 3731 | "file_path": str(entry.get("report_path") or ""), |
| 3732 | "metadata": { |
| 3733 | "source": "daily_note_deep_reading", |
| 3734 | "daily_note": entry.get("daily_note"), |
| 3735 | "daily_note_topic": entry.get("topic"), |
| 3736 | "report_path": entry.get("report_path"), |
| 3737 | "pdf_url": entry.get("pdf_url") or "", |
| 3738 | }, |
| 3739 | "source_type": "reading_report", |
| 3740 | "source_ref": str(entry.get("report_path") or ""), |
| 3741 | "score": 1.0, |
| 3742 | "updated_at": None, |
| 3743 | } |
| 3744 | |
| 3745 | |
| 3746 | def daily_note_mentions(user_id: str, query: str = "", limit: int = 8) -> Dict[str, Any]: |
| 3747 | del user_id |
| 3748 | entries = _daily_note_reading_entries(scope="all", query=query, limit=limit) |
| 3749 | nodes = [] |
| 3750 | for entry in entries: |
| 3751 | node = _daily_note_reading_node(entry) |
| 3752 | nodes.append( |
| 3753 | { |
| 3754 | "node_id": node["node_id"], |
| 3755 | "node_type": "paper", |
| 3756 | "title": node["title"], |
| 3757 | "body": str(entry.get("summary") or "")[:500], |
| 3758 | "score": 1.0, |
| 3759 | "metadata": node["metadata"], |
| 3760 | } |
| 3761 | ) |
| 3762 | return {"nodes": nodes, "source": "daily_note_deep_reading"} |
| 3763 | |
| 3764 | |
| 3765 | def _parse_daily_note_topic_summaries(path: Path) -> Dict[str, Dict[str, Any]]: |
| 3766 | try: |
| 3767 | lines = path.read_text(encoding="utf-8").splitlines() |
| 3768 | except OSError: |
| 3769 | return {} |
| 3770 | summaries: Dict[str, Dict[str, Any]] = {} |
| 3771 | current_topic = "" |
| 3772 | index = 0 |
| 3773 | while index < len(lines): |
| 3774 | line = lines[index].strip() |
| 3775 | if line.startswith("# ") and not line.startswith("## "): |
nothing calls this directly
no test coverage detected