import os import json import time from typing import Optional, List from backend.database import SessionLocal, Integration, Tool from backend.composio_client import get_composio DATA_DIR = os.environ.get("DATA_DIR", "/data") CACHE_FILE = f"{DATA_DIR}/composio_cache.json" CACHE_TTL = int(os.environ.get("CACHE_TTL", "86400")) def _to_dict(obj): if hasattr(obj, "model_dump"): return obj.model_dump(mode="json") if hasattr(obj, "__dict__"): return {k: v for k, v in obj.__dict__.items() if not k.startswith("_")} return {"slug": str(obj), "name": str(obj)} class LocalDB: def __init__(self, cache_file: str = CACHE_FILE): self.cache_file = cache_file os.makedirs(os.path.dirname(cache_file), exist_ok=True) def save_to_json(self, toolkits: List[dict], tools: List[dict]): data = {"toolkits": toolkits, "tools": tools, "cached_at": time.time(), "toolkit_count": len(toolkits), "tool_count": len(tools)} with open(self.cache_file, "w") as f: json.dump(data, f, indent=2) def load_from_json(self) -> Optional[dict]: if os.path.exists(self.cache_file): with open(self.cache_file) as f: return json.load(f) return None def is_json_stale(self) -> bool: data = self.load_from_json() if not data: return True return (time.time() - data.get("cached_at", 0)) > CACHE_TTL def seed_sqlite(self, toolkits: List[dict], tools: List[dict]): db = SessionLocal() try: db.query(Tool).delete() db.query(Integration).delete() db.commit() for tk in toolkits: meta = tk.get("meta", {}) or {} cats = (meta.get("categories") or []) if isinstance(meta, dict) else [] category = cats[0].get("name", "Other") if cats else "Other" auth_type = "oauth2" for s in (tk.get("auth_schemes") or []): if "oauth" in str(s).lower(): auth_type = "oauth2" break auth_type = "api_key" db.add(Integration( id=tk.get("slug", tk.get("id", "")), name=tk.get("name", tk.get("slug", "")), description=tk.get("description", ""), logo_url="", category=category, auth_type=auth_type, tool_count=int(meta.get("tools_count", 0)) if isinstance(meta, dict) else 0, trigger_count=0, is_active=True )) for t in tools: toolkit = t.get("toolkit", {}) or {} ts = toolkit.get("slug", "") if isinstance(toolkit, dict) else "" tags = t.get("tags", ["General"]) or ["General"] db.add(Tool( id=t.get("slug", t.get("id", "")), integration_id=ts, name=t.get("name", ""), description=t.get("description", ""), input_schema=t.get("input_parameters", {"type": "object", "properties": {}}), output_schema=t.get("output_parameters"), category=tags[0] if tags else "General", is_active=not t.get("is_deprecated", False) )) db.commit() self.save_to_json(toolkits, tools) except Exception: db.rollback() raise finally: db.close() async def sync_from_composio(api_key: str = None, force: bool = False) -> dict: composio = get_composio() local_db = LocalDB() if composio is None: data = local_db.load_from_json() if data: local_db.seed_sqlite(data.get("toolkits", []), data.get("tools", [])) return {"source": "local_cache", "toolkits": data.get("toolkit_count", 0), "tools": data.get("tool_count", 0)} return {"source": "none", "toolkits": 0, "tools": 0} if not force and not local_db.is_json_stale(): data = local_db.load_from_json() if data and data.get("toolkits"): local_db.seed_sqlite(data["toolkits"], data.get("tools", [])) return {"source": "local_cache", "toolkits": data["toolkit_count"], "tools": data["tool_count"]} try: raw_tk = composio.toolkits.get() new_toolkits = [_to_dict(t) for t in raw_tk if hasattr(t, "slug") or (isinstance(t, dict) and t.get("slug"))] new_tools = [] for tk in new_toolkits: slug = tk.get("slug", "") if not slug: continue try: raw = composio.tools.get_raw_composio_tools(toolkits=[slug], limit=500) new_tools.extend(_to_dict(r) for r in raw) except Exception: pass local_db.seed_sqlite(new_toolkits, new_tools) return {"source": "api", "toolkits": len(new_toolkits), "tools": len(new_tools)} except Exception as e: print(f"[Sync] SDK failed: {e}") data = local_db.load_from_json() if data: local_db.seed_sqlite(data.get("toolkits", []), data.get("tools", [])) return {"source": "local_cache_fallback", "toolkits": data.get("toolkit_count", 0), "tools": data.get("tool_count", 0)} return {"source": "fallback", "toolkits": 0, "tools": 0} def seed_integrations(): from backend.database import init_db init_db() import asyncio try: result = asyncio.run(sync_from_composio()) print(f"[Seed] Result: {result}") except Exception as e: print(f"[Seed] Error: {e}") if __name__ == "__main__": seed_integrations()