99 lines
3.0 KiB
Python
99 lines
3.0 KiB
Python
"""
|
|
Automated CMA task execution for all 4 sub-tasks
|
|
"""
|
|
import os
|
|
import sys
|
|
|
|
# Set backend directory as working dir
|
|
backend_dir = '/root/cma-management/backend'
|
|
os.chdir(backend_dir)
|
|
sys.path.insert(0, backend_dir)
|
|
|
|
# Load env
|
|
from dotenv import load_dotenv
|
|
load_dotenv()
|
|
|
|
from app.database import get_session_local
|
|
from app.models import DataSourceConfig
|
|
from openpyxl import Workbook, load_workbook
|
|
|
|
# ============================================================
|
|
# Sub-task 1: Insert data source config
|
|
# ============================================================
|
|
print("=" * 60)
|
|
print("子任务1: 插入数据源记录")
|
|
print("=" * 60)
|
|
|
|
db = get_session_local()()
|
|
try:
|
|
existing = db.query(DataSourceConfig).filter(
|
|
DataSourceConfig.name == 'ERP系统 - 博海网络'
|
|
).first()
|
|
if existing:
|
|
print(f"数据源已存在: id={existing.id}, name={existing.name}")
|
|
else:
|
|
source = DataSourceConfig(
|
|
name='ERP系统 - 博海网络',
|
|
source_type='erp',
|
|
api_endpoint='http://127.0.0.1:8300/api/v1',
|
|
api_key='erp-gateway-key-bhwl-2026',
|
|
sync_type='batch',
|
|
status='active',
|
|
)
|
|
db.add(source)
|
|
db.commit()
|
|
db.refresh(source)
|
|
print(f"数据源插入成功: id={source.id}, name={source.name}")
|
|
|
|
all_sources = db.query(DataSourceConfig).all()
|
|
print(f"当前 data_source_config 表记录数: {len(all_sources)}")
|
|
for s in all_sources:
|
|
print(f" id={s.id}, name={s.name}, type={s.source_type}, status={s.status}, endpoint={s.api_endpoint}")
|
|
except Exception as e:
|
|
print(f"错误: {e}")
|
|
db.rollback()
|
|
finally:
|
|
db.close()
|
|
|
|
# ============================================================
|
|
# Sub-task 4: Create Excel import template
|
|
# ============================================================
|
|
print()
|
|
print("=" * 60)
|
|
print("子任务4: 创建Excel导入模板")
|
|
print("=" * 60)
|
|
|
|
template_dir = os.path.join(backend_dir, 'templates')
|
|
os.makedirs(template_dir, exist_ok=True)
|
|
print(f"Templates目录: {template_dir}")
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "KPI导入模板"
|
|
|
|
headers = ['kpi_code', 'kpi_name', 'period', 'actual_value', 'unit', 'dimension']
|
|
for col_idx, header in enumerate(headers, 1):
|
|
ws.cell(row=1, column=col_idx, value=header)
|
|
|
|
sample_data = [
|
|
['F_REVENUE', '营业收入', '2026-06', 500000, '万元', 'finance'],
|
|
['F_PROFIT_RATE', '销售毛利率', '2026-06', 28.5, '%', 'finance'],
|
|
]
|
|
for row_idx, row_data in enumerate(sample_data, 2):
|
|
for col_idx, value in enumerate(row_data, 1):
|
|
ws.cell(row=row_idx, column=col_idx, value=value)
|
|
|
|
output_path = os.path.join(template_dir, 'kpi_import_template.xlsx')
|
|
wb.save(output_path)
|
|
print(f"Excel模板已创建: {output_path}")
|
|
|
|
# Verify
|
|
wb2 = load_workbook(output_path)
|
|
ws2 = wb2.active
|
|
print("验证文件内容:")
|
|
for row in ws2.iter_rows(min_row=1, max_row=3, values_only=True):
|
|
print(f" {row}")
|
|
|
|
print()
|
|
print("子任务1 和 子任务4 已完成。")
|