378 lines
16 KiB
Python
378 lines
16 KiB
Python
"""结题报告初稿生成内核（唯一业务内核）。
|
||
|
||
五步组装（对齐《核心建设内容：各部门"Skill"深度需求规格书》科研处 Skill 002）：
|
||
C1 指标对比：按成果类型汇总实际数量 vs 预期指标 → 超额完成/持平/未达标/数据待补
|
||
C2 完成情况章节：逐指标成文 + 成果明细（来源标注）
|
||
C3 决算转换：按科目汇总金额与占比，生成规范描述句（金额不改写）
|
||
C4 素材要点：代表性成果与年度报告摘句逐条提取（标注来源）
|
||
C5 骨架组装：通用结题结构，叙事章节只放素材要点 + [待补充] 指引
|
||
W1 数据完整性：缺数据生成占位，不阻塞整稿
|
||
|
||
红线：初稿中每个数字都来自台账；严禁编造数据与虚构论文结论。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import uuid
|
||
from datetime import date
|
||
from typing import Any, Optional
|
||
|
||
from db.connection import get_conn, init_db
|
||
from util.logging_config import get_skill_logger
|
||
from util.timeutil import now_unix
|
||
|
||
_PLACEHOLDER = "[待补充：请结合素材要点与原始报告撰写，注意引用本人成果]"
|
||
|
||
_EXPENSE_SENTENCE_TEMPLATES = {
|
||
"设备费": "购置/使用科研设备",
|
||
"材料费": "采购实验材料",
|
||
"测试化验加工费": "开展测试化验与加工",
|
||
"燃料动力费": "支付燃料动力费用",
|
||
"差旅费": "赴外开展调研与学术交流",
|
||
"会议费": "组织/参加学术会议",
|
||
"国际合作与交流费": "开展国际合作与交流",
|
||
"出版/文献/信息传播/知识产权事务费": "支付出版与知识产权事务费用",
|
||
"劳务费": "支付参与项目的研究生与临时聘用人员劳务",
|
||
"专家咨询费": "支付专家咨询费用",
|
||
"其他": "其他与项目研究直接相关的支出",
|
||
}
|
||
|
||
|
||
def _log():
|
||
try:
|
||
return get_skill_logger()
|
||
except RuntimeError:
|
||
return None
|
||
|
||
|
||
def _wan(value: Optional[float]) -> str:
|
||
if value is None:
|
||
return "—"
|
||
return f"{value / 10000:.2f} 万元"
|
||
|
||
|
||
def _camel(name: str) -> str:
|
||
parts = str(name).split("_")
|
||
if len(parts) <= 1:
|
||
return str(name)
|
||
head, *rest = parts
|
||
return head + "".join(p[:1].upper() + p[1:] for p in rest if p)
|
||
|
||
|
||
def _to_float(value: Any) -> Optional[float]:
|
||
if value is None or value == "":
|
||
return None
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def _snake(name: str) -> str:
|
||
return "".join("_" + ch.lower() if ch.isupper() else ch for ch in str(name)).lstrip("_")
|
||
|
||
|
||
def _norm_project(raw: dict[str, Any]) -> dict[str, Any]:
|
||
aliases = {
|
||
"projectNo": ("project_no", "项目编号"),
|
||
"projectName": ("project_name", "项目名称"),
|
||
"piName": ("pi_name", "负责人"),
|
||
"department": ("department", "学院", "所属学院"),
|
||
"program": ("program", "项目类型", "资助渠道"),
|
||
"startDate": ("start_date", "执行期开始", "开始日期"),
|
||
"endDate": ("end_date", "执行期结束", "结束日期"),
|
||
"approvedBudget": ("approved_budget", "立项经费", "批准经费"),
|
||
"expected": ("expected_json", "预期成果指标", "预期指标"),
|
||
"annualReportExcerpt": ("annual_report_excerpt", "年度报告摘句", "年度报告"),
|
||
"status": ("status", "状态"),
|
||
}
|
||
out: dict[str, Any] = {}
|
||
for key, value in raw.items():
|
||
out.setdefault(str(key), value)
|
||
for canonical, names in aliases.items():
|
||
if str(key) in names or _camel(str(key)) == canonical:
|
||
out.setdefault(canonical, value)
|
||
return out
|
||
|
||
|
||
def _norm_achievement(raw: dict[str, Any]) -> dict[str, Any]:
|
||
aliases = {
|
||
"projectNo": ("project_no", "项目编号"),
|
||
"achievementType": ("achievement_type", "成果类型", "类型"),
|
||
"title": ("title", "成果名称", "标题"),
|
||
"venueOrNo": ("venue_or_no", "期刊/专利号", "期刊", "专利号"),
|
||
"achDate": ("ach_date", "日期", "发表日期"),
|
||
"isRepresentative": ("is_representative", "代表性成果"),
|
||
"note": ("note", "备注"),
|
||
}
|
||
out: dict[str, Any] = {}
|
||
for key, value in raw.items():
|
||
out.setdefault(str(key), value)
|
||
for canonical, names in aliases.items():
|
||
if str(key) in names or _camel(str(key)) == canonical:
|
||
out.setdefault(canonical, value)
|
||
return out
|
||
|
||
|
||
def _norm_expense(raw: dict[str, Any]) -> dict[str, Any]:
|
||
aliases = {
|
||
"projectNo": ("project_no", "项目编号"),
|
||
"category": ("category", "科目"),
|
||
"amount": ("amount", "金额"),
|
||
"occurredDate": ("occurred_date", "发生日期"),
|
||
"description": ("description", "原始摘要", "描述"),
|
||
}
|
||
out: dict[str, Any] = {}
|
||
for key, value in raw.items():
|
||
out.setdefault(str(key), value)
|
||
for canonical, names in aliases.items():
|
||
if str(key) in names or _camel(str(key)) == canonical:
|
||
out.setdefault(canonical, value)
|
||
return out
|
||
|
||
|
||
def _c1_compare(expected: list[dict[str, Any]], achievements: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""C1：按类型对比预期指标与实际成果。
|
||
|
||
成果台账整表为空时无从对比，一律判"数据待补"（不下"未达标"结论）；
|
||
台账非空但某类型无记录按实际 0 计（事实陈述）。
|
||
"""
|
||
counts: dict[str, int] = {}
|
||
for ach in achievements:
|
||
ach_type = str(ach.get("achievement_type") or "").strip()
|
||
counts[ach_type] = counts.get(ach_type, 0) + 1
|
||
ledger_empty = not achievements
|
||
rows = []
|
||
for item in expected:
|
||
item_type = str(item.get("type") or "").strip()
|
||
target = item.get("target")
|
||
try:
|
||
target_v = int(float(target))
|
||
except (TypeError, ValueError):
|
||
target_v = None
|
||
unit = str(item.get("unit") or "项")
|
||
actual = counts.get(item_type)
|
||
if target_v is None:
|
||
verdict, detail = "数据待补", f"预期指标未配置数值（类型：{item_type}）"
|
||
elif ledger_empty:
|
||
verdict, detail = "数据待补", f"预期 {target_v} {unit}，成果清单尚未录入，无法对比（W1）"
|
||
elif actual is None:
|
||
verdict, detail = "未达标", f"预期 {target_v} {unit}，实际 0 {unit}（台账中未录入该类成果）"
|
||
actual = 0
|
||
else:
|
||
if actual > target_v:
|
||
verdict = "超额完成"
|
||
elif actual == target_v:
|
||
verdict = "持平"
|
||
else:
|
||
verdict = "未达标"
|
||
detail = f"预期 {target_v} {unit}，实际 {actual} {unit}，{verdict}"
|
||
rows.append({"type": item_type, "target": target_v, "unit": unit, "actual": actual, "verdict": verdict, "detail": detail})
|
||
return rows
|
||
|
||
|
||
def _c3_expense_sentences(expenses: list[dict[str, Any]], total_approved: Optional[float]) -> tuple[list[dict[str, Any]], str]:
|
||
"""C3：按科目汇总并生成规范描述句。返回 (rows, 汇总段落)。"""
|
||
by_category: dict[str, float] = {}
|
||
for item in expenses:
|
||
try:
|
||
amount = float(item.get("amount") or 0)
|
||
except (TypeError, ValueError):
|
||
amount = 0.0
|
||
category = str(item.get("category") or "其他").strip() or "其他"
|
||
by_category[category] = by_category.get(category, 0.0) + amount
|
||
total = round(sum(by_category.values()), 2)
|
||
rows = []
|
||
sentences = []
|
||
for category, amount in sorted(by_category.items(), key=lambda kv: -kv[1]):
|
||
ratio = round(amount / total, 4) if total > 0 else 0.0
|
||
action = _EXPENSE_SENTENCE_TEMPLATES.get(category, f"开展{category}相关支出")
|
||
sentence = f"{action}，支出 {_wan(amount)}，占比 {ratio * 100:.1f}%"
|
||
rows.append({"category": category, "amount": round(amount, 2), "ratio": ratio, "sentence": sentence})
|
||
sentences.append(sentence)
|
||
if not rows:
|
||
return [], "经费决算明细尚未录入，请在补录后重新生成初稿（W1）。"
|
||
tail = f"。项目决算合计支出 {_wan(total)}。"
|
||
if total_approved is not None:
|
||
balance = round(total_approved - total, 2)
|
||
if balance >= 0:
|
||
tail += f"结余 {_wan(balance)}，支出总额控制在立项经费（{_wan(total_approved)}）范围内。"
|
||
else:
|
||
tail += f"超出立项经费（{_wan(total_approved)}）{_wan(-balance)}，请补充说明。"
|
||
return rows, "本项目经费支出按科目如下：" + "；".join(sentences) + "。" + tail
|
||
|
||
|
||
def _c4_narrative_points(achievements: list[dict[str, Any]], excerpt: str) -> list[dict[str, str]]:
|
||
"""C4：从代表性成果与年度报告摘句提取叙事素材要点。"""
|
||
points = []
|
||
for ach in achievements:
|
||
if ach.get("is_representative"):
|
||
title = str(ach.get("title") or "").strip()
|
||
venue = str(ach.get("venue_or_no") or "").strip()
|
||
points.append({"source": "成果清单", "point": f"代表性成果：{title}" + (f"（{venue}）" if venue else "")})
|
||
for sentence in str(excerpt or "").replace("；", "。").split("。"):
|
||
sentence = sentence.strip()
|
||
if sentence:
|
||
points.append({"source": "年度报告", "point": sentence})
|
||
return points
|
||
|
||
|
||
def build_draft_sections(
|
||
project: dict[str, Any],
|
||
achievements: list[dict[str, Any]],
|
||
expenses: list[dict[str, Any]],
|
||
) -> dict[str, Any]:
|
||
"""组装初稿各章节数据（纯计算，不落库不出文件）。"""
|
||
import json as _json
|
||
|
||
project = _norm_project(project)
|
||
raw_expected = project.get("expected_json")
|
||
if raw_expected in (None, ""):
|
||
raw_expected = project.get("expected") or project.get("expectedJson") or []
|
||
if isinstance(raw_expected, str):
|
||
try:
|
||
expected = _json.loads(raw_expected)
|
||
except (TypeError, ValueError):
|
||
expected = []
|
||
elif isinstance(raw_expected, list):
|
||
expected = raw_expected
|
||
else:
|
||
expected = []
|
||
if isinstance(expected, dict):
|
||
expected = expected.get("indicators") or []
|
||
|
||
achievements = [{_snake(k): v for k, v in _norm_achievement(a).items()} for a in achievements]
|
||
expenses = [{_snake(k): v for k, v in _norm_expense(e).items()} for e in expenses]
|
||
|
||
completion = _c1_compare(expected, achievements)
|
||
approved_budget = _to_float(project.get("approved_budget") or project.get("approvedBudget"))
|
||
expense_rows, expense_paragraph = _c3_expense_sentences(expenses, approved_budget)
|
||
excerpt = project.get("annual_report_excerpt") or project.get("annualReportExcerpt") or ""
|
||
narrative_points = _c4_narrative_points(achievements, str(excerpt))
|
||
|
||
achievement_lines = []
|
||
for ach in achievements:
|
||
title = str(ach.get("title") or "").strip()
|
||
venue = str(ach.get("venue_or_no") or "").strip()
|
||
ach_date = str(ach.get("ach_date") or "").strip()
|
||
line = f"{ach.get('achievement_type')}：{title}" + (f"（{venue}，{ach_date}）" if venue or ach_date else "")
|
||
achievement_lines.append(line + "。来源：成果清单")
|
||
|
||
sections = {
|
||
"1. 研究背景与目标": [_PLACEHOLDER],
|
||
"2. 研究内容与进展": (
|
||
[f"· {p['point']}（来源：{p['source']}）" for p in narrative_points] or [f"· {_PLACEHOLDER}"]
|
||
) + [_PLACEHOLDER],
|
||
"3. 预期成果完成情况": [f"{row['type']}：{row['detail']}。来源：预期指标与成果清单" for row in completion]
|
||
or ["预期成果指标未配置，请在项目台账补录后重新生成（W1）。"],
|
||
"4. 经费决算说明": [expense_paragraph, "明细："] + [f"· {row['sentence']}" for row in expense_rows],
|
||
"5. 成果清单": achievement_lines or ["成果清单尚未录入，请在补录后重新生成初稿（W1）。"],
|
||
"6. 存在问题与展望": [_PLACEHOLDER],
|
||
}
|
||
return {
|
||
"completion": completion,
|
||
"expense_rows": expense_rows,
|
||
"expense_paragraph": expense_paragraph,
|
||
"narrative_points": narrative_points,
|
||
"achievement_lines": achievement_lines,
|
||
"sections": sections,
|
||
}
|
||
|
||
|
||
def run_draft(project_raw: dict[str, Any], achievements: list[dict[str, Any]], expenses: list[dict[str, Any]], *, batch_no: str = "") -> dict[str, Any]:
|
||
"""初稿内核：三步生成 → 写 docx + HTML 预览 → 落库。"""
|
||
from service.docx_service import write_docx
|
||
from service.report_service import write_preview_for_kernel
|
||
|
||
log = _log()
|
||
init_db()
|
||
project = _norm_project(project_raw)
|
||
project_no = str(project.get("project_no") or "").strip()
|
||
if not project_no:
|
||
raise ValueError("项目编号（project_no）不能为空")
|
||
batch_no = batch_no or now_unix().to_bytes(5, "big").hex() + uuid.uuid4().hex[:6]
|
||
|
||
if log:
|
||
log.info("draft_kernel_start batch_no=%s project_no=%s stage=kernel", batch_no, project_no)
|
||
|
||
sections_data = build_draft_sections(project, achievements, expenses)
|
||
sections = sections_data["sections"]
|
||
|
||
draft_file = write_docx(project, sections, batch_no)
|
||
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO report_drafts
|
||
(batch_no, project_no, project_name, completion_json, narrative_points_json, draft_file, preview_file)
|
||
VALUES (?, ?, ?, ?, ?, ?, '')
|
||
""",
|
||
(
|
||
batch_no,
|
||
project_no,
|
||
str(project.get("project_name") or ""),
|
||
json.dumps(sections_data["completion"], ensure_ascii=False),
|
||
json.dumps(sections_data["narrative_points"], ensure_ascii=False),
|
||
draft_file,
|
||
),
|
||
)
|
||
draft_id = int(cur.lastrowid)
|
||
# 预览按初稿记录编号命名（draft-{id}.html），多次生成互不覆盖。
|
||
draft = {
|
||
"project_no": project_no,
|
||
"project_name": str(project.get("project_name") or ""),
|
||
"batch_no": batch_no,
|
||
"sections": sections,
|
||
"completion": sections_data["completion"],
|
||
"narrative_points": sections_data["narrative_points"],
|
||
}
|
||
preview_file = write_preview_for_kernel(draft=draft, draft_id=draft_id)
|
||
cur.execute("UPDATE report_drafts SET preview_file = ? WHERE id = ?", (preview_file, draft_id))
|
||
cur.execute(
|
||
"UPDATE closing_projects SET status = '初稿已生成', updated_at = ? WHERE project_no = ?",
|
||
(now_unix(), project_no),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
payload = {
|
||
"ok": True,
|
||
"batch_no": batch_no,
|
||
"draft_id": draft_id,
|
||
"project_no": project_no,
|
||
"draft_file": draft_file,
|
||
"preview_file": preview_file,
|
||
"completion": sections_data["completion"],
|
||
"narrative_points_count": len(sections_data["narrative_points"]),
|
||
"disclaimer": "初稿由台账数据与模板生成；数据章节请核对，叙事章节须本人终审，严禁引入台账之外的成果。",
|
||
}
|
||
if log:
|
||
log.info("draft_kernel_done batch_no=%s project_no=%s draft_id=%s stage=kernel", batch_no, project_no, draft_id)
|
||
return payload
|
||
|
||
|
||
def query_pending_projects(project_no: str = "") -> list[dict[str, Any]]:
|
||
"""读取待结题项目（可按项目编号筛选），附成果与决算数量。"""
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
if project_no:
|
||
cur.execute("SELECT * FROM closing_projects WHERE project_no = ?", (project_no,))
|
||
else:
|
||
cur.execute("SELECT * FROM closing_projects WHERE status = '待结题' ORDER BY id")
|
||
columns = [d[0] for d in cur.description]
|
||
projects = [dict(zip(columns, row)) for row in cur.fetchall()]
|
||
for project in projects:
|
||
pno = project.get("project_no")
|
||
cur.execute("SELECT COUNT(*) FROM achievements WHERE project_no = ?", (pno,))
|
||
project["achievement_count"] = int(cur.fetchone()[0])
|
||
cur.execute("SELECT COUNT(*) FROM expense_items WHERE project_no = ?", (pno,))
|
||
project["expense_count"] = int(cur.fetchone()[0])
|
||
finally:
|
||
conn.close()
|
||
return projects
|