kangshifu-board/scripts/prepare-consumer-quote-inputs.py

"""Prepare fixed consumer-quote inputs and handoff files. No network or business writes."""
from pathlib import Path
import hashlib
import json
import sys
import zipfile

ROOT = Path(__file__).resolve().parents[1]
SOURCE = ROOT / 'work/consumer-research-codex-20261102/delivery'
ARCHIVE = ROOT / 'work/consumer-quotes-codex-20261102.zip'
DEST = ROOT / 'work/consumer-quote-inputs-20261102'
DOCS = Path(r'F:\文档\工作坊当前资料')
TASK = 'KSF-消费者原话-20261102-01'
ARCHIVE_SHA = 'df741b09e0540c21cc311ea81e560beb87c0ec7a85a1872ad704f17a7b5e4ac1'
REMOTE_INPUTS = '/opt/data/ksf-executor-20261102/consumer-quotes-01'
REMOTE_STATE = '/opt/data/ksf-workshop-private/consumer-quotes/KSF-consumer-quotes-20261102-01'
REMOTE_ARCHIVE = '/opt/data/ksf-workshop-private/archive-packages/consumer-quotes-codex-20261102.zip'
PACKAGE = DOCS / 'ksf-consumer-quotes-import-20261102.zip'
sha = lambda blob: hashlib.sha256(blob).hexdigest()
encode = lambda value: (json.dumps(value, ensure_ascii=False, indent=2) + '\n').encode('utf-8')


def retain(path, blob):
    if path.exists():
        assert path.read_bytes() == blob, f'Existing file differs; preserve it: {path}'
    else:
        path.parent.mkdir(parents=True, exist_ok=True)
        with path.open('xb') as f:
            f.write(blob)


