main.py

import json
from datetime import date, datetime, timedelta, timezone
from getpass import getpass
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

BASE_URL = "https://api.limitless.ai/v1/"
DEFAULT_TIMEZONE = "America/Chicago"
MAX_AUDIO_SPAN = timedelta(hours=2)

ROOT_DIR = Path(__file__).resolve().parent / "downloaded-data"
AUDIO_DIR = ROOT_DIR / "audio"
TEXT_DIR = ROOT_DIR / "text"


def prompt_choice() -> str:
    while True:
        choice = input("What do you want to download? (audio/text): ").strip().lower()
        if choice in {"audio", "text"}:
            return choice
        print("Please type 'audio' or 'text'.")


def prompt_date(label: str, earliest: date | None = None) -> date:
    while True:
        raw = input(label).strip()
        try:
            value = datetime.strptime(raw, "%Y-%m-%d").date()
        except ValueError:
            print("Use the YYYY-MM-DD format.")
            continue
        if earliest and value < earliest:
            print(f"Date must be on or after {earliest.isoformat()}.")
            continue
        return value


def prompt_api_key() -> str:
    while True:
        key = getpass("API key: ").strip()
        if not key:
            print("API key cannot be empty.")
            continue
        length = len(key)
        if length == 39:
            return key
        confirm = input(f"Key is {length} characters (expected 39). Use it anyway? [y/N]: ").strip().lower()
        if confirm in {"y", "yes"}:
            return key


def prompt_timezone(default: str = DEFAULT_TIMEZONE) -> str:
    while True:
        raw = input(f"Timezone (IANA name, default {default}): ").strip()
        if raw == "":
            return default
        return raw


def load_timezone(name: str) -> ZoneInfo:
    try:
        return ZoneInfo(name)
    except ZoneInfoNotFoundError as _:
        print(f"Could not load time zone '{name}'. If you're on Windows or lack system tz data, install the 'tzdata' package (pip install tzdata) so daylight savings boundaries are handled correctly.")
        raise


def save_bytes(path: Path, content: bytes):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_bytes(content)


def save_json(path: Path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, indent=2))


def fetch_bytes(url: str, api_key: str) -> bytes | None:
    req = Request(url, headers={"X-API-Key": api_key})
    try:
        with urlopen(req, timeout=90) as resp:
            return resp.read()
    except HTTPError as err:
        body = ""
        try:
            body = err.read().decode("utf-8", errors="replace")
        except Exception:
            body = "<no body>"
        if err.code == 404:
            return None
        raise RuntimeError(f"Request failed ({err.code}): {err.reason}. {body}") from err
    except URLError as err:
        raise RuntimeError(f"Request failed: {err.reason}") from err


def fetch_json(url: str, api_key: str) -> dict:
    req = Request(url, headers={"X-API-Key": api_key})
    try:
        with urlopen(req, timeout=60) as resp:
            return json.load(resp)
    except HTTPError as err:
        body = ""
        try:
            body = err.read().decode("utf-8", errors="replace")
        except Exception:
            body = "<no body>"
        raise RuntimeError(f"Request failed ({err.code}): {err.reason}. {body}") from err
    except URLError as err:
        raise RuntimeError(f"Request failed: {err.reason}") from err


def iter_days(start_day: date, end_day: date):
    current = start_day
    while current <= end_day:
        yield current
        current += timedelta(days=1)


def iter_audio_chunks(day: date, tz: ZoneInfo):
    start = datetime(day.year, day.month, day.day, 0, 0, 0, tzinfo=tz)
    end = datetime(day.year, day.month, day.day, 23, 59, 59, tzinfo=tz)
    cursor = start
    while cursor <= end:
        chunk_end = min(cursor + MAX_AUDIO_SPAN, end)
        yield cursor, chunk_end
        cursor = chunk_end + timedelta(seconds=1)


def download_audio(start_day: date, end_day: date, api_key: str, tz_name: str):
    try:
        tz = load_timezone(tz_name)
    except ZoneInfoNotFoundError:
        return
    AUDIO_DIR.mkdir(parents=True, exist_ok=True)

    for day in iter_days(start_day, end_day):
        print(f"\nDay {day.isoformat()}:")
        for chunk_start, chunk_end in iter_audio_chunks(day, tz):
            start_ms = int(chunk_start.astimezone(timezone.utc).timestamp() * 1000)
            end_ms = int(chunk_end.astimezone(timezone.utc).timestamp() * 1000)

            url = f"{BASE_URL}download-audio?{urlencode({'startMs': start_ms, 'endMs': end_ms})}"
            label = f"{chunk_start.strftime('%H%M')}-{chunk_end.strftime('%H%M')}"
            name = f"{day.isoformat()}_{chunk_start.tzname() or 'LOCAL'}_{label}.ogg"
            target = AUDIO_DIR / name

            try:
                audio_bytes = fetch_bytes(url, api_key)
            except RuntimeError as err:
                print(f"  {label}: failed ({err})")
                continue

            if audio_bytes is None:
                print(f"  {label}: no audio")
                continue

            save_bytes(target, audio_bytes)
            print(f"  {label}: saved to {target}")


def fetch_lifelogs_for_day(day: date, api_key: str) -> list:
    start = f"{day.isoformat()} 00:00:00"
    end = f"{day.isoformat()} 23:59:59"
    all_entries: list = []
    cursor: str | None = None

    while True:
        params = {
            "start": start,
            "end": end,
            "limit": "10",
            "direction": "asc",
            "includeContents": "true",
            "includeHeadings": "true",
            "includeMarkdown": "true",
        }
        if cursor:
            params["cursor"] = cursor

        url = f"{BASE_URL}lifelogs?{urlencode(params)}"

        data = fetch_json(url, api_key)
        entries = data.get("data", {}).get("lifelogs", []) or []
        all_entries.extend(entries)

        cursor = data.get("meta", {}).get("lifelogs", {}).get("nextCursor")
        if not cursor:
            break

    return all_entries


def download_text(start_day: date, end_day: date, api_key: str):
    TEXT_DIR.mkdir(parents=True, exist_ok=True)

    for day in iter_days(start_day, end_day):
        print(f"\nDay {day.isoformat()}:")
        try:
            entries = fetch_lifelogs_for_day(day, api_key)
        except RuntimeError as err:
            print(f"  failed to fetch text: {err}")
            continue

        target = TEXT_DIR / f"lifelogs_{day.isoformat()}.json"
        save_json(target, entries)
        print(f"  saved {len(entries)} entries to {target}")


def main():
    print("Limitless Data Downloader")
    choice = prompt_choice()
    start_day = prompt_date("Start date (YYYY-MM-DD): ")
    end_day   = prompt_date("End date   (YYYY-MM-DD): ", earliest=start_day)
    api_key = prompt_api_key()
    tz_name = prompt_timezone()

    if choice == "audio":
        download_audio(start_day, end_day, api_key, tz_name)
    else:
        download_text(start_day, end_day, api_key)


if __name__ == "__main__":
    main()