taolm/high-risk-scheme-approval-dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: run_all v2管线(6步含工作簿+危大看板)+gen_workbook日期参数化

Browse Source
This commit is contained in:
大师 2026-06-09 04:00:49 +08:00
parent 631ae3f8b0
commit da70789ea8
2 changed files with 51 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

63
run_all.py
View File

@ -1,24 +1,43 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
""" """
专项施工方案数据看板 · 一键管线 专项施工方案数据看板 · 一键管线 v2
用法: python3 run_all.py [YYYY-MM-DD] # 默认 2026-06-08 用法: python3 run_all.py [YYYY-MM-DD] # 默认当月日期
流程: B1(方案清洗) → B2(项目跟踪) → B3(营收财务) → B4(看板HTML) 流程: B1(方案清洗) → B2(项目跟踪) → B3(营收) → B4(工作簿) → B5(危大看板) → B6(综合看板)
""" """
import sys, subprocess, time import sys, subprocess, time, os
from pathlib import Path from pathlib import Path
from datetime import datetime
PROJECT_ROOT = Path(__file__).resolve().parent PROJECT_ROOT = Path(__file__).resolve().parent
date = sys.argv[1] if len(sys.argv) > 1 else '2026-06-08' now = datetime.now()
print(f'🏗️ 专项施工方案数据看板 · 管线启动\n 日期: {date}\n') default_date = sys.argv[1] if len(sys.argv) > 1 else now.strftime('%Y-%m-%d')
print(f'🏗️ 专项施工方案数据看板 · 管线启动\n 日期: {default_date}\n')
scripts = [ scripts = [
('B1 危大方案清洗', 'b1_methods.py', ['验证方案状态分布、审批率、预警信号']), ('B1 危大方案清洗', 'b1_methods.py', ['方案总数','完成']),
('B2 项目启动跟踪', 'b2_tracking.py', ['提取中东区域80个项目的启动任务完成情况']), ('B2 项目启动跟踪', 'b2_tracking.py', ['中东区域','任务']),
('B3 营收财务统计', 'b3_revenue.py', ['统计47个活跃项目营收数据']), ('B3 营收财务统计', 'b3_revenue.py', ['活跃项目','营收']),
('B4 看板HTML生成', 'b4_dashboard_html.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 = [] failed = []
start = time.time() start = time.time()
@ -33,21 +52,22 @@ for name, script, checks in scripts:
try: try:
result = subprocess.run( result = subprocess.run(
['python3', str(path)], ['python3', str(path)],
capture_output=True, text=True, timeout=120, capture_output=True, text=True, timeout=180,
cwd=str(PROJECT_ROOT), env={**__import__('os').environ, 'PYTHONIOENCODING': 'utf-8'} cwd=str(PROJECT_ROOT), env={**os.environ, 'PYTHONIOENCODING': 'utf-8'}
) )
if result.returncode != 0: if result.returncode != 0:
print(f' ❌ 退出码 {result.returncode}') print(f' ❌ 退出码 {result.returncode}')
print(f' {result.stderr.strip()[-200:]}') err = result.stderr.strip()[-300:]
if err:
print(f' {err}')
failed.append(name) failed.append(name)
else: else:
# Show summary lines
for line in result.stdout.strip().split('\n'): for line in result.stdout.strip().split('\n'):
if any(kw in line for kw in ['✅', 'Summary', 'Done', '合同', 'Found', 'Top']): if any(kw in line for kw in checks + ['✅','总计','OA登记']):
print(f' {line.strip()[:100]}') print(f' {line.strip()[:120]}')
print(f' ✅ 完成') print(f' ✅ 完成')
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
print(f' ❌ 超时') print(f' ❌ 超时(180s)')
failed.append(name) failed.append(name)
elapsed = time.time() - start elapsed = time.time() - start
@ -57,12 +77,11 @@ if failed:
print(f'⚠️ 失败: {", ".join(failed)}') print(f'⚠️ 失败: {", ".join(failed)}')
sys.exit(1) sys.exit(1)
# Show output paths out = PROJECT_ROOT / 'data' / default_date / 'cleaned'
out = PROJECT_ROOT / 'data' / date / 'cleaned'
print(f'\n📁 输出目录: {out}') print(f'\n📁 输出目录: {out}')
for f in sorted(out.glob('*.html')): for f in sorted(out.glob('*.html')):
print(f' 🌐 {f.name}') print(f' 🌐 {f.name}')
for f in sorted(out.glob('*.parquet')): for f in sorted(out.glob('*.xlsx')):
print(f' 📊 {f.name}') print(f' 📊 {f.name}')
for f in sorted(out.glob('*_validation.json')): for f in sorted(out.glob('*.pptx')):
print(f' 📋 {f.name}') print(f' 📑 {f.name}')

13
src/gen_workbook.py
View File

@ -1,14 +1,21 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""危大方案看板数据工作簿 v5 — 双数据源(OA登记+公司认定) + GROUPBY/FILTER动态公式""" """危大方案看板数据工作簿 v5 — 双数据源(OA登记+公司认定) + GROUPBY/FILTER动态公式"""
import sys
import pandas as pd import pandas as pd
from openpyxl import Workbook from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.utils import get_column_letter from openpyxl.utils import get_column_letter
from datetime import datetime
BASE = "/mnt/y/Openclaw_Hub/03.资源/实施项目 wiki/dashboard/data/2026-06-08" # 日期：命令行参数 or 默认今天
CERT_DIR = "/mnt/y/Openclaw_Hub/03.资源/实施项目 wiki/dashboard/data/认定数据/2026" # 年度固定目录·一年一次 if len(sys.argv) > 1:
REPORT_DATE = sys.argv[1]
else:
REPORT_DATE = datetime.now().strftime('%Y-%m-%d')
BASE = f"/mnt/y/Openclaw_Hub/03.资源/实施项目 wiki/dashboard/data/{REPORT_DATE}"
CERT_DIR = "/mnt/y/Openclaw_Hub/03.资源/实施项目 wiki/dashboard/data/认定数据/2026" # 年度固定目录
OUT = f"{BASE}/cleaned/危大方案看板数据工作簿.xlsx" OUT = f"{BASE}/cleaned/危大方案看板数据工作簿.xlsx"
REPORT_DATE = "2026-06-08"
REF = "'有效≥2026'" # OA公式引用 REF = "'有效≥2026'" # OA公式引用
CREF = "'认定数据'" # 认定公式引用 CREF = "'认定数据'" # 认定公式引用