273 行
9.5 KiB
Python
273 行
9.5 KiB
Python
#!/usr/bin/env python3
|
||
"""Generate CSP-J/S style feedback for one submission via LLM (with fallback)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from dataclasses import dataclass
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import requests
|
||
|
||
DEFAULT_LINKS: List[Dict[str, str]] = [
|
||
{"title": "NOI 官网（规则与环境）", "url": "https://www.noi.cn/"},
|
||
{"title": "OI Wiki（算法知识库）", "url": "https://oi-wiki.org/"},
|
||
{"title": "cppreference C++14", "url": "https://en.cppreference.com/w/cpp/14"},
|
||
{
|
||
"title": "GCC Warning Options",
|
||
"url": "https://gcc.gnu.org/onlinedocs/gcc/Warning-Options.html",
|
||
},
|
||
{"title": "洛谷（题解与训练）", "url": "https://www.luogu.com.cn/"},
|
||
]
|
||
|
||
|
||
@dataclass
|
||
class LlmResult:
|
||
ok: bool
|
||
feedback_md: str
|
||
links: List[Dict[str, str]]
|
||
model_name: str
|
||
status: str
|
||
|
||
|
||
def env(name: str, default: str = "") -> str:
|
||
value = os.getenv(name, "").strip()
|
||
return value if value 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 detect_cpp14_risk(code: str, compile_log: str) -> List[str]:
|
||
hints: List[str] = []
|
||
joined = f"{code}\n{compile_log}"
|
||
checks = [
|
||
(r"\bif\s+constexpr\b", "检测到 `if constexpr`（C++17），C++14 环境会 CE。"),
|
||
(r"\bstd::optional\b", "检测到 `std::optional`（C++17），建议改为普通变量+标记位。"),
|
||
(r"\bstd::variant\b", "检测到 `std::variant`（C++17），建议改为 struct/enum 分支。"),
|
||
(r"\bstd::string_view\b", "检测到 `std::string_view`（C++17），建议改为 `const string&`。"),
|
||
(r"\[[^\]]+\]\s*=" , "检测到结构化绑定迹象，C++14 不支持，建议改 pair/struct 访问。"),
|
||
(r"%I64d", "检测到 `%I64d`，Linux 评测机应统一使用 `%lld`。"),
|
||
(r"\bvoid\s+main\s*\(", "检测到 `void main()`，需改为 `int main()` 并 `return 0;`。"),
|
||
]
|
||
for pattern, tip in checks:
|
||
if re.search(pattern, joined):
|
||
hints.append(tip)
|
||
if "-Wsign-compare" in compile_log:
|
||
hints.append("存在 `-Wsign-compare`，建议统一使用 `size_t` 或显式类型转换。")
|
||
return hints
|
||
|
||
|
||
def build_fallback_feedback(payload: Dict[str, Any], llm_error: str = "") -> LlmResult:
|
||
status = str(payload.get("status", "Unknown"))
|
||
score = payload.get("score", 0)
|
||
compile_log = str(payload.get("compile_log", ""))
|
||
runtime_log = str(payload.get("runtime_log", ""))
|
||
code = str(payload.get("code", ""))
|
||
|
||
risk_tips = detect_cpp14_risk(code, compile_log)
|
||
if not risk_tips:
|
||
risk_tips = [
|
||
"请确认只使用 C++14 语法，避免 C++17 特性导致 CE。",
|
||
"若题目要求文件输入输出，使用 `freopen(\"xxx.in\",\"r\",stdin)` / `freopen(\"xxx.out\",\"w\",stdout)`。",
|
||
]
|
||
|
||
thought = (
|
||
"代码通过了当前评测，核心思路基本正确，建议继续做规范化和鲁棒性收敛。"
|
||
if status.upper() == "AC"
|
||
else "当前提交未稳定通过，建议先按日志定位错误，再拆分为思路问题与实现问题逐步修复。"
|
||
)
|
||
|
||
lines: List[str] = []
|
||
lines.append("### 评测结论")
|
||
lines.append(f"- 本次状态：**{status}**，分数：**{score}**。")
|
||
lines.append(f"- 思路评价：{thought}")
|
||
lines.append("")
|
||
lines.append("### 福建 CSP-J/S 规范检查（C++14）")
|
||
for tip in risk_tips:
|
||
lines.append(f"- {tip}")
|
||
if compile_log.strip():
|
||
lines.append("- 编译日志有信息，建议逐条清理 warning，减少考场不确定性。")
|
||
if runtime_log.strip():
|
||
lines.append("- 运行日志有输出，建议重点检查边界输入与数组越界风险。")
|
||
lines.append("")
|
||
lines.append("### 修改建议（可执行）")
|
||
lines.append("- 按“先编译通过→再保证正确→最后做优化”的顺序迭代。")
|
||
lines.append("- `long long` 读写统一 `%lld`；不要使用 `%I64d`。")
|
||
lines.append("- 清理 signed/unsigned 警告，降低不同编译器行为差异。")
|
||
lines.append("- 确保 `int main()` 且 `return 0;`。")
|
||
lines.append("")
|
||
lines.append("### 知识点评测")
|
||
lines.append("- 强项：基础实现与调试流程。")
|
||
lines.append("- 待加强：边界构造、类型一致性、赛场环境兼容性。")
|
||
lines.append("")
|
||
lines.append("### 推荐外链资料")
|
||
for item in DEFAULT_LINKS:
|
||
lines.append(f"- [{item['title']}]({item['url']})")
|
||
|
||
if llm_error:
|
||
lines.append("")
|
||
lines.append(f"> 说明：LLM 调用失败，已返回规则兜底建议。错误：{llm_error}")
|
||
|
||
return LlmResult(
|
||
ok=True,
|
||
feedback_md="\n".join(lines).strip(),
|
||
links=DEFAULT_LINKS,
|
||
model_name="fallback-rules",
|
||
status="fallback",
|
||
)
|
||
|
||
|
||
def normalize_links(raw: Any) -> List[Dict[str, str]]:
|
||
links: List[Dict[str, str]] = []
|
||
if isinstance(raw, list):
|
||
for item in raw:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
title = str(item.get("title", "")).strip()
|
||
url = str(item.get("url", "")).strip()
|
||
if title and url:
|
||
links.append({"title": title, "url": url})
|
||
return links if links else DEFAULT_LINKS
|
||
|
||
|
||
def dict_to_markdown(data: Dict[str, Any]) -> str:
|
||
parts: List[str] = []
|
||
for key, value in data.items():
|
||
title = str(key).strip() or "分析项"
|
||
if isinstance(value, str):
|
||
body = value.strip()
|
||
else:
|
||
body = json.dumps(value, ensure_ascii=False, indent=2)
|
||
if not body:
|
||
continue
|
||
parts.append(f"### {title}\n{body}")
|
||
return "\n\n".join(parts)
|
||
|
||
|
||
def call_llm(payload: Dict[str, Any]) -> LlmResult:
|
||
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_prompt = (
|
||
"你是福建省 CSP-J/S 代码规范与评测老师。"
|
||
"请严格按 C++14 旧 GCC 环境给建议，重点指出会导致 CE/RE/爆零的风险。"
|
||
"输出 JSON，不要输出其他文字。"
|
||
)
|
||
user_prompt = {
|
||
"task": "分析这份提交并给出改进建议",
|
||
"required_sections": [
|
||
"评测结论",
|
||
"福建 CSP-J/S 规范检查（C++14）",
|
||
"修改建议",
|
||
"知识点评测",
|
||
"推荐外链资料",
|
||
],
|
||
"submission": payload,
|
||
"output_json_schema": {
|
||
"feedback_md": "markdown string",
|
||
"links": [{"title": "string", "url": "string"}],
|
||
"status": "ready",
|
||
},
|
||
}
|
||
|
||
headers = {"Content-Type": "application/json"}
|
||
if api_key:
|
||
headers["Authorization"] = f"Bearer {api_key}"
|
||
|
||
body = {
|
||
"model": model,
|
||
"stream": False,
|
||
"temperature": 0.1,
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": json.dumps(user_prompt, ensure_ascii=False)},
|
||
],
|
||
}
|
||
|
||
resp = requests.post(api_url, headers=headers, json=body, timeout=50)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
|
||
choices = data.get("choices") if isinstance(data, dict) else None
|
||
if not choices:
|
||
raise RuntimeError("LLM response missing choices")
|
||
first = choices[0] if isinstance(choices, list) and choices else {}
|
||
message = first.get("message") if isinstance(first, dict) else {}
|
||
content = message.get("content", "") if isinstance(message, dict) else ""
|
||
if not isinstance(content, str) or not content.strip():
|
||
raise RuntimeError("LLM content is empty")
|
||
|
||
model_name = str(data.get("model", model)) if isinstance(data, dict) else model
|
||
|
||
parsed: Optional[Dict[str, Any]] = None
|
||
try:
|
||
candidate = json.loads(content)
|
||
if isinstance(candidate, dict):
|
||
parsed = candidate
|
||
except Exception:
|
||
parsed = None
|
||
|
||
if parsed and parsed.get("feedback_md"):
|
||
return LlmResult(
|
||
ok=True,
|
||
feedback_md=str(parsed.get("feedback_md", "")).strip(),
|
||
links=normalize_links(parsed.get("links")),
|
||
model_name=model_name,
|
||
status=str(parsed.get("status", "ready")) or "ready",
|
||
)
|
||
|
||
if parsed:
|
||
return LlmResult(
|
||
ok=True,
|
||
feedback_md=dict_to_markdown(parsed),
|
||
links=DEFAULT_LINKS,
|
||
model_name=model_name,
|
||
status="ready",
|
||
)
|
||
|
||
return LlmResult(
|
||
ok=True,
|
||
feedback_md=content.strip(),
|
||
links=DEFAULT_LINKS,
|
||
model_name=model_name,
|
||
status="ready",
|
||
)
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="Analyze one submission with LLM + fallback")
|
||
parser.add_argument("--input-file", required=True, help="JSON file from backend")
|
||
args = parser.parse_args()
|
||
|
||
payload = load_input(args.input_file)
|
||
try:
|
||
result = call_llm(payload)
|
||
except Exception as exc:
|
||
result = build_fallback_feedback(payload, str(exc))
|
||
|
||
output = {
|
||
"feedback_md": result.feedback_md,
|
||
"links": result.links,
|
||
"model_name": result.model_name,
|
||
"status": result.status,
|
||
}
|
||
sys.stdout.write(json.dumps(output, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|