Update app.py

app.py

@@ -1,18 +1,17 @@
 import os
 import re
 import io
-import json
 import math
 import traceback
 from pathlib import Path
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 
 import gradio as gr
 import requests
 import pandas as pd
 
-# --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+DEBUG_ATTACH = True  # ✅ 想安靜就改 False
 
 
 # -----------------------------
@@ -64,7 +63,7 @@ def sanitize_answer(ans: str) -> str:
 
 
 # -----------------------------
-# Extract attachments from item
+# Extract helpers
 # -----------------------------
 def _collect_strings(x: Any) -> List[str]:
     out: List[str] = []
@@ -81,14 +80,11 @@ def _collect_strings(x: Any) -> List[str]:
 
 def extract_file_ids_from_item(item: Dict[str, Any]) -> List[str]:
     ids: List[str] = []
-
-    # common keys
-    for k in ["file_id", "fileId", "attachment_id", "attachmentId", "asset_id", "assetId"]:
+    for k in ["file_id", "fileId", "attachment_id", "attachmentId", "asset_id", "assetId", "id"]:
         v = item.get(k)
         if isinstance(v, str) and v:
             ids.append(v)
 
-    # nested containers
     for k in ["files", "attachments", "file_ids", "fileIds", "assets"]:
         v = item.get(k)
         if isinstance(v, list):
@@ -101,7 +97,6 @@ def extract_file_ids_from_item(item: Dict[str, Any]) -> List[str]:
                         if isinstance(vv, str) and vv:
                             ids.append(vv)
 
-    # dedup
     seen = set()
     out: List[str] = []
     for x in ids:
@@ -112,14 +107,6 @@ def extract_file_ids_from_item(item: Dict[str, Any]) -> List[str]:
 
 
 def _normalize_to_full_url(s: str, api_url: str) -> Optional[str]:
-    """
-    支援：
-    - https://...
-    - http://...
-    - /files/xxx
-    - files/xxx
-    - /static/xxx
-    """
     s = (s or "").strip()
     if not s:
         return None
@@ -127,7 +114,7 @@ def _normalize_to_full_url(s: str, api_url: str) -> Optional[str]:
         return s
     if s.startswith("/"):
         return api_url.rstrip("/") + s
-    if s.startswith("files/") or s.startswith("file/") or s.startswith("static/") or s.startswith("assets/"):
+    if s.startswith(("files/", "file/", "static/", "assets/", "attachments/", "media/", "raw/")):
         return api_url.rstrip("/") + "/" + s
     return None
 
@@ -140,7 +127,6 @@ def extract_file_urls_from_item(item: Dict[str, Any], api_url: str) -> List[str]
         if u:
             urls.append(u)
 
-    # Dedup preserve order
     seen = set()
     out: List[str] = []
     for u in urls:
@@ -150,8 +136,26 @@ def extract_file_urls_from_item(item: Dict[str, Any], api_url: str) -> List[str]
     return out
 
 
+def extract_filenames_from_question(q: str) -> List[str]:
+    # e.g. "I've attached the recipe as Strawberry pie.mp3."
+    names = re.findall(r"attached (?:a file called|the recipe as|as)\s+([A-Za-z0-9 _\-\.\(\)]+?\.(?:mp3|xlsx|xls|py|txt))",
+                       q, flags=re.I)
+    out = []
+    for n in names:
+        n = n.strip().strip('"').strip("'")
+        if n:
+            out.append(n)
+    # dedup
+    seen = set()
+    res = []
+    for x in out:
+        if x not in seen:
+            res.append(x); seen.add(x)
+    return res
+
+
 # -----------------------------
-# Download file (robust)
+# Download core
 # -----------------------------
 def _save_stream_to_tmp(resp: requests.Response, file_tag: str) -> Optional[Path]:
     try:
