Delete app.py

app.py

@@ -1,533 +0,0 @@
-#!/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()