From 6f849cbb36dbce3789fabc4a01d40e35825972fc Mon Sep 17 00:00:00 2001
From: zlwu <704541015@qq.com>
Date: Tue, 11 Aug 2026 11:57:35 +0800
Subject: [PATCH] chore: add OA published form type stats script against
 zkxw-sc

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 scripts/oa_form_type_stats.py | 91 +++++++++++++++++++++++++++++++++++
 1 file changed, 91 insertions(+)
 create mode 100644 scripts/oa_form_type_stats.py

diff --git a/scripts/oa_form_type_stats.py b/scripts/oa_form_type_stats.py
new file mode 100644
index 00000000..ec023ac1
--- /dev/null
+++ b/scripts/oa_form_type_stats.py
@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+"""统计 zkxw-sc（或当前 DATABASE_URL）已发布 OA 模板 form_schema 中的 type。"""
+from __future__ import annotations
+
+import json
+import sys
+from collections import Counter
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "backend"))
+
+from sqlalchemy import create_engine, text  # noqa: E402
+from app.core.config import settings  # noqa: E402
+
+CUSTOM = {
+    "money",
+    "person",
+    "ownerPerson",
+    "dept",
+    "ownerDept",
+    "post",
+    "receiveAccount",
+    "relatedApproval",
+    "upload",
+    "fcUpload",
+    "leaveKit",
+    "overtimeKit",
+    "confirmationKit",
+    "resignationKit",
+    "tripKit",
+    "tripItinerary",
+}
+
+
+def walk_types(node, out: list) -> None:
+    if node is None:
+        return
+    if isinstance(node, dict):
+        t = node.get("type")
+        if isinstance(t, str) and t.strip():
+            out.append(t.strip())
+        for v in node.values():
+            if isinstance(v, (dict, list)):
+                walk_types(v, out)
+    elif isinstance(node, list):
+        for item in node:
+            walk_types(item, out)
+
+
+def main() -> None:
+    engine = create_engine(settings.DATABASE_URL)
+    sql = text(
+        """
+        SELECT a.app_name, v.form_schema_json
+        FROM oa_app a
+        JOIN oa_app_version v ON v.app_id = a.id
+        WHERE a.del_flag = 0 AND v.del_flag = 0
+          AND v.status = 'published'
+          AND (v.is_latest_published = 1 OR v.is_latest_published = '1')
+        ORDER BY a.app_name
+        """
+    )
+    counter: Counter[str] = Counter()
+    with engine.connect() as conn:
+        rows = conn.execute(sql).mappings().all()
+    print(f"published_latest_apps={len(rows)}")
+    for row in rows:
+        raw = row["form_schema_json"] or ""
+        try:
+            schema = json.loads(raw) if raw else {}
+        except json.JSONDecodeError:
+            print("BAD_JSON", row["app_name"])
+            continue
+        types: list[str] = []
+        if isinstance(schema, list):
+            walk_types(schema, types)
+        else:
+            walk_types(schema.get("rule", schema), types)
+        for t in types:
+            counter[t] += 1
+    print("=== frequency ===")
+    for t, c in counter.most_common():
+        kind = "CUSTOM" if t in CUSTOM else "base"
+        print(f"{c:4d}  {kind:6s}  {t}")
+    print("=== custom used ===")
+    print(sorted(set(counter) & CUSTOM))
+
+
+if __name__ == "__main__":
+    main()