163 lines
8.8 KiB
Python
Executable File
163 lines
8.8 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""books-sync — заметки и рекапы читалки books.ruzzy.dev → Obsidian-vault владельца (папка «Книги/»).
|
||
|
||
Раз в 10 мин (books-sync.timer, юзер astra): GET /api/export с токеном «export:obsidian-rpi» (/etc/books-sync/token);
|
||
если версия не изменилась — ничего не делаем. Файл на книгу: «Книги/<название>.md», главы — разделами (рекап, заметки).
|
||
Диалоги с ИИ целиком не выгружаются (владелец 2026-10-18): нужное сохраняется кнопкой «В заметки» в окне объяснения —
|
||
ИИ сжимает разговор в обычную заметку, она и приходит сюда. Всё выше заголовка «## Мои мысли» перезаписывается из читалки, сам раздел и всё ниже — не трогаем.
|
||
Синхронизация односторонняя: правки заметок — в читалке. Только stdlib.
|
||
|
||
books-sync синхронизировать (таймер)
|
||
books-sync --force переписать файлы, даже если версия та же
|
||
"""
|
||
import datetime as dt
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.request
|
||
from pathlib import Path
|
||
from zoneinfo import ZoneInfo
|
||
|
||
URL = os.environ.get("BOOKS_URL", "https://books.ruzzy.dev")
|
||
TOKEN = Path(os.environ.get("BOOKS_SYNC_TOKEN", "/etc/books-sync/token"))
|
||
STATE = Path(os.environ.get("BOOKS_SYNC_STATE", "/var/lib/books-sync/state.json"))
|
||
VAULT = Path(os.environ.get("ASTRA_VAULT", "/home/astra/vault"))
|
||
FOLDER = "Книги"
|
||
MINE = "## Мои мысли"
|
||
QUOTE_MAX = 1500
|
||
TZ = ZoneInfo("Asia/Nicosia")
|
||
KIND = {"note": "📝 Заметка", "mine": "✎ Стикер", "later": "⏳ Разобрать позже"}
|
||
# обвязка книги — в Obsidian не нужна (владелец 2026-10-18): обложка, выходные данные, предисловия, указатель…
|
||
FRONT_BACK = re.compile(r"(?i)^\s*(cover|copyright|table of contents|contents|preface|foreword|acknowledg\w*|index|"
|
||
r"about the authors?|about this book|colophon|dedication|начало книги|обложка|титул|предисловие|"
|
||
r"благодарност\w*|об этой книге|об автор\w*|иллюстрация на обложке|от издательства|оглавление|"
|
||
r"содержание|предметный указатель|указатель|выходные данные|посвящение)\b")
|
||
|
||
|
||
def fetch(version):
|
||
req = urllib.request.Request(f"{URL}/api/export?v={version}",
|
||
headers={"Authorization": "Bearer " + TOKEN.read_text().strip()})
|
||
with urllib.request.urlopen(req, timeout=60) as r:
|
||
return json.load(r)
|
||
|
||
|
||
def safe(name, n=90):
|
||
name = re.split(r"\s+--\s+", name or "")[0] # «Название -- Автор -- 2024 -- …» из имени файла
|
||
name = re.sub(r'[\\/:*?"<>|#^\[\]\x00-\x1f]', " ", name).strip(" .")
|
||
return re.sub(r"\s+", " ", name)[:n].rstrip(" .") or "Без названия"
|
||
|
||
|
||
def quote(text):
|
||
text = (text or "").strip()
|
||
if len(text) > QUOTE_MAX:
|
||
text = text[:QUOTE_MAX].rstrip() + " …"
|
||
return "\n".join("> " + l if l.strip() else ">" for l in text.splitlines())
|
||
|
||
|
||
def demote(md):
|
||
"""Заголовки внутри рекапа/ответа ИИ — ниже уровня главы, чтобы не ломать структуру файла."""
|
||
return re.sub(r"(?m)^(#{1,4}) ", lambda m: "#" * min(6, len(m.group(1)) + 3) + " ", (md or "").strip())
|
||
|
||
|
||
def day(ts):
|
||
try:
|
||
return dt.datetime.fromisoformat(ts).strftime("%d.%m.%Y")
|
||
except (TypeError, ValueError):
|
||
return ""
|
||
|
||
|
||
def render(site, b, chapters, notes, recaps, dialogs):
|
||
url = f"{site}/book/{b['id']}"
|
||
title = re.split(r"\s+--\s+", b["title"])[0]
|
||
out = ["---", f"книга: {json.dumps(title, ensure_ascii=False)}"]
|
||
if b.get("author"):
|
||
out.append(f"автор: {json.dumps(b['author'], ensure_ascii=False)}")
|
||
out += [f"читалка: {url}", "источник: books-sync — всё выше «Мои мысли» перезаписывается из читалки",
|
||
"tags: [книга]", "---", "", f"# {title}", ""]
|
||
out.append(" · ".join(x for x in (b.get("author") or "", f"[Открыть в читалке]({url})") if x))
|
||
idxs = sorted({x["chapter_idx"] for x in notes + recaps + dialogs})
|
||
for idx in idxs:
|
||
ch = chapters.get(idx, {"title": f"Глава {idx}"})
|
||
out += ["", f"## {ch['title']}", f"[в читалке]({url}/ch/{idx})"]
|
||
r = [x for x in recaps if x["chapter_idx"] == idx]
|
||
if r:
|
||
out += ["", "### Рекап", "", demote(r[0]["md"])]
|
||
ns = [x for x in notes if x["chapter_idx"] == idx]
|
||
if ns:
|
||
out += ["", "### Заметки"]
|
||
for n in ns:
|
||
meta = " · ".join(x for x in (KIND.get(n["kind"], n["kind"]), f"с. {n['page']}" if n.get("page") else "",
|
||
day(n.get("updated") or n.get("created")),
|
||
"✓ разобрано" if n.get("status") == "done" else "") if x)
|
||
out += ["", f"**{meta}**"]
|
||
if n["kind"] != "note" and n.get("quote"): # у «заметки» цитата уже в тексте (как в читалке)
|
||
out.append(quote(n["quote"]))
|
||
if (n.get("body") or "").strip():
|
||
out += ["", n["body"].strip()]
|
||
ds = [x for x in dialogs if x["chapter_idx"] == idx]
|
||
if ds:
|
||
out += ["", "### Диалоги с ИИ"]
|
||
for d in ds:
|
||
head = re.sub(r"\s+", " ", d.get("quote") or "").strip()
|
||
head = head[:70].rstrip() + ("…" if len(head) > 70 else "")
|
||
out += ["", f"#### «{head}» · {day(d.get('created'))}", quote(d.get("quote"))]
|
||
for m in d["msgs"]:
|
||
who = "**Я:**" if m["role"] == "user" else "**ИИ:**"
|
||
out += ["", who, demote(m["text"])]
|
||
return "\n".join(out).rstrip() + "\n"
|
||
|
||
|
||
def write_book(path, generated):
|
||
"""Сгенерированная часть + раздел «Мои мысли» из прежнего файла (или пустой)."""
|
||
mine = f"\n{MINE}\n\n"
|
||
if path.exists():
|
||
old = path.read_text(encoding="utf-8")
|
||
i = old.find("\n" + MINE)
|
||
if i >= 0:
|
||
mine = old[i:]
|
||
text = generated + mine
|
||
if path.exists() and path.read_text(encoding="utf-8") == text:
|
||
return False
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
tmp = path.with_name("." + path.name + ".tmp")
|
||
tmp.write_text(text, encoding="utf-8")
|
||
os.replace(tmp, path)
|
||
return True
|
||
|
||
|
||
def main():
|
||
force = "--force" in sys.argv
|
||
state = json.loads(STATE.read_text()) if STATE.exists() else {}
|
||
data = fetch("" if force else state.get("version", ""))
|
||
if not data.get("changed"):
|
||
return
|
||
chapters = {}
|
||
for ch in data["chapters"]:
|
||
chapters.setdefault(ch["book_id"], {})[ch["idx"]] = ch
|
||
files = state.get("files", {})
|
||
changed = []
|
||
for b in data["books"]:
|
||
chs = chapters.get(b["id"], {})
|
||
sel = lambda xs: [x for x in xs if x["book_id"] == b["id"]
|
||
and not FRONT_BACK.match(chs.get(x["chapter_idx"], {}).get("title", ""))]
|
||
notes, recaps, dialogs = sel(data["notes"]), sel(data["recaps"]), [] # диалоги — только через «В заметки»
|
||
if not (notes or recaps or dialogs) and str(b["id"]) not in files:
|
||
continue # уже был файл, а заметок не осталось — перепишем шапкой, «Мои мысли» сохранятся
|
||
rel = f"{FOLDER}/{safe(b['title'])}.md"
|
||
old = files.get(str(b["id"]))
|
||
if old and old != rel and (VAULT / old).exists() and not (VAULT / rel).exists():
|
||
os.replace(VAULT / old, VAULT / rel) # книгу переименовали — файл вместе с «Моими мыслями»
|
||
if write_book(VAULT / rel, render(data["site"], b, chapters.get(b["id"], {}), notes, recaps, dialogs)):
|
||
changed.append(rel)
|
||
files[str(b["id"])] = rel
|
||
STATE.parent.mkdir(parents=True, exist_ok=True)
|
||
STATE.write_text(json.dumps({"version": data["version"], "files": files,
|
||
"synced": dt.datetime.now(TZ).isoformat(timespec="seconds")}, ensure_ascii=False))
|
||
for rel in changed:
|
||
print("обновлено:", rel)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|