room_cal/room_cal/_utils.py

import asyncio
import json
import os
from dataclasses import dataclass, field
from datetime import date, datetime, timedelta
from importlib.resources import files
from pathlib import Path
from typing import Generic, TypeVar
from zoneinfo import ZoneInfo

_T = TypeVar('_T')


@dataclass
class FetchResult(Generic[_T]):
  """Return type for fetch_* actions: the synced data plus a changed flag."""
  data: _T
  changed: bool

  def __bool__(self) -> bool:
    return self.changed


_source_registry: dict[str, 'EventSource'] = {}


class EventSource:
  """Base class for event-producing objects. Registers the instance by domain name on init."""
  def __init__(self, name: str) -> None:
    self._name = name
    _source_registry[name] = self

  def _domain(self) -> str:
    return self._name

  def _ephemeral(self) -> bool:
    return False


def is_ephemeral_domain(name: str | None) -> bool:
  """Return True if the named domain was registered as ephemeral (e.g. 'api')."""
  if not name:
    return False
  source = _source_registry.get(name)
  return source._ephemeral() if source else False


def registered_domains() -> frozenset[str]:
  """Return the set of all registered source domain names."""
  return frozenset(_source_registry)


def safe_json_loads(s: str | None) -> dict | None:
  if not s:
    return None
  try:
    return json.loads(s)
  except (ValueError, TypeError):
    return None


def iter_dates(start: date, end: date):
  d = start
  while d <= end:
    yield d
    d += timedelta(days=1)


def event_dates(event, start_date: date, end_date: date):
  """Yield dates covered by an EventRecord, clamped to [start_date, end_date]."""
  ev_start = event.start.date()
  ev_end = event.end.date()
  last_day = (ev_end - timedelta(days=1)) if event.all_day else ev_end
  for d in iter_dates(ev_start, last_day):
    if start_date <= d <= end_date:
      yield d


async def drop_stale_events(
  events: list,
  covered: set[date],
  *,
  domain: str,
  id_: str,
  cal,
) -> None:
  """Delete events whose source domain/id match but whose date is no longer covered."""
  to_delete = []
  for ev in events:
    xd = safe_json_loads(ev.x_data)
    stored = xd.get('source', {}) if xd else {}
    if not isinstance(stored, dict):
      continue
    ev_date = ev.start.date()
    if ev_date not in covered and stored.get('domain') == domain and stored.get('id') == id_:
      to_delete.append(ev)
  if to_delete:
    await asyncio.gather(*[cal.delete(ev) for ev in to_delete])


def filter_events(
  events: list,
  prefix: str,
  *,
  all_day: bool | None = None,
  require_x_data: bool = True,
) -> list:
  return [
    e for e in events
    if e.title.startswith(prefix)
    and (all_day is None or e.all_day == all_day)
    and (not require_x_data or e.x_data)
  ]


def data_home() -> Path:
  override = os.environ.get('ROOM_CAL_DATA_HOME')
  return Path(override) if override else Path.home() / '.local' / 'room_cal'


def _deep_merge(base: dict, override: dict) -> dict:
  result = dict(base)
  for k, v in override.items():
    if k in result and isinstance(result[k], dict) and isinstance(v, dict):
      result[k] = _deep_merge(result[k], v)
    else:
      result[k] = v
  return result


# ── Config dataclasses ─────────────────────────────────────────────────────────

@dataclass
class ServiceFees:
  commission: float = 0.0
  client_fee: float = 0.0

  @classmethod
  def from_dict(cls, d: dict) -> 'ServiceFees':
    return cls(commission=d.get('commission', 0.0), client_fee=d.get('client_fee', 0.0))


@dataclass
class ServiceTaxes:
  wrong: float | None = None
  flat_tax: int = 0

  @classmethod
  def from_dict(cls, d: dict) -> 'ServiceTaxes':
    return cls(wrong=d.get('wrong'), flat_tax=d.get('flat_tax', 0))


@dataclass
class ServiceConfig:
  key: str
  name: str
  listing_id: str = ''
  hotel_id: str = ''
  hotel_account_id: str = ''
  host_account_id: str = ''
  room_id: str = ''
  rate_id: str = ''
  fees: ServiceFees = field(default_factory=ServiceFees)
  taxes: ServiceTaxes = field(default_factory=ServiceTaxes)

  @classmethod
  def from_dict(cls, key: str, d: dict) -> 'ServiceConfig':
    return cls(
      key=key,
      name=d['name'],
      listing_id=d.get('listing_id', ''),
      hotel_id=str(d.get('hotel_id', '')),
      hotel_account_id=str(d.get('hotel_account_id', '')),
      host_account_id=str(d.get('host_account_id', '')),
      room_id=str(d.get('room_id', '')),
      rate_id=str(d.get('rate_id', '')),
      fees=ServiceFees.from_dict(d.get('fees', {})),
      taxes=ServiceTaxes.from_dict(d.get('taxes', {})),
    )


