bruno
3b00cbc371
- unified importer framework (app/services/importers/): normalized model, registry, common pipeline (hierarchy, attachments, collections, dedup), async jobs, dry-run preview, column->type mapping - Phase 1: Obsidian, Notion, Logseq/Roam, HTML (Apple Notes/Bear/Ulysses/ OneNote), Google Keep, generic Markdown - Phase 2: typed CSV/TSV, Excel (openpyxl), generic JSON - Phase 3: Word .docx (python-docx), PDF (pypdf), HTML folders - Phase 4: Raindrop, Pocket, Readwise, Shaarli, Netscape bookmarks, .ics, OPML, Standard Notes, Gitea/GitHub issues (+labels/milestones) - Phase 5: incremental re-sync (skip/update/duplicate), partial-error resume, forge repo files, URL web clipper (SSRF guard), batch multi-file + UI queue, Notion relation resolution, exportable JSON reports - /import wizard, API /api/import/*, migration 9 (import_items, import_jobs) - fix: property values stored by property id (correct DB view rendering) - deps: openpyxl, beautifulsoup4, PyYAML, python-docx, pypdf - 43 import tests; full suite 491 green; ruff clean - bump version 5.11.5
97 lines
3.4 KiB
Python
97 lines
3.4 KiB
Python
"""FlowDeck — Standard Notes importer (v5.6.0, Phase 4).
|
|
|
|
Imports a Standard Notes backup (``.json``): each non-encrypted note becomes a
|
|
FlowDeck page. Encrypted notes are reported as warnings.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from app.services.importers.base import (
|
|
Importer,
|
|
ImportPage,
|
|
ImportResult,
|
|
decode_text,
|
|
register_importer,
|
|
)
|
|
|
|
_NOTE_TYPES = ("note", "org.standardnotes.sn", "org.standardnotes.plain-text")
|
|
|
|
|
|
def _note_body(content: Any) -> tuple[str, str, bool]:
|
|
"""Return (title, markdown, encrypted)."""
|
|
if isinstance(content, dict):
|
|
if content.get("encrypted"):
|
|
return "", "", True
|
|
text = content.get("text") or content.get("preview_plain") or ""
|
|
title = content.get("title") or ""
|
|
return title, text, False
|
|
if isinstance(content, str):
|
|
stripped = content.strip()
|
|
if stripped.startswith("{"):
|
|
try:
|
|
parsed = json.loads(stripped)
|
|
if isinstance(parsed, dict) and ("encrypted" in parsed or "000" in parsed):
|
|
return "", "", True
|
|
if isinstance(parsed, dict):
|
|
return parsed.get("title", ""), parsed.get("text", "") or parsed.get("preview_plain", ""), False
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return "", content, False
|
|
return "", "", False
|
|
|
|
|
|
@register_importer
|
|
class StandardNotesImporter(Importer):
|
|
source_id = "standard_notes"
|
|
label = "Standard Notes"
|
|
description = "Sauvegarde JSON Standard Notes → pages (notes chiffrées ignorées)."
|
|
extensions = (".json",)
|
|
order = 39
|
|
|
|
def _items(self, data: bytes) -> list[dict] | None:
|
|
try:
|
|
obj = json.loads(decode_text(data))
|
|
except Exception: # noqa: BLE001
|
|
return None
|
|
if isinstance(obj, dict) and isinstance(obj.get("items"), list):
|
|
return [x for x in obj["items"] if isinstance(x, dict)]
|
|
return None
|
|
|
|
def detect(self, filename: str, data: bytes) -> bool:
|
|
if not filename.lower().endswith(".json"):
|
|
return False
|
|
items = self._items(data)
|
|
if not items:
|
|
return False
|
|
return any("content_type" in it for it in items)
|
|
|
|
def parse(self, filename: str, data: bytes) -> ImportResult:
|
|
result = ImportResult(source=self.source_id)
|
|
items = self._items(data) or []
|
|
count = 0
|
|
for item in items:
|
|
if item.get("deleted"):
|
|
continue
|
|
ctype = str(item.get("content_type", "")).lower()
|
|
if ctype and not any(t in ctype for t in _NOTE_TYPES):
|
|
continue
|
|
title, body, encrypted = _note_body(item.get("content"))
|
|
if encrypted:
|
|
result.warn("Note chiffrée ignorée (déchiffrement non pris en charge)")
|
|
continue
|
|
if not body.strip():
|
|
continue
|
|
if not title:
|
|
title = next((ln.strip(" #") for ln in body.splitlines() if ln.strip()), "Note")
|
|
result.pages.append(ImportPage(
|
|
title=title[:200] or "Note",
|
|
markdown=body,
|
|
source_path=item.get("uuid") or filename,
|
|
external_id=item.get("uuid") or f"{filename}#{count}",
|
|
))
|
|
count += 1
|
|
result.stats["rows"] = count
|
|
return result.finalize()
|