vent_agent_2.0/scripts/run_main_fan_batch.py

#!/usr/bin/env python3
"""显式选矿的本地完整Agent联调；隔离数据库、模拟PDF接口、禁止结果推送。"""
import argparse
import asyncio
from datetime import datetime
import json
import os
from pathlib import Path
import sys
from zoneinfo import ZoneInfo

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))


async def run(args, output):
    from extensions.main_fan.batch import BatchRunner
    from runtime.environment import RuntimeEnvironment
    async with RuntimeEnvironment(prewarm=False, billing_retry=False):
        runner = BatchRunner(output, args.mock_pdf, cache_root=args.cache_dir,
            progress=lambda data: print(json.dumps(data, ensure_ascii=False), flush=True))
        state = await runner.run({'dept_ids': args.dept_id or [], 'all_producing_mines': args.all_producing_mines,
            'snapshot_id': args.snapshot_id,
            'reassess': args.reassess,
            'cutoff': args.cutoff, 'lookback_days': 40, 'user_name': 'main-fan-batch-local'})
        print(json.dumps({'status': state['status'], 'output': str(output),
            'index': str(output / 'index.html'), 'mine_count': len(state['mines']), 'publish': False}, ensure_ascii=False), flush=True)
        return 0 if state['status'] == 'completed' else 1


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    selection = parser.add_mutually_exclusive_group(required=True)
    selection.add_argument('--dept-id', action='append', help='正常生产煤矿的真实部门编号，可重复指定')
    selection.add_argument('--all-producing-mines', action='store_true', help='显式选择本批次全部正常生产煤矿，联调不使用')
    parser.add_argument('--mock-pdf', type=Path, required=True, help='原有测试PDF，不修改内容')
    parser.add_argument('--snapshot-id', help='使用已有冻结清单；默认重新取得deptTree')
    parser.add_argument('--cutoff', help='冻结的北京时间截止时刻，默认批次启动时刻')
    parser.add_argument('--output', type=Path, help='批次输出目录；再次指定同目录可恢复未完成项')
    parser.add_argument('--cache-dir', type=Path, help='可选独立报告档案目录')
    parser.add_argument('--reassess', action='store_true', help='在同一输出目录重新分析，保留旧报告并复用同批次已有数据')
    parser.add_argument('--allow-large-batch', action='store_true', help='显式解除联调最多两个矿的保护')
    args = parser.parse_args()
    if (args.all_producing_mines or len(args.dept_id or []) > 2) and not args.allow_large_batch:
        parser.error('本地联调默认最多两个矿；大批次必须另行显式启用')
    if not args.mock_pdf.is_file():
        parser.error('测试PDF不存在')
    stamp = datetime.now(ZoneInfo('Asia/Shanghai')).strftime('%Y%m%d_%H%M%S')
    output = (args.output or ROOT / 'outputs/main_fan_batches' / stamp).resolve()
    output.mkdir(parents=True, exist_ok=True)
    os.environ.update({'DEEP_AGENT_ENV_PRIORITY': 'process', 'DB_DIALECT': 'sqlite',
        'DB_SQLITE_PATH': str(output / 'runtime/business.sqlite'), 'CHECKPOINTER': 'sqlite',
        'CHECKPOINT_DB_PATH': str(output / 'runtime/checkpoints.sqlite'),
        'RESULT_PLATFORM_ENABLED': '0', 'MAIN_FAN_AUTO_PUBLISH': '0',
        'MAIN_FAN_LOCAL_BATCH_ENABLED': '0', 'MAIN_FAN_ALLOW_MANUAL_CACHE': '0',
        'MAIN_FAN_TARGETS_FILE': '', 'LOG_LEVEL': 'WARNING'})
    (output / 'runtime').mkdir(exist_ok=True)
    raise SystemExit(asyncio.run(run(args, output)))


if __name__ == '__main__':
    main()