Update app.py

app.py

@@ -61,7 +61,7 @@ def sanitize_answer(ans: str) -> str:
 
 
 # -----------------------------
-# Utils: collect strings
+# Utils
 # -----------------------------
 def _collect_strings(x: Any) -> List[str]:
     out: List[str] = []
@@ -95,6 +95,7 @@ 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,7 +113,6 @@ def _normalize_to_full_url(s: str, api_url: str) -> Optional[str]:
         return s
     if s.startswith("/"):
         return api_url.rstrip("/") + s
-    # relative-ish
     if s.startswith(("files/", "file/", "static/", "assets/", "attachments/", "media/", "raw/", "api/")):
         return api_url.rstrip("/") + "/" + s
     return None
@@ -124,7 +124,6 @@ def extract_file_urls_from_item(item: Dict[str, Any], api_url: str) -> List[str]
         u = _normalize_to_full_url(s, api_url)
         if u:
             urls.append(u)
-    # dedup
     seen = set()
     out = []
     for u in urls:
@@ -136,7 +135,7 @@ def extract_file_urls_from_item(item: Dict[str, Any], api_url: str) -> List[str]
 
 def extract_filenames_from_question(q: str) -> List[str]:
     names = re.findall(
-        r"attached (?:a file called|the recipe as|as)\s+([A-Za-z0-9 _\-\.\(\)]+?\.(?:mp3|xlsx|xls|py|txt))",
+        r"(?:attached a file called|attached the recipe as|attached a file|file called)\s+([A-Za-z0-9 _\-\.\(\)]+?\.(?:mp3|xlsx|xls|py|txt))",
         q,
         flags=re.I,
     )
@@ -145,7 +144,6 @@ def extract_filenames_from_question(q: str) -> List[str]:
         n = n.strip().strip('"').strip("'")
         if n:
             out.append(n)
-    # dedup
     seen = set()
     res = []
     for x in out:
@@ -155,8 +153,13 @@ def extract_filenames_from_question(q: str) -> List[str]:
     return res
 
 
+def url_quote_filename(name: str) -> str:
+    # minimal url-encoding for spaces
+    return name.replace(" ", "%20")
+
+
 # -----------------------------
-# Download & save helpers
+# Download helpers
 # -----------------------------
 def _save_stream_to_tmp(resp: requests.Response, file_tag: str) -> Optional[Path]:
     try:
