105 lines
3.1 KiB
Python
105 lines
3.1 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""
|
||
|
|
将现有HTML表单转换为JSON Schema格式
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
||
|
|
from sqlalchemy import text
|
||
|
|
import re
|
||
|
|
import json
|
||
|
|
|
||
|
|
DB_CONFIG = {
|
||
|
|
'host': '123.57.81.127',
|
||
|
|
'port': 3306,
|
||
|
|
'user': 'cpy_admin',
|
||
|
|
'password': 'Tgzz2025+',
|
||
|
|
'database': 'ruoyi-fastapi'
|
||
|
|
}
|
||
|
|
|
||
|
|
DATABASE_URL = (
|
||
|
|
f"mysql+asyncmy://{DB_CONFIG['user']}:{DB_CONFIG['password']}@"
|
||
|
|
f"{DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}"
|
||
|
|
)
|
||
|
|
|
||
|
|
def html_to_json_schema(html):
|
||
|
|
"""将HTML表单转换为JSON Schema"""
|
||
|
|
fields = []
|
||
|
|
|
||
|
|
# 匹配 el-form-item
|
||
|
|
pattern = r'<el-form-item[^>]*label="([^"]*)"[^>]*prop="([^"]*)"[^>]*>(.*?)</el-form-item>'
|
||
|
|
matches = re.findall(pattern, html, re.DOTALL)
|
||
|
|
|
||
|
|
for label, prop, content in matches:
|
||
|
|
field = {
|
||
|
|
'prop': prop,
|
||
|
|
'label': label,
|
||
|
|
'type': 'input',
|
||
|
|
'required': 'required' in content
|
||
|
|
}
|
||
|
|
|
||
|
|
# 判断输入类型
|
||
|
|
if 'el-input-number' in content:
|
||
|
|
field['type'] = 'number'
|
||
|
|
elif 'el-upload' in content:
|
||
|
|
field['type'] = 'upload'
|
||
|
|
elif 'type="textarea"' in content:
|
||
|
|
field['type'] = 'textarea'
|
||
|
|
elif 'show-password' in content:
|
||
|
|
field['type'] = 'password'
|
||
|
|
elif 'el-select' in content:
|
||
|
|
field['type'] = 'select'
|
||
|
|
elif 'el-radio' in content:
|
||
|
|
field['type'] = 'radio'
|
||
|
|
elif 'el-checkbox' in content:
|
||
|
|
field['type'] = 'checkbox'
|
||
|
|
else:
|
||
|
|
field['type'] = 'input'
|
||
|
|
|
||
|
|
# 提取placeholder
|
||
|
|
placeholder_match = re.search(r'placeholder="([^"]*)"', content)
|
||
|
|
if placeholder_match:
|
||
|
|
field['placeholder'] = placeholder_match.group(1)
|
||
|
|
|
||
|
|
# 提取maxlength
|
||
|
|
maxlength_match = re.search(r':maxlength="(\d+)"', content)
|
||
|
|
if maxlength_match:
|
||
|
|
field['maxlength'] = int(maxlength_match.group(1))
|
||
|
|
|
||
|
|
fields.append(field)
|
||
|
|
|
||
|
|
return {'fields': fields}
|
||
|
|
|
||
|
|
async def main():
|
||
|
|
engine = create_async_engine(DATABASE_URL, echo=False)
|
||
|
|
|
||
|
|
async with engine.connect() as conn:
|
||
|
|
# 获取所有表单
|
||
|
|
result = await conn.execute(text("""
|
||
|
|
SELECT id, name, form_json FROM test_form
|
||
|
|
"""))
|
||
|
|
rows = result.fetchall()
|
||
|
|
|
||
|
|
for row in rows:
|
||
|
|
print(f"\nForm ID: {row[0]}, Name: {row[1]}")
|
||
|
|
|
||
|
|
# 转换HTML为JSON Schema
|
||
|
|
schema = html_to_json_schema(row[2])
|
||
|
|
schema_json = json.dumps(schema, ensure_ascii=False, indent=2)
|
||
|
|
|
||
|
|
print(f"JSON Schema:\n{schema_json[:500]}...")
|
||
|
|
|
||
|
|
# 更新数据库
|
||
|
|
await conn.execute(text("""
|
||
|
|
UPDATE test_form SET form_json = :schema WHERE id = :id
|
||
|
|
"""), {'schema': schema_json, 'id': row[0]})
|
||
|
|
await conn.commit()
|
||
|
|
|
||
|
|
print(f"[OK] Updated form {row[0]}")
|
||
|
|
|
||
|
|
await engine.dispose()
|
||
|
|
print("\nAll forms converted to JSON Schema format!")
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
asyncio.run(main())
|