john
e76fa586f1
Ship dual-track learning (daily accumulation vs textbook),沪教/商务词书 APIs and UI, native iOS wrapper with bundled H5, and production book import on deploy. Co-authored-by: Cursor <cursoragent@cursor.com>
107 lines
3.4 KiB
Python
107 lines
3.4 KiB
Python
"""将 backend/data/books/*.json 导入 word_books / word_book_entries 表。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
BACKEND_DIR = Path(__file__).resolve().parents[1]
|
|
if str(BACKEND_DIR) not in sys.path:
|
|
sys.path.insert(0, str(BACKEND_DIR))
|
|
|
|
from database import SessionLocal # noqa: E402
|
|
from models import WordBook, WordBookEntry # noqa: E402
|
|
from services.book_service import DEPRECATED_SLUGS # noqa: E402
|
|
|
|
BOOKS_DIR = BACKEND_DIR / "data" / "books"
|
|
|
|
|
|
def upsert_book(session, data: dict) -> WordBook:
|
|
book = session.query(WordBook).filter(WordBook.slug == data["slug"]).first()
|
|
if not book:
|
|
book = WordBook(slug=data["slug"])
|
|
session.add(book)
|
|
book.title = data["title"]
|
|
book.description = data.get("description")
|
|
book.level = data.get("level", "general")
|
|
book.word_count = data.get("word_count", 0)
|
|
book.unit_count = data.get("unit_count", 0)
|
|
book.sort_order = data.get("sort_order", 0)
|
|
book.daily_target = data.get("daily_target", 15)
|
|
book.master_required_count = data.get("master_required_count", 4)
|
|
book.weak_wrong_threshold = data.get("weak_wrong_threshold", 2)
|
|
book.learn_mode = data.get("learn_mode", "sequential")
|
|
book.is_published = 1
|
|
session.flush()
|
|
return book
|
|
|
|
|
|
def import_book_file(session, path: Path) -> tuple[str, int]:
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
book = upsert_book(session, data)
|
|
session.query(WordBookEntry).filter(WordBookEntry.book_id == book.id).delete(
|
|
synchronize_session=False
|
|
)
|
|
count = 0
|
|
for unit in data.get("units", []):
|
|
unit_no = unit.get("unit", 1)
|
|
unit_title = unit.get("title")
|
|
for w in unit.get("words", []):
|
|
entry = WordBookEntry(
|
|
book_id=book.id,
|
|
unit=unit_no,
|
|
unit_title=unit_title,
|
|
sort_index=w["sort_index"],
|
|
lemma_en=w["lemma_en"].strip().lower()[:128],
|
|
zh=w["zh"].strip()[:256],
|
|
phonetic=(w.get("phonetic") or "")[:128] or None,
|
|
example_en=w.get("example_en"),
|
|
example_cn=w.get("example_cn"),
|
|
)
|
|
session.add(entry)
|
|
count += 1
|
|
book.word_count = count
|
|
book.unit_count = len(data.get("units", []))
|
|
session.commit()
|
|
return book.slug, count
|
|
|
|
|
|
def import_all(books_dir: Path = BOOKS_DIR) -> None:
|
|
files = sorted(books_dir.glob("*.json"))
|
|
files = [
|
|
f
|
|
for f in files
|
|
if f.name != "manifest.json"
|
|
and f.stem not in DEPRECATED_SLUGS
|
|
]
|
|
if not files:
|
|
print(f"未找到词书 JSON: {books_dir}")
|
|
return
|
|
session = SessionLocal()
|
|
try:
|
|
for path in files:
|
|
slug, count = import_book_file(session, path)
|
|
print(f" 导入 {slug}: {count} 词条")
|
|
for slug in DEPRECATED_SLUGS:
|
|
book = session.query(WordBook).filter(WordBook.slug == slug).first()
|
|
if book and book.is_published:
|
|
book.is_published = 0
|
|
session.commit()
|
|
print(f" 下架旧词书 {slug}")
|
|
finally:
|
|
session.close()
|
|
|
|
|
|
def main() -> None:
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description="导入词书 JSON 到数据库")
|
|
parser.add_argument("--dir", type=Path, default=BOOKS_DIR)
|
|
args = parser.parse_args()
|
|
import_all(args.dir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|