#!/usr/bin/env python3 """ quiz server — 靜態檔案 + /api/state 跨裝置同步（token 驗證）+ /api/related-slides 投影片查找 + /api/slides/index、/api/slides/state（slide viewer 後端） + /api/slides/chapter-pdf（章節 PDF 合成 + byte-range 串流） """ import json, os, re, glob, time, io, hashlib, threading from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler from urllib.parse import urlparse, parse_qs BASE_DIR = os.path.dirname(__file__) # === Step 2: DATA_DIR refactor === # 可變狀態（state / cache / log）寫入 DATA_DIR，env 沒設 = 落回 BASE_DIR 維持 staging 行為。 # Zeabur 部署掛 persistent volume 到 /data 然後 DATA_DIR=/data。 DATA_DIR = os.environ.get('DATA_DIR', BASE_DIR) if DATA_DIR != BASE_DIR: os.makedirs(DATA_DIR, exist_ok=True) STATE_FILE = os.path.join(DATA_DIR, 'quiz_state.json') SLIDE_STATE_FILE = os.path.join(DATA_DIR, 'slide_state.json') BOSS_REVIEW_FILE = os.path.join(DATA_DIR, 'boss_review.json') CHAPTER_PDF_CACHE_DIR = os.path.join(DATA_DIR, '_chapter_pdf_cache') # 教材解析 read-only source，留 BASE_DIR（要包進 Docker image） EXPLANATIONS_FILE = os.path.join(BASE_DIR, 'quiz_explanations.json') TOKEN = 'iO3_UTruvI9QQyhrEk7Gbg' # === Step 1: R2 圖檔 redirect === # Cloudflare R2 public base URL（env 沒設 = 走本機 fs，dev/staging 用） # 範例：https://pub-bae6edd1fc614bd9b2d12d983a045aac.r2.dev R2_PUBLIC_BASE = os.environ.get('R2_PUBLIC_BASE', '').rstrip('/') # L-code 前綴 → R2 bucket prefix（產品線分流，跟 tier 名分開） # L1xxxx = 初級 → basic/，L2xxxx = 中級 → mid/，L3xxxx = 高級 → adv/（未來） _R2_LEVEL_PREFIX = {'1': 'basic', '2': 'mid', '3': 'adv'} # 章節 images 路徑：抓 L-code + filename # 例 /科目一_學習指引_分章_v4/output4/L21101_電腦視覺/images/L21101_p01_xxx.png _R2_IMAGE_PATH_RE = re.compile(r'/(L\d{5})_[^/]*/images/([^/?#]+\.png)$') # 通用 L-code 抓取（含 recap/mindmap 等非 images 路徑），tier check 用 _LCODE_GENERIC_RE = re.compile(r'/(L\d{5})(?:[_/]|\.png$)') def r2_url_for_path(path): """把 server URL path 轉成 R2 public URL；無法 map 回 None。 僅處理 LXXXXX_*/images/*.png 結構；recap/mindmap 等其他路徑回 None 走 fs fallback。 """ if not R2_PUBLIC_BASE: return None m = _R2_IMAGE_PATH_RE.search(path) if not m: return None lcode, fname = m.group(1), m.group(2) prefix = _R2_LEVEL_PREFIX.get(lcode[1:2]) if not prefix: return None return '{}/{}/{}/{}'.format(R2_PUBLIC_BASE, prefix, lcode, fname) # === Step 2.5: Tier-based access control + Ghost Members 認證（cookie passthrough）=== # # AUTH_MODE： # token (default) = 維持 ?token=... query 認證，staging/internal 用 # ghost = Ghost cookie 反向驗 + tier 比對，prod 上架用 # disabled = 全開（local dev 用） # # TEACHER_MODE： # enabled (default) = 教師端 endpoint 開放（boss-review / slide-state 寫入 / all_versions=1 / prompts.md） # disabled = 全部 403（prod 切此） # # 環境變數： # AUTH_MODE=token|ghost|disabled # TEACHER_MODE=enabled|disabled # GHOST_URL=https://aiondaily.win (僅 AUTH_MODE=ghost 用；Ghost 跟 server 同 origin，cookie 共享) # PAYWALL_URL_BASE=https://aiondaily.win (root；redirect 拼 ?need=) # # 認證機制（Sein 2026-05-24 spec）： # Ghost 6.x 沒有 issue JWT 端點，session cookie `ghost-members-ssr` 是 koa-session 不是 JWT。 # 改走 cookie passthrough：拿 caller 整個 Cookie header 反向 hit GHOST_URL/members/api/member， # 200 = logged-in（payload 含 tiers），其餘 = 未登入或無效。 # 60s in-process cache 避免 hammering Ghost。 import fnmatch AUTH_MODE = os.environ.get('AUTH_MODE', 'token').strip().lower() TEACHER_MODE = os.environ.get('TEACHER_MODE', 'enabled').strip().lower() GHOST_URL = os.environ.get('GHOST_URL', '').rstrip('/') PAYWALL_URL_BASE = os.environ.get('PAYWALL_URL_BASE', '').rstrip('/') # Tier → L-code fnmatch patterns（鎖死，未來新章節按 L-code 自動歸 tier） TIER_RULES = { 'basic-s1': ['L11???'], # L11000-L11999 初級科目一 'mid-s1': ['L21???'], # L21xxx 中級科目一 'mid-s3': ['L23???'], # L23xxx 中級科目三 'all-access': ['*'], # 全套通吃 } # 教師端 endpoint（prod 上架時 disable） TEACHER_API_PATHS = { '/api/slides/notes', # boss_review GET/POST '/api/slides/state', # slide_state POST } TEACHER_PATH_SUFFIXES = ('prompts.md', 'HANDOFF.md', 'REVIEW.md', 'PROGRESS.md') TEACHER_QUERY_FLAGS = ('all_versions',) # ?all_versions=1 # 需要本機 PNG bytes 才能跑的 endpoint（Zeabur sparse-mirror image 沒真檔，prod 時一律 disable） # chapter-pdf 內部 code=RECAP / code=MINDMAP 也走這條 PIL 合成，封掉這條三種 PDF 一起擋。 # 學員端 UI 的「下載章節 PDF」button 在 prod theme 要同步藏掉避免拿到 403。 PDF_GEN_API_PATHS = { '/api/slides/chapter-pdf', } def tier_allows_lcode(tiers, lcode): """member.tiers 是否允許看此 L-code""" for tier in tiers: for pat in TIER_RULES.get(tier, []): if fnmatch.fnmatchcase(lcode, pat): return True return False def is_teacher_request(parsed): """判斷是否為教師端 request（path/suffix/query 任一命中）， 含 PDF_GEN_API_PATHS（prod sparse-mirror 沒 PNG 跑不了 PIL 合成）。 """ if parsed.path in TEACHER_API_PATHS or parsed.path in PDF_GEN_API_PATHS: return True if any(parsed.path.endswith(s) for s in TEACHER_PATH_SUFFIXES): return True if parsed.query: q = parse_qs(parsed.query) for flag in TEACHER_QUERY_FLAGS: if q.get(flag, ['0'])[0] in ('1', 'true', 'yes'): return True return False def block_teacher_if_disabled(handler, parsed): """TEACHER_MODE=disabled 時擋教師端 / PDF 合成 request；回 True = 已擋下""" if TEACHER_MODE == 'disabled' and is_teacher_request(parsed): handler.send_response(403) handler.send_header('Content-Type', 'application/json') handler.send_header('Access-Control-Allow-Origin', '*') handler.end_headers() handler.wfile.write(b'{"error":"endpoint disabled in prod"}') return True return False # --- Ghost Members cookie passthrough --- _MEMBER_CACHE = {} # cookie_header → {'data': member_dict|None, 'exp': ts} _MEMBER_CACHE_TTL = 60 # 秒；Ghost tier 變更 60s 內生效 def verify_ghost_member(cookie_header): """拿 caller 的整個 Cookie header 反向 hit Ghost /members/api/member。 回 member dict（含 tiers）或 None（未登入 / cookie 失效 / Ghost 不通）。 60s in-process cache（key = 整段 cookie header），避免 hammering Ghost。 """ if not cookie_header or not GHOST_URL: return None now = time.time() cached = _MEMBER_CACHE.get(cookie_header) if cached and cached['exp'] > now: return cached['data'] try: import requests resp = requests.get( GHOST_URL + '/members/api/member', headers={'Cookie': cookie_header, 'Accept': 'application/json'}, timeout=3, allow_redirects=False, ) data = None if resp.status_code == 200: try: data = resp.json() except Exception: data = None # 204 / 401 / 其他 → data 留 None（也 cache，避免反覆打） _MEMBER_CACHE[cookie_header] = {'data': data, 'exp': now + _MEMBER_CACHE_TTL} # 簡單 LRU：cache 超過 2000 條清最舊一半 if len(_MEMBER_CACHE) > 2000: items = sorted(_MEMBER_CACHE.items(), key=lambda kv: kv[1]['exp']) for k, _ in items[:1000]: _MEMBER_CACHE.pop(k, None) return data except Exception as e: print('[ghost-member] verify failed:', e) return None def _missing_tier_for_lcode(lcode): """L-code → 該訂閱的 tier slug（給 paywall ?need= 用）""" if lcode.startswith('L11'): return 'basic-s1' if lcode.startswith('L21'): return 'mid-s1' if lcode.startswith('L23'): return 'mid-s3' return 'all-access' def redirect_to_signin(handler): """未登入 → Ghost Portal signin""" target = (GHOST_URL + '/#/portal/signin') if GHOST_URL else '/' handler._skip_default_headers = True handler.send_response(302) handler.send_header('Location', target) handler.send_header('Cache-Control', 'no-store') handler.end_headers() def redirect_to_paywall(handler, missing_tier): """已登入但 tier 不足 → PAYWALL_URL_BASE/?need=，theme 自己渲染 CTA""" base = PAYWALL_URL_BASE or GHOST_URL or '/' sep = '&' if ('?' in base) else '?' target = '{}{}need={}'.format(base, sep, missing_tier) handler._skip_default_headers = True handler.send_response(302) handler.send_header('Location', target) handler.send_header('Cache-Control', 'no-store') handler.end_headers() def check_auth(handler, lcode=None): """統一鑑權入口。 回 True 通過 / False 已 send response（401/302）。 AUTH_MODE=token: 原 check_token 行為 AUTH_MODE=ghost: cookie passthrough → member.tiers → tier 比對 AUTH_MODE=disabled: 全通過 """ if AUTH_MODE == 'disabled': return True if AUTH_MODE == 'ghost': cookie_header = handler.headers.get('Cookie', '') member = verify_ghost_member(cookie_header) if not member: redirect_to_signin(handler) return False if lcode: tiers = [t.get('slug') for t in (member.get('tiers') or []) if isinstance(t, dict) and t.get('slug')] if not tier_allows_lcode(tiers, lcode): redirect_to_paywall(handler, _missing_tier_for_lcode(lcode)) return False return True # default: token mode return check_token(handler) # PDF 合成鎖（避免同章節並發合成兩次） _PDF_GEN_LOCKS = {} _PDF_GEN_LOCKS_LOCK = threading.Lock() # boss_review.json 寫入鎖（避免並發 race） _BOSS_REVIEW_LOCK = threading.Lock() SUBJECT_NAMES = {'s1': '科目一', 's3': '科目三'} # 科目 → prompts.md 目錄 SUBJ_PROMPTS_DIRS = { 's1': os.path.join(BASE_DIR, '科目一_學習指引_分章_v4', 'output4'), 's3': os.path.join(BASE_DIR, '科目三_學習指引', 'output3'), } def check_token(handler): parsed = urlparse(handler.path) params = parse_qs(parsed.query) t = params.get('token', [None])[0] if t != TOKEN: handler.send_response(401) handler.send_header('Content-Type', 'application/json') handler.end_headers() handler.wfile.write(b'{"error":"unauthorized"}') return False return True # ---- /api/related-slides 邏輯 ---- # Cache：第一次讀 prompts.md 後存到記憶體 _PROMPTS_CACHE = {} def parse_prompts_md(path): """從 prompts.md 抽出每個 Prompt 區段：title, terms, output_filename""" try: with open(path, 'r', encoding='utf-8') as f: text = f.read() except Exception: return [] # 章節 code（從路徑取 L21101 / L23303 等） m = re.search(r'(L\d{5})_', path) code = m.group(1) if m else 'L?????' sections = [] # 用 ## Prompt 切段（每個 Prompt 區段一張 slide） blocks = re.split(r'\n## Prompt ', text) for block in blocks[1:]: # title 在第一行（含 P01 / P22b 之類） first_line = block.split('\n', 1)[0] title_m = re.match(r'P([\d\w\-]+?)[│｜|]\s*(.+)', first_line) if not title_m: continue page_id = 'P' + title_m.group(1).upper() section_title = title_m.group(2).strip() # 教材原詞清單（從「教材原詞必須清楚出現：xxx」段落） terms_m = re.search(r'教材原詞必須清楚出現[：:]([^\n]+)', block) terms_str = terms_m.group(1) if terms_m else '' # 用 / 、 等分隔切詞 terms = [] for piece in re.split(r'[、，,/／\u3001]', terms_str): piece = piece.strip().rstrip('。') if piece and len(piece) >= 2: terms.append(piece) # 輸出檔名（建議輸出檔名：`L21101_p01_xxx.png`） fn_m = re.search(r'建議輸出檔名[：:]\s*`?([^`\n]+\.png)`?', block) filename = fn_m.group(1).strip() if fn_m else '' # 是否為章節導讀（權重要打折，避免被當第一名） is_overview = bool(re.search(r'章節導讀|前言|章節導覽|總覽|overview|sec\d+_', section_title, re.IGNORECASE) or re.search(r'章節導讀|前言與章節導覽|總覽', filename)) # filename 也檢查 sec\d_ 開頭 if filename and re.search(r'_p\d+_sec\d+_', filename) and 'overview' in filename: is_overview = True sections.append({ 'code': code, 'page': page_id, 'title': section_title, 'terms': terms, 'filename': filename, 'is_overview': is_overview, }) return sections def get_subj_slides(subj): """讀該科目所有 prompts.md，回所有 sections（含 cache）""" if subj in _PROMPTS_CACHE: return _PROMPTS_CACHE[subj] base = SUBJ_PROMPTS_DIRS.get(subj) if not base or not os.path.exists(base): return [] all_sections = [] for prompts_path in glob.glob(os.path.join(base, 'L*', 'prompts.md')): chap_dir = os.path.basename(os.path.dirname(prompts_path)) # 確認 images 目錄存在 images_dir = os.path.join(os.path.dirname(prompts_path), 'images') if not os.path.exists(images_dir): continue sections = parse_prompts_md(prompts_path) # 補上 image_path（相對 BASE_DIR，給 quiz HTML 用） rel_subj_dir = os.path.relpath(os.path.dirname(prompts_path), BASE_DIR) for s in sections: if s['filename']: # 找最新版本（_v3 > _v2 > 沒後綴） base_name = re.sub(r'_v\d+\.png$', '', s['filename']).replace('.png', '') candidates = sorted(glob.glob(os.path.join(images_dir, base_name + '*.png')), reverse=True) # 排序後優先 _v 版本，但要找實際存在的 versioned = [c for c in candidates if re.search(r'_v\d+\.png$', c)] no_version = [c for c in candidates if not re.search(r'_v\d+\.png$', c)] actual = (versioned[0] if versioned else (no_version[0] if no_version else None)) if actual: s['image_path'] = os.path.relpath(actual, BASE_DIR) else: s['image_path'] = None else: s['image_path'] = None all_sections.extend([s for s in sections if s.get('image_path')]) _PROMPTS_CACHE[subj] = all_sections return all_sections def score_slide(slide, query_text): """算 slide 與 query_text 命中分數 - 標題命中 +1.5（標題本身就點題，最重要） - 教材原詞命中 +1 - 英文縮寫命中 +0.5 """ score = 0 hits = [] seen = set() # 同一個 hit 字眼只算一次 title = slide.get('title', '') # 標題裡的關鍵詞抽出（移除章節編號 / 中英括號） title_clean = re.sub(r'^[\d.()（）A-Z]+', '', title) title_terms = re.split(r'[—\-/／（）()、,，\s]+', title_clean) for t in title_terms: t = t.strip() if t and len(t) >= 2 and t in query_text and t not in seen: score += 1.5 hits.append(t) seen.add(t) for term in slide.get('terms', []): if term in seen: continue if term in query_text: score += 1 hits.append(term) seen.add(term) continue # 抽 term 裡的英文縮寫單獨比對 eng = re.findall(r'[A-Za-z][A-Za-z\-]+', term) for e in eng: if len(e) >= 3 and e in query_text and e not in seen: score += 0.5 hits.append(e) seen.add(e) break # 章節導讀打折（避免被當第一名） if slide.get('is_overview') and score > 0: score *= 0.6 return score, hits # ---- /api/slides/* slide viewer 邏輯 ---- # Slides index cache（30 秒 TTL）— key 'definitive' / 'all_versions' 分開 cache _SLIDES_INDEX_CACHE = {'definitive': {'data': None, 'timestamp': 0.0}, 'all_versions': {'data': None, 'timestamp': 0.0}} _SLIDES_INDEX_TTL = 30.0 def find_all_image_versions(images_dir, base_filename): """從 base_filename 找所有版本，回 list of (version_int, path)，asc by version。 嚴格只匹配 base_name + '.png' 或 base_name + '_vN.png' — 避免誤抓 base_name + 其他 suffix 的圖。 """ if not base_filename: return [] base_name = re.sub(r'_v\d+\.png$', '', base_filename).replace('.png', '') candidates = glob.glob(os.path.join(images_dir, base_name + '*.png')) versions = [] for c in candidates: name = os.path.basename(c) suffix = name[len(base_name):] if suffix == '.png': versions.append((1, c)) # 無後綴視為 v1 else: m = re.match(r'^_v(\d+)\.png$', suffix) if m: versions.append((int(m.group(1)), c)) versions.sort(key=lambda t: t[0]) return versions def find_definitive_image(images_dir, base_filename): """從 base_filename 找最新版（_v3 > _v2 > 無後綴）""" versions = find_all_image_versions(images_dir, base_filename) if not versions: return None versioned = [v for v in versions if v[0] > 1] return versioned[-1][1] if versioned else versions[-1][1] def parse_chapter_dir_name(dir_name): """從 'L21101_自然語言處理技術與應用' 抽 (code='L21101', title='自然語言處理技術與應用')""" m = re.match(r'^(L\d{5})_(.+)$', dir_name) if not m: return None, dir_name return m.group(1), m.group(2) def build_slides_index(all_versions=False): """掃 SUBJ_PROMPTS_DIRS 產出章節結構。 all_versions=False（預設）：每張 slide 取訂版；同 L-code 多目錄取 sorted asc 後最後者。 all_versions=True（審視模式）：同 L-code 多目錄全部展開（dir_name 區分）； 每張 slide 的所有歷史版本都列為獨立 page。最新版保留原 page_id（兼容 boss_review key）， 舊版本 page_id 加 `__v{N}` 後綴（雙底線避免與 _vN.png 後綴衝突）， title 加「（v{N}）」標示，舊版排在新版之後（時序倒敘 = 最新先看，舊版緊接其後審視）。 """ subjects = [] for subj_id in ('s1', 's3'): base = SUBJ_PROMPTS_DIRS.get(subj_id) if not base or not os.path.exists(base): continue if all_versions: # 同 L-code 多目錄全部保留，按 dir_name asc 列出（v1 章節目錄 → v2 章節目錄） chapters_list = [] for chap_dir in sorted(glob.glob(os.path.join(base, 'L*'))): if not os.path.isdir(chap_dir): continue dir_name = os.path.basename(chap_dir) code, title = parse_chapter_dir_name(dir_name) if not code: continue images_dir = os.path.join(chap_dir, 'images') prompts_path = os.path.join(chap_dir, 'prompts.md') if not os.path.exists(images_dir): continue sections = parse_prompts_md(prompts_path) if os.path.exists(prompts_path) else [] pages = [] for s in sections: if not s.get('filename'): continue all_vers = find_all_image_versions(images_dir, s['filename']) if not all_vers: continue page_id = s['page'].lower() # 找「最新版」 — 與 find_definitive_image 邏輯一致：有 _v 後綴的優先（取最高），其次無後綴 versioned = [v for v in all_vers if v[0] > 1] definitive_v = (versioned[-1][0] if versioned else all_vers[-1][0]) # 排序：最新版在前，舊版倒敘排其後（v3 → v2 → v1）— review 時先看現況再回溯 sorted_vers = sorted(all_vers, key=lambda t: -t[0]) for ver, path in sorted_vers: if ver == definitive_v: pages.append({ 'id': page_id, 'title': s['title'] + (' (現行)' if len(all_vers) > 1 else ''), 'image_path': os.path.relpath(path, BASE_DIR), }) else: pages.append({ 'id': page_id + '__v' + str(ver), 'title': s['title'] + ' (v' + str(ver) + '・舊版)', 'image_path': os.path.relpath(path, BASE_DIR), }) # ---- 孤兒 PNG（在 images_dir 但 prompts.md 沒列）— 拆分前舊主檔之類 ---- # 插在「對應 page hint 的最後一張之後」，例如孤兒 p08 插在 p08a/p08b 後面，方便並排比對 collected = set(p['image_path'] for p in pages) all_pngs = sorted(glob.glob(os.path.join(images_dir, '*.png'))) for png_path in all_pngs: rel = os.path.relpath(png_path, BASE_DIR) if rel in collected: continue fname = os.path.basename(png_path) # 排除 recap/mindmap：它們已被 RECAP/MINDMAP pseudo-chapter 集中收走 if 'recap' in fname.lower() or 'mindmap' in fname.lower(): continue # 從檔名抽 page hint：L21301_p08_feature_processing → p08 pm = re.search(r'_p(\d+\w*)_', fname) page_hint = ('p' + pm.group(1).lower()) if pm else None # 只取數字主編號（p08a → p08）做匹配，孤兒 p08 應插在 p08a/p08b 之後 page_main = re.match(r'^p(\d+)', page_hint).group(0) if (page_hint and re.match(r'^p\d+', page_hint)) else None orphan_entry = { 'id': 'orphan__' + fname.replace('.png', '').lower(), 'title': '🗑️ ' + fname + '（孤兒・不在 prompts.md，可用來與新版並排比對）', 'image_path': rel, } if page_main: # 找 pages 裡 id 開頭 == page_main + (a/b/c/'') 的最後一個 index，insert 在它後面 insert_at = None for idx, p in enumerate(pages): pid = p['id'] # match p08 / p08a / p08b / p08__v2 等同主編號的 page m = re.match(r'^(p\d+)', pid) if m and m.group(1) == page_main: insert_at = idx + 1 if insert_at is not None: pages.insert(insert_at, orphan_entry) continue pages.append(orphan_entry) if not pages: continue chapters_list.append({ 'code': code, 'title': title, 'dir_name': dir_name, 'pages': pages, }) # 按 dir_name asc 排序（同 code 多目錄會自動相鄰 — v1 dir → v2 dir） chapters = sorted(chapters_list, key=lambda c: c['dir_name']) else: # 同 code 後者覆蓋前者（sorted asc 下，v2 在 v1 / 無後綴之後 → 自動選 v2） chapters_by_code = {} for chap_dir in sorted(glob.glob(os.path.join(base, 'L*'))): if not os.path.isdir(chap_dir): continue dir_name = os.path.basename(chap_dir) code, title = parse_chapter_dir_name(dir_name) if not code: continue images_dir = os.path.join(chap_dir, 'images') prompts_path = os.path.join(chap_dir, 'prompts.md') if not os.path.exists(images_dir): continue sections = parse_prompts_md(prompts_path) if os.path.exists(prompts_path) else [] pages = [] for s in sections: if not s.get('filename'): continue actual = find_definitive_image(images_dir, s['filename']) if not actual: continue page_id = s['page'].lower() # 'P01' → 'p01' pages.append({ 'id': page_id, 'title': s['title'], 'image_path': os.path.relpath(actual, BASE_DIR), }) if not pages: continue chapters_by_code[code] = { 'code': code, 'title': title, 'dir_name': dir_name, 'pages': pages, } # 按 code asc 輸出（dict 保序 + 顯式 sort） chapters = sorted(chapters_by_code.values(), key=lambda c: c['code']) subjects.append({ 'id': subj_id, 'name': SUBJECT_NAMES.get(subj_id, subj_id), 'chapters': chapters, }) # 注入 RECAP pseudo-chapter — 合集所有章節 recap 圖按章節順序，當作獨立章節在科目三末尾呈現 try: recap_index = build_recap_index() recap_pages = [] for ch in recap_index.get('chapters', []): for r in ch.get('recaps', []): recap_pages.append({ 'id': 'recap_{}_{}'.format(ch['code'].lower(), r['page_id'].replace('recap_','')), 'title': '{} {}'.format(ch['code'], r.get('title', '')), 'image_path': r['image_path'], }) if recap_pages: # 找科目三 subject 注入；沒有就建獨立 subject s3_subj = next((s for s in subjects if s['id'] == 's3'), None) recap_chapter = { 'code': 'RECAP', 'title': '📌 章節重點速記合集', 'dir_name': 'recap', 'pages': recap_pages, } if s3_subj: s3_subj['chapters'].append(recap_chapter) else: subjects.append({'id':'recap','name':'章節重點速記','chapters':[recap_chapter]}) except Exception as e: print('[slides-index] recap injection failed:', e) # 注入 MINDMAP pseudo-chapter — 合集所有章節 mindmap 圖按章節順序，當作獨立章節在科目三末尾呈現 try: mindmap_index = build_mindmap_index() mindmap_pages = [] for ch in mindmap_index.get('chapters', []): mindmap_pages.append({ 'id': 'mindmap_{}'.format(ch['code'].lower()), 'title': '{} {}'.format(ch['code'], ch.get('title', '')), 'image_path': ch['mindmap_path'], }) if mindmap_pages: # 找科目三 subject 注入；沒有就建獨立 subject s3_subj = next((s for s in subjects if s['id'] == 's3'), None) mindmap_chapter = { 'code': 'MINDMAP', 'title': '🧠 章節心智圖合輯', 'dir_name': 'mindmap', 'pages': mindmap_pages, } if s3_subj: s3_subj['chapters'].append(mindmap_chapter) else: subjects.append({'id':'mindmap','name':'章節心智圖','chapters':[mindmap_chapter]}) except Exception as e: print('[slides-index] mindmap injection failed:', e) return {'subjects': subjects} def get_slides_index(all_versions=False): """有 30 秒 cache 的 slides index — definitive / all_versions 各自分開 cache""" cache_key = 'all_versions' if all_versions else 'definitive' bucket = _SLIDES_INDEX_CACHE[cache_key] now = time.time() if bucket['data'] is not None and (now - bucket['timestamp']) < _SLIDES_INDEX_TTL: return bucket['data'] data = build_slides_index(all_versions=all_versions) bucket['data'] = data bucket['timestamp'] = now return data def read_slide_state(): """讀 slide_state.json；不存在或解析失敗回空殼。 approved 已搬到 boss_review.json _approved_pages，此處不再處理。 """ if not os.path.exists(SLIDE_STATE_FILE): return {'rejected': [], 'review': []} try: with open(SLIDE_STATE_FILE, 'r', encoding='utf-8') as f: d = json.load(f) except Exception: return {'rejected': [], 'review': []} return { 'rejected': list(d.get('rejected', []) or []), 'review': list(d.get('review', []) or []), } # ---- boss_review.json：Boss review 留言（per page）---- def read_boss_review(): """讀 boss_review.json，schema: {: [{id, text, status, ts}, ...]}""" if not os.path.exists(BOSS_REVIEW_FILE): return {} try: with open(BOSS_REVIEW_FILE, 'r', encoding='utf-8') as f: d = json.load(f) return d if isinstance(d, dict) else {} except Exception: return {} def write_boss_review(data): """atomic write: 寫到 .tmp 再 rename""" tmp = BOSS_REVIEW_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) os.replace(tmp, BOSS_REVIEW_FILE) def gen_note_id(): import secrets return 'n_' + str(int(time.time())) + '_' + secrets.token_hex(3) def write_slide_state(payload): """寫 slide_state.json — 只保留 rejected/review（approved 改存 boss_review.json）""" rejected = list(dict.fromkeys(payload.get('rejected', []) or [])) review = list(dict.fromkeys(payload.get('review', []) or [])) data = {'rejected': rejected, 'review': review} with open(SLIDE_STATE_FILE, 'w', encoding='utf-8') as f: json.dump(data, f, ensure_ascii=False, indent=2) return data def get_page_versions(page_key): """page_key 例如 'L23102_p06' → 找該頁所有 PNG 版本 + 關聯 boss_review note。 搜尋邏輯：在科目一 output4/ 跟科目三 output3/ 各章節 images/ 內找 LXXXXX_pXX_*.png """ m = re.match(r'^(L\d{5})_(p[\w]+)$', page_key) if not m: return None chap_code, page_id = m.group(1), m.group(2) # 找該章 images dir — 同 code 多目錄取 sorted 最後者（v2 > v1 > 無後綴） images_dir = None for subj_id, base in SUBJ_PROMPTS_DIRS.items(): if not base or not os.path.exists(base): continue chap_dirs = sorted(glob.glob(os.path.join(base, chap_code + '_*'))) # 從後往前找第一個 valid（含 images/）目錄 for cd in reversed(chap_dirs): if os.path.isdir(cd): _id = os.path.join(cd, 'images') if os.path.exists(_id): images_dir = _id break if images_dir: break if not images_dir or not os.path.exists(images_dir): return {'page_key': page_key, 'versions': []} # 列所有 LXXXXX_pXX_*.png — 嚴格匹配 page_id 後接 _ 或 .（避免 p1 匹配 p10） pattern = os.path.join(images_dir, '{}_{}_*.png'.format(chap_code, page_id)) files = [] for f in glob.glob(pattern): files.append(f) # 也檢查無 suffix 的（直接 LXXXXX_pXX.png 不太可能，跳過） versions = [] for f in files: name = os.path.basename(f) # 從檔名抽版本號（_v? 後綴；無後綴 = v1） vm = re.search(r'_v(\d+)\.png$', name) version = int(vm.group(1)) if vm else 1 rel = os.path.relpath(f, BASE_DIR) versions.append({ 'version': version, 'filename': name, 'path': rel, 'mtime': int(os.path.getmtime(f)), 'size': os.path.getsize(f), }) # 新版本在前 versions.sort(key=lambda v: -v['version']) # 關聯 boss_review note resolution notes_data = read_boss_review() page_notes = notes_data.get(page_key, []) for v in versions: for n in page_notes: res = n.get('resolution') or {} if res.get('actual_filename') == v['filename']: v['note_text'] = n.get('text', '') plan = n.get('heiter_plan') or {} v['plan_summary'] = plan.get('summary', '') v['plan_type'] = plan.get('type', '') v['commit_hash'] = res.get('commit_hash', '') v['verify_note'] = res.get('verify_note', '') v['resolved_ts'] = res.get('resolved_ts', '') break return {'page_key': page_key, 'versions': versions, 'images_dir': os.path.relpath(images_dir, BASE_DIR)} def find_related_slides(subj, query_text, top_n=3): """查 subj 中與 query_text 最相關的 top_n slides 去重：同 code+page 只保留分數最高的版本（v3 > v2 > v1） """ slides = get_subj_slides(subj) scored = [] for s in slides: score, hits = score_slide(s, query_text) if score > 0: scored.append({ 'code': s['code'], 'page': s['page'], 'title': s['title'], 'image_path': s['image_path'], 'score': round(score, 1), 'hits': hits, 'is_overview': s.get('is_overview', False), }) # 去重：同 code+page 只保留 image_path 含 _v 後綴的較新版本 by_key = {} for s in scored: key = (s['code'], s['page']) if key not in by_key: by_key[key] = s else: # 保留分數高的；分數相同則保留 image_path 較新的（含 _v 數字大的） old = by_key[key] new_v = re.search(r'_v(\d+)\.png$', s['image_path'] or '') old_v = re.search(r'_v(\d+)\.png$', old['image_path'] or '') new_vn = int(new_v.group(1)) if new_v else 0 old_vn = int(old_v.group(1)) if old_v else 0 if s['score'] > old['score'] or (s['score'] == old['score'] and new_vn > old_vn): by_key[key] = s deduped = list(by_key.values()) deduped.sort(key=lambda x: -x['score']) return deduped[:top_n] # ---- /api/slides/chapter-pdf（章節合成 PDF） ---- def get_chapter_pdf_path(chap_code, all_versions=False): """根據章節 code 找對應 PDF cache path，需要時生成。 cache key = 該章所有 image path + mtime 的 sha256 前 16 char。 PNG 用 JPEG 壓縮（quality 82, 寬度上限 1920px）後 img2pdf 合成。 all_versions=True → 含舊版/孤兒（review 模式，cache 加 _review 後綴）。 """ index = get_slides_index(all_versions=all_versions) # all_versions=True 模式下，同 code 可能有多個 chap（v1 dir / v2 dir / ...） # 全部收集起來 image_paths 合併，按 dir_name asc 串接（v1 dir → v2 dir） target_chaps = [] for subj in index.get('subjects', []): for chap in subj.get('chapters', []): if chap.get('code') == chap_code: target_chaps.append(chap) if not target_chaps: return None target_chaps.sort(key=lambda c: c.get('dir_name', '')) image_paths = [] for chap in target_chaps: for page in chap.get('pages', []): full = os.path.join(BASE_DIR, page['image_path']) if os.path.exists(full): image_paths.append(full) if not image_paths: return None # 算 cache key fingerprint = '|'.join(p + ':' + str(int(os.path.getmtime(p))) for p in image_paths) h = hashlib.sha256(fingerprint.encode('utf-8')).hexdigest()[:16] if not os.path.exists(CHAPTER_PDF_CACHE_DIR): os.makedirs(CHAPTER_PDF_CACHE_DIR, exist_ok=True) mode_suffix = '_review' if all_versions else '' pdf_path = os.path.join(CHAPTER_PDF_CACHE_DIR, '{}_{}{}.pdf'.format(chap_code, h, mode_suffix)) if os.path.exists(pdf_path): return pdf_path # 拿同章節 lock 避免並發重生（review / definitive 分 lock） lock_key = chap_code + mode_suffix with _PDF_GEN_LOCKS_LOCK: lock = _PDF_GEN_LOCKS.setdefault(lock_key, threading.Lock()) with lock: if os.path.exists(pdf_path): return pdf_path try: from PIL import Image import img2pdf except ImportError as e: print('[pdf] missing dependency:', e) return None jpeg_bytes = [] for p in image_paths: try: img = Image.open(p) if img.mode in ('RGBA', 'P', 'LA'): img = img.convert('RGB') elif img.mode != 'RGB': img = img.convert('RGB') if img.width > 1920: ratio = 1920.0 / img.width img = img.resize((1920, int(img.height * ratio)), Image.LANCZOS) buf = io.BytesIO() img.save(buf, format='JPEG', quality=82, optimize=True) jpeg_bytes.append(buf.getvalue()) except Exception as e: print('[pdf] skip', p, ':', e) if not jpeg_bytes: return None tmp = pdf_path + '.tmp' with open(tmp, 'wb') as f: f.write(img2pdf.convert(jpeg_bytes)) os.rename(tmp, pdf_path) # 清掉同 chap + 同 mode 的舊 cache（不同 mode 不互相吃對方 cache） cleanup_glob = '{}_*{}.pdf'.format(chap_code, mode_suffix) if mode_suffix else '{}_*.pdf'.format(chap_code) for old in glob.glob(os.path.join(CHAPTER_PDF_CACHE_DIR, cleanup_glob)): # 預設模式 cleanup 不能誤掃 _review.pdf if old != pdf_path and (mode_suffix or not old.endswith('_review.pdf')): try: os.remove(old) except Exception: pass print('[pdf] generated', pdf_path, 'size:', os.path.getsize(pdf_path)) return pdf_path # ---- /api/recap/list 章節重點速記頁 ---- # recap subject → 中文 title 對應（從檔名 subject 段抽） _RECAP_SUBJECT_TITLES = { 'regression_summary': '迴歸 N 法重點速記', 'classification_summary': '分類 N 法重點速記', 'unsupervised_summary': '非監督主題重點速記', 'clustering_summary': '聚類 N 法重點速記', 'evaluation_summary': '模型評估重點速記', 'preprocessing_summary': '資料前處理重點速記', } # recap chapter title — 從同 L-code 主章節 dir name 抽，失敗 fallback 此 dict _RECAP_CHAPTER_TITLE_FALLBACK = { 'L23101': '機率統計之機器學習基礎應用', 'L23102': '線性代數之機器學習基礎應用', 'L23103': '數值優化技術與方法', 'L23201': '機器學習原理與技術', 'L23202': '常見機器學習演算法', 'L23203': '深度學習原理與框架', 'L23301': '數據準備與特徵工程', 'L23302': '模型選擇與架構設計', 'L23303': '模型訓練評估與驗證', 'L23304': '模型調整與優化', } def _get_chapter_title_for_recap(code): """從同 L-code 主章節 dir name 抽 title；失敗用 fallback dict""" for subj_id, base in SUBJ_PROMPTS_DIRS.items(): if not base or not os.path.exists(base): continue chap_dirs = sorted(glob.glob(os.path.join(base, code + '_*'))) # 從後往前找（v2 > v1） for cd in reversed(chap_dirs): if os.path.isdir(cd): dir_name = os.path.basename(cd) _code, title = parse_chapter_dir_name(dir_name) if title: # 移除尾巴 v2/v3 等版本後綴 title = re.sub(r'v\d+$', '', title) return title return _RECAP_CHAPTER_TITLE_FALLBACK.get(code, code) def _parse_recap_filename(filename): """從 'L23202_recap_p01_regression_summary.png' parse 出 (code, page_id, subject) 回 None 表示不合法檔名。 """ m = re.match(r'^(L\d{5})_recap_(p\d+[a-z]?)_(.+)\.png$', filename) if not m: return None return m.group(1), m.group(2), m.group(3) def _derive_recap_title(subject): """從 subject 字串對應中文 title；找不到 fallback subject 本身""" return _RECAP_SUBJECT_TITLES.get(subject, subject.replace('_', ' ')) def build_recap_index(): """掃 recap 子資料夾（科目一 output4/recap/ + 科目三 output3/recap/），回章節 list。 支援兩種目錄結構： - `L<5>_/` — 章節內 recap，檔名 `L<5>_recap_p_.png` - `L<5>_L<5>_bridge/` — 跨章節橋接 RECAP，檔名 `recap_.png`（無 L code / page_id 前綴） bridge code 取 first L code + 'B'，sort 上會排在 first chapter 之後、next chapter 之前。 """ recap_bases = [ os.path.join(BASE_DIR, '科目三_學習指引', 'output3', 'recap'), os.path.join(BASE_DIR, '科目一_學習指引_分章_v4', 'output4', 'recap'), ] chapters_by_code = {} for recap_base in recap_bases: if not os.path.exists(recap_base): continue for chap_dir in sorted(glob.glob(os.path.join(recap_base, 'L*'))): if not os.path.isdir(chap_dir): continue dir_name = os.path.basename(chap_dir) # --- 分支一：bridge 目錄（跨章節橋接） --- m_bridge = re.match(r'^(L\d{5})_(L\d{5})_bridge$', dir_name) if m_bridge: first_code, second_code = m_bridge.group(1), m_bridge.group(2) bridge_code = first_code + 'B' # e.g. L21101B → 排在 L21101 之後 L21102 之前 bridge_pngs = sorted(glob.glob(os.path.join(chap_dir, 'recap_*.png'))) for idx, png in enumerate(bridge_pngs, start=1): fn = os.path.basename(png) subject = fn[len('recap_'):-len('.png')] # strip "recap_" + ".png" chap = chapters_by_code.setdefault(bridge_code, { 'code': bridge_code, 'title': '【橋接】{} → {}'.format(first_code, second_code), 'recaps': [], }) chap['recaps'].append({ 'filename': fn, 'page_id': 'recap_bridge_p{:02d}'.format(idx), 'title': _derive_recap_title(subject), 'image_path': os.path.relpath(png, BASE_DIR), }) continue # --- 分支二：標準章節 recap --- for png in sorted(glob.glob(os.path.join(chap_dir, '*_recap_*.png'))): fn = os.path.basename(png) parsed = _parse_recap_filename(fn) if not parsed: continue code, page_id, subject = parsed chap = chapters_by_code.setdefault(code, { 'code': code, 'title': _get_chapter_title_for_recap(code), 'recaps': [], }) chap['recaps'].append({ 'filename': fn, 'page_id': 'recap_' + page_id, 'title': _derive_recap_title(subject), 'image_path': os.path.relpath(png, BASE_DIR), }) # 按 code asc 輸出；每章內 recaps 按 page_id asc sorted_chapters = [] for code in sorted(chapters_by_code.keys()): ch = chapters_by_code[code] ch['recaps'].sort(key=lambda r: r['page_id']) sorted_chapters.append(ch) # 在 L21 / L23 區塊最前面插入「全章節學習地圖」overview chapter def _make_overview_chap(code, title, image_rel): full = os.path.join(BASE_DIR, image_rel) if not os.path.exists(full): return None return { 'code': code, 'title': title, 'recaps': [{ 'filename': os.path.basename(image_rel), 'page_id': 'recap_p00_overview', 'title': '全章節學習地圖', 'image_path': image_rel, }], } l21_ov = _make_overview_chap( 'L21OVERVIEW', '科目一 全章節學習地圖', '科目一_學習指引_分章_v4/output4/L21_overview/images/L21OVERVIEW_pyramid_v1.png') l23_ov = _make_overview_chap( 'L23OVERVIEW', '科目三 全章節學習地圖', '科目三_學習指引/output3/L23_overview/images/L23OVERVIEW_pyramid_v2.png') chapters = [] l21_done = l21_ov is None l23_done = l23_ov is None for ch in sorted_chapters: if not l21_done and ch['code'].startswith('L21'): chapters.append(l21_ov) l21_done = True if not l23_done and ch['code'].startswith('L23'): chapters.append(l23_ov) l23_done = True chapters.append(ch) if not l21_done and l21_ov: chapters.append(l21_ov) if not l23_done and l23_ov: chapters.append(l23_ov) return {'chapters': chapters} def get_recap_pdf_path(): """合成所有章節 recap 一 PDF（按章節順序 + 章內 page_id 順序）。 cache key = 所有 recap image path + mtime 的 sha256 前 16 char。 """ recap_index = build_recap_index() image_paths = [] for chap in recap_index.get('chapters', []): for recap in chap.get('recaps', []): full = os.path.join(BASE_DIR, recap['image_path']) if os.path.exists(full): image_paths.append(full) if not image_paths: return None fingerprint = '|'.join(p + ':' + str(int(os.path.getmtime(p))) for p in image_paths) h = hashlib.sha256(fingerprint.encode('utf-8')).hexdigest()[:16] if not os.path.exists(CHAPTER_PDF_CACHE_DIR): os.makedirs(CHAPTER_PDF_CACHE_DIR, exist_ok=True) pdf_path = os.path.join(CHAPTER_PDF_CACHE_DIR, 'RECAP_ALL_{}.pdf'.format(h)) if os.path.exists(pdf_path): return pdf_path with _PDF_GEN_LOCKS_LOCK: lock = _PDF_GEN_LOCKS.setdefault('__RECAP_ALL__', threading.Lock()) with lock: if os.path.exists(pdf_path): return pdf_path try: from PIL import Image import img2pdf except ImportError as e: print('[recap-pdf] missing dependency:', e) return None jpeg_bytes = [] for p in image_paths: try: img = Image.open(p) if img.mode in ('RGBA', 'P', 'LA'): img = img.convert('RGB') elif img.mode != 'RGB': img = img.convert('RGB') if img.width > 1920: ratio = 1920.0 / img.width img = img.resize((1920, int(img.height * ratio)), Image.LANCZOS) buf = io.BytesIO() img.save(buf, format='JPEG', quality=82, optimize=True) jpeg_bytes.append(buf.getvalue()) except Exception as e: print('[recap-pdf] skip', p, ':', e) if not jpeg_bytes: return None tmp = pdf_path + '.tmp' with open(tmp, 'wb') as f: f.write(img2pdf.convert(jpeg_bytes)) os.rename(tmp, pdf_path) for old in glob.glob(os.path.join(CHAPTER_PDF_CACHE_DIR, 'RECAP_ALL_*.pdf')): if old != pdf_path: try: os.remove(old) except Exception: pass print('[recap-pdf] generated', pdf_path, 'size:', os.path.getsize(pdf_path)) return pdf_path # ---- /api/mindmap/list 章節心智圖頁 ---- # mindmap chapter title — 從同 L-code 主章節 dir name 抽，失敗 fallback 此 dict _MINDMAP_CHAPTER_TITLE_FALLBACK = { 'L23101': '機率統計之機器學習基礎應用', 'L23102': '線性代數之機器學習基礎應用', 'L23103': '數值優化技術與方法', 'L23201': '機器學習原理與技術', 'L23202': '常見機器學習演算法', 'L23203': '深度學習原理與框架', 'L23301': '數據準備與特徵工程', 'L23302': '模型選擇與架構設計', 'L23303': '模型訓練評估與驗證', 'L23304': '模型調整與優化', 'L23401': '數據隱私安全與合規', 'L23402': '演算法偏見與公平性', } def _get_chapter_title_for_mindmap(code): """從同 L-code 主章節 dir name 抽 title；失敗用 fallback dict""" for subj_id, base in SUBJ_PROMPTS_DIRS.items(): if not base or not os.path.exists(base): continue chap_dirs = sorted(glob.glob(os.path.join(base, code + '_*'))) # 從後往前找（v2 > v1） for cd in reversed(chap_dirs): if os.path.isdir(cd): dir_name = os.path.basename(cd) _code, title = parse_chapter_dir_name(dir_name) if title: # 移除尾巴 v2/v3 等版本後綴 title = re.sub(r'v\d+$', '', title) return title return _MINDMAP_CHAPTER_TITLE_FALLBACK.get(code, code) def _parse_mindmap_filename(filename): """從 'L23202_mindmap_overview.png' parse 出 code；不合法回 None""" m = re.match(r'^(L\d{5})_mindmap_overview\.png$', filename) if not m: return None return m.group(1) def build_mindmap_index(): """掃 mindmap 子資料夾（科目三 output3/mindmap/ + 科目一 output4/mindmap/），回章節 list。 每章只有 1 張 mindmap（檔名固定 L{code}_mindmap_overview.png）。 """ mindmap_bases = [ os.path.join(BASE_DIR, '科目三_學習指引', 'output3', 'mindmap'), os.path.join(BASE_DIR, '科目一_學習指引_分章_v4', 'output4', 'mindmap'), ] chapters_by_code = {} for mindmap_base in mindmap_bases: if not os.path.exists(mindmap_base): continue for chap_dir in sorted(glob.glob(os.path.join(mindmap_base, 'L*'))): if not os.path.isdir(chap_dir): continue for png in sorted(glob.glob(os.path.join(chap_dir, '*_mindmap_overview.png'))): fn = os.path.basename(png) code = _parse_mindmap_filename(fn) if not code: continue # 每章只取一張（第一張） if code in chapters_by_code: continue chapters_by_code[code] = { 'code': code, 'title': _get_chapter_title_for_mindmap(code), 'mindmap_path': os.path.relpath(png, BASE_DIR), } # 按 code asc 輸出 chapters = [chapters_by_code[code] for code in sorted(chapters_by_code.keys())] return {'chapters': chapters} def get_mindmap_pdf_path(): """合成所有章節 mindmap 一 PDF（按章節順序，每章 1 page）。 cache key = 所有 mindmap image path + mtime 的 sha256 前 16 char。 """ mindmap_index = build_mindmap_index() image_paths = [] for chap in mindmap_index.get('chapters', []): full = os.path.join(BASE_DIR, chap['mindmap_path']) if os.path.exists(full): image_paths.append(full) if not image_paths: return None fingerprint = '|'.join(p + ':' + str(int(os.path.getmtime(p))) for p in image_paths) h = hashlib.sha256(fingerprint.encode('utf-8')).hexdigest()[:16] if not os.path.exists(CHAPTER_PDF_CACHE_DIR): os.makedirs(CHAPTER_PDF_CACHE_DIR, exist_ok=True) pdf_path = os.path.join(CHAPTER_PDF_CACHE_DIR, 'MINDMAP_ALL_{}.pdf'.format(h)) if os.path.exists(pdf_path): return pdf_path with _PDF_GEN_LOCKS_LOCK: lock = _PDF_GEN_LOCKS.setdefault('__MINDMAP_ALL__', threading.Lock()) with lock: if os.path.exists(pdf_path): return pdf_path try: from PIL import Image import img2pdf except ImportError as e: print('[mindmap-pdf] missing dependency:', e) return None jpeg_bytes = [] for p in image_paths: try: img = Image.open(p) if img.mode in ('RGBA', 'P', 'LA'): img = img.convert('RGB') elif img.mode != 'RGB': img = img.convert('RGB') if img.width > 1920: ratio = 1920.0 / img.width img = img.resize((1920, int(img.height * ratio)), Image.LANCZOS) buf = io.BytesIO() img.save(buf, format='JPEG', quality=82, optimize=True) jpeg_bytes.append(buf.getvalue()) except Exception as e: print('[mindmap-pdf] skip', p, ':', e) if not jpeg_bytes: return None tmp = pdf_path + '.tmp' with open(tmp, 'wb') as f: f.write(img2pdf.convert(jpeg_bytes)) os.rename(tmp, pdf_path) for old in glob.glob(os.path.join(CHAPTER_PDF_CACHE_DIR, 'MINDMAP_ALL_*.pdf')): if old != pdf_path: try: os.remove(old) except Exception: pass print('[mindmap-pdf] generated', pdf_path, 'size:', os.path.getsize(pdf_path)) return pdf_path def parse_range_header(range_header, file_size): """解析 Range: bytes=START-END，回 (start, end)；不合法回 None""" if not range_header: return None m = re.match(r'bytes=(\d*)-(\d*)$', range_header) if not m: return None start_s, end_s = m.group(1), m.group(2) if start_s == '' and end_s == '': return None if start_s == '': # suffix range suffix = int(end_s) if suffix <= 0: return None start = max(0, file_size - suffix) end = file_size - 1 else: start = int(start_s) end = int(end_s) if end_s else file_size - 1 if start > end or start >= file_size: return None end = min(end, file_size - 1) return start, end class Handler(SimpleHTTPRequestHandler): def log_message(self, fmt, *args): # 寫 server access log（debug 用）；改回 pass 可靜音 import sys, time as _t msg = "%s - - [%s] %s\n" % ( self.address_string(), _t.strftime("%Y-%m-%d %H:%M:%S"), fmt % args, ) try: with open(os.path.join(DATA_DIR, 'server-access.log'), 'a', encoding='utf-8') as f: f.write(msg) except Exception: sys.stderr.write(msg) def end_headers(self): # PDF endpoint 自己處理 Cache-Control + Accept-Ranges，跳過全域 no-store if not getattr(self, '_skip_default_headers', False): self.send_header('Cache-Control', 'no-store, no-cache, must-revalidate') super().end_headers() def do_GET(self): parsed = urlparse(self.path) # === Step 2.5: 教師端 endpoint guard（prod 時擋）=== # /api/slides/notes GET 例外：學員端 slides-pdf.html 開機時會打它拿 boss review 留言陣列， # 403 會讓 frontend forEach 炸；prod 改回空陣列保持相容，但 POST 依然走 block。 if TEACHER_MODE == 'disabled' and parsed.path == '/api/slides/notes': self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(b'[]') return if block_teacher_if_disabled(self, parsed): return # === Step 1 + 2.5: 圖檔 tier check + R2 redirect === # 1) ghost mode：抽 L-code 做 tier check，沒過 = redirect to paywall # 2) R2_PUBLIC_BASE 有設且匹配章節 images → 302 redirect 到 R2 # 3) 否則落到 super().do_GET() 走本機 fs（staging 行為） if parsed.path.endswith('.png'): if AUTH_MODE == 'ghost': m_tier = _R2_IMAGE_PATH_RE.search(parsed.path) or _LCODE_GENERIC_RE.search(parsed.path) lcode = m_tier.group(1) if m_tier else None if lcode and not check_auth(self, lcode=lcode): return r2_target = r2_url_for_path(parsed.path) if r2_target: self._skip_default_headers = True self.send_response(302) self.send_header('Location', r2_target) self.send_header('Cache-Control', 'public, max-age=86400') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() return if parsed.path == '/api/state': if not check_auth(self): return data = b'{}' if os.path.exists(STATE_FILE): with open(STATE_FILE, 'rb') as f: data = f.read() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(data) elif parsed.path == '/api/slides/index': if not check_auth(self): return params = parse_qs(parsed.query) all_versions = params.get('all_versions', ['0'])[0] in ('1', 'true', 'yes') data = get_slides_index(all_versions=all_versions) self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8')) elif parsed.path == '/api/slides/state': if not check_auth(self): return data = read_slide_state() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8')) elif parsed.path == '/api/slides/notes': if not check_auth(self): return data = read_boss_review() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8')) elif parsed.path == '/api/slides/versions': if not check_auth(self): return params = parse_qs(parsed.query) page_key = params.get('page_key', [None])[0] if not page_key: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"missing page_key"}') return data = get_page_versions(page_key) if data is None: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"invalid page_key format"}') return self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8')) elif parsed.path == '/api/recap/list': if not check_auth(self): return data = build_recap_index() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8')) elif parsed.path == '/api/mindmap/list': if not check_auth(self): return data = build_mindmap_index() self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8')) elif parsed.path == '/api/slides/chapter-pdf': params = parse_qs(parsed.query) code = params.get('code', [None])[0] # tier check：code 是 L21101 之類則對該 L-code 驗 tier；RECAP/MINDMAP 不限 chap_lcode = code if (code and re.match(r'^L\d{5}$', code)) else None if not check_auth(self, lcode=chap_lcode): return all_versions = params.get('all_versions', ['0'])[0] in ('1', 'true', 'yes') or \ params.get('review', ['0'])[0] in ('1', 'true', 'yes') if not code: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"missing code"}') return # 特殊 code "RECAP" → 合成全部章節 recap 一 PDF if code == 'RECAP': pdf_path = get_recap_pdf_path() elif code == 'MINDMAP': pdf_path = get_mindmap_pdf_path() else: pdf_path = get_chapter_pdf_path(code, all_versions=all_versions) if not pdf_path or not os.path.exists(pdf_path): self.send_response(404) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"chapter pdf not found"}') return file_size = os.path.getsize(pdf_path) # ETag 用 cache filename（含 sha256 hash of png mtimes）— 內容變 = filename 變 = ETag 變 etag = '"' + os.path.basename(pdf_path) + '"' # If-None-Match → 比 ETag → 一致回 304 if_none_match = self.headers.get('If-None-Match', '') self._skip_default_headers = True if if_none_match and etag in if_none_match.split(','): self.send_response(304) self.send_header('ETag', etag) self.send_header('Cache-Control', 'no-cache') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() return range_header = self.headers.get('Range') rng = parse_range_header(range_header, file_size) if rng is not None: start, end = rng length = end - start + 1 self.send_response(206) self.send_header('Content-Type', 'application/pdf') self.send_header('Content-Length', str(length)) self.send_header('Content-Range', 'bytes {}-{}/{}'.format(start, end, file_size)) self.send_header('Accept-Ranges', 'bytes') # no-cache = 可以 cache 但必須 revalidate（帶 If-None-Match）— 內容變立即拿到新版 self.send_header('Cache-Control', 'no-cache') self.send_header('ETag', etag) self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() with open(pdf_path, 'rb') as f: f.seek(start) remaining = length while remaining > 0: chunk = f.read(min(65536, remaining)) if not chunk: break self.wfile.write(chunk) remaining -= len(chunk) else: self.send_response(200) self.send_header('Content-Type', 'application/pdf') self.send_header('Content-Length', str(file_size)) self.send_header('Accept-Ranges', 'bytes') self.send_header('Cache-Control', 'no-cache') self.send_header('ETag', etag) self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() with open(pdf_path, 'rb') as f: while True: chunk = f.read(65536) if not chunk: break self.wfile.write(chunk) else: super().do_GET() def do_POST(self): parsed = urlparse(self.path) # === Step 2.5: 教師端 endpoint guard（prod 時擋）=== if block_teacher_if_disabled(self, parsed): return if parsed.path == '/api/state': if not check_auth(self): return length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length) with open(STATE_FILE, 'wb') as f: f.write(body) self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(b'{"ok":true}') elif parsed.path == '/api/related-slides': if not check_auth(self): return length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode('utf-8') try: req = json.loads(body) except Exception: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"bad_json"}') return subj = req.get('subj', '') qtext = req.get('qtext', '') options = req.get('options', []) answer = req.get('answer', '') top_n = int(req.get('top_n', 3)) # 先讀 explanation（用原 subj），抽「📖 章節來源」段塞進 query 提升命中 explanation = None try: if os.path.exists(EXPLANATIONS_FILE): with open(EXPLANATIONS_FILE, 'r', encoding='utf-8') as f: ex_data = json.load(f) explanation = ex_data.get(subj, {}).get(str(req.get('num', '')), None) except Exception: explanation = None # 抽章節來源段：code 題 `### (📖 )章節來源\n...` / mock 題 `**📖 章節來源**：...` inline source_text = '' source_chap = None if explanation: # heading 形式：## 或 ### 開頭，含/不含 📖 emoji（cover s1/s3 ##、code/mock ###） ms = re.search(r'#{2,3}\s*(?:📖\s*)?章節來源\s*\n+([\s\S]*?)(?=\n#{2,} |\Z)', explanation) if ms: source_text = ms.group(1) else: # inline 形式：**📖 章節來源**：xxx ms = re.search(r'\*\*📖?\s*章節來源\*\*[：:]\s*(.+)', explanation) if ms: source_text = ms.group(1) source_text = re.sub(r'[`*#]', '', source_text) # 抽章節 code（L23103 / L21102 等）做章節 boost mc = re.search(r'L(\d{5})', source_text) if mc: source_chap = 'L' + mc.group(1) # 把題目 + 選項 + 答案 + 章節來源串成 query（章節來源權重高 → 子標題/英文縮寫直接命中） query_parts = [qtext] + (options or []) + [answer] if source_text: query_parts.append(source_text) query = ' '.join(query_parts) # mock_LXXXXX / code_LXXXXX → 自動 route 到對應科目找 slides，但 explanation 鍵保持原 subj slides_subj = subj m = re.match(r'(?:mock|code)_L(\d{5})', subj) if m: code_prefix = m.group(1)[:2] # '21' (科目一) or '23' (科目三) slides_subj = 's1' if code_prefix == '21' else 's3' # 多抓幾張做章節 boost 再 re-sort 取 top_n raw_results = find_related_slides(slides_subj, query, top_n=max(top_n*3, 10)) if source_chap: for r in raw_results: if r.get('code') == source_chap: r['score'] = round(r['score'] * 1.5, 2) raw_results = sorted(raw_results, key=lambda x: -x['score']) results = raw_results[:top_n] self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps({'slides': results, 'explanation': explanation}, ensure_ascii=False).encode('utf-8')) elif parsed.path == '/api/slides/state': if not check_auth(self): return length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode('utf-8') if length > 0 else '{}' try: payload = json.loads(body) except Exception: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"bad_json"}') return write_slide_state(payload) self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(b'{"ok":true}') elif parsed.path == '/api/slides/notes': if not check_auth(self): return length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length).decode('utf-8') if length > 0 else '{}' try: payload = json.loads(body) except Exception: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"bad_json"}') return action = payload.get('action', '') page_key = payload.get('page_key', '') if not page_key: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"missing page_key"}') return with _BOSS_REVIEW_LOCK: data = read_boss_review() notes = list(data.get(page_key, [])) result = {'ok': True} if action == 'add': text = (payload.get('text') or '').strip() if not text: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"empty text"}') return note = { 'id': gen_note_id(), 'text': text, 'status': 'pending', 'ts': time.strftime('%Y-%m-%dT%H:%M:%S'), } notes.append(note) result['note'] = note elif action == 'edit': nid = payload.get('id', '') text = (payload.get('text') or '').strip() for n in notes: if n.get('id') == nid: if text: n['text'] = text if 'status' in payload: n['status'] = payload['status'] n['edited_ts'] = time.strftime('%Y-%m-%dT%H:%M:%S') result['note'] = n break else: self.send_response(404) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"note not found"}') return elif action == 'set_plan': # Heiter Phase 2 結束時呼叫：寫 / 更新該 note 的 heiter_plan nid = payload.get('id', '') plan = payload.get('plan') # dict 或 null（null = 清空 plan） for n in notes: if n.get('id') == nid: if plan is None: n.pop('heiter_plan', None) else: plan['ts'] = time.strftime('%Y-%m-%dT%H:%M:%S') n['heiter_plan'] = plan result['note'] = n break else: self.send_response(404) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"note not found"}') return elif action == 'set_approval': # Boss 在 sidebar 對 heiter_plan 做核可/駁回 # payload.decision = 'approved' | 'rejected' | null（null = 清空） nid = payload.get('id', '') decision = payload.get('decision') reason = (payload.get('reason') or '').strip() for n in notes: if n.get('id') == nid: if decision in (None, ''): n.pop('boss_approval', None) else: ap = { 'decision': decision, 'ts': time.strftime('%Y-%m-%dT%H:%M:%S'), } if reason: ap['reason'] = reason n['boss_approval'] = ap result['note'] = n break else: self.send_response(404) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"note not found"}') return elif action == 'set_resolution': # Heiter Phase 7（驗收+commit 後）呼叫：寫 resolution 區塊 + 同步 status='done' # payload.resolution = {review_pass, commit_hash, handoff_file, actual_filename, notes?} nid = payload.get('id', '') resolution = payload.get('resolution') for n in notes: if n.get('id') == nid: if resolution is None: n.pop('resolution', None) # 清空 resolution 同時把 status 退回 pending if n.get('status') == 'done': n['status'] = 'pending' else: resolution['resolved_ts'] = time.strftime('%Y-%m-%dT%H:%M:%S') resolution.setdefault('resolved_by', 'Heiter') n['resolution'] = resolution # 寫 resolution 時自動把 status 標 done（sidebar UI 才會 hide 紅圈 marks） n['status'] = 'done' result['note'] = n break else: self.send_response(404) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"note not found"}') return elif action == 'delete': nid = payload.get('id', '') notes = [n for n in notes if n.get('id') != nid] elif action == 'set_page_approval': # 一鍵過審：toggle 該頁進 _approved_pages，同時若有 done note 自動補 boss_approval approved = bool(payload.get('approved')) approved_pages = list(data.get('_approved_pages', []) or []) if approved: if page_key not in approved_pages: approved_pages.append(page_key) else: approved_pages = [p for p in approved_pages if p != page_key] data['_approved_pages'] = approved_pages # 同步最新 done note 的 boss_approval done_notes_idx = [i for i, n in enumerate(notes) if n.get('status') == 'done'] if done_notes_idx: latest_i = done_notes_idx[-1] if approved: notes[latest_i]['boss_approval'] = { 'decision': 'approved', 'ts': time.strftime('%Y-%m-%dT%H:%M:%S'), 'via': 'post-resolution-approve', } else: # 取消過審 → 移除 via=post-resolution-approve 的 boss_approval ap = notes[latest_i].get('boss_approval', {}) if ap.get('via') == 'post-resolution-approve': notes[latest_i].pop('boss_approval', None) result['page_key'] = page_key result['approved'] = approved result['approved_pages_count'] = len(approved_pages) else: self.send_response(400) self.send_header('Content-Type', 'application/json') self.end_headers() self.wfile.write(b'{"error":"unknown action"}') return if notes: data[page_key] = notes else: data.pop(page_key, None) write_boss_review(data) self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(json.dumps(result, ensure_ascii=False).encode('utf-8')) else: self.send_response(404) self.end_headers() def do_OPTIONS(self): self.send_response(200) self.send_header('Access-Control-Allow-Origin', '*') self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS') self.send_header('Access-Control-Allow-Headers', 'Content-Type') self.end_headers() if __name__ == '__main__': os.chdir(os.path.dirname(__file__)) # ThreadingHTTPServer：每個 request 一個 thread，避免並發大量 PNG 請求把 server 卡死 _port = int(os.environ.get('PORT', '3030')) server = ThreadingHTTPServer(('', _port), Handler) server.daemon_threads = True print('Serving on :3030 (threaded)') server.serve_forever()