Agent/examples/piaoquan_demand/run.py

"""piaoquan_demand 示例的最小可运行入口。"""

import asyncio
import copy
import importlib
import json
import os
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional

from dotenv import load_dotenv
from sqlalchemy import desc, or_

from examples.piaoquan_demand.config import LOG_LEVEL, ENABLED_TOOLS
from examples.piaoquan_demand.db_manager import DatabaseManager
from examples.piaoquan_demand.models import TopicPatternExecution
from examples.piaoquan_demand.pattern_toos.pattern_service import run_mining
from examples.piaoquan_demand.prepare import prepare
from examples.piaoquan_demand.topic_build_agent_context import TopicBuildAgentContext
from examples.piaoquan_demand.mysql import mysql_db

# Clash Verge TUN 模式兼容：禁止 httpx/urllib 自动检测系统 HTTP 代理
os.environ.setdefault("no_proxy", "*")
# 该示例仅使用项目侧能力，禁用框架内置 skills
os.environ.setdefault("AGENT_DISABLE_BUILTIN_SKILLS", "1")
# 禁用内置工具自动注册，并开启严格工具白名单
os.environ.setdefault("AGENT_DISABLE_BUILTIN_TOOL_REGISTRATION", "1")
os.environ.setdefault("AGENT_STRICT_TOOL_SELECTION", "1")
# 禁用所有侧分支（压缩/反思）
os.environ.setdefault("AGENT_DISABLE_SIDE_BRANCHES", "1")

# 添加项目根目录到 Python 路径
sys.path.insert(0, str(Path(__file__).parent.parent.parent))

load_dotenv()

from agent.core.runner import AgentRunner
from agent.llm import create_openrouter_llm_call
from agent.llm.prompts import SimplePrompt
from agent.trace import FileSystemTraceStore, Message, Trace
from agent.utils import setup_logging
from log_capture import build_log, log

# 导入项目配置
from examples.piaoquan_demand.config import DEBUG, LOG_FILE, LOG_LEVEL, RUN_CONFIG, TRACE_STORE_PATH

CUSTOM_TOOL_MODULES = {
    # piaoquan_demand 示例：严格按工具名白名单加载对应模块
    "think_and_plan": "examples.piaoquan_demand.agent_tools",
    "get_category_tree": "examples.piaoquan_demand.topic_build_pattern_tools",
    "get_frequent_itemsets": "examples.piaoquan_demand.topic_build_pattern_tools",
    "get_itemset_detail": "examples.piaoquan_demand.topic_build_pattern_tools",
    "get_post_elements": "examples.piaoquan_demand.topic_build_pattern_tools",
    "search_elements": "examples.piaoquan_demand.topic_build_pattern_tools",
    "get_element_category_chain": "examples.piaoquan_demand.topic_build_pattern_tools",
    "get_category_detail": "examples.piaoquan_demand.topic_build_pattern_tools",
    "search_categories": "examples.piaoquan_demand.topic_build_pattern_tools",
    "get_category_elements": "examples.piaoquan_demand.topic_build_pattern_tools",
    "get_category_co_occurrences": "examples.piaoquan_demand.topic_build_pattern_tools",
    "get_element_co_occurrences": "examples.piaoquan_demand.topic_build_pattern_tools",
    "get_weight_score_topn": "examples.piaoquan_demand.weight_score_query_tools",
    "get_weight_score_by_name": "examples.piaoquan_demand.weight_score_query_tools",
    "create_demand_item": "examples.piaoquan_demand.demand_build_agent_tools",
    "create_demand_items": "examples.piaoquan_demand.demand_build_agent_tools",
    "write_execution_summary": "examples.piaoquan_demand.demand_build_agent_tools",
}


def get_execution_id_by_merge_level2(cluster_name: str):
    """根据二级品类和平台查询最新的 execution_id。"""
    session = DatabaseManager().get_session()
    try:
        start_of_today = datetime.combine(datetime.now().date(), datetime.min.time())
        query = session.query(TopicPatternExecution).filter(
            TopicPatternExecution.cluster_name == cluster_name,
            TopicPatternExecution.start_time >= start_of_today,
        )
        execution = query.order_by(desc(TopicPatternExecution.id)).first()
        if not execution:
            return None
        return execution.id
    finally:
        session.close()


def resolve_model(prompt: SimplePrompt) -> str:
    model_from_prompt = prompt.config.get("model")
    if model_from_prompt:
        return model_from_prompt
    return f"anthropic/{RUN_CONFIG.model}" if "/" not in RUN_CONFIG.model else RUN_CONFIG.model


def extract_assistant_text(message: Message) -> str:
    if message.role != "assistant":
        return ""
    content = message.content
    if isinstance(content, str):
        return content
    if isinstance(content, dict):
        text = content.get("text", "")
        # 即使本轮包含工具调用，也打印模型给出的文本，便于观察每一步输出
        if text:
            return text
    return ""


