room_cal/scripts/migrate_ics.py

"""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()