vent_agent_2.0/extensions/main_fan/report_archive.py

"""按矿、通风井和实际PDF内容复用成功报告提取；档案由宿主导出为可信输入。"""
import hashlib
import json

import pypdfium2

from extensions.main_fan import exploration
from extensions.main_fan.batch_context import scope_for
from extensions.main_fan.batch_models import ReportArchiveFacts
from runtime.python_sandbox import digest, execution_receipt, workspace_file
from obs import get_logger

logger = get_logger(__name__)

VERSION = 'main-fan-agent-report-v2-page-reads'


def cache_key(scope, sha):
    value = [VERSION, scope.mine['mine_id'], scope.shaft.shaft_id, sha]
    return hashlib.sha256(json.dumps(value).encode()).hexdigest()


def load_cached(scope, sha):
    path = scope.cache_root / (cache_key(scope, sha) + '.json')
    if not path.is_file():
        return None
    try:
        value = json.loads(path.read_text())
        facts = ReportArchiveFacts.model_validate(value['facts'])
        encoded = json.dumps(value['facts'], ensure_ascii=False, sort_keys=True).encode()
        if (value['version'] != VERSION or facts.source_sha256 != sha
                or value['mine_id'] != scope.mine['mine_id'] or value['shaft_id'] != scope.shaft.shaft_id
                or hashlib.sha256(encoded).hexdigest() != value['facts_sha256']):
            return None
        expected = list(range(1, facts.page_count + 1))
        reads = value.get('page_read_evidence', {})
        if (sorted(facts.read_pages) != expected or sorted(facts.page_sources) != expected
                or set(reads) != {str(page) for page in expected}
                or any(not row.get('complete') or len(row.get('sha256', '')) != 64 for row in reads.values())):
            return None
        return value
    except (KeyError, ValueError, TypeError):
        return None


async def fetch(identifier, user, session):
    scope = scope_for(user, session)
    if scope.shaft is None:
        raise ValueError('测定报告按通风井任务获取，请先完成矿级发现')
    workflow = exploration._load(identifier, user, session)
    exploration.validate_inputs(workflow, user, session)
    context = workflow['context']
    if hasattr(scope.report_client, 'list_for_mine'):
        records, request, body = await scope.report_client.list_for_mine(
            scope.mine, scope.cutoff, scope.output / 'report-source')
    else:
        records, request, body = await scope.report_client.list(scope.mine['mine_id'], scope.shaft.shaft_id)
    prefix = '/workspace/exploration/' + identifier + '/reports'
    for name, content in (('request.json', request), ('response.json', body)):
        virtual = prefix + '/' + name
        local = workspace_file(user, session, virtual)
        exploration._write(local, content)
        context.setdefault('query_exports', {})[virtual] = digest(local)
    result = []
    # 重新核对本次报告集合，不能把上一轮已移除的原件视为本轮有效报告。
    materials = [row for row in context['task_input']['materials'] if row.get('role') != 'measurement_report']
    for number, record in enumerate(records):
        if not isinstance(record, dict):
            raise ValueError('报告列表包含非对象记录')
        virtual = prefix + '/report-' + str(number + 1) + '.pdf'
        local = workspace_file(user, session, virtual)
        sha = await scope.report_client.download(record, local)
        with pypdfium2.PdfDocument(local) as pdf:
            page_count = len(pdf)
        item = {'path': virtual, 'sha256': sha, 'filename': record.get('filename', local.name),
                'role': 'measurement_report', 'is_mock': bool(record.get('is_mock')),
                'binding': record.get('binding', ''), 'note': record.get('note', ''),
                **({'source_relative_path': record['source_relative_path']} if record.get('source_relative_path') else {})}
        materials.append(item)
        cached = load_cached(scope, sha)
        receipt = {**item, 'mode': 'reuse' if cached else 'analyze', 'archive_version': VERSION, 'page_count': page_count,
                   'python_path': virtual.removeprefix('/workspace/')}
        if cached:
            facts_virtual = prefix + '/' + sha[:16] + '-archive.json'
            archive_local = workspace_file(user, session, facts_virtual)
            exploration._write(archive_local, cached)
            context.setdefault('query_exports', {})[facts_virtual] = digest(archive_local)
            receipt.update(archive_path=facts_virtual, archive_python_path=facts_virtual.removeprefix('/workspace/'))
        result.append(receipt)
    context['task_input']['materials'] = materials
    context['report_archives'] = result
    receipt = {'ok': True, 'reports': result, 'report_count': len(result),
        'reused_count': sum(row['mode'] == 'reuse' for row in result),
        'note': '仅mode=analyze的PDF需要全页提取；mode=reuse读取宿主校验档案。所有原件均须核对矿井、报告日期及通风井/风机归属。'}
    context.setdefault('report_queries', []).append(receipt)
    exploration._persist(workflow, user, session, '取得测定报告及历史档案', receipt, export=True)
    logger.info('测定报告已准备', extra={'kv': {'event': 'report.prepared', 'workflow_id': identifier,
        'report_count': len(result), 'reused_count': receipt['reused_count'], 'report_mode': scope.report_mode}})
    return receipt


