feat(db): 配置与笔记迁入 Postgres，重启不丢

同步后端迁移：app_config / notes 两张表 + DAO，配置与笔记从本地文件改为入库。
启动 init_db 自动建表，HF 重启不再丢飞书凭证/cookie/笔记。

.gitignore

@@ -10,3 +10,4 @@ backend/app/db/*.db
 __pycache__/
 *.pyc
 .env
+backend/fonts/

backend/app/db/app_config_dao.py

@@ -0,0 +1,68 @@
+"""app_config 键值配置表的读写。
+
+取代原先各 *_config_manager 的 config/*.json 文件存储。各 manager 的公开方法
+（get_config/update_config/get/set/list_all 等）签名与行为不变，只是底层 _read/_write
+改调这里，从而让配置随 Postgres 持久化、HF 容器重启不丢。
+
+兼容旧库：首次读某个 key 时若数据库里还没有、但旧的 config/<x>.json 文件存在，
+就把文件内容一次性导入数据库后返回（幂等）。这样桌面端 / 本地 SQLite 的既有配置不丢。
+"""
+import json
+from pathlib import Path
+from typing import Any, Optional
+
+from app.db.engine import SessionLocal
+from app.db.models.app_config import AppConfig
+
+
+def get_value(key: str) -> Optional[Any]:
+    """读某个配置域的整份 value（dict 或 list）；不存在返回 None。
+
+    容错：启动早期（路由在 init_db 建表前就被 import，某些下载器会在导入期读 cookie 配置）
+    或数据库暂不可用时，按「无配置」返回 None，而不是抛错——与旧的「文件不存在/读失败返回 {}」
+    语义一致，避免把启动 import 链炸掉。
+    """
+    try:
+        with SessionLocal() as db:
+            row = db.get(AppConfig, key)
+            return row.value if row is not None else None
+    except Exception:
+        return None
+
+
+def set_value(key: str, value: Any) -> None:
+    """整份覆盖某个配置域的 value（upsert）。"""
+    with SessionLocal() as db:
+        row = db.get(AppConfig, key)
+        if row is None:
+            db.add(AppConfig(key=key, value=value))
+        else:
+            row.value = value
+        db.commit()
+
+
+def _read_legacy_file(legacy_path: str) -> Optional[Any]:
+    p = Path(legacy_path)
+    if not p.exists():
+        return None
+    try:
+        return json.loads(p.read_text(encoding="utf-8"))
+    except Exception:
+        return None
+
+
+def load_value(key: str, legacy_path: Optional[str] = None, default: Any = None) -> Any:
+    """读配置；数据库没有但旧 JSON 文件存在则一次性导入后返回；都没有返回 default。"""
+    value = get_value(key)
+    if value is not None:
+        return value
+    if legacy_path:
+        legacy = _read_legacy_file(legacy_path)
+        if legacy is not None:
+            try:
+                set_value(key, legacy)
+            except Exception:
+                # 表尚未就绪（早期 import）时先返回文件内容，待 init_db 后下次读再持久化
+                pass
+            return legacy
+    return default

backend/app/db/init_db.py

@@ -1,5 +1,11 @@
+import glob
+import json
+import os
+
+from app.db.models.app_config import AppConfig
 from app.db.models.articles import ArticleItem, ArticleSubscription, ArticleSubscriptionItem
 from app.db.models.models import Model
+from app.db.models.note import Note
 from app.db.models.providers import Provider
 from app.db.models.trend_subscription import (
     NotificationChannel,
@@ -16,6 +22,7 @@ def init_db():
     Base.metadata.create_all(bind=engine)
     _ensure_article_content_text(engine)
     _ensure_model_provider_id_text(engine)
+    _import_legacy_notes(engine)
 
 
 # 注：原 _ensure_model_columns 为 models.supports_multimodal 做的迁移已删除——
@@ -46,6 +53,46 @@ def _ensure_model_provider_id_text(engine):
         break
 
 
+def _import_legacy_notes(engine):
+    # 一次性导入：把旧的 note_results/{task_id}.json（+ {task_id}.status.json）迁进 notes 表。
+    # 仅当 notes 表为空时执行，幂等；HF 临时盘上目录为空时直接跳过。失败不影响启动。
+    from app.db.note_dao import save_note, set_status
+
+    note_dir = os.getenv("NOTE_OUTPUT_DIR", "note_results")
+    if not os.path.isdir(note_dir):
+        return
+    try:
+        with engine.connect() as conn:
+            count = conn.execute(text("SELECT COUNT(*) FROM notes")).scalar()
+        if count and int(count) > 0:
+            return
+    except Exception:
+        return
+
+    imported = 0
+    for path in glob.glob(os.path.join(note_dir, "*.json")):
+        name = os.path.basename(path)
+        stem = name[:-5]  # 去掉 .json
+        # 跳过缓存/状态/检查点：转写、音频缓存、{id}.status、{key}.gpt.checkpoint 等。
+        # 正经笔记的 task_id 是 UUID（不含点），据此排除带点的派生文件。
+        if stem.endswith("_transcript") or stem.endswith("_audio") or "." in stem:
+            continue
+        try:
+            content = json.loads(open(path, "r", encoding="utf-8").read())
+        except Exception:
+            continue
+        save_note(stem, content)
+        status_path = os.path.join(note_dir, f"{stem}.status.json")
+        if os.path.exists(status_path):
+            try:
+                set_status(stem, json.loads(open(status_path, "r", encoding="utf-8").read()))
+            except Exception:
+                pass
+        imported += 1
+    if imported:
+        print(f"[init_db] 已从本地文件导入 {imported} 篇历史笔记到数据库")
+
+
 def _ensure_article_content_text(engine):
     inspector = inspect(engine)
     if "article_items" not in inspector.get_table_names():

backend/app/db/models/app_config.py

@@ -0,0 +1,19 @@
+from sqlalchemy import Column, String, JSON, DateTime, func
+
+from app.db.engine import Base
+
+
+class AppConfig(Base):
+    """通用键值配置表：把原先散落在 config/*.json 的配置迁进数据库，
+    让飞书凭证 / 平台 cookie / 代理 / 转写设置 / 自定义平台等重启不丢。
+
+    - key:   配置域名（feishu / proxy / downloader / transcriber / custom_platforms）
+    - value: 该域的整份配置，任意 JSON（dict 或 list）。Postgres 上是 JSONB，
+             SQLite 上是 JSON 文本，由 SQLAlchemy 的 JSON 类型按方言落地。
+    """
+
+    __tablename__ = "app_config"
+
+    key = Column(String, primary_key=True)
+    value = Column(JSON, nullable=True)
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())

backend/app/db/models/note.py

@@ -0,0 +1,24 @@
+from sqlalchemy import Column, String, JSON, DateTime, func
+
+from app.db.engine import Base
+
+
+class Note(Base):
+    """笔记正文与任务状态：原先是 note_results/{task_id}.json 和
+    {task_id}.status.json 两个本地文件，迁进数据库后重启不丢。
+
+    - content: 笔记结果整体（markdown 版本数组 / transcript / audio_meta 等），
+               对应旧的 {task_id}.json。
+    - status:  任务状态字典 {status, paused, message?, cache?}，
+               对应旧的 {task_id}.status.json。
+    两者各自独立 upsert，互不覆盖（状态机先写 status，生成完才写 content）。
+    转写/音频缓存、_markdown.md、截图等可重建数据仍留在磁盘，不入库。
+    """
+
+    __tablename__ = "notes"
+
+    task_id = Column(String, primary_key=True)
+    content = Column(JSON, nullable=True)
+    status = Column(JSON, nullable=True)
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())

backend/app/db/note_dao.py

@@ -0,0 +1,53 @@
+"""notes 表的读写：笔记正文（content）与任务状态（status）。
+
+取代原先 note_results/{task_id}.json 与 {task_id}.status.json 两个本地文件。
+content 与 status 各自独立 upsert，互不覆盖。
+"""
+from typing import Any, Dict, Optional
+
+from app.db.engine import SessionLocal
+from app.db.models.note import Note
+
+
+def _upsert(task_id: str, **fields) -> None:
+    with SessionLocal() as db:
+        row = db.get(Note, task_id)
+        if row is None:
+            row = Note(task_id=task_id)
+            db.add(row)
+        for key, val in fields.items():
+            setattr(row, key, val)
+        db.commit()
+
+
+def save_note(task_id: str, content: Dict[str, Any]) -> None:
+    """写笔记正文（不动 status）。"""
+    _upsert(task_id, content=content)
+
+
+def load_note(task_id: str) -> Optional[Dict[str, Any]]:
+    """读笔记正文；不存在返回 None。"""
+    with SessionLocal() as db:
+        row = db.get(Note, task_id)
+        return row.content if row is not None and row.content is not None else None
+
+
+def set_status(task_id: str, status_data: Dict[str, Any]) -> None:
+    """写任务状态字典 {status, paused, message?, cache?}（不动 content）。"""
+    _upsert(task_id, status=status_data)
+
+
+def get_status(task_id: str) -> Optional[Dict[str, Any]]:
+    """读任务状态；不存在返回 None。"""
+    with SessionLocal() as db:
+        row = db.get(Note, task_id)
+        return row.status if row is not None and row.status is not None else None
+
+
+def delete_note(task_id: str) -> None:
+    """删除笔记行（正文 + 状态）。"""
+    with SessionLocal() as db:
+        row = db.get(Note, task_id)
+        if row is not None:
+            db.delete(row)
+            db.commit()

backend/app/routers/feishu.py

@@ -46,19 +46,16 @@ def push_task_to_feishu(task_id: str, version_id: Optional[str] = None) -> dict:
     被「手动推送」接口与「生成后自动推送」共用。任何失败都抛 FeishuError，
     由调用方决定是返回错误还是仅记日志（自动推送场景不应中断主流程）。
 
-    直接读写原始 JSON（不走 _read/_write_note_json 的版本归一化），避免把
+    直接读写原始笔记内容（不走 _read/_write_note_json 的版本归一化），避免把
     markdown 字符串就地改写成版本数组、抹掉单版本笔记的 model/style 元信息。
     """
-    import json
-
     # 延迟导入避免与 note 路由的循环依赖；note 路由不在模块级 import 本模块
-    from app.routers.note import NOTE_OUTPUT_DIR, _pick_markdown_version
+    from app.routers.note import _pick_markdown_version
+    from app.db.note_dao import load_note, save_note
 
-    path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json")
-    if not os.path.exists(path):
+    data = load_note(task_id)
+    if data is None:
         raise FeishuError(f"笔记不存在或尚未生成完成：{task_id}")
-    with open(path, "r", encoding="utf-8") as f:
-        data = json.load(f)
 
     # _pick_markdown_version 兼容旧 str 与多版本 list 两种格式，仅读取不改写
     content = _pick_markdown_version(data.get("markdown"), version_id)
@@ -82,8 +79,7 @@ def push_task_to_feishu(task_id: str, version_id: Optional[str] = None) -> dict:
         "pushed_at": datetime.now().isoformat(timespec="seconds"),
     }
     data["feishu"] = feishu_info
-    with open(path, "w", encoding="utf-8") as f:
-        json.dump(data, f, ensure_ascii=False, indent=2)
+    save_note(task_id, data)
     return feishu_info
 
 

backend/app/routers/note.py

@@ -10,6 +10,7 @@ from fastapi import APIRouter, HTTPException, BackgroundTasks, UploadFile, File
 from pydantic import BaseModel, validator, field_validator
 from dataclasses import asdict
 
+from app.db.note_dao import save_note, load_note, get_status
 from app.db.video_task_dao import get_task_by_video
 from app.enmus.exception import NoteErrorEnum
 from app.enmus.note_enums import DownloadQuality
@@ -74,9 +75,8 @@ UPLOAD_DIR = "uploads"
 
 
 def save_note_to_file(task_id: str, note):
-    os.makedirs(NOTE_OUTPUT_DIR, exist_ok=True)
-    with open(os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json"), "w", encoding="utf-8") as f:
-        json.dump(asdict(note), f, ensure_ascii=False, indent=2)
+    # 笔记正文持久化到数据库 notes 表（原先写 note_results/{task_id}.json）。
+    save_note(task_id, asdict(note))
 
 
 def _persist_prefetched_transcript(task_id: str, transcript: dict) -> None:
@@ -227,22 +227,17 @@ def _pick_markdown_version(markdown_field, version_id: Optional[str]) -> str:
 
 
 def _read_note_json(task_id: str) -> dict:
-    """读笔记 JSON 文件，把 markdown 字段就地归一化成版本数组。文件不存在抛 HTTPException。"""
-    path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json")
-    if not os.path.exists(path):
+    """读笔记，把 markdown 字段就地归一化成版本数组。不存在抛 HTTPException。"""
+    data = load_note(task_id)
+    if data is None:
         raise HTTPException(status_code=404, detail="笔记不存在")
-    with open(path, "r", encoding="utf-8") as f:
-        data = json.load(f)
     data["markdown"] = _normalize_versions(data.get("markdown"))
     return data
 
 
 def _write_note_json(task_id: str, data: dict) -> None:
-    """写回笔记 JSON 文件。markdown 一定是 list 形式。"""
-    path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json")
-    os.makedirs(NOTE_OUTPUT_DIR, exist_ok=True)
-    with open(path, "w", encoding="utf-8") as f:
-        json.dump(data, f, ensure_ascii=False, indent=2)
+    """写回笔记到数据库 notes 表。markdown 一定是 list 形式。"""
+    save_note(task_id, data)
 
 
 def _append_version(
@@ -284,13 +279,10 @@ def export_note(task_id: str, format: str = "markdown", version_id: Optional[str
     - format: markdown / pdf / html / word / docx（image / png 暂不支持）
     - version_id: 多版本时指定某一版；不传取最新版（v1 单版兼容旧 str 格式）
     """
-    note_path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json")
-    if not os.path.exists(note_path):
+    note = load_note(task_id)
+    if note is None:
         raise HTTPException(status_code=404, detail="笔记不存在")
 
-    with open(note_path, "r", encoding="utf-8") as f:
-        note = json.load(f)
-
     content = _pick_markdown_version(note.get("markdown"), version_id)
     if not content.strip():
         raise HTTPException(status_code=404, detail="笔记内容为空")
@@ -516,14 +508,12 @@ def generate_note(data: VideoRequest, background_tasks: BackgroundTasks):
 
 @router.get("/task_status/{task_id}")
 def get_task_status(task_id: str):
-    status_path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.status.json")
-    result_path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json")
-
-    # 优先读状态文件
-    if os.path.exists(status_path):
-        with open(status_path, "r", encoding="utf-8") as f:
-            status_content = json.load(f)
+    # 状态与结果都从数据库 notes 表读（原先读 {task_id}.status.json / {task_id}.json）
+    status_content = get_status(task_id)
+    result_content = load_note(task_id)
 
+    # 优先读状态
+    if status_content:
         status = status_content.get("status")
         message = status_content.get("message", "")
         paused = bool(status_content.get("paused", False))
@@ -531,9 +521,7 @@ def get_task_status(task_id: str):
 
         if status == TaskStatus.SUCCESS.value:
             # 成功状态的话，继续读取最终笔记内容
-            if os.path.exists(result_path):
-                with open(result_path, "r", encoding="utf-8") as rf:
-                    result_content = json.load(rf)
+            if result_content is not None:
                 return R.success({
                     "status": status,
                     "result": result_content,
@@ -562,10 +550,8 @@ def get_task_status(task_id: str):
             "task_id": task_id
         })
 
-    # 没有状态文件，但有结果
-    if os.path.exists(result_path):
-        with open(result_path, "r", encoding="utf-8") as f:
-            result_content = json.load(f)
+    # 没有状态，但有结果
+    if result_content is not None:
         return R.success({
             "status": TaskStatus.SUCCESS.value,
             "result": result_content,

backend/app/services/article.py

@@ -10,6 +10,7 @@ from app.article_fetchers.base import ArticleContent, ArticleFetcher
 from app.article_fetchers.generic import GenericArticleFetcher
 from app.article_fetchers.wechat import WechatArticleFetcher
 from app.article_fetchers.xiaohongshu import XiaohongshuArticleFetcher
+from app.db.note_dao import load_note, save_note, set_status
 from app.db.article_dao import (
     create_subscription,
     get_article_item,
@@ -284,12 +285,8 @@ class ArticleService:
     def _content_from_note_result(self, task_id: str) -> str:
         if not task_id:
             return ""
-        result_path = _note_output_dir() / f"{task_id}.json"
-        if not result_path.exists():
-            return ""
-        try:
-            payload = json.loads(result_path.read_text(encoding="utf-8"))
-        except Exception:
+        payload = load_note(task_id)
+        if not payload:
             return ""
         transcript = payload.get("transcript") or {}
         return str(transcript.get("full_text") or "").strip()
@@ -366,17 +363,10 @@ class ArticleService:
             },
             "total_tokens": total_tokens,
         }
-        (_note_output_dir() / f"{task_id}.json").write_text(
-            json.dumps(payload, ensure_ascii=False, indent=2),
-            encoding="utf-8",
-        )
+        save_note(task_id, payload)
 
     def _update_status(self, task_id: str, status: TaskStatus) -> None:
-        payload = {"status": status.value, "paused": False}
-        (_note_output_dir() / f"{task_id}.status.json").write_text(
-            json.dumps(payload, ensure_ascii=False, indent=2),
-            encoding="utf-8",
-        )
+        set_status(task_id, {"status": status.value, "paused": False})
 
     def _index_task(self, task_id: str) -> None:
         try:

backend/app/services/chat_service.py

@@ -8,6 +8,7 @@ from app.models.model_config import ModelConfig
 from app.services.provider import ProviderService
 from app.services.vector_store import VectorStoreManager, NOTE_OUTPUT_DIR
 from app.services.chat_tools import TOOLS, execute_tool
+from app.db.note_dao import load_note
 from app.utils.logger import get_logger
 
 logger = get_logger(__name__)
@@ -15,13 +16,8 @@ logger = get_logger(__name__)
 
 def _load_task_brief(task_id: str) -> dict:
     """读出某篇笔记的标题/平台/URL，用于源卡片展示。失败返回空 dict。"""
-    path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json")
-    if not os.path.exists(path):
-        return {}
-    try:
-        with open(path, "r", encoding="utf-8") as f:
-            data = json.load(f)
-    except Exception:
+    data = load_note(task_id)
+    if not data:
         return {}
     am = data.get("audio_meta", {}) or {}
     raw = am.get("raw_info", {}) or {}

backend/app/services/chat_tools.py

@@ -15,11 +15,8 @@ NOTE_OUTPUT_DIR = os.getenv("NOTE_OUTPUT_DIR", "note_results")
 
 
 def _load_note_data(task_id: str) -> Optional[dict]:
-    path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json")
-    if not os.path.exists(path):
-        return None
-    with open(path, "r", encoding="utf-8") as f:
-        return json.load(f)
+    from app.db.note_dao import load_note
+    return load_note(task_id)
 
 
 # ── 工具定义（OpenAI function calling 格式）──────────────────────

backend/app/services/cookie_manager.py

@@ -1,25 +1,21 @@
-import json
-from pathlib import Path
 from typing import Optional, Dict
 
+from app.db.app_config_dao import load_value, set_value
+
 
 class CookieConfigManager:
+    # 平台 cookie 持久化在数据库 app_config 表（key="downloader"）；
+    # filepath 仅用于把旧的 config/downloader.json 一次性导入。
+    _KEY = "downloader"
+
     def __init__(self, filepath: str = "config/downloader.json"):
-        self.path = Path(filepath)
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        if not self.path.exists():
-            self._write({})
+        self._legacy_path = filepath
 
     def _read(self) -> Dict[str, Dict[str, str]]:
-        try:
-            with self.path.open("r", encoding="utf-8") as f:
-                return json.load(f)
-        except Exception:
-            return {}
+        return load_value(self._KEY, self._legacy_path, {}) or {}
 
     def _write(self, data: Dict[str, Dict[str, str]]):
-        with self.path.open("w", encoding="utf-8") as f:
-            json.dump(data, f, ensure_ascii=False, indent=2)
+        set_value(self._KEY, data)
 
     def get(self, platform: str) -> Optional[str]:
         data = self._read()

backend/app/services/custom_platform_manager.py

@@ -6,34 +6,29 @@
   - name:  展示名。
   - match: URL 子串匹配。如 "vimeo.com"。命中即视为该平台。
 """
-import json
 import re
-from pathlib import Path
 from typing import Optional
 
+from app.db.app_config_dao import load_value, set_value
 
-_PATH = Path("config/custom_platforms.json")
+
+# 自定义平台持久化在数据库 app_config 表（key="custom_platforms"，value 是列表）。
+# _LEGACY_PATH 仅用于把旧的 config/custom_platforms.json 一次性导入。
+_KEY = "custom_platforms"
+_LEGACY_PATH = "config/custom_platforms.json"
 _KEY_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{1,31}$")
 
 
 def _read() -> list[dict]:
-    if not _PATH.exists():
-        return []
-    try:
-        data = json.loads(_PATH.read_text(encoding="utf-8"))
-        if isinstance(data, dict):
-            data = data.get("platforms", [])
-        return [p for p in (data or []) if isinstance(p, dict) and p.get("key")]
-    except Exception:
-        return []
+    data = load_value(_KEY, _LEGACY_PATH, [])
+    # 兼容旧文件 {"platforms": [...]} 的形态（导入后首次读到的就是这个 dict）。
+    if isinstance(data, dict):
+        data = data.get("platforms", [])
+    return [p for p in (data or []) if isinstance(p, dict) and p.get("key")]
 
 
 def _write(items: list[dict]) -> None:
-    _PATH.parent.mkdir(parents=True, exist_ok=True)
-    _PATH.write_text(
-        json.dumps({"platforms": items}, ensure_ascii=False, indent=2),
-        encoding="utf-8",
-    )
+    set_value(_KEY, items)
 
 
 def list_all() -> list[dict]:

backend/app/services/feishu_config_manager.py

@@ -1,8 +1,8 @@
-import json
 import os
-from pathlib import Path
 from typing import Any, Dict, Optional
 
+from app.db.app_config_dao import load_value, set_value
+
 # 飞书 / Lark 开放平台默认域名。海外租户用 open.larksuite.com，
 # 国内租户用 open.feishu.cn（默认）。用户可在设置页切换。
 DEFAULT_FEISHU_BASE_URL = "https://open.feishu.cn"
@@ -16,22 +16,18 @@ class FeishuConfigManager:
     返回给前端时通过 get_public_config() 隐去明文。
     """
 
+    # 配置现在持久化在数据库 app_config 表（key="feishu"）。filepath 仅用于把旧的
+    # config/feishu.json 一次性导入数据库，保证桌面端/本地库的既有配置不丢。
+    _KEY = "feishu"
+
     def __init__(self, filepath: str = "config/feishu.json"):
-        self.path = Path(filepath)
-        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._legacy_path = filepath
 
     def _read(self) -> Dict[str, Any]:
-        if not self.path.exists():
-            return {}
-        try:
-            with self.path.open("r", encoding="utf-8") as f:
-                return json.load(f)
-        except Exception:
-            return {}
+        return load_value(self._KEY, self._legacy_path, {}) or {}
 
     def _write(self, data: Dict[str, Any]):
-        with self.path.open("w", encoding="utf-8") as f:
-            json.dump(data, f, ensure_ascii=False, indent=2)
+        set_value(self._KEY, data)
 
     def get_config(self) -> Dict[str, Any]:
         """内部使用：含 app_secret 明文。"""

backend/app/services/note.py

@@ -15,6 +15,7 @@ from app.downloaders.bilibili_downloader import BilibiliDownloader
 from app.downloaders.douyin_downloader import DouyinDownloader
 from app.downloaders.local_downloader import LocalDownloader
 from app.downloaders.youtube_downloader import YoutubeDownloader
+from app.db.note_dao import load_note, set_status
 from app.db.video_task_dao import delete_task_by_video, insert_video_task
 from app.enmus.exception import NoteErrorEnum, ProviderErrorEnum
 from app.enmus.task_status_enums import TaskStatus
@@ -511,9 +512,7 @@ class NoteGenerator:
         if not task_id:
             return
 
-        NOTE_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
-        status_file = NOTE_OUTPUT_DIR / f"{task_id}.status.json"
-        print(f"写入状态文件: {status_file} 当前状态: {status} paused={paused} cache={cache}")
+        # 任务状态持久化到数据库 notes 表的 status 字段（原先写 {task_id}.status.json）。
         data = {"status": status.value if isinstance(status, TaskStatus) else status, "paused": paused}
         if message:
             data["message"] = message
@@ -521,25 +520,9 @@ class NoteGenerator:
             data["cache"] = cache
 
         try:
-            # First create a temporary file
-            temp_file = status_file.with_suffix('.tmp')
-
-            # Write to temporary file
-            with temp_file.open('w', encoding='utf-8') as f:
-                json.dump(data, f, ensure_ascii=False, indent=2)
-
-            # Atomic rename operation
-            temp_file.replace(status_file)
-
-            print(f"状态文件写入成功: {status_file}")
+            set_status(task_id, data)
         except Exception as e:
-            logger.error(f"写入状态文件失败 (task_id={task_id})：{e}")
-            # Try to write error to file directly as fallback
-            try:
-                with status_file.open('w', encoding='utf-8') as f:
-                    f.write(f"Error writing status: {str(e)}")
-            except:
-                logger.error(f"写入错误  {e}")
+            logger.error(f"写入任务状态失败 (task_id={task_id})：{e}")
 
     def _handle_exception(self, task_id, exc):
         logger.error(f"任务异常 (task_id={task_id})", exc_info=True)
@@ -896,13 +879,12 @@ class NoteGenerator:
         - 跳过下载、转写、截图替换等重型环节
         - 只调 LLM 拿一段新 markdown 文本返回（由路由层负责追加到版本列表）
         """
-        note_path = NOTE_OUTPUT_DIR / f"{task_id}.json"
-        if not note_path.exists():
+        data = load_note(task_id)
+        if data is None:
             raise NoteError(
                 code=NoteErrorEnum.PLATFORM_NOT_SUPPORTED.code,
                 message=f"笔记不存在：{task_id}",
             )
-        data = json.loads(note_path.read_text(encoding="utf-8"))
 
         audio_meta_d = data.get("audio_meta") or {}
         transcript_d = data.get("transcript") or {}

backend/app/services/proxy_config_manager.py

@@ -1,8 +1,8 @@
-import json
 import os
-from pathlib import Path
 from typing import Any, Dict, Optional
 
+from app.db.app_config_dao import load_value, set_value
+
 
 class ProxyConfigManager:
     """全局代理配置，存 JSON 文件，支持前端动态修改。
@@ -12,22 +12,17 @@ class ProxyConfigManager:
     这样桌面端/web 用户在设置页填，docker/服务器部署用环境变量兜底。
     """
 
+    # 配置持久化在数据库 app_config 表（key="proxy"）；filepath 仅用于旧文件一次性导入。
+    _KEY = "proxy"
+
     def __init__(self, filepath: str = "config/proxy.json"):
-        self.path = Path(filepath)
-        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._legacy_path = filepath
 
     def _read(self) -> Dict[str, Any]:
-        if not self.path.exists():
-            return {}
-        try:
-            with self.path.open("r", encoding="utf-8") as f:
-                return json.load(f)
-        except Exception:
-            return {}
+        return load_value(self._KEY, self._legacy_path, {}) or {}
 
     def _write(self, data: Dict[str, Any]):
-        with self.path.open("w", encoding="utf-8") as f:
-            json.dump(data, f, ensure_ascii=False, indent=2)
+        set_value(self._KEY, data)
 
     def get_config(self) -> Dict[str, Any]:
         data = self._read()

backend/app/services/transcriber_config_manager.py

@@ -1,28 +1,23 @@
-import json
 import os
-from pathlib import Path
 from typing import Optional, Dict, Any
 
+from app.db.app_config_dao import load_value, set_value
+
 
 class TranscriberConfigManager:
-    """管理转写器配置，存储在 JSON 文件中，支持前端动态修改。"""
+    """管理转写器配置，持久化在数据库 app_config 表（key="transcriber"），支持前端动态修改。"""
+
+    # filepath 仅用于把旧的 config/transcriber.json 一次性导入数据库。
+    _KEY = "transcriber"
 
     def __init__(self, filepath: str = "config/transcriber.json"):
-        self.path = Path(filepath)
-        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._legacy_path = filepath
 
     def _read(self) -> Dict[str, Any]:
-        if not self.path.exists():
-            return {}
-        try:
-            with self.path.open("r", encoding="utf-8") as f:
-                return json.load(f)
-        except Exception:
-            return {}
+        return load_value(self._KEY, self._legacy_path, {}) or {}
 
     def _write(self, data: Dict[str, Any]):
-        with self.path.open("w", encoding="utf-8") as f:
-            json.dump(data, f, ensure_ascii=False, indent=2)
+        set_value(self._KEY, data)
 
     def get_config(self) -> Dict[str, Any]:
         """获取当前转写器配置，fallback 到环境变量默认值。

backend/app/services/vector_store.py

@@ -6,6 +6,7 @@ from typing import Optional
 import chromadb
 from chromadb.config import Settings
 
+from app.db.note_dao import load_note
 from app.utils.logger import get_logger
 
 logger = get_logger(__name__)
@@ -117,14 +118,11 @@ class VectorStoreManager:
 
     def index_task(self, task_id: str) -> None:
         """读取笔记结果并建立向量索引。"""
-        result_path = os.path.join(NOTE_OUTPUT_DIR, f"{task_id}.json")
-        if not os.path.exists(result_path):
-            logger.warning(f"笔记文件不存在，跳过索引: {result_path}")
+        note_data = load_note(task_id)
+        if note_data is None:
+            logger.warning(f"笔记不存在，跳过索引: {task_id}")
             return
 
-        with open(result_path, "r", encoding="utf-8") as f:
-            note_data = json.load(f)
-
         markdown = note_data.get("markdown", "")
         transcript = note_data.get("transcript", {})
         segments = transcript.get("segments", [])

backend/tests/conftest.py

@@ -0,0 +1,16 @@
+import os
+import tempfile
+
+# 把数据库与笔记输出目录强制指向临时位置：
+#  - 避免测试污染真实的 video_memo.db / note_results（配置/笔记已迁入数据库后，
+#    管理器与 note_dao 的读写都走全局 engine，不隔离会写进开发库）。
+#  - 顺带建好 app_config / notes 等新表，供依赖数据库的单测使用。
+# 必须在任何 app 模块导入前设置：pytest 最先加载 conftest，此处 import app.db.* 时
+# engine 才会按这里的 DATABASE_URL 初始化。load_dotenv(override=False) 不会覆盖已设的值。
+_TEST_DIR = tempfile.mkdtemp(prefix="videomemo-test-")
+os.environ["DATABASE_URL"] = "sqlite:///" + os.path.join(_TEST_DIR, "test.db")
+os.environ.setdefault("NOTE_OUTPUT_DIR", os.path.join(_TEST_DIR, "note_results"))
+
+from app.db.init_db import init_db  # noqa: E402
+
+init_db()

backend/tests/test_article_service.py

@@ -99,9 +99,12 @@ def test_generate_from_url_saves_note_json(tmp_path, monkeypatch):
         task_id="task-1",
     )
 
-    saved = json.loads((tmp_path / "notes" / "task-1.json").read_text(encoding="utf-8"))
-    status = json.loads((tmp_path / "notes" / "task-1.status.json").read_text(encoding="utf-8"))
+    # 笔记正文与状态现在存数据库 notes 表（原先是 note_results/{task_id}.json 文件）
+    import app.db.note_dao as note_dao
+    saved = note_dao.load_note("task-1")
+    status = note_dao.get_status("task-1")
     assert result["task_id"] == "task-1"
+    assert saved is not None
     assert saved["markdown"] == "# 总结\n\n- 要点"
     assert saved["transcript"]["full_text"] == "正文内容"
     assert saved["audio_meta"]["title"] == "文章标题"