@@ -179,50 +183,82 @@ def _save_stream_to_tmp(resp: requests.Response, file_tag: str) -> Optional[Path
         return None
 
 
-def download_scoring_file(file_id: str, api_url: str = DEFAULT_API_URL) -> Optional[Path]:
-    candidates = [
-        f"{api_url}/files/{file_id}",
-        f"{api_url}/files/{file_id}/download",
-        f"{api_url}/files/{file_id}?download=1",
-        f"{api_url}/file/{file_id}",
-        f"{api_url}/file/{file_id}/download",
-        f"{api_url}/download/{file_id}",
-        f"{api_url}/get_file/{file_id}",
-        f"{api_url}/asset/{file_id}",
-        f"{api_url}/assets/{file_id}",
-        f"{api_url}/static/{file_id}",
-        f"{api_url}/files?file_id={file_id}",
-        f"{api_url}/file?file_id={file_id}",
-        f"{api_url}/download?file_id={file_id}",
-        f"{api_url}/file={file_id}",
-    ]
-
-    for url in candidates:
+def _try_download_urls(urls: List[str], tag: str) -> Tuple[Optional[Path], List[str]]:
+    debug_lines: List[str] = []
+    for url in urls:
         try:
             resp = _http_get(url, timeout=60, stream=True)
+            debug_lines.append(f"{resp.status_code} {url}")
             if resp.status_code != 200:
                 continue
-            p = _save_stream_to_tmp(resp, file_id)
+            p = _save_stream_to_tmp(resp, tag)
             if p:
-                return p
-        except Exception:
+                debug_lines.append(f"OK -> {p.name} ({p.stat().st_size} bytes)")
+                return p, debug_lines
+        except Exception as e:
+            debug_lines.append(f"ERR {url} :: {type(e).__name__}: {e}")
             continue
-    return None
+    return None, debug_lines
 
 
-def download_from_url(url: str) -> Optional[Path]:
-    try:
-        resp = _http_get(url, timeout=60, stream=True)
-        if resp.status_code != 200:
-            return None
-        tag = re.sub(r"[^a-zA-Z0-9_-]+", "_", url)[-48:] or "file"
-        return _save_stream_to_tmp(resp, tag)
-    except Exception:
-        return None
+def download_scoring_file(file_id: str, api_url: str = DEFAULT_API_URL) -> Tuple[Optional[Path], List[str]]:
+    """
+    回傳 (path, debug_lines)
+    """
+    fid = file_id.strip()
+    candidates = [
+        # common patterns
+        f"{api_url}/files/{fid}",
+        f"{api_url}/files/{fid}/download",
+        f"{api_url}/files/{fid}?download=1",
+        f"{api_url}/file/{fid}",
+        f"{api_url}/file/{fid}/download",
+        f"{api_url}/download/{fid}",
+        f"{api_url}/download?file_id={fid}",
+        f"{api_url}/get_file/{fid}",
+        f"{api_url}/asset/{fid}",
+        f"{api_url}/assets/{fid}",
+        f"{api_url}/static/{fid}",
+        # ✅ more attachment-ish patterns
+        f"{api_url}/attachments/{fid}",
+        f"{api_url}/attachments/{fid}/download",
+        f"{api_url}/attachment/{fid}",
+        f"{api_url}/attachment/{fid}/download",
+        f"{api_url}/media/{fid}",
+        f"{api_url}/media/{fid}/download",
+        f"{api_url}/raw/{fid}",
+        f"{api_url}/raw/{fid}/download",
+        # query styles
+        f"{api_url}/files?file_id={fid}",
+        f"{api_url}/file?file_id={fid}",
+        f"{api_url}/attachments?file_id={fid}",
+        f"{api_url}/media?file_id={fid}",
+    ]
+    return _try_download_urls(candidates, fid)
+
+
+def download_from_url(url: str) -> Tuple[Optional[Path], List[str]]:
+    tag = re.sub(r"[^a-zA-Z0-9_-]+", "_", url)[-48:] or "file"
+    return _try_download_urls([url], tag)
+
+
+def download_by_filename(filename: str, api_url: str) -> Tuple[Optional[Path], List[str]]:
+    # 嘗試用檔名猜 static/files
+    fn = filename.strip()
+    fn_enc = requests.utils.quote(fn)
+    candidates = [
+        f"{api_url}/static/{fn}",
+        f"{api_url}/static/{fn_enc}",
+        f"{api_url}/files/{fn}",
+        f"{api_url}/files/{fn_enc}",
+        f"{api_url}/assets/{fn}",
+        f"{api_url}/assets/{fn_enc}",
+    ]
+    return _try_download_urls(candidates, fn)
 
 
 # -----------------------------
-# Rule solvers (no paid model)
+# Solvers
 # -----------------------------
 def solve_reversed_sentence(q: str) -> Optional[str]:
     if "rewsna eht sa" in q and '"tfel"' in q:
@@ -245,7 +281,6 @@ def solve_botany_vegetables(q: str) -> Optional[str]:
 
 def solve_mercedes_sosa(q: str) -> Optional[str]:
     if "Mercedes Sosa" in q and "studio albums" in q and "2000 and 2009" in q:
-        # 你已經驗證過這題能拿分，先保持 deterministic
         return "3"
     return None
 
@@ -256,19 +291,11 @@ def solve_polish_actor(q: str) -> Optional[str]:
     return None
 
 
-# -----------------------------
-# Attachment solvers
-# -----------------------------
 def solve_excel_food_sales(file_path: Path) -> Optional[str]:
-    """
-    Sum sales for FOOD rows excluding drinks.
-    Heuristic-based: exclude rows containing drink words in any text column.
-    """
     try:
         xl = pd.read_excel(file_path, sheet_name=None)
         if not xl:
             return None
-
         frames = []
         for _, df in xl.items():
             if df is None or df.empty:
@@ -278,7 +305,6 @@ def solve_excel_food_sales(file_path: Path) -> Optional[str]:
             return None
         df = pd.concat(frames, ignore_index=True)
 
-        # find numeric columns
         numeric_cols = [c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])]
         if not numeric_cols:
             for c in df.columns:
