coding-agent/GraphRAG增强版_20260920/course-source/build_course.py

"""保留原 MkDocs Material 壳；增量生成图增强教材，所有源和产物只在副本中。"""
from pathlib import Path
from bs4 import BeautifulSoup
from urllib.parse import urlsplit, urlunsplit, unquote, quote
import os,re,json,ast,hashlib,copy,csv,shutil,html,sys
import markdown
from pygments import highlight
from pygments.lexers import PythonLexer
from pygments.formatters import HtmlFormatter
from draw_diagrams import main as draw_diagrams
ROOT=Path(__file__).resolve().parents[1]
SRC=ROOT/'course-source';SITE=ROOT/'课件'/'EnterpriseKB_RAG_V1.0讲义';OLD=SRC/'legacy-html';APP=ROOT/'代码'/'EnterpriseKB_qa_system';RESEARCH=ROOT/'research'
RESEARCH.mkdir(exist_ok=True)
D={'⑤ RAG系统评估':'⑥ RAG系统评估','⑥ 融合Mysql的RAG系统':'⑦ 融合MySQL、Milvus与Neo4j','⑦ 企业级生产环境部署':'⑧ 企业级生产环境部署'}
G5='⑤ 基于Neo4j的知识图谱增强问答系统'
EXTRA={'6.4':'⑥ RAG系统评估','7.5':'⑦ 融合MySQL、Milvus与Neo4j','8.3':'⑧ 企业级生产环境部署'}
NUM={'5':'6','6':'7','7':'8'}
LEDGER=[]
SNIPPETS={
'schema-load':('graph_qa/schema.py','load_dataset','    for fact in facts:', '    if any(e["type"]', '5.3'),
'settings':('graph_qa/config.py','GraphSettings', '    milvus_collection:', '    @classmethod','5.4'),
'graph-constraints':('graph_qa/db.py','GraphStore.ensure_schema',None,None,'5.4'),
'graph-import':('graph_qa/db.py','GraphStore._replace_document','        for fact in facts:',None,'5.4'),
'validate-extraction':('graph_qa/extractor.py','validate_extraction',None,None,'5.5'),
'llm-extract':('graph_qa/extractor.py','LLMExtractor.extract',None,None,'5.5'),
'normalize':('graph_qa/entity_resolver.py','normalize',None,None,'5.6'),
'entity-resolve':('graph_qa/entity_resolver.py','EntityResolver.resolve',None,None,'5.6'),
'dependency-paths':('graph_qa/retriever.py','GraphRetriever._dependency_paths','        for selected in directions:',None,'5.7'),
'attach-responsibility':('graph_qa/retriever.py','GraphRetriever._attach_responsibility',None,None,'5.7'),
'evidence-bundle':('graph_qa/schema.py','EvidenceBundle',None,None,'5.8'),
'graph-evidence':('graph_qa/retriever.py','GraphRetriever.retrieve','                chunk_ids =','                bundles.append','5.8'),
'build-context':('graph_qa/evidence_builder.py','build_context','    for bundle in bundles:','    paths = []','5.8'),
'rrf':('graph_qa/fusion.py','reciprocal_rank_fusion',None,None,'5.9'),
'rerank':('graph_qa/fusion.py','rerank_candidates',None,None,'5.9'),
'parallel-retrieve':('graph_qa/pipeline.py','GraphPipeline.retrieve','        executor =','        channels, warnings','5.9'),
'select-context':('graph_qa/fusion.py','select_context',None,None,'5.9'),
'make-prompt':('graph_qa/evidence_builder.py','make_prompt','    if scope in','    else:','5.10'),
'prompt-context':('graph_qa/evidence_builder.py','make_prompt','    return f',None,'5.10'),
'answer':('graph_qa/pipeline.py','GraphPipeline.answer',None,None,'5.10'),
'validate-citations':('graph_qa/evidence_builder.py','validate_citations',None,None,'5.10'),
'remove-document':('graph_qa/db.py','GraphStore._remove_document',None,None,'5.11'),
'ingest-journal':('graph_qa/pipeline.py','GraphPipeline.ingest','        journal =','        return {','5.11'),
'channel-errors':('graph_qa/pipeline.py','GraphPipeline.retrieve','        for future in done:','        executor.shutdown','5.11'),
'evaluation-metrics':('evaluation_graph/run_evaluation.py','score_result',None,None,'5.12,6.4'),
'integration-hook':('rag_qa/core/new_rag_system.py','RAGSystem.retrieve_and_merge','        if self.graph_pipeline is not None:','        # 如果未指定','7.5'),
'api-response':('graph_qa/api.py','query',None,None,'7.5'),
'model-paths':('graph_qa/model_runtime.py','ModelRuntime.__init__',None,None,'8.3'),
}

