Update app.py

app.py

@@ -1,6 +1,5 @@
-# app.py  – handles images, txt/py, PDFs, any file…
-
-import os, json, time, io, mimetypes
+# app.py — handles images, PDFs, text/code, Excel, audio, etc.
+import os, json, time, io, tempfile, mimetypes
 from functools import lru_cache
 
 import gradio as gr
@@ -8,20 +7,20 @@ import requests
 import pandas as pd
 from openai import OpenAI, RateLimitError, APIError
 from duckduckgo_search import DDGS
-from PyPDF2 import PdfReader   # <- new dependency
+from PyPDF2 import PdfReader
 
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 OPENAI_MODEL    = "gpt-4o-mini"
-TEXT_CHAR_LIMIT = 8_000
-PDF_PAGE_LIMIT  = 3
+TEXT_LIMIT      = 8_000
+PDF_PAGES       = 3
+AUDIO_SIZE_CAP  = 16 * 1024 * 1024       # 16 MB
 
-# ───────────────────────── helpers ─────────────────────────
+# ─────────────── helpers ───────────────
 def duckduckgo_search(query: str, max_results: int = 5) -> str:
-    bullets = []
     with DDGS() as ddgs:
-        for r in ddgs.text(query, max_results=max_results):
-            bullets.append(f"- {r['title']} – {r['href']}")
-    return "\n".join(bullets) or "No results."
+        hits = [f"- {r['title']} – {r['href']}"
+                for r in ddgs.text(query, max_results=max_results)]
+    return "\n".join(hits) or "No results found."
 
 DDG_SCHEMA = {
     "name": "duckduckgo_search",
@@ -36,57 +35,81 @@ DDG_SCHEMA = {
     },
 }
 
-def fetch_text_file(url: str) -> str:
+def download_bytes(url: str, cap: int | None = None) -> bytes:
+    r = requests.get(url, timeout=20)
+    r.raise_for_status()
+    data = r.content
+    if cap and len(data) > cap:
+        raise ValueError("File too large")
+    return data
+
+def extract_text_file(url: str) -> str:
     try:
-        txt = requests.get(url, timeout=15).text
-        return txt[:TEXT_CHAR_LIMIT]
+        txt = download_bytes(url).decode(errors="replace")
+        return txt[:TEXT_LIMIT]
     except Exception as e:
-        return f"[Could not download text file: {e}]"
+        return f"[Could not fetch text file: {e}]"
 
-def fetch_pdf_text(url: str) -> str:
+def extract_pdf(url: str) -> str:
     try:
-        resp = requests.get(url, timeout=20)
-        resp.raise_for_status()
-        reader = PdfReader(io.BytesIO(resp.content))
-        pages = []
-        for i, page in enumerate(reader.pages[:PDF_PAGE_LIMIT]):
-            pages.append(page.extract_text() or "")
-        return ("\n\n".join(pages))[:TEXT_CHAR_LIMIT]
+        reader = PdfReader(io.BytesIO(download_bytes(url)))
+        pages = [reader.pages[i].extract_text() or "" for i in range(min(PDF_PAGES, len(reader.pages)))]
+        return ("\n\n".join(pages))[:TEXT_LIMIT]
     except Exception as e:
         return f"[Could not read PDF: {e}]"
 
-# ─────────────────────────── agent ─────────────────────────
-class GPT4oMiniAgentWithFiles:
-    def __init__(self, retries:int = 3, backoff:float = 2.0):
+def extract_excel(url: str) -> str:
+    try:
+        buf = io.BytesIO(download_bytes(url))
+        df  = pd.read_excel(buf, nrows=15, engine="openpyxl")
+        return df.to_csv(index=False, header=True)[:TEXT_LIMIT]
+    except Exception as e:
+        return f"[Could not read Excel: {e}]"
+
+def transcribe_audio(url: str, client: OpenAI) -> str:
+    try:
+        data = download_bytes(url, cap=AUDIO_SIZE_CAP)
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".audio") as tmp:
+            tmp.write(data); tmp.flush()
+            tr = client.audio.transcriptions.create(model="whisper-1", file=open(tmp.name, "rb"))
+        return tr.text[:2000]
+    except Exception as e:
+        return f"[Could not transcribe audio: {e}]"
+
+# ─────────────── Agent ───────────────
+class GPT4oMiniAgent:
+    def __init__(self, retries=3, backoff=2.0):
         key = os.getenv("OPENAI_API_KEY")
         if not key:
-            raise EnvironmentError("OPENAI_API_KEY missing in Secrets.")
+            raise EnvironmentError("Add OPENAI_API_KEY in Space Secrets")
         self.client, self.retries, self.backoff = OpenAI(api_key=key), retries, backoff
