kangshifu-board/scripts/record-workshop-dialogue-r13.py

"""Append an explicitly simulated conversation, preserving the exact r13 handbook prompts."""
import argparse
import concurrent.futures
import datetime
import hashlib
import html
import importlib.util
import json
import shutil
import urllib.error
import urllib.request
from decimal import Decimal
from pathlib import Path

ROOT=Path(__file__).resolve().parents[1]
BASE=ROOT/'docs/workshop-quality-r13-20261110'
OUT=ROOT/'docs/workshop-dialogue-r13-20261110'
WORK=ROOT/'work/workshop-dialogue-r13-20261110'
sha=lambda p:hashlib.sha256(p.read_bytes()).hexdigest()
dump=lambda p,d:p.write_text(json.dumps(d,ensure_ascii=False,indent=2)+'\n',encoding='utf8')
NOTICE='本记录由同一助手模拟员工和研究助手，按r13手册的15节顺序推进。员工的目标、回答、取舍与质询均为模拟，不是真实员工输入；工具执行另有原始回执。不是豆包客户端或独立双模型实测。只在本地保存，不写业务库、不建企业账套。'


def read_json(p):return json.loads(p.read_bytes())


def initialize():
    OUT.mkdir(exist_ok=False)
    (OUT/'evidence').mkdir()
    (OUT/'阶段产出').mkdir()
    data=read_json(BASE/'prompts-and-outputs.json')
    shutil.copyfile(BASE/'prompts-and-outputs.json',OUT/'使用的手册提示词.json')
    shutil.copyfile(BASE/'证据与编码-r13.json',OUT/'使用的证据与编码.json')
    meta={'schema':'ksf.workshop-simulated-dialogue.v1','version':'r13-local-quality','notice':NOTICE,
          'startedAt':datetime.datetime.now(datetime.timezone(datetime.timedelta(hours=8))).isoformat(),
          'manual':str(BASE/'康师傅工作坊操作手册-r13本地修订.html'),
          'manualSha256':sha(BASE/'康师傅工作坊操作手册-r13本地修订.html'),
          'skillRoot':str(BASE/'installation/skills/ksf-workshop-runtime'),
          'skillFiles':read_json(BASE/'installation/release.json')['workshop']['files'],
          'promptsSha256':sha(OUT/'使用的手册提示词.json'),
          'evidenceLedgerSha256':sha(OUT/'使用的证据与编码.json'),
          'order':[s['id'] for s in data['steps']],
          'simulation':True,'employeeStatementsAreSimulated':True,'independentModels':False,
          'actualHostAutoLoading':False,'productionWrites':0,'completedStages':[]}
    dump(OUT/'运行说明.json',meta)
    (OUT/'对话原始记录.jsonl').write_text('',encoding='utf8')
    print(json.dumps({'initialized':True,'root':str(OUT)},ensure_ascii=False))


def messages():
    return [json.loads(line) for line in (OUT/'对话原始记录.jsonl').read_text('utf8').splitlines() if line]


def add(stage,role,content,kind='dialogue',**extra):
    row={'seq':len(messages())+1,'stage':stage,'role':role,'kind':kind,'content':content,
         'simulation':role!='execution',**extra}
    with (OUT/'对话原始记录.jsonl').open('a',encoding='utf8') as f:f.write(json.dumps(row,ensure_ascii=False)+'\n')
    return row


