Files
dss/tools/api/npm_search.py
Digital Production Factory 276ed71f31 Initial commit: Clean DSS implementation
Migrated from design-system-swarm with fresh git history.
Old project history preserved in /home/overbits/apps/design-system-swarm

Core components:
- MCP Server (Python FastAPI with mcp 1.23.1)
- Claude Plugin (agents, commands, skills, strategies, hooks, core)
- DSS Backend (dss-mvp1 - token translation, Figma sync)
- Admin UI (Node.js/React)
- Server (Node.js/Express)
- Storybook integration (dss-mvp1/.storybook)

Self-contained configuration:
- All paths relative or use DSS_BASE_PATH=/home/overbits/dss
- PYTHONPATH configured for dss-mvp1 and dss-claude-plugin
- .env file with all configuration
- Claude plugin uses ${CLAUDE_PLUGIN_ROOT} for portability

Migration completed: $(date)
🤖 Clean migration with full functionality preserved
2025-12-09 18:45:48 -03:00

353 lines
11 KiB
Python

"""
npm Registry Search for Design Systems.
This module provides npm registry search capabilities to find
design system packages when they're not in our built-in registry.
"""
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import List, Optional, Dict, Any
import re
NPM_REGISTRY_URL = "https://registry.npmjs.org"
NPM_SEARCH_URL = "https://registry.npmjs.org/-/v1/search"
@dataclass
class NpmPackageInfo:
"""Information about an npm package."""
name: str
version: str
description: str
keywords: List[str]
homepage: Optional[str]
repository: Optional[str]
npm_url: str
downloads_weekly: int = 0
is_design_system: bool = False
confidence_score: float = 0.0
def to_dict(self) -> Dict[str, Any]:
return {
"name": self.name,
"version": self.version,
"description": self.description,
"keywords": self.keywords,
"homepage": self.homepage,
"repository": self.repository,
"npm_url": self.npm_url,
"downloads_weekly": self.downloads_weekly,
"is_design_system": self.is_design_system,
"confidence_score": self.confidence_score,
}
# Keywords that indicate a design system package
DESIGN_SYSTEM_KEYWORDS = {
# High confidence
"design-system": 3.0,
"design-tokens": 3.0,
"ui-kit": 2.5,
"component-library": 2.5,
"design tokens": 3.0,
# Medium confidence
"ui-components": 2.0,
"react-components": 1.5,
"vue-components": 1.5,
"css-framework": 2.0,
"theme": 1.5,
"theming": 1.5,
"tokens": 1.5,
"styles": 1.0,
"components": 1.0,
# Low confidence (common in design systems)
"ui": 0.5,
"react": 0.3,
"vue": 0.3,
"css": 0.3,
"scss": 0.3,
"tailwind": 1.5,
"material": 1.0,
"bootstrap": 1.0,
}
def calculate_design_system_score(package_info: Dict[str, Any]) -> float:
"""
Calculate a confidence score that a package is a design system.
Returns a score from 0.0 to 1.0.
"""
score = 0.0
# Check keywords
keywords = package_info.get("keywords", []) or []
for keyword in keywords:
keyword_lower = keyword.lower()
for ds_keyword, weight in DESIGN_SYSTEM_KEYWORDS.items():
if ds_keyword in keyword_lower:
score += weight
# Check description
description = (package_info.get("description", "") or "").lower()
if "design system" in description:
score += 3.0
if "design tokens" in description:
score += 2.5
if "component library" in description:
score += 2.0
if "ui components" in description:
score += 1.5
if "ui kit" in description:
score += 2.0
if any(word in description for word in ["react", "vue", "angular", "svelte"]):
score += 0.5
if "css" in description:
score += 0.3
# Check name patterns
name = package_info.get("name", "").lower()
if any(term in name for term in ["design", "theme", "ui", "components", "tokens"]):
score += 1.0
if name.startswith("@"): # Scoped packages often are design systems
score += 0.5
# Normalize to 0-1 range
normalized_score = min(1.0, score / 10.0)
return normalized_score
async def search_npm(
query: str,
limit: int = 10,
design_systems_only: bool = True
) -> List[NpmPackageInfo]:
"""
Search npm registry for packages matching the query.
Args:
query: Search query
limit: Maximum number of results
design_systems_only: If True, filter to likely design system packages
Returns:
List of NpmPackageInfo objects
"""
params = {
"text": query,
"size": limit * 3 if design_systems_only else limit, # Get more to filter
}
results = []
try:
async with aiohttp.ClientSession() as session:
async with session.get(NPM_SEARCH_URL, params=params) as response:
if response.status != 200:
return []
data = await response.json()
for obj in data.get("objects", []):
package = obj.get("package", {})
# Calculate design system score
ds_score = calculate_design_system_score(package)
is_design_system = ds_score >= 0.3
if design_systems_only and not is_design_system:
continue
# Extract repository URL
repo = package.get("links", {}).get("repository")
if not repo:
repo_info = package.get("repository")
if isinstance(repo_info, dict):
repo = repo_info.get("url", "")
elif isinstance(repo_info, str):
repo = repo_info
info = NpmPackageInfo(
name=package.get("name", ""),
version=package.get("version", ""),
description=package.get("description", ""),
keywords=package.get("keywords", []) or [],
homepage=package.get("links", {}).get("homepage"),
repository=repo,
npm_url=f"https://www.npmjs.com/package/{package.get('name', '')}",
downloads_weekly=obj.get("downloads", {}).get("weekly", 0) if obj.get("downloads") else 0,
is_design_system=is_design_system,
confidence_score=ds_score,
)
results.append(info)
if len(results) >= limit:
break
except Exception as e:
print(f"npm search error: {e}")
return []
# Sort by confidence score
results.sort(key=lambda x: x.confidence_score, reverse=True)
return results
async def get_package_info(package_name: str) -> Optional[NpmPackageInfo]:
"""
Get detailed information about a specific npm package.
"""
url = f"{NPM_REGISTRY_URL}/{package_name}"
try:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status != 200:
return None
data = await response.json()
# Get latest version info
latest_version = data.get("dist-tags", {}).get("latest", "")
version_info = data.get("versions", {}).get(latest_version, {})
# Extract repository URL
repo = data.get("repository")
repo_url = None
if isinstance(repo, dict):
repo_url = repo.get("url", "")
elif isinstance(repo, str):
repo_url = repo
# Clean up repo URL
if repo_url:
repo_url = re.sub(r'^git\+', '', repo_url)
repo_url = re.sub(r'\.git$', '', repo_url)
repo_url = repo_url.replace('git://', 'https://')
repo_url = repo_url.replace('ssh://git@', 'https://')
ds_score = calculate_design_system_score({
"name": data.get("name", ""),
"description": data.get("description", ""),
"keywords": data.get("keywords", []),
})
return NpmPackageInfo(
name=data.get("name", ""),
version=latest_version,
description=data.get("description", ""),
keywords=data.get("keywords", []) or [],
homepage=data.get("homepage"),
repository=repo_url,
npm_url=f"https://www.npmjs.com/package/{package_name}",
is_design_system=ds_score >= 0.3,
confidence_score=ds_score,
)
except Exception as e:
print(f"npm package info error: {e}")
return None
async def verify_package_exists(package_name: str) -> bool:
"""Check if an npm package exists."""
url = f"{NPM_REGISTRY_URL}/{package_name}"
try:
async with aiohttp.ClientSession() as session:
async with session.head(url) as response:
return response.status == 200
except Exception:
return False
async def get_package_files(package_name: str, version: str = "latest") -> List[str]:
"""
Get list of files in an npm package (for finding token files).
Uses unpkg.com to browse package contents.
"""
url = f"https://unpkg.com/{package_name}@{version}/?meta"
try:
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
if response.status != 200:
return []
data = await response.json()
def extract_files(node: Dict, prefix: str = "") -> List[str]:
files = []
if node.get("type") == "file":
files.append(f"{prefix}/{node.get('path', '')}")
elif node.get("type") == "directory":
for child in node.get("files", []):
files.extend(extract_files(child, prefix))
return files
return extract_files(data)
except Exception as e:
print(f"Error getting package files: {e}")
return []
def find_token_files(file_list: List[str]) -> Dict[str, List[str]]:
"""
Identify potential design token files from a list of package files.
"""
token_files = {
"json_tokens": [],
"css_variables": [],
"scss_variables": [],
"js_tokens": [],
"style_dictionary": [],
}
for file_path in file_list:
lower_path = file_path.lower()
# JSON tokens
if lower_path.endswith(".json"):
if any(term in lower_path for term in ["token", "theme", "color", "spacing", "typography"]):
token_files["json_tokens"].append(file_path)
# CSS variables
elif lower_path.endswith(".css"):
if any(term in lower_path for term in ["variables", "tokens", "theme", "custom-properties"]):
token_files["css_variables"].append(file_path)
# SCSS variables
elif lower_path.endswith((".scss", ".sass")):
if any(term in lower_path for term in ["variables", "tokens", "_variables", "_tokens"]):
token_files["scss_variables"].append(file_path)
# JS/TS tokens
elif lower_path.endswith((".js", ".ts", ".mjs")):
if any(term in lower_path for term in ["theme", "tokens", "colors", "spacing"]):
token_files["js_tokens"].append(file_path)
# Style Dictionary
elif "style-dictionary" in lower_path or "tokens.json" in lower_path:
token_files["style_dictionary"].append(file_path)
return token_files
# Synchronous wrapper for use in non-async contexts
def search_npm_sync(query: str, limit: int = 10, design_systems_only: bool = True) -> List[NpmPackageInfo]:
"""Synchronous wrapper for search_npm."""
return asyncio.run(search_npm(query, limit, design_systems_only))
def get_package_info_sync(package_name: str) -> Optional[NpmPackageInfo]:
"""Synchronous wrapper for get_package_info."""
return asyncio.run(get_package_info(package_name))