Update app.py

app.py

@@ -1,272 +1,135 @@
-import re
+import os
 import json
 import tempfile
-from urllib.parse import urlparse
-
+import requests
+import pandas as pd
+import gradio as gr
+
+# =========================
+# Constants (不要改)
+# =========================
+DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+
+# =========================
+# Basic Agent
+# =========================
 class BasicAgent:
-    """
-    A pragmatic baseline agent:
-    - If question contains a URL, fetch it and try to extract the answer.
-    - If question implies there is an attached file, download it via /files/{task_id}.
-    - Uses simple heuristics to return an exact-match style answer (no extra words).
-    """
-
-    def __init__(self, api_url: str = DEFAULT_API_URL):
+    def __init__(self, api_url: str):
         self.api_url = api_url
-        self.session = requests.Session()
-        self.session.headers.update({
-            "User-Agent": "Mozilla/5.0 (compatible; BasicAgent/1.0)"
-        })
         print("BasicAgent initialized.")
 
     def __call__(self, question: str, task_id: str | None = None) -> str:
-        try:
-            # 1) If there is a file for this task, download and use it (when needed)
-            # Some questions explicitly mention "provided in the image" or "attached file"
-            if task_id and self._looks_like_file_task(question):
-                file_path = self._download_task_file(task_id)
-                if file_path:
-                    ans = self._answer_from_file(question, file_path)
-                    if ans:
-                        return self._finalize(ans)
-
-            # 2) If question contains URL(s), fetch and try to answer from the page
-            urls = self._extract_urls(question)
-            if urls:
-                for u in urls:
-                    html_or_text = self._safe_get(u)
-                    if html_or_text:
-                        ans = self._answer_from_web(question, u, html_or_text)
-                        if ans:
-                            return self._finalize(ans)
-
-            # 3) If no URL/file, try direct heuristics (grocery list, simple parsing, etc.)
-            ans = self._answer_from_text_only(question)
-            return self._finalize(ans) if ans else "unknown"
-
-        except Exception as e:
-            print(f"[Agent Error] {e}")
-            return "unknown"
-
-    # -------------------------
-    # Helpers
-    # -------------------------
-    def _finalize(self, s: str) -> str:
-        # EXACT MATCH friendly: strip spaces, avoid extra punctuation/labels
-        return str(s).strip()
+        print(f"Agent received question: {question[:80]}")
 
-    def _extract_urls(self, text: str) -> list[str]:
-        # robust URL regex
-        urls = re.findall(r"(https?://[^\s)>\]]+)", text)
-        # remove trailing punctuation
-        clean = []
-        for u in urls:
-            clean.append(u.rstrip(".,;:!?\"'"))
-        return clean
+        # 若沒有 task_id，直接回答文字題
+        if task_id is None:
+            return "I don't know."
 
-    def _safe_get(self, url: str, timeout: int = 20) -> str | None:
+        # 嘗試抓附件（有些題目會有）
         try:
-            r = self.session.get(url, timeout=timeout)
-            r.raise_for_status()
-            # return raw text; many pages are html
-            return r.text
+            file_url = f"{self.api_url}/files/{task_id}"
+            r = requests.get(file_url, timeout=10)
+            if r.status_code == 200 and r.headers.get("content-type", "").startswith("application/json"):
+                data = r.json()
+                return json.dumps(data)[:500]
         except Exception as e:
-            print(f"[GET failed] {url} -> {e}")
-            return None
+            print("File fetch failed:", e)
 
-    def _looks_like_file_task(self, question: str) -> bool:
-        q = question.lower()
-        keywords = [
-            "provided in the image", "provided in the file", "see the image",
-            "in the attached", "download", "the file", "the image",
-            "in the pdf", "in this spreadsheet", "in the document"
-        ]
-        return any(k in q for k in keywords)
-
-    def _download_task_file(self, task_id: str) -> str | None:
-        """
-        Download task file from /files/{task_id}.
-        """
-        url = f"{self.api_url}/files/{task_id}"
-        try:
-            r = self.session.get(url, timeout=30)
-            r.raise_for_status()
+        # 預設保底回答（至少不會 crash）
+        return "I don't know."
 
