333 lines
16 KiB
Python
333 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""Code4History two-way workbench. All operations are local, stdlib-only.
|
|
|
|
Use --root to target a test or separate project. No daemon or background process.
|
|
This tool safely archives upload bytes and moves completed human/agent tasks;
|
|
semantic historical interpretation remains the responsibility of the research skill.
|
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
|
from datetime import datetime, timezone
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SRC_KINDS = {'primary': ('SRC-P', 'primary'), 'secondary': ('SRC-S', 'secondary')}
|
|
WORK_FILES = {'readme.md', '.gitkeep'}
|
|
SIDE_SUFFIX = '.source.md'
|
|
TEMP_SUFFIXES = ('.part', '.tmp', '.crdownload', '.download', '.swp')
|
|
|
|
|
|
def stamp():
|
|
return datetime.now(timezone.utc).isoformat(timespec='seconds')
|
|
|
|
|
|
def sha256(path):
|
|
d = hashlib.sha256()
|
|
with path.open('rb') as f:
|
|
for part in iter(lambda: f.read(1024*1024), b''):
|
|
d.update(part)
|
|
return d.hexdigest()
|
|
|
|
|
|
def safe_file(root: Path, relative: str, *, required=True) -> Path:
|
|
# Never accept absolute, dotted segments, traversal, directory / symlink.
|
|
if not relative or '\\' in relative or '/' in relative or relative in {'.', '..'} or Path(relative).is_absolute():
|
|
raise ValueError('Only a direct filename inside the target inbox is allowed.')
|
|
p = root / relative
|
|
if required and (not p.is_file() or p.is_symlink()):
|
|
raise ValueError(f'Not a regular file: {p}')
|
|
if p.exists() and p.is_symlink():
|
|
raise ValueError(f'Unsafe symbolic link: {p}')
|
|
return p
|
|
|
|
|
|
def atomic_bytes(dest: Path, content: bytes):
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
if dest.exists():
|
|
raise FileExistsError(dest)
|
|
with tempfile.NamedTemporaryFile(dir=dest.parent, prefix='.c4h-incoming-', delete=False) as f:
|
|
temp = Path(f.name)
|
|
f.write(content)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
try:
|
|
os.replace(temp, dest)
|
|
finally:
|
|
temp.unlink(missing_ok=True)
|
|
|
|
|
|
def atomic_replace(dest: Path, content: bytes):
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile(dir=dest.parent, prefix='.c4h-write-', delete=False) as f:
|
|
temp=Path(f.name); f.write(content); f.flush(); os.fsync(f.fileno())
|
|
try:
|
|
os.replace(temp, dest)
|
|
finally:
|
|
temp.unlink(missing_ok=True)
|
|
|
|
|
|
def safe_copy(source: Path, destination: Path, digest: str):
|
|
destination.parent.mkdir(parents=True,exist_ok=True)
|
|
if destination.exists():
|
|
if sha256(destination) == digest:
|
|
return False
|
|
raise FileExistsError(f'Destination exists with different bytes: {destination}')
|
|
with tempfile.NamedTemporaryFile(dir=destination.parent,prefix='.c4h-upload-',delete=False) as f:
|
|
temp=Path(f.name)
|
|
try:
|
|
shutil.copyfile(source,temp)
|
|
if sha256(temp) != digest:
|
|
raise IOError(f'Hash mismatch while staging: {source}')
|
|
os.replace(temp,destination)
|
|
finally:
|
|
temp.unlink(missing_ok=True)
|
|
if sha256(destination) != digest:
|
|
raise IOError(f'Hash mismatch after archive write: {destination}')
|
|
return True
|
|
|
|
|
|
def user_answer(doc: str) -> str:
|
|
if '## 用户回复' not in doc:
|
|
return ''
|
|
reply = doc.split('## 用户回复',1)[1].split('## Agent 处理记录',1)[0].strip()
|
|
if not reply or reply.startswith('（请在此填写'):
|
|
return ''
|
|
return reply
|
|
|
|
|
|
def status(root: Path) -> dict:
|
|
upl=root/'upload'
|
|
files=[p for p in sorted(upl.iterdir()) if p.name not in WORK_FILES] if upl.exists() else []
|
|
attachments=[]; source_notes=[]; blocked=[]
|
|
for p in files:
|
|
if p.is_symlink() or not p.is_file():
|
|
blocked.append(p.name);continue
|
|
if p.name.startswith('.') or p.name.lower().endswith(TEMP_SUFFIXES):
|
|
blocked.append(p.name);continue
|
|
if p.name.endswith(SIDE_SUFFIX):
|
|
source_notes.append(p.name);continue
|
|
companion=p.name+SIDE_SUFFIX
|
|
attachments.append({'file':p.name,'source_note':companion if (upl/companion).is_file() else None,'bytes':p.stat().st_size})
|
|
paired={x['source_note'] for x in attachments if x['source_note']}
|
|
todo=[]
|
|
for p in sorted((root/'todo').glob('TODO-*.md')):
|
|
if p.is_symlink(): blocked.append('todo/'+p.name);continue
|
|
body=p.read_text(encoding='utf-8')
|
|
m=re.search(r'^status:\s*([a-z_-]+)\s*$',body,re.M)
|
|
state=m.group(1) if m else 'invalid'
|
|
if state=='open' and user_answer(body):state='answered'
|
|
todo.append({'file':p.name,'status':state})
|
|
req=[]
|
|
for p in sorted((root/'requests').iterdir()) if (root/'requests').exists() else []:
|
|
if p.name in WORK_FILES or p.name=='archive':continue
|
|
if p.is_file() and not p.is_symlink():req.append(p.name)
|
|
else:blocked.append('requests/'+p.name)
|
|
return {
|
|
'upload':attachments,
|
|
'orphan_source_notes':sorted(set(source_notes)-paired),
|
|
'blocked':blocked,
|
|
'todo':todo,
|
|
'requests':req,
|
|
'summary':{
|
|
'upload_pending':len(attachments)+len(set(source_notes)-paired)+len([x for x in blocked if '/' not in x]),
|
|
'todo_open':sum(x['status']=='open' for x in todo),
|
|
'todo_answered':sum(x['status']=='answered' for x in todo),
|
|
'requests_pending':len(req)
|
|
}
|
|
}
|
|
|
|
|
|
def next_number(folder: Path, prefix: str) -> int:
|
|
seq=[]
|
|
for p in folder.glob(f'{prefix}-*.md'):
|
|
m=re.match(re.escape(prefix)+r'-(\d{4})(?:-|\.md)',p.name)
|
|
if m:seq.append(int(m.group(1)))
|
|
return max(seq,default=0)+1
|
|
|
|
|
|
def next_source_id(root: Path,kind: str):
|
|
prefix, sub=SRC_KINDS[kind]
|
|
folder=root/'01_sources/records'/sub
|
|
return f'{prefix}-{next_number(folder,prefix):04d}'
|
|
|
|
|
|
def yaml_set(doc: str,key: str,val: str) -> str:
|
|
value=json.dumps(val,ensure_ascii=False)
|
|
pattern=re.compile(r'^'+re.escape(key)+r':.*$',re.M)
|
|
if pattern.search(doc.split('\n---\n',1)[0]):
|
|
return pattern.sub(lambda _:f'{key}: {value}',doc,count=1)
|
|
return doc.replace('\n---\n',f'\n{key}: {value}\n---\n',1)
|
|
|
|
|
|
def intake(root:Path,filename:str,kind:str,slug:str,dry_run=False):
|
|
if not re.fullmatch(r'[a-z0-9]+(?:-[a-z0-9]+)*',slug):
|
|
raise ValueError('slug must be lowercase ASCII kebab-case')
|
|
if filename in WORK_FILES or filename.endswith(SIDE_SUFFIX) or filename.startswith('.') or filename.lower().endswith(TEMP_SUFFIXES):
|
|
raise ValueError('Not an upload attachment filename')
|
|
inbox=root/'upload'
|
|
source=safe_file(inbox,filename)
|
|
if source.stat().st_size==0:
|
|
raise ValueError('Zero-byte upload must be checked, not archived')
|
|
metadata=safe_file(inbox,filename+SIDE_SUFFIX,required=False)
|
|
has_meta=metadata.is_file()
|
|
digest=sha256(source); note_hash=sha256(metadata) if has_meta else None
|
|
# A previous round may have been interrupted after successful archival but
|
|
# before upload cleanup, or the user may have uploaded identical bytes.
|
|
# Reuse an existing verified accession; NEVER discard a changed sidecar.
|
|
for prior in sorted((root/'01_sources/provenance').glob('SRC-*-intake.json')):
|
|
try:
|
|
receipt_old=json.loads(prior.read_text(encoding='utf-8'))
|
|
archived=root/receipt_old['archive_path']
|
|
if (receipt_old.get('sha256')!=digest or not archived.is_file()
|
|
or sha256(archived)!=digest):
|
|
continue
|
|
if receipt_old.get('source_note_sha256')!=note_hash:
|
|
raise ValueError('已归档相同原件，但此次来源说明不同；保留上传件供人工合并新的来源信息。')
|
|
record_id=receipt_old['source_id']
|
|
record_matches=list((root/'01_sources/records'/SRC_KINDS[kind][1]).glob(record_id+'-*.md'))
|
|
if not record_id.startswith(SRC_KINDS[kind][0]+'-') or not record_matches:
|
|
continue
|
|
if dry_run:return {'id':record_id,'reused':True,'sha256':digest}
|
|
if sha256(source)!=digest or (has_meta and sha256(metadata)!=note_hash):
|
|
raise IOError('Upload changed while checking duplicate')
|
|
source.unlink()
|
|
if has_meta:metadata.unlink()
|
|
return {'id':record_id,'reused':True,'sha256':digest}
|
|
except (KeyError, json.JSONDecodeError):
|
|
continue
|
|
ident=next_source_id(root,kind)
|
|
suffix=Path(filename).suffix.lower()
|
|
original=root/'01_sources/originals'/SRC_KINDS[kind][1]/f'{ident}--{filename}'
|
|
meta_dest=root/'01_sources/provenance'/f'{ident}-upload-source.md'
|
|
receipt=root/'01_sources/provenance'/f'{ident}-intake.json'
|
|
record=root/'01_sources/records'/SRC_KINDS[kind][1]/f'{ident}-{slug}.md'
|
|
if any(p.exists() for p in (original,receipt,record,meta_dest)):
|
|
raise FileExistsError('Generated destination is occupied; inspect before retrying')
|
|
plan={'id':ident,'original_path':original.relative_to(root).as_posix(),'record_path':record.relative_to(root).as_posix(),'metadata_path':meta_dest.relative_to(root).as_posix() if has_meta else None,'sha256':digest,'metadata_sha256':note_hash,'metadata_missing':not has_meta}
|
|
if dry_run:return plan
|
|
# Validate all inputs, prepare copies and documentation before removing originals.
|
|
template=root/'10_templates'/('primary.md' if kind=='primary' else 'secondary.md')
|
|
card=template.read_text(encoding='utf-8').replace('__ID__',ident)
|
|
for field,value in [('original_path',plan['original_path']),('sha256',digest),('original_filename',filename),('ingested_utc',stamp()),('upload_source_note',plan['metadata_path'] or ''),('source_metadata_status','provided_unverified' if has_meta else 'missing')]:
|
|
card=yaml_set(card,field,value)
|
|
# Staging and creating records may fail. Until receipt + record + archive checks pass,
|
|
# never remove the user's source. Partial copies are left for recovery.
|
|
safe_copy(source,original,digest)
|
|
if has_meta:safe_copy(metadata,meta_dest,note_hash)
|
|
atomic_bytes(record,card.encode('utf-8'))
|
|
receipt_data={'source_id':ident,'timestamp_utc':stamp(),'upload_name':filename,'archive_path':plan['original_path'],'sha256':digest,'source_note_path':plan['metadata_path'],'source_note_sha256':note_hash,'ingestion':'archived_not_necessarily_read'}
|
|
atomic_bytes(receipt,(json.dumps(receipt_data,ensure_ascii=False,indent=2)+'\n').encode('utf-8'))
|
|
if sha256(original)!=digest or not record.is_file() or not receipt.is_file() or (has_meta and sha256(meta_dest)!=note_hash):
|
|
raise IOError('Archive integrity check failed; upload retained')
|
|
# Raw catalog updates can be retried by Agent if interrupted; receipt is durable.
|
|
catalog=root/'01_sources/catalog.md'
|
|
current=catalog.read_text(encoding='utf-8') if catalog.exists() else '# Source catalog\n'
|
|
line=f'\n- `{ident}` · `{plan["original_path"]}` · sha256 `{digest}` · [SRC card]({record.relative_to(root).as_posix()}) · metadata: {"provided" if has_meta else "pending"}\n'
|
|
atomic_replace(catalog,(current+line).encode('utf-8'))
|
|
# Recheck bytes before removing inbox: if researcher modified during intake, retain.
|
|
if sha256(source)!=digest or (has_meta and sha256(metadata)!=note_hash):
|
|
raise IOError('Upload changed during intake; retaining input to avoid data loss')
|
|
# At this stage all copies & receipt are verified; unlink originals only now.
|
|
source.unlink()
|
|
if has_meta:metadata.unlink()
|
|
return plan
|
|
|
|
|
|
def todo_new(root:Path,title:str,kind:str,body:str,ref:str,dry_run=False):
|
|
if not title.strip() or not body.strip():raise ValueError('TODO title and body required')
|
|
n=next_number(root/'todo','TODO');ident=f'TODO-{n:04d}'
|
|
doc=f'''---
|
|
id: {ident}
|
|
status: open
|
|
kind: {kind}
|
|
created: "{stamp()}"
|
|
reference: {json.dumps(ref,ensure_ascii=False)}
|
|
---
|
|
|
|
# {title}
|
|
|
|
## 为什么需要用户
|
|
{body}
|
|
|
|
## 用户回复
|
|
（请在此填写回复；扫描器会自动识别）
|
|
|
|
## Agent 处理记录
|
|
（处理后说明关联的正式研究记录）
|
|
'''
|
|
dest=root/'todo'/f'{ident}-request.md'
|
|
if not dry_run:atomic_bytes(dest,doc.encode('utf-8'))
|
|
return str(dest.relative_to(root))
|
|
|
|
|
|
def archive_task(root:Path,folder:str,name:str,record:str,require_answer=False,dry_run=False):
|
|
source=safe_file(root/folder,name)
|
|
if source.name=='readme.md' or not source.name.endswith('.md'):
|
|
raise ValueError('Expected a Markdown task file')
|
|
evidence=(root/record).resolve()
|
|
if not evidence.is_file() or (root.resolve()!=evidence and root.resolve() not in evidence.parents):
|
|
raise ValueError('The --record path must be an existing file in this research repository')
|
|
if folder=='todo':
|
|
if not re.fullmatch(r'TODO-\d{4}-[a-zA-Z0-9_-]+\.md',name):
|
|
raise ValueError('Invalid TODO task file')
|
|
doc=source.read_text('utf-8')
|
|
status_line=re.search(r'^status:\s*([a-z_-]+)\s*$',doc,re.M)
|
|
if not status_line or status_line.group(1) not in ('open','answered') or not user_answer(doc):
|
|
raise ValueError('TODO needs an actual user answer in ## 用户回复')
|
|
dest=root/folder/'archive'/name
|
|
if dest.exists():raise FileExistsError(f'Already archived: {dest}')
|
|
if dry_run:return str(dest.relative_to(root))
|
|
doc=source.read_text('utf-8')
|
|
if folder=='todo':
|
|
doc=re.sub(r'^status:\s*open\s*$', 'status: answered', doc, count=1, flags=re.M)
|
|
applied=f'\n\n## 归档说明\n- 时间：{stamp()}\n- Agent 已处理并记录于：`{record}`\n'
|
|
# Preserve the original reply while adding an archival statement.
|
|
atomic_bytes(dest,(doc+applied).encode('utf-8'))
|
|
if not dest.read_bytes().startswith(doc.encode('utf-8')):
|
|
raise IOError('Archive content mismatch')
|
|
source.unlink()
|
|
return str(dest.relative_to(root))
|
|
|
|
|
|
def finish(root:Path)->int:
|
|
s=status(root)
|
|
print(json.dumps(s['summary'],ensure_ascii=False,indent=2))
|
|
if s['summary']['upload_pending'] or s['summary']['todo_answered'] or s['summary']['requests_pending']:
|
|
print('有未完成的交接事项：请更新 handoff/latest.md；未成功归档的上传原件必须保留。')
|
|
return 1
|
|
print('已无待摄入 upload 文件、未处理用户答复或 requests；开放 TODO 可保留等待用户。')
|
|
return 0
|
|
|
|
|
|
def main(argv=None):
|
|
p=argparse.ArgumentParser(description=__doc__)
|
|
p.add_argument('--root',type=Path,default=ROOT)
|
|
sub=p.add_subparsers(dest='command',required=True)
|
|
sub.add_parser('scan')
|
|
intake_p=sub.add_parser('intake');intake_p.add_argument('--file',required=True);intake_p.add_argument('--kind',choices=tuple(SRC_KINDS),required=True);intake_p.add_argument('--slug',required=True);intake_p.add_argument('--dry-run',action='store_true')
|
|
todo_p=sub.add_parser('todo-new');todo_p.add_argument('--title',required=True);todo_p.add_argument('--kind',choices=('question','verification','decision','authorization','assistance'),default='question');todo_p.add_argument('--body',required=True);todo_p.add_argument('--ref',default='');todo_p.add_argument('--dry-run',action='store_true')
|
|
close_p=sub.add_parser('todo-close');close_p.add_argument('--id',required=True);close_p.add_argument('--record',required=True);close_p.add_argument('--dry-run',action='store_true')
|
|
req_p=sub.add_parser('request-done');req_p.add_argument('--file',required=True);req_p.add_argument('--record',required=True);req_p.add_argument('--dry-run',action='store_true')
|
|
sub.add_parser('finish')
|
|
args=p.parse_args(argv);root=args.root.resolve()
|
|
try:
|
|
if args.command=='scan':print(json.dumps(status(root),ensure_ascii=False,indent=2))
|
|
elif args.command=='intake':print(json.dumps(intake(root,args.file,args.kind,args.slug,args.dry_run),ensure_ascii=False,indent=2))
|
|
elif args.command=='todo-new':print(todo_new(root,args.title,args.kind,args.body,args.ref,args.dry_run))
|
|
elif args.command=='todo-close':print(archive_task(root,'todo',safe_file(root/'todo',next((x.name for x in (root/'todo').glob(args.id+'-*.md')),args.id)).name,args.record,True,args.dry_run))
|
|
elif args.command=='request-done':print(archive_task(root,'requests',args.file,args.record,False,args.dry_run))
|
|
elif args.command=='finish':return finish(root)
|
|
except (ValueError,OSError,RuntimeError,StopIteration) as ex:
|
|
print(f'交接失败：{ex}',file=sys.stderr)
|
|
return 2
|
|
return 0
|
|
|
|
if __name__=='__main__':sys.exit(main())
|