Add ICS-to-SQLite migration script

scripts/migrate_ics.py

@@ -0,0 +1,198 @@
+"""Migrate events from an ICS file into a room_cal SQLite store."""
+
+import argparse
+import sqlite3
+import sys
+from datetime import date, datetime
+from pathlib import Path
+
+
+_X_DATA_PROP = 'X-ROOMCAL-DATA'
+
+_CREATE_TABLE = '''
+CREATE TABLE IF NOT EXISTS events (
+  uid           TEXT NOT NULL,
+  calendar_name TEXT NOT NULL,
+  start         TEXT NOT NULL,
+  end           TEXT NOT NULL,
+  all_day       INTEGER NOT NULL,
+  title         TEXT NOT NULL,
+  description   TEXT NOT NULL DEFAULT '',
+  x_data        TEXT,
+  PRIMARY KEY (uid, calendar_name)
+)
+'''
+
+_INSERT = '''
+INSERT INTO events (uid, calendar_name, start, end, all_day, title, description, x_data)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+'''
+
+_INSERT_OR_REPLACE = '''
+INSERT OR REPLACE INTO events (uid, calendar_name, start, end, all_day, title, description, x_data)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+'''
+
+
+def _naive(dt) -> datetime:
+  if isinstance(dt, datetime):
+    return dt.replace(tzinfo=None)
+  return datetime(dt.year, dt.month, dt.day)
+
+
+def _parse_vevent(vevent) -> dict | None:
+  dtstart_prop = vevent.get('DTSTART')
+  if dtstart_prop is None:
+    return None
+
+  dtstart = dtstart_prop.dt
+  dtend_prop = vevent.get('DTEND') or vevent.get('DUE')
+  if dtend_prop is not None:
+    dtend = dtend_prop.dt
+  else:
+    dtend = dtstart
+
+  all_day = isinstance(dtstart, date) and not isinstance(dtstart, datetime)
+
+  uid = str(vevent.get('UID', ''))
+  if not uid:
+    return None
+
+  x_data_prop = vevent.get(_X_DATA_PROP)
+  x_data = str(x_data_prop) if x_data_prop is not None else None
+
+  return {
+    'uid': uid,
+    'start': _naive(dtstart),
+    'end': _naive(dtend),
+    'all_day': all_day,
+    'title': str(vevent.get('SUMMARY', '')),
+    'description': str(vevent.get('DESCRIPTION', '')),
+    'x_data': x_data,
+  }
+
+
+_CALENDAR_NAME = 'Room'
+
+
+def migrate(ics_path: Path, db_path: Path, overwrite: bool, dry_run: bool) -> None:
+  calendar_name = _CALENDAR_NAME
+  try:
+    from icalendar import Calendar as _iCal
+  except ImportError:
+    print('error: icalendar is not installed', file=sys.stderr)
+    sys.exit(1)
+
+  raw = ics_path.read_bytes()
+  cal = _iCal.from_ical(raw)
+
+  events = []
+  skipped = 0
+  for component in cal.walk():
+    if component.name != 'VEVENT':
+      continue
+    parsed = _parse_vevent(component)
+    if parsed is None:
+      skipped += 1
+      continue
+    events.append(parsed)
+
+  if skipped:
+    print(f'warning: skipped {skipped} VEVENT(s) with missing UID or DTSTART', file=sys.stderr)
+
+  if not events:
+    print('no events to import')
+    return
+
+  sql = _INSERT_OR_REPLACE if overwrite else _INSERT
+
+  if dry_run:
+    print(f'dry-run: would import {len(events)} event(s) into {db_path}')
+    for e in events:
+      flag = ' [has x_data]' if e['x_data'] else ''
+      print(f"  {e['start'].date()} {'(all-day)' if e['all_day'] else e['start'].time()!s:8}  {e['title']!r}{flag}")
+    return
+
+  db_path.parent.mkdir(parents=True, exist_ok=True)
+  conn = sqlite3.connect(db_path)
+  try:
+    conn.execute(_CREATE_TABLE)
+    inserted = 0
+    replaced = 0
+    conflicts = 0
+    for e in events:
+      row = (
+        e['uid'], calendar_name,
+        e['start'].isoformat(), e['end'].isoformat(),
+        int(e['all_day']), e['title'], e['description'], e['x_data'],
+      )
+      if overwrite:
+        conn.execute(sql, row)
+        replaced += 1
+      else:
+        try:
+          conn.execute(sql, row)
+          inserted += 1
+        except sqlite3.IntegrityError:
+          conflicts += 1
+    conn.commit()
+  finally:
+    conn.close()
+
+  if overwrite:
+    print(f'imported {replaced} event(s) into {db_path}')
+  else:
+    msg = f'imported {inserted} event(s) into {db_path}'
+    if conflicts:
+      msg += f'; {conflicts} skipped (uid already exists — use --overwrite to replace)'
+    print(msg)
+
+
+def main() -> None:
+  parser = argparse.ArgumentParser(
+    description='Import events from an ICS file into a room_cal SQLite store.',
+  )
+  parser.add_argument('ics_file', type=Path, help='Path to the .ics file to import')
+  parser.add_argument(
+    '--db',
+    type=Path,
+    default=None,
+    metavar='PATH',
+    help='SQLite database path (default: from config, or store.db next to config.json)',
+  )
+  parser.add_argument(
+    '--overwrite',
+    action='store_true',
+    help='Replace existing events with the same UID instead of skipping',
+  )
+  parser.add_argument(
+    '--dry-run',
+    action='store_true',
+    help='Parse and list events without writing anything',
+  )
+  args = parser.parse_args()
+
+  if not args.ics_file.exists():
+    print(f'error: {args.ics_file} not found', file=sys.stderr)
+    sys.exit(1)
+
+  db_path = args.db
+  if db_path is None:
+    try:
+      from room_cal._utils import get_config, data_home
+      cfg = get_config()
+      raw = cfg.store.sqlite.db_path
+      if raw:
+        p = Path(raw)
+        db_path = p if p.is_absolute() else data_home() / p
+      else:
+        db_path = data_home() / 'store.db'
+    except Exception:
+      db_path = Path('store.db')
+      print(f'warning: could not read config; defaulting db path to {db_path}', file=sys.stderr)
+
+  migrate(args.ics_file, db_path, args.overwrite, args.dry_run)
+
+
+if __name__ == '__main__':
+  main()