319 lines
9.8 KiB
Python
319 lines
9.8 KiB
Python
"""
|
||
模板字段关联查询示例脚本
|
||
演示如何查询模板关联的输入和输出字段
|
||
"""
|
||
import pymysql
|
||
import os
|
||
from typing import Dict, List, Optional
|
||
|
||
# 数据库连接配置
|
||
DB_CONFIG = {
|
||
'host': os.getenv('DB_HOST', '152.136.177.240'),
|
||
'port': int(os.getenv('DB_PORT', 5012)),
|
||
'user': os.getenv('DB_USER', 'finyx'),
|
||
'password': os.getenv('DB_PASSWORD', '6QsGK6MpePZDE57Z'),
|
||
'database': os.getenv('DB_NAME', 'finyx'),
|
||
'charset': 'utf8mb4'
|
||
}
|
||
|
||
TENANT_ID = 615873064429507639
|
||
|
||
|
||
def get_template_fields_by_name(template_name: str) -> Optional[Dict]:
|
||
"""
|
||
根据模板名称获取关联的字段
|
||
|
||
Args:
|
||
template_name: 模板名称，如 '初步核实审批表'
|
||
|
||
Returns:
|
||
dict: 包含 template_id, template_name, input_fields 和 output_fields 的字典
|
||
"""
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
try:
|
||
sql = """
|
||
SELECT
|
||
fc.id AS template_id,
|
||
fc.name AS template_name,
|
||
f.id AS field_id,
|
||
f.name AS field_name,
|
||
f.filed_code AS field_code,
|
||
f.field_type
|
||
FROM f_polic_file_config fc
|
||
INNER JOIN f_polic_file_field fff ON fc.id = fff.file_id
|
||
INNER JOIN f_polic_field f ON fff.filed_id = f.id
|
||
WHERE fc.tenant_id = %s
|
||
AND fc.name = %s
|
||
AND fc.state = 1
|
||
AND fff.state = 1
|
||
AND f.state = 1
|
||
ORDER BY f.field_type, f.name
|
||
"""
|
||
cursor.execute(sql, (TENANT_ID, template_name))
|
||
rows = cursor.fetchall()
|
||
|
||
if not rows:
|
||
return None
|
||
|
||
result = {
|
||
'template_id': rows[0]['template_id'],
|
||
'template_name': rows[0]['template_name'],
|
||
'input_fields': [],
|
||
'output_fields': []
|
||
}
|
||
|
||
for row in rows:
|
||
field_info = {
|
||
'id': row['field_id'],
|
||
'name': row['field_name'],
|
||
'field_code': row['field_code'],
|
||
'field_type': row['field_type']
|
||
}
|
||
|
||
if row['field_type'] == 1:
|
||
result['input_fields'].append(field_info)
|
||
elif row['field_type'] == 2:
|
||
result['output_fields'].append(field_info)
|
||
|
||
return result
|
||
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
def get_template_fields_by_id(template_id: int) -> Optional[Dict]:
|
||
"""
|
||
根据模板ID获取关联的字段
|
||
|
||
Args:
|
||
template_id: 模板ID
|
||
|
||
Returns:
|
||
dict: 包含 template_id, template_name, input_fields 和 output_fields 的字典
|
||
"""
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
try:
|
||
# 先获取模板名称
|
||
sql_template = """
|
||
SELECT id, name
|
||
FROM f_polic_file_config
|
||
WHERE id = %s AND tenant_id = %s AND state = 1
|
||
"""
|
||
cursor.execute(sql_template, (template_id, TENANT_ID))
|
||
template = cursor.fetchone()
|
||
|
||
if not template:
|
||
return None
|
||
|
||
# 获取字段
|
||
sql_fields = """
|
||
SELECT
|
||
f.id AS field_id,
|
||
f.name AS field_name,
|
||
f.filed_code AS field_code,
|
||
f.field_type
|
||
FROM f_polic_file_field fff
|
||
INNER JOIN f_polic_field f ON fff.filed_id = f.id
|
||
WHERE fff.file_id = %s
|
||
AND fff.tenant_id = %s
|
||
AND fff.state = 1
|
||
AND f.state = 1
|
||
ORDER BY f.field_type, f.name
|
||
"""
|
||
cursor.execute(sql_fields, (template_id, TENANT_ID))
|
||
rows = cursor.fetchall()
|
||
|
||
result = {
|
||
'template_id': template['id'],
|
||
'template_name': template['name'],
|
||
'input_fields': [],
|
||
'output_fields': []
|
||
}
|
||
|
||
for row in rows:
|
||
field_info = {
|
||
'id': row['field_id'],
|
||
'name': row['field_name'],
|
||
'field_code': row['field_code'],
|
||
'field_type': row['field_type']
|
||
}
|
||
|
||
if row['field_type'] == 1:
|
||
result['input_fields'].append(field_info)
|
||
elif row['field_type'] == 2:
|
||
result['output_fields'].append(field_info)
|
||
|
||
return result
|
||
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
def get_all_templates_with_field_stats() -> List[Dict]:
|
||
"""
|
||
获取所有模板及其字段统计信息
|
||
|
||
Returns:
|
||
list: 模板列表，每个模板包含字段统计
|
||
"""
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
try:
|
||
sql = """
|
||
SELECT
|
||
fc.id AS template_id,
|
||
fc.name AS template_name,
|
||
COUNT(DISTINCT CASE WHEN f.field_type = 1 THEN f.id END) AS input_field_count,
|
||
COUNT(DISTINCT CASE WHEN f.field_type = 2 THEN f.id END) AS output_field_count,
|
||
COUNT(DISTINCT f.id) AS total_field_count
|
||
FROM f_polic_file_config fc
|
||
LEFT JOIN f_polic_file_field fff ON fc.id = fff.file_id AND fff.state = 1
|
||
LEFT JOIN f_polic_field f ON fff.filed_id = f.id AND f.state = 1
|
||
WHERE fc.tenant_id = %s
|
||
AND fc.state = 1
|
||
GROUP BY fc.id, fc.name
|
||
ORDER BY fc.name
|
||
"""
|
||
cursor.execute(sql, (TENANT_ID,))
|
||
templates = cursor.fetchall()
|
||
|
||
return [
|
||
{
|
||
'template_id': t['template_id'],
|
||
'template_name': t['template_name'],
|
||
'input_field_count': t['input_field_count'] or 0,
|
||
'output_field_count': t['output_field_count'] or 0,
|
||
'total_field_count': t['total_field_count'] or 0
|
||
}
|
||
for t in templates
|
||
]
|
||
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
def find_templates_using_field(field_code: str) -> List[Dict]:
|
||
"""
|
||
查找使用特定字段的所有模板
|
||
|
||
Args:
|
||
field_code: 字段编码，如 'target_name'
|
||
|
||
Returns:
|
||
list: 使用该字段的模板列表
|
||
"""
|
||
conn = pymysql.connect(**DB_CONFIG)
|
||
cursor = conn.cursor(pymysql.cursors.DictCursor)
|
||
|
||
try:
|
||
sql = """
|
||
SELECT DISTINCT
|
||
fc.id AS template_id,
|
||
fc.name AS template_name
|
||
FROM f_polic_file_config fc
|
||
INNER JOIN f_polic_file_field fff ON fc.id = fff.file_id
|
||
INNER JOIN f_polic_field f ON fff.filed_id = f.id
|
||
WHERE fc.tenant_id = %s
|
||
AND f.tenant_id = %s
|
||
AND f.filed_code = %s
|
||
AND fc.state = 1
|
||
AND fff.state = 1
|
||
AND f.state = 1
|
||
ORDER BY fc.name
|
||
"""
|
||
cursor.execute(sql, (TENANT_ID, TENANT_ID, field_code))
|
||
templates = cursor.fetchall()
|
||
|
||
return [
|
||
{
|
||
'template_id': t['template_id'],
|
||
'template_name': t['template_name']
|
||
}
|
||
for t in templates
|
||
]
|
||
|
||
finally:
|
||
cursor.close()
|
||
conn.close()
|
||
|
||
|
||
def print_template_fields(result: Dict):
|
||
"""打印模板字段信息"""
|
||
if not result:
|
||
print("未找到模板")
|
||
return
|
||
|
||
print("="*80)
|
||
print(f"模板: {result['template_name']} (ID: {result['template_id']})")
|
||
print("="*80)
|
||
|
||
print(f"\n输入字段 ({len(result['input_fields'])} 个):")
|
||
if result['input_fields']:
|
||
for field in result['input_fields']:
|
||
print(f" - {field['name']} ({field['field_code']})")
|
||
else:
|
||
print(" (无)")
|
||
|
||
print(f"\n输出字段 ({len(result['output_fields'])} 个):")
|
||
if result['output_fields']:
|
||
for field in result['output_fields']:
|
||
print(f" - {field['name']} ({field['field_code']})")
|
||
else:
|
||
print(" (无)")
|
||
|
||
|
||
def main():
|
||
"""主函数 - 演示各种查询方式"""
|
||
print("="*80)
|
||
print("模板字段关联查询示例")
|
||
print("="*80)
|
||
|
||
# 示例1: 根据模板名称查询
|
||
print("\n【示例1】根据模板名称查询字段")
|
||
print("-" * 80)
|
||
# 注意：模板名称需要完全匹配，如 "2.初步核实审批表（XXX）"
|
||
result = get_template_fields_by_name('2.初步核实审批表（XXX）')
|
||
if not result:
|
||
# 尝试其他可能的名称
|
||
result = get_template_fields_by_name('初步核实审批表')
|
||
print_template_fields(result)
|
||
|
||
# 示例2: 获取所有模板的字段统计
|
||
print("\n\n【示例2】获取所有模板的字段统计")
|
||
print("-" * 80)
|
||
templates = get_all_templates_with_field_stats()
|
||
print(f"共找到 {len(templates)} 个模板:\n")
|
||
for template in templates[:5]: # 只显示前5个
|
||
print(f" {template['template_name']} (ID: {template['template_id']})")
|
||
print(f" 输入字段: {template['input_field_count']} 个")
|
||
print(f" 输出字段: {template['output_field_count']} 个")
|
||
print(f" 总字段数: {template['total_field_count']} 个\n")
|
||
|
||
if len(templates) > 5:
|
||
print(f" ... 还有 {len(templates) - 5} 个模板")
|
||
|
||
# 示例3: 查找使用特定字段的模板
|
||
print("\n\n【示例3】查找使用 'target_name' 字段的模板")
|
||
print("-" * 80)
|
||
templates_using_field = find_templates_using_field('target_name')
|
||
print(f"共找到 {len(templates_using_field)} 个模板使用该字段:")
|
||
for template in templates_using_field:
|
||
print(f" - {template['template_name']} (ID: {template['template_id']})")
|
||
|
||
print("\n" + "="*80)
|
||
print("查询完成")
|
||
print("="*80)
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|
||
|