145 lines
4.7 KiB
Python
145 lines
4.7 KiB
Python
"""Cursor 编辑层：JSON 数据 → 主题 / 速览 / 中文描述。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from daily.config import ROOT, env
|
||
from daily.llm_client import extract_json_object, has_llm_configured, llm_chat
|
||
from daily.text_utils import clip_text
|
||
from daily.report_data import editorial_json_path, save_json, skill_id
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_SKILL_DIR = ROOT / "skills" / "daily-editor"
|
||
|
||
|
||
def is_enabled() -> bool:
|
||
raw = (env("DAILY_CURSOR_EDITOR") or "").strip().lower()
|
||
if raw in {"1", "true", "yes", "on"}:
|
||
return has_llm_configured()
|
||
if raw in {"0", "false", "no", "off"}:
|
||
return False
|
||
return False
|
||
|
||
|
||
def _load_skill_prompt() -> str:
|
||
skill_path = _SKILL_DIR / "SKILL.md"
|
||
if skill_path.exists():
|
||
return skill_path.read_text(encoding="utf-8").strip()
|
||
return "你是技术早报编辑。根据输入 JSON 输出编辑结果 JSON。"
|
||
|
||
|
||
def _build_system_prompt() -> str:
|
||
skill = _load_skill_prompt()
|
||
return (
|
||
f"{skill}\n\n"
|
||
"再次强调：只输出 JSON 对象，包含 theme_line、highlights（3条）、descriptions。"
|
||
)
|
||
|
||
|
||
def run_editorial(llm_input: dict[str, Any], *, date_str: str) -> dict[str, Any] | None:
|
||
"""调用 LLM 生成 editorial；失败返回 None。"""
|
||
if not is_enabled():
|
||
return None
|
||
system = _build_system_prompt()
|
||
user = json.dumps(llm_input, ensure_ascii=False, indent=2)
|
||
try:
|
||
raw = llm_chat(system, user)
|
||
except Exception as exc:
|
||
logger.warning("Cursor 编辑失败，回退规则模式：%s", exc)
|
||
return None
|
||
if not raw:
|
||
logger.warning("Cursor 编辑无响应，回退规则模式")
|
||
return None
|
||
parsed = extract_json_object(raw)
|
||
if not parsed.get("theme_line") and not parsed.get("descriptions"):
|
||
logger.warning("Cursor 编辑 JSON 无效，回退规则模式")
|
||
return None
|
||
editorial = _normalize_editorial(parsed)
|
||
save_json(editorial_json_path(date_str), editorial)
|
||
return editorial
|
||
|
||
|
||
def _normalize_editorial(raw: dict[str, Any]) -> dict[str, Any]:
|
||
theme = str(raw.get("theme_line") or "").strip()
|
||
highlights_raw = raw.get("highlights") or []
|
||
highlights: list[str] = []
|
||
if isinstance(highlights_raw, list):
|
||
for item in highlights_raw:
|
||
if isinstance(item, str) and item.strip():
|
||
highlights.append(item.strip())
|
||
descriptions_raw = raw.get("descriptions") or {}
|
||
descriptions: dict[str, str] = {}
|
||
if isinstance(descriptions_raw, dict):
|
||
for key, value in descriptions_raw.items():
|
||
if isinstance(value, str) and value.strip():
|
||
limit = 40 if str(key).startswith("github:") else 36
|
||
descriptions[str(key)] = clip_text(value, limit)
|
||
return {
|
||
"theme_line": theme,
|
||
"highlights": highlights[:3],
|
||
"descriptions": descriptions,
|
||
}
|
||
|
||
|
||
def theme_line_from_editorial(editorial: dict[str, Any]) -> str:
|
||
theme = editorial.get("theme_line", "")
|
||
if not theme:
|
||
return ""
|
||
if "今日主题" in theme:
|
||
return theme if theme.startswith("**") else f"**{theme}**"
|
||
return f"**今日主题**：{theme}"
|
||
|
||
|
||
def apply_descriptions(
|
||
*,
|
||
trending: list[dict[str, Any]],
|
||
hot: list[dict[str, Any]],
|
||
github_trending: list[dict[str, Any]],
|
||
github_emerging: list[dict[str, Any]],
|
||
github_topic: list[dict[str, Any]],
|
||
ai_news: dict[str, Any],
|
||
descriptions: dict[str, str],
|
||
) -> None:
|
||
if not descriptions:
|
||
return
|
||
|
||
for item in trending + hot:
|
||
key = f"skill:{skill_id(item)}"
|
||
if key in descriptions:
|
||
item["description"] = descriptions[key]
|
||
|
||
for repo_list in (github_trending, github_emerging, github_topic):
|
||
for item in repo_list:
|
||
key = f"github:{item.get('repo', '')}"
|
||
if key in descriptions:
|
||
item["description"] = descriptions[key]
|
||
|
||
if not ai_news.get("enabled"):
|
||
return
|
||
|
||
def _apply_news_item(item: dict[str, Any]) -> None:
|
||
key = f"news:{item.get('link', '')}"
|
||
if key in descriptions:
|
||
item["summary"] = descriptions[key]
|
||
|
||
for cat in ai_news.get("categories") or []:
|
||
for item in cat.get("items") or []:
|
||
_apply_news_item(item)
|
||
for item in ai_news.get("flat") or []:
|
||
_apply_news_item(item)
|
||
|
||
|
||
def load_cached_editorial(date_str: str) -> dict[str, Any] | None:
|
||
path = editorial_json_path(date_str)
|
||
if not path.exists():
|
||
return None
|
||
try:
|
||
return _normalize_editorial(json.loads(path.read_text(encoding="utf-8")))
|
||
except (OSError, json.JSONDecodeError):
|
||
return None
|