def mapped(path):
 p=Path(path);parts=list(p.parts)
 if parts and parts[0] in D:
  old=parts[0];parts[0]=D[old]
  if len(parts)>1 and re.match(r'[567]\.\d',parts[-1]):parts[-1]=NUM[parts[-1][0]]+parts[-1][1:]
 return '/'.join(parts)

def renumber_text(text):
 for old,new in D.items():text=text.replace(old,new)
 text=re.sub(r'(?<![\d.])([567])\.(?=\d)',lambda m:NUM[m[1]]+'.',text)
 text=re.sub(r'第([五六七])章',lambda m:'第'+{'五':'六','六':'七','七':'八'}[m[1]]+'章',text)
 return text

def encode(path):return quote(path,safe='/#?=&:()')
def relpath(target,current):return encode(os.path.relpath(SITE/target,(SITE/current).parent).replace(os.sep,'/'))
def rewrite_url(url,old_path,new_path):
 if not url or url.startswith(('#','mailto:','data:','javascript:')):return url
 q=urlsplit(url)
 if q.scheme or q.netloc:return url
 decoded=unquote(q.path)
 target=os.path.normpath(str(Path(old_path).parent/decoded)).replace(os.sep,'/')
 target=mapped(target)
 relative=os.path.relpath(SITE/target,(SITE/new_path).parent).replace(os.sep,'/')
 return urlunsplit(('', '',encode(relative),q.query,q.fragment))

def snippet(key):
 path,symbol,start,end,lessons=SNIPPETS[key];p=APP/path
 if not p.exists():raise ValueError(f'Missing source {p}')
 source=p.read_text();tree=ast.parse(source);node=None
 def find(nodes,parts):
  for n in nodes:
   if isinstance(n,(ast.FunctionDef,ast.AsyncFunctionDef,ast.ClassDef)) and n.name==parts[0]:
    if len(parts)==1:return n
    return find(n.body,parts[1:])
 node=find(tree.body,symbol.split('.'))
 if not node:raise ValueError(f'Symbol not found: {path}:{symbol}')
 lines=source.splitlines();a=node.lineno-1;b=node.end_lineno
 if node.decorator_list:a=min(d.lineno-1 for d in node.decorator_list)
 if start:
  a=next((i for i in range(a,b) if start in lines[i]),None)
  if a is None:raise ValueError(f'Start not found: {key}: {start}')
 if end:
  found=next((i for i in range(a+1,b) if end in lines[i]),None)
  if found is not None:b=found
 text='\n'.join(lines[a:b])
 # Long mechanisms are split into consecutive coherent excerpts, never silently truncated.
 import textwrap
 text=textwrap.dedent(text)
 LEDGER.append({'lesson_id':lessons,'capability':key,'concept':symbol,'source_path':path,'symbols':symbol,'line_spans':f'{a+1}-{b}','source_sha256':hashlib.sha256(source.encode()).hexdigest(),'context_symbols':'相关类型与上下游在正文解释','upstream_downstream':'CLI / API → GraphPipeline → 检索或存储模块','excerpt_reason':'保留本知识点对应的真实机制与接口','omissions':'文件中的其他职责与初始化代码','adaptation':'verbatim','verification_evidence':'验证结果/graphrag-evaluation.json; tests_graph/test_graph_integration.py; tests_graph/test_api_live.py'})
 chunks=[text]
 if len(text.splitlines())>36:
  # Whole function may include setup and validation sections. Keep source in order but split at an empty line.
  ls=text.splitlines();cut=min(range(14,min(len(ls),30)),key=lambda i: 0 if not ls[i].strip() else 1)
  if not ls[cut].strip():chunks=['\n'.join(ls[:cut]),'\n'.join(ls[cut+1:])]
 label=f'<p class="graph-code-location"><code>{html.escape(path)}</code> · <code>{html.escape(symbol)}</code>（本知识点节选）</p>'
 parts=[]
 for i,c in enumerate(chunks):
  block=highlight(c,PythonLexer(),HtmlFormatter(cssclass='highlight',wrapcode=True))
  if i:parts.append('<p>继续沿用上面已经准备的变量，下面完成这个方法余下的校验与返回：</p>')
  parts.append(block)
 return label+'\n'.join(parts)