@dataclass
class GlobalTaxes:
  sales: float = 0.0

  @classmethod
  def from_dict(cls, d: dict) -> 'GlobalTaxes':
    return cls(sales=d.get('sales', 0.0))


@dataclass
class DefaultPrices:
  weekday: int = 62
  weekend: int = 72

  @classmethod
  def from_dict(cls, d: dict) -> 'DefaultPrices':
    return cls(weekday=d.get('weekday', 62), weekend=d.get('weekend', 72))


@dataclass
class PropertyConfig:
  timezone: str = 'America/Los_Angeles'
  checkin_hour: int = 14
  checkout_hour: int = 12

  @classmethod
  def from_dict(cls, d: dict) -> 'PropertyConfig':
    return cls(
      timezone=d.get('timezone', 'America/Los_Angeles'),
      checkin_hour=d.get('checkin_hour', 14),
      checkout_hour=d.get('checkout_hour', 12),
    )


@dataclass
class GlobalConfig:
  taxes: GlobalTaxes = field(default_factory=GlobalTaxes)
  property: PropertyConfig = field(default_factory=PropertyConfig)
  prices: DefaultPrices = field(default_factory=DefaultPrices)
  reservation_window_days: int = 90
  busy_calendar_name: str = 'Busy'

  @classmethod
  def from_dict(cls, d: dict) -> 'GlobalConfig':
    return cls(
      taxes=GlobalTaxes.from_dict(d.get('taxes', {})),
      property=PropertyConfig.from_dict(d.get('property', {})),
      prices=DefaultPrices.from_dict(d.get('prices', {})),
      reservation_window_days=d.get('reservation_window_days', 90),
      busy_calendar_name=d.get('busy_calendar_name', 'Busy'),
    )


@dataclass
class PipelineConfig:
  type: str
  rules: list[str]

  @classmethod
  def from_dict(cls, d: dict) -> 'PipelineConfig':
    return cls(type=d['type'], rules=d.get('rules', []))


@dataclass
class FutureBlockRule:
  days_out: int = 150

  @classmethod
  def from_dict(cls, d: dict) -> 'FutureBlockRule':
    return cls(days_out=d.get('days_out', 150))


@dataclass
class ShortNoticeRule:
  max_days: int = 14
  ramp_days: int = 7

  @classmethod
  def from_dict(cls, d: dict) -> 'ShortNoticeRule':
    return cls(max_days=d.get('max_days', 14), ramp_days=d.get('ramp_days', 7))


@dataclass
class FillMonthRule:
  discount_pct: int = 10
  max_days: int = 30
  ramp_days: int = 7

  @classmethod
  def from_dict(cls, d: dict) -> 'FillMonthRule':
    return cls(
      discount_pct=d.get('discount_pct', 10),
      max_days=d.get('max_days', 30),
      ramp_days=d.get('ramp_days', 7),
    )


@dataclass
class FireRule:
  day_of: int = 314
  day_before: int = 111
  day_after: int = 222
  feed_name: str = 'Fire'

  @classmethod
  def from_dict(cls, d: dict) -> 'FireRule':
    return cls(
      day_of=d.get('day_of', 314),
      day_before=d.get('day_before', 111),
      day_after=d.get('day_after', 222),
      feed_name=d.get('feed_name', 'Fire'),
    )


@dataclass
class RulesConfig:
  future_block: FutureBlockRule = field(default_factory=FutureBlockRule)
  short_notice: ShortNoticeRule = field(default_factory=ShortNoticeRule)
  fill_month: FillMonthRule = field(default_factory=FillMonthRule)
  fire: FireRule = field(default_factory=FireRule)
  pipelines: dict[str, PipelineConfig] = field(default_factory=dict)

  @classmethod
  def from_dict(cls, d: dict) -> 'RulesConfig':
    return cls(
      future_block=FutureBlockRule.from_dict(d.get('future_block', {})),
      short_notice=ShortNoticeRule.from_dict(d.get('short_notice', {})),
      fill_month=FillMonthRule.from_dict(d.get('fill_month', {})),
      fire=FireRule.from_dict(d.get('fire', {})),
      pipelines={k: PipelineConfig.from_dict(v) for k, v in d.get('pipelines', {}).items()},
    )


