319 lines
12 KiB
Python
319 lines
12 KiB
Python
"""结题项目 / 成果清单 / 决算明细 / 初稿记录 的维护与查询。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any, Optional
|
||
|
||
from db.connection import get_conn, init_db
|
||
from service.draft_service import _norm_achievement, _norm_expense, _norm_project
|
||
from util.logging_config import get_skill_logger
|
||
from util.timeutil import now_unix
|
||
|
||
_PROJECT_TEXT_FIELDS = ("project_name", "pi_name", "department", "program", "start_date", "end_date", "annual_report_excerpt", "status")
|
||
_KNOWN_STATUS = ("待结题", "初稿已生成", "已提交")
|
||
_KNOWN_ACH_TYPES = ("论文", "专利", "人才培养", "其他")
|
||
|
||
|
||
def _log():
|
||
try:
|
||
return get_skill_logger()
|
||
except RuntimeError:
|
||
return None
|
||
|
||
|
||
def _snake(name: str) -> str:
|
||
return "".join("_" + ch.lower() if ch.isupper() else ch for ch in str(name)).lstrip("_")
|
||
|
||
|
||
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 upsert_project(fields: dict[str, Any]) -> dict[str, Any]:
|
||
fields = {_snake(k): v for k, v in _norm_project(fields).items()}
|
||
project_no = str(fields.get("project_no") or "").strip()
|
||
if not project_no:
|
||
raise ValueError("项目编号（project_no）不能为空")
|
||
values: dict[str, Any] = {"project_no": project_no}
|
||
for key in _PROJECT_TEXT_FIELDS:
|
||
if fields.get(key) not in (None, ""):
|
||
values[key] = str(fields[key]).strip()
|
||
if values.get("status") and values["status"] not in _KNOWN_STATUS:
|
||
raise ValueError(f"status 只接受 {'、'.join(_KNOWN_STATUS)}：{values['status']}")
|
||
if fields.get("approved_budget") not in (None, ""):
|
||
approved = _to_float(fields.get("approved_budget"))
|
||
if approved is None or approved < 0:
|
||
raise ValueError(f"立项经费必须是非负数字（元）：{fields.get('approved_budget')}")
|
||
values["approved_budget"] = round(approved, 2)
|
||
if fields.get("expected_json") not in (None, ""):
|
||
raw = fields["expected_json"]
|
||
try:
|
||
expected = json.loads(raw) if isinstance(raw, str) else raw
|
||
except (TypeError, ValueError) as exc:
|
||
raise ValueError(f"预期指标 JSON 解析失败：{exc}") from exc
|
||
if not isinstance(expected, list):
|
||
raise ValueError("预期指标应为数组，如 [{\"type\":\"论文\",\"target\":3,\"unit\":\"篇\"}]")
|
||
values["expected_json"] = json.dumps(expected, ensure_ascii=False)
|
||
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT id FROM closing_projects WHERE project_no = ?", (project_no,))
|
||
row = cur.fetchone()
|
||
if row:
|
||
project_id = int(row[0])
|
||
set_clause = ", ".join(f"{k} = ?" for k in values if k != "project_no")
|
||
cur.execute(
|
||
f"UPDATE closing_projects SET {set_clause}, updated_at = ? WHERE id = ?",
|
||
[values[k] for k in values if k != "project_no"] + [now_unix(), project_id],
|
||
)
|
||
action = "updated"
|
||
else:
|
||
columns = ", ".join(values.keys())
|
||
placeholders = ", ".join("?" for _ in values)
|
||
cur.execute(
|
||
f"INSERT INTO closing_projects ({columns}, created_at, updated_at) VALUES ({placeholders}, ?, ?)",
|
||
list(values.values()) + [now_unix(), now_unix()],
|
||
)
|
||
project_id = int(cur.lastrowid)
|
||
action = "created"
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
log = _log()
|
||
if log:
|
||
log.info("closing_project_upserted project_no=%s action=%s stage=ledger", project_no, action)
|
||
return {"ok": True, "project_no": project_no, "id": project_id, "action": action}
|
||
|
||
|
||
def upsert_achievement(fields: dict[str, Any]) -> dict[str, Any]:
|
||
achievement = {_snake(k): v for k, v in _norm_achievement(fields).items()}
|
||
project_no = str(achievement.get("project_no") or "").strip()
|
||
title = str(achievement.get("title") or "").strip()
|
||
ach_type = str(achievement.get("achievement_type") or "").strip()
|
||
if not project_no:
|
||
raise ValueError("项目编号（project_no）不能为空")
|
||
if not ach_type:
|
||
raise ValueError("成果类型（achievement_type）不能为空")
|
||
if ach_type not in _KNOWN_ACH_TYPES:
|
||
raise ValueError(f"成果类型只接受 {'、'.join(_KNOWN_ACH_TYPES)}：{ach_type}")
|
||
if not title:
|
||
raise ValueError("成果名称（title）不能为空")
|
||
|
||
values = {
|
||
"project_no": project_no,
|
||
"achievement_type": ach_type,
|
||
"title": title,
|
||
"venue_or_no": str(achievement.get("venue_or_no") or "").strip(),
|
||
"ach_date": str(achievement.get("ach_date") or "").strip(),
|
||
"is_representative": 1 if str(achievement.get("is_representative") or "").strip() in ("1", "true", "True", "是") else 0,
|
||
"note": str(achievement.get("note") or "").strip(),
|
||
}
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT id FROM closing_projects WHERE project_no = ?", (project_no,))
|
||
if not cur.fetchone():
|
||
raise ValueError(f"项目不存在，请先录入结题项目：{project_no}")
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO achievements
|
||
(project_no, achievement_type, title, venue_or_no, ach_date, is_representative, note, created_at, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
list(values.values()) + [now_unix(), now_unix()],
|
||
)
|
||
achievement_id = int(cur.lastrowid)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
log = _log()
|
||
if log:
|
||
log.info("achievement_added project_no=%s type=%s stage=ledger", project_no, ach_type)
|
||
return {"ok": True, "id": achievement_id, "project_no": project_no, "title": title}
|
||
|
||
|
||
def upsert_expense(fields: dict[str, Any]) -> dict[str, Any]:
|
||
expense = {_snake(k): v for k, v in _norm_expense(fields).items()}
|
||
project_no = str(expense.get("project_no") or "").strip()
|
||
category = str(expense.get("category") or "").strip()
|
||
amount = _to_float(expense.get("amount"))
|
||
if not project_no:
|
||
raise ValueError("项目编号（project_no）不能为空")
|
||
if not category:
|
||
raise ValueError("科目（category）不能为空")
|
||
if amount is None or amount < 0:
|
||
raise ValueError(f"金额必须是非负数字：{fields.get('amount')}")
|
||
|
||
values = {
|
||
"project_no": project_no,
|
||
"category": category,
|
||
"amount": round(amount, 2),
|
||
"occurred_date": str(expense.get("occurred_date") or "").strip(),
|
||
"description": str(expense.get("description") or "").strip(),
|
||
}
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT id FROM closing_projects WHERE project_no = ?", (project_no,))
|
||
if not cur.fetchone():
|
||
raise ValueError(f"项目不存在，请先录入结题项目：{project_no}")
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO expense_items
|
||
(project_no, category, amount, occurred_date, description, created_at, updated_at)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
list(values.values()) + [now_unix(), now_unix()],
|
||
)
|
||
expense_id = int(cur.lastrowid)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
log = _log()
|
||
if log:
|
||
log.info("expense_added project_no=%s category=%s stage=ledger", project_no, category)
|
||
return {"ok": True, "id": expense_id, "project_no": project_no, "category": category, "amount": values["amount"]}
|
||
|
||
|
||
def _import_generic(rows: Any, upsert, label: str) -> dict[str, Any]:
|
||
if not isinstance(rows, list) or not rows:
|
||
raise ValueError(f"导入内容应为非空 JSON 数组（{label}）")
|
||
if len(rows) > 500:
|
||
raise ValueError(f"单次导入条数过多（{len(rows)}），上限 500 条")
|
||
results = []
|
||
for idx, raw in enumerate(rows):
|
||
if not isinstance(raw, dict):
|
||
raise ValueError(f"第 {idx + 1} 条不是 JSON 对象")
|
||
results.append(upsert(raw))
|
||
return {
|
||
"ok": True,
|
||
"count": len(results),
|
||
"created": sum(1 for r in results if r["action"] == "created"),
|
||
"updated": sum(1 for r in results if r["action"] == "updated"),
|
||
"items": results,
|
||
}
|
||
|
||
|
||
def import_projects(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
||
return _import_generic(rows, lambda row: upsert_project({_snake(k): v for k, v in _norm_project(row).items()}), "结题项目")
|
||
|
||
|
||
def import_achievements(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
||
return _import_generic(rows, lambda row: upsert_achievement({_snake(k): v for k, v in _norm_achievement(row).items()}), "成果清单")
|
||
|
||
|
||
def import_expenses(rows: list[dict[str, Any]]) -> dict[str, Any]:
|
||
return _import_generic(rows, lambda row: upsert_expense({_snake(k): v for k, v in _norm_expense(row).items()}), "决算明细")
|
||
|
||
|
||
def list_projects(status: str = "") -> list[dict[str, Any]]:
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
if status:
|
||
cur.execute("SELECT * FROM closing_projects WHERE status = ? ORDER BY id", (status,))
|
||
else:
|
||
cur.execute("SELECT * FROM closing_projects ORDER BY id")
|
||
columns = [d[0] for d in cur.description]
|
||
rows = [dict(zip(columns, row)) for row in cur.fetchall()]
|
||
finally:
|
||
conn.close()
|
||
return rows
|
||
|
||
|
||
def list_achievements(project_no: str) -> list[dict[str, Any]]:
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT * FROM achievements WHERE project_no = ? ORDER BY id", (project_no,))
|
||
columns = [d[0] for d in cur.description]
|
||
rows = [dict(zip(columns, row)) for row in cur.fetchall()]
|
||
finally:
|
||
conn.close()
|
||
return rows
|
||
|
||
|
||
def list_expenses(project_no: str) -> list[dict[str, Any]]:
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT * FROM expense_items WHERE project_no = ? ORDER BY id", (project_no,))
|
||
columns = [d[0] for d in cur.description]
|
||
rows = [dict(zip(columns, row)) for row in cur.fetchall()]
|
||
finally:
|
||
conn.close()
|
||
return rows
|
||
|
||
|
||
def list_drafts(limit: int = 20) -> list[dict[str, Any]]:
|
||
init_db()
|
||
limit = max(1, min(int(limit), 200))
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"SELECT id, batch_no, project_no, project_name, draft_file, preview_file, created_at FROM report_drafts ORDER BY id DESC LIMIT ?",
|
||
(limit,),
|
||
)
|
||
columns = [d[0] for d in cur.description]
|
||
rows = [dict(zip(columns, row)) for row in cur.fetchall()]
|
||
finally:
|
||
conn.close()
|
||
return rows
|
||
|
||
|
||
def get_draft(draft_id: int) -> dict[str, Any]:
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT * FROM report_drafts WHERE id = ?", (draft_id,))
|
||
row = cur.fetchone()
|
||
if not row:
|
||
raise ValueError(f"初稿记录不存在：{draft_id}")
|
||
columns = [d[0] for d in cur.description]
|
||
draft = dict(zip(columns, row))
|
||
finally:
|
||
conn.close()
|
||
for key in ("completion_json", "narrative_points_json"):
|
||
try:
|
||
draft[key[:-5]] = json.loads(draft.get(key) or "[]")
|
||
except (TypeError, ValueError):
|
||
draft[key[:-5]] = []
|
||
return draft
|
||
|
||
|
||
def stats() -> dict[str, Any]:
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT status, COUNT(*) FROM closing_projects GROUP BY status")
|
||
by_status = {str(r[0]): int(r[1]) for r in cur.fetchall()}
|
||
cur.execute("SELECT COUNT(*) FROM achievements")
|
||
achievement_total = int(cur.fetchone()[0])
|
||
cur.execute("SELECT COUNT(*) FROM expense_items")
|
||
expense_total = int(cur.fetchone()[0])
|
||
cur.execute("SELECT COUNT(*) FROM report_drafts")
|
||
draft_total = int(cur.fetchone()[0])
|
||
finally:
|
||
conn.close()
|
||
return {
|
||
"ok": True,
|
||
"projects_by_status": by_status,
|
||
"achievement_total": achievement_total,
|
||
"expense_total": expense_total,
|
||
"draft_total": draft_total,
|
||
}
|