88 lines
3.4 KiB
Python
88 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
专项施工方案数据看板 · 一键管线 v2
|
|
用法: python3 run_all.py [YYYY-MM-DD] # 默认当月日期
|
|
流程: B1(方案清洗) → B2(项目跟踪) → B3(营收) → B4(工作簿) → B5(危大看板) → B6(综合看板)
|
|
"""
|
|
import sys, subprocess, time, os
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent
|
|
|
|
now = datetime.now()
|
|
default_date = sys.argv[1] if len(sys.argv) > 1 else now.strftime('%Y-%m-%d')
|
|
print(f'🏗️ 专项施工方案数据看板 · 管线启动\n 日期: {default_date}\n')
|
|
|
|
scripts = [
|
|
('B1 危大方案清洗', 'b1_methods.py', ['方案总数','完成']),
|
|
('B2 项目启动跟踪', 'b2_tracking.py', ['中东区域','任务']),
|
|
('B3 营收财务统计', 'b3_revenue.py', ['活跃项目','营收']),
|
|
('B4 数据工作簿', 'gen_workbook.py', ['OA登记','认定危大']),
|
|
('B5 危大方案看板', 'b4b_certified_dashboard_2026.py', ['危大方案']),
|
|
('B6 综合看板', 'b4_dashboard_html.py', ['综合看板']),
|
|
]
|
|
|
|
# 检查认定数据是否已就绪
|
|
cert_dir = PROJECT_ROOT / 'data' / '认定数据' / str(now.year)
|
|
if not (cert_dir / 'certified_schemes_detail.csv').exists():
|
|
print('⚠️ 认定数据未找到，运行年度认定清洗...')
|
|
cert_script = PROJECT_ROOT / 'src' / 'clean_certified.py'
|
|
if cert_script.exists():
|
|
r = subprocess.run(['python3', str(cert_script)], capture_output=True, text=True,
|
|
cwd=str(PROJECT_ROOT), timeout=120)
|
|
if r.returncode != 0:
|
|
print(f' ❌ 认定清洗失败，工作簿将缺少认定数据\n')
|
|
else:
|
|
print(f' ✅ 认定清洗完成\n')
|
|
else:
|
|
print(f' ⚠️ clean_certified.py 未找到，跳过\n')
|
|
|
|
failed = []
|
|
start = time.time()
|
|
|
|
for name, script, checks in scripts:
|
|
path = PROJECT_ROOT / 'src' / script
|
|
if not path.exists():
|
|
print(f' ⚠️ {name}: 脚本不存在 {path}')
|
|
failed.append(name)
|
|
continue
|
|
|
|
print(f'▶ {name} ...')
|
|
try:
|
|
result = subprocess.run(
|
|
['python3', str(path), default_date],
|
|
capture_output=True, text=True, timeout=180,
|
|
cwd=str(PROJECT_ROOT), env={**os.environ, 'PYTHONIOENCODING': 'utf-8'}
|
|
)
|
|
if result.returncode != 0:
|
|
print(f' ❌ 退出码 {result.returncode}')
|
|
err = result.stderr.strip()[-300:]
|
|
if err:
|
|
print(f' {err}')
|
|
failed.append(name)
|
|
else:
|
|
for line in result.stdout.strip().split('\n'):
|
|
if any(kw in line for kw in checks + ['✅','总计','OA登记']):
|
|
print(f' {line.strip()[:120]}')
|
|
print(f' ✅ 完成')
|
|
except subprocess.TimeoutExpired:
|
|
print(f' ❌ 超时(180s)')
|
|
failed.append(name)
|
|
|
|
elapsed = time.time() - start
|
|
print(f'\n{"="*50}')
|
|
print(f'🏁 管线完成 | 耗时 {elapsed:.0f}s | 成功 {len(scripts)-len(failed)}/{len(scripts)}')
|
|
if failed:
|
|
print(f'⚠️ 失败: {", ".join(failed)}')
|
|
sys.exit(1)
|
|
|
|
out = PROJECT_ROOT / 'data' / default_date / 'cleaned'
|
|
print(f'\n📁 输出目录: {out}')
|
|
for f in sorted(out.glob('*.html')):
|
|
print(f' 🌐 {f.name}')
|
|
for f in sorted(out.glob('*.xlsx')):
|
|
print(f' 📊 {f.name}')
|
|
for f in sorted(out.glob('*.pptx')):
|
|
print(f' 📑 {f.name}')
|