文件内容
scripts/run.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""功能自测评估 — 自包含skill，解读自评量表结果"""
import argparse, json, re, sys
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.request import Request, urlopen
from urllib.error import URLError, HTTPError
API_URL = "https://maas-api.hivoice.cn/v1/chat/completions"
MODEL = "u1-insuremed"
from preprocess import (
PreprocessError, SUPPORTED_FILE_TYPES,
detect_input_type, load_input_artifact, first_matching_index, normalize_header,
)
FIELD_ALIASES = {
"assessment_id": ["assessment_id", "assessmentId", "评估ID"],
"questions": ["questions", "问题列表"],
"answers": ["answers", "答案"],
"assessed_at": ["assessed_at", "评估时间"],
}
def _call_llm(system_prompt: str, user_prompt: str, appkey: str) -> str:
payload = {"model": MODEL, "temperature": 0.0, "messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}]}
try:
req = Request(API_URL, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
headers={"Content-Type": "application/json", "Authorization": f"Bearer {appkey}"})
resp = urlopen(req, timeout=120)
return json.loads(resp.read().decode("utf-8"))["choices"][0]["message"]["content"]
except HTTPError as exc: raise RuntimeError(f"API HTTP {exc.code}")
except URLError as exc: raise RuntimeError(f"API unreachable: {exc.reason}")
def as_list(value: Any) -> List[Any]:
if value is None or value == "": return []
if isinstance(value, list): return value
return [value]
def to_float(v: Any) -> Optional[float]:
if v is None: return None
try: return float(v)
except (ValueError, TypeError): return None
SYSTEM_PROMPT = """你是一位康复评估师，帮助患者解读功能自测结果。
你的任务：
1. 解读每个问题的得分和含义
2. 计算总分并评估整体恢复情况
3. 指出强项和待改善项
4. 给出针对性的下一步建议
5. 用Markdown格式展示，包含简单的进度条或评级
输出Markdown格式，鼓励性语气。末尾加免责提示。"""
def build(data: Dict[str, Any], appkey: str) -> Dict[str, Any]:
assessment_id = data.get("assessment_id", "")
questions = as_list(data.get("questions", []))
answers = data.get("answers", {})
assessed_at = data.get("assessed_at", "")
# 本地计算分数
answer_summary = []
total = 0.0
for q in questions:
qid = q.get("id", "")
qtext = q.get("text", "")
ans = answers.get(qid, 0)
score = to_float(ans) or 0.0
total += score
answer_summary.append({"id": qid, "question": qtext, "answer": ans, "score": score})
user_prompt = f"""请解读以下功能自测评估结果：
评估ID：{assessment_id}
评估时间：{assessed_at}
问题数：{len(questions)}
总分：{total}（满分{len(questions)*5}，每个问题1-5分）
各题得分：
```json
{json.dumps(answer_summary, ensure_ascii=False, indent=2)}
```
请解读每个维度的得分，评估整体恢复阶段，给出建议。"""
text = _call_llm(SYSTEM_PROMPT, user_prompt, appkey)
return {
"skill": "功能自测评估",
"status": "ok",
"data": {
"assessment_id": assessment_id, "assessed_at": assessed_at,
"score": total, "max_score": len(questions) * 5,
"answer_summary": answer_summary,
},
"text": text.strip(),
}
# ── 多格式解析 ──
def load_json(path: str) -> Dict[str, Any]:
data = json.loads(Path(path).read_text(encoding="utf-8"))
if not isinstance(data, dict): raise ValueError("input must be a JSON object")
return data
def parse_text_kv(text: str, field_aliases: Dict[str, List[str]]) -> Dict[str, Any]:
result = {}
lines = [line.strip() for line in text.strip().splitlines() if line.strip()]
pattern = re.compile(r"^\s*([A-Za-z一-鿿_][A-Za-z0-9一-鿿_\s]*)\s*[:：]\s*(.+?)\s*$")
header_map = {}
for canonical, aliases in field_aliases.items():
for alias in aliases: header_map[normalize_header(alias)] = canonical; header_map[normalize_header(canonical)] = canonical
for line in lines:
match = pattern.match(line)
if not match: continue
key = header_map.get(normalize_header(match.group(1).strip()))
if key is None: continue
value_str = match.group(2).strip()
try: result[key] = json.loads(value_str)
except (json.JSONDecodeError, ValueError): result[key] = value_str
return result
def normalize_artifact(artifact, field_aliases):
kind = artifact.get("kind")
if kind == "json":
data = artifact["data"]
if isinstance(data, dict): return data
raise PreprocessError("JSON input must be an object.")
if kind == "text":
text = artifact.get("text", "")
try:
data = json.loads(text)
if isinstance(data, dict): return data
except (json.JSONDecodeError, ValueError): pass
return parse_text_kv(text, field_aliases)
if kind == "tables":
all_rows = []
for table in artifact.get("tables", []): all_rows.extend(table.get("rows", []))
if not all_rows: raise PreprocessError("No data rows found.")
header_row = all_rows[0]
col_map = {}
for canonical, aliases in field_aliases.items():
for alias in aliases:
idx = first_matching_index({normalize_header(cell): i for i, cell in enumerate(header_row) if cell.strip()}, (alias, canonical))
if idx is not None: col_map[canonical] = idx; break
if len(all_rows) < 2: return {}
data_row = all_rows[1]
result = {}
for key, idx in col_map.items():
if idx < len(data_row) and data_row[idx].strip():
val = data_row[idx].strip()
try: result[key] = json.loads(val)
except (json.JSONDecodeError, ValueError): result[key] = val
return result
raise PreprocessError(f"Unsupported artifact kind: {kind}")
def write_json(data, output):
text = json.dumps(data, ensure_ascii=False, indent=2)
if output: Path(output).parent.mkdir(parents=True, exist_ok=True); Path(output).write_text(text + "\n", encoding="utf-8")
else: print(text)
def main():
parser = argparse.ArgumentParser(description="功能自测评估")
parser.add_argument("--input", required=True)
parser.add_argument("--output", default="")
parser.add_argument("--input-type", default="auto", choices=["auto", *sorted(SUPPORTED_FILE_TYPES)])
parser.add_argument("--sheet", default=""); parser.add_argument("--encoding", default="utf-8")
parser.add_argument("--save-prepared", action="store_true")
parser.add_argument("--appkey", required=True, help="内部医疗大模型鉴权key（必填）")
args = parser.parse_args()
try:
input_path = Path(args.input)
if not input_path.exists(): print(f"ERROR: Input file not found: {input_path}", file=sys.stderr); return 1
input_type = detect_input_type(input_path, args.input_type)
if input_type == "json": data = load_json(args.input)
else: data = normalize_artifact(load_input_artifact(input_path, input_type, args.encoding, args.sheet), FIELD_ALIASES)
if args.save_prepared:
pp = Path(args.output).with_suffix(".prepared.json") if args.output else input_path.with_suffix(".prepared.json")
pp.parent.mkdir(parents=True, exist_ok=True); pp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
result = build(data, args.appkey)
write_json(result, args.output)
return 0
except PreprocessError as exc:
# 回退到 _shared/doc-preprocess 尝试处理
try:
_shared_dir = Path(__file__).resolve().parent.parents[3] / "_shared" / "doc-preprocess" / "scripts"
if not _shared_dir.exists():
print(f"ERROR: 无法读取输入文件，本地预处理失败且 _shared/doc-preprocess 不可用。原因：{exc}", file=sys.stderr)
return 1
import importlib.util as _iu
_spec = _iu.spec_from_file_location("_shared_preprocess", _shared_dir / "preprocess.py")
_sp = _iu.module_from_spec(_spec)
_spec.loader.exec_module(_sp)
input_type = _sp.detect_input_type(input_path, args.input_type)
if input_type == "json":
data = load_json(args.input)
else:
artifact = _sp.load_input_artifact(input_path, input_type, args.encoding, args.sheet)
data = normalize_artifact(artifact, FIELD_ALIASES)
if args.save_prepared:
prepared_path = Path(args.output).with_suffix(".prepared.json") if args.output else input_path.with_suffix(".prepared.json")
prepared_path.parent.mkdir(parents=True, exist_ok=True)
prepared_path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
result = build(data, args.appkey)
write_json(result, args.output)
return 0
except Exception as shared_exc:
print(f"ERROR: 无法读取输入文件。本地预处理失败：{exc}；_shared/doc-preprocess 回退也失败：{shared_exc}", file=sys.stderr)
return 1
except Exception as exc:
print(f"ERROR: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())