def prepare():
    archive = ARCHIVE.read_bytes()
    assert len(archive) == 1287263 and sha(archive) == ARCHIVE_SHA
    with zipfile.ZipFile(ARCHIVE) as z:
        assert z.testzip() is None and len(z.namelist()) == 30
        manifest = z.read('FILES.sha256').decode('utf-8').splitlines()
        assert len(manifest) == 29
        for line in manifest:
            expected, name = line.split('  ', 1)
            assert sha(z.read(name)) == expected
            assert z.read(name) == (SOURCE / name).read_bytes()
        data = json.loads(z.read('consumer-quotes.json'))
    assert len(data['records']) == 4 and data['businessDataWritten'] is False
    inputs = {}
    for row in data['records']:
        key = row['id']
        original_locator = row['locator']
        raw = row['completeMainBody']
        start, end = original_locator['startCharacter'], original_locator['endCharacter']
        assert raw[start:end] == row['quote']
        assert sha(raw.encode('utf-8')) == row['hashes']['mainTextSha256']
        locator = {
            'url': row['sourceUrl'], 'section': '原帖主文章正文中的使用体验自述',
            'startCharacter': start, 'endCharacter': end,
            'unit': 'Unicode code point', 'startBase': 0, 'endExclusive': True,
            'publishedAtReported': row['publishedAtReported'], 'publicationTimezone': 'unknown',
            'httpCollectedAt': row['httpCollectedAt'], 'browserReadAt': row['browserCollectedAt'],
            'archiveSha256': ARCHIVE_SHA,
            'textOrigin': {'file': original_locator['file'], 'sha256': row['hashes']['mainTextSha256'],
                           'extraction': original_locator['extraction']},
            'responseOriginal': {'file': row['files']['response'], 'sha256': row['hashes']['originalResponseSha256']},
            'screenshot': {'file': row['files']['screenshot'], 'sha256': row['hashes']['screenshotSha256']},
            'articleDom': {'file': row['files']['articleDom'], 'sha256': row['hashes']['articleDomSha256']},
            'browserReadFile': row['files']['browserRead'],
            'collectionPerformedBy': 'Codex',
            'productScope': row['productScope'], 'productIdentificationBasis': row['productIdentificationBasis'],
            'context': row['context'], 'purchaseVerified': False, 'realIdentityVerified': False,
            'commercialRelationship': 'unknown', 'visibleTopic': row['visibleTopic'],
            'limitations': data['limits'],
        }
        item = {
            'itemKey': key + '-experience', 'className': 'KfEvidence',
            'externalKey': TASK + ':' + key + ':experience:v1',
            'data': {
                'dataMode': 'live', 'quote': row['quote'],
                'evidenceKind': 'public_user_experience_self_report',
                'platform': 'smzdm', 'authorLabel': row['authorDisplay'],
                'parentObjectId': row['sourceUrl'].rstrip('/').split('/')[-1],
                'quoteForm': 'verbatim_excerpt',
                'verificationState': 'archived_text_and_screenshot_checked',
                'publishedAt': None, 'collectedAt': row['browserCollectedAt'],
                'locator': locator,
            },
        }
        task = {
            'source': {'title': row['title'], 'sourceKind': 'social', 'dataMode': 'live',
                       'sourceUrl': row['sourceUrl'], 'platform': 'smzdm',
                       'rawText': raw, 'collectedAt': row['httpCollectedAt']},
            'records': [item],
        }
        assert 'productRef' not in item['data'] and 'sourceRef' not in item['data']
        inputs[key + '.input.json'] = task
        retain(DEST / (key + '.input.json'), encode(task))

    review = {
        'ok': True, 'operation': 'consumer-quote-draft-reviewed',
        'basis': '王刚001草稿由用户转述；原话与固定本地原件独立对照。远端文件权限和其哈希验算只记为王刚001回报。',
        'taskKey': TASK, 'originalArchiveSha256': ARCHIVE_SHA, 'originalArchiveBytes': len(archive),
        'sourceCount': 4, 'evidenceCount': 4, 'platformCount': 1,
        'corrections': [
            '明确表达面量少的是①cq-03、②cq-04，共2篇；③cq-02没有说量少。问题表中“③②同系列”应为“①②”。',
            '①既说“面量真少”“真不够吃”，也批评青豆；不能说唯一明确批评是青豆，也不将“看着还可以吧”写成整体正面。',
            '“作为一餐是否足够”是研究问题，不是①②的逐字原话；自加食材的动机与具体用餐场景保持待验证。',
            '最新帖发表于2024-10-10，距2026-11-02已超过2年。“近两年没有样本”成立；“快2年”应改为“超过2年”。',
            '④是作者留水改善口感的经验，不能推断所有留水量的效果，也不能因作者语气正面就断言其从未感觉干。',
            '“吃腻泡面想尝鲜”“元旦前买”等明确自述可保留，人口特征、真实用餐场景、动机及需求强度不能补写。',
        ],
        'sampleObservations': {'portionComments': ['cq-03', 'cq-04'],
                               'boilingMentioned': ['cq-02', 'cq-03', 'cq-04'],
                               'unnoticedBeforePurchase': ['cq-02'],
                               'drainingWaterAdvice': ['cq-01'], 'peaDislike': ['cq-03']},
        'sampleProportionsArePopulationEstimates': False,
        'draftReadyForInputPreparation': True, 's2Saved': False,
        'inputHashes': {name: sha(encode(value)) for name, value in inputs.items()},
        'businessWrites': 0,
    }
    retain(DEST / 'draft-review.json', encode(review))
    review_text = '''消费者原话草稿核对结论

草稿可以用于继续准备资料入库，尚不是已保存的S2成果。四篇原话、出处与产品区分符合原件，保留正负上下文、小样本和历史时间边界。

保存前修正以下表述：
1. 量少的是①和②，共2篇，不是3篇；问题表中“③②”应改成“①②”。
2. ①明确批评了面量和青豆，不能说唯一批评是青豆。“看着还可以吧”指加料后的外观，不足以断定整体正面。
3. “作为一餐是否足够”是我们提出的研究问题，不是作者逐字说的场景。自加挂面、蛋、菜的实际动机待验证。
4. 最新一篇为2024年10月10日，距2026年11月2日已超过2年；“近两年没有样本”成立，“快2年”应改为“超过2年”。
5. 留水建议属于作者改善口感的经验，不推出普遍嫌干或普遍不嫌干。已有的尝鲜、元旦前购买等原话可保留，未明示的人群和用餐场景保持未知。

下一步只导入4份完整正文及4条主引文，不创建产品或分类，不把燃魂与香辣合成SKU。来源为公开用户自述，购买、真实身份及商业关系未认证。发表时间只保留页面原字符串，时区未知；本次取得时间使用原采集日志，不混淆两者。

固定输入、原件哈希与本地隔离预检完成后交运维准备，再由王刚001真实消息任务执行。入库后独立回读通过，才准备S2保存模板；S1第2版G8IrkkDG7X保持原样。
'''
    retain(DOCS / '19-消费者原话草稿核对结果.txt', review_text.encode('utf-8'))
    return inputs, review, review_text