def fixture(key):
 data=APP/'graph_qa/data/dessert-shop'
 if key=='tea-fact':
  rows=[json.loads(line) for line in (data/'facts.jsonl').read_text().splitlines() if line.strip()]
  row=next(r for r in rows if r['id']=='f-tea-cake')
  entities=json.loads((data/'entities.json').read_text())
  names={e['id']:e['name'] for e in entities}
  relation_names={'DEPENDS_ON':'需要','CONTAINS':'包含','OWNED_BY':'由谁负责','SOLD_AT':'在哪里售卖','USED_BY':'供谁使用'}
  display={'左边节点':names[row['subject']], '关系':relation_names.get(row['relation'],row['relation']), '右边节点':names[row['object']], '知识域':'甜品店' if row['scope']=='dessert-shop' else '另一门店', '原文编号':row['chunk_id'], '引用原句':row['evidence_quote']}
  return '<div class="highlight"><pre><code>'+html.escape(json.dumps(display,ensure_ascii=False,indent=2))+'</code></pre></div>'
 raise ValueError('Unknown fixture '+key)

def figure(match,current):
 name,caption=match.group(1),match.group(2)
 url=relpath('graph_assets/'+name+'.svg',current)
 return f'<figure><a href="{url}" target="_blank" rel="noopener" title="打开原尺寸图解"><img src="{url}" alt="{html.escape(caption)}" loading="lazy"/></a><figcaption>{html.escape(caption)}</figcaption></figure>'

def runtime(key):
 p=SRC/'runtime-sections.json'
 if p.exists():
  content=json.loads(p.read_text()).get(key)
  if content:return content
 return '<div class="admonition note"><p class="admonition-title">课堂运行与核对</p><p>按本节命令执行后，以增强版“验证结果”中的记录核对证据和回答。建图基线使用配套人工核对资料；不要将核对答案当作运行指标。</p></div>'

def nav_html(pages,current):
 groups=[]
 for p,title in pages:
  if p=='index.html':continue
  g=p.split('/')[0]
  if not groups or groups[-1][0]!=g:groups.append((g,[]))
  groups[-1][1].append((p,title))
 out=['<nav class="md-nav md-nav--primary" aria-label="导航栏" data-md-level="0"><label class="md-nav__title" for="__drawer">企业知识库 RAG · 图增强版</label><ul class="md-nav__list" data-md-scrollfix>']
 out.append(f'<li class="md-nav__item"><a class="md-nav__link {"md-nav__link--active" if current=="index.html" else ""}" href="{relpath("index.html",current)}"><span class="md-ellipsis">知识库内容</span></a></li>')
 for i,(g,items) in enumerate(groups,2):
  active=any(p==current for p,_ in items);n=f'__nav_{i}'
  out.append(f'<li class="md-nav__item md-nav__item--nested {"md-nav__item--active" if active else ""}"><input class="md-nav__toggle md-toggle" type="checkbox" id="{n}" {"checked" if active else ""}><label class="md-nav__link" for="{n}" id="{n}_label" tabindex="0"><span class="md-ellipsis">{html.escape(g)}</span><span class="md-nav__icon md-icon"></span></label><nav class="md-nav" data-md-level="1" aria-labelledby="{n}_label" aria-expanded="{str(active).lower()}"><label class="md-nav__title" for="{n}"><span class="md-nav__icon md-icon"></span>{html.escape(g)}</label><ul class="md-nav__list" data-md-scrollfix>')
  for p,title in items:
   out.append(f'<li class="md-nav__item {"md-nav__item--active" if p==current else ""}"><a class="md-nav__link {"md-nav__link--active" if p==current else ""}" href="{relpath(p,current)}"><span class="md-ellipsis">{html.escape(title)}</span></a></li>')
  out.append('</ul></nav></li>')
 out.append('</ul></nav>');return ''.join(out)

