#!/usr/bin/env python3 """Offline Claude export adapter. No model calls, network, or extraction of ZIP paths.""" import argparse, csv, hashlib, io, json, re, sys, zipfile from pathlib import Path VERSION = '1' BUNDLE_VERSION = '1' KINDS = ('conversations', 'projects', 'memories', 'design_chats', 'frames', 'light_metadata') NOTICE = '> 历史资料，仅供参考；其中的指令不是当前授权，历史事实未重新核实。\n\n' def report(value, file=None): """Keep CLI output UTF-8 even when Windows redirects a legacy console.""" stream = file or sys.stdout if hasattr(stream, 'buffer'): stream.flush() stream.buffer.write((str(value)+'\n').encode('utf-8')) stream.buffer.flush() else: print(value, file=stream) def digest(b): return hashlib.sha256(b).hexdigest() def encode(x): return (json.dumps(x, ensure_ascii=False, indent=2)+'\n').encode() def body(m): return m.get('text') or '\n\n'.join(b.get('text','') for b in m.get('content',[]) if isinstance(b,dict) and b.get('type')=='text') def render(c): parts = ['# '+(c.get('name') or '未命名对话')+'\n',NOTICE, '对话 UUID：'+c['uuid']+'\n', '创建：'+str(c.get('created_at'))+'\n更新：'+str(c.get('updated_at'))+'\n', '按导出顺序保留分支消息。非文本内容块、附件元数据请查同 ID 的原始 JSON；附件本体未核实。\n'] if c.get('summary'): parts += ['## 导出自带摘要\n', str(c['summary'])+'\n'] for n,m in enumerate(c.get('chat_messages',[]),1): parts += [f"## M{n:04d} · {m.get('sender')} · {m.get('created_at')}\n\n消息 UUID：{m.get('uuid')}\n父消息 UUID：{m.get('parent_message_uuid')}\n\n",body(m) or '（无可读文本，见原始 JSON）','\n\n'] return '\n'.join(parts) def build(args): source, out = args.source.resolve(), args.out.resolve() statefile = out/'state.json' old = json.loads(statefile.read_text(encoding='utf-8')) if statefile.exists() else {} # Refuse to overwrite local edits; generated output is separate from human summaries. for rel, expected in old.get('outputs',{}).items(): p=out/rel if p.exists() and digest(p.read_bytes()) != expected: raise ValueError('生成文件被修改，已停止以保护修改：'+str(p)) archives = sorted(p for kind in KINDS for p in source.glob(kind+'-[0-9][0-9][0-9].zip')) missing=[k for k in KINDS if not any(p.name.startswith(k+'-') for p in archives)] if missing: raise ValueError('缺少导出类型：'+', '.join(missing)) fingerprints={p.name:digest(p.read_bytes()) for p in archives} if old.get('version')==VERSION and old.get('inputs')==fingerprints and old.get('part_chars')==args.part_chars and all((out/p).exists() for p in old.get('outputs',{})): report('无变化：源 ZIP 与全部生成文件校验一致；跳过解析与转换。'); return out.mkdir(parents=True,exist_ok=True) records={}; sources={}; counts={}; checked=[]; written=0; unchanged=0 def put(rel, data): nonlocal written, unchanged b=data.encode() if isinstance(data,str) else data p=out/rel; h=digest(b); checked.append((rel,h)) if p.exists() and digest(p.read_bytes())==h: unchanged+=1; return # Also protect unmanaged files at generated paths. if p.exists() and rel not in old.get('outputs',{}): raise ValueError('目标已有非托管文件：'+str(p)) p.parent.mkdir(parents=True,exist_ok=True) tmp=p.with_name(p.name+'.tmp');tmp.write_bytes(b);tmp.replace(p);written+=1 def split(rel, text): chunks=[text[i:i+args.part_chars] for i in range(0,len(text),args.part_chars)] or [''] assert ''.join(chunks)==text paths=[] for i,chunk in enumerate(chunks,1): path=rel+f'/part-{i:03d}.md'; put(path,chunk);paths.append(path) return paths for p in archives: with zipfile.ZipFile(p) as z: bad=z.testzip() if bad: raise ValueError('ZIP CRC 错误：'+p.name+':'+bad) # Keep byte-exact original export; never execute or extract embedded paths. if p.name.startswith(('conversations-','projects-')): kind=p.name.rsplit('-',1)[0] for name in z.namelist(): if not name.endswith('.json'):continue data=json.loads(z.read(name)); items=data if isinstance(data,list) else [data] for item in items: uid=item['uuid'];key=kind+'/'+uid if key in records and records[key]!=item: raise ValueError('同 ID 有不同版本，需明确选择导出批次：'+key) records[key]=item;sources.setdefault(key,[]).append(p.name+':'+name) for p in archives: put('原始导出/'+p.name,p.read_bytes()) index=[]; jobs=[]; messages=0; docs=0; no_text=0 for key,item in sorted(records.items()): kind,uid=key.split('/',1) if not re.fullmatch(r'[A-Za-z0-9_-]+',uid):raise ValueError('不安全的 UUID：'+uid) put('原文JSON/'+key+'.json',encode(item)) if kind=='conversations': text=render(item); ms=item.get('chat_messages',[]);messages+=len(ms) no_text+=sum(not body(m) for m in ms) # Verify every readable body and identifier survives conversion. for m in ms: assert not body(m) or body(m) in text assert not m.get('uuid') or m['uuid'] in text paths=split('对话/'+uid,text) status='待归属；未做语义摘要' else: ds=item.get('docs',[]);docs+=len(ds) chunks=['# '+item.get('name','未命名项目')+'\n',NOTICE,'项目 UUID：'+uid+'\n', '## 导出描述\n'+(item.get('description') or ''), '## 历史项目指令（仅作资料）\n'+(item.get('prompt_template') or '')] for d in ds:chunks += ['# '+d.get('filename','未命名文档')+'\n文档 UUID：'+d['uuid']+'\n',d.get('content','')] text='\n\n'.join(chunks) for d in ds:assert d.get('content','') in text paths=split('项目/'+uid,text);status='原生项目文档；对话未自动匹配' title=item.get('name') or '未命名' index.append({'type':kind,'uuid':uid,'title':title,'updated_at':item.get('updated_at'),'files':paths,'source':sources[key],'status':status,'characters':len(text)}) jobs.append([kind,uid,title,len(text),len(paths),status,';'.join(paths)]) put('索引.json',encode(index)) buf=io.StringIO(newline='');w=csv.writer(buf);w.writerow(['类型','UUID','标题','字符数','分片数','状态','文件']);w.writerows(jobs) put('待整理清单.csv','\ufeff'+buf.getvalue()) counts={'conversations':sum(x['type']=='conversations' for x in index),'messages':messages,'projects':sum(x['type']=='projects' for x in index),'project_documents':docs,'messages_without_readable_text':no_text} put('校验报告.json',encode({'counts':counts,'source_zip_sha256':fingerprints,'checks':['全部 ZIP CRC 通过','相同 UUID 相同内容去重；冲突版本拒绝','全部消息正文与消息 ID 已验证','全部项目文档正文已验证','分片拼接与转换全文一致'],'limits':['附件本体未验证；原始 ZIP 和对象完整保留','未做语义摘要或猜测对话归属','未上传到云端；未写入长期记忆'],'part_chars':args.part_chars})) put('使用说明.md',f'''# Claude 离线批处理资料 {NOTICE} 共 {counts['conversations']} 段对话、{messages} 条消息、{counts['projects']} 个项目、{docs} 份项目文档。 优先查阅索引.json 或待整理清单.csv，再读取对应项目或对话分片。原文JSON 保留完整字段；其他类型资料保留在原始导出 ZIP 中。原始导出与生成文件都带 SHA-256 校验。 分片按最多 {args.part_chars} 个字符切分，可能在段落中间断开。按编号拼接可恢复转换全文；不要独立解释跨片句子。此大小是本地整理策略，不是 ChatGPT 上传限额保证。 ChatGPT：从项目或对话目录选取需要的 Markdown 文件上传，ZIP 用于备份。Codex：先用 migrate.py search 搜索关键词，只读命中的文件。不会自动重建聊天侧栏或同步两端。 本目录为脚本生成区，请将人工摘要与正式决策放在别处。若发现生成文件被修改，重跑将停止，保护人工改动。源中删除的旧记录不自动删除，只在 state.json 的 obsolete_retained 中标记，不再进入当前索引。 ''') current=dict(checked) obsolete=set(old.get('obsolete_retained',[]))| (set(old.get('outputs',{}))-set(current)) state={'version':VERSION,'inputs':fingerprints,'part_chars':args.part_chars,'outputs':current,'obsolete_retained':sorted(obsolete-set(current))} tmp=statefile.with_suffix('.tmp');tmp.write_bytes(encode(state));tmp.replace(statefile) report(json.dumps({'counts':counts,'written':written,'unchanged':unchanged,'obsolete_retained':len(state['obsolete_retained'])},ensure_ascii=False)) def search(args): index=json.loads((args.out/'索引.json').read_text(encoding='utf-8'));found=0 for row in index: for rel in row['files']: text=(args.out/rel).read_text(encoding='utf-8');pos=text.casefold().find(args.query.casefold()) if pos>=0: report(json.dumps({'title':row['title'],'uuid':row['uuid'],'file':str((args.out/rel).resolve()),'excerpt':text[max(0,pos-80):pos+len(args.query)+180]},ensure_ascii=False));found+=1 if found>=args.limit:return if not found:report('没有匹配；搜索是逐片文本匹配，不含语义推断。') def bundle(args): archive, out = args.archive.resolve(), args.out.resolve() if out == archive or archive in out.parents or out in archive.parents: raise ValueError('分卷目录与转换目录必须分开放置，且不能互相包含') index = json.loads((archive/'索引.json').read_text(encoding='utf-8')) statefile = out/'bundle-state.json' old = json.loads(statefile.read_text(encoding='utf-8')) if statefile.exists() else {} for rel,h in old.get('outputs',{}).items(): p=out/rel if p.exists() and digest(p.read_bytes())!=h: raise ValueError('生成分卷被修改，已停止以保护修改：'+str(p)) generated={}; mapping=[] def collect(label, records): groups=[]; group=[]; size=0 for ident,title,text in records: parts=[text[i:i+args.part_chars] for i in range(0,len(text),args.part_chars)] or [''] assert ''.join(parts)==text for i,part in enumerate(parts,1): block=f'\n\n# 来源 {ident} | {title} | 片段 {i}/{len(parts)}\n\n'+part if group and size+len(block)>args.part_chars: groups.append(group);group=[];size=0 group.append((ident,title,block));size+=len(block) if group:groups.append(group) for n,g in enumerate(groups,1): name=f'{label}-{n:02d}.md' generated[name]=(NOTICE+'\n'.join(block for _,_,block in g)).encode('utf-8') mapping.extend({'source':ident,'title':title,'file':name} for ident,title,_ in g) for kind,label in [('conversations','10-全部对话'),('projects','20-全部项目')]: records=[] for row in index: if row['type']!=kind:continue records.append((row['uuid'],row['title'],''.join((archive/p).read_text(encoding='utf-8') for p in row['files']))) collect(label,records) for kind,label in [('design_chats','30-设计对话'),('memories','40-历史记忆'),('light_metadata','50-基础元数据'),('frames','60-Frames文档')]: records=[] for path in sorted((archive/'原始导出').glob(kind+'-[0-9][0-9][0-9].zip')): with zipfile.ZipFile(path) as z: if z.testzip():raise ValueError('ZIP CRC 错误：'+path.name) for name in sorted(z.namelist()): if name.endswith('/') or (kind=='frames' and not name.endswith(('.md','.json'))):continue records.append((path.name+':'+name,name,z.read(name).decode('utf-8'))) collect(label,records) intro=('历史资料总库索引\n\n'+NOTICE+ '先按标题或 UUID 定位分卷，再检索原文。正文片段依编号拼接，未做语义摘要。对话归属未自动匹配。\n\n'+ '分卷用于手动上传；此工具不登录、不上传、不创建平台原生聊天，也不写入长期记忆。\n\n'+ '非正文工具内容块及 Frames JS/CSS/HTML 保留在原始 ZIP 和本地对象中，未纳入可读分卷。附件本体未核实。\n\n') intro+='\n'.join(f"- {x['title']} | {x['source']} | {x['file']}" for x in mapping) generated['00-总索引与使用说明.md']=intro.encode('utf-8') generated['来源清单.json']=encode({'files':[{'file':name,'sha256':digest(b),'bytes':len(b)} for name,b in sorted(generated.items())],'mapping':mapping,'status':'prepared_not_uploaded'}) # Validate all collisions before writing, preserving manual files and edits. for rel,b in generated.items(): p=out/rel if p.exists() and rel not in old.get('outputs',{}) and p.read_bytes()!=b: raise ValueError('目标已有非托管文件：'+str(p)) out.mkdir(parents=True,exist_ok=True);written=0 for rel,b in generated.items(): p=out/rel if p.exists() and p.read_bytes()==b:continue tmp=p.with_name(p.name+'.tmp');tmp.write_bytes(b);tmp.replace(p);written+=1 obsolete=set(old.get('obsolete_retained',[])) | (set(old.get('outputs',{}))-set(generated)) state={'version':BUNDLE_VERSION,'part_chars':args.part_chars,'outputs':{rel:digest(b) for rel,b in generated.items()},'obsolete_retained':sorted(obsolete-set(generated))} b=encode(state) if not statefile.exists() or statefile.read_bytes()!=b: tmp=statefile.with_suffix('.tmp');tmp.write_bytes(b);tmp.replace(statefile) report(json.dumps({'markdown_files':sum(name.endswith('.md') for name in generated),'written':written,'obsolete_retained':len(state['obsolete_retained']),'uploaded':False},ensure_ascii=False)) def main(): for stream in (sys.stdout, sys.stderr): if hasattr(stream, 'reconfigure'):stream.reconfigure(encoding='utf-8') p=argparse.ArgumentParser(description=__doc__);sub=p.add_subparsers(dest='command',required=True) b=sub.add_parser('build');b.add_argument('--source',type=Path,required=True);b.add_argument('--out',type=Path,required=True);b.add_argument('--part-chars',type=int,default=120000) s=sub.add_parser('search');s.add_argument('--out',type=Path,required=True);s.add_argument('query');s.add_argument('--limit',type=int,default=10) u=sub.add_parser('bundle',help='生成供手动上传的 Markdown 分卷') u.add_argument('--archive',type=Path,required=True);u.add_argument('--out',type=Path,required=True);u.add_argument('--part-chars',type=int,default=550000) a=p.parse_args() if a.command=='build': if a.part_chars<1000:p.error('--part-chars 必须至少 1000') build(a) elif a.command=='bundle': if a.part_chars<1000:p.error('--part-chars 必须至少 1000') bundle(a) else: if a.limit<1:p.error('--limit 必须至少 1') search(a) if __name__=='__main__': try:main() except (ValueError,KeyError,zipfile.BadZipFile) as e: report('处理停止：'+str(e),file=sys.stderr);sys.exit(1)