Final_Assignment_Template

Sleeping

App Files Files Community

johnnychiang commited on Jan 9

Commit

70ff4fd

verified ·

1 Parent(s): 5b1f03f

Update app.py

Browse files

Files changed (1) hide show

app.py +77 -99

app.py CHANGED Viewed

@@ -5,7 +5,6 @@ import json
 import math
 import base64
 import traceback
-from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
@@ -30,21 +29,13 @@ def _http_get(url: str, timeout: int = 30, stream: bool = False) -> requests.Res
 def _looks_like_html(b: bytes) -> bool:
-    head = (b or b"")[:400].lower()
     return (b"<!doctype html" in head) or (b"<html" in head) or (b"<head" in head) or (b"<body" in head)
 def _safe_filename_from_headers(resp: requests.Response, fallback: str) -> str:
     cd = resp.headers.get("content-disposition", "")
-    # filename*=UTF-8''xxx or filename="xxx"
-    m = re.search(r"filename\*=(?:UTF-8'')?([^;]+)", cd, flags=re.I)
-    if m:
-        name = m.group(1).strip().strip('"').strip("'")
-        name = name.split("/")[-1].split("\\")[-1]
-        if name:
-            return name
-    m = re.search(r'filename="?([^";]+)"?', cd, flags=re.I)
     if m:
         name = m.group(1).strip().strip('"').strip("'")
         name = name.split("/")[-1].split("\\")[-1]
@@ -56,9 +47,7 @@ def _safe_filename_from_headers(resp: requests.Response, fallback: str) -> str:
         return fallback + ".xlsx"
     if "audio" in ct or "mpeg" in ct or "mp3" in ct:
         return fallback + ".mp3"
-    if "python" in ct:
-        return fallback + ".py"
-    if "text" in ct:
         return fallback + ".txt"
     return fallback
@@ -106,6 +95,7 @@ def extract_file_ids_from_item(item: Dict[str, Any]) -> List[str]:
                         if isinstance(vv, str) and vv:
                             ids.append(vv)
     seen = set()
     out: List[str] = []
     for x in ids:
@@ -164,23 +154,16 @@ def extract_filenames_from_question(q: str) -> List[str]:
 def url_quote_filename(name: str) -> str:
-    # minimal url-encoding for spaces only
-    return (name or "").replace(" ", "%20")
 # -----------------------------
-# Download helpers (FIXED streaming)
 # -----------------------------
 def _save_stream_to_tmp(resp: requests.Response, file_tag: str) -> Optional[Path]:
-    """
-    Correct way:
-    - read first chunk from iter_content (not resp.raw.read)
-    - if HTML -> abort
-    - else write first chunk + rest
-    """
     try:
-        it = resp.iter_content(chunk_size=64 * 1024)
-        first = next(it, b"")
         if not first:
             return None
         if _looks_like_html(first):