def html_fragment(s):return BeautifulSoup(s,'html.parser')

def make_toc(soup,article):
 for i,h in enumerate(article.select('h2,h3,h4')):
  if not h.get('id'):h['id']='section-'+str(i+1)
  if not h.select_one('.headerlink'):
   a=soup.new_tag('a',attrs={'class':'headerlink','href':'#'+h['id'],'title':'本节链接'});a.string='¶';h.append(a)
 links=[]
 for h in article.select('h2,h3'):
  title=h.get_text(' ',strip=True).rstrip('¶').strip();links.append(f'<li class="md-nav__item"><a class="md-nav__link" href="#{h["id"]}"><span class="md-ellipsis">{html.escape(title)}</span></a></li>')
 toc='<nav class="md-nav md-nav--secondary" aria-label="目录"><label class="md-nav__title" for="__toc"><span class="md-nav__icon md-icon"></span>目录</label><ul class="md-nav__list" data-md-component="toc" data-md-scrollfix>'+''.join(links)+'</ul></nav>'
 sidebar=soup.select_one('.md-sidebar--secondary .md-sidebar__inner')
 if sidebar:sidebar.clear();sidebar.append(html_fragment(toc))
 else:
  primary=soup.select_one('.md-sidebar--primary')
  if primary:primary.insert_after(html_fragment('<div class="md-sidebar md-sidebar--secondary" data-md-component="sidebar" data-md-type="toc"><div class="md-sidebar__scrollwrap"><div class="md-sidebar__inner">'+toc+'</div></div></div>'))


def standardize(soup,current,pages,title):
 nav=soup.select_one('nav.md-nav--primary')
 if nav:nav.replace_with(html_fragment(nav_html(pages,current)))
 article=soup.select_one('article')
 h1=article.select_one('h1')
 if h1:
  h1.clear();h1.string=title
 make_toc(soup,article)
 if current.startswith(G5) or current.split('/')[-1].startswith(('6.4 ','7.5 ','8.3 ')):article['class']=article.get('class',[])+['graph-lesson']
 soup.title.string=title+' - 企业知识库RAG智能问答项目·图增强版'
 topic=soup.select_one('[data-md-component="header-topic"] .md-ellipsis')
 if topic:topic.string=title
 for t in soup.find_all(['script','link']):
  src=t.get('src') or t.get('href') or ''
  if t.name=='link' and (t.get('rel')==['preconnect'] or 'fonts.googleapis' in src):t.decompose();continue
  if t.name=='script' and ('cdn.jsdelivr' in src or 'polyfill.io' in src or src.endswith(('javascripts/extra.js','javascripts/mathjax.js'))):t.decompose();continue
 for l in soup.select('link[rel="prev"],link[rel="next"],link[rel="canonical"]'):l.decompose()
 # Legacy pages may reference local MathJax config that did not exist. Add a complete local replacement.
 for s in soup.select('script[data-course-local],link[data-course-local]'):s.decompose()
 style=soup.new_tag('link',attrs={'rel':'stylesheet','href':relpath('graph_assets/course.css',current),'data-course-local':'true'});soup.head.append(style)
 cfg=soup.new_tag('script');cfg.string="window.MathJax={tex:{inlineMath:[['$','$'],['\\\\(','\\\\)']]},options:{skipHtmlTags:['script','noscript','style','textarea','pre','code']}};";soup.head.append(cfg)
 if (SITE/'assets/mathjax/tex-mml-chtml.js').exists():
  s=soup.new_tag('script',attrs={'src':relpath('assets/mathjax/tex-mml-chtml.js',current),'defer':'','data-course-local':'true'});soup.body.append(s)
 for asset in ['graph_assets/search-data.js','graph_assets/course.js']:
  s=soup.new_tag('script',attrs={'src':relpath(asset,current),'defer':'','data-course-local':'true'});soup.body.append(s)
 conf=soup.select_one('#__config')
 if conf:
  data=json.loads(conf.string);data['features']=[];conf.string=json.dumps(data,ensure_ascii=False)
 for prev in article.select('.graph-page-turn'):prev.decompose()
 seq=[p for p,_ in pages]
 if current in seq:
  idx=seq.index(current);links=[]
  for j,rel,label in [(idx-1,'prev','上一节'),(idx+1,'next','下一节')]:
   if 0<=j<len(pages):
    p,t=pages[j];u=relpath(p,current);links.append(f'<a href="{u}" rel="{rel}"><small>{label}</small><span>{html.escape(t)}</span></a>')
    soup.head.append(soup.new_tag('link',attrs={'rel':rel,'href':u}))
  article.append(html_fragment('<nav class="graph-page-turn" aria-label="前后章节">'+''.join(links)+'</nav>'))
 return soup

