main.py

from __future__ import annotations

import asyncio
import json
import os
import sys

from dotenv import load_dotenv

from client import create_search_client, disconnect_client
from search import KnowledgeDoc, add_knowledge_base, list_docs, seed_search_docs, semantic_search

load_dotenv()


def require_env(name: str) -> str:
    value = (os.getenv(name) or "").strip()
    if not value:
        print(f"Missing required env var: {name}", file=sys.stderr)
        print("Copy .env.example to .env and add your DNotifier credentials.", file=sys.stderr)
        sys.exit(1)
    return value


def env_flag(name: str, default: bool) -> bool:
    raw = (os.getenv(name) or "").strip().lower()
    if not raw:
        return default
    return raw in {"1", "true", "yes"}


def print_help() -> None:
    print(
        """
Commands:
  /help                              Show this help
  /seed                              Index sample docs
  /add <title> | <content>           Add one document
  /list                              List indexed documents
  /limit <n>                         Set search limit
  /min <0-1>                         Set minSimilarity
  /source <name|off>                 Set filterbySource (or off)
  /quit                              Exit

Type any other text as a semantic search query.
"""
    )


async def main() -> None:
    app_id = require_env("DNOTIFIER_APP_ID")
    secret = require_env("DNOTIFIER_SECRET")
    user_id = (os.getenv("DNOTIFIER_USER_ID") or "semantic-search-demo").strip()

    limit = int(os.getenv("SEARCH_LIMIT") or 5)
    min_similarity = float(os.getenv("SEARCH_MIN_SIMILARITY") or 0.2)
    filterby_source = None

    notifier = create_search_client(app_id=app_id, secret=secret, user_id=user_id)
    await notifier.connect()

    if env_flag("SEED_KB", True):
        print("Seeding sample search documents…")
        await seed_search_docs(notifier, user_id)

    print_help()
    print(f"Ready. user_id={user_id}  limit={limit}  min_similarity={min_similarity}")

    try:
        while True:
            try:
                line = input("you> ").strip()
            except EOFError:
                break
            if not line:
                continue
            try:
                if line in {"/quit", "/exit"}:
                    break
                if line == "/help":
                    print_help()
                    continue
                if line == "/seed":
                    await seed_search_docs(notifier, user_id)
                    continue
                if line == "/list":
                    print(json.dumps(await list_docs(notifier, user_id), indent=2, default=str))
                    continue
                if line.startswith("/limit "):
                    limit = int(line[7:].strip())
                    print(f"limit={limit}")
                    continue
                if line.startswith("/min "):
                    min_similarity = float(line[5:].strip())
                    print(f"min_similarity={min_similarity}")
                    continue
                if line.startswith("/source "):
                    arg = line[8:].strip()
                    filterby_source = None if arg in {"", "off"} else arg
                    print(f"filterby_source={filterby_source or '(none)'}")
                    continue
                if line.startswith("/add "):
                    payload = line[5:]
                    sep = payload.find("|")
                    if sep == -1:
                        print("Usage: /add <title> | <content>")
                        continue
                    title = payload[:sep].strip()
                    content = payload[sep + 1 :].strip()
                    if not title or not content:
                        print("Usage: /add <title> | <content>")
                        continue
                    result = await add_knowledge_base(
                        notifier,
                        user_id,
                        KnowledgeDoc(title=title, content=content, type="docs"),
                    )
                    print("Document added:", json.dumps(result, indent=2, default=str))
                    continue

                hits = await semantic_search(
                    notifier,
                    user_id,
                    line,
                    limit=limit,
                    min_similarity=min_similarity,
                    filterby_source=filterby_source,
                )
                print(json.dumps(hits, indent=2, default=str))
                print()
            except Exception as err:  # noqa: BLE001
                print(f"Error: {err}")
    finally:
        await disconnect_client(notifier)


if __name__ == "__main__":
    asyncio.run(main())