Spaces:
Running
Running
| import streamlit as st | |
| import io, re, json, html | |
| from typing import List | |
| import numpy as np | |
| import math | |
| # Optional Gemini client | |
| try: | |
| from google import genai | |
| except Exception: | |
| genai = None | |
| # Optional transformers (Hugging Face pipeline) | |
| try: | |
| from transformers import pipeline, AutoTokenizer, AutoModelForCausalLM | |
| TRANSFORMERS_AVAILABLE = True | |
| except Exception: | |
| TRANSFORMERS_AVAILABLE = False | |
| pipeline = None | |
| AutoTokenizer = None | |
| AutoModelForCausalLM = None | |
| # PDF extraction helpers | |
| use_pymupdf = False | |
| try: | |
| import fitz | |
| use_pymupdf = True | |
| except Exception: | |
| try: | |
| from PyPDF2 import PdfReader | |
| except Exception: | |
| PdfReader = None | |
| # Sentence-transformers embedder | |
| try: | |
| from sentence_transformers import SentenceTransformer | |
| except Exception: | |
| SentenceTransformer = None | |
| st.set_page_config(page_title="StudyMate — PDF Q&A", layout="wide") | |
| st.title("📘 StudyMate — PDF Q&A") | |
| # ---------------- Sidebar ---------------- | |
| st.sidebar.header("Config") | |
| if "gemini_api_key" not in st.session_state: | |
| st.session_state.gemini_api_key = "" | |
| if "gemini_model" not in st.session_state: | |
| st.session_state.gemini_model = "models/gemini-2.5-pro" | |
| # Add HF pipeline settings | |
| if "use_hf_pipeline" not in st.session_state: | |
| st.session_state.use_hf_pipeline = False | |
| if "hf_model_id" not in st.session_state: | |
| st.session_state.hf_model_id = "ibm-granite/granite-3.3-2b-instruct" | |
| if "hf_api_token" not in st.session_state: | |
| st.session_state.hf_api_token = "" | |
| st.session_state.gemini_api_key = st.sidebar.text_input("Gemini API Key", value=st.session_state.gemini_api_key, type="password") | |
| st.session_state.gemini_model = st.sidebar.text_input("Gemini Model ID", value=st.session_state.gemini_model) | |
| st.sidebar.markdown("---") | |
| st.session_state.use_hf_pipeline = st.sidebar.checkbox("Enable Hugging Face pipeline (IBM Granite)", value=st.session_state.use_hf_pipeline) | |
| st.session_state.hf_model_id = st.sidebar.text_input("HF Model ID", value=st.session_state.hf_model_id) | |
| st.session_state.hf_api_token = st.sidebar.text_input("HF API token (optional)", value=st.session_state.hf_api_token, type="password") | |
| chunk_size = st.sidebar.number_input("Chunk size (chars)", 200, 5000, 1000, 100) | |
| chunk_overlap = st.sidebar.number_input("Chunk overlap (chars)", 0, 500, 200, 50) | |
| top_k = st.sidebar.slider("Top-k retrieval", 1, 8, 3) | |
| use_llm = st.sidebar.checkbox("Use Gemini/HF for generation", value=True) | |
| # ---------------- Session state ---------------- | |
| if "docs" not in st.session_state: st.session_state.docs = [] | |
| if "chunks" not in st.session_state: st.session_state.chunks = [] | |
| if "chunk_sources" not in st.session_state: st.session_state.chunk_sources = [] | |
| if "embeddings" not in st.session_state: st.session_state.embeddings = np.zeros((0, 384), dtype=np.float32) | |
| if "quiz_data" not in st.session_state: st.session_state.quiz_data = None | |
| if "flashcards" not in st.session_state: st.session_state.flashcards = None | |
| if "last_gemini_raw" not in st.session_state: st.session_state.last_gemini_raw = None | |
| if "embedder_loaded" not in st.session_state: st.session_state.embedder_loaded = False | |
| # ---------------- Helpers ---------------- | |
| def load_embedder(name: str = "all-MiniLM-L6-v2"): | |
| if SentenceTransformer is None: | |
| raise RuntimeError("sentence_transformers not installed.") | |
| return SentenceTransformer(name) | |
| def get_embedder(): | |
| if not st.session_state.embedder_loaded: | |
| st.session_state.embedder = load_embedder() | |
| st.session_state.embedder_loaded = True | |
| return st.session_state.embedder | |
| def extract_text_from_pdf_stream(file_stream: io.BytesIO) -> str: | |
| file_stream.seek(0) | |
| if use_pymupdf: | |
| try: | |
| doc = fitz.open(stream=file_stream.read(), filetype="pdf") | |
| pages = [p.get_text("text") or "" for p in doc] | |
| return "\n".join(pages) | |
| except Exception: | |
| file_stream.seek(0) | |
| if PdfReader is not None: | |
| try: | |
| file_stream.seek(0) | |
| reader = PdfReader(file_stream) | |
| pages = [] | |
| for p in reader.pages: | |
| try: | |
| pages.append(p.extract_text() or "") | |
| except Exception: | |
| pages.append("") | |
| return "\n".join(pages) | |
| except Exception: | |
| return "" | |
| return "" | |
| def clean_text(s: str) -> str: | |
| return re.sub(r"\s+", " ", (s or "")).strip() | |
| def chunk_text(text: str, size: int, overlap: int) -> List[str]: | |
| text = text.strip() | |
| if not text: | |
| return [] | |
| chunks = [] | |
| start = 0 | |
| L = len(text) | |
| while start < L: | |
| end = start + size | |
| chunks.append(text[start:end].strip()) | |
| start = end - overlap | |
| if start < 0: | |
| start = end | |
| rem = text[start:].strip() | |
| if rem: chunks.append(rem) | |
| return [c for c in chunks if c] | |
| def embed_texts(embedder, texts): | |
| if not texts: return np.zeros((0, embedder.get_sentence_embedding_dimension()), dtype=np.float32) | |
| embs = embedder.encode(texts, convert_to_numpy=True, show_progress_bar=False) | |
| norms = np.linalg.norm(embs, axis=1, keepdims=True) | |
| norms[norms==0] = 1.0 | |
| embs = embs / norms | |
| return embs.astype(np.float32) | |
| def similarity_search(query_emb, all_embs, top_k=3): | |
| if all_embs.shape[0] == 0: return [] | |
| q_norm = np.linalg.norm(query_emb) | |
| if q_norm == 0: return [] | |
| q = query_emb / q_norm | |
| sims = np.dot(all_embs, q) | |
| if top_k < len(sims): | |
| top_k_indices = np.argpartition(-sims, top_k)[:top_k] | |
| sorted_top_k = top_k_indices[np.argsort(-sims[top_k_indices])] | |
| idxs = sorted_top_k | |
| else: | |
| idxs = np.argsort(-sims)[:top_k] | |
| return [(int(i), float(sims[i])) for i in idxs] | |
| def get_context_from_hits(hits): | |
| if not hits: return "" | |
| ctx_chunks = [] | |
| for idx, score in hits: | |
| src = st.session_state.chunk_sources[idx] if idx < len(st.session_state.chunk_sources) else "unknown" | |
| chunk_text = st.session_state.chunks[idx] | |
| ctx_chunks.append(f"CHUNK (source: {src}, score: {score:.4f})\n{chunk_text}") | |
| return "\n\n---\n\n".join(ctx_chunks) | |
| # ---------------- Gemini wrapper ---------------- | |
| def make_gemini_client(): | |
| if genai is None: | |
| raise RuntimeError("google-genai package not installed.") | |
| key = st.session_state.get("gemini_api_key") | |
| if not key: | |
| raise RuntimeError("Gemini API key not set.") | |
| return genai.Client(api_key=key) | |
| def call_gemini_with_context(prompt: str, model_id: str, context: str, max_output_tokens: int = 512): | |
| client = make_gemini_client() | |
| MAX_CTX = 20000 | |
| ctx = context if len(context) <= MAX_CTX else context[:MAX_CTX] | |
| full_prompt = f"{prompt}\n\nCONTEXT:\n{ctx}\n\nAnswer:" | |
| variants = [ | |
| lambda: client.models.generate_content(model=model_id, contents=full_prompt, max_output_tokens=max_output_tokens), | |
| lambda: client.models.generate_content(model=model_id, contents=full_prompt, max_tokens=max_output_tokens), | |
| lambda: client.models.generate_content(model=model_id, inputs=full_prompt, max_output_tokens=max_output_tokens), | |
| lambda: client.models.generate_content(model=model_id, contents=full_prompt), | |
| ] | |
| last_exc = None | |
| resp = None | |
| for fn in variants: | |
| try: | |
| resp = fn() | |
| break | |
| except Exception as e: | |
| last_exc = e | |
| emsg = str(e).lower() | |
| if any(x in emsg for x in ("quota", "unauthorized", "permission", "forbidden", "not found", "403", "401")): | |
| raise | |
| continue | |
| if resp is None: | |
| raise RuntimeError(f"Gemini call failed. Last error: {last_exc}") | |
| out_text = "" | |
| try: | |
| if hasattr(resp, "candidates") and resp.candidates: | |
| cand = resp.candidates[0] | |
| if hasattr(cand, "content") and cand.content: | |
| parts = getattr(cand.content, "parts", None) | |
| if parts: | |
| texts = [] | |
| for p in parts: | |
| if hasattr(p, "text") and p.text: | |
| texts.append(str(p.text)) | |
| elif isinstance(p, dict) and p.get("text"): | |
| texts.append(str(p.get("text"))) | |
| if texts: out_text = "\n".join(texts) | |
| except Exception: | |
| out_text = "" | |
| if not out_text: | |
| try: | |
| out_text = getattr(resp, "output_text", "") or out_text | |
| except Exception: | |
| out_text = out_text | |
| if not out_text: | |
| try: | |
| d = resp.to_dict() if hasattr(resp, "to_dict") else (resp if isinstance(resp, dict) else None) | |
| if isinstance(d, dict): | |
| if "candidates" in d and isinstance(d["candidates"], list) and d["candidates"]: | |
| first = d["candidates"][0] | |
| if isinstance(first, dict): | |
| cont = first.get("content") or first.get("message") or {} | |
| if isinstance(cont, dict): | |
| parts = cont.get("parts") or [] | |
| texts = [] | |
| for p in parts: | |
| if isinstance(p, dict) and p.get("text"): | |
| texts.append(p.get("text")) | |
| if texts: | |
| out_text = "\n".join(texts) | |
| for k in ("generated_text", "output_text", "text", "content", "result"): | |
| if k in d and isinstance(d[k], str) and d[k].strip(): | |
| out_text = d[k]; break | |
| except Exception: | |
| pass | |
| if not out_text: | |
| try: | |
| out_text = str(resp) | |
| except Exception: | |
| out_text = "" | |
| # store raw only for debugging; do NOT show by default | |
| st.session_state.last_gemini_raw = out_text | |
| return out_text | |
| # ---------------- Hugging Face pipeline wrapper ---------------- | |
| def init_hf_pipeline(model_id: str): | |
| if not TRANSFORMERS_AVAILABLE: | |
| raise RuntimeError("transformers package not installed.") | |
| # allow pass-through of token via environment or huggingface-cli; this snippet does not set env | |
| # We'll load model/tokenizer using Auto classes for safety; for very large models you may want to use from_pretrained with device_map | |
| try: | |
| # Use the pipeline abstraction for text generation | |
| gen_pipe = pipeline("text-generation", model=model_id, trust_remote_code=True) | |
| return gen_pipe | |
| except Exception as e: | |
| # Try a lighter loader fallback | |
| try: | |
| tokenizer = AutoTokenizer.from_pretrained(model_id) | |
| model = AutoModelForCausalLM.from_pretrained(model_id) | |
| gen_pipe = pipeline("text-generation", model=model, tokenizer=tokenizer) | |
| return gen_pipe | |
| except Exception as e2: | |
| raise RuntimeError(f"Failed to initialize HF pipeline for {model_id}: {e}; fallback error: {e2}") | |
| def call_hf_pipeline(prompt: str, model_id: str, max_new_tokens: int = 512): | |
| """ | |
| Calls HF pipeline. Returns generated text string. | |
| """ | |
| if not TRANSFORMERS_AVAILABLE: | |
| raise RuntimeError("transformers package not installed.") | |
| # initialize | |
| try: | |
| pipe = init_hf_pipeline(model_id) | |
| except Exception as e: | |
| raise RuntimeError(f"Failed to init HF pipeline: {e}") | |
| # The user's example used a 'messages' structure. Pipeline expects text; we pass prompt text. | |
| try: | |
| out = pipe(prompt, max_new_tokens=max_new_tokens, do_sample=False) | |
| # pipeline returns list of dicts [{'generated_text': '...'}] | |
| if isinstance(out, list) and out: | |
| text = out[0].get("generated_text") or out[0].get("text") or str(out[0]) | |
| return text | |
| return str(out) | |
| except Exception as e: | |
| raise RuntimeError(f"HF pipeline generation failed: {e}") | |
| # ---------------- JSON/HTML cleaning helpers ---------------- | |
| def extract_json_substring(text: str): | |
| if not text: | |
| return None | |
| t = re.sub(r"^```(?:json)?\s*|\s*```$", "", text.strip(), flags=re.IGNORECASE) | |
| fenced = re.findall(r"```(?:json)?\s*(\{.*?\}|\[.*?\])\s*```", text, flags=re.DOTALL | re.IGNORECASE) | |
| if fenced: | |
| return fenced[0] | |
| arr = re.search(r"(\[.*\])", t, flags=re.DOTALL) | |
| if arr: return arr.group(1) | |
| obj = re.search(r"(\{.*\})", t, flags=re.DOTALL) | |
| if obj: return obj.group(1) | |
| return None | |
| def clean_json_like_output(raw: str): | |
| if not raw: return raw | |
| s = raw.strip() | |
| s = re.sub(r"^```(?:json)?\s*", "", s, flags=re.IGNORECASE) | |
| s = re.sub(r"```$", "", s, flags=re.IGNORECASE).strip() | |
| js = extract_json_substring(s) | |
| if js: return js | |
| return s | |
| def html_term_def_pairs_from_text(html_text: str): | |
| out = [] | |
| if not html_text: | |
| return out | |
| term_classes = r'(?:card-term|flashcard-term|term|front-term|term-title)' | |
| def_classes = r'(?:card-def|flashcard-back-content|definition|def|desc|front-def|back-def)' | |
| term_pat = rf'<[^>]*class=["\']{term_classes}["\'][^>]*>(.*?)</' | |
| def_pat = rf'<[^>]*class=["\']{def_classes}["\'][^>]*>(.*?)</' | |
| term_matches = re.findall(term_pat, html_text, flags=re.IGNORECASE | re.DOTALL) | |
| def_matches = re.findall(def_pat, html_text, flags=re.IGNORECASE | re.DOTALL) | |
| if term_matches and def_matches and len(term_matches) == len(def_matches): | |
| for t, d in zip(term_matches, def_matches): | |
| t_clean = re.sub(r'<[^>]+>', '', t).strip() | |
| d_clean = re.sub(r'<[^>]+>', '', d).strip() | |
| out.append({"term": t_clean, "definition": d_clean}) | |
| return out | |
| divs = re.findall(r'<div[^>]*>(.*?)</div>', html_text, flags=re.DOTALL | re.IGNORECASE) | |
| divs = [re.sub(r'<[^>]+>', '', d).strip() for d in divs if d.strip()] | |
| for i in range(0, len(divs)-1, 2): | |
| t = divs[i]; d = divs[i+1] | |
| if len(t) < 100 and len(d) < 2000: | |
| out.append({"term": t, "definition": d}) | |
| return out | |
| # ---------------- Upload & Index ---------------- | |
| st.subheader("1) Upload PDFs & Index") | |
| uploaded_files = st.file_uploader("Upload PDF files", type=["pdf"], accept_multiple_files=True) | |
| col1, col2 = st.columns([1,3]) | |
| with col1: | |
| if st.button("Process & Index Files"): | |
| if not uploaded_files: | |
| st.warning("Upload at least one PDF.") | |
| else: | |
| st.session_state.docs = [] | |
| st.session_state.chunks = [] | |
| st.session_state.chunk_sources = [] | |
| embedder = get_embedder() | |
| all_texts = [] | |
| with st.spinner("Extracting text..."): | |
| for f in uploaded_files: | |
| try: | |
| bio = io.BytesIO(f.read()) | |
| text = extract_text_from_pdf_stream(bio) | |
| text = clean_text(text) | |
| st.session_state.docs.append({"name": f.name, "text": text}) | |
| all_texts.append((f.name, text)) | |
| except Exception as e: | |
| st.error(f"Failed to read {f.name}: {e}") | |
| for name, text in all_texts: | |
| chs = chunk_text(text, size=chunk_size, overlap=chunk_overlap) | |
| for c in chs: | |
| st.session_state.chunks.append(c) | |
| st.session_state.chunk_sources.append(name) | |
| if st.session_state.chunks: | |
| with st.spinner("Creating embeddings..."): | |
| embs = embed_texts(embedder, st.session_state.chunks) | |
| st.session_state.embeddings = embs | |
| st.success(f"Indexed {len(st.session_state.chunks)} chunks from {len(all_texts)} files.") | |
| else: | |
| st.warning("No text extracted from the uploaded files.") | |
| st.session_state.quiz_data = None | |
| st.session_state.flashcards = None | |
| st.session_state.last_gemini_raw = None | |
| with col2: | |
| if st.session_state.docs: | |
| with st.expander("Indexed documents"): | |
| for d in st.session_state.docs: | |
| st.write(f"- **{d['name']}** — {len(d['text'])} chars") | |
| st.write(f"**Total chunks:** {len(st.session_state.chunks)}") | |
| # ---------------- Tools ---------------- | |
| st.markdown("---") | |
| st.subheader("2) Tools") | |
| def build_context_for_action(query_text: str = None): | |
| if st.session_state.chunks and st.session_state.embeddings.shape[0] > 0 and query_text: | |
| try: | |
| embedder = get_embedder() | |
| q_emb = embedder.encode(query_text, convert_to_numpy=True) | |
| hits = similarity_search(q_emb, st.session_state.embeddings, top_k=top_k) | |
| ctx = get_context_from_hits(hits) | |
| if ctx: | |
| return ctx | |
| except Exception: | |
| pass | |
| return "\n\n".join([d["text"] for d in st.session_state.docs]) if st.session_state.docs else "" | |
| def call_tool(prompt: str, question: str = None, max_out=512): | |
| """ | |
| Main LLM caller. Priority: | |
| - Gemini (if use_llm and gemini key present) | |
| - Hugging Face pipeline (if enabled) | |
| - returns helpful error messages if neither available | |
| """ | |
| if not use_llm: | |
| return "LLM disabled in sidebar." | |
| context = build_context_for_action(question or prompt) | |
| if not context: | |
| return "No document text available. Upload & index PDFs first." | |
| # 1) Try Gemini if user has set key & use_llm True | |
| if st.session_state.get("gemini_api_key"): | |
| try: | |
| raw = call_gemini_with_context(prompt, st.session_state.gemini_model, context, max_output_tokens=max_out) | |
| return raw | |
| except Exception as e: | |
| # store the error in last_gemini_raw for debugging | |
| st.session_state.last_gemini_raw = f"Gemini failed: {e}" | |
| # fall through to HF pipeline if enabled | |
| # 2) Try HF pipeline (if enabled) | |
| if st.session_state.get("use_hf_pipeline"): | |
| hf_model = st.session_state.get("hf_model_id") or st.session_state.hf_model_id | |
| try: | |
| # build prompt: we include context so the HF model sees it (be mindful of token limits) | |
| MAX_CTX = 20000 | |
| ctx = context if len(context) <= MAX_CTX else context[:MAX_CTX] | |
| full_prompt = f"{prompt}\n\nCONTEXT:\n{ctx}\n\nAnswer:" | |
| raw = call_hf_pipeline(full_prompt, hf_model, max_new_tokens=max_out) | |
| return raw | |
| except Exception as e: | |
| return f"HF pipeline failed: {e}" | |
| # Nothing available | |
| return "No LLM available: set Gemini API key in sidebar or enable HF pipeline." | |
| # Summary | |
| if st.button("Generate Summary"): | |
| PROMPT = "Summarize the context in 5 concise bullet points. Use only the context." | |
| raw = call_tool(PROMPT, question="summary", max_out=400) | |
| st.subheader("Summary") | |
| st.write(raw or "No output") | |
| # Q&A | |
| query = st.text_input("Ask a question about your PDFs:") | |
| if st.button("Get Answer"): | |
| if not query.strip(): | |
| st.warning("Enter a question.") | |
| else: | |
| PROMPT = f"Answer the question using ONLY the context. QUESTION: {query}" | |
| raw = call_tool(PROMPT, question=query, max_out=512) | |
| st.subheader("Answer") | |
| st.write(raw or "No output") | |
| # Quiz generation | |
| if st.button("Generate Quiz (5 MCQs)"): | |
| PROMPT = ( | |
| "Generate exactly 5 MCQs based ONLY on the context. " | |
| "Return JSON array only: [{\"question\":\"...\",\"options\":[\"A\",\"B\",\"C\",\"D\"],\"answer\":\"A\"}, ...]" | |
| ) | |
| raw = call_tool(PROMPT, question="quiz", max_out=800) | |
| cleaned = clean_json_like_output(raw or "") | |
| parsed = None | |
| try: | |
| parsed = json.loads(cleaned) | |
| except Exception: | |
| parsed = None | |
| if isinstance(parsed, list): | |
| norm = [] | |
| for it in parsed: | |
| qtext = it.get("question") if isinstance(it.get("question"), str) else str(it.get("question")) | |
| opts = list(it.get("options") or []) | |
| if len(opts) < 4: opts = opts + ["(option)"]*(4-len(opts)) | |
| norm.append({"question": qtext, "options": opts, "answer": str(it.get("answer") or "").strip()}) | |
| st.session_state.quiz_data = norm | |
| st.success("Quiz parsed.") | |
| st.session_state.last_gemini_raw = None | |
| else: | |
| st.warning("Could not parse quiz output. Use 'Inspect raw output' to debug.") | |
| if st.session_state.last_gemini_raw: | |
| with st.expander("Inspect raw Gemini output (parsing failed)"): | |
| st.code(st.session_state.last_gemini_raw) | |
| # Render interactive quiz | |
| if st.session_state.get("quiz_data"): | |
| st.markdown("### 🧠 Quiz (interactive)") | |
| for i, q in enumerate(st.session_state.quiz_data, start=1): | |
| st.markdown(f"**Q{i}. {q.get('question','')}**") | |
| options = q.get("options", []) | |
| choice = st.radio(f"Select for Q{i}:", options, key=f"quiz_q{i}") | |
| if st.button(f"Check Q{i}", key=f"check_{i}"): | |
| correct_field = q.get("answer","") | |
| correct_text = None | |
| if isinstance(correct_field, str) and len(correct_field.strip()) == 1 and correct_field.strip().upper() in "ABCD": | |
| idx = ord(correct_field.strip().upper()) - 65 | |
| if 0 <= idx < len(options): correct_text = options[idx] | |
| if correct_text is None: | |
| if correct_field in options: correct_text = correct_field | |
| if correct_text is None: | |
| cf = str(correct_field).strip() | |
| if cf and cf[0].upper() in "ABCD": | |
| idx = ord(cf[0].upper()) - 65 | |
| if 0 <= idx < len(options): correct_text = options[idx] | |
| if correct_text is None: correct_text = str(correct_field) | |
| if choice == correct_text: | |
| st.success("✅ Correct") | |
| else: | |
| # show the option text if we have it, else show raw answer | |
| show_ans = correct_text if correct_text else str(correct_field) | |
| st.error(f"❌ Incorrect — correct: {show_ans}") | |
| st.markdown("---") | |
| # Flashcards generation & rendering (kept your block, with HF/Gemini caller preserved) | |
| if st.button("Generate Flashcards (7)"): | |
| PROMPT = "Generate 7 flashcards in pure JSON: [{\"term\":\"T\",\"definition\":\"D\"}, ...]. Return JSON only." | |
| raw = call_tool(PROMPT, question="flashcards", max_out=800) | |
| cleaned = clean_json_like_output(raw or "") | |
| parsed_cards = [] | |
| parsed_ok = False | |
| # Try JSON parse | |
| try: | |
| j = json.loads(cleaned) | |
| if isinstance(j, list) and j: | |
| for item in j: | |
| t = item.get("term") if isinstance(item.get("term"), str) else str(item.get("term") or "") | |
| d = item.get("definition") if isinstance(item.get("definition"), str) else str(item.get("definition") or "") | |
| parsed_cards.append({"term": t.strip(), "definition": d.strip()}) | |
| parsed_ok = True | |
| except Exception: | |
| parsed_ok = False | |
| # Fallback: attempt smart HTML extraction | |
| if not parsed_ok and raw: | |
| pairs = html_term_def_pairs_from_text(raw) | |
| if pairs: | |
| parsed_cards = pairs | |
| parsed_ok = True | |
| # Fallback: try extracting JSON-like objects in text | |
| if not parsed_ok and raw: | |
| objs = re.findall(r'\{[^{}]{10,}\}', raw, flags=re.DOTALL) | |
| for o in objs: | |
| try: | |
| od = json.loads(o) | |
| if isinstance(od, dict) and (od.get("term") or od.get("definition")): | |
| t = od.get("term") or od.get("question") or "" | |
| d = od.get("definition") or od.get("answer") or "" | |
| parsed_cards.append({"term": str(t).strip(), "definition": str(d).strip()}) | |
| except Exception: | |
| continue | |
| if parsed_cards: | |
| parsed_ok = True | |
| if parsed_ok and parsed_cards: | |
| # dedupe & normalize | |
| final = [] | |
| seen = set() | |
| for c in parsed_cards: | |
| term = (c.get("term") or "").strip() | |
| definition = (c.get("definition") or "").strip() | |
| if not term and not definition: | |
| continue | |
| key = (term.lower(), definition[:80]) | |
| if key in seen: | |
| continue | |
| seen.add(key) | |
| final.append({"term": term or "Untitled", "definition": definition or ""}) | |
| st.session_state.flashcards = final[:20] | |
| st.success(f"Flashcards parsed: {len(st.session_state.flashcards)}") | |
| st.session_state.last_gemini_raw = None | |
| else: | |
| st.session_state.flashcards = None | |
| st.warning("Could not parse flashcards automatically. Inspect raw output to debug.") | |
| if st.session_state.last_gemini_raw: | |
| with st.expander("Inspect raw Gemini output (parsing failed)"): | |
| st.code(st.session_state.last_gemini_raw) | |
| # Render flashcards (colored basic view). This renders HTML via st.components.v1.html for reliability | |
| if st.session_state.get("flashcards"): | |
| st.markdown("### 💡 Flashcards — Colored View") | |
| # CSS — make cards size consistent and content fill inside | |
| card_css = ( | |
| "<style>" | |
| ".cards-grid { display:flex; flex-wrap:wrap; gap:18px; align-items:flex-start; margin-bottom:18px; }" | |
| ".basic-card { width: 320px; min-height: 150px; border-radius: 12px; padding: 14px; box-sizing: border-box; display:flex; flex-direction:column; color: #fff; box-shadow: 0 6px 18px rgba(0,0,0,0.25); overflow:hidden; }" | |
| ".card-term { font-weight:800; font-size:16px; margin-bottom:10px; text-align:center; }" | |
| ".card-def { flex:1 1 auto; overflow:auto; font-size:14px; line-height:1.45; white-space:normal; word-wrap:break-word; }" | |
| ".card-color-0 { background: linear-gradient(135deg, #6EE7B7 0%, #3B82F6 100%); color:#042c2e; }" | |
| ".card-color-1 { background: linear-gradient(135deg, #FDE68A 0%, #F59E0B 100%); color:#2b2b0b; }" | |
| ".card-color-2 { background: linear-gradient(135deg, #A78BFA 0%, #7C3AED 100%); color:#ffffff; }" | |
| ".card-color-3 { background: linear-gradient(135deg, #FDB4BF 0%, #F472B6 100%); color:#2b0b10; }" | |
| ".card-color-4 { background: linear-gradient(135deg, #93C5FD 0%, #2563EB 100%); color:#04203b; }" | |
| ".card-color-5 { background: linear-gradient(135deg, #C7F9CC 0%, #10B981 100%); color:#05321f; }" | |
| "@media (max-width:900px) { .basic-card { width: 95%; } }" | |
| "</style>" | |
| ) | |
| # Build HTML for all cards (escape text to avoid raw HTML injection) | |
| card_html = card_css + '<div class="cards-grid">' | |
| palette_count = 6 | |
| for idx, card in enumerate(st.session_state.flashcards): | |
| cls = f"card-color-{idx % palette_count}" | |
| # escape values so nothing renders raw | |
| term = html.escape(card.get("term", "")) # safe | |
| definition = html.escape(card.get("definition", "")).replace("\n", "<br/>") | |
| # ensure small content fits — definition inside .card-def will scroll if long | |
| card_html += f""" | |
| <div class="basic-card {cls}"> | |
| <div class="card-term">{term}</div> | |
| <div class="card-def">{definition}</div> | |
| </div> | |
| """ | |
| card_html += "</div>" | |
| # Render the card HTML using the HTML component so it's interpreted as HTML reliably. | |
| # We keep escaping of term/definition above (html.escape) to avoid injection. | |
| try: | |
| # estimate height: base 220 + 180 per row (approx 3 cards per row) | |
| cards_per_row = 3 | |
| rows = math.ceil(len(st.session_state.flashcards) / cards_per_row) | |
| height = min(1200, 220 + rows * 180) | |
| except Exception: | |
| height = 600 | |
| st.components.v1.html(card_html, height=height, scrolling=True) | |
| st.caption("Tip: long definitions will scroll inside the card. Use 'Generate Flashcards' again to refresh.") | |
| # ----------------- OCR integration (APPENDED; does not change anything above) ----------------- | |
| # The user asked not to modify any code above; OCR helpers and UI are added below. | |
| def ensure_tesseract(): | |
| """Install or check Tesseract at runtime if missing.""" | |
| try: | |
| import subprocess | |
| subprocess.run(["tesseract", "--version"], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | |
| # don't write sidebar messages during import-time in some runtimes; safe to attempt | |
| except Exception: | |
| try: | |
| import subprocess | |
| subprocess.run(["apt-get", "update"], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | |
| subprocess.run([ | |
| "apt-get", "install", "-y", | |
| "tesseract-ocr", "libtesseract-dev", | |
| "tesseract-ocr-eng", "tesseract-ocr-ara" | |
| ], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE) | |
| except Exception: | |
| # best-effort; if installation fails, OCR later will error and show message | |
| pass | |
| # OCR UI | |
| st.markdown("---") | |
| st.subheader("📷 Image to Text & Highlights (OCR)") | |
| uploaded_images = st.file_uploader( | |
| "Upload one or more images (JPG, PNG)", | |
| type=["png", "jpg", "jpeg"], | |
| accept_multiple_files=True, | |
| key="ocr_images_uploader" | |
| ) | |
| if st.button("Run OCR & Summarize (Images)"): | |
| if not uploaded_images: | |
| st.warning("Please upload at least one image.") | |
| else: | |
| # ensure tesseract binary present (best-effort) | |
| ensure_tesseract() | |
| try: | |
| import pytesseract | |
| from PIL import Image | |
| except Exception: | |
| st.error("OCR dependencies missing. Install pytesseract and pillow: pip install pytesseract pillow") | |
| else: | |
| ocr_texts = [] | |
| with st.spinner("Running OCR on uploaded images..."): | |
| for img_file in uploaded_images: | |
| try: | |
| img = Image.open(img_file) | |
| text = pytesseract.image_to_string(img) | |
| clean_txt = re.sub(r"\s+", " ", text.strip()) | |
| ocr_texts.append((img_file.name, clean_txt)) | |
| except Exception as e: | |
| st.error(f"OCR failed for {img_file.name}: {e}") | |
| if not ocr_texts: | |
| st.warning("No text extracted from any image.") | |
| else: | |
| combined_text = "\n\n".join(t for _, t in ocr_texts if t) | |
| # Optional Gemini summarization (if user enabled & key present) | |
| if use_llm and st.session_state.get("gemini_api_key") and genai: | |
| try: | |
| client = genai.Client(api_key=st.session_state.gemini_api_key) | |
| prompt = f"Summarize the following extracted text into 5 concise bullet points:\n\n{combined_text}" | |
| resp = client.models.generate_content(model=st.session_state.gemini_model, contents=prompt) | |
| # resp parsing best-effort | |
| summary_text = getattr(resp, 'text', None) or str(resp) | |
| st.markdown("### ✨ Gemini Highlights") | |
| st.markdown(summary_text) | |
| except Exception as e: | |
| st.error(f"Gemini summarization failed: {e}") | |
| else: | |
| # Local simple summarization: pick short sentences | |
| sentences = re.split(r'(?<=[.!?])\s+', combined_text) | |
| bullet_points = [] | |
| for s in sentences: | |
| s = s.strip() | |
| if len(s) > 15 and len(s) < 180: | |
| bullet_points.append("• " + s) | |
| if len(bullet_points) >= 6: | |
| break | |
| if not bullet_points: | |
| bullet_points = ["No clear textual content detected."] | |
| st.success("OCR and summarization complete ✅") | |
| st.subheader("Extracted Highlights:") | |
| st.markdown("\n".join(bullet_points)) | |
| with st.expander("Show Full Extracted OCR Text"): | |
| for name, txt in ocr_texts: | |
| st.markdown(f"*{name}*") | |
| st.text_area("Text", value=txt, height=200) | |
| # End of appended OCR integration | |