-            # Try to infer extension from headers
-            ctype = (r.headers.get("Content-Type") or "").lower()
-            ext = ""
-            if "pdf" in ctype:
-                ext = ".pdf"
-            elif "image" in ctype:
-                ext = ".png"
-            elif "text" in ctype:
-                ext = ".txt"
-            elif "csv" in ctype:
-                ext = ".csv"
-
-            fd, path = tempfile.mkstemp(suffix=ext)
-            os.close(fd)
-            with open(path, "wb") as f:
-                f.write(r.content)
-
-            print(f"[Downloaded] task file -> {path} ({ctype})")
-            return path
-        except Exception as e:
-            print(f"[File download failed] {url} -> {e}")
-            return None
+# =========================
+# Main runner
+# =========================
+def run_and_submit_all(profile: gr.OAuthProfile | None):
+    space_id = os.getenv("SPACE_ID")
 
-    # -------------------------
-    # Answering strategies
-    # -------------------------
-    def _answer_from_web(self, question: str, url: str, page_text: str) -> str | None:
-        q = question.lower()
+    if not profile:
+        return "Please login first.", None
 
-        # If question asks about Wikipedia (common): try to extract numbers / key fact around entity
-        if "wikipedia" in q or "wikipedia.org" in url:
-            return self._wiki_style_extract(question, page_text)
+    username = profile.username
+    api_url = DEFAULT_API_URL
+    questions_url = f"{api_url}/questions"
+    submit_url = f"{api_url}/submit"
 
-        # If question asks about "how many" and includes a year range, try find integer near keyword
-        if "how many" in q or "number of" in q:
-            num = self._find_best_number(question, page_text)
-            if num is not None:
-                return str(num)
+    # 建立 Agent（⚠️ 這裡已經修正）
+    agent = BasicAgent(api_url=api_url)
 
-        # If question asks "what is the surname" etc., try simple pattern match
-        if "surname" in q:
-            # naive: find "Surname" lines
-            m = re.search(r"Surname[:\s]+([A-Z][a-zA-Z-]+)", page_text)
-            if m:
-                return m.group(1)
+    agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
 
-        # Generic fallback: if question contains quoted phrase, find it and return nearby
-        return None
+    # 取得題目
+    response = requests.get(questions_url, timeout=15)
+    response.raise_for_status()
+    questions_data = response.json()
 
-    def _wiki_style_extract(self, question: str, page_text: str) -> str | None:
-        """
-        Very light heuristic:
-        - if asked "how many studio albums ... between YEAR and YEAR": count occurrences of 'studio album'
-          won't be accurate from raw html, so fallback to best-number extractor.
-        """
-        q = question.lower()
-        if "how many" in q:
-            num = self._find_best_number(question, page_text)
-            if num is not None:
-                return str(num)
-        # more could be added, but keep robust
-        return None
+    answers_payload = []
+    results_log = []
 
-    def _find_best_number(self, question: str, page_text: str) -> int | None:
-        """
-        Find a plausible answer number by looking at context keywords from question.
-        """
-        # Extract keywords (very simple)
-        q = question.lower()
-        # pick a few anchor words
-        anchors = []
-        for w in ["studio", "albums", "species", "camera", "published", "between", "highest", "number"]:
-            if w in q:
-                anchors.append(w)
+    for item in questions_data:
+        task_id = item.get("task_id")
+        question = item.get("question")
 
-        # Search within a reduced slice if possible
-        text = page_text
-        if anchors:
-            # try find a region around first anchor appearance
-            idx = text.lower().find(anchors[0])
-            if idx != -1:
-                start = max(0, idx - 2000)
-                end = min(len(text), idx + 2000)
-                text = text[start:end]
-
-        # grab integers
-        nums = re.findall(r"\b(\d{1,4})\b", text)
-        if not nums:
-            return None
-
-        # Heuristic: prefer smaller counts (1-300) over years (e.g., 2008)
-        candidates = []
-        for n in nums:
-            v = int(n)
-            if 0 <= v <= 500:
-                candidates.append(v)
-
-        if candidates:
-            # choose the most frequent candidate
-            from collections import Counter
-            c = Counter(candidates)
-            return c.most_common(1)[0][0]
-
-        return None
-
-    def _answer_from_file(self, question: str, file_path: str) -> str | None:
-        """
-        Minimal file handling:
-        - If it's text/csv: read and try parse.
-        - If pdf/image: we won't OCR here; return None.
-        """
         try:
