267 lines
8.4 KiB
Python
267 lines
8.4 KiB
Python
"""
|
|
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())
|