tools/executor.py

import json
from pathlib import Path
from typing import Any

from config import MAX_TOOL_OUTPUT_CHARS
from tools.attachment_loader import download_task_file
from tools.code_runner import solve_python_output
from tools.common import normalize_answer, read_plain_file, truncate_text
from tools.direct_rules import solve_direct
from tools.sports_solver import solve_sports
from tools.spreadsheet_solver import solve_excel_food_sales
from tools.structured_web_tools import solve_structured_web, wikipedia_parse_html, wikipedia_search_titles
from tools.tool_specs import ToolSpec
from tools.web_tools import fetch_url_text, search_web


def solver_result_to_observation(result, tool_name: str) -> dict[str, Any]:
    return {
        "tool": tool_name,
        "ok": result.solved,
        "answer": result.answer,
        "confidence": result.confidence,
        "source": result.source,
        "evidence": truncate_text(result.evidence, MAX_TOOL_OUTPUT_CHARS),
        "error": result.error,
    }


def disabled_tool_observation(tool_name: str, reason: str) -> dict[str, Any]:
    return {
        "tool": tool_name,
        "ok": False,
        "answer": None,
        "confidence": "low",
        "source": f"{tool_name}.disabled",
        "evidence": "",
        "error": reason,
    }


def direct_answer_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    question = args.get("question") or context["question"]
    return solver_result_to_observation(solve_direct(question), "direct_answer_tool")


def python_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    return solver_result_to_observation(
        solve_python_output(
            context["question"],
            args.get("task_id") or context.get("task_id", ""),
            args.get("file_name") or context.get("file_name", ""),
        ),
        "python_tool",
    )


def spreadsheet_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    return solver_result_to_observation(
        solve_excel_food_sales(
            context["question"],
            args.get("task_id") or context.get("task_id", ""),
            args.get("file_name") or context.get("file_name", ""),
        ),
        "spreadsheet_tool",
    )


def wikipedia_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    question = args.get("question") or context["question"]
    result = solve_structured_web(question)
    if result.solved:
        return solver_result_to_observation(result, "wikipedia_tool")

    query = args.get("query") or question
    try:
        titles = wikipedia_search_titles(query, limit=5)
        evidence_parts = [f"search_titles={titles}"]
        if titles:
            html = wikipedia_parse_html(titles[0])
            evidence_parts.append(truncate_text(html, 5000))
        return {
            "tool": "wikipedia_tool",
            "ok": bool(titles),
            "answer": None,
            "confidence": "medium" if titles else "low",
            "source": "wikipedia_tool.search",
            "evidence": truncate_text("\n\n".join(evidence_parts), MAX_TOOL_OUTPUT_CHARS),
            "error": "" if titles else "Wikipedia search returned no titles.",
        }
    except Exception as exc:
        return {
            "tool": "wikipedia_tool",
            "ok": False,
            "answer": None,
            "confidence": "low",
            "source": "wikipedia_tool.error",
            "evidence": "",
            "error": str(exc),
        }


def sports_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    question = args.get("question") or context["question"]
    return solver_result_to_observation(solve_sports(question), "sports_tool")


def web_search_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    query = args.get("query") or context["question"]
    results = search_web(query, max_results=int(args.get("max_results", 5)))
    return {
        "tool": "web_search_tool",
        "ok": bool(results),
        "answer": None,
        "confidence": "medium" if results else "low",
        "source": "web_search_tool",
        "evidence": truncate_text(json.dumps(results, ensure_ascii=False), MAX_TOOL_OUTPUT_CHARS),
        "error": "" if results else "No search results.",
    }


def web_read_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    url = args.get("url", "")
    if not url:
        return {
            "tool": "web_read_tool",
            "ok": False,
            "answer": None,
            "confidence": "low",
            "source": "web_read_tool",
            "evidence": "",
            "error": "Missing url.",
        }
    text = fetch_url_text(url, limit=MAX_TOOL_OUTPUT_CHARS)
    return {
        "tool": "web_read_tool",
        "ok": bool(text and not text.startswith("无法读取网页")),
        "answer": None,
        "confidence": "medium",
        "source": "web_read_tool",
        "evidence": truncate_text(text, MAX_TOOL_OUTPUT_CHARS),
        "error": "" if not text.startswith("无法读取网页") else text,
    }


def attachment_text_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    task_id = args.get("task_id") or context.get("task_id", "")
    file_name = args.get("file_name") or context.get("file_name", "")
    file_path, note = download_task_file(task_id, file_name)
    if not file_path:
        return {
            "tool": "attachment_text_tool",
            "ok": False,
            "answer": None,
            "confidence": "low",
            "source": "attachment_text_tool",
            "evidence": note,
            "error": note,
        }
    text = read_plain_file(Path(file_path), limit=MAX_TOOL_OUTPUT_CHARS)
    return {
        "tool": "attachment_text_tool",
        "ok": True,
        "answer": None,
        "confidence": "medium",
        "source": "attachment_text_tool",
        "evidence": f"{note}\n{text}",
        "error": "",
    }


