chore: add OA published form type stats script against zkxw-sc
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
1 parent
d92472d2cd
commit
6f849cbb36
1 file changed
+91
@@ -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()
|
||||
Reference in new issue
Block a user