把"FDE 到底干什么"落成一张表：交付工单系统与考核指标实战

CREATE TABLE work_orders (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    industry        TEXT NOT NULL,          -- 行业：law / ecommerce / manufacturing
    req_source      TEXT NOT NULL,          -- 需求来源：manual(客户提出) / mined(现场挖掘)
    req_text        TEXT NOT NULL,          -- 需求原始描述
    clarify_rounds  INTEGER NOT NULL DEFAULT 0,  -- 需求澄清轮次
    dev_hours       REAL NOT NULL,          -- 开发工时
    deploy_hours    REAL NOT NULL,          -- 部署联调工时
    first_pass      INTEGER NOT NULL,       -- 是否一次验收通过：0/1
    revise_rounds   INTEGER NOT NULL DEFAULT 0,  -- 验收后修改轮次
    biz_metric_name TEXT,                   -- 业务指标名，如"单证录入时长"
    biz_before      REAL,                   -- 改造前数值
    biz_after       REAL,                   -- 改造后数值
    created_at      TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

#!/usr/bin/env python3
"""FDE 交付工单台账：录入 + 指标计算。仅依赖标准库。"""
import sqlite3
import sys

SCHEMA = open("schema.sql", encoding="utf-8").read()


def connect(db="ledger.db"):
    conn = sqlite3.connect(db)
    conn.executescript(SCHEMA)
    return conn


def add_order(conn, row: dict):
    cols = ("industry, req_source, req_text, clarify_rounds, dev_hours,"
            " deploy_hours, first_pass, revise_rounds,"
            " biz_metric_name, biz_before, biz_after")
    conn.execute(
        f"INSERT INTO work_orders ({cols}) VALUES (?,?,?,?,?,?,?,?,?,?,?)",
        (row["industry"], row["req_source"], row["req_text"],
         row["clarify_rounds"], row["dev_hours"], row["deploy_hours"],
         row["first_pass"], row["revise_rounds"],
         row.get("biz_metric_name"), row.get("biz_before"),
         row.get("biz_after")))
    conn.commit()


def metrics(conn) -> dict:
    cur = conn.cursor()
    total, mined = cur.execute(
        "SELECT COUNT(*), SUM(req_source='mined') FROM work_orders").fetchone()
    avg_clarify = cur.execute(
        "SELECT AVG(clarify_rounds) FROM work_orders").fetchone()[0]
    pass_rate = cur.execute(
        "SELECT AVG(first_pass) FROM work_orders").fetchone()[0]
    density = cur.execute(
        """SELECT SUM(biz_before - biz_after) / SUM(dev_hours)
           FROM work_orders
           WHERE biz_before IS NOT NULL AND biz_after IS NOT NULL"""
    ).fetchone()[0]
    return {
        "需求挖掘率": round((mined or 0) / total, 3) if total else None,
        "平均澄清轮次": round(avg_clarify or 0, 2),
        "一次验收通过率": round(pass_rate or 0, 3),
        "交付价值密度": round(density, 3) if density else None,
    }


if __name__ == "__main__":
    conn = connect(sys.argv[1] if len(sys.argv) > 1 else "ledger.db")
    if len(sys.argv) > 2 and sys.argv[2] == "metrics":
        for k, v in metrics(conn).items():
            print(f"{k}: {v}")
    else:
        print("用法: python3 fde_ledger.py [db文件] metrics")

from fde_ledger import connect, add_order, metrics

conn = connect(":memory:")
add_order(conn, dict(industry="law", req_source="manual",
    req_text="合同模板批量生成", clarify_rounds=3, dev_hours=16,
    deploy_hours=8, first_pass=1, revise_rounds=1,
    biz_metric_name="单份合同起草时长(分钟)", biz_before=90, biz_after=25))
add_order(conn, dict(industry="ecommerce", req_source="mined",
    req_text="客服高频问题自动归类", clarify_rounds=1, dev_hours=24,
    deploy_hours=4, first_pass=0, revise_rounds=2,
    biz_metric_name="人工客服日均工单", biz_before=800, biz_after=520))

for k, v in metrics(conn).items():
    print(f"{k}: {v}")

需求挖掘率: 0.5
平均澄清轮次: 2.0
一次验收通过率: 0.5
交付价值密度: 8.625