mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-30 18:56:16 +00:00
allaun
779f4fe413
Add concept extraction/orchestration scripts under scripts/ and infra metadata in infra.json. Generated extraction outputs live under extraction/ and are regenerated from scripts; add extraction/ to .gitignore.
212 lines
7.2 KiB
Python
212 lines
7.2 KiB
Python
import json, re
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
OUT = ROOT / "extraction"
|
|
OUT.mkdir(exist_ok=True)
|
|
|
|
SKIP_DIRS = {".git","node_modules","__pycache__","venv",".venv","result",
|
|
".lake","target","lean_binned","scratch","shared-data",
|
|
"archive","result-devcontainer","extraction"}
|
|
|
|
LEAN_TOP = re.compile(r'^(def|theorem|structure|inductive|class|abbrev)\s+(\w+)\b', re.MULTILINE)
|
|
LEAN_DOC = re.compile(r'/-(.*?)-/', re.DOTALL)
|
|
PY_DEF = re.compile(r'^(def|class|async\s+def)\s+(\w+)\(')
|
|
MD_HEADER = re.compile(r'^(#{1,6})\s+(.+)$', re.MULTILINE)
|
|
|
|
|
|
def infer_lean_status(kind, text):
|
|
if kind in ('structure','inductive','class','abbrev'): return 'DEFINED'
|
|
if 'sorry' in text.lower(): return 'SORRY'
|
|
return 'DEFINED'
|
|
|
|
|
|
def scan_lean_file(path, rel):
|
|
text = path.read_text(errors='replace')
|
|
doc_matches = list(LEAN_DOC.finditer(text))
|
|
module_doc = doc_matches[0].group(1).strip().splitlines()[0] if doc_matches else ''
|
|
results = []
|
|
for m in LEAN_TOP.finditer(text):
|
|
kind, name = m.group(1), m.group(2)
|
|
start = m.start()
|
|
context = text[max(0, start-100):start+300]
|
|
ns = str(rel.with_suffix('')).replace('/', '.')
|
|
full = f'Semantics.{ns}.{name}'
|
|
desc = module_doc[:100]
|
|
for dm in reversed(doc_matches):
|
|
if dm.start() < start:
|
|
snippet = [l.strip() for l in dm.group(1).strip().splitlines() if l.strip()]
|
|
desc = next((l for l in snippet if not l.startswith('-') and len(l) > 5), desc)
|
|
break
|
|
status = infer_lean_status(kind, context)
|
|
related = []
|
|
for mm in LEAN_TOP.finditer(text, start+1, start+2000):
|
|
if mm.group(2) != name:
|
|
related.append(f'math_{mm.group(2).lower()}')
|
|
if len(related) >= 3:
|
|
break
|
|
results.append({
|
|
'id': f'math_{name.lower()}',
|
|
'type': kind if kind != 'abbrev' else 'type',
|
|
'name': name,
|
|
'expression': name,
|
|
'latex': name,
|
|
'description': desc[:200],
|
|
'source_file': f'Semantics/Semantics/{rel}',
|
|
'lean_name': full,
|
|
'lean_status': status,
|
|
'tags': [ns.split('.')[-2] if '.' in ns else ns],
|
|
'related': related
|
|
})
|
|
return results
|
|
|
|
|
|
def scan_python_file(path, rel):
|
|
text = path.read_text(errors='replace')
|
|
comment = ''
|
|
first = text.lstrip()
|
|
if first.startswith('#'):
|
|
comment = first.split('\n', 1)[0][:120]
|
|
results = []
|
|
for _, name in PY_DEF.findall(text):
|
|
results.append({
|
|
'id': f'idea_{name.lower()}',
|
|
'label': name,
|
|
'title': name,
|
|
'summary': comment or f'Python function in {rel}',
|
|
'status': 'IMPLEMENTED',
|
|
'source': str(rel),
|
|
'tags': ['python','shim'],
|
|
'edges': []
|
|
})
|
|
return results[:200]
|
|
|
|
|
|
def scan_md_file(path, rel):
|
|
text = path.read_text(errors='replace')
|
|
results = []
|
|
stem = path.stem.replace('-', '')
|
|
results.append({
|
|
'id': f'idea_{stem.lower()}_doc',
|
|
'label': stem[:40],
|
|
'title': f'Doc: {path.name}',
|
|
'summary': text[:200].replace('\n', ' '),
|
|
'status': 'DOCUMENTED',
|
|
'source': str(rel),
|
|
'tags': ['markdown','doc'],
|
|
'edges': []
|
|
})
|
|
for m in MD_HEADER.finditer(text):
|
|
title = m.group(2).strip()
|
|
if len(title) > 5:
|
|
h = hash(title) & 0xFFFFFFFF
|
|
results.append({
|
|
'id': f'idea_doc_{h:08x}',
|
|
'label': title[:40].replace(' ', ''),
|
|
'title': title,
|
|
'summary': text[m.end():m.end()+200].replace('\n', ' '),
|
|
'status': 'DOCUMENTED',
|
|
'source': str(rel),
|
|
'tags': ['markdown','section'],
|
|
'edges': []
|
|
})
|
|
return results[:80]
|
|
|
|
|
|
def walk(base, suffixes):
|
|
stack = [base]
|
|
while stack:
|
|
p = stack.pop()
|
|
if p.is_dir():
|
|
if p.name in SKIP_DIRS:
|
|
continue
|
|
try:
|
|
for c in sorted(p.iterdir()):
|
|
stack.append(c)
|
|
except PermissionError:
|
|
continue
|
|
elif p.is_file() and p.suffix in suffixes:
|
|
yield p
|
|
|
|
all_math = []
|
|
|
|
def main():
|
|
global all_math
|
|
docs_ideas, infra_ideas, app_ideas = [], [], []
|
|
|
|
print('[1/4] Lean...', flush=True)
|
|
ld = ROOT / '0-Core-Formalism' / 'lean' / 'Semantics' / 'Semantics'
|
|
if ld.exists():
|
|
for p in walk(ld, {'.lean'}):
|
|
try:
|
|
all_math.extend(scan_lean_file(p, p.relative_to(ld)))
|
|
except Exception:
|
|
pass
|
|
print(f' {len(all_math)} Lean concepts', flush=True)
|
|
|
|
print('[2/4] Docs...', flush=True)
|
|
dd = ROOT / '6-Documentation'
|
|
if dd.exists():
|
|
for p in walk(dd, {'.md'}):
|
|
try:
|
|
docs_ideas.extend(scan_md_file(p, p.relative_to(ROOT)))
|
|
except Exception:
|
|
pass
|
|
for p in [ROOT/'README.md', ROOT/'AGENTS.md', ROOT/'ARCHITECTURE.md',
|
|
ROOT/'CHANGELOG.md']:
|
|
if p.exists():
|
|
try:
|
|
docs_ideas.extend(scan_md_file(p, p.name))
|
|
except Exception:
|
|
pass
|
|
print(f' {len(docs_ideas)} doc concepts', flush=True)
|
|
|
|
print('[3/4] Infrastructure...', flush=True)
|
|
sd = ROOT / '4-Infrastructure' / 'shim'
|
|
if sd.exists():
|
|
for p in walk(sd, {'.py'}):
|
|
try:
|
|
infra_ideas.extend(scan_python_file(p, p.relative_to(ROOT)))
|
|
except Exception:
|
|
pass
|
|
print(f' {len(infra_ideas)} infra concepts', flush=True)
|
|
|
|
print('[4/4] Applications...', flush=True)
|
|
app_dir = ROOT / '5-Applications'
|
|
if app_dir.exists():
|
|
for p in walk(app_dir, {'.py'}):
|
|
try:
|
|
app_ideas.extend(scan_python_file(p, p.relative_to(ROOT)))
|
|
except Exception:
|
|
pass
|
|
print(f' {len(app_ideas)} app concepts', flush=True)
|
|
|
|
print('Writing outputs...', flush=True)
|
|
|
|
def write_json(name, obj, count):
|
|
p = OUT / name
|
|
p.write_text(json.dumps(obj, indent=2))
|
|
print(f' {name}: {count} items ({p.stat().st_size:,} bytes)', flush=True)
|
|
|
|
write_json('lean_concepts.json', {'math': all_math}, len(all_math))
|
|
write_json('docs_concepts.json', {'ideas': docs_ideas}, len(docs_ideas))
|
|
write_json('infra_concepts.json', {'ideas': infra_ideas}, len(infra_ideas))
|
|
|
|
merged = []
|
|
for it in docs_ideas + infra_ideas + app_ideas:
|
|
merged.append({
|
|
'id': it['id'], 'type': 'concept', 'name': it['title'],
|
|
'expression': '', 'latex': '', 'description': it['summary'][:200],
|
|
'source_file': it['source'], 'lean_name': '',
|
|
'lean_status': 'IDEA', 'tags': it.get('tags', []), 'related': []
|
|
})
|
|
merged.extend(all_math)
|
|
merged = merged[:40000]
|
|
p = OUT / 'all_concepts_merged.json'
|
|
p.write_text(json.dumps({'math': merged}, indent=2))
|
|
print(f'\nTotal: {len(merged)} concepts extracted.', flush=True)
|
|
print(f' all_concepts_merged.json: {len(merged)} items ({p.stat().st_size:,} bytes)', flush=True)
|
|
print('Done.', flush=True)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|