def probe(name,body):
    assert body['op'] in {'contract','version','browse','detail'}
    path=OUT/'evidence'/(name+'.json')
    if path.exists():return read_json(path)
    req=urllib.request.Request('https://kangshifu.fmode.cn/api/library',data=json.dumps(body,ensure_ascii=False).encode(),
        headers={'Content-Type':'application/json'},method='POST')
    try:
        try:response=urllib.request.urlopen(req,timeout=25)
        except urllib.error.HTTPError as exc:response=exc
        raw=response.read();status=response.code
        try:d=json.loads(raw)
        except ValueError:d={'ok':False,'error':'NON_JSON_RESPONSE'}
    except (OSError,urllib.error.URLError,TimeoutError) as exc:
        raw=b'';status=None;d={'ok':False,'error':type(exc).__name__}
    path.write_bytes(raw if raw else json.dumps(d).encode())
    receipt={'name':name,'request':body,'url':req.full_url,'httpStatus':status,
        'checkedAt':datetime.datetime.now(datetime.timezone.utc).isoformat(),
        'responseSha256':sha(path),'ok':d.get('ok'),'generation':d.get('generation'),
        'readOnly':d.get('readOnly'),'databaseWrites':d.get('databaseWrites'),'error':d.get('error')}
    dump(OUT/'evidence'/(name+'.request.json'),receipt)
    if d.get('ok'):assert d.get('readOnly') is True and d.get('databaseWrites')==0
    return d


def stage_start(stage):
    meta=read_json(OUT/'运行说明.json');index=len(meta['completedStages'])
    assert meta['order'][index]==stage,(stage,index)
    s=next(s for s in read_json(OUT/'使用的手册提示词.json')['steps'] if s['id']==stage)
    add(stage,'employee',s['fullPrompt'],kind='handbook-prompt',title=s['title'],
        promptSha256=hashlib.sha256(s['fullPrompt'].encode()).hexdigest())


def append_stage(stage):
    from workshop_dialogue_r13_content import TURNS
    stage_start(stage)
    if stage=='install':
        receipt=read_json(WORK/'install-result/install-receipt.json')
        assert receipt['ok'] and receipt['filesInstalled']
        shutil.copyfile(WORK/'install-result/install-receipt.json',OUT/'evidence/install-receipt.json')
        add(stage,'execution','实际在work下的新隔离目录执行r13安装器；安装回执见evidence/install-receipt.json。既有宿主技能目录未改。',kind='actual-execution')
    if stage=='category':
        # Two independent read-only capability checks; the content run reuses pinned local research material.
        jobs=[('library-version',{'op':'version'}),('library-contract',{'op':'contract'})]
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:
            results=list(pool.map(lambda job:probe(*job),jobs))
        add(stage,'execution','实际只读核查当前资料入口。'+json.dumps([{'name':n,'ok':d.get('ok'),'generation':d.get('generation'),'error':d.get('error')} for (n,_),d in zip(jobs,results)],ensure_ascii=False),kind='actual-execution')
    if stage=='research':
        gen=read_json(OUT/'使用的证据与编码.json')['generation']
        jobs=[('mixed-products',dict(op='browse',entity='products',filters={'brand':'ksf','category':'banmian'},keyword='混合',limit=50,generation=gen)),
              ('four-bowls',dict(op='browse',entity='products',filters={'brand':'ksf','category':'banmian'},keyword='4碗',limit=50,generation=gen))]
        with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool:results=list(pool.map(lambda job:probe(*job),jobs))
        add(stage,'execution','按T1实际追加两项只读补查。'+json.dumps([{'name':n,'ok':d.get('ok'),'matched':d.get('totalMatched'),'returned':len(d.get('records',[])),'generation':d.get('generation'),'error':d.get('error')} for (n,_),d in zip(jobs,results)],ensure_ascii=False),kind='actual-execution')
        for name in ['existing-two-flavour','existing-custom-six']:
            r=read_json(OUT/'evidence'/(name+'.request.json'));assert r['ok']
            add(stage,'execution','从补查命中实际精确回读商品：'+json.dumps({'request':r['request'],'ok':r['ok'],'generation':r['generation'],'readOnly':r['readOnly'],'databaseWrites':r['databaseWrites']},ensure_ascii=False),kind='actual-execution')
    if stage=='feasibility':
        values={'R0':'20','V0':'15','Rn':'21','Vn':'15.5','F':'1000','Qn':'100','Qc':'1000'}
        v={k:Decimal(n) for k,n in values.items()}
        u0=v['R0']-v['V0'];un=v['Rn']-v['Vn'];du=un-u0
        value=v['Qn']*un+v['Qc']*du-v['F']
        assert value==50 and du==Decimal('.5')
        dump(OUT/'evidence/假设算术复算.json',{'inputs':values,'empiricalCompanyData':False,'unit':'元/六碗销售单位；非报价或销量预测',
             'result':{'u0':str(u0),'un':str(un),'deltaUnit':str(du),'incrementalContribution':str(value),'sameVolumeBreakEven':2000}})
        add(stage,'execution','实际本地复算一组明确为假设的数字，结果见evidence/假设算术复算.json；没有获取企业成本或交易数据。',kind='actual-execution')
    for row in TURNS[stage]:
        add(stage,row['role'],row['content'],kind=row.get('kind','dialogue'))
    meta=read_json(OUT/'运行说明.json');meta['completedStages'].append(stage);dump(OUT/'运行说明.json',meta)
    out=[r['content'] for r in messages() if r['stage']==stage and r['kind']=='stage-output']
    assert out,stage
    (OUT/'阶段产出'/f'{len(meta["completedStages"]):02}-{stage}.txt').write_text('\n\n'.join(out),encoding='utf-8-sig')
    render()
    print(json.dumps({'stage':stage,'messages':len(messages()),'stagesCompleted':len(meta['completedStages'])},ensure_ascii=False))