def register_selected_tools(tool_names: list[str]) -> None:
    for tool_name in tool_names:
        module_path = CUSTOM_TOOL_MODULES.get(tool_name)
        if not module_path:
            raise ValueError(f"未配置工具模块映射: {tool_name}")
        importlib.import_module(module_path)


def _join_element_names_to_name(element_names: object) -> str:
    """把 tool 入参 element_names 转成 demand_content.name（逗号分隔）。"""
    if element_names is None:
        return ""
    if isinstance(element_names, list):
        parts = [str(x).strip() for x in element_names if x is not None and str(x).strip()]
        return ",".join(parts)
    # 兼容异常数据：比如历史版本可能传了字符串
    return str(element_names).strip()


def _safe_truncate(s: object, max_len: int) -> str:
    if s is None:
        return ""
    s_str = str(s)
    if max_len and len(s_str) > max_len:
        return s_str[:max_len]
    return s_str


def _load_name_score_map(execution_id: int) -> dict:
    """读取 data/{execution_id} 下所有 JSON 的 name->score（同名取最高分）。"""
    data_dir = Path(__file__).parent / "data" / str(execution_id)
    if not data_dir.exists():
        return {}

    score_map = {}
    for json_path in data_dir.glob("*.json"):
        try:
            with open(json_path, "r", encoding="utf-8") as f:
                payload = json.load(f)
        except Exception:
            continue

        if not isinstance(payload, list):
            continue

        for item in payload:
            if not isinstance(item, dict):
                continue
            name = item.get("name")
            score = item.get("score")
            if isinstance(name, str) and isinstance(score, (int, float)):
                prev = score_map.get(name)
                score_f = float(score)
                if prev is None or score_f > prev:
                    score_map[name] = score_f
    return score_map


def _avg_score_for_joined_name(name: str, score_map: dict) -> float:
    """按逗号拆分 name，分别取分后求平均。"""
    parts = [part.strip() for part in str(name).split(",") if part and part.strip()]
    if not parts:
        return 0.0
    return sum(float(score_map.get(part, 0.0)) for part in parts) / len(parts)


def _create_demand_task(execution_id: int) -> Optional[int]:
    """创建 demand_task 记录，返回任务ID。"""
    try:
        task_id = mysql_db.insert(
            "demand_task",
            {
                "execution_id": execution_id,
                "status": 0,
                "log": "",
            },
        )
        log(f"[task] 创建 demand_task 成功，task_id={task_id}, execution_id={execution_id}")
        return task_id
    except Exception as e:
        log(f"[task] 创建 demand_task 失败，execution_id={execution_id}, error={e}")
        return None


def _finish_demand_task(task_id: Optional[int], status: int, task_log: str) -> None:
    """更新 demand_task 状态与日志。"""
    if not task_id:
        return
    try:
        mysql_db.update(
            "demand_task",
            {
                "status": int(status),
                "log": task_log or "",
            },
            "id = %s",
            (task_id,),
        )
        log(f"[task] 更新 demand_task 成功，task_id={task_id}, status={status}")
    except Exception as e:
        log(f"[task] 更新 demand_task 失败，task_id={task_id}, status={status}, error={e}")


def write_demand_items_to_mysql(execution_id: int, merge_level2: str) -> int:
    """
    把 result/{execution_id}/execution_id_{execution_id}_demand_items.json
    写入 MySQL 表 demand_content
    """
    # create_demand_item(s) 使用 Path.cwd()/result 作为输出目录。
    # 为了兼容“从不同目录启动脚本”的情况，这里同时尝试 cwd 和脚本目录两种结果位置。
    demand_items_path = (
            Path.cwd()
            / "result"
            / str(execution_id)
            / f"execution_id_{execution_id}_demand_items.json"
    )
    if not demand_items_path.exists():
        alt_path = (
                Path(__file__).parent
                / "result"
                / str(execution_id)
                / f"execution_id_{execution_id}_demand_items.json"
        )
        if alt_path.exists():
            demand_items_path = alt_path
        else:
            log(f"[mysql] 未找到需求 JSON：{demand_items_path}（也未找到 {alt_path}），跳过写入")
            return 0

    try:
        with open(demand_items_path, "r", encoding="utf-8") as f:
            loaded = json.load(f)
    except Exception as e:
        log(f"[mysql] 读取需求 JSON 失败：{demand_items_path}，error={e}")
        return 0

    items = loaded["items"] if isinstance(loaded, dict) and isinstance(loaded.get("items"), list) else loaded
    if not isinstance(items, list):
        log(f"[mysql] 需求 JSON 非数组，跳过写入：type={type(items)}")
        return 0

    score_map = _load_name_score_map(execution_id)
    rows: list[dict] = []
    for di in items:
        if not isinstance(di, dict):
            continue

        name = _join_element_names_to_name(di.get("element_names"))
        if not name:
            continue

        score = _avg_score_for_joined_name(name, score_map)
        reason = di.get("reason")
        desc_value = di.get("desc")
        ext_data = {"reason": reason, "desc": desc_value}

        rows.append(
            {
                "merge_leve2": _safe_truncate(merge_level2, 32),
                "name": _safe_truncate(name, 64),
                "score": float(score),
                "ext_data": json.dumps(ext_data, ensure_ascii=False),
            }
        )

    if not rows:
        log("[mysql] 生成行为空，跳过写入")
        return 0

    affected = mysql_db.insert_many("demand_content", rows)
    log(f"[mysql] 写入 demand_content 完成，rows={len(rows)}, affected={affected}")
    return len(rows)