def final_format_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    answer = normalize_answer(str(args.get("answer", "")))
    return {
        "tool": "final_format_tool",
        "ok": bool(answer),
        "answer": answer or None,
        "confidence": args.get("confidence", "medium"),
        "source": "final_format_tool",
        "evidence": "Normalized final answer.",
        "error": "" if answer else "Missing answer.",
    }


def audio_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    return disabled_tool_observation("audio_tool", "当前版本未启用音频转写工具。")


def video_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    return disabled_tool_observation("video_tool", "当前版本未启用视频/YouTube 工具。")


def vision_tool(args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    return disabled_tool_observation("vision_tool", "当前版本未启用图片/棋局视觉工具。")


TOOL_REGISTRY: dict[str, ToolSpec] = {
    "direct_answer_tool": ToolSpec(
        name="direct_answer_tool",
        description="Low-cost deterministic solver for self-contained logic questions: reversed text, non-commutative table, botanical vegetable list.",
        input_schema={"question": "optional string; omit to use current question"},
        run=direct_answer_tool,
        cost_level="free",
    ),
    "python_tool": ToolSpec(
        name="python_tool",
        description="Download and execute/analyze an attached Python file, then return the final numeric/text output if recoverable.",
        input_schema={"task_id": "optional string", "file_name": "optional string"},
        run=python_tool,
        cost_level="free",
    ),
    "spreadsheet_tool": ToolSpec(
        name="spreadsheet_tool",
        description="Download and analyze an attached Excel spreadsheet; currently solves food-sales-not-drinks style calculations.",
        input_schema={"task_id": "optional string", "file_name": "optional string"},
        run=spreadsheet_tool,
        cost_level="free",
    ),
    "wikipedia_tool": ToolSpec(
        name="wikipedia_tool",
        description="Use MediaWiki/Wikipedia APIs and structured parsers for Wikipedia-style questions.",
        input_schema={"query": "optional search query", "question": "optional question override"},
        run=wikipedia_tool,
        cost_level="low",
    ),
    "sports_tool": ToolSpec(
        name="sports_tool",
        description="Use sports data APIs; currently handles the 1977 Yankees walks/at-bats question.",
        input_schema={"question": "optional question override"},
        run=sports_tool,
        cost_level="low",
    ),
    "web_search_tool": ToolSpec(
        name="web_search_tool",
        description="Search the web and return result titles and URLs.",
        input_schema={"query": "string", "max_results": "optional integer, default 5"},
        run=web_search_tool,
        cost_level="low",
    ),
    "web_read_tool": ToolSpec(
        name="web_read_tool",
        description="Read a specific URL and return extracted page text.",
        input_schema={"url": "string"},
        run=web_read_tool,
        cost_level="low",
    ),
    "attachment_text_tool": ToolSpec(
        name="attachment_text_tool",
        description="Download an attachment and read it as text when it is a plain text/CSV/JSON/Markdown-like file.",
        input_schema={"task_id": "optional string", "file_name": "optional string"},
        run=attachment_text_tool,
        cost_level="free",
    ),
    "final_format_tool": ToolSpec(
        name="final_format_tool",
        description="Normalize a candidate final answer string to submission-safe format.",
        input_schema={"answer": "string", "confidence": "optional string"},
        run=final_format_tool,
        cost_level="free",
    ),
    "audio_tool": ToolSpec(
        name="audio_tool",
        description="Disabled audio transcription tool. Calling it reports unavailable.",
        input_schema={},
        run=audio_tool,
        enabled=False,
        cost_level="disabled",
    ),
    "video_tool": ToolSpec(
        name="video_tool",
        description="Disabled video/YouTube analysis tool. Calling it reports unavailable.",
        input_schema={},
        run=video_tool,
        enabled=False,
        cost_level="disabled",
    ),
    "vision_tool": ToolSpec(
        name="vision_tool",
        description="Disabled image/chess vision tool. Calling it reports unavailable.",
        input_schema={},
        run=vision_tool,
        enabled=False,
        cost_level="disabled",
    ),
}


def tool_prompt() -> str:
    return "\n".join(spec.prompt_block() for spec in TOOL_REGISTRY.values())


def execute_tool(action: str, args: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
    spec = TOOL_REGISTRY.get(action)
    if not spec:
        return {
            "tool": action,
            "ok": False,
            "answer": None,
            "confidence": "low",
            "source": "tool_executor.unknown_tool",
            "evidence": "",
            "error": f"Unknown tool: {action}",
        }
    if not spec.enabled:
        return disabled_tool_observation(action, f"Tool {action} is disabled.")
    try:
        return spec.run(args or {}, context)
    except Exception as exc:
        return {
            "tool": action,
            "ok": False,
            "answer": None,
            "confidence": "low",
            "source": f"{action}.exception",
            "evidence": "",
            "error": str(exc),
        }