|
|
@@ -1,4 +1,4 @@
|
|
|
-"""按矿复用main_fan Agent的本地批次:先发现、再逐井多轮执行,结果不进入推送队列。"""
|
|
|
+"""固定代码准备数据,逐井研判先落本地,可选进入持久化结果推送队列。"""
|
|
|
import asyncio
|
|
|
from contextlib import contextmanager
|
|
|
from datetime import datetime
|
|
|
@@ -8,19 +8,22 @@ from html import escape
|
|
|
import json
|
|
|
from pathlib import Path
|
|
|
import shutil
|
|
|
+import time
|
|
|
from uuid import uuid4
|
|
|
from zoneinfo import ZoneInfo
|
|
|
|
|
|
import config
|
|
|
from agents.workspace import session_root
|
|
|
-from db import chat_store
|
|
|
-from extensions.main_fan import exploration, mine_catalog
|
|
|
+from db import chat_store, main_fan_store, business_store
|
|
|
+from extensions.main_fan import data_export, discovery, exploration, mine_catalog, preparation
|
|
|
from extensions.main_fan.gateway_queries import parse_time
|
|
|
from extensions.main_fan.batch_context import BatchScope, current_scope
|
|
|
from extensions.main_fan.batch_models import BatchRequest, MinePlan
|
|
|
-from extensions.main_fan.report_api import ReportClient, mock_report_client
|
|
|
+from extensions.main_fan.crawler_reports import open_report_source, report_source
|
|
|
+from obs import bind_log_context, unbind_log_context, get_logger
|
|
|
from runtime import execution
|
|
|
-from runtime.python_sandbox import digest, workspace_file
|
|
|
+
|
|
|
+logger = get_logger(__name__)
|
|
|
|
|
|
|
|
|
@contextmanager
|
|
|
@@ -59,7 +62,9 @@ def export_index(folder, state):
|
|
|
table{border-collapse:collapse;width:100%;background:white}td,th{padding:12px;border-bottom:1px solid #dbe3e5;text-align:left}
|
|
|
a{color:#076b78}.note{padding:16px;background:#fff2d7;border-radius:8px}code{font-size:14px}</style>
|
|
|
<h1>主通风机判识 · 本地预览</h1>'''
|
|
|
- body += '<p class="note">测定PDF使用既有测试原件的模拟接口,保留原件真实身份。结果仅写入本地文件,未推送。</p>'
|
|
|
+ source_note = '测定PDF使用测试原件模拟接口。' if state['pdf_mode'] == 'mock' else '测定PDF来自资料采集服务及只读挂载目录。'
|
|
|
+ delivery_note = '结果先保存在本地,符合提交条件的结果进入推送队列;推送状态见业务任务记录。' if state['publish'] else '结果仅保存在本地,推送关闭。'
|
|
|
+ body += '<p class="note">' + source_note + delivery_note + '</p>'
|
|
|
body += '<p>批次:<code>' + escape(state['batch_id']) + '</code> 状态:' + escape(state['status']) + '</p>'
|
|
|
body += '<p>截止时刻:' + escape(state['cutoff']) + ',回查 ' + str(state['lookback_days']) + ' 天。</p>'
|
|
|
verification = state.get('local_verification')
|
|
|
@@ -72,9 +77,11 @@ a{color:#076b78}.note{padding:16px;background:#fff2d7;border-radius:8px}code{fon
|
|
|
|
|
|
|
|
|
class BatchRunner:
|
|
|
- def __init__(self, output, pdf_path, *, cache_root=None, executor=None, progress=None, max_mines=None):
|
|
|
+ def __init__(self, output, pdf_path=None, *, cache_root=None, executor=None, progress=None, max_mines=None):
|
|
|
self.output = Path(output).resolve()
|
|
|
- self.pdf_path = Path(pdf_path).resolve()
|
|
|
+ self.pdf_path = Path(pdf_path).resolve() if pdf_path is not None else None
|
|
|
+ self.source = report_source(self.pdf_path)
|
|
|
+ self.publish = config.get_bool('RESULT_PLATFORM_ENABLED', False)
|
|
|
self.cache_root = Path(cache_root or config.ROOT / 'data/main_fan/report_archives').resolve()
|
|
|
self.executor = executor or execution.execute
|
|
|
self.progress = progress
|
|
|
@@ -89,7 +96,21 @@ class BatchRunner:
|
|
|
if len(set(request.dept_ids)) != len(request.dept_ids):
|
|
|
raise ValueError('选择的煤矿部门不能重复')
|
|
|
with batch_lock(self.output):
|
|
|
- return await self._run(request, snapshot)
|
|
|
+ started = time.monotonic()
|
|
|
+ token = bind_log_context(batch_output=str(self.output))
|
|
|
+ try:
|
|
|
+ logger.info('主通风机批次启动', extra={'kv': {'event': 'batch.started',
|
|
|
+ 'report_mode': self.source['mode'], 'publish': self.publish, 'lookback_days': request.lookback_days}})
|
|
|
+ state = await self._run(request, snapshot)
|
|
|
+ logger.info('主通风机批次结束', extra={'kv': {'event': 'batch.finished',
|
|
|
+ 'batch_id': state['batch_id'], 'status': state['status'],
|
|
|
+ 'duration_ms': round((time.monotonic() - started) * 1000)}})
|
|
|
+ return state
|
|
|
+ except Exception:
|
|
|
+ logger.exception('主通风机批次失败', extra={'kv': {'event': 'batch.failed'}})
|
|
|
+ raise
|
|
|
+ finally:
|
|
|
+ unbind_log_context(token)
|
|
|
|
|
|
async def _run(self, request, snapshot):
|
|
|
state_path = self.output / 'batch.json'
|
|
|
@@ -98,11 +119,12 @@ class BatchRunner:
|
|
|
if (state.get('requested_dept_ids', state['dept_ids']) != request.dept_ids
|
|
|
or state.get('all_producing_mines', False) != request.all_producing_mines
|
|
|
or state['lookback_days'] != request.lookback_days
|
|
|
- or state['pdf_sha256'] != digest(self.pdf_path)
|
|
|
+ or state.get('report_source', {'mode': 'mock', 'sha256': state.get('pdf_sha256')}) != self.source
|
|
|
+ or state.get('publish', False) != self.publish
|
|
|
or state['user_name'] != request.user_name
|
|
|
or (request.cutoff and parse_time(request.cutoff).replace(microsecond=0).isoformat() != state['cutoff'])
|
|
|
or (request.snapshot_id and request.snapshot_id != state['snapshot_id'])):
|
|
|
- raise ValueError('恢复批次的矿井、时段、报告或用户与原记录不一致')
|
|
|
+ raise ValueError('恢复批次的矿井、时段、报告源、推送开关或用户与原记录不一致')
|
|
|
snapshot = json.loads((self.output / 'mine-catalog.json').read_text())
|
|
|
if request.reassess:
|
|
|
for item in state['mines']:
|
|
|
@@ -113,10 +135,13 @@ class BatchRunner:
|
|
|
if entry.get('analysis_id') and folder.is_dir():
|
|
|
revision = folder / 'revisions' / entry['analysis_id']
|
|
|
revision.mkdir(parents=True, exist_ok=True)
|
|
|
- for name in ('报告.md', '结果.json', '来源与覆盖.json', '待审阅报文.json'):
|
|
|
+ for name in ('报告.md', '结果.json', '来源与覆盖.json', '待审阅报文.json',
|
|
|
+ '推送记录.json', '给Agent的数据.json', '研判数据.zip', '数据清单.json'):
|
|
|
if (folder / name).is_file():
|
|
|
shutil.copyfile(folder / name, revision / name)
|
|
|
entry['status'] = 'pending'
|
|
|
+ entry.pop('analysis_saved', None)
|
|
|
+ entry['assessment_revision'] = uuid4().hex
|
|
|
state['status'] = 'running'
|
|
|
else:
|
|
|
snapshot = snapshot or (mine_catalog.load(request.snapshot_id) if request.snapshot_id
|
|
|
@@ -130,8 +155,10 @@ class BatchRunner:
|
|
|
cutoff = (parse_time(request.cutoff)
|
|
|
if request.cutoff else datetime.now(ZoneInfo('Asia/Shanghai'))).replace(microsecond=0).isoformat()
|
|
|
state = {'schema_version': 1, 'batch_id': uuid4().hex, 'snapshot_id': snapshot['snapshot_id'],
|
|
|
- 'status': 'running', 'publish': False, 'delivery': 'local_files', 'pdf_mode': 'mock',
|
|
|
- 'pdf_sha256': digest(self.pdf_path), 'cutoff': cutoff, 'lookback_days': request.lookback_days,
|
|
|
+ 'status': 'running', 'publish': self.publish,
|
|
|
+ 'delivery': 'local_files_and_push_queue' if self.publish else 'local_files', 'pdf_mode': self.source['mode'],
|
|
|
+ 'report_source': self.source, 'pdf_sha256': self.source.get('sha256'),
|
|
|
+ 'cutoff': cutoff, 'lookback_days': request.lookback_days,
|
|
|
'dept_ids': dept_ids, 'requested_dept_ids': request.dept_ids,
|
|
|
'all_producing_mines': request.all_producing_mines, 'user_name': request.user_name,
|
|
|
'mines': [{**{key: selected[dept][key] for key in ('dept_id', 'mine_id', 'mine_name')},
|
|
|
@@ -139,9 +166,9 @@ class BatchRunner:
|
|
|
exploration._write(self.output / 'mine-catalog.json', snapshot)
|
|
|
if state['status'] != 'completed':
|
|
|
state['status'] = 'running'
|
|
|
+ state['data_preparation'] = 'fixed_code'
|
|
|
self.persist(state)
|
|
|
- async with mock_report_client(self.pdf_path) as http:
|
|
|
- report_client = ReportClient(http)
|
|
|
+ async with open_report_source(self.source, self.pdf_path) as report_client:
|
|
|
for item in state['mines']:
|
|
|
if item['status'] == 'completed':
|
|
|
continue
|
|
|
@@ -154,37 +181,61 @@ class BatchRunner:
|
|
|
raise ValueError('恢复批次的原矿级会话不存在,请使用原业务数据库')
|
|
|
scope = BatchScope(state['batch_id'], item, request.user_name, item['session_id'],
|
|
|
mine_dir, self.cache_root, state['cutoff'], state['lookback_days'], report_client,
|
|
|
- progress=self.progress)
|
|
|
+ progress=self.progress, report_mode=self.source['mode'], publish=self.publish)
|
|
|
token = current_scope.set(scope)
|
|
|
+ log_token = bind_log_context(batch_id=scope.batch_id, dept_id=item['dept_id'],
|
|
|
+ mine_name=item['mine_name'], session_id=scope.session)
|
|
|
+ mine_started = time.monotonic()
|
|
|
try:
|
|
|
+ logger.info('矿井处理开始', extra={'kv': {'event': 'batch.mine.started'}})
|
|
|
item.update(status='running', error='')
|
|
|
self.persist(state)
|
|
|
if not item.get('plan'):
|
|
|
data = self.input_for(scope, 'discovery')
|
|
|
- result = await self.call(scope, data, 'discovery')
|
|
|
- plan = MinePlan.model_validate(result['assessment']['plan'])
|
|
|
+ _, prepared = self.prepare_workflow(scope, data, 'discovery')
|
|
|
+ plan = await discovery.discover(prepared['workflow_id'], scope.user, scope.session)
|
|
|
+ main_fan_store.update_workflow(prepared['workflow_id'], status='completed')
|
|
|
item['plan'] = plan.model_dump(mode='json')
|
|
|
item['shafts'] = [{**shaft.model_dump(mode='json'), 'status': 'pending'} for shaft in plan.shafts]
|
|
|
- exploration._write(mine_dir / 'discovery.json', result)
|
|
|
+ exploration._write(mine_dir / 'discovery.json', {'mode': 'fixed_code', 'plan': item['plan']})
|
|
|
self.persist(state)
|
|
|
plan = MinePlan.model_validate(item['plan'])
|
|
|
for shaft, entry in zip(plan.shafts, item['shafts']):
|
|
|
if entry['status'] == 'completed':
|
|
|
continue
|
|
|
scope.shaft = shaft
|
|
|
- entry.update(status='running', error='')
|
|
|
- self.persist(state)
|
|
|
+ scope.assessment_revision = entry.get('assessment_revision', '')
|
|
|
+ shaft_log = bind_log_context(shaft_id=shaft.shaft_id)
|
|
|
try:
|
|
|
- result = await self.call(scope, self.input_for(scope, 'assessment'), shaft.shaft_id)
|
|
|
- self.export_shaft(scope, result)
|
|
|
+ entry.update(status='running', error='')
|
|
|
+ self.persist(state)
|
|
|
+ folder = mine_dir / shaft.shaft_id
|
|
|
+ # 推送入队失败后恢复时直接使用已落地的结果,避免重复调用模型。
|
|
|
+ if entry.get('analysis_saved') and (folder / '结果.json').is_file():
|
|
|
+ result = json.loads((folder / '结果.json').read_text(encoding='utf-8'))
|
|
|
+ else:
|
|
|
+ result = await self.call(scope, self.input_for(scope, 'assessment'), shaft.shaft_id)
|
|
|
+ self.export_shaft(scope, result)
|
|
|
+ entry['analysis_saved'] = True
|
|
|
+ self.persist(state)
|
|
|
+ entry['data_archive'] = await asyncio.to_thread(data_export.export, scope, result)
|
|
|
+ logger.info('研判所用数据已与结果一同归档', extra={'kv': {
|
|
|
+ 'event': 'batch.data.exported', **entry['data_archive']}})
|
|
|
+ self.queue_delivery(scope, result)
|
|
|
+ (folder / 'failure.json').unlink(missing_ok=True)
|
|
|
entry.update(status='completed', result_code=result['platform_payload']['judgment']['resultCode'],
|
|
|
analysis_id=result['analysis_id'], workflow_id=result['workflow_id'])
|
|
|
+ logger.info('通风井研判已保存', extra={'kv': {'event': 'batch.shaft.completed',
|
|
|
+ 'analysis_id': result['analysis_id'], 'publish': self.publish}})
|
|
|
except asyncio.CancelledError:
|
|
|
entry['status'] = 'interrupted'
|
|
|
raise
|
|
|
except Exception as exc:
|
|
|
+ logger.exception('通风井处理失败', extra={'kv': {'event': 'batch.shaft.failed'}})
|
|
|
entry.update(status='failed', error=type(exc).__name__ + ': ' + str(exc)[:500])
|
|
|
exploration._write(mine_dir / shaft.shaft_id / 'failure.json', {'error': entry['error']})
|
|
|
+ finally:
|
|
|
+ unbind_log_context(shaft_log)
|
|
|
self.persist(state)
|
|
|
item['status'] = 'completed' if all(row['status'] == 'completed' for row in item['shafts']) else 'partial'
|
|
|
except asyncio.CancelledError:
|
|
|
@@ -193,17 +244,23 @@ class BatchRunner:
|
|
|
self.persist(state)
|
|
|
raise
|
|
|
except Exception as exc:
|
|
|
+ logger.exception('矿井处理失败', extra={'kv': {'event': 'batch.mine.failed'}})
|
|
|
item.update(status='failed', error=type(exc).__name__ + ': ' + str(exc)[:500])
|
|
|
exploration._write(mine_dir / 'failure.json', {'error': item['error']})
|
|
|
finally:
|
|
|
+ logger.info('矿井处理结束', extra={'kv': {'event': 'batch.mine.finished',
|
|
|
+ 'status': item['status'], 'duration_ms': round((time.monotonic() - mine_started) * 1000)}})
|
|
|
current_scope.reset(token)
|
|
|
# 包含真实请求、响应、PDF、脚本和证据,便于用户本地检查;没有服务端凭证。
|
|
|
- source = session_root(scope.user, scope.session)
|
|
|
- for directory in ('workspace', 'python_runs'):
|
|
|
- if (source / directory).is_dir():
|
|
|
- shutil.copytree(source / directory, mine_dir / directory,
|
|
|
- dirs_exist_ok=True, ignore=exportable_files)
|
|
|
- self.persist(state)
|
|
|
+ try:
|
|
|
+ source = session_root(scope.user, scope.session)
|
|
|
+ for directory in ('workspace', 'python_runs'):
|
|
|
+ if (source / directory).is_dir():
|
|
|
+ shutil.copytree(source / directory, mine_dir / directory,
|
|
|
+ dirs_exist_ok=True, ignore=exportable_files)
|
|
|
+ self.persist(state)
|
|
|
+ finally:
|
|
|
+ unbind_log_context(log_token)
|
|
|
state['status'] = 'completed' if all(row['status'] == 'completed' for row in state['mines']) else 'partial'
|
|
|
state['finished_at'] = datetime.now(ZoneInfo('Asia/Shanghai')).isoformat()
|
|
|
self.persist(state)
|
|
|
@@ -215,33 +272,50 @@ class BatchRunner:
|
|
|
'dept_id': scope.mine['dept_id'], 'use_gateway': True,
|
|
|
'fan_pair': scope.shaft.fan_pair if scope.shaft else [],
|
|
|
'task_description': '本地批次主通风机判识',
|
|
|
- 'measurements': {'batch': {'phase': phase, 'batch_id': scope.batch_id,
|
|
|
+ 'measurements': {'batch': {'phase': phase, 'batch_id': scope.batch_id, 'data_preparation': 'fixed_code',
|
|
|
'cutoff': scope.cutoff, 'lookback_days': scope.lookback_days,
|
|
|
+ 'assessment_revision': scope.assessment_revision,
|
|
|
'shaft_plan': scope.shaft.model_dump(mode='json') if scope.shaft else None}}}
|
|
|
|
|
|
- async def call(self, scope, data, phase):
|
|
|
- if self.progress:
|
|
|
- self.progress({'stage': phase, 'dept_id': scope.mine['dept_id'], 'session_id': scope.session})
|
|
|
+ @staticmethod
|
|
|
+ def prepare_workflow(scope, data, phase):
|
|
|
mode = data['measurements']['batch']['phase']
|
|
|
- task_id = hashlib.sha256((scope.batch_id + scope.mine['dept_id'] + mode + ':' + phase).encode()).hexdigest()[:32]
|
|
|
+ seed = scope.batch_id + scope.mine['dept_id'] + mode + ':' + phase
|
|
|
+ if scope.assessment_revision:
|
|
|
+ seed += ':' + scope.assessment_revision
|
|
|
+ task_id = hashlib.sha256(seed.encode()).hexdigest()[:32]
|
|
|
prepared = exploration.prepare(data, scope.user, scope.session, trusted=data, job_id=task_id)
|
|
|
scope.active_workflow_id = prepared['workflow_id']
|
|
|
+ return task_id, prepared
|
|
|
+
|
|
|
+ async def call(self, scope, data, phase):
|
|
|
+ started = time.monotonic()
|
|
|
+ logger.info('固定代码准备分析输入', extra={'kv': {'event': 'batch.preparation.started'}})
|
|
|
+ if self.progress:
|
|
|
+ self.progress({'stage': 'fixed_data_preparation', 'dept_id': scope.mine['dept_id'], 'session_id': scope.session})
|
|
|
+ task_id, prepared = self.prepare_workflow(scope, data, phase)
|
|
|
+ inputs = await preparation.assessment(prepared['workflow_id'], scope.user, scope.session)
|
|
|
+ logger.info('分析输入已准备,开始调用 Agent', extra={'kv': {'event': 'batch.agent.started',
|
|
|
+ 'workflow_id': prepared['workflow_id'], 'duration_ms': round((time.monotonic() - started) * 1000)}})
|
|
|
message = ('这是主通风机本地批次,先读取业务Skill及references/batch.md。完整输入已由宿主注入,'
|
|
|
'调用prepare_main_fan_task后读取input.json和其中measurements.batch。'
|
|
|
- '本轮阶段为' + ('矿级发现:查询本矿两个实时列表并取齐分页,自主确认通风井和测点,最后submit_main_fan_mine_plan。'
|
|
|
- if mode == 'discovery' else
|
|
|
- '井级分析:fetch_main_fan_reports核对PDF及历史档案;query_main_fan_history按天查满开停和风量。'
|
|
|
+ '本轮仅进行分析:固定代码已完成筛矿、两个实时列表的分页和测点选择、40天开停和风量查询、'
|
|
|
+ '开停区间及切换候选、前后1小时目标的单点风量对比,报告接口与hash核验也已完成。'
|
|
|
+ '无需规划取数或再次调用实时/历史接口。直接读取整理后的Agent输入:' + inputs['agent_data_path'] + '。'
|
|
|
+ '其中reports给出PDF/可信档案及analyze/reuse模式。'
|
|
|
'首次PDF完整读取后archive_main_fan_report保存提取,复用档案时跳过PDF识别。'
|
|
|
'新的档案契约要求page_sources和每页实际read_file记录,渲染不能代替阅读。合订PDF中的每份报告都要单独提取,不能只整理第一份。'
|
|
|
- '自编Python完成时序、稳定窗口分析和报告,最后submit_main_fan_report。') +
|
|
|
- '历史取数后先读取agent_data_path的观测状态区间、变化边界、异常和风量统计。'
|
|
|
- '各日原件在单个ZIP中,按manifest的archive成员用Python按需复核;不要逐份读取或打印数十万条采样。'
|
|
|
- '切换分析必须基于具体时间而不只是开停占比。在details.switch_analysis中交付status、method、events、limitations;'
|
|
|
- '每个event记录候选或确认状态、原机停机时刻、另一台启动时刻、间隔和测点证据。'
|
|
|
- '无法确认整机时仍逐条保留相关测点的状态变化时刻作为候选,说明限制,不能仅凭占比宣称已发现切换。'
|
|
|
- '重大设备历史已有真实样例data[].sbNodeCurve[],其中dataTime为时刻、realData为原始读数。'
|
|
|
- '优先采用整理结果中的实际样本数和质量信息;这些嵌套数组有数据时不能称为“无历史数据”,全为0也须报告实际样本和数据有效性限制。'
|
|
|
- '结果仅存本地,不调用推送。模拟PDF保留原件真实身份,禁止移植为本矿正式报告。'
|
|
|
+ '本轮重点是同叶片角度、同频率下的风压、功率和效率数值表格对比。'
|
|
|
+ '读取Skill的references/performance-tables.md,用Python对齐共同风量并计算性能指标,按rules.json.curve的固定阈值逐工况判定。'
|
|
|
+ '型号和额定功率只作基础核对,不能代替性能通过;未知工况或数据不足时写明限制,不降低阈值。'
|
|
|
+ '根据整理数据分析、核对证据并用Python生成报告,最后submit_main_fan_report。'
|
|
|
+ '开停、切换和风量直接采用JSON中的states、flows、switch_candidates;'
|
|
|
+ 'comparison.before_value/after_value/change_percent就是本轮用于比较的现成结果。'
|
|
|
+ '不重新识别切换、选择风量点或计算比例,不读取原始日段或解压历史ZIP,也不补算稳定窗口、完整率、CV或均值。'
|
|
|
+ '只有JSON标记截断时,按details_path读取剩余整理结果。零值保留,缺失直接说明,不自行补查。'
|
|
|
+ 'details.switch_analysis保留status、method、events、limitations,events引用JSON中的候选ID、方向、时间范围和风量对照。'
|
|
|
+ 'time_range不是准确停启时刻或切换耗时;未提供的准确时刻与间隔不要补填。保留notes和comparison.reason中的限制。'
|
|
|
+ '由宿主负责本地保存及配置开启后的推送,Agent不调用推送。原件须核对目标矿井、通风井及风机归属,模拟PDF不得移植为本矿正式报告。'
|
|
|
'本地任务已获授权,身份或数据不足时提交受限结果,不请求人工确认。'
|
|
|
'\n本轮已准备好的唯一workflow_id:' + prepared['workflow_id'] +
|
|
|
';请使用此ID,前一阶段的ID已经结束。输入文件:' + prepared['input_path'] +
|
|
|
@@ -252,11 +326,32 @@ class BatchRunner:
|
|
|
exploration._write(scope.output / ('execution-' + phase + '.json'), {
|
|
|
'status': result.status, 'run_id': result.run_id, 'error': result.error, 'output': result.output,
|
|
|
'usage': result.usage, 'interrupts': getattr(result, 'interrupts', []),
|
|
|
- 'structured_output': result.structured_output})
|
|
|
+ 'structured_output': result.structured_output, 'diagnostics': getattr(result, 'diagnostics', {})})
|
|
|
if result.status != 'success' or not result.structured_output:
|
|
|
- raise ValueError('Agent未完成实际提交:' + (result.error or result.status))
|
|
|
+ reason = result.error or ('缺少 submit_main_fan_report 提交的正式结果'
|
|
|
+ if result.status == 'success' else result.status)
|
|
|
+ raise ValueError('Agent未完成实际提交:' + reason)
|
|
|
+ logger.info('Agent 研判完成', extra={'kv': {'event': 'batch.agent.completed',
|
|
|
+ 'workflow_id': prepared['workflow_id'], 'run_id': result.run_id, 'usage': result.usage,
|
|
|
+ 'duration_ms': round((time.monotonic() - started) * 1000)}})
|
|
|
return result.structured_output
|
|
|
|
|
|
+ def queue_delivery(self, scope, result):
|
|
|
+ folder = scope.output / scope.shaft.shaft_id
|
|
|
+ receipt = {'enabled': self.publish, 'status': 'local_only', 'task_id': result['platform_payload']['taskId']}
|
|
|
+ if self.publish:
|
|
|
+ if result.get('status') != 'completed':
|
|
|
+ receipt['status'] = 'skipped_incomplete'
|
|
|
+ else:
|
|
|
+ from extensions.result_platform.settings import ResultPlatformSettings
|
|
|
+ settings = ResultPlatformSettings.from_config()
|
|
|
+ job = business_store.record_completed_result('main_fan', scope.user, scope.session,
|
|
|
+ self.input_for(scope, 'assessment'), result, settings.url)
|
|
|
+ receipt.update(status=job['push_status'], job_id=job['id'])
|
|
|
+ logger.info('本地结果已进入持久化推送队列', extra={'kv': {'event': 'batch.delivery.queued',
|
|
|
+ 'task_id': job['id'], 'push_status': job['push_status']}})
|
|
|
+ exploration._write(folder / '推送记录.json', receipt)
|
|
|
+
|
|
|
@staticmethod
|
|
|
def export_shaft(scope, result):
|
|
|
folder = scope.output / scope.shaft.shaft_id
|
|
|
@@ -268,7 +363,7 @@ class BatchRunner:
|
|
|
exploration._write(folder / '待审阅报文.json', payload)
|
|
|
metadata = {'mine': {key: scope.mine[key] for key in ('dept_id', 'mine_id', 'mine_name')},
|
|
|
'shaft': scope.shaft.model_dump(mode='json'), 'cutoff': scope.cutoff,
|
|
|
- 'lookback_days': scope.lookback_days, 'publish': False, 'pdf_mode': 'mock',
|
|
|
+ 'lookback_days': scope.lookback_days, 'publish': scope.publish, 'pdf_mode': scope.report_mode,
|
|
|
'history_coverage': result.get('assessment', {}).get('history_coverage'),
|
|
|
'report_archives': result.get('assessment', {}).get('report_archives', [])}
|
|
|
exploration._write(folder / '来源与覆盖.json', metadata)
|