Upload app.py

app.py

@@ -0,0 +1,533 @@
+#!/usr/bin/env python3
+"""
+Gradio UI for Vehicle Spec RAG — standalone, same pipeline as the notebook.
+
+Hugging Face Space files (nothing else required):
+  app.py
+  requirements.txt
+  vehicle_spec_extraction.ipynb   (optional, for the demo notebook)
+
+Upload a workshop-manual PDF in the UI, then ask for specs.
+Optional Space secrets: OPENAI_API_KEY, MISTRAL_API_KEY, HF_TOKEN
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import warnings
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+warnings.filterwarnings("ignore", category=DeprecationWarning)
+
+os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
+os.environ.setdefault("OMP_NUM_THREADS", "1")
+os.environ.setdefault("KMP_DUPLICATE_LIB_OK", "TRUE")
+
+ROOT = Path(__file__).resolve().parent
+try:
+    from dotenv import load_dotenv
+    load_dotenv(ROOT / ".env")
+except Exception:
+    pass
+
+import gradio as gr
+import pymupdf
+from langchain_community.vectorstores import FAISS
+from langchain_core.documents import Document
+from langchain_core.output_parsers import StrOutputParser
+from langchain_core.prompts import ChatPromptTemplate
+from langchain_core.runnables import RunnablePassthrough
+from langchain_huggingface import HuggingFaceEmbeddings
+
+OPENAI_KEY = os.getenv("OPENAI_API_KEY", "")
+MISTRAL_KEY = os.getenv("MISTRAL_API_KEY", "")
+HF_TOKEN = os.getenv("HF_TOKEN", "")
+has_openai = bool(OPENAI_KEY)
+has_mistral = bool(MISTRAL_KEY)
+has_hf = bool(HF_TOKEN)
+
+_BATCH = 128
+_LOCAL_PDF = next(ROOT.glob("*.pdf"), None)
+
+# ── PDF parsing (PyMuPDF, same as notebook) ───────────────────────────────────
+_NOISE = [
+    re.compile(r"file:///C:/TSO/[^\n]+", re.I),
+    re.compile(r"^\s*repair4less\s*$", re.I | re.M),
+    re.compile(r"^\s*Page\s+\d+\s+sur\s+\d+\s*$", re.I | re.M),
+    re.compile(r"^\s*\d{4}-\d{2}-\d{2}\s*$", re.M),
+    re.compile(r"^\s*2014 F-150 Workshop Manual\s*$", re.I | re.M),
+]
+_SECTION_RE = re.compile(r"SECTION\s+([\d\-A-Za-z]+):\s*([^\n]+)", re.I)
+
+
+@dataclass
+class PageText:
+    page_number: int
+    text: str
+    section_id: str | None = None
+    section_title: str | None = None
+    char_count: int = 0
+
+
+def clean_text(text: str) -> str:
+    cleaned = text.replace("\x00", " ")
+    for pattern in _NOISE:
+        cleaned = pattern.sub("", cleaned)
+    cleaned = re.sub(r"[ \t]+", " ", cleaned)
+    return re.sub(r"\n{3,}", "\n\n", cleaned).strip()
+
+
+def extract_pages(pdf_path: str | Path) -> list[PageText]:
+    pages: list[PageText] = []
+    last_id = last_title = None
+    with pymupdf.open(str(pdf_path)) as doc:
+        for i, page in enumerate(doc):
+            raw = page.get_text("text") or ""
+            text = clean_text(raw)
+            match = _SECTION_RE.search(raw)
+            if match:
+                last_id, last_title = match.group(1).strip(), match.group(2).strip()
+            pages.append(
+                PageText(
+                    page_number=i + 1,
+                    text=text,
+                    section_id=last_id,
+                    section_title=last_title,
+                    char_count=len(text),
+                )
+            )
+    return pages
+
+
+def chunk_pages(pages: list[PageText], max_chars: int = 1200, overlap: int = 150) -> list[dict]:
+    spec_markers = ("Torque Specifications", "General Specifications", "Fill Capacity", "Fill\nCapacity")
+    chunks: list[dict] = []
+
+    def split_long(text: str) -> list[str]:
+        if len(text) <= max_chars:
+            return [text]
+        pieces = re.split(r"\n(?=\d+\.\n|\n)", text)
+        out, buf = [], ""
+        for piece in pieces:
+            piece = piece.strip()
+            if not piece:
+                continue
+            if buf and len(buf) + len(piece) + 1 > max_chars:
+                out.append(buf.strip())
+                buf = (buf[-overlap:] + "\n" + piece).strip() if overlap else piece
+            else:
+                buf = f"{buf}\n{piece}".strip() if buf else piece
+        if buf:
+            out.append(buf.strip())
+        final = []
+        for c in out:
+            if len(c) <= max_chars:
+                final.append(c)
+                continue
+            start = 0
+            while start < len(c):
+                end = min(start + max_chars, len(c))
+                final.append(c[start:end].strip())
+                if end >= len(c):
+                    break
+                start = max(0, end - overlap)
+        return [c for c in final if c]
+
+    for page in pages:
+        text = page.text.strip()
+        if page.char_count < 40 or not text:
+            continue
+        is_spec = any(m in text for m in spec_markers) and len(text) <= max_chars * 2
+        parts = [text] if is_spec else split_long(text)
+        for idx, part in enumerate(parts):
+            header = f"Page {page.page_number}"
+            if page.section_id:
+                header = f"SECTION {page.section_id}: {page.section_title or ''} | {header}"
+            cid = hashlib.md5(f"{page.page_number}|{idx}|{part[:80]}".encode()).hexdigest()[:12]
+            chunks.append(
+                {
+                    "chunk_id": cid,
+                    "text": f"{header}\n{part}",
+                    "page": page.page_number,
+                    "section_id": page.section_id or "",
+                    "section_title": page.section_title or "",
+                    "chunk_type": "spec" if is_spec else ("page" if len(parts) == 1 else "section"),
+                }
+            )
+    return chunks
+
+
+# ── Offline spec parser (same idea as notebook fallback) ──────────────────────
+_TORQUE_ROW = re.compile(
+    r"(?P<component>[A-Za-z][A-Za-z0-9\-/()., ]{2,80}?)\s*\n\s*"
+    r"(?P<nm>\d+(?:\.\d+)?)\s*\n?\s*"
+    r"(?P<lbft>\d+(?:\.\d+)?|—|-)\s*\n?\s*"
+    r"(?P<lbin>\d+(?:\.\d+)?|—|-)",
+    re.M,
+)
+_INLINE_TORQUE = re.compile(
+    r"(?:Tighten(?:\s+the\s+(?P<fastener>\w+(?:\s+\w+)?))?\s+to|torque(?:d)?\s+to)\s+"
+    r"(?P<nm>\d+(?:\.\d+)?)\s*Nm(?:\s*\((?P<lbft>\d+(?:\.\d+)?)\s*lb-ft\))?",
+    re.I,
+)
+_CAPACITY = re.compile(
+    r"(?P<value>[\d,]+(?:\.\d+)?)\s*(?P<unit>ml|L)\b(?:\s*\((?P<alt>[\d.]+)\s*(?P<alt_unit>pt|qt|ml|L)\))?",
+    re.I,
+)
+_DIM = re.compile(
+    r"(?P<item>[A-Za-z][A-Za-z0-9 /\-]{4,70}?)\s+"
+    r"(?P<value>\d+(?:\.\d+)?)\s*(?P<unit>mm)\s*(?:\((?P<alt>[\d.]+)\s*in\))?",
+    re.I,
+)
+_STOP = {"the", "for", "and", "of", "a", "an", "to", "in", "on", "what", "is", "with",
+         "torque", "value", "spec", "specification", "capacity", "how"}
+_DISTINCTIVE = {
+    "caliper", "shock", "absorber", "hub", "bearing", "stabilizer", "disc", "pad",
+    "axle", "strut", "knuckle", "driveshaft",
+}
+
+
+def _tokens(query: str) -> set[str]:
+    return {t for t in re.findall(r"[a-z0-9]+", query.lower()) if t not in _STOP and len(t) > 2}
+
+
+def _relevant(name: str, content: set[str]) -> bool:
+    if not content:
+        return True
+    lower = name.lower()
+    required = content & _DISTINCTIVE
+    if required and not all(t in lower for t in required):
+        return False
+    for q in ("lower", "upper", "front", "rear"):
+        if q in content and q not in lower:
+            return False
+    roots = {t for t in content if len(t) >= 5} or content
+    return sum(t in lower for t in roots) / len(roots) >= 0.34
+
+
+def extract_with_rules(query: str, docs: list[Document]) -> list[dict[str, Any]]:
+    content = _tokens(query) - {"bolt", "bolts", "nut", "nuts", "screw"}
+    wants_torque = bool(re.search(r"torque|tighten|nm|bolt|nut", query, re.I))
+    wants_fluid = bool(re.search(r"fluid|capacity|fill|ml|liter", query, re.I))
+    wants_dim = bool(re.search(r"thickness|clearance|dimension|minimum|maximum", query, re.I))
+    if not any([wants_torque, wants_fluid, wants_dim]):
+        wants_torque = True
+
+    results, seen = [], set()
+    for doc in docs:
+        text, page, section = doc.page_content, doc.metadata.get("page"), doc.metadata.get("section_id")
+        if wants_torque:
+            for m in _TORQUE_ROW.finditer(text):
+                component = re.sub(r"\s+", " ", m.group("component")).strip(" -")
+                if len(component) < 4 or component.lower() in {"description", "item", "specification"}:
+                    continue
+                if not _relevant(component, content):
+                    continue
+                try:
+                    nm = float(m.group("nm"))
+                except ValueError:
+                    continue
+                if not 0 < nm <= 800:
+                    continue
+                key = (component.lower(), m.group("nm"), page)
+                if key in seen:
+                    continue
+                seen.add(key)
+                notes = []
+                if m.group("lbft") not in {"—", "-", "–"}:
+                    notes.append(f"{m.group('lbft')} lb-ft")
+                if m.group("lbin") not in {"—", "-", "–"}:
+                    notes.append(f"{m.group('lbin')} lb-in")
+                results.append({
+                    "component": component, "spec_type": "Torque",
+                    "value": m.group("nm"), "unit": "Nm",
+                    "notes": "; ".join(notes) or None,
+                    "source_page": page, "section": section,
+                })
+            for m in _INLINE_TORQUE.finditer(text):
+                fastener = (m.group("fastener") or "Fastener").strip()
+                if not _relevant(fastener + " " + text[:200], content) and content:
+                    continue
+                key = (fastener.lower(), m.group("nm"), page)
+                if key in seen:
+                    continue
+                seen.add(key)
+                results.append({
+                    "component": fastener, "spec_type": "Torque",
+                    "value": m.group("nm"), "unit": "Nm",
+                    "notes": f"{m.group('lbft')} lb-ft" if m.group("lbft") else None,
+                    "source_page": page, "section": section,
+                })
+        if wants_fluid:
+            for m in _CAPACITY.finditer(text):
+                value = m.group("value").replace(",", "")
+                try:
+                    if m.group("unit").lower() == "ml" and float(value) < 50:
+                        continue
+                except ValueError:
+                    continue
+                key = ("brake fluid", value, page)
+                if key in seen:
+                    continue
+                seen.add(key)
+                alt = f"{m.group('alt')} {m.group('alt_unit')}" if m.group("alt") else None
+                results.append({
+                    "component": "Brake Fluid", "spec_type": "Fluid Capacity",
+                    "value": value, "unit": m.group("unit"), "notes": alt,
+                    "source_page": page, "section": section,
+                })
+        if wants_dim:
+            for m in _DIM.finditer(text):
+                item = re.sub(r"\s+", " ", m.group("item")).strip(" -")
+                if not _relevant(item, content):
+                    continue
+                key = (item.lower(), m.group("value"), page)
+                if key in seen:
+                    continue
+                seen.add(key)
+                results.append({
+                    "component": item, "spec_type": "Dimension",
+                    "value": m.group("value"), "unit": "mm",
+                    "notes": f"{m.group('alt')} in" if m.group("alt") else None,
+                    "source_page": page, "section": section,
+                })
+    return results
+
+
+# ── LangChain RAG (same LCEL pattern as the notebook) ─────────────────────────
+SPEC_PROMPT = ChatPromptTemplate.from_messages([
+    ("system", """You are an expert automotive technician extracting specifications from a service manual.
+
+Return ONLY a valid JSON array. Each element must have these exact keys:
+  component, spec_type, value, unit, notes, source_page, section
+
+Rules:
+- spec_type ∈ {Torque, Fluid Capacity, Dimension, Part Number, Clearance}
+- Primary unit: Nm (torque), ml (fluid), mm (dimension)
+- Put lb-ft / lb-in / pt in the notes field
+- Do NOT invent values. Output raw JSON only."""),
+    ("user", "Manual excerpts:\n{context}\n\nQuery: {question}\n\nJSON array:"),
+])
+
+
+def format_docs(docs: list) -> str:
+    return "\n\n---\n".join(
+        f"[Page {d.metadata.get('page', '?')} | {d.metadata.get('section_id', '')}]\n{d.page_content}"
+        for d in docs
+    )
+
+
+def parse_spec_json(raw: str) -> list[dict]:
+    raw = re.sub(r"^```(?:json)?", "", raw.strip(), flags=re.M).strip(" `\n")
+    try:
+        data = json.loads(raw)
+        if isinstance(data, list):
+            return data
+        if isinstance(data, dict):
+            for key in ("results", "specifications", "data", "items"):
+                if isinstance(data.get(key), list):
+                    return data[key]
+    except json.JSONDecodeError:
+        match = re.search(r"\[.*?\]", raw, re.S)
+        if match:
+            try:
+                return json.loads(match.group())
+            except Exception:
+                pass
+    return []
+
+
+def get_best_llm():
+    if has_openai:
+        from langchain_openai import ChatOpenAI
+        for name, label in (("gpt-4", "GPT-4"), ("gpt-3.5-turbo", "GPT-3.5-turbo")):
+            try:
+                return ChatOpenAI(model=name, temperature=0, openai_api_key=OPENAI_KEY), label
+            except Exception:
+                continue
+    if has_mistral:
+        try:
+            from langchain_mistralai import ChatMistralAI
+            return ChatMistralAI(model="mistral-small", temperature=0, mistral_api_key=MISTRAL_KEY), "Mistral"
+        except Exception:
+            pass
+    if has_hf:
+        try:
+            from langchain_huggingface import HuggingFaceEndpoint
+            return HuggingFaceEndpoint(
+                repo_id="mistralai/Mistral-7B-Instruct-v0.3",
+                huggingfacehub_api_token=HF_TOKEN,
+                temperature=0.01,
+            ), "Mistral-7B (HF)"
+        except Exception:
+            pass
+    return None, "offline (rule-based)"
+
+
+def embeddings():
+    if has_openai:
+        from langchain_openai import OpenAIEmbeddings
+        return OpenAIEmbeddings(model="text-embedding-3-small", openai_api_key=OPENAI_KEY)
+    return HuggingFaceEmbeddings(
+        model_name="sentence-transformers/all-MiniLM-L6-v2",
+        model_kwargs={"device": "cpu"},
+        encode_kwargs={"normalize_embeddings": True, "batch_size": 32},
+        multi_process=False,
+        show_progress=False,
+    )
+
+
+class SpecRAG:
+    def __init__(self) -> None:
+        self.store: FAISS | None = None
+        self.retriever = None
+        self.llm = None
+        self.llm_label = "offline (rule-based)"
+        self.n_pages = 0
+        self.n_chunks = 0
+        self.pdf_name = ""
+
+    def ingest(self, pdf_path: str | Path) -> str:
+        import faiss as _faiss
+        _faiss.omp_set_num_threads(1)
+
+        pages = [p for p in extract_pages(pdf_path) if p.char_count >= 40]
+        chunks = chunk_pages(pages)
+        docs = [
+            Document(
+                page_content=c["text"],
+                metadata={k: c[k] for k in ("page", "section_id", "section_title", "chunk_type", "chunk_id")},
+            )
+            for c in chunks
+        ]
+        emb = embeddings()
+        store = FAISS.from_documents(docs[:_BATCH], emb)
+        for start in range(_BATCH, len(docs), _BATCH):
+            store.add_documents(docs[start : start + _BATCH])
+
+        self.store = store
+        self.retriever = store.as_retriever(search_type="mmr", search_kwargs={"k": 6, "fetch_k": 20})
+        self.n_pages, self.n_chunks = len(pages), len(docs)
+        self.pdf_name = Path(pdf_path).name
+        llm, label = get_best_llm()
+        self.llm, self.llm_label = llm, label
+        emb_label = "OpenAI" if has_openai else "Sentence-Transformers"
+        return (
+            f"Indexed {self.pdf_name}: {self.n_pages} pages → {self.n_chunks} chunks | "
+            f"embedding={emb_label} | llm={self.llm_label}"
+        )
+
+    def query(self, question: str, top_k: int = 6) -> dict[str, Any]:
+        if self.retriever is None:
+            raise RuntimeError("Index a PDF first.")
+        retriever = self.store.as_retriever(
+            search_type="mmr",
+            search_kwargs={"k": int(top_k), "fetch_k": max(20, int(top_k) * 3)},
+        )
+        docs = retriever.invoke(question)
+        retrieved = [
+            f"#{i} p.{d.metadata.get('page')} ({d.metadata.get('section_id')}) — "
+            f"{d.page_content[:140].replace(chr(10), ' ')}…"
+            for i, d in enumerate(docs, start=1)
+        ]
+        results, method = [], "rules"
+        if self.llm:
+            try:
+                chain = (
+                    {"context": retriever | format_docs, "question": RunnablePassthrough()}
+                    | SPEC_PROMPT | self.llm | StrOutputParser()
+                )
+                results = parse_spec_json(chain.invoke(question))
+                if results:
+                    method = self.llm_label
+            except Exception:
+                results = []
+        if not results:
+            results = extract_with_rules(question, docs)
+            method = "rules"
+        return {"method": method, "results": results, "retrieved": retrieved}
+
+
+pipeline = SpecRAG()
+
+
+def _pdf_path(upload) -> Path | None:
+    if upload is not None:
+        return Path(upload if isinstance(upload, str) else getattr(upload, "name", upload))
+    if _LOCAL_PDF and _LOCAL_PDF.exists():
+        return _LOCAL_PDF
+    return None
+
+
+def index_pdf(upload) -> str:
+    path = _pdf_path(upload)
+    if path is None:
+        return "Upload a workshop-manual PDF, then click Index."
+    return pipeline.ingest(path)
+
+
+def ask(query: str, top_k: int, upload) -> tuple[str, str]:
+    if not query.strip():
+        return "[]", "Enter a query."
+    if pipeline.retriever is None:
+        path = _pdf_path(upload)
+        if path is None:
+            return "[]", "Upload and index a PDF first."
+        index_pdf(upload)
+    out = pipeline.query(query.strip(), top_k=int(top_k))
+    return (
+        json.dumps(out["results"], indent=2, ensure_ascii=False),
+        f"method={out['method']}\n" + "\n".join(out["retrieved"]),
+    )
+
+
+def on_load():
+    if _LOCAL_PDF and _LOCAL_PDF.exists():
+        try:
+            return index_pdf(str(_LOCAL_PDF))
+        except Exception as exc:
+            return f"Ready. Upload a PDF to index. ({exc})"
+    return "Ready. Upload a workshop-manual PDF and click Index."
+
+
+with gr.Blocks(title="Vehicle Spec RAG") as demo:
+    gr.Markdown(
+        "# Vehicle Specification Extraction\n"
+        "Same pipeline as `vehicle_spec_extraction.ipynb`: "
+        "PyMuPDF → chunk → Sentence-Transformers / OpenAI embeddings → FAISS (MMR) → "
+        "GPT / Mistral or rule-based JSON extraction.\n\n"
+        "Upload a service-manual PDF, index it, then ask for torque, fluid, or dimension specs."
+    )
+    pdf = gr.File(label="Workshop manual PDF", file_types=[".pdf"])
+    status = gr.Textbox(label="Index status", interactive=False)
+    index_btn = gr.Button("Index PDF")
+    with gr.Row():
+        query = gr.Textbox(label="Query", placeholder="Torque for brake caliper bolts", scale=4)
+        top_k = gr.Slider(3, 12, value=6, step=1, label="Top-K")
+    ask_btn = gr.Button("Extract specs", variant="primary")
+    results = gr.Code(label="Structured JSON", language="json")
+    debug = gr.Textbox(label="Retrieval debug", lines=8)
+
+    demo.load(on_load, outputs=status)
+    index_btn.click(index_pdf, inputs=pdf, outputs=status)
+    ask_btn.click(ask, inputs=[query, top_k, pdf], outputs=[results, debug])
+    query.submit(ask, inputs=[query, top_k, pdf], outputs=[results, debug])
+    gr.Examples(
+        examples=[
+            ["Torque for brake caliper bolts"],
+            ["Brake fluid capacity"],
+            ["Shock absorber lower nuts torque"],
+            ["Front brake disc minimum thickness"],
+            ["Wheel bearing and wheel hub bolts torque"],
+        ],
+        inputs=query,
+    )
+
+
+if __name__ == "__main__":
+    demo.launch()