| |
| """ |
| TRTOI 官方網站 + 內容編輯後台 (繁體中文) |
| Taipei Renal Transcriptomics and Outcomes Investigation |
| |
| 架構參考 UPenn TRIDENT。內容存在 content/*.json，可由團隊成員登入後台編輯。 |
| |
| 用法： |
| python web.py 啟動網站 (本機 http://127.0.0.1:5000) |
| python web.py adduser <名稱> 新增可編輯的成員帳號 (互動輸入密碼) |
| python web.py setpass <名稱> 重設某成員密碼 |
| python web.py deluser <名稱> 刪除成員帳號 |
| python web.py listusers 列出成員帳號 |
| |
| 多人協作：各成員各自登入、不同時間分別編輯（後儲存者覆蓋，適合非同時編輯情境）。 |
| """ |
| import os |
| import sys |
| import json |
| import getpass |
| import tempfile |
| import functools |
|
|
| from flask import (Flask, render_template, abort, request, redirect, |
| url_for, session, flash, g, Response) |
| from werkzeug.security import generate_password_hash, check_password_hash |
|
|
| BASE = os.path.dirname(os.path.abspath(__file__)) |
| CONTENT_DIR = os.path.join(BASE, "content") |
| AUTH_FILE = os.path.join(BASE, "auth.json") |
| HERO_DIR = os.path.join(BASE, "static", "uploads", "hero") |
| ALLOWED_IMG_EXT = {".png", ".jpg", ".jpeg", ".gif", ".webp"} |
| TEAM_FILE = os.path.join(CONTENT_DIR, "team_members.json") |
| SITE_FILE = os.path.join(CONTENT_DIR, "site.json") |
| TEAM_PHOTO_DIR = os.path.join(BASE, "static", "uploads", "team") |
| ANALYTICS_DIR = os.path.join(BASE, "analytics") |
| ANALYTICS_FILE = os.path.join(ANALYTICS_DIR, "visits.json") |
| BACKUP_DIR = os.path.join(BASE, "backup") |
|
|
| |
| ORDER = ["index", "ckd", "about", "different", "aims", "goals", |
| "team", "members", "publications", "trtoi2", "links", "contact", "analytics"] |
|
|
| SITE = { |
| "name_zh": "TRTOI 台北腎臟世代研究", |
| "name_full": "Taipei Renal Transcriptomics and Outcomes Investigation", |
| "abbr": "TRTOI", |
| "tagline": "以轉錄體學為核心的腎臟病多體學整合研究平台", |
| } |
|
|
| |
| LANGS = ("en", "zh") |
| DEFAULT_LANG = "en" |
|
|
| |
| UI_STRINGS = { |
| "en": { |
| "html_lang": "en", |
| "site_name": "TRTOI · Taipei Renal Cohort", |
| "tagline": "A transcriptomics-driven multi-omics research platform for kidney disease", |
| "brand": "Division of Nephrology, Taipei Veterans General Hospital", |
| "login": "Member login", |
| "logout": "Log out (%s)", |
| "footer_tagline": "Integrating kidney-biopsy transcriptomics, proteomics, digital pathology and clinical follow-up to advance precision nephrology in Taiwan.", |
| "quick_links": "Quick Links", |
| "ql_about": "What is TRTOI", |
| "ql_aims": "Research Aims", |
| "ql_pubs": "Publications", |
| "ql_contact": "Contact Us", |
| "partners": "International Kidney Biopsy Studies", |
| "copyright": "© 2018–2026 TRTOI Consortium. For academic research and health education only; not medical advice.", |
| "hero_reorder": "Drag to reorder", |
| "hero_delete": "Delete", |
| "hero_upload": "＋ Upload image", |
| "edit_page": "Edit this page", |
| "edit_hint": "Edit mode: click any text to edit", |
| "save": "Save", |
| "cancel": "Cancel", |
| "lang_en": "EN", |
| "lang_zh": "中文", |
| "an_today": "Today's clicks", "an_total": "Total clicks", |
| "an_unique": "Unique visitors", "an_countries": "Countries / regions", |
| "an_trend": "Click trend", "an_day": "Day", "an_week": "Week", "an_month": "Month", |
| "an_map": "Global visitor distribution", "an_ranking": "Top countries / regions", |
| "an_times": "hits", "an_sep": ": ", |
| "an_none_geo": "No locatable visitors yet (a test IP needs internet access to show your location).", |
| "an_nodata": "No traffic data yet.", |
| "cf_name": "Name", "cf_name_ph": "Your name", |
| "cf_subject": "Subject", "cf_subject_ph": "Collaboration / data request / other", |
| "cf_message": "Message", "cf_message_ph": "Please enter your message…", |
| "cf_submit": "Send", "cf_thanks": "Thank you for your message. We will reply soon!", |
| "cf_send_fail": "Sending failed. Please email us directly.", |
| "lg_backend": "TRTOI Content Editor", |
| "lg_intro": "Log in with your research-member account to edit the site.", |
| "lg_user": "Username", "lg_pass": "Password", "lg_submit": "Log in", |
| "lg_back": "← Back to homepage", "lg_error": "Incorrect username or password.", |
| "lg_locked": "Too many failed attempts. Please try again in about %d min.", |
| }, |
| "zh": { |
| "html_lang": "zh-Hant", |
| "site_name": "TRTOI 台北腎臟世代研究", |
| "tagline": "以轉錄體學為核心的腎臟病多體學整合研究平台", |
| "brand": "臺北榮民總醫院 腎臟科", |
| "login": "成員登入", |
| "logout": "登出（%s）", |
| "footer_tagline": "整合腎臟切片轉錄體學、蛋白體學、數位病理與臨床追蹤，推動台灣腎臟病精準醫療。", |
| "quick_links": "快速連結", |
| "ql_about": "什麼是 TRTOI", |
| "ql_aims": "研究目標", |
| "ql_pubs": "研究發表", |
| "ql_contact": "聯絡我們", |
| "partners": "國際腎臟切片研究", |
| "copyright": "© 2018–2026 TRTOI Consortium. 本網站僅供學術研究與衛教用途，不構成醫療建議。", |
| "hero_reorder": "拖曳可調整順序", |
| "hero_delete": "刪除", |
| "hero_upload": "＋ 上傳圖片", |
| "edit_page": "編輯此頁", |
| "edit_hint": "編輯模式：直接點文字修改", |
| "save": "儲存", |
| "cancel": "取消", |
| "lang_en": "EN", |
| "lang_zh": "中文", |
| "an_today": "今日點擊", "an_total": "累積點擊", |
| "an_unique": "不重複訪客", "an_countries": "涵蓋國家／地區", |
| "an_trend": "點擊趨勢", "an_day": "日", "an_week": "週", "an_month": "月", |
| "an_map": "全球訪客分布", "an_ranking": "訪客國家／地區排行", |
| "an_times": "次", "an_sep": "：", |
| "an_none_geo": "目前尚無可定位的訪客（本機測試 IP 需連得上網路才會顯示你目前的所在地）。", |
| "an_nodata": "尚無流量資料。", |
| "cf_name": "姓名", "cf_name_ph": "您的稱呼", |
| "cf_subject": "主旨", "cf_subject_ph": "合作／資料申請／其他", |
| "cf_message": "內容", "cf_message_ph": "請輸入您的訊息…", |
| "cf_submit": "送出", "cf_thanks": "感謝您的來信，我們會盡快回覆！", |
| "cf_send_fail": "送出失敗，請直接來信與我們聯繫。", |
| "lg_backend": "TRTOI 編輯後台", |
| "lg_intro": "請以研究成員帳號登入以編輯網站內容。", |
| "lg_user": "帳號", "lg_pass": "密碼", "lg_submit": "登入", |
| "lg_back": "← 返回網站首頁", "lg_error": "帳號或密碼錯誤", |
| "lg_locked": "登入失敗次數過多，請約 %d 分鐘後再試。", |
| }, |
| } |
|
|
| |
| CHART_LABELS_EN = { |
| "糖尿病": "Diabetes", "性別": "Sex", "CKD 分期": "CKD stage", |
| "前六大病理診斷": "Top 6 pathological diagnoses", |
| "有": "Yes", "無": "No", "男": "Male", "女": "Female", |
| } |
|
|
|
|
| def current_lang(): |
| """目前語言：優先 ?lang= 查詢參數（讓中文頁有可被爬蟲索引的獨立網址）， |
| 其次 cookie `lang`，預設 en。請求外（CLI/import）回 DEFAULT_LANG。""" |
| try: |
| code = request.args.get("lang") or request.cookies.get("lang") |
| except Exception: |
| code = None |
| return code if code in LANGS else DEFAULT_LANG |
|
|
| |
| |
| |
| import configparser |
| import sqlite3 |
| from pathlib import Path |
|
|
| DASHBOARD_DIR = os.environ.get( |
| "TRTOI_DASHBOARD_DIR", |
| r"M:\Also Dropbox\Shuo-Ming Ou\歐朔銘\11 Journal Study\000ClaudeAI\05_TRTOIDashBoard", |
| ) |
|
|
|
|
| def _local_db_path(): |
| """依 Dashboard 的 myconfig.ini [LocalDB] db_path 解析本機 DB 路徑（與儀表板一致）。""" |
| direct = os.environ.get("TRTOI_COHORT_DB") |
| if direct: |
| return direct |
| rel = "../04_PythonAnywhere_Local/example.db" |
| try: |
| cfg = configparser.ConfigParser() |
| cfg.read(os.path.join(DASHBOARD_DIR, "myconfig.ini"), encoding="utf-8") |
| rel = cfg.get("LocalDB", "db_path", fallback=rel) |
| except Exception: |
| pass |
| return os.path.normpath(os.path.join(DASHBOARD_DIR, rel)) |
|
|
|
|
| |
| _charts_cache = {"mtime": None, "data": None} |
|
|
| |
| STATS_FILE = os.path.join(BASE, "data_cloud", "cohort_stats.json") |
| _stats_cache = {"mtime": None, "data": None} |
|
|
|
|
| def _cloud_stats(): |
| try: |
| m = os.path.getmtime(STATS_FILE) |
| if _stats_cache["mtime"] != m: |
| with open(STATS_FILE, "r", encoding="utf-8") as f: |
| _stats_cache["data"] = json.load(f) |
| _stats_cache["mtime"] = m |
| return _stats_cache["data"] |
| except Exception: |
| return None |
|
|
|
|
| def _egfr_ckdepi_2021(scr, age, female): |
| """CKD-EPI 2021（race-free）eGFR；缺值回 None。""" |
| if not scr or not age or scr <= 0: |
| return None |
| k = 0.7 if female else 0.9 |
| a = -0.241 if female else -0.302 |
| r = scr / k |
| e = 142 * (min(r, 1) ** a) * (max(r, 1) ** -1.200) * (0.9938 ** age) |
| return e * 1.012 if female else e |
|
|
|
|
| def _ckd_stage(e): |
| if e is None: |
| return None |
| if e >= 90: return "G1" |
| if e >= 60: return "G2" |
| if e >= 45: return "G3a" |
| if e >= 30: return "G3b" |
| if e >= 15: return "G4" |
| return "G5" |
|
|
|
|
| |
| PATHO_MAP = { |
| "1": "Minimal change disease", "2": "Membranous nephropathy", "3": "FSGS", |
| "4": "Diabetic nephropathy", "5": "Nephrosclerosis", "6": "IgA nephropathy", |
| "7": "TMA", "8": "Acute tubulointerstitial nephritis", "9": "Secondary FSGS", |
| "10": "Mesangial proliferative GN", "11": "Minor glomerular abnormalities", |
| "12": "C3-dominant GN", "13": "Fibrillary GN", "14": "Metastatic neuroendocrine carcinoma", |
| "15": "Oxalate nephropathy", "16": "Dense deposit disease", "17": "Lupus nephritis", |
| "18": "Light chain cast nephropathy", "19": "Membranoproliferative GN", |
| "20": "Pauci-immune crescentic GN", "21": "Nodular glomerulopathy (suspicious HCDD)", |
| "22": "Amyloidosis", "23": "Focal mesangiolysis", "24": "Fabry disease", |
| "25": "Acute tubular injury (kappa)", "26": "Immune complex-mediated GN", |
| "27": "Inadequate specimen", "28": "Malignant HTN", "29": "No biopsy", |
| "30": "Chronic tubulointerstitial nephritis", "31": "Xanthogranulomatous inflammation", |
| "32": "Acute tubular necrosis", "33": "Thin GBM (potential Alport)", "34": "IgA-dominant MPGN", |
| "35": "Ischemic change", "36": "Mild mesangial proliferation", |
| "37": "Idiopathic nodular glomerulosclerosis", "38": "Cryoglobulinemic GN", |
| "39": "Chronic pyelonephritis", "40": "Anti-GBM disease", |
| "41": "Monoclonal Ig deposition disease", "42": "IgM nephropathy", |
| "43": "Type III collagen glomerulopathy", |
| } |
|
|
|
|
| def cohort_charts(): |
| """回傳性別 / 糖尿病 / CKD 分期（圓餅）+ 前六大病理診斷（橫條）的即時分佈；失敗回 None。""" |
| import re |
| from collections import Counter |
| try: |
| db = _local_db_path() |
| m = os.path.getmtime(db) |
| if _charts_cache["mtime"] != m: |
| uri = Path(db).as_uri() + "?mode=ro" |
| con = sqlite3.connect(uri, uri=True) |
| try: |
| rows = con.execute( |
| "SELECT Gender, Diabetes, CREA, Age, Patho_Diagnosis FROM patient" |
| ).fetchall() |
| finally: |
| con.close() |
|
|
| def dig(v): |
| return re.sub(r"\D", "", str(v or "")) |
|
|
| def num(v): |
| mm = re.search(r"[-+]?\d*\.?\d+", str(v)) if v is not None else None |
| return float(mm.group()) if mm else None |
|
|
| gender, dm, ckd, patho = Counter(), Counter(), Counter(), Counter() |
| for g_raw, d_raw, crea, age, pat in rows: |
| g = dig(g_raw) |
| gender[{"1": "男", "0": "女"}.get(g)] += 1 |
| dm[{"1": "有", "0": "無"}.get(dig(d_raw))] += 1 |
| s = _ckd_stage(_egfr_ckdepi_2021(num(crea), num(age), g == "0")) |
| if s: |
| ckd[s] += 1 |
| ps = str(pat or "").strip() |
| if ps: |
| first = re.sub(r"\.0$", "", ps.split(";")[0].strip()) |
| name = PATHO_MAP.get(first) |
| if name: |
| patho[name] += 1 |
|
|
| def build(title, counter, order): |
| labels, data = [], [] |
| for k in order: |
| if counter.get(k): |
| labels.append(k) |
| data.append(counter[k]) |
| return {"title": title, "labels": labels, "data": data} |
|
|
| top = patho.most_common(6) |
| _charts_cache["data"] = { |
| "pies": [ |
| build("糖尿病", dm, ["有", "無"]), |
| build("性別", gender, ["男", "女"]), |
| build("CKD 分期", ckd, ["G1", "G2", "G3a", "G3b", "G4", "G5"]), |
| ], |
| "bar": { |
| "title": "前六大病理診斷", |
| "labels": [k for k, _ in top], |
| "data": [v for _, v in top], |
| "total": sum(patho.values()), |
| }, |
| } |
| _charts_cache["mtime"] = m |
| return _charts_cache["data"] |
| except Exception: |
| pass |
| s = _cloud_stats() |
| return s.get("charts") if s else None |
|
|
|
|
| def cohort_charts_i18n(lang="zh"): |
| """依語言回傳圖表：en 時把固定中文標籤換成英文（病理名 PATHO_MAP 本就是英文）。""" |
| data = cohort_charts() |
| if not data or lang != "en": |
| return data |
| import copy |
| d = copy.deepcopy(data) |
| tr = lambda x: CHART_LABELS_EN.get(x, x) |
| for pie in d.get("pies", []): |
| pie["title"] = tr(pie.get("title")) |
| pie["labels"] = [tr(x) for x in pie.get("labels", [])] |
| bar = d.get("bar") |
| if bar: |
| bar["title"] = tr(bar.get("title")) |
| bar["labels"] = [tr(x) for x in bar.get("labels", [])] |
| return d |
|
|
|
|
| _cohort_cache = {"mtime": None, "count": None} |
|
|
|
|
| def cohort_count(): |
| """回傳切片世代目前人數（patient 表列數），依 DB 檔修改時間快取；失敗回 None。""" |
| try: |
| db = _local_db_path() |
| m = os.path.getmtime(db) |
| if _cohort_cache["mtime"] != m: |
| uri = Path(db).as_uri() + "?mode=ro" |
| con = sqlite3.connect(uri, uri=True) |
| try: |
| _cohort_cache["count"] = con.execute( |
| "SELECT COUNT(*) FROM patient").fetchone()[0] |
| finally: |
| con.close() |
| _cohort_cache["mtime"] = m |
| return _cohort_cache["count"] |
| except Exception: |
| pass |
| s = _cloud_stats() |
| return s.get("count") if s else None |
|
|
| app = Flask(__name__) |
| app.config["MAX_CONTENT_LENGTH"] = 16 * 1024 * 1024 |
|
|
|
|
| HERO_ORDER_FILE = os.path.join(HERO_DIR, "order.json") |
|
|
|
|
| def _hero_files(): |
| try: |
| return [f for f in os.listdir(HERO_DIR) |
| if os.path.splitext(f)[1].lower() in ALLOWED_IMG_EXT] |
| except FileNotFoundError: |
| return [] |
|
|
|
|
| def _hero_read_order(): |
| try: |
| with open(HERO_ORDER_FILE, "r", encoding="utf-8") as fh: |
| return json.load(fh) |
| except Exception: |
| return [] |
|
|
|
|
| def _hero_write_order(order): |
| os.makedirs(HERO_DIR, exist_ok=True) |
| with open(HERO_ORDER_FILE, "w", encoding="utf-8") as fh: |
| json.dump(order, fh, ensure_ascii=False) |
| _persist_mark("static/uploads/hero/order.json", HERO_ORDER_FILE) |
|
|
|
|
| def hero_images(): |
| """頂部輪播圖片檔名清單，依儲存的排序；新上傳者接在後面。""" |
| files = _hero_files() |
| fileset = set(files) |
| order = [f for f in _hero_read_order() if f in fileset] |
| extra = sorted((f for f in files if f not in set(order)), |
| key=lambda f: os.path.getmtime(os.path.join(HERO_DIR, f))) |
| return order + extra |
|
|
|
|
| def team_members(lang="zh"): |
| """研究團隊成員清單：[{id, photo, bio}, ...]（en 缺檔回退中文）。""" |
| path = _lang_path(TEAM_FILE, lang) |
| if not os.path.exists(path): |
| path = TEAM_FILE |
| try: |
| with open(path, "r", encoding="utf-8") as fh: |
| data = json.load(fh) |
| return data if isinstance(data, list) else [] |
| except Exception: |
| return [] |
|
|
|
|
| def save_team(members, lang="zh"): |
| os.makedirs(CONTENT_DIR, exist_ok=True) |
| target = _lang_path(TEAM_FILE, lang) |
| fd, tmp = tempfile.mkstemp(dir=CONTENT_DIR, suffix=".tmp") |
| try: |
| with os.fdopen(fd, "w", encoding="utf-8") as fh: |
| json.dump(members, fh, ensure_ascii=False, indent=2) |
| _replace_retry(tmp, target) |
| finally: |
| if os.path.exists(tmp): |
| os.remove(tmp) |
| _backup(target) |
| _persist_mark("content/" + os.path.basename(target), target) |
|
|
|
|
| |
| SITE_TEXT_DEFAULTS = { |
| "side_note": "TRTOI 為台灣最大腎臟 RNA 資料庫，整合全基因定序、腎臟切片轉錄體學、" |
| "蛋白體學、數位病理與長期臨床追蹤。", |
| } |
|
|
|
|
| def load_site(lang="zh"): |
| """站台層級文字：先讀中文再以該語言覆蓋（en 缺欄位 → 回退中文/預設）。""" |
| data = dict(SITE_TEXT_DEFAULTS) |
| paths = [SITE_FILE] if lang == "zh" else [SITE_FILE, _lang_path(SITE_FILE, lang)] |
| for p in paths: |
| try: |
| with open(p, "r", encoding="utf-8") as fh: |
| saved = json.load(fh) |
| if isinstance(saved, dict): |
| data.update(saved) |
| except Exception: |
| pass |
| return data |
|
|
|
|
| def save_site(data, lang="zh"): |
| os.makedirs(CONTENT_DIR, exist_ok=True) |
| target = _lang_path(SITE_FILE, lang) |
| fd, tmp = tempfile.mkstemp(dir=CONTENT_DIR, suffix=".tmp") |
| try: |
| with os.fdopen(fd, "w", encoding="utf-8") as fh: |
| json.dump(data, fh, ensure_ascii=False, indent=2) |
| _replace_retry(tmp, target) |
| finally: |
| if os.path.exists(tmp): |
| os.remove(tmp) |
| _backup(target) |
| _persist_mark("content/" + os.path.basename(target), target) |
|
|
|
|
| |
| |
| |
| import time as _time |
|
|
| _analytics = None |
| _analytics_last_write = 0.0 |
| _analytics_persist_last = 0.0 |
|
|
|
|
| def _today_str(): |
| import datetime |
| return datetime.date.today().isoformat() |
|
|
|
|
| def _load_analytics(): |
| global _analytics |
| if _analytics is None: |
| if _persist_enabled(): |
| |
| try: |
| from huggingface_hub import hf_hub_download |
| p = hf_hub_download(repo_id=HF_ANALYTICS_REPO, repo_type="dataset", |
| filename="visits.json", token=HF_WRITE_TOKEN) |
| with open(p, "r", encoding="utf-8") as f: |
| _analytics = json.load(f) |
| except Exception: |
| _analytics = None |
| if _analytics is None: |
| try: |
| with open(ANALYTICS_FILE, "r", encoding="utf-8") as f: |
| _analytics = json.load(f) |
| except Exception: |
| _analytics = {} |
| for k, v in (("daily", {}), ("ip_hits", {}), ("ip_geo", {}), ("total", 0)): |
| _analytics.setdefault(k, v) |
| return _analytics |
|
|
|
|
| def _save_analytics(force=False): |
| global _analytics_last_write |
| if _analytics is None: |
| return |
| now = _time.time() |
| if not force and now - _analytics_last_write < 5: |
| return |
| try: |
| os.makedirs(ANALYTICS_DIR, exist_ok=True) |
| fd, tmp = tempfile.mkstemp(dir=ANALYTICS_DIR, suffix=".tmp") |
| with os.fdopen(fd, "w", encoding="utf-8") as f: |
| json.dump(_analytics, f, ensure_ascii=False) |
| _replace_retry(tmp, ANALYTICS_FILE) |
| _analytics_last_write = now |
| except Exception: |
| pass |
|
|
|
|
| def _maybe_persist_analytics(force=False): |
| """把流量檔節流上傳到 dataset repo（HF_ANALYTICS_REPO）。 |
| ⚠️ 不可存回 Space repo：Space repo 的每個 commit 都會觸發 rebuild，而 rebuild |
| 重啟後節流計時器歸零 + keep-alive ping 馬上又 persist → 無限 rebuild 迴圈 |
| （2026-08-21 的 BUILD_ERROR 即此因）。dataset 的 commit 不會觸發 rebuild。""" |
| global _analytics_persist_last |
| if not _persist_enabled(): |
| return |
| now = _time.time() |
| if not force and now - _analytics_persist_last < 600: |
| return |
| _analytics_persist_last = now |
| _save_analytics(force=True) |
| threading.Thread(target=_upload_analytics_dataset, daemon=True).start() |
|
|
|
|
| def _upload_analytics_dataset(): |
| try: |
| from huggingface_hub import HfApi |
| HfApi(token=HF_WRITE_TOKEN).upload_file( |
| path_or_fileobj=ANALYTICS_FILE, path_in_repo="visits.json", |
| repo_id=HF_ANALYTICS_REPO, repo_type="dataset", |
| commit_message="persist analytics") |
| except Exception: |
| pass |
|
|
|
|
| def _client_ip(): |
| xff = request.headers.get("X-Forwarded-For", "") |
| return xff.split(",")[0].strip() if xff else (request.remote_addr or "unknown") |
|
|
|
|
| def _is_private_ip(ip): |
| import ipaddress |
| try: |
| a = ipaddress.ip_address(ip) |
| return a.is_private or a.is_loopback |
| except Exception: |
| return True |
|
|
|
|
| def _geo_lookup(ip): |
| """IP → 國家 / 座標（ip-api.com，免費無金鑰）；私有/本機 IP 改查伺服器自身公網位置。""" |
| import urllib.request |
| q = "" if _is_private_ip(ip) else ip |
| url = "http://ip-api.com/json/%s?fields=status,countryCode,country,lat,lon" % q |
| try: |
| d = json.loads(urllib.request.urlopen(url, timeout=3).read().decode("utf-8")) |
| if d.get("status") == "success": |
| return {"cc": d.get("countryCode"), "country": d.get("country"), |
| "lat": d.get("lat"), "lng": d.get("lon")} |
| except Exception: |
| pass |
| return None |
|
|
|
|
| @app.before_request |
| def _force_canonical(): |
| """直接開 *.hf.space（未經 Cloudflare Worker、沒帶祕鑰）→ 301 轉到正式網域， |
| 讓所有人統一走 trtoi.org。未設 PROXY_KEY（本機開發）則不強制。""" |
| try: |
| if not PROXY_KEY: |
| return |
| host = (request.host or "").split(":")[0].lower() |
| if host.endswith(".hf.space") and request.headers.get("X-Proxy-Key", "") != PROXY_KEY: |
| path = request.full_path |
| if path.endswith("?"): |
| path = path[:-1] |
| return redirect("https://%s%s" % (CANONICAL_HOST, path), code=301) |
| except Exception: |
| pass |
|
|
|
|
| @app.before_request |
| def _set_lang(): |
| """把目前語言掛到 g.lang，供路由與模板使用。""" |
| g.lang = current_lang() |
|
|
|
|
| @app.route("/lang/<code>") |
| def set_lang(code): |
| """切換語言：設 cookie 後回到原頁。""" |
| if code not in LANGS: |
| code = DEFAULT_LANG |
| resp = redirect(request.referrer or url_for("index")) |
| resp.set_cookie("lang", code, max_age=60 * 60 * 24 * 365, samesite="Lax") |
| return resp |
|
|
|
|
| @app.before_request |
| def _track_visit(): |
| try: |
| p = request.path |
| if request.method != "GET": |
| return |
| if p.startswith("/static") or p.startswith("/admin") or p.startswith("/favicon"): |
| return |
| a = _load_analytics() |
| today = _today_str() |
| a["daily"][today] = a["daily"].get(today, 0) + 1 |
| a["total"] += 1 |
| ip = _client_ip() |
| a["ip_hits"][ip] = a["ip_hits"].get(ip, 0) + 1 |
| _save_analytics() |
| _maybe_persist_analytics() |
| except Exception: |
| pass |
|
|
|
|
| def analytics_report(): |
| """流量摘要（供流量統計頁）。地理定位延後到此，且每次上限 40 個新 IP（尊重 API 限速）。""" |
| a = _load_analytics() |
| missing = [ip for ip in a["ip_hits"] if ip not in a["ip_geo"]][:40] |
| for ip in missing: |
| g = _geo_lookup(ip) |
| if g and g.get("cc"): |
| a["ip_geo"][ip] = g |
| if missing: |
| _save_analytics(force=True) |
| _maybe_persist_analytics(force=True) |
| countries = {} |
| for ip, hits in a["ip_hits"].items(): |
| g = a["ip_geo"].get(ip) |
| if g and g.get("cc"): |
| c = countries.setdefault(g["cc"], {"cc": g["cc"], "country": g["country"], |
| "hits": 0, "lat": g["lat"], "lng": g["lng"]}) |
| c["hits"] += hits |
| import datetime |
| day_items = sorted(a["daily"].items()) |
| |
| day_series = [{"label": d[5:], "hits": h} for d, h in day_items[-30:]] |
| |
| weekly = {} |
| monthly = {} |
| for d, h in a["daily"].items(): |
| try: |
| dt = datetime.date.fromisoformat(d) |
| except Exception: |
| continue |
| monday = (dt - datetime.timedelta(days=dt.weekday())).isoformat() |
| weekly[monday] = weekly.get(monday, 0) + h |
| monthly[d[:7]] = monthly.get(d[:7], 0) + h |
| def _wk_label(monday_iso): |
| m = datetime.date.fromisoformat(monday_iso) |
| sun = m + datetime.timedelta(days=6) |
| return f"{monday_iso[5:]}~{sun.isoformat()[5:]}" |
| week_series = [{"label": _wk_label(k), "hits": v} for k, v in sorted(weekly.items())[-12:]] |
| month_series = [{"label": k, "hits": v} for k, v in sorted(monthly.items())[-12:]] |
| return { |
| "today": a["daily"].get(_today_str(), 0), |
| "total": a["total"], |
| "unique": len(a["ip_hits"]), |
| "series": {"day": day_series, "week": week_series, "month": month_series}, |
| "countries": sorted(countries.values(), key=lambda x: -x["hits"]), |
| } |
|
|
|
|
| |
| |
| |
| |
| import threading |
|
|
| _persist_dirty = {} |
| _persist_lock = threading.Lock() |
| _persist_last = 0.0 |
| _persist_started = False |
|
|
|
|
| def _persist_enabled(): |
| return bool(HF_PERSIST and HF_SPACE_REPO and HF_WRITE_TOKEN) |
|
|
|
|
| def _persist_mark(repo_path, local_path): |
| """標記一個檔案要同步回 repo（local_path=None 表示刪除）。""" |
| if not _persist_enabled(): |
| return |
| global _persist_last, _persist_started |
| with _persist_lock: |
| _persist_dirty[repo_path] = local_path |
| _persist_last = _time.time() |
| if not _persist_started: |
| _persist_started = True |
| threading.Thread(target=_persist_worker, daemon=True).start() |
|
|
|
|
| def _persist_worker(): |
| from huggingface_hub import HfApi, CommitOperationAdd, CommitOperationDelete |
| api = HfApi(token=HF_WRITE_TOKEN) |
| while True: |
| _time.sleep(5) |
| with _persist_lock: |
| if not _persist_dirty or (_time.time() - _persist_last) < 20: |
| continue |
| batch = dict(_persist_dirty) |
| _persist_dirty.clear() |
| ops = [] |
| for repo_path, local_path in batch.items(): |
| try: |
| if local_path is None: |
| ops.append(CommitOperationDelete(path_in_repo=repo_path)) |
| elif os.path.exists(local_path): |
| ops.append(CommitOperationAdd(path_in_repo=repo_path, |
| path_or_fileobj=local_path)) |
| except Exception: |
| pass |
| if not ops: |
| continue |
| try: |
| api.create_commit(repo_id=HF_SPACE_REPO, repo_type="space", |
| operations=ops, |
| commit_message="edit via site (%d files)" % len(ops)) |
| except Exception: |
| with _persist_lock: |
| for k, v in batch.items(): |
| _persist_dirty.setdefault(k, v) |
|
|
|
|
| |
| |
| |
| def _lang_path(path, lang): |
| """en → 在副檔名前插入 .en（xxx.json → xxx.en.json）；zh → 原路徑。""" |
| if lang == "en": |
| root, ext = os.path.splitext(path) |
| return root + ".en" + ext |
| return path |
|
|
|
|
| def content_path(slug, lang="zh"): |
| return _lang_path(os.path.join(CONTENT_DIR, slug + ".json"), lang) |
|
|
|
|
| def load_page(slug, lang="zh"): |
| """讀取某分頁內容 JSON（en 缺檔 → 回退中文，未翻譯頁仍可顯示）。不存在回 None。""" |
| if slug not in ORDER: |
| return None |
| path = content_path(slug, lang) |
| if not os.path.exists(path): |
| path = content_path(slug, "zh") |
| if not os.path.exists(path): |
| return None |
| with open(path, "r", encoding="utf-8") as f: |
| return json.load(f) |
|
|
|
|
| def _backup(local_path): |
| """把剛存好的內容檔複製一份到 backup/ 做時間戳版本備份（每檔保留最近 20 份）。 |
| 本機與雲端都會做；雲端另有 repo commit 歷史作為版本備份。""" |
| try: |
| if not os.path.exists(local_path): |
| return |
| import datetime |
| import glob |
| import shutil |
| os.makedirs(BACKUP_DIR, exist_ok=True) |
| base = os.path.basename(local_path) |
| stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") |
| shutil.copy2(local_path, os.path.join(BACKUP_DIR, "%s.%s.bak" % (base, stamp))) |
| olds = sorted(glob.glob(os.path.join(BACKUP_DIR, base + ".*.bak"))) |
| for old in olds[:-20]: |
| try: |
| os.remove(old) |
| except Exception: |
| pass |
| except Exception: |
| pass |
|
|
|
|
| def _replace_retry(src, dst, tries=6): |
| """os.replace 加重試：Dropbox／防毒同步時可能暫時鎖檔（WinError 5）。""" |
| import time |
| for _ in range(tries - 1): |
| try: |
| os.replace(src, dst) |
| return |
| except PermissionError: |
| time.sleep(0.15) |
| os.replace(src, dst) |
|
|
|
|
| def save_page(slug, data, lang="zh"): |
| """原子性寫回內容 JSON（en 寫入 <slug>.en.json）。""" |
| os.makedirs(CONTENT_DIR, exist_ok=True) |
| target = content_path(slug, lang) |
| fd, tmp = tempfile.mkstemp(dir=CONTENT_DIR, suffix=".tmp") |
| try: |
| with os.fdopen(fd, "w", encoding="utf-8") as f: |
| json.dump(data, f, ensure_ascii=False, indent=2) |
| _replace_retry(tmp, target) |
| finally: |
| if os.path.exists(tmp): |
| os.remove(tmp) |
| _backup(target) |
| _persist_mark("content/" + os.path.basename(target), target) |
|
|
|
|
| def build_nav(lang="zh"): |
| """依 ORDER 建立左側選單；標題取自各分頁 JSON 的 nav 欄位（依語言）。""" |
| nav = [] |
| for slug in ORDER: |
| data = load_page(slug, lang) |
| if not data: |
| continue |
| label = data.get("nav") or data.get("title") or slug |
| url = "/" if slug == "index" else "/" + slug |
| nav.append({"slug": slug, "url": url, "label": label}) |
| return nav |
|
|
|
|
| |
| |
| |
| def load_auth(): |
| """讀取（或初次建立）認證檔。首次會建立預設帳號 admin。""" |
| if not os.path.exists(AUTH_FILE): |
| data = { |
| "secret": os.urandom(24).hex(), |
| "users": {"admin": generate_password_hash("trtoi2026")}, |
| } |
| with open(AUTH_FILE, "w", encoding="utf-8") as f: |
| json.dump(data, f, ensure_ascii=False, indent=2) |
| print("※ 已建立預設管理員帳號 admin（預設密碼見 README，請盡快用 " |
| "`python web.py setpass admin` 更改）。") |
| return data |
| with open(AUTH_FILE, "r", encoding="utf-8") as f: |
| return json.load(f) |
|
|
|
|
| def save_auth(data): |
| with open(AUTH_FILE, "w", encoding="utf-8") as f: |
| json.dump(data, f, ensure_ascii=False, indent=2) |
|
|
|
|
| AUTH = load_auth() |
| |
| HF_MODE = os.environ.get("HF_MODE") == "1" |
| |
| app.secret_key = os.environ.get("FLASK_SECRET_KEY") or AUTH["secret"] |
|
|
| |
| HF_PERSIST = os.environ.get("HF_PERSIST") == "1" |
| HF_SPACE_REPO = os.environ.get("SPACE_ID", "") |
| HF_WRITE_TOKEN = os.environ.get("HF_WRITE_TOKEN", "") |
| |
| HF_ANALYTICS_REPO = os.environ.get("HF_ANALYTICS_REPO") or \ |
| (HF_SPACE_REPO + "-analytics" if HF_SPACE_REPO else "") |
| |
| PROXY_KEY = os.environ.get("PROXY_KEY", "") |
| CANONICAL_HOST = os.environ.get("CANONICAL_HOST", "trtoi.org") |
| |
| _env_admin_user = os.environ.get("ADMIN_USER") |
| _env_admin_pw = os.environ.get("ADMIN_PASSWORD") |
| if _env_admin_user and _env_admin_pw: |
| AUTH.setdefault("users", {})[_env_admin_user] = generate_password_hash(_env_admin_pw) |
|
|
|
|
| def login_required(view): |
| @functools.wraps(view) |
| def wrapped(*args, **kwargs): |
| if not session.get("user"): |
| return redirect(url_for("admin_login", next=request.path)) |
| return view(*args, **kwargs) |
| return wrapped |
|
|
|
|
| |
| |
| |
| def asset_ver(filename): |
| """回傳靜態檔的修改時間當版本號，讓瀏覽器在檔案更新後自動抓新版（破快取）。""" |
| try: |
| return str(int(os.path.getmtime(os.path.join(app.static_folder, filename)))) |
| except OSError: |
| return "1" |
|
|
|
|
| @app.context_processor |
| def inject_globals(): |
| lang = getattr(g, "lang", DEFAULT_LANG) |
| return {"NAV": build_nav(lang), "SITE": SITE, "site_text": load_site(lang), |
| "T": UI_STRINGS.get(lang, UI_STRINGS[DEFAULT_LANG]), "LANG": lang, |
| "user": session.get("user"), "ver": asset_ver, |
| "cohort": cohort_count(), "charts": cohort_charts_i18n(lang), |
| "hero_images": hero_images(), "team_members": team_members(lang), |
| "readonly": HF_MODE} |
|
|
|
|
| |
| |
| |
| @app.route("/") |
| def index(): |
| page = load_page("index", g.lang) |
| if not page: |
| abort(404) |
| return render_template("page.html", page=page, active="index") |
|
|
|
|
| |
| |
| |
| SEO_HOST = "https://%s" % CANONICAL_HOST |
| SEO_SLUGS = [s for s in ORDER if s != "analytics"] |
|
|
|
|
| @app.route("/robots.txt") |
| def robots_txt(): |
| body = ("User-agent: *\n" |
| "Allow: /\n" |
| "Disallow: /admin\n" |
| "Sitemap: %s/sitemap.xml\n" % SEO_HOST) |
| return Response(body, mimetype="text/plain") |
|
|
|
|
| @app.route("/sitemap.xml") |
| def sitemap_xml(): |
| parts = ['<?xml version="1.0" encoding="UTF-8"?>', |
| '<urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9">'] |
| for slug in SEO_SLUGS: |
| path = "/" if slug == "index" else "/" + slug |
| |
| parts.append("<url><loc>%s</loc></url>" % (SEO_HOST + path)) |
| parts.append("</urlset>") |
| return Response("\n".join(parts), mimetype="application/xml") |
|
|
|
|
| @app.route("/<slug>") |
| def page(slug): |
| if slug not in ORDER or slug == "index": |
| abort(404) |
| data = load_page(slug, g.lang) |
| if not data: |
| abort(404) |
| extra = {"analytics": analytics_report()} if slug == "analytics" else {} |
| return render_template("page.html", page=data, active=slug, **extra) |
|
|
|
|
| |
| |
| |
| |
| LOGIN_MAX_FAILS = 5 |
| LOGIN_WINDOW = 600 |
| LOGIN_LOCKOUT = 900 |
| _login_lock = threading.Lock() |
| _login_fails = {} |
|
|
|
|
| def _login_locked(ip): |
| """回傳剩餘鎖定秒數（>0 表示鎖定中）。""" |
| now = _time.time() |
| with _login_lock: |
| rec = _login_fails.get(ip) |
| if rec and rec.get("until", 0) > now: |
| return int(rec["until"] - now) |
| return 0 |
|
|
|
|
| def _login_fail(ip): |
| now = _time.time() |
| with _login_lock: |
| rec = _login_fails.setdefault(ip, {"ts": [], "until": 0}) |
| rec["ts"] = [t for t in rec["ts"] if now - t < LOGIN_WINDOW] |
| rec["ts"].append(now) |
| if len(rec["ts"]) >= LOGIN_MAX_FAILS: |
| rec["until"] = now + LOGIN_LOCKOUT |
| rec["ts"] = [] |
| if len(_login_fails) > 5000: |
| for k in [k for k, v in _login_fails.items() |
| if v.get("until", 0) < now and not v.get("ts")]: |
| _login_fails.pop(k, None) |
|
|
|
|
| def _login_ok(ip): |
| with _login_lock: |
| _login_fails.pop(ip, None) |
|
|
|
|
| @app.route("/admin/login", methods=["GET", "POST"]) |
| def admin_login(): |
| if HF_MODE: |
| abort(404) |
| T = UI_STRINGS.get(g.lang, UI_STRINGS[DEFAULT_LANG]) |
| if request.method == "POST": |
| ip = _client_ip() |
| locked = _login_locked(ip) |
| if locked: |
| flash(T["lg_locked"] % max(1, locked // 60)) |
| return render_template("admin/login.html", active=None) |
| username = (request.form.get("username") or "").strip() |
| password = request.form.get("password") or "" |
| pw_hash = AUTH.get("users", {}).get(username) |
| if pw_hash and check_password_hash(pw_hash, password): |
| _login_ok(ip) |
| session["user"] = username |
| nxt = request.args.get("next") or url_for("index") |
| return redirect(nxt) |
| _login_fail(ip) |
| flash(T["lg_error"]) |
| return render_template("admin/login.html", active=None) |
|
|
|
|
| @app.route("/admin/logout") |
| def admin_logout(): |
| session.pop("user", None) |
| return redirect(url_for("index")) |
|
|
|
|
| |
| |
| |
| |
| |
| def _mail_cfg(key, default=None): |
| v = os.environ.get(key) |
| if v: |
| return v |
| try: |
| cfg = configparser.ConfigParser() |
| cfg.read(os.path.join(BASE, "hf_deploy.ini"), encoding="utf-8") |
| return cfg.get("MAIL", key.lower(), fallback=default) |
| except Exception: |
| return default |
|
|
|
|
| def _send_contact_mail(name, email, subject, message): |
| to_addr = _mail_cfg("CONTACT_TO") |
| user = _mail_cfg("SMTP_USER") |
| pw = _mail_cfg("SMTP_PASS") |
| host = _mail_cfg("SMTP_HOST", "smtp.gmail.com") |
| port = int(_mail_cfg("SMTP_PORT", "587") or 587) |
| if not (to_addr and user and pw): |
| return False, "not_configured" |
| import smtplib |
| import ssl |
| from email.message import EmailMessage |
| from email.utils import formataddr |
| msg = EmailMessage() |
| msg["Subject"] = "[TRTOI 網站來信] " + (subject or "(無主旨)") |
| msg["From"] = formataddr(("TRTOI Website", user)) |
| msg["To"] = to_addr |
| if email: |
| msg["Reply-To"] = email |
| msg.set_content("姓名 Name: %s\nEmail: %s\n主旨 Subject: %s\n\n%s" |
| % (name, email, subject, message)) |
| try: |
| with smtplib.SMTP(host, port, timeout=20) as s: |
| s.starttls(context=ssl.create_default_context()) |
| s.login(user, pw) |
| s.send_message(msg) |
| return True, "sent" |
| except Exception as e: |
| return False, type(e).__name__ |
|
|
|
|
| _contact_last = {} |
|
|
|
|
| @app.route("/contact/send", methods=["POST"]) |
| def contact_send(): |
| d = request.get_json(silent=True) or {} |
| name = (d.get("name") or "").strip()[:200] |
| email = (d.get("email") or "").strip()[:200] |
| subject = (d.get("subject") or "").strip()[:300] |
| message = (d.get("message") or "").strip()[:5000] |
| if not (name and email and message): |
| return {"ok": False, "error": "missing"} |
| ip = _client_ip() |
| now = _time.time() |
| if now - _contact_last.get(ip, 0) < 30: |
| return {"ok": False, "error": "rate"} |
| ok, info = _send_contact_mail(name, email, subject, message) |
| if ok: |
| _contact_last[ip] = now |
| return {"ok": bool(ok), "error": None if ok else info} |
|
|
|
|
| def apply_patch(data, path, value): |
| """依 'blocks.2.items.0.title' 之類的路徑，把 value 寫入 data 對應位置。""" |
| parts = path.split(".") |
| cur = data |
| for p in parts[:-1]: |
| cur = cur[int(p)] if p.isdigit() else cur[p] |
| last = parts[-1] |
| if last.isdigit(): |
| cur[int(last)] = value |
| else: |
| cur[last] = value |
|
|
|
|
| @app.route("/admin/hero/upload", methods=["POST"]) |
| @login_required |
| def hero_upload(): |
| """上傳頂部輪播圖片（可多張）。""" |
| import uuid |
| os.makedirs(HERO_DIR, exist_ok=True) |
| saved = 0 |
| for f in request.files.getlist("images"): |
| if not f or not f.filename: |
| continue |
| ext = os.path.splitext(f.filename)[1].lower() |
| if ext not in ALLOWED_IMG_EXT: |
| continue |
| name = uuid.uuid4().hex + ext |
| f.save(os.path.join(HERO_DIR, name)) |
| _persist_mark("static/uploads/hero/" + name, os.path.join(HERO_DIR, name)) |
| saved += 1 |
| if saved: |
| _hero_write_order(hero_images()) |
| return {"ok": saved > 0, "saved": saved} |
|
|
|
|
| @app.route("/admin/hero/delete", methods=["POST"]) |
| @login_required |
| def hero_delete(): |
| """刪除指定的輪播圖片。""" |
| data = request.get_json(silent=True) or {} |
| name = os.path.basename(data.get("name", "")) |
| path = os.path.join(HERO_DIR, name) |
| if os.path.splitext(name)[1].lower() in ALLOWED_IMG_EXT and os.path.exists(path): |
| os.remove(path) |
| _persist_mark("static/uploads/hero/" + name, None) |
| _hero_write_order([f for f in _hero_read_order() if f != name]) |
| return {"ok": True} |
| return {"ok": False} |
|
|
|
|
| @app.route("/admin/hero/reorder", methods=["POST"]) |
| @login_required |
| def hero_reorder(): |
| """儲存拖曳後的圖片順序。""" |
| data = request.get_json(silent=True) or {} |
| fileset = set(_hero_files()) |
| order = [os.path.basename(n) for n in data.get("order", []) |
| if os.path.basename(n) in fileset] |
| _hero_write_order(order) |
| return {"ok": True} |
|
|
|
|
| def _other_lang(lang): |
| return "zh" if lang == "en" else "en" |
|
|
|
|
| def _photo_referenced(photo, lang): |
| """該照片是否仍被另一語言的團隊清單引用（避免刪到跨語言共用檔）。""" |
| if not photo: |
| return False |
| other = _other_lang(lang) |
| if not os.path.exists(_lang_path(TEAM_FILE, other)): |
| return False |
| bn = os.path.basename(photo) |
| return any(os.path.basename((m or {}).get("photo") or "") == bn |
| for m in team_members(other)) |
|
|
|
|
| @app.route("/admin/team/add", methods=["POST"]) |
| @login_required |
| def team_add(): |
| import uuid |
| bio_template = ( |
| "<strong>姓名 Name</strong><br>" |
| "現職：<br>" |
| "學歷：<br>" |
| "經歷：<br>" |
| "研究專長：" |
| ) |
| members = team_members(g.lang) |
| members.append({"id": uuid.uuid4().hex, "title": "", |
| "photo": "", "bio": bio_template}) |
| save_team(members, g.lang) |
| return {"ok": True} |
|
|
|
|
| @app.route("/admin/team/delete", methods=["POST"]) |
| @login_required |
| def team_delete(): |
| data = request.get_json(silent=True) or {} |
| mid = data.get("id", "") |
| members = team_members(g.lang) |
| for m in members: |
| if m.get("id") == mid and m.get("photo") and not _photo_referenced(m["photo"], g.lang): |
| bn = os.path.basename(m["photo"]) |
| p = os.path.join(TEAM_PHOTO_DIR, bn) |
| if os.path.exists(p): |
| os.remove(p) |
| _persist_mark("static/uploads/team/" + bn, None) |
| save_team([m for m in members if m.get("id") != mid], g.lang) |
| return {"ok": True} |
|
|
|
|
| @app.route("/admin/team/bio", methods=["POST"]) |
| @login_required |
| def team_bio(): |
| data = request.get_json(silent=True) or {} |
| mid, bio = data.get("id", ""), data.get("bio", "") |
| members = team_members(g.lang) |
| for m in members: |
| if m.get("id") == mid: |
| m["bio"] = bio |
| save_team(members, g.lang) |
| return {"ok": True} |
|
|
|
|
| @app.route("/admin/team/title", methods=["POST"]) |
| @login_required |
| def team_title(): |
| data = request.get_json(silent=True) or {} |
| mid, title = data.get("id", ""), data.get("title", "") |
| members = team_members(g.lang) |
| for m in members: |
| if m.get("id") == mid: |
| m["title"] = title |
| save_team(members, g.lang) |
| return {"ok": True} |
|
|
|
|
| @app.route("/admin/team/photo", methods=["POST"]) |
| @login_required |
| def team_photo(): |
| import uuid |
| mid = request.form.get("id", "") |
| f = request.files.get("photo") |
| if not f or not f.filename: |
| return {"ok": False, "error": "no file"} |
| ext = os.path.splitext(f.filename)[1].lower() |
| if ext not in ALLOWED_IMG_EXT: |
| return {"ok": False, "error": "bad type"} |
| os.makedirs(TEAM_PHOTO_DIR, exist_ok=True) |
| name = uuid.uuid4().hex + ext |
| f.save(os.path.join(TEAM_PHOTO_DIR, name)) |
| _persist_mark("static/uploads/team/" + name, os.path.join(TEAM_PHOTO_DIR, name)) |
| members = team_members(g.lang) |
| for m in members: |
| if m.get("id") == mid: |
| if m.get("photo") and not _photo_referenced(m["photo"], g.lang): |
| oldname = os.path.basename(m["photo"]) |
| oldp = os.path.join(TEAM_PHOTO_DIR, oldname) |
| if os.path.exists(oldp): |
| os.remove(oldp) |
| _persist_mark("static/uploads/team/" + oldname, None) |
| m["photo"] = name |
| save_team(members, g.lang) |
| return {"ok": True, "photo": name} |
|
|
|
|
| @app.route("/admin/team/reorder", methods=["POST"]) |
| @login_required |
| def team_reorder(): |
| data = request.get_json(silent=True) or {} |
| order = data.get("order", []) |
| members = team_members(g.lang) |
| by_id = {m.get("id"): m for m in members} |
| new = [by_id[i] for i in order if i in by_id] |
| new += [m for m in members if m.get("id") not in set(order)] |
| save_team(new, g.lang) |
| return {"ok": True} |
|
|
|
|
| def _pub_block(data): |
| for b in data.get("blocks", []): |
| if b.get("type") == "publications": |
| return b |
| return None |
|
|
|
|
| @app.route("/admin/pub/add", methods=["POST"]) |
| @login_required |
| def pub_add(): |
| data = load_page("publications", g.lang) |
| b = _pub_block(data) |
| if b is None: |
| return {"ok": False} |
| b.setdefault("items", []).insert( |
| 0, {"year": "", "title": "", "journal": "", "desc": "", "badge": ""}) |
| save_page("publications", data, g.lang) |
| return {"ok": True} |
|
|
|
|
| @app.route("/admin/pub/delete", methods=["POST"]) |
| @login_required |
| def pub_delete(): |
| idx = (request.get_json(silent=True) or {}).get("index") |
| data = load_page("publications", g.lang) |
| b = _pub_block(data) |
| if b and isinstance(idx, int) and 0 <= idx < len(b.get("items", [])): |
| b["items"].pop(idx) |
| save_page("publications", data, g.lang) |
| return {"ok": True} |
| return {"ok": False} |
|
|
|
|
| @app.route("/admin/pub/field", methods=["POST"]) |
| @login_required |
| def pub_field(): |
| d = request.get_json(silent=True) or {} |
| idx, field, value = d.get("index"), d.get("field"), d.get("value", "") |
| if field not in {"year", "title", "journal", "desc", "badge"}: |
| return {"ok": False} |
| |
| neutral = field in {"year", "title", "journal"} |
| langs = list(LANGS) if neutral else [g.lang] |
| ok = False |
| for lg in langs: |
| if lg != "zh" and not os.path.exists(content_path("publications", lg)): |
| continue |
| data = load_page("publications", lg) |
| b = _pub_block(data) if data else None |
| if b and isinstance(idx, int) and 0 <= idx < len(b.get("items", [])): |
| b["items"][idx][field] = value |
| save_page("publications", data, lg) |
| ok = True |
| return {"ok": ok} |
|
|
|
|
| @app.route("/admin/save-inline/<slug>", methods=["POST"]) |
| @login_required |
| def admin_save_inline(slug): |
| """就地編輯的儲存端點：接收 {patches: {路徑: 新值}} 並寫回 JSON。 |
| slug 為 'site' 時寫回站台層級文字（side-note 等每頁共用元素），其餘為分頁內容。""" |
| payload = request.get_json(silent=True) or {} |
| patches = payload.get("patches", {}) |
| lang = payload.get("lang") if payload.get("lang") in LANGS else g.lang |
| if not isinstance(patches, dict): |
| return {"ok": False, "error": "bad payload"} |
| if slug == "site": |
| data = load_site(lang) |
| saver = lambda d: save_site(d, lang) |
| elif slug in ORDER: |
| data = load_page(slug, lang) |
| if not data: |
| return {"ok": False, "error": "page not found"} |
| saver = lambda d: save_page(slug, d, lang) |
| else: |
| return {"ok": False, "error": "unknown page"} |
| for path, val in patches.items(): |
| try: |
| apply_patch(data, str(path), val) |
| except Exception: |
| pass |
| saver(data) |
| return {"ok": True} |
|
|
|
|
| |
| |
| |
| def cli(): |
| args = sys.argv[1:] |
| if not args: |
| app.run(host="127.0.0.1", port=5000, debug=True) |
| return |
|
|
| cmd = args[0] |
| auth = load_auth() |
|
|
| if cmd == "listusers": |
| print("成員帳號：", ", ".join(sorted(auth["users"].keys())) or "(無)") |
|
|
| elif cmd in ("adduser", "setpass") and len(args) >= 2: |
| name = args[1] |
| pw1 = getpass.getpass("輸入密碼：") |
| pw2 = getpass.getpass("再次輸入：") |
| if pw1 != pw2: |
| print("兩次密碼不一致，取消。") |
| return |
| if not pw1: |
| print("密碼不可為空，取消。") |
| return |
| auth["users"][name] = generate_password_hash(pw1) |
| save_auth(auth) |
| print(f"已{'新增' if cmd == 'adduser' else '更新'}帳號：{name}") |
|
|
| elif cmd == "deluser" and len(args) >= 2: |
| name = args[1] |
| if name in auth["users"]: |
| del auth["users"][name] |
| save_auth(auth) |
| print(f"已刪除帳號：{name}") |
| else: |
| print(f"找不到帳號：{name}") |
|
|
| else: |
| print(__doc__) |
|
|
|
|
| if __name__ == "__main__": |
| cli() |
|
|
