import gradio as gr import os import json from huggingface_hub import HfApi, hf_hub_download from datetime import datetime import shutil from pathlib import Path from uuid import uuid4 from zoneinfo import ZoneInfo # --- 配置 (优先从环境变量读取) --- DATASET_REPO_ID = os.environ.get("DATASET_REPO_ID", "mingyang22/huggingface-notes") HF_TOKEN = os.environ.get("HF_TOKEN") # 必须在 Space 设置中配置 REMOTE_NOTES_PATH = "db/notes.json" BEIJING_TZ = ZoneInfo("Asia/Shanghai") PWA_HEAD = """ """ def get_default_data_dir(): # 如果在 Space 环境,优先使用当前目录下的 cache_data 文件夹,避免 /root 权限问题 if os.environ.get("SPACE_ID") or os.environ.get("HF_SPACE"): return str(Path.cwd() / "cache_data") custom_dir = os.environ.get("HF_NOTES_DATA_DIR") if custom_dir: return custom_dir if os.name == "nt": base = os.environ.get("LOCALAPPDATA") or str(Path.home() / "AppData" / "Local") else: base = os.environ.get("XDG_DATA_HOME") or str(Path.home() / ".local" / "share") return str(Path(base) / "hf-note-app-pro") DATA_DIR = get_default_data_dir() LOCAL_NOTES_PATH = str(Path(DATA_DIR) / "notes.json") def ensure_local_notes(): Path(DATA_DIR).mkdir(parents=True, exist_ok=True) p = Path(LOCAL_NOTES_PATH) if not p.exists(): p.write_text("[]", encoding="utf-8") def read_notes(): ensure_local_notes() try: data = json.loads(Path(LOCAL_NOTES_PATH).read_text(encoding="utf-8-sig")) if isinstance(data, list): valid_notes = [] for item in data: if not isinstance(item, dict): continue # 关键修复:同时支持 C# 风格 (Uppercase) 和 Python 风格 (Lowercase) 的键名 n_id = item.get("Id") or item.get("id", "") n_title = item.get("Title") or item.get("title", "") n_content = item.get("Content") or item.get("content", "") n_updated = item.get("UpdatedAt") or item.get("updated_at", "") n_pinned = item.get("IsPinned") if "IsPinned" in item else item.get("is_pinned", False) n_deleted = item.get("IsDeleted") if "IsDeleted" in item else item.get("is_deleted", False) valid_notes.append({ "id": str(n_id), "title": str(n_title), "content": str(n_content), "updated_at": str(n_updated), "is_pinned": bool(n_pinned), "is_deleted": bool(n_deleted) }) return valid_notes except Exception as e: print(f"读取笔记失败: {e}") return [] def write_notes(notes): ensure_local_notes() Path(LOCAL_NOTES_PATH).write_text( json.dumps(notes, ensure_ascii=False, indent=2), encoding="utf-8", ) def now_beijing(): return datetime.now(BEIJING_TZ) # --- 持久化管理 --- class CloudSync: def __init__(self): self.api = HfApi(token=HF_TOKEN) def pull(self): try: ensure_local_notes() print(f"🔄 正在从 Dataset {DATASET_REPO_ID} 拉取 {REMOTE_NOTES_PATH}...") downloaded_path = hf_hub_download( repo_id=DATASET_REPO_ID, filename=REMOTE_NOTES_PATH, repo_type="dataset", token=HF_TOKEN, force_download=True, revision="main", ) shutil.copy(downloaded_path, LOCAL_NOTES_PATH) return True, f"✅ 云端拉取同步完成" except Exception as e: msg = str(e) print(f"拉取失败详情: {msg}") # 如果是 401/404,通常是 Token 没设或权限问题 if "401" in msg or "404" in msg: return False, f"⚠️ 拉取失败: 请检查 Space 的 HF_TOKEN 是否已正确配置 (Dataset 可能为私有)" return False, f"⚠️ 拉取失败: {msg}" def push(self): ensure_local_notes() if not os.path.exists(LOCAL_NOTES_PATH): return False, "❌ 文件丢失" try: self.api.upload_file( path_or_fileobj=LOCAL_NOTES_PATH, path_in_repo=REMOTE_NOTES_PATH, repo_id=DATASET_REPO_ID, repo_type="dataset", commit_message=f"Web Update Pro at {now_beijing().strftime('%Y-%m-%d %H:%M:%S %z')}" ) return True, "✅ 已备份至云端" except Exception as e: return False, f"❌ 备份失败: {e}" sync_manager = CloudSync() # --- 业务逻辑 --- def load_notes_list(filter_type="all", search_query=""): notes = read_notes() query = search_query.lower() if search_query else "" filtered = [] for n in notes: # Tab 过滤 is_deleted = n.get("is_deleted", False) is_pinned = n.get("is_pinned", False) if filter_type == "trash": if not is_deleted: continue else: if is_deleted: continue if filter_type == "pinned" and not is_pinned: continue # 搜索过滤 if query and query not in n["title"].lower() and query not in n["content"].lower(): continue filtered.append(n) # 排序:置顶优先,时间倒序 sorted_notes = sorted(filtered, key=lambda x: (x.get("is_pinned", False), x.get("updated_at", "")), reverse=True) return [ [n["id"], f"{'📌 ' if n.get('is_pinned') else ''}{n['title'] or '未命名'}", n["updated_at"]] for n in sorted_notes ] def get_note_detail(note_id): if not note_id: return "", "", "" notes = read_notes() for n in notes: if n["id"] == note_id: return n["title"], n["content"], n["updated_at"] return "", "", "" def handle_save(note_id, title, content, push_cloud=False): if not title and not content: return "无内容可保存", load_notes_list(), note_id notes = read_notes() now = now_beijing().isoformat(timespec="seconds") found = False for n in notes: if n["id"] == note_id: n["title"], n["content"], n["updated_at"] = title, content, now found = True break if not found: new_id = uuid4().hex new_note = { "id": new_id, "title": title or "新笔记", "content": content, "updated_at": now, "is_pinned": False, "is_deleted": False } notes.insert(0, new_note) note_id = new_id write_notes(notes) if push_cloud: _, msg = sync_manager.push() status = f"已保存并同步 | {msg}" else: status = "已自动保存到本地" return status, load_notes_list(), note_id def handle_delete(note_id, current_filter): if not note_id: return "未选择笔记", load_notes_list(current_filter), "" notes = read_notes() for n in notes: if n["id"] == note_id: if current_filter == "trash": notes.remove(n) else: n["is_deleted"] = True n["is_pinned"] = False break write_notes(notes) sync_manager.push() return "已移至回收站" if current_filter != "trash" else "已彻底删除", load_notes_list(current_filter), "" def handle_pin(note_id, current_filter): if not note_id: return load_notes_list(current_filter) notes = read_notes() for n in notes: if n["id"] == note_id: n["is_pinned"] = not n.get("is_pinned", False) break write_notes(notes) sync_manager.push() return load_notes_list(current_filter) # --- Gradio UI --- with gr.Blocks(theme=gr.themes.Default(), head=PWA_HEAD) as demo: current_filter_state = gr.State("all") selected_note_id = gr.State("") with gr.Row(equal_height=True): # 1. 导航栏 (ClassNote 风格) with gr.Column(scale=1, min_width=150): gr.HTML("