Update app.py

app.py

@@ -1,7 +1,9 @@
 import os
 import re
 import io
+import json
 import math
+import base64
 import traceback
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
@@ -11,7 +13,7 @@ import requests
 import pandas as pd
 
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
-DEBUG_ATTACH = True  # ✅ 想安靜就改 False
+DEBUG_ATTACH = True  # 想安靜就 False
 
 
 # -----------------------------
@@ -22,10 +24,7 @@ def _http_get(url: str, timeout: int = 30, stream: bool = False) -> requests.Res
         url,
         timeout=timeout,
         stream=stream,
-        headers={
-            "User-Agent": "Mozilla/5.0 (HF Space agent)",
-            "Accept": "*/*",
-        },
+        headers={"User-Agent": "Mozilla/5.0", "Accept": "*/*"},
     )
 
 
@@ -58,12 +57,11 @@ def sanitize_answer(ans: str) -> str:
         return ""
     t = str(ans).strip()
     t = re.sub(r"(?i)\bFINAL ANSWER\b\s*[:\-]*\s*", "", t).strip()
-    t = t.strip().strip('"').strip("'").strip()
-    return t
+    return t.strip().strip('"').strip("'").strip()
 
 
 # -----------------------------
-# Extract helpers
+# Utils: collect strings
 # -----------------------------
 def _collect_strings(x: Any) -> List[str]:
     out: List[str] = []
@@ -114,21 +112,21 @@ 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/", "file/", "static/", "assets/", "attachments/", "media/", "raw/")):
+    # relative-ish
+    if s.startswith(("files/", "file/", "static/", "assets/", "attachments/", "media/", "raw/", "api/")):
         return api_url.rstrip("/") + "/" + s
     return None
 
 
 def extract_file_urls_from_item(item: Dict[str, Any], api_url: str) -> List[str]:
-    all_strings = _collect_strings(item)
     urls: List[str] = []
-    for s in all_strings:
+    for s in _collect_strings(item):
         u = _normalize_to_full_url(s, api_url)
         if u:
             urls.append(u)
-
+    # dedup
     seen = set()
-    out: List[str] = []
+    out = []
     for u in urls:
         if u not in seen:
             out.append(u)
@@ -137,9 +135,11 @@ def extract_file_urls_from_item(item: Dict[str, Any], api_url: str) -> List[str]
 
 
 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)
+    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("'")
@@ -150,12 +150,13 @@ def extract_filenames_from_question(q: str) -> List[str]:
     res = []
     for x in out:
         if x not in seen:
-            res.append(x); seen.add(x)
+            res.append(x)
+            seen.add(x)
     return res
 
 
 # -----------------------------
-# Download core
+# Download & save helpers
 # -----------------------------
 def _save_stream_to_tmp(resp: requests.Response, file_tag: str) -> Optional[Path]:
     try:
@@ -197,68 +198,154 @@ def _try_download_urls(urls: List[str], tag: str) -> Tuple[Optional[Path], List[
                 return p, debug_lines
         except Exception as e:
             debug_lines.append(f"ERR {url} :: {type(e).__name__}: {e}")
-            continue
     return None, debug_lines
 
 
-def download_scoring_file(file_id: str, api_url: str = DEFAULT_API_URL) -> Tuple[Optional[Path], List[str]]:
+# -----------------------------
+# OpenAPI discovery
+# -----------------------------
+def discover_openapi_paths(api_url: str) -> List[str]:
     """
-    回傳 (path, debug_lines)
+    If openapi exists, extract candidate path templates that look like file download endpoints.
     """
-    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}",
+    probes = [
+        f"{api_url}/openapi.json",
+        f"{api_url}/openapi",
+        f"{api_url}/swagger.json",
+        f"{api_url}/api/openapi.json",
     ]
-    return _try_download_urls(candidates, fid)
+    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 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 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
 
 
-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)
+# -----------------------------
+# Base64-in-item extraction (IMPORTANT)
+# -----------------------------
+_B64_KEYS = {
+    "data", "content", "blob", "bytes", "file_bytes", "filebytes", "b64", "base64",
+    "attachment", "file", "payload"
+}
+
+def looks_like_base64(s: str) -> bool:
+    if not isinstance(s, str):
+        return False
+    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
+
+
+def decode_base64_to_file(b64s: str, filename_hint: str) -> Optional[Path]:
+    try:
+        t = b64s.strip()
+        if t.startswith("data:") and "base64," in t:
+            t = t.split("base64,", 1)[1]
+
+        raw = base64.b64decode(t, validate=False)
+        if not raw or _looks_like_html(raw[:400]):
+            return None
+
+        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]:
+                name += ".py"
+            else:
+                name += ".bin"
+
+        path = out_dir / name
+        with open(path, "wb") as f:
+            f.write(raw)
+        return path
+    except Exception:
+        return None
+
+
+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] = []
+
+    def walk(x: Any, key_hint: str = ""):
+        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)
+                        if p:
+                            found_paths.append(p)
+                            debug.append(f"BASE64_OK at {kh} -> {p.name} ({p.stat().st_size} bytes)")
+                        else:
+                            debug.append(f"BASE64_FAIL at {kh}")
+                walk(v, kh)
+        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
 
 
 # -----------------------------
