408 lines
18 KiB
Python
408 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""Code4History cross-version research-data migration (stdlib only).
|
||
|
||
Usage:
|
||
python migrate.py export --source OLD --out research.c4h.zip
|
||
python migrate.py restore --bundle research.c4h.zip --target NEW
|
||
python migrate.py upgrade --source OLD --template code4history-v0.8.zip --target NEW
|
||
|
||
This tool never deletes the old project, never overwrites existing user-edited
|
||
content in the target, and never installs captured framework customizations
|
||
as executable instructions. Review the migration report after every restore.
|
||
"""
|
||
from __future__ import annotations
|
||
import argparse
|
||
import hashlib
|
||
import json
|
||
import os
|
||
from pathlib import Path, PurePosixPath
|
||
import shutil
|
||
import stat
|
||
import sys
|
||
import tempfile
|
||
from datetime import datetime, timezone
|
||
from zipfile import ZIP_DEFLATED, ZipFile, ZipInfo
|
||
|
||
BUNDLE_TYPE = 'code4history-research-export'
|
||
FORMAT_VERSION = 1
|
||
BASELINE_FILE = 'migration_baselines.json'
|
||
MUTABLE_ROOT = {'topic.md', 'USERINFO.md', 'STATE.md'}
|
||
MUTABLE_PATHS = {
|
||
'handoff/latest.md', '00_governance/bibliography.bib', '00_governance/checkpoints.md',
|
||
'00_governance/decisions.md', '00_governance/research-log.md',
|
||
'01_sources/catalog.md', '04_context/chronology.md',
|
||
'05_historiography/debates.md', '07_manuscript/outline.md',
|
||
'07_manuscript/sections/order.txt',
|
||
'08_review/citation-audit.md', '08_review/contradictions.md',
|
||
'08_review/counterevidence.md', '08_review/submission-checklist.md',
|
||
}
|
||
FRAMEWORK_ROOTS = {
|
||
'AGENTS.md', 'agent.md', 'AGENT.md', 'code4history.md', 'README.md',
|
||
'INTERPRETATION.md', 'ARGUMENT_DENSITY.md', 'SYNTHESIS_AND_PROSE.md',
|
||
'DISCUSSION_TO_TOPIC.md', 'QUICK_FIX.md', 'PUBLISHING.md', 'MIGRATION.md',
|
||
'CHANGELOG.md', '.gitignore', 'UPGRADE_v0.4.md', 'UPGRADE_v0.5.md',
|
||
'UPGRADE_v0.6.md', 'UPGRADE_v0.7.md', 'UPGRADE_v0.8.md', 'INTERACTION.md',
|
||
}
|
||
SKIP_DIRS = {
|
||
'.git', '.venv', 'venv', 'node_modules', '__pycache__', '.pytest_cache',
|
||
'.mypy_cache', '.ruff_cache', '.idea', '.vscode', '.code4history-migration',
|
||
}
|
||
SKIP_NAMES = {'.DS_Store', 'Thumbs.db', '.gitkeep'}
|
||
SELF_GENERATED_PATHS = {'tools/migration_baselines.json'}
|
||
BLOCKED_SUFFIXES = {'.pem', '.key', '.p12', '.pfx', '.credentials', '.pyc'}
|
||
|
||
|
||
def timestamp() -> str:
|
||
return datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
|
||
|
||
|
||
def digest_bytes(data: bytes) -> str:
|
||
return hashlib.sha256(data).hexdigest()
|
||
|
||
|
||
def digest_file(path: Path) -> str:
|
||
sha = hashlib.sha256()
|
||
with path.open('rb') as f:
|
||
for chunk in iter(lambda: f.read(1024 * 1024), b''):
|
||
sha.update(chunk)
|
||
return sha.hexdigest()
|
||
|
||
|
||
def validate_relpath(name: str) -> str:
|
||
# Treat archive paths as POSIX even on Windows; reject Windows drive/UNC.
|
||
if not isinstance(name, str) or not name or '\\' in name or ':' in name:
|
||
raise ValueError(f'非法相对路径：{name!r}')
|
||
p = PurePosixPath(name)
|
||
if p.is_absolute() or name.startswith('/') or any(s in ('', '.', '..') for s in name.split('/')):
|
||
raise ValueError(f'路径穿越或绝对路径：{name!r}')
|
||
return p.as_posix()
|
||
|
||
|
||
def is_sensitive(rel: str) -> bool:
|
||
p = PurePosixPath(rel)
|
||
if any(part in SKIP_DIRS for part in p.parts[:-1]):
|
||
return True
|
||
n = p.name
|
||
return n in SKIP_NAMES or n == '.env' or n.startswith('.env.') or p.suffix.lower() in BLOCKED_SUFFIXES
|
||
|
||
|
||
def baseline_data() -> dict:
|
||
source = Path(__file__).resolve().with_name(BASELINE_FILE)
|
||
if not source.is_file():
|
||
raise RuntimeError(f'缺少官方框架基线 {source}；请将 migrate.py 与 migration_baselines.json 放在同一目录。')
|
||
doc = json.loads(source.read_text(encoding='utf-8'))
|
||
if doc.get('schema') != 1 or not isinstance(doc.get('files'), dict):
|
||
raise ValueError('不支持的框架基线格式')
|
||
return doc['files']
|
||
|
||
|
||
def is_framework_area(rel: str) -> bool:
|
||
p = PurePosixPath(rel)
|
||
if rel in FRAMEWORK_ROOTS or p.name.lower() == 'readme.md':
|
||
return True
|
||
return p.parts[0] in {'.agents', '10_templates', 'tools', 'tests'}
|
||
|
||
|
||
def classify(rel: str, sha: str, baselines: dict) -> str:
|
||
known = baselines.get(rel)
|
||
if rel in SELF_GENERATED_PATHS:
|
||
return 'stock'
|
||
if known and sha in known:
|
||
return 'stock' # exact byte match against official releases
|
||
if rel in MUTABLE_ROOT or rel in MUTABLE_PATHS:
|
||
return 'research'
|
||
# A shipped but modified document is a framework customization unless
|
||
# explicitly listed as a mutable research file above.
|
||
if known:
|
||
return 'framework_customization'
|
||
# All changes in framework / skill / tools areas require manual approval.
|
||
# Never silently install executable code or agent instructions.
|
||
if is_framework_area(rel):
|
||
return 'framework_customization'
|
||
return 'research'
|
||
|
||
|
||
def project_files(root: Path):
|
||
if not root.is_dir():
|
||
raise FileNotFoundError(f'项目目录不存在：{root}')
|
||
for base, dirs, files in os.walk(root, followlinks=False):
|
||
base_path = Path(base)
|
||
dirs[:] = [x for x in sorted(dirs) if x not in SKIP_DIRS and not (base_path / x).is_symlink()]
|
||
for name in sorted(files):
|
||
file = base_path / name
|
||
rel = file.relative_to(root).as_posix()
|
||
if is_sensitive(rel):
|
||
continue
|
||
if file.is_symlink() or not file.is_file():
|
||
print(f'警告：跳过软链接/非常规文件 {rel}', file=sys.stderr)
|
||
continue
|
||
validate_relpath(rel)
|
||
yield rel, file
|
||
|
||
|
||
def export(source: Path, output: Path, *, dry_run=False, quiet=False) -> dict:
|
||
source = source.resolve()
|
||
output = output.absolute()
|
||
if source == output or source in output.parents:
|
||
raise ValueError('导出包必须位于旧项目目录之外，避免循环打包或污染研究资料')
|
||
baselines = baseline_data()
|
||
manifest = {
|
||
'format': BUNDLE_TYPE, 'version': FORMAT_VERSION,
|
||
'created_utc': datetime.now(timezone.utc).isoformat(),
|
||
'source_dir_name': source.name, 'entries': [], 'excluded': [],
|
||
}
|
||
for rel, file in project_files(source):
|
||
sha = digest_file(file)
|
||
kind = classify(rel, sha, baselines)
|
||
if kind == 'stock':
|
||
continue
|
||
entry = {'path': rel, 'kind': kind, 'size': file.stat().st_size, 'sha256': sha}
|
||
manifest['entries'].append(entry)
|
||
if not quiet:
|
||
kinds = {k: sum(x['kind'] == k for x in manifest['entries']) for k in ('research', 'research_extension', 'framework_customization')}
|
||
print(f'导出计划：研究文件 {kinds["research"]}，自定义扩展 {kinds["research_extension"]}，需手动审阅的框架改动 {kinds["framework_customization"]}')
|
||
print('注意：.git 历史、.env/密钥、虚拟环境及缓存不在导出包内。原始 PDF 和研究产出会包含在内。')
|
||
if dry_run:
|
||
return manifest
|
||
output.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = None
|
||
try:
|
||
with tempfile.NamedTemporaryFile(prefix='.c4h-export-', suffix='.zip', dir=output.parent, delete=False) as f:
|
||
tmp = Path(f.name)
|
||
with ZipFile(tmp, 'w', compression=ZIP_DEFLATED, allowZip64=True, compresslevel=6) as archive:
|
||
for entry in manifest['entries']:
|
||
rel = entry['path']
|
||
category = 'research' if entry['kind'] != 'framework_customization' else 'framework_customizations'
|
||
archive.write(source.joinpath(*PurePosixPath(rel).parts), arcname=f'{category}/{rel}')
|
||
archive.writestr('manifest.json', json.dumps(manifest, ensure_ascii=False, indent=2) + '\n')
|
||
os.replace(tmp, output)
|
||
finally:
|
||
if tmp is not None and tmp.exists():
|
||
tmp.unlink()
|
||
if not quiet:
|
||
print(f'导出完成：{output}（旧项目未修改）')
|
||
return manifest
|
||
|
||
|
||
def read_bundle(bundle: Path):
|
||
with ZipFile(bundle, 'r') as z:
|
||
members = z.namelist()
|
||
if members.count('manifest.json') != 1 or len(set(members)) != len(members):
|
||
raise ValueError('迁移包中的清单不存在或有重复路径')
|
||
meta = json.loads(z.read('manifest.json'))
|
||
if meta.get('format') != BUNDLE_TYPE or meta.get('version') != FORMAT_VERSION:
|
||
raise ValueError('不是受支持的 Code4History 迁移包')
|
||
entries = meta.get('entries')
|
||
if not isinstance(entries, list):
|
||
raise ValueError('无效的清单内容')
|
||
listed = {'manifest.json'}
|
||
seen = set()
|
||
for item in entries:
|
||
if not isinstance(item, dict):
|
||
raise ValueError('无效的文件条目')
|
||
rel = validate_relpath(item['path'])
|
||
if rel in seen or item['kind'] not in ('research', 'research_extension', 'framework_customization'):
|
||
raise ValueError('重复文件或无效分类')
|
||
seen.add(rel)
|
||
pre = 'framework_customizations' if item['kind'] == 'framework_customization' else 'research'
|
||
archive_name = f'{pre}/{rel}'
|
||
if archive_name not in members:
|
||
raise ValueError(f'迁移包缺少文件 {archive_name}')
|
||
info = z.getinfo(archive_name)
|
||
if stat.S_ISLNK(info.external_attr >> 16):
|
||
raise ValueError(f'拒绝迁移包中的软链接 {rel}')
|
||
if info.file_size != item['size'] or len(item['sha256']) != 64:
|
||
raise ValueError(f'文件大小或校验值无效：{rel}')
|
||
listed.add(archive_name)
|
||
unknown = [x for x in members if x not in listed and not x.endswith('/')]
|
||
if unknown:
|
||
raise ValueError(f'迁移包存在未列入清单的文件：{unknown[:3]}')
|
||
return meta
|
||
|
||
|
||
def secure_target(target: Path, rel: str) -> Path:
|
||
validate_relpath(rel)
|
||
cur = target
|
||
for part in PurePosixPath(rel).parts:
|
||
cur = cur / part
|
||
if cur.is_symlink():
|
||
raise ValueError(f'目标路径含软链接，拒绝写入：{rel}')
|
||
return cur
|
||
|
||
|
||
def write_verified(zipobj: ZipFile, archive_name: str, dest: Path, entry: dict):
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
temp = None
|
||
try:
|
||
with tempfile.NamedTemporaryFile(prefix='.c4h-', dir=dest.parent, delete=False) as output:
|
||
temp = Path(output.name)
|
||
sha = hashlib.sha256()
|
||
length = 0
|
||
with zipobj.open(archive_name, 'r') as incoming:
|
||
for chunk in iter(lambda: incoming.read(1024 * 1024), b''):
|
||
sha.update(chunk)
|
||
length += len(chunk)
|
||
output.write(chunk)
|
||
if sha.hexdigest() != entry['sha256'] or length != entry['size']:
|
||
raise ValueError(f'迁移包完整性校验失败：{entry["path"]}')
|
||
os.replace(temp, dest)
|
||
finally:
|
||
if temp is not None and temp.exists():
|
||
temp.unlink()
|
||
|
||
|
||
def restore(bundle: Path, target: Path, *, dry_run=False, quiet=False) -> dict:
|
||
meta = read_bundle(bundle)
|
||
baselines = baseline_data()
|
||
target = target.resolve()
|
||
if not target.is_dir() or not (target / 'code4history.md').is_file():
|
||
raise ValueError('目标目录必须是一个已解压、可读取 code4history.md 的新版框架')
|
||
# Preverify SHA256 for *all* payloads before writing anything.
|
||
with ZipFile(bundle, 'r') as z:
|
||
for item in meta['entries']:
|
||
rel, kind = item['path'], item['kind']
|
||
archive_name = ('framework_customizations' if kind == 'framework_customization' else 'research') + '/' + rel
|
||
sha = hashlib.sha256(); size = 0
|
||
with z.open(archive_name) as stream:
|
||
for part in iter(lambda: stream.read(1024*1024), b''):
|
||
sha.update(part); size += len(part)
|
||
if size != item['size'] or sha.hexdigest() != item['sha256']:
|
||
raise ValueError(f'迁移包损坏：{rel}，未修改目标项目')
|
||
|
||
stamp = timestamp()
|
||
# Conflict file directory stays outside the tracked research directories.
|
||
review_root = target / '.code4history-migration' / 'conflicts' / stamp
|
||
operations = []
|
||
for item in meta['entries']:
|
||
rel = item['path']
|
||
dest = secure_target(target, rel)
|
||
if item['kind'] == 'framework_customization':
|
||
action = 'review-framework'
|
||
elif dest.exists():
|
||
if not dest.is_file():
|
||
action = 'conflict'
|
||
elif digest_file(dest) == item['sha256']:
|
||
action = 'unchanged'
|
||
elif digest_file(dest) in baselines.get(rel, []) and (rel in MUTABLE_PATHS or rel in MUTABLE_ROOT):
|
||
action = 'fill-stock-placeholder'
|
||
else:
|
||
action = 'conflict'
|
||
else:
|
||
action = 'copy-new'
|
||
operations.append((item, action))
|
||
summary = {k: sum(a == k for _, a in operations) for k in ('copy-new','fill-stock-placeholder','unchanged','conflict','review-framework')}
|
||
if not quiet:
|
||
print('恢复计划：' + '，'.join(f'{k}={v}' for k, v in summary.items()))
|
||
if any(a in ('conflict', 'review-framework') for _, a in operations):
|
||
print('已有内容不会被覆盖：冲突与旧框架修改保存到 .code4history-migration/conflicts/ 中供人工处理。')
|
||
if dry_run:
|
||
return summary
|
||
report = {'source_bundle': str(bundle.resolve()), 'timestamp_utc': stamp, 'actions': []}
|
||
with ZipFile(bundle, 'r') as z:
|
||
for item, action in operations:
|
||
rel, kind = item['path'], item['kind']
|
||
srcname = ('framework_customizations' if kind == 'framework_customization' else 'research') + '/' + rel
|
||
if action == 'unchanged':
|
||
report['actions'].append({'path': rel, 'result': 'unchanged'})
|
||
continue
|
||
if action == 'review-framework':
|
||
dest = secure_target(target, f'.code4history-migration/conflicts/{stamp}/framework_customizations/{rel}')
|
||
elif action == 'conflict':
|
||
dest = secure_target(target, f'.code4history-migration/conflicts/{stamp}/incoming/{rel}')
|
||
else:
|
||
dest = secure_target(target, rel)
|
||
write_verified(z, srcname, dest, item)
|
||
report['actions'].append({'path': rel, 'result': action, 'saved_to': str(dest.relative_to(target))})
|
||
log = target / '.code4history-migration' / 'latest-report.json'
|
||
log.parent.mkdir(parents=True, exist_ok=True)
|
||
log.write_text(json.dumps(report, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
|
||
if not quiet:
|
||
print(f'恢复完成：{target}；报告：{log}')
|
||
return summary
|
||
|
||
|
||
def initialize_template(template: Path, dest: Path):
|
||
"""Only create a new directory; extract safely from a known scaffold layout."""
|
||
if dest.exists():
|
||
raise FileExistsError(f'升级目标已存在：{dest}')
|
||
dest.parent.mkdir(parents=True, exist_ok=True)
|
||
stage = Path(tempfile.mkdtemp(prefix='.c4h-upgrade-', dir=dest.parent))
|
||
try:
|
||
with ZipFile(template, 'r') as z:
|
||
allfiles = [x for x in z.namelist() if not x.endswith('/')]
|
||
names = []
|
||
for name in allfiles:
|
||
# ZIP is expected to have a single code4history/ root.
|
||
if not name.startswith('code4history/'):
|
||
raise ValueError('不支持的模板根目录')
|
||
rel = validate_relpath(name[len('code4history/'):])
|
||
info = z.getinfo(name)
|
||
if stat.S_ISLNK(info.external_attr >> 16):
|
||
raise ValueError('框架模板不得包含软链接')
|
||
names.append((name, rel))
|
||
for name, rel in names:
|
||
output = secure_target(stage, rel)
|
||
output.parent.mkdir(parents=True, exist_ok=True)
|
||
with z.open(name) as incoming, output.open('wb') as f:
|
||
shutil.copyfileobj(incoming, f)
|
||
if not (stage/'code4history.md').is_file():
|
||
raise ValueError('新框架包缺少 code4history.md')
|
||
stage.rename(dest)
|
||
finally:
|
||
if stage.exists():
|
||
shutil.rmtree(stage)
|
||
|
||
|
||
def upgrade(source: Path, template: Path, target: Path, *, dry_run=False):
|
||
source, template, target = source.resolve(), template.resolve(), target.absolute()
|
||
if not source.is_dir() or not template.is_file():
|
||
raise ValueError('旧项目目录或新版框架 ZIP 不存在')
|
||
if source == target or source in target.parents or target in source.parents:
|
||
raise ValueError('升级目标应放在旧目录的平级位置，不得是旧目录自身或其内部目录')
|
||
if target.exists():
|
||
raise FileExistsError('目标已存在；请使用新的目标目录，或 export + restore 更新已有项目')
|
||
if dry_run:
|
||
export(source, target.with_suffix('.preview.zip'), dry_run=True)
|
||
return
|
||
with tempfile.TemporaryDirectory(prefix='c4h-bundle-') as d:
|
||
pkg = Path(d)/'research.c4h.zip'
|
||
export(source, pkg)
|
||
initialize_template(template, target)
|
||
try:
|
||
restore(pkg, target)
|
||
except Exception:
|
||
print(f'迁移中断，新目录仍保留用于检查：{target}；旧项目未修改', file=sys.stderr)
|
||
raise
|
||
print(f'一键升级完成：{target}；旧项目保留：{source}')
|
||
|
||
|
||
def main(argv=None):
|
||
parser = argparse.ArgumentParser(description='Code4History 跨版本研究数据迁移（只复制，不删旧研究）')
|
||
sub = parser.add_subparsers(dest='command', required=True)
|
||
a = sub.add_parser('export', help='从旧项目导出研究内容与框架个性化修改')
|
||
a.add_argument('--source', type=Path, required=True)
|
||
a.add_argument('--out', type=Path, required=True)
|
||
a.add_argument('--dry-run', action='store_true')
|
||
b = sub.add_parser('restore', help='恢复研究资料至已解压的新框架目录')
|
||
b.add_argument('--bundle', type=Path, required=True)
|
||
b.add_argument('--target', type=Path, required=True)
|
||
b.add_argument('--dry-run', action='store_true')
|
||
c = sub.add_parser('upgrade', help='一键创建新框架并恢复旧研究')
|
||
c.add_argument('--source', type=Path, required=True)
|
||
c.add_argument('--template', type=Path, required=True)
|
||
c.add_argument('--target', type=Path, required=True)
|
||
c.add_argument('--dry-run', action='store_true')
|
||
args = parser.parse_args(argv)
|
||
try:
|
||
if args.command == 'export': export(args.source, args.out, dry_run=args.dry_run)
|
||
elif args.command == 'restore': restore(args.bundle, args.target, dry_run=args.dry_run)
|
||
elif args.command == 'upgrade': upgrade(args.source, args.template, args.target, dry_run=args.dry_run)
|
||
except (ValueError, OSError, RuntimeError) as e:
|
||
print(f'错误：{e}', file=sys.stderr)
|
||
return 1
|
||
return 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main())
|