async def run_once(execution_id, merge_level2) -> str:
    task_id = _create_demand_task(execution_id=execution_id)
    task_status = 2
    task_log_text = ""

    TopicBuildAgentContext.set_execution_id(execution_id)
    prepare(execution_id)

    base_dir = Path(__file__).parent
    output_dir = base_dir / "output"
    output_dir.mkdir(exist_ok=True)

    setup_logging(level=LOG_LEVEL, file=LOG_FILE)
    register_selected_tools(ENABLED_TOOLS)

    prompt = SimplePrompt(base_dir / "demand.md")

    model = resolve_model(prompt)

    run_config = copy.deepcopy(RUN_CONFIG)
    run_config.temperature = float(prompt.config.get("temperature", run_config.temperature))
    run_config.max_iterations = int(prompt.config.get("max_iterations", run_config.max_iterations))
    run_config.tools = ENABLED_TOOLS.copy()
    # 禁用反思/总结经验相关流程（避免进入 reflection 侧分支）
    run_config.enable_research_flow = False
    run_config.goal_compression = "none"
    run_config.force_side_branch = None
    run_config.knowledge.enable_extraction = False
    run_config.knowledge.enable_completion_extraction = False
    run_config.knowledge.enable_injection = False
    run_config.trace_id = None

    initial_messages = prompt.build_messages(merge_level2=merge_level2)

    store = FileSystemTraceStore(base_path=TRACE_STORE_PATH)
    runner = AgentRunner(
        trace_store=store,
        llm_call=create_openrouter_llm_call(model=model),
        skills_dir=None,
        debug=DEBUG,
    )

    final_text = ""
    total_tokens = 0
    total_cost = 0.0
    has_completed_trace_cost = False
    log_file_path = output_dir / f"{execution_id}" / f"run_log_{execution_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt"
    log_file_path.parent.mkdir(parents=True, exist_ok=True)
    try:
        with build_log(execution_id) as log_buffer:
            async for item in runner.run(messages=initial_messages, config=run_config):
                if isinstance(item, Trace):
                    if getattr(item, "status", None) == "completed":
                        total_tokens = int(getattr(item, "total_tokens", 0) or 0)
                        total_cost = float(getattr(item, "total_cost", 0.0) or 0.0)
                        has_completed_trace_cost = True
                    continue
                elif isinstance(item, Message):
                    text = extract_assistant_text(item)
                    if text:
                        final_text = text
                        log(f"[assistant] {text}")
                    if not has_completed_trace_cost:
                        total_tokens += int(getattr(item, "total_tokens", 0) or 0)
                        total_cost += float(getattr(item, "cost", 0.0) or 0.0)

            if final_text:
                output_file = output_dir / f"{execution_id}" / "result.txt"
                output_file.parent.mkdir(parents=True, exist_ok=True)
                with open(output_file, "w", encoding="utf-8") as f:
                    f.write(final_text)

            log(f"[cost] total_tokens={total_tokens}, total_cost=${total_cost:.6f}")

            # agent 执行完成后：把本地 result JSON 写入 MySQL 表 demand_content
            # element_names -> name（逗号分隔）；reason/desc -> ext_data JSON；merge_leve2 -> demand_content.merge_leve2
            try:
                write_demand_items_to_mysql(execution_id=execution_id, merge_level2=merge_level2)
            except Exception as e:
                log(f"[mysql] 写入 demand_content 异常：{e}")

            task_log_text = log_buffer.getvalue()
            with open(log_file_path, "w", encoding="utf-8") as f:
                f.write(task_log_text)

            task_status = 1
    except Exception as e:
        if not task_log_text:
            task_log_text = f"[run] 执行异常: {e}"
        task_status = 2
        raise
    finally:
        _finish_demand_task(task_id=task_id, status=task_status, task_log=task_log_text)

    return final_text


async def main() -> None:
    cluster_name = '历史名人'
    execution_id = get_execution_id_by_merge_level2(cluster_name=cluster_name)
    if execution_id is None:
        execution_id = run_mining(cluster_name=cluster_name, merge_leve2=cluster_name)
    print(execution_id)

    # await run_once(execution_id, cluster_name)


if __name__ == "__main__":
    asyncio.run(main())