91 lines
4.6 KiB
Python
91 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
||
"""规范机械自检：目录结构 / models 四段式 / CRUD 根键 / 三处同步 / 无硬编码库名 / i18n / SQL 方言。"""
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
PKG = os.path.join(ROOT, 'pbl_blueprint')
|
||
sys.path.insert(0, ROOT)
|
||
FAIL = []
|
||
|
||
|
||
def chk(cond, msg):
|
||
print(('PASS ' if cond else 'FAIL ') + msg)
|
||
if not cond:
|
||
FAIL.append(msg)
|
||
|
||
|
||
# 1 目录结构
|
||
for must in ['pyproject.toml', 'scripts/load_path.py', 'skill/SKILL.md', 'tests',
|
||
os.path.join('pbl_blueprint', '__init__.py'), os.path.join('pbl_blueprint', 'init.py'),
|
||
os.path.join('pbl_blueprint', 'models'), os.path.join('pbl_blueprint', 'json'),
|
||
os.path.join('pbl_blueprint', 'wwwroot'), os.path.join('pbl_blueprint', 'i18n')]:
|
||
chk(os.path.exists(os.path.join(ROOT, must)), '目录结构存在 %s' % must)
|
||
for stray in ['models.py', 'db.py', 'blueprint_crud.py', 'subobjects.py', '__init__.py']:
|
||
chk(not os.path.exists(os.path.join(ROOT, stray)), '仓库根无游离包文件 %s' % stray)
|
||
|
||
# 2 models 四段式 + tenant 打头
|
||
from pbl_blueprint import models as M
|
||
for t in M.table_names():
|
||
chk(M.validate_model(t) == [], 'models/%s.json 四段式自检: %s' % (t, M.validate_model(t) or 'ok'))
|
||
|
||
# 3 CRUD json 根键
|
||
for f in sorted(os.listdir(os.path.join(PKG, 'json'))):
|
||
d = json.load(open(os.path.join(PKG, 'json', f), encoding='utf-8'))
|
||
chk('tblname' in d and 'params' in d and 'table' not in d and 'list' not in d,
|
||
'CRUD %s 根键 tblname+params' % f)
|
||
chk(d.get('tblname') in M.table_names(), 'CRUD %s tblname 有对应 models 定义' % f)
|
||
for key in ('browserfields', 'editexclouded', 'edit_exclouded_fields'):
|
||
for fld in (d['params'].get('browserfields', {}).get('alters', {}) if key == 'browserfields'
|
||
else d['params'].get(key, []) if not isinstance(d['params'].get(key), dict) else []):
|
||
chk(fld in M.fields_of(d['tblname']), 'CRUD %s 字段 %s 存在于表定义' % (f, fld))
|
||
|
||
# 4 三处同步
|
||
import pbl_blueprint as P
|
||
from pbl_blueprint import init as I
|
||
impl = set()
|
||
for mod in ['blueprint_crud', 'subobjects', 'templates']:
|
||
m = __import__('pbl_blueprint.' + mod, fromlist=['x'])
|
||
impl |= {n for n in dir(m) if re.match(r'^(blueprint|subobject|template)_\w+$', n) and callable(getattr(m, n))}
|
||
exported = {n for n in dir(P) if re.match(r'^(blueprint|subobject|template)_\w+$', n)}
|
||
src = open(os.path.join(PKG, 'init.py'), encoding='utf-8').read()
|
||
registered = set(re.findall(r'env\.(\w+)\s*=', src)) | {n for n in I.CONTRACT_FUNCS.__iter__()} if False else set(re.findall(r'env\.(\w+)\s*=', src)) | {f.__name__ for f in I.CONTRACT_FUNCS}
|
||
chk(impl == exported, '①定义==②导出: 差集 %s' % (impl ^ exported))
|
||
chk(impl <= registered, '①定义⊆③init注册: 缺 %s' % (impl - registered))
|
||
|
||
# 5 无硬编码 DB 名
|
||
for dirpath, _, names in os.walk(PKG):
|
||
for n in names:
|
||
if n.endswith('.py'):
|
||
text = open(os.path.join(dirpath, n), encoding='utf-8').read()
|
||
chk(not re.search(r"(?i)^\s*DBNAME\s*=", text, re.M), '%s 无硬编码 DBNAME' % n)
|
||
chk('get_module_dbname' in text or 'db.py' == n or n not in ('blueprint_crud.py', 'subobjects.py', 'templates.py', '__init__.py'),
|
||
'%s 库名走 get_module_dbname 链路' % n)
|
||
|
||
# 6 i18n 覆盖接口 message 键
|
||
i18n = json.load(open(os.path.join(PKG, 'i18n', 'pbl_blueprint.zh_CN.json'), encoding='utf-8'))
|
||
keys = set()
|
||
for mod in ['blueprint_crud', 'subobjects', 'templates']:
|
||
text = open(os.path.join(PKG, mod + '.py'), encoding='utf-8').read()
|
||
keys |= set(re.findall(r"_err\('([\w.]+)'", text))
|
||
chk(keys <= set(i18n), 'i18n 覆盖全部 message 键, 缺 %s' % (keys - set(i18n)))
|
||
en = json.load(open(os.path.join(PKG, 'i18n', 'pbl_blueprint.en_US.json'), encoding='utf-8'))
|
||
chk(set(en) == set(i18n), 'zh_CN/en_US 键一致')
|
||
|
||
# 7 SQL 方言禁项 + engine 声明
|
||
sql = open(os.path.join(PKG, 'sql', 'pbl_blueprint.core.sql'), encoding='utf-8').read()
|
||
chk(not re.search(r'(?i)\b(bigserial|serial|nextval)\b', sql), 'SQL 无 BIGSERIAL/SERIAL/nextval')
|
||
chk(re.search(r'(?m)^--\s*engine:', sql) is not None, 'SQL 声明 engine')
|
||
chk(not re.search(r'\bboolean\b|\bint\(11\)\b', sql), 'SQL 无 boolean/int(11)')
|
||
|
||
# 8 dspy 无 import/f-string/print/uuid（复用 audit）
|
||
sys.path.insert(0, os.path.join(ROOT, 'scripts'))
|
||
import audit_dspy
|
||
files, problems = audit_dspy.audit()
|
||
chk(not problems, 'dspy 审计零命中（%d 文件）: %s' % (len(files), problems[:3]))
|
||
|
||
print('SELFCHECK_%s (%d fails)' % ('PASS' if not FAIL else 'FAIL', len(FAIL)))
|
||
sys.exit(1 if FAIL else 0)
|