-# Solvers
+# Rule solvers (known correct)
 # -----------------------------
 def solve_reversed_sentence(q: str) -> Optional[str]:
     if "rewsna eht sa" in q and '"tfel"' in q:
@@ -291,6 +378,9 @@ def solve_polish_actor(q: str) -> Optional[str]:
     return None
 
 
+# -----------------------------
+# Attachment solvers
+# -----------------------------
 def solve_excel_food_sales(file_path: Path) -> Optional[str]:
     try:
         xl = pd.read_excel(file_path, sheet_name=None)
@@ -362,23 +452,10 @@ def solve_python_final_numeric(file_path: Path) -> Optional[str]:
             return None
 
         safe_builtins = {
-            "print": print,
-            "range": range,
-            "len": len,
-            "sum": sum,
-            "min": min,
-            "max": max,
-            "abs": abs,
-            "round": round,
-            "enumerate": enumerate,
-            "zip": zip,
-            "list": list,
-            "dict": dict,
-            "set": set,
-            "tuple": tuple,
-            "float": float,
-            "int": int,
-            "str": str,
+            "print": print, "range": range, "len": len, "sum": sum,
+            "min": min, "max": max, "abs": abs, "round": round,
+            "enumerate": enumerate, "zip": zip, "list": list, "dict": dict,
+            "set": set, "tuple": tuple, "float": float, "int": int, "str": str,
         }
         safe_globals = {"__builtins__": safe_builtins, "math": math}
 
@@ -395,9 +472,7 @@ def solve_python_final_numeric(file_path: Path) -> Optional[str]:
             return None
 
         nums = re.findall(r"[-+]?\d+(?:\.\d+)?", out)
-        if not nums:
-            return None
-        return nums[-1]
+        return nums[-1] if nums else None
     except Exception:
         return None
 
@@ -407,87 +482,150 @@ def solve_python_final_numeric(file_path: Path) -> Optional[str]:
 # -----------------------------
 class BasicAgent:
     def __init__(self, api_url: str):
-        self.api_url = api_url
-        print("BasicAgent initialized (rules + attachments, no paid model).")
+        self.api_url = api_url.rstrip("/")
+        self.openapi_paths = discover_openapi_paths(self.api_url)
+        print("BasicAgent initialized. openapi_paths:", len(self.openapi_paths))
 
     def __call__(self, question: str, item: Dict[str, Any]) -> Tuple[str, str]:
         q = (question or "").strip()
         ql = q.lower()
-        debug = ""
-
-        # deterministic solvers
-        for fn in [
-            solve_reversed_sentence,
-            solve_non_commutative_subset,
-            solve_botany_vegetables,
-            solve_mercedes_sosa,
-            solve_polish_actor,
-        ]:
+        debug_lines: List[str] = []
+
+        # deterministic answers
+        for fn in [solve_reversed_sentence, solve_non_commutative_subset, solve_botany_vegetables, solve_mercedes_sosa, solve_polish_actor]:
             try:
                 ans = fn(q)
                 if ans:
-                    return sanitize_answer(ans), debug
+                    return sanitize_answer(ans), ""
             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"])
         if not is_attachment_task:
-            return "", debug
+            return "", ""
 
-        # 0) try filenames in question
+        # filename hint
         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
+        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 ""
 
-        # 1) urls in item
+        # 2) try URLs found in item
         urls = extract_file_urls_from_item(item, api_url=self.api_url)
         for u in urls:
