csp/scripts/analyze_learning_note.py

#!/usr/bin/env python3
"""Score a learning note (0-100) and map to rating (1-10) via LLM with fallback."""

from __future__ import annotations

import argparse
import json
import os
import time
from typing import Any, Dict, Optional

import requests


def env(name: str, default: str = "") -> str:
    v = os.getenv(name, "").strip()
    return v if v else default


def load_input(path: str) -> Dict[str, Any]:
    with open(path, "r", encoding="utf-8") as f:
        data = json.load(f)
    if not isinstance(data, dict):
        raise ValueError("input json must be object")
    return data


def fallback(note: str) -> Dict[str, Any]:
    n = note.strip()
    score = 40
    if len(n) >= 300:
        score += 15
    if len(n) >= 800:
        score += 10
    if "```" in n:
        score += 15
    if "踩坑" in n or "错误" in n or "debug" in n.lower():
        score += 10
    if "总结" in n or "注意" in n:
        score += 10
    score = min(100, score)
    rating = max(1, min(10, round(score / 10)))
    feedback_md = (
        "### 笔记评分（规则兜底）\n"
        f"- 评分：**{score}/100**，评级：**{rating}/10**\n"
        "\n### 你做得好的地方\n"
        "- 记录了学习过程（已检测到一定的笔记内容）。\n"
        "\n### 建议补充\n"
        "- 写清：本节课**学习目标**、**关键概念**、**代码模板**。\n"
        "- 至少写 1 个你遇到的坑（如输入输出格式、编译报错）以及解决方案。\n"
        "- 最后用 3-5 行做总结，方便复习。\n"
    )
    return {"score": score, "rating": rating, "feedback_md": feedback_md, "model_name": "fallback-rules"}


def call_llm(payload: Dict[str, Any]) -> Dict[str, Any]:
    api_url = env("OI_LLM_API_URL") or env("CSP_LLM_API_URL")
    api_key = env("OI_LLM_API_KEY") or env("CSP_LLM_API_KEY")
    model = env("OI_LLM_MODEL", "qwen3-max")
    if not api_url:
        raise RuntimeError("missing OI_LLM_API_URL")

    system = (
        "你是一位面向小学生的C++竞赛教练，请对学习笔记打分。"
        "评分满分100分，并给出10分制评级（rating=round(score/10)，范围1-10）。"
        "评分维度：覆盖度30、正确性30、可操作性20、反思总结20。"
        "输出必须是JSON，不要输出其他任何文字。"
    )
    user = {
        "task": "对学习笔记评分并给出改进建议",
        "problem": {
            "id": payload.get("problem_id"),
            "title": payload.get("problem_title"),
        },
        "note": payload.get("note", ""),
        "output_json_schema": {
            "score": "integer 0-100",
            "rating": "integer 1-10",
            "feedback_md": "markdown string",
            "model_name": "string",
        },
    }

    headers = {"Content-Type": "application/json"}
    if api_key:
        headers["Authorization"] = f"Bearer {api_key}"

    body = {
        "model": model,
        "stream": False,
        "temperature": 0.2,
        "messages": [
            {"role": "system", "content": system},
            {"role": "user", "content": json.dumps(user, ensure_ascii=False)},
        ],
    }

    last: Optional[Exception] = None
    for attempt in range(4):
        try:
            resp = requests.post(api_url, headers=headers, json=body, timeout=50)
            if resp.status_code < 500:
                resp.raise_for_status()
            else:
                raise RuntimeError(f"HTTP {resp.status_code}")
            data = resp.json()
            content = data["choices"][0]["message"]["content"]
            parsed = json.loads(content)
            if not isinstance(parsed, dict):
                raise ValueError("model output not object")
            score = int(parsed.get("score", 0))
            score = max(0, min(100, score))
            rating = int(parsed.get("rating", round(score / 10)))
            rating = max(1, min(10, rating))
            feedback_md = str(parsed.get("feedback_md", "")).strip() or "### 笔记评分\n- 请补充更多内容（学习目标/代码/总结）。"
            model_name = str(parsed.get("model_name", model)).strip() or model
            return {"score": score, "rating": rating, "feedback_md": feedback_md, "model_name": model_name}
        except Exception as e:  # noqa: BLE001
            last = e
            time.sleep(0.6 * (attempt + 1))

    raise RuntimeError(str(last) if last else "llm failed")


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--input-file", required=True)
    args = ap.parse_args()

    payload = load_input(args.input_file)
    note = str(payload.get("note", ""))
    if not note.strip():
        print(
            json.dumps(
                {"score": 0, "rating": 1, "feedback_md": "### 笔记为空\n请先写笔记再评分。", "model_name": "validator"},
                ensure_ascii=False,
            )
        )
        return 0

    try:
        out = call_llm(payload)
    except Exception:
        out = fallback(note)

    print(json.dumps(out, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())