def package(inputs, review, review_text):
    verification = ROOT / 'work/consumer-quote-inputs-verified-20261102.json'
    local = json.loads(verification.read_text('utf-8'))
    assert local['ok'] and local['productionBusinessWrites'] == 0 and len(local['checks']) == 4
    assert local['inputHashes'] == review['inputHashes']
    ops = f'''运维001：准备4篇消费者原话入库。只落地固定输入和留存目录，不代跑业务。

包：ksf-consumer-quotes-import-20261102.zip。先核FILES.sha256全部项。inputs下4份cq-01至cq-04.input.json原样放到{REMOTE_INPUTS}/，root:root 0644、hermes可读。沿用现有ksf-secure.mjs，不安装新执行器，不改Skill/Function/schema，不重启。

包内archives/consumer-quotes-codex-20261102.zip为原采集包原字节，1287263B，SHA-256={ARCHIVE_SHA}。保存到{REMOTE_ARCHIVE}，hermes:hermes 0600，父目录0700。若该路径已有同字节原包，核对后直接复用；已有不同文件则保留现场并停止，不覆盖、不重打包。

王刚001已回报解压原件目录/opt/data/ksf-workshop-private/consumer-quotes-codex-20261102/；若存在，核其29项FILES.sha256与原包一致后复用，不覆盖。若不存在，用已核原包在全新目录解压，目录0700、文件0600。不得将既有0条采集目录改写或清空。

只读核对固定任务键{TASK}的历史、本地回执与在途，并按输入4个sourceUrl、source.rawText内容哈希及4个externalKey查服务端。已有相同资料可复用；如发现该任务已执行或在途，回报实际编号并暂停，不换键重跑。

以hermes(10000)、HOME=/opt/data准备留存父目录及{REMOTE_STATE}，0700。目标若已存在，不清空或换目录，先核历史。本批只计划KfSource+KfEvidence，不需要新身份文件。

回报：4份输入哈希/权限、同一原包实际路径/字节/完整哈希及hermes实读、原件清单核对、历史/在途与留存目录状态。没有执行workshop-import就是预期。准备完成后才交王刚001按包内执行单入库一次；无需立即保存S2。
'''
    wang = f'''王刚001：运维准备完成后，在一条新的真实消息任务里，使用本条已核实的现有agentProfileKey，通过现有secure入口导入4份固定消费者原话输入。你不需要重新设计或修改内容。

草稿修正：量少为①②共2篇；①同时批评面量和青豆；“作为一餐”是研究问题；最新帖距今超过2年。完整核对见包内草稿核对结果.txt。

固定任务键{TASK}。沿用身份，不登记新Agent或员工。先查本任务无既有执行回执、无在途；若有则只回报，不重放。以hermes(10000)、HOME=/opt/data执行。

按cq-01、cq-02、cq-03、cq-04顺序逐份执行。前一份导入及来源/证据精确回读都通过后才做下一份。下面以cq-01为例，其余只将命令和回执前缀中的cq-01替换为对应cq-02/03/04；不改输入、不换任务键。

1. 只读查重：
node /opt/data/skills/ksf-workshop-runtime/scripts/ksf-secure.mjs workshop-dedup --input {REMOTE_INPUTS}/cq-01.input.json --request-id {TASK}:cq-01:preflight:v1

新来源no-exact-match、条目deferred/SOURCE_NOT_REGISTERED是正常组合；来源unchanged和条目reuse也是可接受组合。先确认外层ok和entryResult.ok都为true，只有以上可解释结果才继续，其他异常停止回报。

2. 导入一次：
node /opt/data/skills/ksf-workshop-runtime/scripts/ksf-secure.mjs workshop-import --agent-profile <本条已核实agentProfileKey> --input {REMOTE_INPUTS}/cq-01.input.json --request-id {TASK}:cq-01:import:v1

核外层ok和entryResult.ok均为true，取entryResult.result的sourceId、sourceOperation、items[0].objectId、items[0].jobId、operation及counts。不以退出码替代两层回执核对。

3. 分别用真实返回编号独占写两份只读输入（0600），格式为{{"className":"KfSource","objectId":"<实际sourceId>"}}和{{"className":"KfEvidence","objectId":"<实际evidenceId>"}}，然后：
node /opt/data/skills/ksf-workshop-runtime/scripts/ksf-secure.mjs workshop-read --input <来源回读输入绝对路径> --request-id {TASK}:cq-01:read-source:v1
node /opt/data/skills/ksf-workshop-runtime/scripts/ksf-secure.mjs workshop-read --input <证据回读输入绝对路径> --request-id {TASK}:cq-01:read-evidence:v1

核对来源输入各字段逐项一致：social/live/smzdm、标题、URL、完整正文、collectedAt；证据输入各字段逐项一致：quote、locator、authorLabel、parentObjectId、evidenceKind=public_user_experience_self_report、verificationState=archived_text_and_screenshot_checked、quoteForm=verbatim_excerpt、publishedAt=null和实际collectedAt。sourceRef指回本次来源，productRef为空。locator按Unicode码点切片等于quote，原包/正文/截图哈希与实际文件吻合。发表时间保留locator.publishedAtReported原字符串，时区未知；不得用采集时间替代发表时间。不能只看“读接口成功”就算内容通过。

两份回读均通过后再处理下一份。所有命令各执行一次。输入快照、只读输入、stdout/stderr、退出码和两层回执独占留存到{REMOTE_STATE}/，0600，不覆盖既有文件。查重、写入、回读任何一步失败即停，保留原requestId和已取得编号；不自动重试/换键/删记录，错误不代表零写入。

成功回报4篇各自sourceId、evidenceId、jobId、sourceOperation、created/unchanged、import两层invocationId及逐项回读。可为unchanged，不为凑新增改键。collectionPerformedBy为Codex，实际导入执行者为你，不声称你刚刚重新抓取了平台。

本次只做原话资料入库：4来源、每来源1条主引文；不建KfCategory/KfProduct，不保存S2，不生成机会卡/项目，不改S1 G8IrkkDG7X，不重跑旧K轮或旧导入。3篇燃魂、1篇香辣不合SKU，均不挂到-KbX0Agka-。不称已验证购买、真实身份或普遍需求，也不据历史帖子认定当前商品价格/规格/库存。
'''
    retain(DOCS / '20-发给运维001-消费者原话入库准备.txt', ops.encode('utf-8'))
    retain(DOCS / '21-发给王刚001-消费者原话入库.txt', wang.encode('utf-8'))
    blobs = {'inputs/' + name: encode(value) for name, value in inputs.items()}
    blobs.update({
        'archives/consumer-quotes-codex-20261102.zip': ARCHIVE.read_bytes(),
        'draft-review.json': encode(review),
        '草稿核对结果.txt': review_text.encode('utf-8'),
        '给运维001.txt': ops.encode('utf-8'), '给王刚001.txt': wang.encode('utf-8'),
        'local-verification.json': verification.read_bytes(),
    })
    history_file = ROOT / 'work/consumer-quote-import-history-20261102.json'
    if history_file.exists():
        blobs['readonly-history.json'] = history_file.read_bytes()
    blobs['FILES.sha256'] = ''.join(sha(b) + '  ' + n + '\n' for n, b in blobs.items()).encode('utf-8')
    if PACKAGE.exists():
        with zipfile.ZipFile(PACKAGE) as z:
            assert set(z.namelist()) == set(blobs)
            for n, b in blobs.items():
                assert z.read(n) == b
    else:
        with zipfile.ZipFile(PACKAGE, 'x', compression=zipfile.ZIP_DEFLATED) as z:
            for name, blob in blobs.items():
                z.writestr(name, blob)
    with zipfile.ZipFile(PACKAGE) as z:
        assert z.testzip() is None
        for line in z.read('FILES.sha256').decode('utf-8').splitlines():
            expected, name = line.split('  ', 1)
            assert sha(z.read(name)) == expected
    receipt = {'ok': True, 'operation': 'consumer-quote-inputs-packaged',
               'taskKey': TASK, 'file': str(PACKAGE), 'bytes': PACKAGE.stat().st_size,
               'sha256': sha(PACKAGE.read_bytes()), 'manifestChecked': len(blobs) - 1,
               'sourcesPlanned': 4, 'evidencePlanned': 4, 'productsPlanned': 0,
               'reportsPlanned': 0, 'targetExecuted': False, 'businessWrites': 0,
               'originalArchiveSha256': ARCHIVE_SHA, 'inputHashes': review['inputHashes']}
    retain(ROOT / 'work/consumer-quote-import-package-receipt-20261102.json', encode(receipt))
    return receipt


if __name__ == '__main__':
    sys.stdout.reconfigure(encoding='utf-8')
    prepared = prepare()
    if '--package' in sys.argv:
        print(json.dumps(package(*prepared), ensure_ascii=False, indent=2))
    else:
        print(json.dumps({'ok': True, 'operation': 'consumer-quote-inputs-prepared',
                          'taskKey': TASK, 'inputHashes': prepared[1]['inputHashes'],
                          'businessWrites': 0}, ensure_ascii=False, indent=2))