def render():
    meta=read_json(OUT/'运行说明.json');rows=messages();steps={s['id']:s for s in read_json(OUT/'使用的手册提示词.json')['steps']}
    text=['康师傅工作坊｜r13 本地模拟对话','',NOTICE,'','员工的回答是本次模拟设定，不是企业事实；本地引用键不是看板编号。原文提及/观察价不代表总体需求率/企业净收入。','']
    roles={'employee':'员工（模拟）','assistant':'助手（模拟）','execution':'执行记录（实际）'}
    blocks=[];last=None
    for r in rows:
        if last!=r['stage']:
            title=steps[r['stage']]['number']+' '+steps[r['stage']]['title']
            text+=['','='*35,title,'='*35,'']
            blocks.append('<h2 id="'+r['stage']+'">'+html.escape(title)+'</h2>');last=r['stage']
        label=f'{r["seq"]:03}　{roles[r["role"]]}'
        if r['kind']=='handbook-prompt':label+='｜粘贴手册原文'
        if r['kind']=='stage-output':label+='｜本节产出'
        text += [label,r['content'],'']
        blocks.append('<div class="message '+r['role']+'"><b>'+html.escape(label)+'</b><div class="body">'+html.escape(r['content'])+'</div></div>')
    (OUT/'完整模拟对话.txt').write_text('\n'.join(text),encoding='utf-8-sig')
    page='''<!doctype html><html lang="zh-CN"><meta charset="utf-8"><meta name="viewport" content="width=device-width,initial-scale=1"><title>康师傅工作坊｜本地模拟对话</title><style>body{max-width:960px;margin:24px auto;padding:0 18px;font:15px/1.8 "Microsoft YaHei",sans-serif;color:#222}h1{font-size:23px}h2{font-size:18px;margin:35px 0 15px;border-top:2px solid #444;padding-top:12px}.note{color:#666;font-size:13px}.message{padding:14px 0;border-bottom:1px solid #ddd}.message.employee{background:#f7f7f7;padding:14px 12px}.message.execution{font-size:12px;color:#666}.body{white-space:pre-wrap;overflow-wrap:anywhere;margin-top:7px}a{color:#1458a2}b{font-size:13px}@media(max-width:600px){body{margin:14px auto;font-size:14px}}</style><h1>康师傅工作坊｜r13 本地模拟对话</h1><p class="note">'''+html.escape(NOTICE)+'''</p><p><a href="完整模拟对话.txt">纯文本记录</a> · <a href="对话原始记录.jsonl">逐条原始记录</a></p>'''+''.join(blocks)+'</html>'
    (OUT/'完整模拟对话.html').write_text(page,encoding='utf8')


if __name__=='__main__':
    p=argparse.ArgumentParser();p.add_argument('stage');a=p.parse_args()
    if a.stage=='init':initialize()
    elif a.stage=='render':render()
    else:append_stage(a.stage)