""" Import ka-note inventory (assets/rooms/asset_images/asset_persons) from SQLite into 03 Bereiche/Inventar/. Source: D:\\projects\\chrka\\myNote\\ka-note\\server\\data\\ka-note.db Target: 03 Bereiche/Inventar/ (merge with existing UpNote imports) Images: 07 Anhänge/inventar/ (extracted from image_blobs BLOBs) """ from __future__ import annotations import argparse import json import re import sqlite3 import sys from pathlib import Path try: sys.stdout.reconfigure(encoding="utf-8") # type: ignore[attr-defined] except Exception: pass DB = Path(r"D:\projects\chrka\myNote\ka-note\server\data\ka-note.db") CTX_JSON = Path( r"D:\projects\chrka\myNote\ka-note\server\data\backup-remote\contexts.json" ) VAULT = Path(r"D:\projects\chrka\brain") INVENTAR = VAULT / "03 Bereiche" / "Inventar" ATTACH = VAULT / "07 Anhänge" / "inventar" # YAML frontmatter keys used in existing inventory notes (order preserved) YAML_ORDER = [ "typ", "ort", "details", "firma", "zuordnung", "modell", "seriennummer", "hersteller", "preis", "anzahl", "kaufdatum", "rechnung", "handbuch", "tests_shop", "garantie", # ka-note additions "kategorie", "zustand", "status", ] CONDITION_DE = { "new": "neu", "good": "gut", "fair": "mittel", "poor": "schlecht", } EXT_BY_MIME = { "image/jpeg": ".jpg", "image/jpg": ".jpg", "image/png": ".png", "image/webp": ".webp", "image/gif": ".gif", "image/heic": ".heic", } def slug_filename(title: str) -> str: t = re.sub(r"[\\/:*?\"<>|]", "-", title) return re.sub(r"\s+", " ", t).strip() def yaml_escape(v) -> str: if v is None or v == "": return "" s = str(v) if re.search(r'[:\[\]{}#&*!|>\'"%@`,]', s) or s.startswith("-") or "\n" in s: return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"' return s def load_person_map() -> dict[str, str]: ctxs = json.load(open(CTX_JSON, encoding="utf-8")) out = {} for c in ctxs: if c.get("type") == "person" and c.get("deletedAt") is None: meta = c.get("meta") or {} name = meta.get("fullName") or c["name"].removeprefix("Person ").strip() out[c["id"]] = name return out def extract_image(con, image_id: str, user_id: str, slug: str, suffix: str) -> str | None: """Write image blob to attachments. suffix e.g. 'cover' or '2'.""" row = con.execute( "SELECT mime_type, data FROM image_blobs " "WHERE id=? AND user_id=? AND deleted_at IS NULL", (image_id, user_id), ).fetchone() if not row: return None mime, data = row["mime_type"], row["data"] ext = EXT_BY_MIME.get(mime.lower(), ".bin") ATTACH.mkdir(parents=True, exist_ok=True) fname = f"{slug} - {suffix}{ext}" target = ATTACH / fname # avoid overwrite across distinct image_ids with same suffix n = 2 while target.exists() and target.stat().st_size != len(data): fname = f"{slug} - {suffix} ({n}){ext}" target = ATTACH / fname n += 1 if not target.exists(): target.write_bytes(data) return fname def build_note(asset, room_name: str | None, person_names: list[str], cover_fname: str | None, extra_fnames: list[str]) -> str: fm: dict[str, str] = {} if room_name: fm["ort"] = room_name if asset["brand"]: fm["hersteller"] = asset["brand"] if asset["model"]: fm["modell"] = asset["model"].strip() if asset["serial_number"]: fm["seriennummer"] = asset["serial_number"] if asset["purchase_price"] is not None: fm["preis"] = f"{asset['purchase_price']}" if asset["purchase_year"]: fm["kaufdatum"] = str(asset["purchase_year"]) if asset["category"]: fm["kategorie"] = asset["category"] if asset["condition"]: fm["zustand"] = CONDITION_DE.get(asset["condition"], asset["condition"]) if asset["status"]: fm["status"] = asset["status"] if person_names: fm["zuordnung"] = ", ".join(person_names) lines = ["---", "tags:", " - inventar", " - kanote-import"] for k in YAML_ORDER: if k in fm: lines.append(f"{k}: {yaml_escape(fm[k])}") lines.append("---") lines.append("") lines.append(f"# {asset['title']}") lines.append("") if cover_fname: lines.append(f"![[{cover_fname}]]") lines.append("") if asset["notes"]: lines.append(asset["notes"].strip()) lines.append("") if extra_fnames: lines.append("## Bilder") lines.append("") for f in extra_fnames: lines.append(f"![[{f}]]") lines.append("") return "\n".join(lines).rstrip() + "\n" def main() -> int: ap = argparse.ArgumentParser() ap.add_argument("--dry-run", action="store_true") args = ap.parse_args() INVENTAR.mkdir(parents=True, exist_ok=True) con = sqlite3.connect(str(DB)) con.row_factory = sqlite3.Row rooms = {r["id"]: r["name"] for r in con.execute("SELECT id,name FROM rooms WHERE deleted_at IS NULL")} person_map = load_person_map() # preload asset_images + persons imgs: dict[str, list[sqlite3.Row]] = {} for r in con.execute( "SELECT * FROM asset_images WHERE deleted_at IS NULL ORDER BY sort_order" ): imgs.setdefault(r["asset_id"], []).append(r) pers: dict[str, list[str]] = {} for r in con.execute( "SELECT asset_id, person_id FROM asset_persons WHERE deleted_at IS NULL" ): name = person_map.get(r["person_id"]) if name: pers.setdefault(r["asset_id"], []).append(name) assets = list(con.execute( "SELECT * FROM assets WHERE deleted_at IS NULL ORDER BY title COLLATE NOCASE" )) print(f"Found {len(assets)} assets, {len(rooms)} rooms") written = merged = 0 _written_this_run: set[Path] = set() for a in assets: room = rooms.get(a["room_id"]) if a["room_id"] else None persons = pers.get(a["id"], []) cover_fname = None extra = [] user_id = a["user_id"] slug = slug_filename(a["title"]) if not args.dry_run: if a["cover_image_id"]: cover_fname = extract_image( con, a["cover_image_id"], user_id, slug, "cover") idx = 2 for img_row in imgs.get(a["id"], []): img_id = img_row["image_id"] if img_id == a["cover_image_id"]: continue f = extract_image(con, img_id, user_id, slug, str(idx)) if f: extra.append(f) idx += 1 body = build_note(a, room, persons, cover_fname, extra) base = slug_filename(a["title"]) # avoid collision within this run (duplicate titles in ka-note) fname = base + ".md" target = INVENTAR / fname if target.exists() and target in _written_this_run: n = 2 while (INVENTAR / f"{base} ({n}).md") in _written_this_run or \ (INVENTAR / f"{base} ({n}).md").exists(): n += 1 fname = f"{base} ({n}).md" target = INVENTAR / fname if args.dry_run: info = [f"room={room or '-'}"] if persons: info.append(f"persons={persons}") if a["cover_image_id"]: ni = len(imgs.get(a["id"], [])) info.append(f"images={ni}") print(f" {'MERGE' if target.exists() else 'NEW '} " f"{a['title']} [{', '.join(info)}]") continue if target.exists(): existing = target.read_text(encoding="utf-8") block = [f"\n\n## ka-note Import\n"] if cover_fname: block.append(f"![[{cover_fname}]]\n") if a["notes"]: block.append(a["notes"].strip() + "\n") if extra: block.append("\n### Bilder\n") for f in extra: block.append(f"![[{f}]]") target.write_text(existing.rstrip() + "".join(block) + "\n", encoding="utf-8") merged += 1 else: target.write_text(body, encoding="utf-8") written += 1 _written_this_run.add(target) if args.dry_run: return 0 print(f"Written: {written}, merged: {merged}") return 0 if __name__ == "__main__": sys.exit(main())