@dataclass
class CalendarEntry:
  name: str
  url: str | None = None
  caldav: str | None = None
  past_days: int | None = None
  future_days: int | None = None

  @classmethod
  def from_dict(cls, d: dict) -> 'CalendarEntry':
    return cls(
      name=d['name'],
      url=d.get('url'),
      caldav=d.get('caldav'),
      past_days=d.get('past_days'),
      future_days=d.get('future_days'),
    )


@dataclass
class CaldavStoreConfig:
  url: str = ''
  username: str = ''
  password: str = ''
  calendar_name: str = 'Room'

  @classmethod
  def from_dict(cls, d: dict) -> 'CaldavStoreConfig':
    return cls(
      url=d.get('url', ''),
      username=d.get('username', ''),
      password=d.get('password', ''),
      calendar_name=d.get('calendar_name', 'Room'),
    )


@dataclass
class SqliteStoreConfig:
  db_path: str = 'store.db'

  @classmethod
  def from_dict(cls, d: dict) -> 'SqliteStoreConfig':
    return cls(db_path=d.get('db_path', 'store.db'))


@dataclass
class StoreConfig:
  active: str = 'caldav'
  caldav: CaldavStoreConfig = field(default_factory=CaldavStoreConfig)
  sqlite: SqliteStoreConfig = field(default_factory=SqliteStoreConfig)

  @classmethod
  def from_dict(cls, d: dict) -> 'StoreConfig':
    return cls(
      active=d.get('active', 'caldav'),
      caldav=CaldavStoreConfig.from_dict(d.get('caldav', {})),
      sqlite=SqliteStoreConfig.from_dict(d.get('sqlite', {})),
    )


@dataclass
class Config:
  global_: GlobalConfig = field(default_factory=GlobalConfig)
  services: dict[str, ServiceConfig] = field(default_factory=dict)
  rules: RulesConfig = field(default_factory=RulesConfig)
  calendars: list[CalendarEntry] = field(default_factory=list)
  api_url: str | None = None
  store: StoreConfig = field(default_factory=StoreConfig)

  @classmethod
  def from_dict(cls, d: dict) -> 'Config':
    store_d = d.get('store', {})
    return cls(
      global_=GlobalConfig.from_dict(d.get('global', {})),
      services={k: ServiceConfig.from_dict(k, v) for k, v in d.get('services', {}).items()},
      rules=RulesConfig.from_dict(d.get('rules', {})),
      calendars=[CalendarEntry.from_dict(e) for e in d.get('calendars', [])],
      api_url=d.get('api_url'),
      store=StoreConfig.from_dict(store_d),
    )


# ── Config loading ─────────────────────────────────────────────────────────────

def get_config() -> Config:
  """Load room_cal config.

  Always starts from the bundled default config.json, then deep-merges
  data_home()/config.json on top. Dict keys are merged recursively; lists
  and scalar values in the user config replace their defaults entirely.
  """
  default = json.loads(files('room_cal').joinpath('config.json').read_text(encoding='utf-8'))
  override_path = data_home() / 'config.json'
  if not override_path.exists():
    return Config.from_dict(default)
  override = json.loads(override_path.read_text(encoding='utf-8'))
  return Config.from_dict(_deep_merge(default, override))


def local_today() -> date:
  """Return today's date in the configured property timezone."""
  tz = ZoneInfo(get_config().global_.property.timezone)
  return datetime.now(tz).date()


def services() -> list[ServiceConfig]:
  """Return list of ServiceConfig for all configured services."""
  return list(get_config().services.values())


def service_name(key: str) -> str:
  """Return the display name for a service key, falling back to key.capitalize()."""
  svc = get_config().services.get(key)
  return svc.name if svc else key.capitalize()


def source_line(source: dict) -> str:
  """Format a source dict {name, id} into a human-readable string."""
  name = source.get('name', '')
  id_ = source.get('id', '')
  return f"{name} {id_}".strip()


def read_env(path: Path) -> dict[str, str]:
  env: dict[str, str] = {}
  try:
    for line in path.read_text().splitlines():
      line = line.strip()
      if not line or line.startswith('#') or '=' not in line:
        continue
      k, _, v = line.partition('=')
      env[k.strip()] = v.strip().strip("'\"")
  except FileNotFoundError:
    pass
  return env