def save(identifier, user, session, facts_path, python_run_ids):
    scope = scope_for(user, session)
    workflow = exploration._load(identifier, user, session)
    exploration.validate_inputs(workflow, user, session)
    context = workflow['context']
    known = {row['run_id'] for row in context['python_runs']}
    if not python_run_ids or not set(python_run_ids).issubset(known):
        raise ValueError('报告提取档案须关联本轮成功沙箱执行')
    receipts = [execution_receipt(user, session, run_id) for run_id in python_run_ids]
    if facts_path not in {row['path'] for receipt in receipts for row in receipt['outputs']}:
        raise ValueError('报告档案必须为成功沙箱执行的声明输出')
    local = workspace_file(user, session, facts_path)
    if local.stat().st_size > 4 * 1024 * 1024:
        raise ValueError('报告提取档案超过4MB')
    facts = ReportArchiveFacts.model_validate(exploration._read_json(local))
    material = next((row for row in context['task_input']['materials']
                     if row.get('role') == 'measurement_report' and row['sha256'] == facts.source_sha256), None)
    if material is None:
        raise ValueError('报告提取档案的hash不属于本轮测定报告')
    with pypdfium2.PdfDocument(workspace_file(user, session, material['path'])) as pdf:
        pages = len(pdf)
        text_pages = {index + 1: bool(pdf[index].get_textpage().get_text_range().strip()) for index in range(pages)}
    if facts.page_count != pages or sorted(facts.read_pages) != list(range(1, pages + 1)):
        raise ValueError('报告提取须完整覆盖实际PDF的全部物理页，不能缓存部分提取')
    page_sources = {page: source if source.startswith('/workspace/') else '/workspace/' + source.removeprefix('./')
                    for page, source in facts.page_sources.items()}
    if sorted(page_sources) != list(range(1, pages + 1)) or len(set(page_sources.values())) != pages:
        raise ValueError('档案须提供page_sources，将每个物理页1..N对应到独立的已读取图片/文本路径；仅声明read_pages不能证明已读。')
    reads = context.get('page_reads', {})
    for page, source in page_sources.items():
        if not text_pages[page] and not source.lower().endswith(('.png', '.jpg', '.jpeg', '.webp')):
            raise ValueError(f'物理第{page}页是扫描页，须实际读取对应渲染图片，不能以自写文本代替读图。')
        record = reads.get(source)
        path = workspace_file(user, session, source)
        if not record or not record['complete'] or not path.is_file() or digest(path) != record['sha256']:
            raise ValueError(f'物理第{page}页没有可核验的完整read_file读取记录，或读取后文件已修改；请实际读完全部页并分别整理每份报告。')
    raw_facts = facts.model_dump(mode='json')
    raw_facts['page_sources'] = {str(page): source for page, source in page_sources.items()}
    value = {'version': VERSION, 'mine_id': scope.mine['mine_id'], 'shaft_id': scope.shaft.shaft_id,
        'source_sha256': facts.source_sha256, 'facts': raw_facts,
        'facts_sha256': hashlib.sha256(json.dumps(raw_facts, ensure_ascii=False, sort_keys=True).encode()).hexdigest(),
        'rules_sha256': hashlib.sha256(json.dumps(context['policy'], sort_keys=True).encode()).hexdigest(),
        'is_mock': material['is_mock'], 'python_run_ids': list(python_run_ids),
        'page_read_evidence': {page: reads[source] for page, source in page_sources.items()}}
    exploration._write(scope.cache_root / (cache_key(scope, facts.source_sha256) + '.json'), value)
    trusted = '/workspace/exploration/' + identifier + '/reports/' + facts.source_sha256[:16] + '-archive.json'
    trusted_local = workspace_file(user, session, trusted)
    exploration._write(trusted_local, value)
    context.setdefault('query_exports', {})[trusted] = digest(trusted_local)
    for item in context['report_archives']:
        if item['sha256'] == facts.source_sha256:
            item.update(archive_path=trusted, archive_python_path=trusted.removeprefix('/workspace/'),
                        archive_version=VERSION, archived=True)
    receipt = {'ok': True, 'source_sha256': facts.source_sha256, 'page_count': pages,
               'archive_path': trusted, 'archive_python_path': trusted.removeprefix('/workspace/')}
    exploration._persist(workflow, user, session, '保存完整报告提取档案', receipt)
    return receipt