"""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))