@@ -202,60 +205,7 @@ def _try_download_urls(urls: List[str], tag: str) -> Tuple[Optional[Path], List[
 
 
 # -----------------------------
-# OpenAPI discovery
-# -----------------------------
-def discover_openapi_paths(api_url: str) -> List[str]:
-    """
-    If openapi exists, extract candidate path templates that look like file download endpoints.
-    """
-    probes = [
-        f"{api_url}/openapi.json",
-        f"{api_url}/openapi",
-        f"{api_url}/swagger.json",
-        f"{api_url}/api/openapi.json",
-    ]
-    candidates: List[str] = []
-    for p in probes:
-        try:
-            r = _http_get(p, timeout=20, stream=False)
-            if r.status_code != 200:
-                continue
-            data = r.json()
-            paths = data.get("paths", {}) if isinstance(data, dict) else {}
-            for path in paths.keys():
-                low = path.lower()
-                if any(k in low for k in ["file", "files", "attachment", "download", "asset", "media"]):
-                    candidates.append(path)
-        except Exception:
-            continue
-
-    # dedup
-    seen = set()
-    out = []
-    for x in candidates:
-        if x not in seen:
-            out.append(x)
-            seen.add(x)
-    return out
-
-
-def build_openapi_url_candidates(api_url: str, fid: str, openapi_paths: List[str]) -> List[str]:
-    urls = []
-    for path in openapi_paths:
-        # Replace common params
-        u = path
-        u = u.replace("{file_id}", fid).replace("{fileId}", fid).replace("{id}", fid).replace("{attachment_id}", fid)
-        if "{" in u and "}" in u:
-            # still has unknown template vars
-            continue
-        if not u.startswith("/"):
-            u = "/" + u
-        urls.append(api_url.rstrip("/") + u)
-    return urls
-
-
-# -----------------------------
-# Base64-in-item extraction (IMPORTANT)
+# Base64-in-item extraction (備用)
 # -----------------------------
 _B64_KEYS = {
     "data", "content", "blob", "bytes", "file_bytes", "filebytes", "b64", "base64",
@@ -268,13 +218,10 @@ def looks_like_base64(s: str) -> bool:
     t = s.strip()
     if len(t) < 200:
         return False
-    # allow data:...;base64,....
     if t.startswith("data:") and "base64," in t:
         return True
-    # base64 charset check (loose)
     if re.fullmatch(r"[A-Za-z0-9+/=\s]+", t) is None:
         return False
-    # must have padding-ish or length multiple-ish (loose)
     return True
 
 
@@ -291,15 +238,13 @@ def decode_base64_to_file(b64s: str, filename_hint: str) -> Optional[Path]:
         out_dir = Path("/tmp/gaia_files")
         out_dir.mkdir(parents=True, exist_ok=True)
 
-        # infer suffix by magic if missing
         name = filename_hint or "attachment"
         if "." not in name:
-            # quick magic guesses
             if raw[:2] == b"PK":
                 name += ".xlsx"
             elif raw[:3] == b"ID3" or raw[:2] == b"\xff\xfb":
                 name += ".mp3"
-            elif raw[:1] == b"#" or b"import" in raw[:200]:
+            elif b"import" in raw[:200]:
                 name += ".py"
             else:
                 name += ".bin"
@@ -313,9 +258,6 @@ def decode_base64_to_file(b64s: str, filename_hint: str) -> Optional[Path]:
 
 
 def extract_base64_files_from_item(item: Any, filename_hint: str) -> Tuple[List[Path], List[str]]:
-    """
-    Deep-scan dict/list for base64 strings under likely keys.
-    """
     found_paths: List[Path] = []
     debug: List[str] = []
 
@@ -323,7 +265,6 @@ def extract_base64_files_from_item(item: Any, filename_hint: str) -> Tuple[List[
         if isinstance(x, dict):
             for k, v in x.items():
                 kh = f"{key_hint}.{k}" if key_hint else str(k)
-                # if key suggests file-like, attempt decode when value is b64-ish string
                 if isinstance(v, str) and (k.lower() in _B64_KEYS or "base64" in k.lower() or "b64" in k.lower()):
                     if looks_like_base64(v):
                         p = decode_base64_to_file(v, filename_hint)
@@ -336,16 +277,13 @@ def extract_base64_files_from_item(item: Any, filename_hint: str) -> Tuple[List[
         elif isinstance(x, list):
             for i, y in enumerate(x):
                 walk(y, f"{key_hint}[{i}]")
-        else:
-            # not traversable
-            return
 
     walk(item)
     return found_paths, debug
 
 
 # -----------------------------
-# Rule solvers (known correct)
+# Deterministic solvers (你已經答對的)
 # -----------------------------
 def solve_reversed_sentence(q: str) -> Optional[str]:
     if "rewsna eht sa" in q and '"tfel"' in q:
@@ -483,8 +421,7 @@ def solve_python_final_numeric(file_path: Path) -> Optional[str]:
 class BasicAgent:
     def __init__(self, api_url: str):
         self.api_url = api_url.rstrip("/")
-        self.openapi_paths = discover_openapi_paths(self.api_url)
-        print("BasicAgent initialized. openapi_paths:", len(self.openapi_paths))
+        print("BasicAgent initialized.")
 
     def __call__(self, question: str, item: Dict[str, Any]) -> Tuple[str, str]:
         q = (question or "").strip()
@@ -500,40 +437,96 @@ class BasicAgent:
             except Exception:
                 pass
 
-        is_attachment_task = any(k in ql for k in ["attached excel", "attached python", "attached a file", "i've attached", ".mp3", ".xlsx", ".py"])
+        is_attachment_task = any(k in ql for k in ["attached excel", "attached python", "i've attached", ".mp3", ".xlsx", ".py"])
         if not is_attachment_task:
             return "", ""
 
-        # filename hint
+        task_id = str(item.get("task_id", "")).strip()
+        file_name = str(item.get("file_name", "")).strip()  # <<<<<< 你缺的就是用它
         filenames = extract_filenames_from_question(q)
-        filename_hint = filenames[0] if filenames else "attachment"
-
-        # 1) try base64 inside item (NEW)
-        paths, dbg = extract_base64_files_from_item(item, filename_hint=filename_hint)
-        debug_lines.extend(dbg)
-        for fp in paths:
-            ans = self._solve_from_file(q, fp)
-            if ans:
-                return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
-
-        # 2) try URLs found in item
-        urls = extract_file_urls_from_item(item, api_url=self.api_url)
-        for u in urls:
-            fp, dbg2 = _try_download_urls([u], tag=filename_hint)
-            debug_lines.extend(dbg2)
+        filename_hint = filenames[0] if filenames else (file_name or "attachment")
+
+        # 0) 先嘗試打題目詳情 (很多系統附件藏在這裡)
+        detail_candidates = [
+            f"{self.api_url}/question/{task_id}",
+            f"{self.api_url}/questions/{task_id}",
+            f"{self.api_url}/task/{task_id}",
+            f"{self.api_url}/tasks/{task_id}",
+            f"{self.api_url}/api/question/{task_id}",
+            f"{self.api_url}/api/questions/{task_id}",
+        ]
+        detail_json = None
+        for u in detail_candidates:
+            try:
+                r = _http_get(u, timeout=20, stream=False)
+                debug_lines.append(f"{r.status_code} {u}")
+                if r.status_code == 200 and "application/json" in (r.headers.get("content-type","").lower()):
+                    detail_json = r.json()
+                    debug_lines.append("DETAIL_OK: got json")
+                    break
+            except Exception as e:
+                debug_lines.append(f"ERR {u} :: {type(e).__name__}: {e}")
+
+        # 1) base64 in detail/item
+        for src_name, src in [("DETAIL", detail_json), ("ITEM", item)]:
+            if src:
+                paths, dbg = extract_base64_files_from_item(src, filename_hint=filename_hint)
+                debug_lines.extend([f"{src_name}::{x}" for x in dbg])
+                for fp in paths:
+                    ans = self._solve_from_file(q, fp)
+                    if ans:
+                        return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
+
+        # 2) url strings in detail/item
+        for src_name, src in [("DETAIL", detail_json), ("ITEM", item)]:
+            if src:
+                urls = extract_file_urls_from_item(src, api_url=self.api_url)
+                if urls:
+                    fp, dbg2 = _try_download_urls(urls, tag=filename_hint)
+                    debug_lines.extend([f"{src_name}::{x}" for x in dbg2])
+                    if fp:
+                        ans = self._solve_from_file(q, fp)
+                        if ans:
+                            return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
+
+        # 3) 用 file_name 組路徑（你目前最缺的）
+        #    (你的 debug 顯示 item 就只有這個線索)
+        if file_name:
+            fn_q = url_quote_filename(file_name)
+            fn_candidates = [
+                # direct filename
+                f"{self.api_url}/static/{fn_q}",
+                f"{self.api_url}/files/{fn_q}",
+                f"{self.api_url}/assets/{fn_q}",
+                f"{self.api_url}/media/{fn_q}",
+                f"{self.api_url}/raw/{fn_q}",
+                f"{self.api_url}/api/static/{fn_q}",
+                f"{self.api_url}/api/files/{fn_q}",
+                f"{self.api_url}/api/assets/{fn_q}",
+                f"{self.api_url}/api/media/{fn_q}",
+                # task_id + filename (常見)
+                f"{self.api_url}/files/{task_id}/{fn_q}",
+                f"{self.api_url}/files/{task_id}/download/{fn_q}",
+                f"{self.api_url}/download/{task_id}/{fn_q}",
+                f"{self.api_url}/api/files/{task_id}/{fn_q}",
+                f"{self.api_url}/api/download/{task_id}/{fn_q}",
+                # query style
+                f"{self.api_url}/download?task_id={task_id}&file_name={fn_q}",
+                f"{self.api_url}/download?task_id={task_id}&filename={fn_q}",
+                f"{self.api_url}/api/download?task_id={task_id}&file_name={fn_q}",
+                f"{self.api_url}/api/download?task_id={task_id}&filename={fn_q}",
+            ]
+            fp, dbg3 = _try_download_urls(fn_candidates, tag=file_name)
+            debug_lines.extend(dbg3)
             if fp:
                 ans = self._solve_from_file(q, fp)
                 if ans:
                     return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
 
-        # 3) try ids in item
+        # 4) id-based fallback（保留）
         file_ids = extract_file_ids_from_item(item)
-        # fallback: task_id also as id
-        tid = item.get("task_id")
-        if isinstance(tid, str) and tid:
-            file_ids.append(tid)
-
-        # dedup ids
+        if task_id:
+            file_ids.append(task_id)
         seen = set()
         file_ids2 = []
         for x in file_ids:
@@ -541,80 +534,44 @@ class BasicAgent:
                 file_ids2.append(x); seen.add(x)
 
         for fid in file_ids2:
-            fp, dbg3 = self._download_by_id(fid, filename_hint)
-            debug_lines.extend(dbg3)
+            candidates = [
+                f"{self.api_url}/files/{fid}",
+                f"{self.api_url}/files/{fid}/download",
+                f"{self.api_url}/file/{fid}",
+                f"{self.api_url}/download/{fid}",
+                f"{self.api_url}/get_file/{fid}",
+                f"{self.api_url}/assets/{fid}",
+                f"{self.api_url}/static/{fid}",
+                f"{self.api_url}/attachments/{fid}",
+                f"{self.api_url}/media/{fid}",
+                f"{self.api_url}/raw/{fid}",
+                f"{self.api_url}/api/files/{fid}",
+                f"{self.api_url}/api/files/{fid}/download",
+                f"{self.api_url}/api/file/{fid}",
+                f"{self.api_url}/api/download/{fid}",
+                f"{self.api_url}/file={fid}",
+                f"{self.api_url}/gradio_api/file={fid}",
+                f"{self.api_url}/download?file_id={fid}",
+                f"{self.api_url}/api/download?file_id={fid}",
+            ]
+            fp, dbg4 = _try_download_urls(candidates, tag=filename_hint)
+            debug_lines.extend(dbg4)
             if fp:
                 ans = self._solve_from_file(q, fp)
                 if ans:
                     return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
 
-        # last: add item keys snapshot (so you can see where file is)
         if DEBUG_ATTACH:
             try:
                 keys = sorted(list(item.keys()))
                 debug_lines.append("ITEM_KEYS: " + ", ".join(keys))
+                if file_name:
+                    debug_lines.append(f"ITEM_FILE_NAME: {file_name}")
             except Exception:
                 pass
 
         return "", "\n".join(debug_lines).strip() if DEBUG_ATTACH else ""
 
-    def _download_by_id(self, fid: str, tag: str) -> Tuple[Optional[Path], List[str]]:
-        fid = fid.strip()
-        api = self.api_url
-
-        candidates = []
-
-        # openapi discovered paths
-        candidates += build_openapi_url_candidates(api, fid, self.openapi_paths)
-
-        # common REST-ish
-        candidates += [
-            f"{api}/files/{fid}",
-            f"{api}/files/{fid}/download",
-            f"{api}/file/{fid}",
-            f"{api}/download/{fid}",
-            f"{api}/get_file/{fid}",
-            f"{api}/assets/{fid}",
-            f"{api}/static/{fid}",
-            f"{api}/attachments/{fid}",
-            f"{api}/media/{fid}",
-            f"{api}/raw/{fid}",
-        ]
-
-        # API prefix (very common)
-        candidates += [
-            f"{api}/api/files/{fid}",
-            f"{api}/api/files/{fid}/download",
-            f"{api}/api/file/{fid}",
-            f"{api}/api/download/{fid}",
-            f"{api}/api/attachments/{fid}",
-            f"{api}/api/media/{fid}",
-        ]
-
-        # HF/Gradio-style file serving (common on Spaces)
-        candidates += [
-            f"{api}/file={fid}",
-            f"{api}/gradio_api/file={fid}",
-            f"{api}/gradio_api/file={fid}&download=1",
-        ]
-
-        # query styles
-        candidates += [
-            f"{api}/download?file_id={fid}",
-            f"{api}/api/download?file_id={fid}",
-            f"{api}/files?file_id={fid}",
-            f"{api}/api/files?file_id={fid}",
-        ]
-
-        # dedup preserve order
-        seen = set()
-        cand2 = []
-        for u in candidates:
-            if u not in seen:
-                cand2.append(u); seen.add(u)
-
-        return _try_download_urls(cand2, tag)
-
     def _solve_from_file(self, q: str, fp: Path) -> Optional[str]:
         suf = fp.suffix.lower()
         ql = q.lower()
@@ -625,7 +582,7 @@ class BasicAgent:
         if ("attached python code" in ql) or (suf in [".py", ".txt"]):
             return solve_python_final_numeric(fp)
 
-        # mp3 tasks are intentionally skipped (no audio pipeline)
+        # mp3 tasks 仍然 skip（你目前沒做音訊辨識）
         return None
 
 
@@ -704,7 +661,7 @@ def run_and_submit_all(profile: gr.OAuthProfile | None = None):
 # -----------------------------
 with gr.Blocks() as demo:
     gr.Markdown("# Basic Agent Evaluation Runner (No Paid Model)")
-    gr.Markdown("✅ This version tries **base64-in-item** + **openapi discovery** + more HF Space download patterns.\n\nDebug column shows what happened.")
+    gr.Markdown("✅ This version tries: **question detail endpoints** + **file_name path patterns** + url/base64 scan.\n\nDebug欄會顯示嘗試過哪些網址。")
     gr.LoginButton()
     run_button = gr.Button("Run Evaluation & Submit All Answers")
     status_output = gr.Textbox(label="Run Status / Submission Result", lines=14, interactive=False)