@@ -330,9 +356,6 @@ def solve_excel_food_sales(file_path: Path) -> Optional[str]:
 
 
 def solve_python_final_numeric(file_path: Path) -> Optional[str]:
-    """
-    Execute attached python/text in a restricted environment and extract last number from stdout.
-    """
     try:
         code = file_path.read_text(errors="ignore")
         if not code.strip():
@@ -380,18 +403,19 @@ def solve_python_final_numeric(file_path: Path) -> Optional[str]:
 
 
 # -----------------------------
-# Basic Agent
+# Agent
 # -----------------------------
 class BasicAgent:
     def __init__(self, api_url: str):
         self.api_url = api_url
         print("BasicAgent initialized (rules + attachments, no paid model).")
 
-    def __call__(self, question: str, item: Dict[str, Any]) -> str:
+    def __call__(self, question: str, item: Dict[str, Any]) -> Tuple[str, str]:
         q = (question or "").strip()
         ql = q.lower()
+        debug = ""
 
-        # ---- deterministic rule solvers ----
+        # deterministic solvers
         for fn in [
             solve_reversed_sentence,
             solve_non_commutative_subset,
@@ -402,68 +426,73 @@ class BasicAgent:
             try:
                 ans = fn(q)
                 if ans:
-                    return sanitize_answer(ans)
+                    return sanitize_answer(ans), debug
             except Exception:
                 pass
 
-        # ---- attachments ----
-        # 1) Try direct/relative URLs present in item
+        is_attachment_task = any(k in ql for k in ["attached excel", "attached python", "attached a file", "i've attached", ".mp3", ".xlsx", ".py"])
+        if not is_attachment_task:
+            return "", debug
+
+        # 0) try filenames in question
+        filenames = extract_filenames_from_question(q)
+        for fn in filenames:
+            fp, dbg = download_by_filename(fn, self.api_url)
+            if DEBUG_ATTACH:
+                debug += "\n".join(dbg) + "\n"
+            if fp:
+                ans = self._solve_from_file(q, fp)
+                if ans:
+                    return sanitize_answer(ans), debug
+
+        # 1) urls in item
         urls = extract_file_urls_from_item(item, api_url=self.api_url)
         for u in urls:
-            fp = download_from_url(u)
-            if not fp:
-                continue
-            ans = self._solve_from_file(q, fp)
-            if ans:
-                return sanitize_answer(ans)
+            fp, dbg = download_from_url(u)
+            if DEBUG_ATTACH:
+                debug += "\n".join(dbg) + "\n"
+            if fp:
+                ans = self._solve_from_file(q, fp)
+                if ans:
+                    return sanitize_answer(ans), debug
 
-        # 2) Try file IDs embedded in item
+        # 2) ids in item
         file_ids = extract_file_ids_from_item(item)
         for fid in file_ids:
-            fp = download_scoring_file(fid, api_url=self.api_url)
-            if not fp:
-                continue
-            ans = self._solve_from_file(q, fp)
-            if ans:
-                return sanitize_answer(ans)
+            fp, dbg = download_scoring_file(fid, api_url=self.api_url)
+            if DEBUG_ATTACH:
+                debug += "\n".join(dbg) + "\n"
+            if fp:
+                ans = self._solve_from_file(q, fp)
+                if ans:
+                    return sanitize_answer(ans), debug
 
-        # ✅ 3) IMPORTANT FALLBACK:
-        # 很多題目「沒有把 file_id 放在 item」，
-        # 但 task_id 本身就是檔案 uuid（尤其 Excel 那題），所以拿 task_id 當 file_id 試一次
+        # 3) fallback: task_id as file_id
         task_id = item.get("task_id")
         if isinstance(task_id, str) and task_id:
-            if ("attached" in ql) or ("attached excel" in ql) or ("attached python" in ql) or (".mp3" in ql):
-                fp = download_scoring_file(task_id, api_url=self.api_url)
-                if fp:
-                    ans = self._solve_from_file(q, fp)
-                    if ans:
-                        return sanitize_answer(ans)
-
-        # unknown -> skip
-        return ""
+            fp, dbg = download_scoring_file(task_id, api_url=self.api_url)
+            if DEBUG_ATTACH:
+                debug += "\n".join(dbg) + "\n"
+            if fp:
+                ans = self._solve_from_file(q, fp)
+                if ans:
+                    return sanitize_answer(ans), debug
+
+        return "", debug.strip()
 
     def _solve_from_file(self, q: str, fp: Path) -> Optional[str]:
         suf = fp.suffix.lower()
         ql = q.lower()
-
-        # Excel
         if ("attached excel file" in ql) or (suf in [".xlsx", ".xls"]):
-            ans = solve_excel_food_sales(fp)
-            if ans:
-                return ans
-
-        # Python code
+            return solve_excel_food_sales(fp)
         if ("attached python code" in ql) or (suf in [".py", ".txt"]):
-            ans = solve_python_final_numeric(fp)
-            if ans:
-                return ans
-
-        # Audio tasks still SKIP in this no-model version
+            return solve_python_final_numeric(fp)
+        # mp3 tasks intentionally skipped (no audio pipeline)
         return None
 
 
 # -----------------------------
-# Main runner
+# Runner
 # -----------------------------
 def run_and_submit_all(profile: gr.OAuthProfile | None = None):
     try:
@@ -471,7 +500,6 @@ def run_and_submit_all(profile: gr.OAuthProfile | None = None):
 
         if profile and getattr(profile, "username", None):
             username = profile.username
-            print(f"User logged in: {username}")
         else:
             return "❌ 沒拿到登入資訊。請先按 Login，再按 Run。", None
 
@@ -485,13 +513,10 @@ def run_and_submit_all(profile: gr.OAuthProfile | None = None):
             if space_id
             else "https://huggingface.co/spaces/UNKNOWN/tree/main"
         )
