Files
twoJob-demo/scripts/gen_spec.py
T

247 lines
9.6 KiB
Python
Raw Normal View History

2026-09-03 15:35:23 +08:00
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
任务4: 生成结构化开发规格 docs/spec.json (机器可直接消费)。
把审核决策(features.keep/note) + 完整字段字典(field_specs) + 隐藏入口(page_extras)
+ 真实导出样本(downloads, 含表头/行列预览) + 审批模板库(approval_templates)
汇总为一份"按模块->功能页"组织的结构化规格, 供下一步开发直接读取。
输出结构:
{
meta, stats,
modules: [ { module, page_count, pages: [ {
node_key, name, group, full_path, url, keep,
description, understanding, review_note, screenshot,
field_count, fields:[{group,name,order}],
hidden_entries:[...], # "更多"菜单发现的隐藏功能界面
data_samples:[{file,kind,size,sheet,headers,preview_rows,total_rows}]
} ] } ],
approval_templates: [ {category,name,provider,description} ]
}
"""
import os, json, sqlite3, time, re, collections, warnings
BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DB = os.path.join(BASE, "db", "research.db")
OUT = os.path.join(BASE, "docs", "spec.json")
warnings.filterwarnings("ignore")
def db():
c = sqlite3.connect(DB); c.row_factory = sqlite3.Row; return c
def excel_preview(relpath, max_scan=6, max_head=40):
"""读导出文件表头 + 前几行预览。.xls 用 xlrd, .xlsx 用 openpyxl。"""
p = os.path.join(BASE, relpath)
if not os.path.isfile(p):
return None
low = p.lower()
try:
if low.endswith(".xls"):
import xlrd
wb = xlrd.open_workbook(p); ws = wb.sheet_by_index(0)
rows = []
for r in range(min(max_scan, ws.nrows)):
rows.append([str(ws.cell_value(r, c)).strip() for c in range(min(max_head, ws.ncols))])
return {"sheet": ws.name, "total_rows": ws.nrows, "total_cols": ws.ncols, "rows": rows}
if low.endswith(".xlsx"):
import openpyxl
wb = openpyxl.load_workbook(p, read_only=True, data_only=True); ws = wb.active
rows = []
for i, row in enumerate(ws.iter_rows(values_only=True)):
if i >= max_scan:
break
rows.append([("" if v is None else str(v)).strip() for v in row[:max_head]])
r = {"sheet": ws.title, "total_rows": ws.max_row, "total_cols": ws.max_column, "rows": rows}
wb.close(); return r
except Exception as e:
return {"error": str(e)[:120]}
return None
def best_header(rows):
"""从前几行里挑字段数最多的一行作为表头(过滤单格标题行)。"""
if not rows:
return []
best, cnt = [], 0
for r in rows:
vals = [v for v in r if v]
if len(vals) > cnt:
best, cnt = vals, len(vals)
return [] if cnt <= 1 else best
def load_leaves():
c = db()
rows = [dict(r) for r in c.execute(
"SELECT node_key,parent_key,level,name,full_path,url,description,"
"understanding,screenshot_path,keep,note,sort_order FROM features ORDER BY sort_order,id")]
c.close()
by_parent = {}
for r in rows:
by_parent.setdefault(r["parent_key"] or "", []).append(r)
leaves = []
def dfs(pk, module, group):
for r in by_parent.get(pk, []):
m = r["name"] if r["level"] == 0 else module
g = r["name"] if r["level"] == 1 else group
kids = by_parent.get(r["node_key"])
if kids:
dfs(r["node_key"], m, g)
else:
r["_module"] = m
r["_group"] = g if g and g != r["name"] else ""
leaves.append(r)
dfs("", "", "")
return leaves
def load_fields():
c = db(); out = collections.defaultdict(list)
try:
for r in c.execute("SELECT node_key,field,field_group,source,sort_order "
"FROM field_specs ORDER BY node_key,sort_order"):
out[r["node_key"]].append({"group": r["field_group"] or "", "name": r["field"],
"order": r["sort_order"], "source": r["source"]})
except sqlite3.OperationalError:
pass
c.close(); return out
def load_extras():
c = db(); out = collections.defaultdict(list)
try:
for r in c.execute("SELECT node_key,label FROM page_extras ORDER BY node_key,id"):
out[r["node_key"]].append(r["label"])
except sqlite3.OperationalError:
pass
c.close(); return out
def load_downloads():
c = db(); out = collections.defaultdict(list)
try:
for r in c.execute("SELECT node_key,file_path,kind,size FROM downloads WHERE status='ok' OR status IS NULL"):
out[r["node_key"]].append(dict(r))
except sqlite3.OperationalError:
pass
c.close(); return out
def load_deep_shots():
"""深层界面截图(非 main): 主操作弹窗/表单(add)、筛选(filter)、自定义列(columns)、行详情(detail)。"""
kind_cn = {"add": "主操作弹窗/表单", "filter": "筛选面板",
"columns": "自定义列/选择表头", "detail": "行详情/编辑"}
c = db(); out = collections.defaultdict(list)
try:
for r in c.execute("SELECT node_key,kind,label,path FROM page_shots "
"WHERE kind<>'main' ORDER BY node_key,id"):
out[r["node_key"]].append({"kind": r["kind"], "kind_cn": kind_cn.get(r["kind"], r["kind"]),
"label": r["label"] or "", "path": r["path"]})
except sqlite3.OperationalError:
pass
c.close(); return out
def load_templates():
c = db(); out = []
try:
for r in c.execute("SELECT category,name,provider,description,sort_order "
"FROM approval_templates ORDER BY sort_order,id"):
out.append({"category": r["category"], "name": r["name"],
"provider": r["provider"] or "", "description": r["description"] or ""})
except sqlite3.OperationalError:
pass
c.close(); return out
def norm(s):
return re.sub(r"^\d+_", "", s or "").strip()
def main():
leaves = load_leaves()
fields = load_fields()
extras = load_extras()
downloads = load_downloads()
deep_shots = load_deep_shots()
templates = load_templates()
modules = collections.OrderedDict()
n_fields_pages = 0
for lf in leaves:
nk = lf["node_key"]
mod = lf["_module"] or "未分类"
fl = fields.get(nk, [])
if fl:
n_fields_pages += 1
samples = []
for d in downloads.get(nk, []):
pv = excel_preview(d["file_path"])
item = {"file": d["file_path"], "kind": d["kind"], "size": d["size"]}
if pv and "rows" in pv:
item.update({"sheet": pv["sheet"], "total_rows": pv["total_rows"],
"total_cols": pv["total_cols"], "headers": best_header(pv["rows"])})
samples.append(item)
page = {
"node_key": nk, "name": lf["name"], "group": lf["_group"],
"full_path": lf["full_path"], "url": lf["url"] or "",
"keep": 1 if (lf["keep"] is None or lf["keep"] == 1) else 0,
"description": lf["description"] or "",
"understanding": lf["understanding"] or "",
"review_note": lf["note"] or "",
"screenshot": lf["screenshot_path"] or "",
"field_count": len(fl),
"fields": [{"group": f["group"], "name": f["name"], "order": f["order"]} for f in fl],
"hidden_entries": extras.get(nk, []),
"deep_shots": deep_shots.get(nk, []),
"data_samples": samples,
}
modules.setdefault(mod, []).append(page)
mod_list = [{"module": m, "page_count": len(ps),
"field_pages": sum(1 for p in ps if p["field_count"]),
"pages": ps} for m, ps in modules.items()]
total_fields = sum(len(v) for v in fields.values())
total_samples = sum(len(v) for v in downloads.values())
spec = {
"meta": {
"product": "2号人事部",
"source": "https://2haohr.com 功能调研",
"generated_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
"schema_version": 1,
"notes": "按 模块->功能页 组织; fields 为自定义列/选择表头弹窗采集的完整字段(远超截图);"
" hidden_entries 为'更多'菜单发现的隐藏功能界面; data_samples 为真实导出样本(含表头预览);"
" screenshot 为去遮挡后重拍的整页主图; deep_shots 为点击按钮抵达的深层界面截图(新增表单/筛选/自定义列/行详情)。",
},
"stats": {
"modules": len(mod_list),
"pages": len(leaves),
"kept_pages": sum(1 for lf in leaves if (lf["keep"] is None or lf["keep"] == 1)),
"pages_with_fields": n_fields_pages,
"total_fields": total_fields,
"pages_with_hidden_entries": sum(1 for k in extras if extras[k]),
"total_hidden_entries": sum(len(v) for v in extras.values()),
"data_samples": total_samples,
"pages_with_deep_shots": sum(1 for k in deep_shots if deep_shots[k]),
"total_deep_shots": sum(len(v) for v in deep_shots.values()),
"approval_templates": len(templates),
},
"modules": mod_list,
"approval_templates": templates,
}
os.makedirs(os.path.dirname(OUT), exist_ok=True)
with open(OUT, "w", encoding="utf-8") as f:
json.dump(spec, f, ensure_ascii=False, indent=2)
size = os.path.getsize(OUT)
print("生成 docs/spec.json (%.1f KB)" % (size / 1024))
print("统计:", json.dumps(spec["stats"], ensure_ascii=False))
if __name__ == "__main__":
main()