-            fp, dbg = download_from_url(u)
-            if DEBUG_ATTACH:
-                debug += "\n".join(dbg) + "\n"
+            fp, dbg2 = _try_download_urls([u], tag=filename_hint)
+            debug_lines.extend(dbg2)
             if fp:
                 ans = self._solve_from_file(q, fp)
                 if ans:
-                    return sanitize_answer(ans), debug
+                    return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
 
-        # 2) ids in item
+        # 3) try ids in item
         file_ids = extract_file_ids_from_item(item)
-        for fid in file_ids:
-            fp, dbg = download_scoring_file(fid, api_url=self.api_url)
-            if DEBUG_ATTACH:
-                debug += "\n".join(dbg) + "\n"
+        # fallback: task_id also as id
+        tid = item.get("task_id")
+        if isinstance(tid, str) and tid:
+            file_ids.append(tid)
+
+        # dedup ids
+        seen = set()
+        file_ids2 = []
+        for x in file_ids:
+            if x and x not in seen:
+                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)
             if fp:
                 ans = self._solve_from_file(q, fp)
                 if ans:
-                    return sanitize_answer(ans), debug
-
-        # 3) fallback: task_id as file_id
-        task_id = item.get("task_id")
-        if isinstance(task_id, str) and task_id:
-            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 sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
 
-        return "", debug.strip()
+        # 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))
+            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()
+
         if ("attached excel file" in ql) or (suf in [".xlsx", ".xls"]):
             return solve_excel_food_sales(fp)
+
         if ("attached python code" in ql) or (suf in [".py", ".txt"]):
             return solve_python_final_numeric(fp)
-        # mp3 tasks intentionally skipped (no audio pipeline)
+
+        # mp3 tasks are intentionally skipped (no audio pipeline)
         return None
 
 
@@ -508,11 +646,7 @@ def run_and_submit_all(profile: gr.OAuthProfile | None = None):
         submit_url = f"{api_url}/submit"
 
         agent = BasicAgent(api_url=api_url)
-        agent_code = (
-            f"https://huggingface.co/spaces/{space_id}/tree/main"
-            if space_id
-            else "https://huggingface.co/spaces/UNKNOWN/tree/main"
-        )
+        agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main" if space_id else "https://huggingface.co/spaces/UNKNOWN/tree/main"
 
         r = requests.get(questions_url, timeout=45)
         r.raise_for_status()
@@ -534,30 +668,16 @@ def run_and_submit_all(profile: gr.OAuthProfile | None = None):
 
             if isinstance(submitted_answer, str) and submitted_answer.strip() == "":
                 skipped += 1
-                results_log.append({
-                    "Task ID": task_id,
-                    "Question": question_text,
-                    "Submitted Answer": "SKIPPED",
-                    "Debug": debug if DEBUG_ATTACH else ""
-                })
+                results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": "SKIPPED", "Debug": debug})
                 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,
-                "Debug": debug if DEBUG_ATTACH else ""
-            })
+            results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer, "Debug": debug})
 
         if not answers_payload:
-            return "⚠️ 全部 SKIPPED（代表目前沒有穩定可解題，或附件抓不到）。", pd.DataFrame(results_log)
+            return "⚠️ 全部 SKIPPED（目前沒有穩定可解題，或附件仍抓不到）。", pd.DataFrame(results_log)
 
-        submission_data = {
-            "username": username.strip(),
-            "agent_code": agent_code,
-            "answers": answers_payload,
-        }
+        submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
 
         r2 = requests.post(submit_url, json=submission_data, timeout=180)
         r2.raise_for_status()
@@ -584,20 +704,11 @@ 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(
-        """
-1) Login  
-2) Click **Run Evaluation & Submit All Answers**  
-
-✅ This version adds **Debug** column for attachment tasks, showing tried URLs and status codes.
-"""
-    )
-
+    gr.Markdown("✅ This version tries **base64-in-item** + **openapi discovery** + more HF Space download patterns.\n\nDebug column shows what happened.")
     gr.LoginButton()
     run_button = gr.Button("Run Evaluation & Submit All Answers")
     status_output = gr.Textbox(label="Run Status / Submission Result", lines=14, interactive=False)
     results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
-
     run_button.click(fn=run_and_submit_all, outputs=[status_output, results_table])
 
 if __name__ == "__main__":