323 lines
12 KiB
Python
323 lines
12 KiB
Python
"""项目台账维护：录入（upsert）、批量导入、查询。
|
||
|
||
台账是播报的唯一数据源；金额单位元，日期 YYYY-MM-DD。
|
||
project_no 唯一，重复导入按项目编号幂等更新（不覆盖未提供字段）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any
|
||
|
||
from db.connection import get_conn, init_db
|
||
from service.fund_service import _norm_project, _parse_date, _to_float
|
||
from util.logging_config import get_skill_logger
|
||
from util.timeutil import now_unix
|
||
|
||
_DATE_FIELDS = ("frozen_since", "start_date", "end_date")
|
||
_NUMERIC_FIELDS = ("total_approved", "total_received", "total_spent", "frozen_amount")
|
||
_TEXT_FIELDS = ("project_name", "project_type", "pi_name", "department", "status", "remark")
|
||
|
||
|
||
def _log():
|
||
try:
|
||
return get_skill_logger()
|
||
except RuntimeError:
|
||
return None
|
||
|
||
|
||
def _snake(name: str) -> str:
|
||
"""camelCase → snake_case（_norm_project 输出 canonical camelCase 键）。"""
|
||
return "".join("_" + ch.lower() if ch.isupper() else ch for ch in str(name)).lstrip("_")
|
||
|
||
|
||
def upsert_project(fields: dict[str, Any]) -> dict[str, Any]:
|
||
"""按 project_no 幂等写入台账；未提供的字段不覆盖已有值。"""
|
||
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 _NUMERIC_FIELDS:
|
||
if key in fields and fields[key] not in (None, ""):
|
||
number = _to_float(fields[key])
|
||
if number is None or number < 0:
|
||
raise ValueError(f"字段 {key} 必须是非负数字：{fields[key]}")
|
||
values[key] = number
|
||
for key in _DATE_FIELDS:
|
||
if key in fields and fields[key] not in (None, ""):
|
||
parsed = _parse_date(fields[key])
|
||
if parsed is None:
|
||
raise ValueError(f"字段 {key} 日期格式无效（应为 YYYY-MM-DD）：{fields[key]}")
|
||
values[key] = parsed.isoformat()
|
||
for key in _TEXT_FIELDS:
|
||
if key in fields and fields[key] is not None:
|
||
values[key] = str(fields[key]).strip()
|
||
if values.get("status") and values["status"] not in ("在研", "已结题", "暂缓"):
|
||
raise ValueError(f"status 只接受 在研/已结题/暂缓：{values['status']}")
|
||
if values.get("start_date") and values.get("end_date") and values["end_date"] <= values["start_date"]:
|
||
raise ValueError("执行期结束日期必须晚于开始日期")
|
||
|
||
init_db()
|
||
log = _log()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT id FROM fund_projects WHERE project_no = ?", (project_no,))
|
||
row = cur.fetchone()
|
||
if row:
|
||
project_id = int(row[0])
|
||
set_clause = ", ".join(f"{key} = ?" for key in values if key != "project_no")
|
||
params = [values[key] for key in values if key != "project_no"] + [now_unix(), project_id]
|
||
cur.execute(
|
||
f"UPDATE fund_projects SET {set_clause}, updated_at = ? WHERE id = ?",
|
||
params,
|
||
)
|
||
action = "updated"
|
||
else:
|
||
columns = ", ".join(values.keys())
|
||
placeholders = ", ".join("?" for _ in values)
|
||
cur.execute(
|
||
f"INSERT INTO fund_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()
|
||
if log:
|
||
log.info("project_upserted project_no=%s action=%s stage=project", project_no, action)
|
||
return {"ok": True, "project_no": project_no, "id": project_id, "action": action}
|
||
|
||
|
||
def import_projects(projects_raw: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""批量导入台账（upsert），返回统计。"""
|
||
if not isinstance(projects_raw, list) or not projects_raw:
|
||
raise ValueError("导入内容应为非空 JSON 数组")
|
||
if len(projects_raw) > 500:
|
||
raise ValueError(f"单次导入项目数过多（{len(projects_raw)}），上限 500 个")
|
||
results = []
|
||
for idx, raw in enumerate(projects_raw):
|
||
if not isinstance(raw, dict):
|
||
raise ValueError(f"第 {idx + 1} 个项目不是 JSON 对象")
|
||
normalized = {_snake(canonical): value for canonical, value in _norm_project(raw).items()}
|
||
results.append(upsert_project(normalized))
|
||
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"),
|
||
"projects": results,
|
||
}
|
||
|
||
|
||
def list_projects(pi_name: str = "", status: str = "") -> list[dict[str, Any]]:
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
sql = "SELECT * FROM fund_projects WHERE 1=1"
|
||
params: list[Any] = []
|
||
if pi_name:
|
||
sql += " AND pi_name LIKE ?"
|
||
params.append(f"%{pi_name}%")
|
||
if status:
|
||
sql += " AND status = ?"
|
||
params.append(status)
|
||
sql += " ORDER BY id"
|
||
cur.execute(sql, params)
|
||
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_project_by_id(project_id: int) -> dict[str, Any]:
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT * FROM fund_projects WHERE id = ?", (project_id,))
|
||
row = cur.fetchone()
|
||
if not row:
|
||
raise ValueError(f"项目不存在：{project_id}")
|
||
columns = [d[0] for d in cur.description]
|
||
return dict(zip(columns, row))
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def parse_projects_json(raw: str) -> list[dict[str, Any]]:
|
||
"""解析导入 JSON：'-'（stdin）/ 文件路径 / 内联数组或 {"projects": [...]} 包装。"""
|
||
import os
|
||
import sys
|
||
|
||
text = (raw or "").strip()
|
||
if not text:
|
||
raise ValueError("缺少导入 JSON（--projects-json）")
|
||
if text == "-":
|
||
text = sys.stdin.read().strip()
|
||
elif text.startswith(("[", "{")):
|
||
pass
|
||
elif os.path.isfile(text):
|
||
with open(text, encoding="utf-8") as f:
|
||
text = f.read().strip()
|
||
else:
|
||
raise ValueError(f"--projects-json 不是有效文件路径或 JSON：{text[:80]}")
|
||
try:
|
||
payload = json.loads(text)
|
||
except json.JSONDecodeError as exc:
|
||
raise ValueError(f"导入 JSON 解析失败：{exc}") from exc
|
||
if isinstance(payload, dict) and isinstance(payload.get("projects"), list):
|
||
payload = payload["projects"]
|
||
if not isinstance(payload, list):
|
||
raise ValueError("导入 JSON 应为对象数组或 {\"projects\": [...]}")
|
||
return payload
|
||
|
||
|
||
def list_reports(limit: int = 20, status_tag: str = "") -> list[dict[str, Any]]:
|
||
init_db()
|
||
limit = max(1, min(int(limit), 200))
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
if status_tag:
|
||
cur.execute(
|
||
"""
|
||
SELECT id, batch_no, as_of, project_no, project_name, pi_name,
|
||
spend_rate, time_progress, deviation, status_tag, summary, report_file, created_at
|
||
FROM progress_reports WHERE status_tag = ? ORDER BY id DESC LIMIT ?
|
||
""",
|
||
(status_tag, limit),
|
||
)
|
||
else:
|
||
cur.execute(
|
||
"""
|
||
SELECT id, batch_no, as_of, project_no, project_name, pi_name,
|
||
spend_rate, time_progress, deviation, status_tag, summary, report_file, created_at
|
||
FROM progress_reports 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_report(report_id: int) -> dict[str, Any]:
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT * FROM progress_reports WHERE id = ?", (report_id,))
|
||
row = cur.fetchone()
|
||
if not row:
|
||
raise ValueError(f"播报记录不存在：{report_id}")
|
||
columns = [d[0] for d in cur.description]
|
||
report = dict(zip(columns, row))
|
||
finally:
|
||
conn.close()
|
||
try:
|
||
report["suggestions"] = json.loads(report.get("suggestions_json") or "[]")
|
||
except (TypeError, ValueError):
|
||
report["suggestions"] = []
|
||
return report
|
||
|
||
|
||
def view_rules() -> list[dict[str, Any]]:
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"SELECT id, rule_key, display_name, threshold_value, description, updated_at FROM alert_rules 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
|
||
|
||
|
||
_KNOWN_RULE_KEYS = ("lag_threshold", "overshoot_threshold", "pending_aging_days", "closing_warn_days")
|
||
|
||
|
||
def set_rule(rule_key: str, threshold_value: float) -> dict[str, Any]:
|
||
"""按 rule_key 更新阈值（须为已知规则键，数值须为正）。"""
|
||
rule_key = (rule_key or "").strip()
|
||
if rule_key not in _KNOWN_RULE_KEYS:
|
||
raise ValueError(f"未知规则键：{rule_key}（可选：{'、'.join(_KNOWN_RULE_KEYS)}）")
|
||
threshold_value = float(threshold_value)
|
||
if threshold_value <= 0 or threshold_value >= 1.5:
|
||
raise ValueError(f"阈值必须是 0～1.5 之间的正数：{threshold_value}")
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT id FROM alert_rules WHERE rule_key = ?", (rule_key,))
|
||
row = cur.fetchone()
|
||
if not row:
|
||
raise ValueError(f"规则键尚未初始化：{rule_key}（请先执行 init-db）")
|
||
cur.execute(
|
||
"UPDATE alert_rules SET threshold_value = ?, updated_at = ? WHERE id = ?",
|
||
(threshold_value, now_unix(), int(row[0])),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
log = _log()
|
||
if log:
|
||
log.info("rule_updated rule_key=%s value=%s stage=rules", rule_key, threshold_value)
|
||
return {"ok": True, "rule_key": rule_key, "threshold_value": threshold_value}
|
||
|
||
|
||
def stats() -> dict[str, Any]:
|
||
init_db()
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT COUNT(*) FROM fund_projects")
|
||
project_total = int(cur.fetchone()[0])
|
||
cur.execute("SELECT status, COUNT(*) FROM fund_projects GROUP BY status")
|
||
by_status = {str(r[0]): int(r[1]) for r in cur.fetchall()}
|
||
cur.execute("SELECT status_tag, COUNT(*) FROM progress_reports GROUP BY status_tag")
|
||
reports_by_tag = {str(r[0]): int(r[1]) for r in cur.fetchall()}
|
||
cur.execute("SELECT COUNT(*) FROM progress_reports")
|
||
report_total = int(cur.fetchone()[0])
|
||
cur.execute(
|
||
"""
|
||
SELECT project_no, project_name, pi_name, status_tag, spend_rate, time_progress, as_of
|
||
FROM progress_reports
|
||
WHERE status_tag = '严重滞后'
|
||
ORDER BY deviation ASC LIMIT 5
|
||
"""
|
||
)
|
||
lag_top = [
|
||
{
|
||
"project_no": r[0],
|
||
"project_name": r[1],
|
||
"pi_name": r[2],
|
||
"status_tag": r[3],
|
||
"spend_rate": r[4],
|
||
"time_progress": r[5],
|
||
"as_of": r[6],
|
||
}
|
||
for r in cur.fetchall()
|
||
]
|
||
cur.execute("SELECT COALESCE(SUM(frozen_amount), 0) FROM fund_projects WHERE status = '在研'")
|
||
frozen_total = float(cur.fetchone()[0])
|
||
finally:
|
||
conn.close()
|
||
return {
|
||
"ok": True,
|
||
"project_total": project_total,
|
||
"projects_by_status": by_status,
|
||
"report_total": report_total,
|
||
"reports_by_tag": reports_by_tag,
|
||
"lag_top": lag_top,
|
||
"frozen_total": frozen_total,
|
||
}
|