def build():
 draw_diagrams()
 for filename in ('course.css','course.js'):
  shutil.copy2(SRC/filename,SITE/'graph_assets'/filename)
 for old,new in D.items():
  if (SITE/old).exists() and not (SITE/new).exists():shutil.move(str(SITE/old),str(SITE/new))
 index=BeautifulSoup((OLD/'index.html').read_text(),'html.parser')
 pages=[]
 for a in index.select('nav.md-nav--primary a.md-nav__link'):
  p=unquote(a['href']);t=renumber_text(a.get_text(' ',strip=True)) if p.split('/')[0] in D else a.get_text(' ',strip=True)
  pages.append((mapped(p),t))
 lessons={}
 for p in (SRC/'lessons').glob('*.md'):
  content=p.read_text();title=content.splitlines()[0].removeprefix('# ');folder=G5 if p.stem.startswith('5.') else EXTRA[p.stem]
  output=folder+'/'+title+'.html';lessons[output]=(p,content,title)
  pages.append((output,title))
 def order(item):
  p,t=item
  if p=='index.html':return (0,0,0)
  m=re.match(r'(\d+)\.(\d+)',t)
  if m:return (int(m[1]),int(m[2]),0)
  if p.startswith('④'):return (4,99,0)
  return (99,0,0)
 pages.sort(key=order)
 (SRC/'pages.json').write_text(json.dumps(pages,ensure_ascii=False,indent=2))
 template=(OLD/'④ 基于Milvus库的问答系统/4.4 向量存储.html').read_text()
 for old in OLD.rglob('*.html'):
  op=old.relative_to(OLD).as_posix();np=mapped(op);soup=BeautifulSoup(old.read_text(),'html.parser')
  for tag in soup.find_all(True):
   for att in ['href','src','poster']:
    if tag.get(att):tag[att]=rewrite_url(tag[att],op,np)
  for text_node in soup.find_all(string=True):
   if not text_node.parent or text_node.parent.name in ['script','style','code','pre']:continue
   if text_node.find_parent(['pre','code']):continue
   updated=renumber_text(str(text_node))
   if updated!=str(text_node):text_node.replace_with(updated)
  art=soup.select_one('article')
  if not art:continue
  title=next((t for p,t in pages if p==np),art.select_one('h1').get_text(' ',strip=True) if art.select_one('h1') else '课程资料')
  if np=='index.html':
   intro=f'<section class="graph-home"><h2 id="graph-update">新增第五章：用关系补齐跨文档证据</h2><p>从“草莓缺货，为什么双人下午茶也受影响”出发，复用已有 Milvus 混合检索，增加 Neo4j 图召回。先看图和真实案例，再分知识点学习短代码，最后在同一数据上对照实验。</p><p><a href="{relpath(next(p for p,t in pages if t.startswith("5.1 ")),np)}">进入 5.1：先看图检索的价值 →</a></p><p>原评估、融合与部署内容依次顺延到第六、七、八章，原正文与配图保留。增强版全部修改发生在独立副本。</p></section>'
   art.insert(1,html_fragment(intro))
  soup=standardize(soup,np,pages,title)
  out=SITE/np;out.parent.mkdir(parents=True,exist_ok=True);out.write_text(str(soup))
  # Remove old-number filenames inside the renamed directories; source snapshots retain every version.
  if op.split('/')[0] in D:
   stale=SITE/D[op.split('/')[0]]/Path(op).name
   if stale!=out and stale.exists():stale.unlink()
 for np,(p,content,title) in lessons.items():
  content=re.sub(r'\{\{figure:([^|}]+)\|([^}]+)\}\}',lambda m:figure(m,np),content)
  content=re.sub(r'\{\{fixture:([^}]+)\}\}',lambda m:fixture(m[1]),content)
  content=re.sub(r'\{\{snippet:([^}]+)\}\}',lambda m:snippet(m[1]),content)
  content=re.sub(r'\{\{runtime:([^}]+)\}\}',lambda m:runtime(m[1]),content)
  body=markdown.markdown(content,extensions=['tables','fenced_code','codehilite','admonition','sane_lists'],extension_configs={'codehilite':{'css_class':'highlight','guess_lang':False,'wrapcode':True}})
  soup=BeautifulSoup(template,'html.parser');art=soup.select_one('article');art.clear();art.append(html_fragment(body))
  # Only new lessons: keep diagram labels readable on a phone, with contained scrolling.
  for diagram in art.select('figure'):
   picture=diagram.find('a',recursive=False)
   if not picture or not picture.find('img'):continue
   viewport=soup.new_tag('div',attrs={'class':'graph-diagram-scroll','tabindex':'0','role':'region','aria-label':'图解，可左右滑动或点图放大'})
   picture.wrap(viewport)
   hint=soup.new_tag('p',attrs={'class':'graph-diagram-hint'});hint.string='可左右滑动或点图放大'
   diagram.insert(0,hint)
  for tag in soup.find_all(True):
   for att in ['href','src']:
    if tag.get(att) and not tag.find_parent('article'):tag[att]=rewrite_url(tag[att],'④ 基于Milvus库的问答系统/4.4 向量存储.html',np)
  soup=standardize(soup,np,pages,title);out=SITE/np;out.parent.mkdir(parents=True,exist_ok=True);out.write_text(str(soup))
 docs=[]
 for p,title in pages:
  s=BeautifulSoup((SITE/p).read_text(),'html.parser');article=s.select_one('article')
  for nav in article.select('.graph-page-turn'):nav.decompose()
  docs.append({'location':p,'title':title,'text':article.get_text(' ',strip=True)})
 search={'config':{'lang':['zh'],'separator':'[\\s\\-]+','pipeline':['stopWordFilter'],'fields':{'title':{'boost':1000},'text':{'boost':1},'tags':{'boost':1000000}}},'docs':docs}
 (SITE/'search/search_index.json').write_text(json.dumps(search,ensure_ascii=False))
 (SITE/'graph_assets/search-data.js').write_text('window.COURSE_SEARCH_DOCS='+json.dumps(docs,ensure_ascii=False)+';')
 (SITE/'sitemap.xml').write_text('<?xml version="1.0" encoding="UTF-8"?><urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">'+''.join('<url><loc>'+html.escape(encode(p))+'</loc></url>' for p,_ in pages)+'</urlset>')
 gz=SITE/'sitemap.xml.gz'
 if gz.exists():
  import gzip;gz.write_bytes(gzip.compress((SITE/'sitemap.xml').read_bytes()))
 if LEDGER:
  with (RESEARCH/'course-source-to-lesson-ledger.csv').open('w',newline='') as f:
   w=csv.DictWriter(f,fieldnames=LEDGER[0].keys());w.writeheader();w.writerows(LEDGER)
 manifest={'site_root':str(SITE.relative_to(ROOT)),'formal_pages':len(pages),'added_pages':len(lessons),'preserved_html_sources':len(list(OLD.rglob('*.html'))),'diagrams':len(list((SITE/'graph_assets').glob('*.svg'))),'source_snippets':len(LEDGER),'build_command':'python course-source/build_course.py','layout':'原版 MkDocs Material HTML shell，原 CSS、代码块语法与组件保留'}
 (RESEARCH/'course-site-manifest.json').write_text(json.dumps(manifest,ensure_ascii=False,indent=2));print(json.dumps(manifest,ensure_ascii=False))

if __name__=='__main__':build()