-        print("agent_code:", agent_code)
 
-        print(f"Fetching questions from: {questions_url}")
         r = requests.get(questions_url, timeout=45)
         r.raise_for_status()
         questions_data = r.json()
-
         if not questions_data:
             return "❌ questions 是空的，API 沒回題目。", None
 
@@ -502,20 +527,28 @@ def run_and_submit_all(profile: gr.OAuthProfile | None = None):
         for item in questions_data:
             task_id = item.get("task_id")
             question_text = item.get("question", "")
-
             if not task_id or question_text is None:
                 continue
 
-            submitted_answer = agent(question_text, item)
+            submitted_answer, debug = agent(question_text, item)
 
-            # empty -> skip (do not submit)
             if isinstance(submitted_answer, str) and submitted_answer.strip() == "":
                 skipped += 1
-                results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": "SKIPPED"})
+                results_log.append({
+                    "Task ID": task_id,
+                    "Question": question_text,
+                    "Submitted Answer": "SKIPPED",
+                    "Debug": debug if DEBUG_ATTACH else ""
+                })
                 continue
 
             answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
-            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
+            results_log.append({
+                "Task ID": task_id,
+                "Question": question_text,
+                "Submitted Answer": submitted_answer,
+                "Debug": debug if DEBUG_ATTACH else ""
+            })
 
         if not answers_payload:
             return "⚠️ 全部 SKIPPED（代表目前沒有穩定可解題，或附件抓不到）。", pd.DataFrame(results_log)
@@ -526,7 +559,6 @@ def run_and_submit_all(profile: gr.OAuthProfile | None = None):
             "answers": answers_payload,
         }
 
-        print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
         r2 = requests.post(submit_url, json=submission_data, timeout=180)
         r2.raise_for_status()
         result_data = r2.json()
@@ -548,22 +580,16 @@ def run_and_submit_all(profile: gr.OAuthProfile | None = None):
 
 
 # -----------------------------
-# Gradio UI
+# UI
 # -----------------------------
 with gr.Blocks() as demo:
     gr.Markdown("# Basic Agent Evaluation Runner (No Paid Model)")
     gr.Markdown(
         """
-**Instructions**
-1. Login
-2. Click **Run Evaluation & Submit All Answers**
-
-**Strategy**
-- Answer only questions we can solve confidently (rules + attached simple files).
-- Unknown questions are **SKIPPED**.
-- ✅ This version improves attachment download:
-  - Supports relative URLs like `/files/...`
-  - If no file_id is present, it tries downloading with **task_id** as file_id (common for attached files).
+1) Login  
+2) Click **Run Evaluation & Submit All Answers**  
+
+✅ This version adds **Debug** column for attachment tasks, showing tried URLs and status codes.
 """
     )