daily-robots/daily/agent_workflow.py

"""Agent 工作流：趋势分析 → 叙事化企微早报。"""

from __future__ import annotations

import json
import logging
import re
from pathlib import Path
from typing import Any

from daily.config import OUTPUT_DIR, ROOT, env
from daily.llm_client import extract_json_object, has_llm_configured, llm_chat
from daily.report_data import save_json

logger = logging.getLogger(__name__)

_SKILL_DIR = ROOT / "skills" / "daily-agent"
_MD_BLOCK = re.compile(r"```(?:markdown|md)?\s*([\s\S]*?)```", re.IGNORECASE)
_WECOM_NEW_ENTRY_NOTE = re.compile(r"（新入[^）]*）")


def _strip_new_entry_notes(md: str) -> str:
    md = _WECOM_NEW_ENTRY_NOTE.sub("", md)
    return re.sub(r"\*\*—", "** —", md)


def report_mode() -> str:
    return (env("DAILY_REPORT_MODE") or "classic").strip().lower()


def is_agent_mode() -> bool:
    if report_mode() != "agent":
        return False
    if not has_llm_configured():
        logger.warning("DAILY_REPORT_MODE=agent 但未配置 LLM，回退 classic")
        return False
    return True


def trends_json_path(date_str: str) -> Path:
    return OUTPUT_DIR / f"{date_str}.trends.json"


def _load_skill() -> str:
    path = _SKILL_DIR / "SKILL.md"
    if path.exists():
        return path.read_text(encoding="utf-8").strip()
    return "你是早报主编 Agent。"


def _extract_markdown(text: str) -> str:
    text = text.strip()
    match = _MD_BLOCK.search(text)
    if match:
        return match.group(1).strip()
    if text.startswith("📰"):
        return text
    return text


def analyze_trends(llm_input: dict[str, Any], *, date_str: str) -> dict[str, Any] | None:
    from daily.config import theme_ban_days
    from daily.narrative_axis import (
        enforce_narrative_axis,
        load_recent_axes,
        load_recent_theme_summaries,
        pick_narrative_axis,
    )

    skill = _load_skill()
    featured_note = ""
    if llm_input.get("featured_pick"):
        featured_note = (
            "\n输入已含 **featured_pick**（编辑指定今日首推）；"
            "top_picks.skill 必须以 featured_pick 为准；"
            "why/opening 不得向读者提及「编辑指定」。\n"
        )
    used_axes = set(load_recent_axes(date_str))
    axis = pick_narrative_axis(used_axes)
    llm_input["required_narrative_axis"] = axis
    llm_input["narrative_axis"] = axis
    theme_ban = load_recent_theme_summaries(date_str, theme_ban_days())
    ban_note = ""
    if theme_ban:
        ban_note = (
            "\n近几日已用过的主题/导语（请软避开同类开场，勿原样复用）：\n- "
            + "\n- ".join(theme_ban)
            + "\n"
        )
    system = (
        f"{skill}\n\n"
        f"{featured_note}"
        f"{ban_note}"
        "当前执行 **Step 1：趋势分析**。\n"
        f"**required_narrative_axis** = `{axis}`；输出 JSON 必须含 `narrative_axis` 且等于该值。\n"
        "只输出 trends JSON（headline, opening, themes, top_picks, signals, narrative_axis），不要 Markdown。"
    )
    user = json.dumps(llm_input, ensure_ascii=False, indent=2)
    try:
        raw = llm_chat(system, user)
    except Exception as exc:
        logger.warning("Agent Step1 趋势分析失败：%s", exc)
        return None
    if not raw:
        return None
    parsed = extract_json_object(raw)
    if not parsed.get("headline") and not parsed.get("opening"):
        logger.warning("Agent Step1 JSON 无效")
        return None
    parsed = enforce_narrative_axis(parsed, axis)
    save_json(trends_json_path(date_str), parsed)
    logger.info("Agent Step1 完成：%s [%s]", parsed.get("headline", "?"), axis)
    return parsed


def write_wecom_report(
    llm_input: dict[str, Any],
    trends: dict[str, Any],
    *,
    date_str: str,
    time_str: str,
    updated: str,
) -> str | None:
    skill = _load_skill()
    featured_note = ""
    if llm_input.get("featured_pick"):
        featured_note = (
            "\n输入 data 已含 **featured_pick**；"
            "今日首推区块须使用 featured_pick.why_today；"
            "链接行用 Markdown [标题](URL)，勿用反引号裸 URL；"
            "读者可见文案不得出现「编辑指定」等元信息。\n"
        )
    system = (
        f"{skill}\n\n"
        f"{featured_note}"
        "当前执行 **Step 2：撰写企微早报**。\n"
        f"日期={date_str}，时间={time_str}，数据截至={updated}。\n"
        "只输出企微 Markdown 正文，不要代码块，不要 JSON。"
    )
    payload = {"data": llm_input, "trends": trends}
    user = json.dumps(payload, ensure_ascii=False, indent=2)
    try:
        raw = llm_chat(system, user)
    except Exception as exc:
        logger.warning("Agent Step2 写稿失败：%s", exc)
        return None
    if not raw:
        return None
    md = _extract_markdown(raw)
    if not md.startswith("📰"):
        md = f"📰 **早报 · {date_str}**\n> ⏱ {time_str} · 数据截至 {updated}\n\n{md}"
    md = _strip_new_entry_notes(md)
    logger.info("Agent Step2 完成：%d bytes", len(md.encode("utf-8")))
    return md


def run_agent_workflow(
    llm_input: dict[str, Any],
    *,
    date_str: str,
    time_str: str,
    updated: str,
) -> str | None:
    """两步 Agent 工作流；成功返回企微 Markdown，失败返回 None。"""
    trends = analyze_trends(llm_input, date_str=date_str)
    if not trends:
        return None
    return write_wecom_report(
        llm_input,
        trends,
        date_str=date_str,
        time_str=time_str,
        updated=updated,
    )