Python 64.6%
TypeScript 33.7%
CSS 0.8%
1"""Ingestion: course website HTML (generated notes), PDF, DOCX, PPTX, MD → chunks → DB."""23from __future__ import annotations45import hashlib6import re7from pathlib import Path89from selectolax.parser import HTMLParser, Node10from sqlalchemy import delete, select1112from app.core.logging import get_logger13from app.db import SessionLocal14from app.models import CourseChunk, CourseDocument15from app.rag.chunking import Chunk, chunk_text1617log = get_logger("rag.ingest")1819SITE_URLS = {"IMM1003": "https://www.uqo-imm1003.app", "IMM1033": "https://www.uqo-imm1033.app"}202122# ------------------------------------------------------------------ HTML helpers23def _katex_to_tex(root: Node) -> None:24 for k in root.css("span.katex, span.katex-display"):25 ann = k.css_first("annotation")26 tex = ann.text() if ann else ""27 display = "katex-display" in (k.attributes.get("class") or "")28 k.replace_with(f" $${tex}$$ " if display else f" ${tex}$ ")293031def _clean(root: Node) -> None:32 for sel in ("script", "style", "svg", "nav", "button", ".h-anchor", ".box-anchor",33 ".widget", ".explorer", "form", "input", ".ch-hero .deco", ".watermark"):34 for n in root.css(sel):35 n.decompose()363738def _node_text(n: Node) -> str:39 txt = n.text(separator=" ", strip=False)40 return re.sub(r"[ \t]+", " ", txt).strip()414243def _block_to_text(n: Node) -> str:44 tag = n.tag45 cls = n.attributes.get("class") or ""46 if tag in {"h2", "h3", "h4"}:47 return f"\n\n{'#' * (int(tag[1]))} {_node_text(n)}\n\n"48 if "box" in cls.split() or cls.startswith("box"):49 label = n.css_first(".box-label")50 title = n.css_first(".box-title")51 body = n.css_first(".box-body")52 head = " — ".join(x.text(strip=True) for x in (label, title) if x is not None)53 return f"\n\n**{head}** : {_node_text(body) if body else _node_text(n)}\n\n"54 if tag in {"ul", "ol"}:55 items = [f"- {_node_text(li)}" for li in n.css("li")]56 return "\n" + "\n".join(items) + "\n\n"57 if tag == "table":58 rows = []59 for tr in n.css("tr"):60 cells = [_node_text(td) for td in tr.css("th, td")]61 rows.append(" | ".join(cells))62 return "\n" + "\n".join(rows) + "\n\n"63 if tag in {"p", "div", "blockquote", "figure", "figcaption", "dl", "pre"}:64 return "\n\n" + _node_text(n) + "\n\n"65 return " " + _node_text(n) + " "666768def _flatten(container: Node) -> str:69 out = []70 for child in container.iter(include_text=True):71 if child.tag == "-text":72 t = child.text(strip=False)73 if t and t.strip():74 out.append(" " + t.strip() + " ")75 else:76 out.append(_block_to_text(child))77 text = "".join(out)78 text = re.sub(r"[ \t]+", " ", text)79 text = re.sub(r"\n{3,}", "\n\n", text)80 return text.strip()818283def _split_sections(article: Node) -> list[tuple[str, str, str]]:84 """Return [(section_id, section_title, text)] split on h2."""85 sections: list[tuple[str, str, list[str]]] = []86 current: tuple[str, str, list[str]] | None = ("", "Introduction", [])87 for child in article.iter(include_text=True):88 if child.tag == "h2":89 if current and "".join(current[2]).strip():90 sections.append(current)91 sid = child.attributes.get("id") or ""92 current = (sid, _node_text(child), [])93 elif child.tag == "-text":94 t = child.text(strip=False)95 if t and t.strip() and current is not None:96 current[2].append(" " + t.strip() + " ")97 else:98 if current is not None:99 current[2].append(_block_to_text(child))100 if current and "".join(current[2]).strip():101 sections.append(current)102 out = []103 for sid, title, parts in sections:104 text = re.sub(r"\n{3,}", "\n\n", re.sub(r"[ \t]+", " ", "".join(parts))).strip()105 out.append((sid, title, text))106 return out107108109def parse_seance_html(html: str, course: str, num: str) -> tuple[str, list[Chunk]]:110 tree = HTMLParser(html)111 main = tree.css_first("main")112 if main is None:113 return "", []114 _clean(main)115 _katex_to_tex(main)116 title_attr = main.attributes.get("data-title") or ""117 h1 = main.css_first("h1")118 chapter = title_attr or (h1.text(strip=True) if h1 else f"Séance {num}")119 module = chapter if chapter.lower().startswith("séance") else f"Séance {int(num)} — {chapter}"120 article = main.css_first("article") or main121 base = f"{SITE_URLS.get(course, '')}/seance/{num}/"122 chunks: list[Chunk] = []123 for sid, sec_title, text in _split_sections(article):124 if len(text) < 80:125 continue126 url = f"{base}#{sid}" if sid else base127 for c in chunk_text(text, module=module, section=sec_title, page=f"S{int(num)}", url=url):128 c.metadata = {"kind": "notes", "seance": int(num)}129 chunks.append(c)130 return module, chunks131132133def parse_generic_page(html: str, course: str, slug: str, title: str) -> list[Chunk]:134 tree = HTMLParser(html)135 main = tree.css_first("main")136 if main is None:137 return []138 _clean(main)139 _katex_to_tex(main)140 entries = main.css(".gl-entry")141 chunks: list[Chunk] = []142 url = f"{SITE_URLS.get(course, '')}/{slug}/"143 if entries: # glossary: pack definitions144 buf: list[str] = []145 for e in entries:146 term = e.css_first("h3")147 body = e.css_first(".gl-def")148 buf.append(f"**{term.text(strip=True) if term else ''}** : "149 f"{_node_text(body) if body else _node_text(e)}")150 text = "\n\n".join(buf)151 for c in chunk_text(text, module=title, section="Définitions", page="G", url=url):152 c.metadata = {"kind": "glossary"}153 chunks.append(c)154 return chunks155 text = _flatten(main)156 for c in chunk_text(text, module=title, section=title, page="", url=url):157 c.metadata = {"kind": slug}158 chunks.append(c)159 return chunks160161162# ------------------------------------------------------------------ file parsers163def parse_pdf(path: Path) -> list[tuple[str, str]]:164 import pdfplumber165166 pages = []167 with pdfplumber.open(str(path)) as pdf:168 for i, page in enumerate(pdf.pages, 1):169 text = page.extract_text() or ""170 for table in page.extract_tables() or []:171 text += "\n\n" + "\n".join(" | ".join(str(c or "") for c in row) for row in table)172 if text.strip():173 pages.append((str(i), text))174 return pages175176177def parse_docx(path: Path) -> str:178 import docx179180 d = docx.Document(str(path))181 parts = [p.text for p in d.paragraphs if p.text.strip()]182 for t in d.tables:183 for row in t.rows:184 parts.append(" | ".join(c.text for c in row.cells))185 return "\n\n".join(parts)186187188def parse_pptx(path: Path) -> list[tuple[str, str]]:189 from pptx import Presentation190191 prs = Presentation(str(path))192 out = []193 for i, slide in enumerate(prs.slides, 1):194 texts = []195 for shape in slide.shapes:196 if shape.has_text_frame:197 texts.append(shape.text_frame.text)198 if any(t.strip() for t in texts):199 out.append((str(i), "\n".join(texts)))200 return out201202203# ------------------------------------------------------------------ persistence204def _checksum(data: bytes) -> str:205 return hashlib.sha256(data).hexdigest()206207208async def _replace_document(course: str, filename: str, title: str, checksum: str,209 chunks: list[Chunk], visibility: str) -> CourseDocument:210 async with SessionLocal() as session:211 existing = await session.execute(212 select(CourseDocument).where(CourseDocument.course_code == course,213 CourseDocument.filename == filename))214 for doc in existing.scalars():215 await session.execute(delete(CourseChunk).where(CourseChunk.document_id == doc.id))216 await session.delete(doc)217 doc = CourseDocument(course_code=course, filename=filename, title=title,218 checksum=checksum, visibility=visibility, n_chunks=len(chunks))219 session.add(doc)220 await session.flush()221 for c in chunks:222 session.add(CourseChunk(document_id=doc.id, course_code=course, module=c.module,223 section=c.section, page=c.page, url=c.url,224 content=c.content, metadata_=c.metadata,225 visibility=visibility))226 await session.commit()227 return doc228229230async def ingest_site(course: str, dist_dir: Path, visibility: str = "students") -> int:231 """Ingest a generated course website (dist/<course>/)."""232 course = course.upper()233 total = 0234 for seance_dir in sorted((dist_dir / "seance").glob("*/")):235 page = seance_dir / "index.html"236 if not page.exists():237 continue238 html = page.read_bytes()239 module, chunks = parse_seance_html(html.decode("utf-8", errors="replace"), course,240 seance_dir.name)241 if chunks:242 await _replace_document(course, f"seance-{seance_dir.name}.html", module,243 _checksum(html), chunks, visibility)244 total += len(chunks)245 log.info("ingested", course=course, doc=module, n=len(chunks))246 for slug, title in (("glossaire", "Glossaire"), ("aide-memoire", "Aide-mémoire des formules"),247 ("fiches", "Fiches synthèse"), ("definitions", "Définitions"),248 ("exercices", "Exercices"), ("ressources", "Ressources")):249 page = dist_dir / slug / "index.html"250 if not page.exists():251 continue252 html = page.read_bytes()253 chunks = parse_generic_page(html.decode("utf-8", errors="replace"), course, slug, title)254 if chunks:255 await _replace_document(course, f"{slug}.html", title, _checksum(html), chunks,256 visibility)257 total += len(chunks)258 log.info("ingested", course=course, doc=title, n=len(chunks))259 return total260261262async def ingest_file(course: str, path: Path, visibility: str = "students",263 module: str | None = None) -> int:264 course = course.upper()265 data = path.read_bytes()266 ext = path.suffix.lower()267 title = module or path.stem.replace("_", " ")268 chunks: list[Chunk] = []269 if ext == ".pdf":270 for page, text in parse_pdf(path):271 chunks += chunk_text(text, module=title, section=f"p. {page}", page=page)272 elif ext == ".docx":273 chunks = chunk_text(parse_docx(path), module=title, section=title)274 elif ext == ".pptx":275 for slide, text in parse_pptx(path):276 chunks += chunk_text(text, module=title, section=f"diapo {slide}", page=slide)277 elif ext in {".md", ".txt", ".tex"}:278 text = data.decode("utf-8", errors="replace")279 if ext == ".tex":280 text = re.sub(r"\\[a-zA-Z]+\*?(\[[^\]]*\])?", " ", text)281 text = re.sub(r"[{}%]", " ", text)282 chunks = chunk_text(text, module=title, section=title)283 elif ext in {".html", ".htm"}:284 chunks = parse_generic_page(data.decode("utf-8", errors="replace"), course, path.stem,285 title)286 else:287 raise ValueError(f"Format non pris en charge : {ext}")288 for c in chunks:289 c.metadata = {"kind": "upload", "file": path.name}290 await _replace_document(course, path.name, title, _checksum(data), chunks, visibility)291 return len(chunks)292