-            # try as text
-            if file_path.endswith(".txt"):
-                with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
-                    content = f.read()
-                # try number extraction
-                num = self._find_best_number(question, content)
-                if num is not None:
-                    return str(num)
-                return None
-
-            if file_path.endswith(".csv"):
-                df = pd.read_csv(file_path)
-                # If asked something about a column, very naive: return first cell
-                if df.shape[0] > 0 and df.shape[1] > 0:
-                    return str(df.iloc[0, 0])
-                return None
-
-            # pdf/png: not handled in this baseline
-            return None
-
+            answer = agent(question, task_id=task_id)  # ⚠️ 關鍵修正點
         except Exception as e:
-            print(f"[File parse failed] {file_path} -> {e}")
-            return None
-
-    def _answer_from_text_only(self, question: str) -> str | None:
-        """
-        Pure text heuristics (no web/file).
-        Covers common GAIA-L1 style tasks like grocery categorization, alphabetize, comma-separated.
-        """
-        q = question.strip()
-
-        # Grocery list vegetable extraction pattern (example from your screenshot)
-        if "grocery list" in q.lower() and "vegetables" in q.lower() and "comma" in q.lower():
-            # Extract list after "Here's the list I have so far:"
-            m = re.search(r"Here's the list I have so far:\s*(.+?)\.\s*I need to make headings", q, re.S | re.I)
-            if not m:
-                m = re.search(r"Here's the list I have so far:\s*(.+)", q, re.S | re.I)
-            if m:
-                items_blob = m.group(1)
-                items = [x.strip().lower() for x in items_blob.split(",")]
-                items = [x for x in items if x]
+            answer = f"ERROR: {e}"
 
-                # Simple botany rule of thumb for that sample: treat these as vegetables
-                # (keep conservative: include obvious veggies, exclude fruits)
-                veggies_set = {
-                    "bell pepper", "broccoli", "celery", "corn", "green beans",
-                    "lettuce", "sweet potatoes", "zucchini"
-                }
-                veggies = []
-                for it in items:
-                    it_norm = it.strip()
-                    # normalize plurals
-                    if it_norm in veggies_set:
-                        veggies.append(it_norm)
-
-                veggies = sorted(set(veggies))
-                return ", ".join(veggies)
+        answers_payload.append({
+            "task_id": task_id,
+            "submitted_answer": answer
+        })
 
-        # If asked to reverse a sentence (common trick)
-        if "reverse" in q.lower() and "sentence" in q.lower():
-            # find quoted
-            m = re.search(r'"([^"]+)"', q)
-            if m:
-                return m.group(1)[::-1]
+        results_log.append({
+            "Task ID": task_id,
+            "Question": question,
+            "Submitted Answer": answer
+        })
 
-        return None
+    submission_data = {
+        "username": username,
+        "agent_code": agent_code,
+        "answers": answers_payload
+    }
+
+    # 提交
+    r = requests.post(submit_url, json=submission_data, timeout=60)
+    r.raise_for_status()
+    result = r.json()
+
+    status = (
+        f"Submission Successful!\n"
+        f"User: {result.get('username')}\n"
+        f"Score: {result.get('score')}%\n"
+        f"{result.get('correct_count')}/{result.get('total_attempted')} correct\n"
+        f"{result.get('message')}"
+    )
+
+    return status, pd.DataFrame(results_log)
+
+# =========================
+# Gradio UI
+# =========================
+with gr.Blocks() as demo:
+    gr.Markdown("# Basic Agent Evaluation Runner")
+
+    gr.Markdown("""
+    **Instructions**
+    1. Login with Hugging Face  
+    2. Click the button  
+    3. Wait for submission result  
+    """)
+
+    gr.LoginButton()
+
+    run_btn = gr.Button("Run Evaluation & Submit All Answers")
+
+    status_out = gr.Textbox(label="Status", lines=6)
+    table_out = gr.DataFrame(label="Results", wrap=True)
+
+    run_btn.click(
+        fn=run_and_submit_all,
+        outputs=[status_out, table_out]
+    )
+
+if __name__ == "__main__":
+    demo.launch()