-        self.sys_prompt = (
-            "You are a concise, accurate assistant. "
-            "If certain, answer directly; otherwise call duckduckgo_search."
+        self.system_prompt = (
+            "You are a concise, accurate assistant. If certain, answer directly; "
+            "if not, call duckduckgo_search first."
         )
 
     @lru_cache(maxsize=512)
-    def __call__(self, question:str, file_url:str|None=None) -> str:
-        user_content = [{"type": "text", "text": question}]
+    def __call__(self, question: str, file_url: str | None = None) -> str:
+        user_parts = [{"type": "text", "text": question}]
+
         if file_url:
-            kind = (file_url.split("?")[0].split("#")[0]).lower()
-            ext  = os.path.splitext(kind)[1]
+            ext = os.path.splitext(file_url.split("?")[0].split("#")[0])[1].lower()
             if ext in {".png", ".jpg", ".jpeg", ".gif", ".webp"}:
-                user_content.append({"type":"image_url","image_url":{"url":file_url}})
-            elif ext == ".pdf":
-                text = fetch_pdf_text(file_url)
-                user_content.append({"type":"text","text": f"(PDF extract)\n{text}"})
+                user_parts.append({"type": "image_url", "image_url": {"url": file_url}})
+            elif ext in {".pdf"}:
+                user_parts.append({"type": "text", "text": "(PDF extract)\n" + extract_pdf(file_url)})
+            elif ext in {".xls", ".xlsx"}:
+                user_parts.append({"type": "text", "text": "(Excel preview)\n" + extract_excel(file_url)})
             elif ext in {".txt", ".py", ".md", ".json", ".csv", ".html"}:
-                text = fetch_text_file(file_url)
-                user_content.append({"type":"text","text": f"(File content)\n{text}"})
+                user_parts.append({"type": "text", "text": "(File content)\n" + extract_text_file(file_url)})
+            elif ext in {".mp3", ".wav", ".m4a", ".flac", ".ogg"}:
+                user_parts.append({"type": "text", "text": "(Audio transcript)\n" + transcribe_audio(file_url, self.client)})
             else:
-                user_content.append({"type":"text","text": f"[File available here] {file_url}"})
+                user_parts.append({"type": "text", "text": f"[File available: {file_url}]"} )
 
         msgs = [
-            {"role":"system","content":self.sys_prompt},
-            {"role":"user","content":user_content},
+            {"role": "system", "content": self.system_prompt},
+            {"role": "user",   "content": user_parts},
         ]
 
         resp = self._chat(msgs, tools=[DDG_SCHEMA], tool_choice="auto")
@@ -94,64 +117,56 @@ class GPT4oMiniAgentWithFiles:
         if resp.choices[0].message.tool_calls:
             for call in resp.choices[0].message.tool_calls:
                 args = json.loads(call.function.arguments or "{}")
-                tool_out = duckduckgo_search(**args)
-                msgs.append({
-                    "role":"tool",
-                    "tool_call_id":call.id,
-                    "name":call.function.name,
-                    "content":tool_out,
-                })
+                search_out = duckduckgo_search(**args)
+                msgs.append({"role": "tool", "tool_call_id": call.id, "name": call.function.name, "content": search_out})
             resp = self._chat(msgs)
 
         return resp.choices[0].message.content.strip()
 
     def _chat(self, messages, **kw):
-        for i in range(1, self.retries+1):
+        for i in range(1, self.retries + 1):
             try:
                 return self.client.chat.completions.create(
-                    model=OPENAI_MODEL,
-                    messages=messages,
-                    temperature=0.0,
-                    max_tokens=512,
-                    **kw
+                    model=OPENAI_MODEL, messages=messages,
+                    temperature=0.0, max_tokens=512, **kw
                 )
             except (RateLimitError, APIError):
                 time.sleep(self.backoff * i)
         raise RuntimeError("OpenAI API failed after retries.")
 
-# ────────────────── run + submit ──────────────────
-def run_and_submit_all(profile: gr.OAuthProfile|None):
+# ─────────────── pipeline ───────────────
+def run_and_submit_all(profile: gr.OAuthProfile | None):
     if not profile:
         return "Please log in ↑", None
     username = profile.username
-    agent = GPT4oMiniAgentWithFiles()
-    space_id = os.getenv("SPACE_ID","local")
+    agent = GPT4oMiniAgent()
+    space_id = os.getenv("SPACE_ID", "local")
     agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
 
-    qs = requests.get(f"{DEFAULT_API_URL}/questions", timeout=15).json()
+    questions = requests.get(f"{DEFAULT_API_URL}/questions", timeout=15).json()
 
     rows, answers = [], []
-    for item in qs:
-        qid  = item["task_id"]
-        text = item["question"]
-        file_url = item.get("filename") or item.get("file_url")
-        ans  = agent(text, file_url)
+    for q in questions:
+        qid   = q["task_id"]
+        qtext = q["question"]
+        fileu = q.get("filename") or q.get("file_url")
+        ans   = agent(qtext, fileu)
         answers.append({"task_id": qid, "submitted_answer": ans})
-        rows.append({"Task ID": qid, "Question": text, "File": file_url or "", "Answer": ans})
+        rows.append({"Task ID": qid, "Question": qtext, "File": fileu or "", "Answer": ans})
 
     payload = {"username": username, "agent_code": agent_code, "answers": answers}
     res = requests.post(f"{DEFAULT_API_URL}/submit", json=payload, timeout=60).json()
-    status = f"Score {res['score']} % ({res['correct_count']}/{res['total_attempted']})"
+    status = f"Score {res['score']} %  ({res['correct_count']}/{res['total_attempted']})"
     return status, pd.DataFrame(rows)
 
-# ────────────────────── UI ───────────────────────
+# ─────────────── UI ───────────────
 with gr.Blocks() as demo:
-    gr.Markdown("# Unit-4 Agent – handles images, text/code files & PDFs")
+    gr.Markdown("# Unit-4 Agent – images, PDFs, Excel, audio, text, etc.")
     gr.LoginButton()
-    btn = gr.Button("Run Evaluation & Submit All Answers")
-    status = gr.Textbox(label="Status", interactive=False)
-    table  = gr.DataFrame(label="Log", wrap=True)
-    btn.click(run_and_submit_all, outputs=[status, table])
+    run = gr.Button("Run Evaluation & Submit All Answers")
+    out_status = gr.Textbox(label="Status", interactive=False)
+    out_table  = gr.DataFrame(label="Log", wrap=True)
+    run.click(run_and_submit_all, outputs=[out_status, out_table])
 
 if __name__ == "__main__":
     demo.launch(debug=True, share=False)