@@ -193,7 +176,7 @@ def _save_stream_to_tmp(resp: requests.Response, file_tag: str) -> Optional[Path
         with open(out_path, "wb") as f:
             f.write(first)
-            for chunk in it:
                 if chunk:
                     f.write(chunk)
@@ -222,13 +205,14 @@ def _try_download_urls(urls: List[str], tag: str) -> Tuple[Optional[Path], List[
 # -----------------------------
-# Base64-in-item extraction (backup)
 # -----------------------------
 _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
@@ -300,7 +284,7 @@ def extract_base64_files_from_item(item: Any, filename_hint: str) -> Tuple[List[
 # -----------------------------
-# Deterministic solvers (your correct ones)
 # -----------------------------
 def solve_reversed_sentence(q: str) -> Optional[str]:
     if "rewsna eht sa" in q and '"tfel"' in q:
@@ -341,7 +325,6 @@ def solve_excel_food_sales(file_path: Path) -> Optional[str]:
         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:
@@ -351,14 +334,8 @@ def solve_excel_food_sales(file_path: Path) -> Optional[str]:
             return None
         df = pd.concat(frames, ignore_index=True)
-        # pick numeric sales column
-        for c in df.columns:
-            if df[c].dtype == object:
-                continue
-            # allow numeric
         numeric_cols = [c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])]
         if not numeric_cols:
-            # try coercion
             for c in df.columns:
                 df[c] = pd.to_numeric(df[c], errors="ignore")
             numeric_cols = [c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])]
@@ -466,18 +443,16 @@ class BasicAgent:
             except Exception:
                 pass
-        # attachment tasks?
-        is_attachment_task = any(k in ql for k in ["attached", ".mp3", ".xlsx", ".xls", ".py"])
         if not is_attachment_task:
             return "", ""
         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 (file_name or "attachment")
-        fn_q = url_quote_filename(filename_hint)
-        # 0) detail endpoints
         detail_candidates = [
             f"{self.api_url}/question/{task_id}",
             f"{self.api_url}/questions/{task_id}",
@@ -508,7 +483,7 @@ class BasicAgent:
                     if ans:
                         return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
-        # 2) urls in json
         for src_name, src in [("DETAIL", detail_json), ("ITEM", item)]:
             if src:
                 urls = extract_file_urls_from_item(src, api_url=self.api_url)
@@ -520,48 +495,40 @@ class BasicAgent:
                         if ans:
                             return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
-        # 3) filename patterns (MOST IMPORTANT)
-        # try with item file_name first, else filename_hint
-        fn_core = url_quote_filename(file_name or filename_hint or "attachment")
-        candidates = [
-            # direct
-            f"{self.api_url}/static/{fn_core}",
-            f"{self.api_url}/files/{fn_core}",
-            f"{self.api_url}/assets/{fn_core}",
-            f"{self.api_url}/media/{fn_core}",
-            f"{self.api_url}/raw/{fn_core}",
-            f"{self.api_url}/api/static/{fn_core}",
-            f"{self.api_url}/api/files/{fn_core}",
-            f"{self.api_url}/api/assets/{fn_core}",
-            f"{self.api_url}/api/media/{fn_core}",
-            # task_id + filename (very common)
-            f"{self.api_url}/files/{task_id}/{fn_core}",
-            f"{self.api_url}/files/{task_id}/download/{fn_core}",
-            f"{self.api_url}/download/{task_id}/{fn_core}",
-            f"{self.api_url}/api/files/{task_id}/{fn_core}",
-            f"{self.api_url}/api/download/{task_id}/{fn_core}",
-            # query style
-            f"{self.api_url}/download?task_id={task_id}&file_name={fn_core}",
-            f"{self.api_url}/download?task_id={task_id}&filename={fn_core}",
-            f"{self.api_url}/api/download?task_id={task_id}&file_name={fn_core}",
-            f"{self.api_url}/api/download?task_id={task_id}&filename={fn_core}",
-        ]
-        fp, dbg3 = _try_download_urls(candidates, tag=(file_name or filename_hint))
-        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 ""
         # 4) id-based fallback
         file_ids = extract_file_ids_from_item(item)
         if task_id:
             file_ids.append(task_id)
         seen = set()
         file_ids2 = []
         for x in file_ids:
@@ -570,7 +537,7 @@ class BasicAgent:
                 seen.add(x)
         for fid in file_ids2:
-            candidates2 = [
                 f"{self.api_url}/files/{fid}",
                 f"{self.api_url}/files/{fid}/download",
                 f"{self.api_url}/file/{fid}",
@@ -590,10 +557,10 @@ class BasicAgent:
                 f"{self.api_url}/download?file_id={fid}",
                 f"{self.api_url}/api/download?file_id={fid}",
             ]
-            fp2, dbg4 = _try_download_urls(candidates2, tag=filename_hint)
             debug_lines.extend(dbg4)
-            if fp2:
-                ans = self._solve_from_file(q, fp2)
                 if ans:
                     return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
@@ -612,28 +579,26 @@ class BasicAgent:
         suf = fp.suffix.lower()
         ql = q.lower()
-        if ("excel" in ql) or (suf in [".xlsx", ".xls"]):
             return solve_excel_food_sales(fp)
-        if ("python" in ql) or (suf in [".py", ".txt"]):
             return solve_python_final_numeric(fp)
-        # mp3 你還沒做語音辨識，就先 None
         return None
 # -----------------------------
 # Runner
 # -----------------------------
-def run_and_submit_all(profile: Any = None):
     try:
         space_id = os.getenv("SPACE_ID", "").strip()
-        username = None
-        if profile is not None:
-            username = getattr(profile, "username", None)
-        if not username:
             return "❌ 沒拿到登入資訊。請先按 Login，再按 Run。", None
         api_url = DEFAULT_API_URL
@@ -663,11 +628,15 @@ def run_and_submit_all(profile: Any = 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})
                 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 not answers_payload:
             return "⚠️ 全部 SKIPPED（目前沒有穩定可解題，或附件仍抓不到）。", pd.DataFrame(results_log)
@@ -679,7 +648,7 @@ def run_and_submit_all(profile: Any = None):
         result_data = r2.json()
         final_status = (
-            f"✅ Submission Successful!\n"
             f"User: {result_data.get('username')}\n"
             f"Overall Score: {result_data.get('score', 'N/A')}% "
             f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
@@ -693,17 +662,26 @@ def run_and_submit_all(profile: Any = None):
         tb = traceback.format_exc()
         return f"❌ Runtime Error:\n{e}\n\n--- Traceback ---\n{tb}", None
 # -----------------------------
-# UI
 # -----------------------------
 with gr.Blocks() as demo:
     gr.Markdown("# Basic Agent Evaluation Runner (No Paid Model)")
-    gr.Markdown("✅ Try: **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)
     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__":
-    demo.launch(debug=True, share=False, show_error=True)

 import math
 import base64
 import traceback
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple
 def _looks_like_html(b: bytes) -> bool:
+    head = b[:400].lower()
     return (b"<!doctype html" in head) or (b"<html" in head) or (b"<head" in head) or (b"<body" in head)
 def _safe_filename_from_headers(resp: requests.Response, fallback: str) -> str:
     cd = resp.headers.get("content-disposition", "")
+    m = re.search(r'filename\*?="?([^";]+)"?', cd, flags=re.I)
     if m:
         name = m.group(1).strip().strip('"').strip("'")
         name = name.split("/")[-1].split("\\")[-1]
         return fallback + ".xlsx"
     if "audio" in ct or "mpeg" in ct or "mp3" in ct:
         return fallback + ".mp3"
+    if "text" in ct or "python" in ct:
         return fallback + ".txt"
     return fallback
                         if isinstance(vv, str) and vv:
                             ids.append(vv)
+    # dedup
     seen = set()
     out: List[str] = []
     for x in ids:
 def url_quote_filename(name: str) -> str:
+    # minimal url-encoding for spaces
+    return name.replace(" ", "%20")
 # -----------------------------
+# Download helpers
 # -----------------------------
 def _save_stream_to_tmp(resp: requests.Response, file_tag: str) -> Optional[Path]:
     try:
+        first = resp.raw.read(4096)
         if not first:
             return None
         if _looks_like_html(first):
         with open(out_path, "wb") as f:
             f.write(first)
+            for chunk in resp.iter_content(chunk_size=1024 * 64):
                 if chunk:
                     f.write(chunk)
 # -----------------------------
+# Base64-in-item extraction (備用)
 # -----------------------------
 _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
 # -----------------------------
+# Deterministic solvers (你已經答對的)
 # -----------------------------
 def solve_reversed_sentence(q: str) -> Optional[str]:
     if "rewsna eht sa" in q and '"tfel"' in q:
         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:
             return None
         df = pd.concat(frames, ignore_index=True)
         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:
                 df[c] = pd.to_numeric(df[c], errors="ignore")
             numeric_cols = [c for c in df.columns if pd.api.types.is_numeric_dtype(df[c])]
             except Exception:
                 pass
+        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 "", ""
         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 (file_name or "attachment")
+        # 0) 嘗試抓題目詳情 JSON
         detail_candidates = [
             f"{self.api_url}/question/{task_id}",
             f"{self.api_url}/questions/{task_id}",
                     if ans:
                         return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
+        # 2) url strings
         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 ans:
                             return sanitize_answer(ans), "\n".join(debug_lines) if DEBUG_ATTACH else ""
+        # 3) file_name 組路徑（重點）
+        if file_name:
+            fn_q = url_quote_filename(file_name)
+            fn_candidates = [
+                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}",
+                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}",
+                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 ""
         # 4) id-based fallback
         file_ids = extract_file_ids_from_item(item)
         if task_id:
             file_ids.append(task_id)
         seen = set()
         file_ids2 = []
         for x in file_ids:
                 seen.add(x)
         for fid in file_ids2:
+            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?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 ""
         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 / youtube 類題目前先不做（要加 whisper/yt-dlp 才能真的解）
         return None
 # -----------------------------
 # Runner
 # -----------------------------
+def run_and_submit_all(profile: Optional[gr.OAuthProfile]):
     try:
         space_id = os.getenv("SPACE_ID", "").strip()
+        if profile and getattr(profile, "username", None):
+            username = profile.username
+        else:
             return "❌ 沒拿到登入資訊。請先按 Login，再按 Run。", None
         api_url = DEFAULT_API_URL
             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}
+                )
                 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 not answers_payload:
             return "⚠️ 全部 SKIPPED（目前沒有穩定可解題，或附件仍抓不到）。", pd.DataFrame(results_log)
         result_data = r2.json()
         final_status = (
+            "✅ Submission Successful!\n"
             f"User: {result_data.get('username')}\n"
             f"Overall Score: {result_data.get('score', 'N/A')}% "
             f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
         tb = traceback.format_exc()
         return f"❌ Runtime Error:\n{e}\n\n--- Traceback ---\n{tb}", None
 # -----------------------------
+# UI (OAuth profile 正確傳遞版)
 # -----------------------------
 with gr.Blocks() as demo:
     gr.Markdown("# Basic Agent Evaluation Runner (No Paid Model)")
+    gr.Markdown("✅ This version tries: **detail endpoints** + **file_name path patterns** + url/base64 scan.\n\nDebug欄會顯示嘗試過哪些網址。")
+    # ✅ 用 State 存 OAuthProfile，避免「畫面登入但 callback 收不到 profile」
+    profile_state = gr.State(value=None)
+    login_btn = gr.LoginButton()
+    login_btn.click(fn=lambda p: p, outputs=profile_state)
     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, inputs=profile_state, outputs=[status_output, results_table])
 if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860, debug=True)