Compare commits

...

2 Commits

Author SHA1 Message Date
0496262cd5 feat(skills): author root SKILL.md for reference-curator suite + 7 sub-skills
Finish the migration for the dirs the bulk pass couldn't auto-handle:

- 90-reference-curator/SKILL.md: hand-authored suite orchestrator (pipeline overview,
  7-stage table, /reference-curator run modes, install) — the single loadable entry.
- 90-reference-curator/0{1..7}-*/SKILL.md: generated from each sub-skill's desktop/SKILL.md.
- scripts/migrate_skill_root.py: generalized discovery to find nested suite sub-skills
  (rglob desktop/code SKILL.md), so the migrator now handles suites too.

81-mac-optimizer, 91-multi-agent-guide, 94-dintel-bootstrap need NO root SKILL.md: they
are Claude Code plugins whose skill correctly lives at skills/<name>/SKILL.md (validated).
Adding a root SKILL.md there would violate plugin structure.

All SKILL.md repo-wide validate: flat-root=65, suite-sub=7, plugin-skills=3, 0 failures.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-28 01:14:46 +09:00
60734dbde7 Recalibrate estimate for SMB acceptability
Real-world feedback: list-rate calc overshot SMB-acceptable levels.
- scaling: driver properties_total -> subbrands_total (chains share templates),
  cap x6.5 -> x2.0, applied to On-page only (Technical now fixed site-wide)
- add 'smb' entry tier (lean hours @ 0.55 billing); 3-tier auto-select
  (smb/basic/treatment) by portfolio size; per-tier billing; --baseline smb enabled
- docs (findings_to_service.md, SKILL.md) synced to 3-tier model

Effect: SHR 25-property chain 71.5M -> 29.5M; SMB single hotel ~3.0M;
basic/treatment still reproduce real 10.5M/25.0M at 1 property.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-28 01:13:38 +09:00
15 changed files with 1844 additions and 65 deletions

View File

@@ -0,0 +1,190 @@
---
name: 01-reference-discovery
description: |
Search and discover authoritative reference sources with credibility validation.
Triggers: find sources, search documentation, discover references, source validation.
---
# Reference Discovery
Searches for authoritative sources, validates credibility, and produces curated URL lists for crawling.
## Source Priority Hierarchy
| Tier | Source Type | Examples |
|------|-------------|----------|
| **Tier 1** | Official documentation | docs.anthropic.com, docs.claude.com, platform.openai.com/docs |
| **Tier 1** | Engineering blogs (official) | anthropic.com/news, openai.com/blog |
| **Tier 1** | Official GitHub repos | github.com/anthropics/*, github.com/openai/* |
| **Tier 2** | Research papers | arxiv.org, papers with citations |
| **Tier 2** | Verified community guides | Cookbook examples, official tutorials |
| **Tier 3** | Community content | Blog posts, tutorials, Stack Overflow |
## Discovery Workflow
### Step 1: Define Search Scope
```python
search_config = {
"topic": "prompt engineering",
"vendors": ["anthropic", "openai", "google"],
"source_types": ["official_docs", "engineering_blog", "github_repo"],
"freshness": "past_year", # past_week, past_month, past_year, any
"max_results_per_query": 20
}
```
### Step 2: Generate Search Queries
For a given topic, generate targeted queries:
```python
def generate_queries(topic, vendors):
queries = []
# Official documentation queries
for vendor in vendors:
queries.append(f"site:docs.{vendor}.com {topic}")
queries.append(f"site:{vendor}.com/docs {topic}")
# Engineering blog queries
for vendor in vendors:
queries.append(f"site:{vendor}.com/blog {topic}")
queries.append(f"site:{vendor}.com/news {topic}")
# GitHub queries
for vendor in vendors:
queries.append(f"site:github.com/{vendor} {topic}")
# Research queries
queries.append(f"site:arxiv.org {topic}")
return queries
```
### Step 3: Execute Search
Use web search tool for each query:
```python
def execute_discovery(queries):
results = []
for query in queries:
search_results = web_search(query)
for result in search_results:
results.append({
"url": result.url,
"title": result.title,
"snippet": result.snippet,
"query_used": query
})
return deduplicate_by_url(results)
```
### Step 4: Validate and Score Sources
```python
def score_source(url, title):
score = 0.0
# Domain credibility
if any(d in url for d in ['docs.anthropic.com', 'docs.claude.com', 'docs.openai.com']):
score += 0.40 # Tier 1 official docs
elif any(d in url for d in ['anthropic.com', 'openai.com', 'google.dev']):
score += 0.30 # Tier 1 official blog/news
elif 'github.com' in url and any(v in url for v in ['anthropics', 'openai', 'google']):
score += 0.30 # Tier 1 official repos
elif 'arxiv.org' in url:
score += 0.20 # Tier 2 research
else:
score += 0.10 # Tier 3 community
# Freshness signals (from title/snippet)
if any(year in title for year in ['2025', '2024']):
score += 0.20
elif any(year in title for year in ['2023']):
score += 0.10
# Relevance signals
if any(kw in title.lower() for kw in ['guide', 'documentation', 'tutorial', 'best practices']):
score += 0.15
return min(score, 1.0)
def assign_credibility_tier(score):
if score >= 0.60:
return 'tier1_official'
elif score >= 0.40:
return 'tier2_verified'
else:
return 'tier3_community'
```
### Step 5: Output URL Manifest
```python
def create_manifest(scored_results, topic):
manifest = {
"discovery_date": datetime.now().isoformat(),
"topic": topic,
"total_urls": len(scored_results),
"urls": []
}
for result in sorted(scored_results, key=lambda x: x['score'], reverse=True):
manifest["urls"].append({
"url": result["url"],
"title": result["title"],
"credibility_tier": result["tier"],
"credibility_score": result["score"],
"source_type": infer_source_type(result["url"]),
"vendor": infer_vendor(result["url"])
})
return manifest
```
## Output Format
Discovery produces a JSON manifest for the crawler:
```json
{
"discovery_date": "2025-01-28T10:30:00",
"topic": "prompt engineering",
"total_urls": 15,
"urls": [
{
"url": "https://docs.anthropic.com/en/docs/prompt-engineering",
"title": "Prompt Engineering Guide",
"credibility_tier": "tier1_official",
"credibility_score": 0.85,
"source_type": "official_docs",
"vendor": "anthropic"
}
]
}
```
## Known Authoritative Sources
Pre-validated sources for common topics:
| Vendor | Documentation | Blog/News | GitHub |
|--------|--------------|-----------|--------|
| Anthropic | docs.anthropic.com, docs.claude.com | anthropic.com/news | github.com/anthropics |
| OpenAI | platform.openai.com/docs | openai.com/blog | github.com/openai |
| Google | ai.google.dev/docs | blog.google/technology/ai | github.com/google |
## Integration
**Output:** URL manifest JSON → `web-crawler-orchestrator`
**Database:** Register new sources in `sources` table via `content-repository`
## Deduplication
Before outputting, deduplicate URLs:
- Normalize URLs (remove trailing slashes, query params)
- Check against existing `documents` table via `content-repository`
- Merge duplicate entries, keeping highest credibility score

View File

@@ -0,0 +1,236 @@
---
name: 02-web-crawler-orchestrator
description: |
Multi-backend web crawler with Firecrawl MCP, rate limiting, and format handling.
Triggers: crawl URLs, fetch pages, scrape content, web crawler.
---
# Web Crawler Orchestrator
Manages crawling operations using Firecrawl MCP with rate limiting and format handling.
## Prerequisites
- Firecrawl MCP server connected
- Config file at `~/.config/reference-curator/crawl_config.yaml`
- Storage directory exists: `~/reference-library/raw/`
## Crawl Configuration
```yaml
# ~/.config/reference-curator/crawl_config.yaml
firecrawl:
rate_limit:
requests_per_minute: 20
concurrent_requests: 3
default_options:
timeout: 30000
only_main_content: true
include_html: false
processing:
max_content_size_mb: 50
raw_content_dir: ~/reference-library/raw/
```
## Crawl Workflow
### Step 1: Load URL Manifest
Receive manifest from `reference-discovery`:
```python
def load_manifest(manifest_path):
with open(manifest_path) as f:
manifest = json.load(f)
return manifest["urls"]
```
### Step 2: Determine Crawl Strategy
```python
def select_strategy(url):
"""Select optimal crawl strategy based on URL characteristics."""
if url.endswith('.pdf'):
return 'pdf_extract'
elif 'github.com' in url and '/blob/' in url:
return 'raw_content' # Get raw file content
elif 'github.com' in url:
return 'scrape' # Repository pages
elif any(d in url for d in ['docs.', 'documentation']):
return 'scrape' # Documentation sites
else:
return 'scrape' # Default
```
### Step 3: Execute Firecrawl
Use Firecrawl MCP for crawling:
```python
# Single page scrape
firecrawl_scrape(
url="https://docs.anthropic.com/en/docs/prompt-engineering",
formats=["markdown"], # markdown | html | screenshot
only_main_content=True,
timeout=30000
)
# Multi-page crawl (documentation sites)
firecrawl_crawl(
url="https://docs.anthropic.com/en/docs/",
max_depth=2,
limit=50,
formats=["markdown"],
only_main_content=True
)
```
### Step 4: Rate Limiting
```python
import time
from collections import deque
class RateLimiter:
def __init__(self, requests_per_minute=20):
self.rpm = requests_per_minute
self.request_times = deque()
def wait_if_needed(self):
now = time.time()
# Remove requests older than 1 minute
while self.request_times and now - self.request_times[0] > 60:
self.request_times.popleft()
if len(self.request_times) >= self.rpm:
wait_time = 60 - (now - self.request_times[0])
if wait_time > 0:
time.sleep(wait_time)
self.request_times.append(time.time())
```
### Step 5: Save Raw Content
```python
import hashlib
from pathlib import Path
def save_content(url, content, content_type='markdown'):
"""Save crawled content to raw storage."""
# Generate filename from URL hash
url_hash = hashlib.sha256(url.encode()).hexdigest()[:16]
# Determine extension
ext_map = {'markdown': '.md', 'html': '.html', 'pdf': '.pdf'}
ext = ext_map.get(content_type, '.txt')
# Create dated subdirectory
date_dir = datetime.now().strftime('%Y/%m')
output_dir = Path.home() / 'reference-library/raw' / date_dir
output_dir.mkdir(parents=True, exist_ok=True)
# Save file
filepath = output_dir / f"{url_hash}{ext}"
if content_type == 'pdf':
filepath.write_bytes(content)
else:
filepath.write_text(content, encoding='utf-8')
return str(filepath)
```
### Step 6: Generate Crawl Manifest
```python
def create_crawl_manifest(results):
manifest = {
"crawl_date": datetime.now().isoformat(),
"total_crawled": len([r for r in results if r["status"] == "success"]),
"total_failed": len([r for r in results if r["status"] == "failed"]),
"documents": []
}
for result in results:
manifest["documents"].append({
"url": result["url"],
"status": result["status"],
"raw_content_path": result.get("filepath"),
"content_size": result.get("size"),
"crawl_method": "firecrawl",
"error": result.get("error")
})
return manifest
```
## Error Handling
| Error | Action |
|-------|--------|
| Timeout | Retry once with 2x timeout |
| Rate limit (429) | Exponential backoff, max 3 retries |
| Not found (404) | Log and skip |
| Access denied (403) | Log, mark as `failed` |
| Connection error | Retry with backoff |
```python
def crawl_with_retry(url, max_retries=3):
for attempt in range(max_retries):
try:
result = firecrawl_scrape(url)
return {"status": "success", "content": result}
except RateLimitError:
wait = 2 ** attempt * 10 # 10, 20, 40 seconds
time.sleep(wait)
except TimeoutError:
if attempt == 0:
# Retry with doubled timeout
result = firecrawl_scrape(url, timeout=60000)
return {"status": "success", "content": result}
except NotFoundError:
return {"status": "failed", "error": "404 Not Found"}
except Exception as e:
if attempt == max_retries - 1:
return {"status": "failed", "error": str(e)}
return {"status": "failed", "error": "Max retries exceeded"}
```
## Firecrawl MCP Reference
**scrape** - Single page:
```
firecrawl_scrape(url, formats, only_main_content, timeout)
```
**crawl** - Multi-page:
```
firecrawl_crawl(url, max_depth, limit, formats, only_main_content)
```
**map** - Discover URLs:
```
firecrawl_map(url, limit) # Returns list of URLs on site
```
## Integration
| From | Input | To |
|------|-------|-----|
| reference-discovery | URL manifest | web-crawler-orchestrator |
| web-crawler-orchestrator | Crawl manifest + raw files | content-repository |
| quality-reviewer (deep_research) | Additional queries | reference-discovery → here |
## Output Structure
```
~/reference-library/raw/
└── 2025/01/
├── a1b2c3d4e5f6g7h8.md # Markdown content
├── b2c3d4e5f6g7h8i9.md
└── c3d4e5f6g7h8i9j0.pdf # PDF documents
```

View File

@@ -0,0 +1,164 @@
---
name: 03-content-repository
description: |
MySQL storage manager for reference library with versioning and deduplication.
Triggers: store content, manage repository, document database, content storage.
---
# Content Repository
Manages MySQL storage for the reference library system. Handles document storage, version control, deduplication, and retrieval.
## Prerequisites
- MySQL 8.0+ with utf8mb4 charset
- Config file at `~/.config/reference-curator/db_config.yaml`
- Database `reference_library` initialized with schema
## Quick Reference
### Connection Setup
```python
import yaml
import os
from pathlib import Path
def get_db_config():
config_path = Path.home() / ".config/reference-curator/db_config.yaml"
with open(config_path) as f:
config = yaml.safe_load(f)
# Resolve environment variables
mysql = config['mysql']
return {
'host': mysql['host'],
'port': mysql['port'],
'database': mysql['database'],
'user': os.environ.get('MYSQL_USER', mysql.get('user', '')),
'password': os.environ.get('MYSQL_PASSWORD', mysql.get('password', '')),
'charset': mysql['charset']
}
```
### Core Operations
**Store New Document:**
```python
def store_document(cursor, source_id, title, url, doc_type, raw_content_path):
sql = """
INSERT INTO documents (source_id, title, url, doc_type, crawl_date, crawl_status, raw_content_path)
VALUES (%s, %s, %s, %s, NOW(), 'completed', %s)
ON DUPLICATE KEY UPDATE
version = version + 1,
previous_version_id = doc_id,
crawl_date = NOW(),
raw_content_path = VALUES(raw_content_path)
"""
cursor.execute(sql, (source_id, title, url, doc_type, raw_content_path))
return cursor.lastrowid
```
**Check Duplicate:**
```python
def is_duplicate(cursor, url):
cursor.execute("SELECT doc_id FROM documents WHERE url_hash = SHA2(%s, 256)", (url,))
return cursor.fetchone() is not None
```
**Get Document by Topic:**
```python
def get_docs_by_topic(cursor, topic_slug, min_quality=0.80):
sql = """
SELECT d.doc_id, d.title, d.url, dc.structured_content, dc.quality_score
FROM documents d
JOIN document_topics dt ON d.doc_id = dt.doc_id
JOIN topics t ON dt.topic_id = t.topic_id
LEFT JOIN distilled_content dc ON d.doc_id = dc.doc_id
WHERE t.topic_slug = %s
AND (dc.review_status = 'approved' OR dc.review_status IS NULL)
ORDER BY dt.relevance_score DESC
"""
cursor.execute(sql, (topic_slug,))
return cursor.fetchall()
```
## Table Quick Reference
| Table | Purpose | Key Fields |
|-------|---------|------------|
| `sources` | Authorized content sources | source_type, credibility_tier, vendor |
| `documents` | Crawled document metadata | url_hash (dedup), version, crawl_status |
| `distilled_content` | Processed summaries | review_status, compression_ratio |
| `review_logs` | QA decisions | quality_score, decision, refactor_instructions |
| `topics` | Taxonomy | topic_slug, parent_topic_id |
| `document_topics` | Many-to-many linking | relevance_score |
| `export_jobs` | Export tracking | export_type, output_format, status |
## Status Values
**crawl_status:** `pending``completed` | `failed` | `stale`
**review_status:** `pending``in_review``approved` | `needs_refactor` | `rejected`
**decision (review):** `approve` | `refactor` | `deep_research` | `reject`
## Common Queries
### Find Stale Documents (needs re-crawl)
```sql
SELECT d.doc_id, d.title, d.url, d.crawl_date
FROM documents d
JOIN crawl_schedule cs ON d.source_id = cs.source_id
WHERE d.crawl_date < DATE_SUB(NOW(), INTERVAL
CASE cs.frequency
WHEN 'daily' THEN 1
WHEN 'weekly' THEN 7
WHEN 'biweekly' THEN 14
WHEN 'monthly' THEN 30
END DAY)
AND cs.is_enabled = TRUE;
```
### Get Pending Reviews
```sql
SELECT dc.distill_id, d.title, d.url, dc.token_count_distilled
FROM distilled_content dc
JOIN documents d ON dc.doc_id = d.doc_id
WHERE dc.review_status = 'pending'
ORDER BY dc.distill_date ASC;
```
### Export-Ready Content
```sql
SELECT d.title, d.url, dc.structured_content, t.topic_slug
FROM documents d
JOIN distilled_content dc ON d.doc_id = dc.doc_id
JOIN document_topics dt ON d.doc_id = dt.doc_id
JOIN topics t ON dt.topic_id = t.topic_id
JOIN review_logs rl ON dc.distill_id = rl.distill_id
WHERE rl.decision = 'approve'
AND rl.quality_score >= 0.85
ORDER BY t.topic_slug, dt.relevance_score DESC;
```
## Workflow Integration
1. **From crawler-orchestrator:** Receive URL + raw content path → `store_document()`
2. **To content-distiller:** Query pending documents → send for processing
3. **From quality-reviewer:** Update `review_status` based on decision
4. **To markdown-exporter:** Query approved content by topic
## Error Handling
- **Duplicate URL:** Silent update (version increment) via `ON DUPLICATE KEY UPDATE`
- **Missing source_id:** Validate against `sources` table before insert
- **Connection failure:** Implement retry with exponential backoff
## Full Schema Reference
See `references/schema.sql` for complete table definitions including indexes and constraints.
## Config File Template
See `references/db_config_template.yaml` for connection configuration template.

View File

@@ -0,0 +1,240 @@
---
name: 04-content-distiller
description: |
Raw content summarizer extracting key concepts, code snippets, and structured output.
Triggers: distill content, summarize document, extract key concepts, compress content.
---
# Content Distiller
Transforms raw crawled content into structured, high-quality reference materials.
## Distillation Goals
1. **Compress** - Reduce token count while preserving essential information
2. **Structure** - Organize content for easy retrieval and reference
3. **Extract** - Pull out code snippets, key concepts, and actionable patterns
4. **Annotate** - Add metadata for searchability and categorization
## Distillation Workflow
### Step 1: Load Raw Content
```python
def load_for_distillation(cursor):
"""Get documents ready for distillation."""
sql = """
SELECT d.doc_id, d.title, d.url, d.raw_content_path,
d.doc_type, s.source_type, s.credibility_tier
FROM documents d
JOIN sources s ON d.source_id = s.source_id
LEFT JOIN distilled_content dc ON d.doc_id = dc.doc_id
WHERE d.crawl_status = 'completed'
AND dc.distill_id IS NULL
ORDER BY s.credibility_tier ASC
"""
cursor.execute(sql)
return cursor.fetchall()
```
### Step 2: Analyze Content Structure
Identify content type and select appropriate distillation strategy:
```python
def analyze_structure(content, doc_type):
"""Analyze document structure for distillation."""
analysis = {
"has_code_blocks": bool(re.findall(r'```[\s\S]*?```', content)),
"has_headers": bool(re.findall(r'^#+\s', content, re.MULTILINE)),
"has_lists": bool(re.findall(r'^\s*[-*]\s', content, re.MULTILINE)),
"has_tables": bool(re.findall(r'\|.*\|', content)),
"estimated_tokens": len(content.split()) * 1.3, # Rough estimate
"section_count": len(re.findall(r'^#+\s', content, re.MULTILINE))
}
return analysis
```
### Step 3: Extract Key Components
**Extract Code Snippets:**
```python
def extract_code_snippets(content):
"""Extract all code blocks with language tags."""
pattern = r'```(\w*)\n([\s\S]*?)```'
snippets = []
for match in re.finditer(pattern, content):
snippets.append({
"language": match.group(1) or "text",
"code": match.group(2).strip(),
"context": get_surrounding_text(content, match.start(), 200)
})
return snippets
```
**Extract Key Concepts:**
```python
def extract_key_concepts(content, title):
"""Use Claude to extract key concepts and definitions."""
prompt = f"""
Analyze this document and extract key concepts:
Title: {title}
Content: {content[:8000]} # Limit for context
Return JSON with:
- concepts: [{{"term": "...", "definition": "...", "importance": "high|medium|low"}}]
- techniques: [{{"name": "...", "description": "...", "use_case": "..."}}]
- best_practices: ["..."]
"""
# Use Claude API to process
return claude_extract(prompt)
```
### Step 4: Create Structured Summary
**Summary Template:**
```markdown
# {title}
**Source:** {url}
**Type:** {source_type} | **Tier:** {credibility_tier}
**Distilled:** {date}
## Executive Summary
{2-3 sentence overview}
## Key Concepts
{bulleted list of core concepts with brief definitions}
## Techniques & Patterns
{extracted techniques with use cases}
## Code Examples
{relevant code snippets with context}
## Best Practices
{actionable recommendations}
## Related Topics
{links to related content in library}
```
### Step 5: Optimize for Tokens
```python
def optimize_content(structured_content, target_ratio=0.30):
"""
Compress content to target ratio while preserving quality.
Target: 30% of original token count.
"""
original_tokens = count_tokens(structured_content)
target_tokens = int(original_tokens * target_ratio)
# Prioritized compression strategies
strategies = [
remove_redundant_explanations,
condense_examples,
merge_similar_sections,
trim_verbose_descriptions
]
optimized = structured_content
for strategy in strategies:
if count_tokens(optimized) > target_tokens:
optimized = strategy(optimized)
return optimized
```
### Step 6: Store Distilled Content
```python
def store_distilled(cursor, doc_id, summary, key_concepts,
code_snippets, structured_content,
original_tokens, distilled_tokens):
sql = """
INSERT INTO distilled_content
(doc_id, summary, key_concepts, code_snippets, structured_content,
token_count_original, token_count_distilled, distill_model, review_status)
VALUES (%s, %s, %s, %s, %s, %s, %s, 'claude-opus-4-5', 'pending')
"""
cursor.execute(sql, (
doc_id, summary,
json.dumps(key_concepts),
json.dumps(code_snippets),
structured_content,
original_tokens,
distilled_tokens
))
return cursor.lastrowid
```
## Distillation Prompts
**For Prompt Engineering Content:**
```
Focus on:
1. Specific techniques with before/after examples
2. Why techniques work (not just what)
3. Common pitfalls and how to avoid them
4. Actionable patterns that can be directly applied
```
**For API Documentation:**
```
Focus on:
1. Endpoint specifications and parameters
2. Request/response examples
3. Error codes and handling
4. Rate limits and best practices
```
**For Research Papers:**
```
Focus on:
1. Key findings and conclusions
2. Novel techniques introduced
3. Practical applications
4. Limitations and caveats
```
## Quality Metrics
Track compression efficiency:
| Metric | Target |
|--------|--------|
| Compression Ratio | 25-35% of original |
| Key Concept Coverage | ≥90% of important terms |
| Code Snippet Retention | 100% of relevant examples |
| Readability | Clear, scannable structure |
## Handling Refactor Requests
When `quality-reviewer` returns `refactor` decision:
```python
def handle_refactor(distill_id, instructions):
"""Re-distill based on reviewer feedback."""
# Load original content and existing distillation
original = load_raw_content(distill_id)
existing = load_distilled_content(distill_id)
# Apply specific improvements based on instructions
improved = apply_improvements(existing, instructions)
# Update distilled_content
update_distilled(distill_id, improved)
# Reset review status
set_review_status(distill_id, 'pending')
```
## Integration
| From | Input | To |
|------|-------|-----|
| content-repository | Raw document records | content-distiller |
| content-distiller | Distilled content | quality-reviewer |
| quality-reviewer | Refactor instructions | content-distiller (loop) |

View File

@@ -0,0 +1,229 @@
---
name: 05-quality-reviewer
description: |
Content quality evaluator with multi-criteria scoring and decision routing.
Triggers: review quality, score content, QA review, approve refactor reject.
---
# Quality Reviewer
Evaluates distilled content for quality, routes decisions, and triggers refactoring or additional research when needed.
## Review Workflow
```
[Distilled Content]
┌─────────────────┐
│ Score Criteria │ → accuracy, completeness, clarity, PE quality, usability
└─────────────────┘
┌─────────────────┐
│ Calculate Total │ → weighted average
└─────────────────┘
├── ≥ 0.85 → APPROVE → markdown-exporter
├── 0.60-0.84 → REFACTOR → content-distiller (with instructions)
├── 0.40-0.59 → DEEP_RESEARCH → web-crawler-orchestrator (with queries)
└── < 0.40 → REJECT → archive with reason
```
## Scoring Criteria
| Criterion | Weight | Checks |
|-----------|--------|--------|
| **Accuracy** | 0.25 | Factual correctness, up-to-date info, proper attribution |
| **Completeness** | 0.20 | Covers key concepts, includes examples, addresses edge cases |
| **Clarity** | 0.20 | Clear structure, concise language, logical flow |
| **PE Quality** | 0.25 | Demonstrates techniques, before/after examples, explains why |
| **Usability** | 0.10 | Easy to reference, searchable keywords, appropriate length |
## Decision Thresholds
| Score Range | Decision | Action |
|-------------|----------|--------|
| ≥ 0.85 | `approve` | Proceed to export |
| 0.60 - 0.84 | `refactor` | Return to distiller with feedback |
| 0.40 - 0.59 | `deep_research` | Gather more sources, then re-distill |
| < 0.40 | `reject` | Archive, log reason |
## Review Process
### Step 1: Load Content for Review
```python
def get_pending_reviews(cursor):
sql = """
SELECT dc.distill_id, dc.doc_id, d.title, d.url,
dc.summary, dc.key_concepts, dc.structured_content,
dc.token_count_original, dc.token_count_distilled,
s.credibility_tier
FROM distilled_content dc
JOIN documents d ON dc.doc_id = d.doc_id
JOIN sources s ON d.source_id = s.source_id
WHERE dc.review_status = 'pending'
ORDER BY s.credibility_tier ASC, dc.distill_date ASC
"""
cursor.execute(sql)
return cursor.fetchall()
```
### Step 2: Score Each Criterion
Evaluate content against each criterion using this assessment template:
```python
assessment_template = {
"accuracy": {
"score": 0.0, # 0.00 - 1.00
"notes": "",
"issues": [] # Specific factual errors if any
},
"completeness": {
"score": 0.0,
"notes": "",
"missing_topics": [] # Concepts that should be covered
},
"clarity": {
"score": 0.0,
"notes": "",
"confusing_sections": [] # Sections needing rewrite
},
"prompt_engineering_quality": {
"score": 0.0,
"notes": "",
"improvements": [] # Specific PE technique gaps
},
"usability": {
"score": 0.0,
"notes": "",
"suggestions": []
}
}
```
### Step 3: Calculate Final Score
```python
WEIGHTS = {
"accuracy": 0.25,
"completeness": 0.20,
"clarity": 0.20,
"prompt_engineering_quality": 0.25,
"usability": 0.10
}
def calculate_quality_score(assessment):
return sum(
assessment[criterion]["score"] * weight
for criterion, weight in WEIGHTS.items()
)
```
### Step 4: Route Decision
```python
def determine_decision(score, assessment):
if score >= 0.85:
return "approve", None, None
elif score >= 0.60:
instructions = generate_refactor_instructions(assessment)
return "refactor", instructions, None
elif score >= 0.40:
queries = generate_research_queries(assessment)
return "deep_research", None, queries
else:
return "reject", f"Quality score {score:.2f} below minimum threshold", None
def generate_refactor_instructions(assessment):
"""Extract actionable feedback from low-scoring criteria."""
instructions = []
for criterion, data in assessment.items():
if data["score"] < 0.80:
if data.get("issues"):
instructions.extend(data["issues"])
if data.get("missing_topics"):
instructions.append(f"Add coverage for: {', '.join(data['missing_topics'])}")
if data.get("improvements"):
instructions.extend(data["improvements"])
return "\n".join(instructions)
def generate_research_queries(assessment):
"""Generate search queries for content gaps."""
queries = []
if assessment["completeness"]["missing_topics"]:
for topic in assessment["completeness"]["missing_topics"]:
queries.append(f"{topic} documentation guide")
if assessment["accuracy"]["issues"]:
queries.append("latest official documentation verification")
return queries
```
### Step 5: Log Review Decision
```python
def log_review(cursor, distill_id, assessment, score, decision, instructions=None, queries=None):
# Get current round number
cursor.execute(
"SELECT COALESCE(MAX(review_round), 0) + 1 FROM review_logs WHERE distill_id = %s",
(distill_id,)
)
review_round = cursor.fetchone()[0]
sql = """
INSERT INTO review_logs
(distill_id, review_round, reviewer_type, quality_score, assessment,
decision, refactor_instructions, research_queries)
VALUES (%s, %s, 'claude_review', %s, %s, %s, %s, %s)
"""
cursor.execute(sql, (
distill_id, review_round, score,
json.dumps(assessment), decision, instructions,
json.dumps(queries) if queries else None
))
# Update distilled_content status
status_map = {
"approve": "approved",
"refactor": "needs_refactor",
"deep_research": "needs_refactor",
"reject": "rejected"
}
cursor.execute(
"UPDATE distilled_content SET review_status = %s WHERE distill_id = %s",
(status_map[decision], distill_id)
)
```
## Prompt Engineering Quality Checklist
When scoring `prompt_engineering_quality`, verify:
- [ ] Demonstrates specific techniques (CoT, few-shot, etc.)
- [ ] Shows before/after examples
- [ ] Explains *why* techniques work, not just *what*
- [ ] Provides actionable patterns
- [ ] Includes edge cases and failure modes
- [ ] References authoritative sources
## Auto-Approve Rules
Tier 1 (official) sources with score ≥ 0.80 may auto-approve without human review if configured:
```yaml
# In export_config.yaml
quality:
auto_approve_tier1_sources: true
auto_approve_min_score: 0.80
```
## Integration Points
| From | Action | To |
|------|--------|-----|
| content-distiller | Sends distilled content | quality-reviewer |
| quality-reviewer | APPROVE | markdown-exporter |
| quality-reviewer | REFACTOR + instructions | content-distiller |
| quality-reviewer | DEEP_RESEARCH + queries | web-crawler-orchestrator |

View File

@@ -0,0 +1,296 @@
---
name: 06-markdown-exporter
description: |
Export approved content to markdown files or JSONL for fine-tuning.
Triggers: export markdown, generate files, create JSONL, export content.
---
# Markdown Exporter
Exports approved content as structured markdown files for Claude Projects or fine-tuning.
## Export Configuration
```yaml
# ~/.config/reference-curator/export_config.yaml
output:
base_path: ~/reference-library/exports/
project_files:
structure: nested_by_topic # flat | nested_by_topic | nested_by_source
index_file: INDEX.md
include_metadata: true
fine_tuning:
format: jsonl
max_tokens_per_sample: 4096
include_system_prompt: true
quality:
min_score_for_export: 0.80
```
## Export Workflow
### Step 1: Query Approved Content
```python
def get_exportable_content(cursor, min_score=0.80, topic_filter=None):
"""Get all approved content meeting quality threshold."""
sql = """
SELECT d.doc_id, d.title, d.url,
dc.summary, dc.key_concepts, dc.code_snippets, dc.structured_content,
t.topic_slug, t.topic_name,
rl.quality_score, s.credibility_tier, s.vendor
FROM documents d
JOIN distilled_content dc ON d.doc_id = dc.doc_id
JOIN document_topics dt ON d.doc_id = dt.doc_id
JOIN topics t ON dt.topic_id = t.topic_id
JOIN review_logs rl ON dc.distill_id = rl.distill_id
JOIN sources s ON d.source_id = s.source_id
WHERE rl.decision = 'approve'
AND rl.quality_score >= %s
AND rl.review_id = (
SELECT MAX(review_id) FROM review_logs
WHERE distill_id = dc.distill_id
)
"""
params = [min_score]
if topic_filter:
sql += " AND t.topic_slug IN (%s)" % ','.join(['%s'] * len(topic_filter))
params.extend(topic_filter)
sql += " ORDER BY t.topic_slug, rl.quality_score DESC"
cursor.execute(sql, params)
return cursor.fetchall()
```
### Step 2: Organize by Structure
**Nested by Topic (recommended):**
```
exports/
├── INDEX.md
├── prompt-engineering/
│ ├── _index.md
│ ├── 01-chain-of-thought.md
│ ├── 02-few-shot-prompting.md
│ └── 03-system-prompts.md
├── claude-models/
│ ├── _index.md
│ ├── 01-model-comparison.md
│ └── 02-context-windows.md
└── agent-building/
├── _index.md
└── 01-tool-use.md
```
**Flat Structure:**
```
exports/
├── INDEX.md
├── prompt-engineering-chain-of-thought.md
├── prompt-engineering-few-shot.md
└── claude-models-comparison.md
```
### Step 3: Generate Files
**Document File Template:**
```python
def generate_document_file(doc, include_metadata=True):
content = []
if include_metadata:
content.append("---")
content.append(f"title: {doc['title']}")
content.append(f"source: {doc['url']}")
content.append(f"vendor: {doc['vendor']}")
content.append(f"tier: {doc['credibility_tier']}")
content.append(f"quality_score: {doc['quality_score']:.2f}")
content.append(f"exported: {datetime.now().isoformat()}")
content.append("---")
content.append("")
content.append(doc['structured_content'])
return "\n".join(content)
```
**Topic Index Template:**
```python
def generate_topic_index(topic_slug, topic_name, documents):
content = [
f"# {topic_name}",
"",
f"This section contains {len(documents)} reference documents.",
"",
"## Contents",
""
]
for i, doc in enumerate(documents, 1):
filename = generate_filename(doc['title'])
content.append(f"{i}. [{doc['title']}]({filename})")
return "\n".join(content)
```
**Root INDEX Template:**
```python
def generate_root_index(topics_with_counts, export_date):
content = [
"# Reference Library",
"",
f"Exported: {export_date}",
"",
"## Topics",
""
]
for topic in topics_with_counts:
content.append(f"- [{topic['name']}]({topic['slug']}/) ({topic['count']} documents)")
content.extend([
"",
"## Quality Standards",
"",
"All documents in this library have:",
"- Passed quality review (score ≥ 0.80)",
"- Been distilled for conciseness",
"- Verified source attribution"
])
return "\n".join(content)
```
### Step 4: Write Files
```python
def export_project_files(content_list, config):
base_path = Path(config['output']['base_path'])
structure = config['output']['project_files']['structure']
# Group by topic
by_topic = defaultdict(list)
for doc in content_list:
by_topic[doc['topic_slug']].append(doc)
# Create directories and files
for topic_slug, docs in by_topic.items():
if structure == 'nested_by_topic':
topic_dir = base_path / topic_slug
topic_dir.mkdir(parents=True, exist_ok=True)
# Write topic index
topic_index = generate_topic_index(topic_slug, docs[0]['topic_name'], docs)
(topic_dir / '_index.md').write_text(topic_index)
# Write document files
for i, doc in enumerate(docs, 1):
filename = f"{i:02d}-{slugify(doc['title'])}.md"
file_content = generate_document_file(doc)
(topic_dir / filename).write_text(file_content)
# Write root INDEX
topics_summary = [
{"slug": slug, "name": docs[0]['topic_name'], "count": len(docs)}
for slug, docs in by_topic.items()
]
root_index = generate_root_index(topics_summary, datetime.now().isoformat())
(base_path / 'INDEX.md').write_text(root_index)
```
### Step 5: Fine-tuning Export (Optional)
```python
def export_fine_tuning_dataset(content_list, config):
"""Export as JSONL for fine-tuning."""
output_path = Path(config['output']['base_path']) / 'fine_tuning.jsonl'
max_tokens = config['output']['fine_tuning']['max_tokens_per_sample']
with open(output_path, 'w') as f:
for doc in content_list:
sample = {
"messages": [
{
"role": "system",
"content": "You are an expert on AI and prompt engineering."
},
{
"role": "user",
"content": f"Explain {doc['title']}"
},
{
"role": "assistant",
"content": truncate_to_tokens(doc['structured_content'], max_tokens)
}
],
"metadata": {
"source": doc['url'],
"topic": doc['topic_slug'],
"quality_score": doc['quality_score']
}
}
f.write(json.dumps(sample) + '\n')
```
### Step 6: Log Export Job
```python
def log_export_job(cursor, export_name, export_type, output_path,
topic_filter, total_docs, total_tokens):
sql = """
INSERT INTO export_jobs
(export_name, export_type, output_format, topic_filter, output_path,
total_documents, total_tokens, status, started_at, completed_at)
VALUES (%s, %s, 'markdown', %s, %s, %s, %s, 'completed', NOW(), NOW())
"""
cursor.execute(sql, (
export_name, export_type,
json.dumps(topic_filter) if topic_filter else None,
str(output_path), total_docs, total_tokens
))
```
## Cross-Reference Generation
Link related documents:
```python
def add_cross_references(doc, all_docs):
"""Find and link related documents."""
related = []
doc_concepts = set(c['term'].lower() for c in doc['key_concepts'])
for other in all_docs:
if other['doc_id'] == doc['doc_id']:
continue
other_concepts = set(c['term'].lower() for c in other['key_concepts'])
overlap = len(doc_concepts & other_concepts)
if overlap >= 2:
related.append({
"title": other['title'],
"path": generate_relative_path(doc, other),
"overlap": overlap
})
return sorted(related, key=lambda x: x['overlap'], reverse=True)[:5]
```
## Output Verification
After export, verify:
- [ ] All files readable and valid markdown
- [ ] INDEX.md links resolve correctly
- [ ] No broken cross-references
- [ ] Total token count matches expectation
- [ ] No duplicate content
## Integration
| From | Input | To |
|------|-------|-----|
| quality-reviewer | Approved content IDs | markdown-exporter |
| markdown-exporter | Structured files | Project knowledge / Fine-tuning |

View File

@@ -0,0 +1,286 @@
---
name: 07-pipeline-orchestrator
description: |
Full reference curation pipeline coordinator with QA loop and state management.
Triggers: run pipeline, orchestrate workflow, full curation, pipeline start.
---
# Pipeline Orchestrator
Coordinates the full reference curation workflow, handling QA loops and state management.
## Pipeline Architecture
```
[Input: Topic | URLs | Manifest]
1. reference-discovery ──────────────────┐
(skip if URLs/manifest) │
│ │
▼ │
2. web-crawler-orchestrator │
│ │
▼ │
3. content-repository │
│ │
▼ │
4. content-distiller ◄───────────────────┤
│ │
▼ │
5. quality-reviewer │
│ │
┌─────┼─────┬────────────────┐ │
▼ ▼ ▼ ▼ │
APPROVE REJECT REFACTOR DEEP_RESEARCH│
│ │ │ │ │
│ │ └─────────────┤ │
│ │ └───────┘
▼ ▼
6. markdown-exporter archive
[Complete]
```
## Input Modes
| Mode | Example Input | Pipeline Start |
|------|--------------|----------------|
| **Topic** | `"Claude system prompts"` | Stage 1 (discovery) |
| **URLs** | `["https://docs.anthropic.com/..."]` | Stage 2 (crawler) |
| **Manifest** | Path to `manifest.json` | Stage 2 (crawler) |
## Configuration Options
```yaml
pipeline:
max_sources: 10 # Discovery limit
max_pages: 50 # Pages per source
auto_approve: false # Auto-approve above threshold
approval_threshold: 0.85
qa_loop:
max_refactor_iterations: 3
max_deep_research_iterations: 2
max_total_iterations: 5
export:
format: project_files # or fine_tuning, jsonl
```
## Pipeline Execution
### Stage 1: Reference Discovery
For topic-based input, search and validate authoritative sources:
```python
def run_discovery(topic, max_sources=10):
# Uses WebSearch to find sources
# Validates credibility
# Outputs manifest.json with source URLs
sources = search_authoritative_sources(topic, max_sources)
validate_and_rank_sources(sources)
write_manifest(sources)
return manifest_path
```
### Stage 2: Web Crawler
Crawl URLs from manifest or direct input:
```python
def run_crawler(input_source, max_pages=50):
# Selects optimal crawler backend
# Respects rate limits
# Stores raw content
urls = load_urls(input_source)
for url in urls:
crawl_with_best_backend(url, max_pages)
return crawl_results
```
### Stage 3: Content Repository
Store crawled content with deduplication:
```python
def run_repository(crawl_results):
# Deduplicates by URL hash
# Tracks versions
# Returns stored doc IDs
for result in crawl_results:
store_document(result)
return stored_doc_ids
```
### Stage 4: Content Distiller
Process raw content into structured summaries:
```python
def run_distiller(doc_ids, refactor_instructions=None):
# Extracts key concepts
# Generates summaries
# Creates structured markdown
for doc_id in doc_ids:
distill_document(doc_id, instructions=refactor_instructions)
return distilled_ids
```
### Stage 5: Quality Reviewer
Score and route content based on quality:
```python
def run_reviewer(distilled_ids, auto_approve=False, threshold=0.85):
decisions = {}
for distill_id in distilled_ids:
score, assessment = score_content(distill_id)
if auto_approve and score >= threshold:
decisions[distill_id] = ('approve', None)
elif score >= 0.85:
decisions[distill_id] = ('approve', None)
elif score >= 0.60:
instructions = generate_feedback(assessment)
decisions[distill_id] = ('refactor', instructions)
elif score >= 0.40:
queries = generate_research_queries(assessment)
decisions[distill_id] = ('deep_research', queries)
else:
decisions[distill_id] = ('reject', assessment)
return decisions
```
### Stage 6: Markdown Exporter
Export approved content:
```python
def run_exporter(approved_ids, format='project_files'):
# Organizes by topic
# Generates INDEX.md
# Creates cross-references
export_documents(approved_ids, format=format)
return export_path
```
## QA Loop Handling
```python
def handle_qa_loop(distill_id, decision, iteration_tracker):
counts = iteration_tracker.get(distill_id, {'refactor': 0, 'deep_research': 0})
if decision == 'refactor':
if counts['refactor'] >= MAX_REFACTOR:
return 'needs_manual_review'
counts['refactor'] += 1
iteration_tracker[distill_id] = counts
return 're_distill'
if decision == 'deep_research':
if counts['deep_research'] >= MAX_DEEP_RESEARCH:
return 'needs_manual_review'
counts['deep_research'] += 1
iteration_tracker[distill_id] = counts
return 're_crawl'
return decision
```
## State Management
### MySQL Backend (Preferred)
```sql
SELECT run_id, status, current_stage, stats
FROM pipeline_runs
WHERE run_id = ?;
```
### File-Based Fallback
```
~/reference-library/pipeline_state/
├── run_001/
│ ├── state.json # Pipeline state
│ ├── manifest.json # Discovered sources
│ ├── crawl_results.json
│ └── review_log.json # QA decisions
```
State JSON format:
```json
{
"run_id": "run_001",
"run_type": "topic",
"input_value": "Claude system prompts",
"status": "running",
"current_stage": "distilling",
"stats": {
"sources_discovered": 5,
"pages_crawled": 45,
"approved": 0,
"refactored": 0
},
"started_at": "2026-01-29T10:00:00Z"
}
```
## Checkpointing
Checkpoint after each stage to enable resume:
| Checkpoint | Trigger | Resume From |
|------------|---------|-------------|
| `discovery_complete` | Manifest saved | → crawler |
| `crawl_complete` | All pages crawled | → repository |
| `store_complete` | Docs in database | → distiller |
| `distill_complete` | Content processed | → reviewer |
| `review_complete` | Decisions logged | → exporter |
| `export_complete` | Files generated | Done |
## Output Summary
```json
{
"run_id": 123,
"status": "completed",
"duration_minutes": 15,
"stats": {
"sources_discovered": 5,
"pages_crawled": 45,
"documents_stored": 45,
"documents_distilled": 45,
"approved": 40,
"refactored": 8,
"deep_researched": 2,
"rejected": 3,
"needs_manual_review": 2
},
"exports": {
"format": "project_files",
"path": "~/reference-library/exports/",
"document_count": 40
}
}
```
## Error Handling
On stage failure:
1. Save checkpoint with error state
2. Log error details
3. Report to user with resume instructions
```python
try:
run_stage(stage_name)
save_checkpoint(stage_name, 'complete')
except Exception as e:
save_checkpoint(stage_name, 'failed', error=str(e))
report_error(f"Pipeline paused at {stage_name}: {e}")
```

View File

@@ -0,0 +1,78 @@
---
name: 90-reference-curator
description: |
Full reference-documentation curation pipeline: discover authoritative sources →
crawl → store → distill → quality-review (QA loop) → export to markdown / project files
/ fine-tuning JSONL. A suite of 7 composable sub-skills with a single orchestrator entry.
Triggers: reference curator, curate documentation, build reference library, research
pipeline, discover sources, crawl docs, distill content, export reference, 레퍼런스 큐레이션,
문서 수집 파이프라인.
version: "1.0"
author: OurDigital
environment: Code
---
# Reference Curator (90)
A modular suite that turns a topic or a set of URLs into a curated reference library.
Six stages run as a pipeline with a quality-review **QA loop**; each stage is also a
standalone sub-skill you can run on its own.
## Pipeline
```
discover → crawl → store → distill → review (QA loop) → export
01 02 03 04 05 06
orchestrated by 07
```
| # | Sub-skill | Does | Root directive |
|---|-----------|------|----------------|
| 01 | reference-discovery | Find + credibility-rank authoritative sources | `01-reference-discovery/SKILL.md` |
| 02 | web-crawler-orchestrator | Multi-backend crawl (Firecrawl / Node / aiohttp / Scrapy) | `02-web-crawler-orchestrator/SKILL.md` |
| 03 | content-repository | MySQL storage with version tracking | `03-content-repository/SKILL.md` |
| 04 | content-distiller | Summarize + extract key concepts | `04-content-distiller/SKILL.md` |
| 05 | quality-reviewer | QA loop: approve / refactor / re-research routing | `05-quality-reviewer/SKILL.md` |
| 06 | markdown-exporter | Export to markdown / project files / fine-tuning JSONL | `06-markdown-exporter/SKILL.md` |
| 07 | pipeline-orchestrator | Coordinates all stages + QA loop + state | `07-pipeline-orchestrator/SKILL.md` |
## How to run
**Orchestrated (recommended)** — the `/reference-curator` command (see
`commands/reference-curator-pipeline.md`):
```bash
# From a topic (starts at discovery)
/reference-curator "Claude Code best practices" --max-sources 5
# From URLs (skips discovery)
/reference-curator https://docs.anthropic.com/en/docs/prompt-caching
# Auto-approve + fine-tuning output
/reference-curator "MCP servers" --auto-approve --export-format fine_tuning
```
Input modes: **topic** (→ discovery), **URLs** (→ crawl), **manifest** (→ resume).
Key flags: `--depth light|standard|deep|full`, `--output`, `--max-sources`, `--max-pages`,
`--auto-approve`, `--threshold`, `--max-iterations`, `--export-format project_files|fine_tuning|jsonl`.
**Individual sub-skills** — each has a slash command in `commands/` and its own root
`SKILL.md`; run any stage standalone (e.g. just discovery, or just export).
## Install
```bash
cd custom-skills/90-reference-curator
./install.sh # interactive (storage dir, MySQL optional, crawler backend)
./install.sh --minimal # Firecrawl only, no MySQL
./install.sh --check # verify
```
Full guide: `USER-GUIDE.md`. Changelog: `CHANGELOG.md`.
## Notes
- Crawler backends and MySQL are optional — `--minimal` runs Firecrawl-only with no DB.
- The QA loop (stage 05) gates quality before export; `--threshold` / `--max-iterations`
tune how strict and how many refinement passes per document.
- This is a one-shot-per-topic curation workflow, not an always-on service.

View File

@@ -56,9 +56,9 @@ headless Chrome, python-pptx. Create `data/` subfolder. Initialize `findings.jso
## Stage 5 — Estimate (견적) — REVIEW GATE
- `python scripts/estimate.py --findings <out>/data/findings.json --rate-card references/rate_card.yaml --sow references/sow_templates.yaml --out-dir <out> --seq <N> [--baseline basic|treatment] [--billing 0.70]`
- **Effort-based** (OurDigital real model): cost = role_rate × 청구율 70% × 표준 업무시간, by module; 제안가 = 합계 절사. findings auto-select baseline (basic/treatment) and scale Technical/On-page hours by `properties_total`.
- **Effort-based** (OurDigital real model): cost = role_rate × 청구율 × 표준 업무시간, by module; 제안가 = 합계 절사. findings auto-select a tier (**smb / basic / treatment**) and scale **On-page** hours sub-linearly by `subbrands_total` (cap ×2.0); Technical is fixed. `smb` tier bills at 0.55, others 0.70. Override with `--baseline` / `--billing` (quote premium single properties as basic/treatment manually).
- Produces `05_estimate_ko.md`, `05_estimate.xlsx`, `data/estimate.json`. Present the 견적; get sign-off.
- Logic in `references/findings_to_service.md`; rates/hours in `rate_card.yaml` + `sow_templates.yaml` (edit together). Reproduces real Basic ₩10.5M / Treatment ₩25.0M quotes.
- Logic in `references/findings_to_service.md`; rates/hours in `rate_card.yaml` + `sow_templates.yaml` (edit together). Reproduces real Basic ₩10.5M / Treatment ₩25.0M; SMB entry ~₩3M; chains stay SMB-acceptable (e.g. 25-property → ~₩29.5M).
## Stage 6 — Deliverables — REVIEW GATE before send
- **Client PDF**: author the short brief HTML from `templates/client_brief.html` (fill the content; keep the CSS),

View File

@@ -1,19 +1,22 @@
# Findings → estimate mapping (effort-based)
# Findings → estimate mapping (effort-based, SMB-aware)
`estimate.py` builds the 견적 from `sow_templates.yaml` priced via `rate_card.yaml`:
**cost = role_rate × billing_rate (0.70) × standard_hours**, grouped by module;
제안가 = 합계 floored to `rounding_unit`. This mirrors OurDigital/D.intelligence's
real SOW-based quoting.
**cost = role_rate × billing_rate × standard_hours**, grouped by module; 제안가 = 합계
floored to `rounding_unit`. Mirrors OurDigital/D.intelligence's real SOW-based quoting.
## Baseline selection (basic vs treatment)
- `treatment` if any finding `severity == critical` **OR** `entity.properties_total > 3`
- else `basic`
- override with `--baseline`.
## Tier selection (smb / basic / treatment)
- `treatment` if `properties_total > 5` OR `subbrands_total > 3` (multi-brand / chain)
- `smb` if `properties_total <= 1` AND `subbrands_total == 0` (single-property SMB)
- `basic` otherwise (small multi-property / mid)
- override with `--baseline`. **Note:** tiering is by portfolio size only — quote a
single *premium* property as `basic`/`treatment` manually.
Per-tier billing: `smb` uses `billing_rate: 0.55` (set in sow_templates); `basic`/
`treatment` use `rate_card.billing_rate` (0.70). Override with `--billing`.
## Module inclusion
Each baseline carries the standard module set (P&M · Technical SEO · On-page SEO ·
SEO Growth), matching real quotes. Findings justify modules via the `trigger` field
in `sow_templates.yaml`:
Each tier carries its standard module set (P&M · Technical SEO · On-page SEO ·
SEO Growth). Findings annotate modules via the `trigger` field:
| Module | trigger finding classes |
|---|---|
@@ -23,18 +26,21 @@ in `sow_templates.yaml`:
| SEO Growth | measurement, always |
## Hours scaling (portfolio)
Tasks marked `scale: true` (Technical SEO + On-page SEO) have their **hours**
multiplied sub-linearly by `entity.properties_total` per `rate_card.scaling.bands`.
P&M and SEO Growth stay fixed (management/KPI overhead is ~flat). A single-property
prospect → ×1.0, which reproduces the real single-site quotes exactly.
Only **On-page** tasks (`scale: true`) scale, sub-linearly, by the
`rate_card.scaling.driver` (= `subbrands_total`, i.e. distinct brands/templates),
capped at **×2.0**. Rationale: chains share page templates, so on-page work grows
with template variety, not raw property count. **Technical SEO is fixed** site-wide
work (`scale: false`); P&M and SEO Growth are fixed too. Single-brand → ×1.0.
## Tools & terms
`tools` (e.g. SEMrush Guru) are listed separately — client-subscribed, with +15%
procurement markup if billed through us. VAT 별도 · 유효기간 14d · 현금 · 절사 from
`rate_card.terms`.
`tools` (e.g. SEMrush Guru) listed separately — client-subscribed, +15% procurement
markup if billed through us. VAT 별도 · 유효기간 14d · 현금 · 절사 from `rate_card.terms`.
## Validated reproduction (2026-05-28)
## Validated (2026-05-28)
- `basic` ×1.0 → 합계 10,612,000 → 제안가 **10,500,000** (real Basic quote)
- `treatment` ×1.0 → 합계 25,340,000 → 제안가 **25,000,000** (real Treatment quote)
- `basic` ×1.0 → 합계 10,612,000 → 제안가 **10,500,000** (real Basic quote)
- `smb` single property → ~**3,000,000** (lean × 55% billing)
- chain example (SHR, 5 sub-brands ×1.6, treatment) → ~**29,500,000** (vs naive 71.5M)
Edit `rate_card.yaml` and `sow_templates.yaml` together when rates or standard hours change.
Edit `rate_card.yaml` + `sow_templates.yaml` together when rates, hours, tiers, or
scaling change.

View File

@@ -42,11 +42,13 @@ role_rates:
associate: 30000 # 사원
intern: 12000 # 인턴
# 포트폴리오 규모에 따른 '시간' 스케일(서브선형). scale:true task 에만 적용.
# driver: findings.entity 의 카운트. bands: [최대값, 배수]; count <= 최대값 인 첫 밴드.
# 포트폴리오 규모에 따른 '시간' 스케일(서브선형). On-page(scale:true) task 에만 적용.
# driver = findings.entity 의 카운트. 체인은 페이지 템플릿을 공유하므로 '프로퍼티 수'가 아닌
# '브랜드/템플릿 수(subbrands_total)'를 기준으로 하고, 캡을 ×2.0 로 낮게 둔다.
# (Technical SEO 는 사이트 단위 고정 작업이므로 스케일하지 않음 — sow_templates 의 scale:false)
scaling:
driver: properties_total
bands: [[1, 1.0], [5, 1.6], [15, 2.8], [30, 4.5], [999999, 6.5]]
driver: subbrands_total
bands: [[1, 1.0], [3, 1.3], [6, 1.6], [999999, 2.0]]
# 별도 조달 항목(인력비와 분리). 청구 시 procurement_markup 적용 가능.
tools:

View File

@@ -1,13 +1,40 @@
# SOW task templates — standard 업무 시간(hours) by module.
# Seeded from the two real OurDigital quotes so estimate.py reproduces them at
# billing_rate 0.70:
# basic -> 제안가 ₩10,500,000 (SEO Basic & Coaching, 3개월 프로젝트)
# treatment -> 제안가 ₩25,000,000 (SEO Audit & Treatment, 월 정기)
# task.role references rate_card.role_rates. scale:true → hours scaled by portfolio.
# trigger: finding classes that justify the module (for annotation + selection).
# Priced via rate_card.yaml: cost = role_rate × billing_rate × hours.
# Tiers (제안가 at default billing): smb < basic < treatment.
# basic -> ₩10,500,000 (real SEO Basic & Coaching quote, billing 0.70)
# treatment -> ₩25,000,000 (real SEO Audit & Treatment quote, billing 0.70)
# smb -> lean entry tier for single-property SMBs (billing 0.55)
# Scaling: only On-page tasks (scale:true) scale by sub-brands/templates
# (rate_card.scaling); Technical SEO is fixed site-wide work. P&M & Growth fixed.
# Each baseline may override billing_rate; else rate_card.billing_rate (0.70).
baselines:
smb:
service: "SEO Quick Audit (SMB)"
billing_rate: 0.55 # SMB 진입 티어 — 낮은 청구율
modules:
- name: "Planning & Management"
trigger: [always]
tasks:
- {task: "업무 관리·착수", desc: "과업 정의·일정·리포팅", role: senior_manager, hours: 8, scale: false}
- {task: "웹 사이트 분석", desc: "유입·행동·전환 flow 요약 분석", role: senior_manager, hours: 8, scale: false}
- name: "Technical SEO"
trigger: [crawlability, cwv, schema_entity]
tasks:
- {task: "Crawling·Indexing 점검", desc: "검색사이트 등록·수집·Site Health 점검", role: manager, hours: 8, scale: false}
- {task: "속도·CWV 점검", desc: "로딩속도·Core Web Vitals·모바일 점검", role: manager, hours: 8, scale: false}
- {task: "구조·메타·사이트맵 점검", desc: "사이트/URL 구조·메타·사이트맵·색인", role: manager, hours: 8, scale: false}
- name: "On-page SEO"
trigger: [onpage, schema_entity]
tasks:
- {task: "키워드·메타 진단", desc: "중점 키워드·페이지 메타·템플릿 진단", role: manager, hours: 12, scale: true}
- {task: "on-page 퀵윈 가이드", desc: "타이틀·메타·헤더·링크·이미지 핵심 개선 가이드", role: manager, hours: 16, scale: true}
- name: "SEO Growth"
trigger: [measurement, always]
tasks:
- {task: "기본 성과 지표 설정", desc: "핵심 SEO 지표·중점 키워드 트래킹 설정", role: manager, hours: 8, scale: false}
basic:
service: "SEO Audit & Basic Treatment"
modules:
@@ -19,10 +46,10 @@ baselines:
- name: "Technical SEO"
trigger: [crawlability, cwv, schema_entity]
tasks:
- {task: "Crawling & Indexing 설정", desc: "검색사이트 등록/수집 관리, Site Health Check 도구 설정", role: technical_advisor, hours: 16, scale: true}
- {task: "속도·UX·수집 설정", desc: "로딩속도·페이지 UX·링크·수집 제외 설정", role: manager, hours: 16, scale: true}
- {task: "사이트/URL 구조·메타", desc: "구조·URL·메타데이터·사이트맵·리다이렉션", role: senior_manager, hours: 12, scale: true}
- {task: "색인·CWV 진단", desc: "GSC·SEO Tools 활용 색인/크롤오류/Core Web Vitals 진단", role: senior_manager, hours: 16, scale: true}
- {task: "Crawling & Indexing 설정", desc: "검색사이트 등록/수집 관리, Site Health Check 도구 설정", role: technical_advisor, hours: 16, scale: false}
- {task: "속도·UX·수집 설정", desc: "로딩속도·페이지 UX·링크·수집 제외 설정", role: manager, hours: 16, scale: false}
- {task: "사이트/URL 구조·메타", desc: "구조·URL·메타데이터·사이트맵·리다이렉션", role: senior_manager, hours: 12, scale: false}
- {task: "색인·CWV 진단", desc: "GSC·SEO Tools 활용 색인/크롤오류/Core Web Vitals 진단", role: senior_manager, hours: 16, scale: false}
- name: "On-page SEO"
trigger: [onpage, schema_entity]
tasks:
@@ -46,10 +73,10 @@ baselines:
- name: "Technical SEO"
trigger: [crawlability, cwv, schema_entity]
tasks:
- {task: "Crawling & Indexing 설정", desc: "검색사이트 등록/수집 관리, Site Health Check 도구 설정", role: manager, hours: 24, scale: true}
- {task: "속도·UX·리다이렉트", desc: "로딩속도·페이지 UX·링크·리다이렉트·수집 제외 설정", role: technical_advisor, hours: 30, scale: true}
- {task: "사이트/URL 구조·보안", desc: "구조·URL·메타데이터·사이트맵·보안 관리 진단", role: senior_manager, hours: 24, scale: true}
- {task: "모바일·CWV·개선과제", desc: "모바일 최적화·Core Web Vitals 진단·개선 과제 도출", role: manager, hours: 20, scale: true}
- {task: "Crawling & Indexing 설정", desc: "검색사이트 등록/수집 관리, Site Health Check 도구 설정", role: manager, hours: 24, scale: false}
- {task: "속도·UX·리다이렉트", desc: "로딩속도·페이지 UX·링크·리다이렉트·수집 제외 설정", role: technical_advisor, hours: 30, scale: false}
- {task: "사이트/URL 구조·보안", desc: "구조·URL·메타데이터·사이트맵·보안 관리 진단", role: senior_manager, hours: 24, scale: false}
- {task: "모바일·CWV·개선과제", desc: "모바일 최적화·Core Web Vitals 진단·개선 과제 도출", role: manager, hours: 20, scale: false}
- name: "On-page SEO"
trigger: [onpage, schema_entity]
tasks:

View File

@@ -286,7 +286,8 @@ def main():
sc = EST.get("scope", {})
note = f"청구율 {int(EST.get('billing_rate', 0.7) * 100)}% · 일8h/월4주 · SOW 기반"
if sc.get("hours_multiplier", 1.0) != 1.0:
note += f" · 프로퍼티 {sc.get('properties_total')}×{sc['hours_multiplier']:g}"
dl = "브랜드/템플릿" if sc.get("driver") == "subbrands_total" else "프로퍼티"
note += f" · {dl} {sc.get('driver_count')}×{sc['hours_multiplier']:g}"
textbox(s, 0.85, 2.2 + 0.5 * rows, 11.6, 1.3, [
[(note, 10, GREY, False)],
[(EST.get("disclaimer", ""), 9, GREY, False)],

View File

@@ -32,28 +32,34 @@ def won(n):
return f"{int(round(n)):,}"
def scope_multiplier(rate, count):
def scope_multiplier(rate, f):
"""Sub-linear hours multiplier from the configured driver (default subbrands_total)."""
sc = rate.get("scaling", {})
driver = sc.get("driver", "subbrands_total")
bands = sc.get("bands", [[1, 1.0]])
c = max(int(count or 0), 1)
count = max(int(f.get("entity", {}).get(driver, 0) or 0), 1)
for mx, m in bands:
if c <= mx:
return float(m)
return float(bands[-1][1])
if count <= mx:
return float(m), driver, count
return float(bands[-1][1]), driver, count
def pick_baseline(f, override):
if override:
return override
severities = {x.get("severity") for x in f.get("findings", [])}
props = f.get("entity", {}).get("properties_total", 0) or 0
return "treatment" if ("critical" in severities or props > 3) else "basic"
e = f.get("entity", {})
props = e.get("properties_total", 0) or 0
subs = e.get("subbrands_total", 0) or 0
if props > 5 or subs > 3: # multi-brand / chain
return "treatment"
if props <= 1 and subs == 0: # single-property SMB
return "smb"
return "basic" # small multi-property / mid
def assemble(f, rate, sow, baseline, billing):
roles = rate["role_rates"]
props = f.get("entity", {}).get("properties_total", 0)
mult = scope_multiplier(rate, props)
mult, driver, dcount = scope_multiplier(rate, f)
tpl = sow["baselines"][baseline]
modules = []
grand = 0.0
@@ -72,7 +78,7 @@ def assemble(f, rate, sow, baseline, billing):
sub += amount
modules.append({"name": mod["name"], "subtotal": sub, "tasks": tasks})
grand += sub
return modules, grand, mult, props, tpl["service"]
return modules, grand, mult, driver, dcount, tpl["service"]
ROLE_KO = {
@@ -91,7 +97,8 @@ def write_md(path, q):
f"- **산정 기준**: SOW 기반 · 청구율 {int(q['billing_rate']*100)}% · 일 8시간/월 4주 · {q['terms']['vat']} · 지급 {q['terms']['payment']}",
""]
if q["scope"]["hours_multiplier"] != 1.0:
L.append(f"> 포트폴리오 규모 반영: 프로퍼티 {q['scope']['properties_total']}개 기준 Technical/On-page 업무시간 ×{q['scope']['hours_multiplier']:g} (서브선형)")
dl = "브랜드/템플릿" if q["scope"]["driver"] == "subbrands_total" else "프로퍼티"
L.append(f"> 규모 반영: {dl} {q['scope']['driver_count']}개 기준 On-page 업무시간 ×{q['scope']['hours_multiplier']:g} (서브선형)")
L.append("")
L += ["## 견적 내역", "",
"| 구분 | 세부 업무 | 담당 | 시간(h) | 합계 |",
@@ -158,7 +165,7 @@ def main():
ap.add_argument("--sow", required=True)
ap.add_argument("--out-dir", default=".")
ap.add_argument("--seq", type=int, default=1)
ap.add_argument("--baseline", choices=["basic", "treatment"], default=None)
ap.add_argument("--baseline", choices=["smb", "basic", "treatment"], default=None)
ap.add_argument("--billing", type=float, default=None)
args = ap.parse_args()
@@ -169,9 +176,13 @@ def main():
with open(args.sow, encoding="utf-8") as fh:
sow = yaml.safe_load(fh)
billing = args.billing if args.billing is not None else rate["billing_rate"]
baseline = pick_baseline(f, args.baseline)
modules, grand, mult, props, service = assemble(f, rate, sow, baseline, billing)
tpl_billing = sow["baselines"][baseline].get("billing_rate")
billing = (args.billing if args.billing is not None
else tpl_billing if tpl_billing is not None else rate["billing_rate"])
modules, grand, mult, driver, dcount, service = assemble(f, rate, sow, baseline, billing)
props = f.get("entity", {}).get("properties_total", 0)
subs = f.get("entity", {}).get("subbrands_total", 0)
rounding = rate["rounding_unit"]
proposal = int(math.floor(grand / rounding) * rounding)
@@ -190,8 +201,8 @@ def main():
"prospect": f.get("prospect", {}).get("name", "(prospect)"),
"service": service, "baseline": baseline, "billing_rate": billing,
"company": rate["company"], "terms": rate["terms"],
"scope": {"properties_total": props,
"subbrands_total": f.get("entity", {}).get("subbrands_total", 0),
"scope": {"driver": driver, "driver_count": dcount,
"properties_total": props, "subbrands_total": subs,
"hours_multiplier": mult},
"modules": modules, "subtotal_sum": grand, "proposal": proposal,
"rounding_unit": rounding,
@@ -209,7 +220,7 @@ def main():
json.dump(q, fh, ensure_ascii=False, indent=2)
print(f"견적 {quote_no} [{baseline}] 제안가 {won(proposal)} (합계 {won(grand)}) "
f"| 프로퍼티 {props} ×{mult:g} | 청구율 {int(billing*100)}%")
f"| {driver}={dcount} ×{mult:g} | 청구율 {int(billing*100)}%")
for m in modules:
print(f" {m['name']:24} {won(m['subtotal'])}")

View File

@@ -45,6 +45,21 @@ def find_source(d):
return None
def discover_skill_dirs():
"""Top-level skill dirs PLUS any nested dir that has a desktop/ or code/ SKILL.md
(e.g. the sub-skills of a suite like 90-reference-curator). Plugins keep their skill
at skills/<name>/SKILL.md with no desktop/ or code/, so they are correctly left alone."""
found = {}
for d in SKILLS.iterdir(): # top-level: report SKIP/MANUAL/CREATE
if is_skill_dir(d):
found[d] = True
for src in list(SKILLS.rglob("desktop/SKILL.md")) + list(SKILLS.rglob("code/SKILL.md")):
d = src.parent.parent # nested sub-skills (suite members)
if is_skill_dir(d):
found[d] = True
return sorted(found)
def set_name(frontmatter, name):
"""Replace the single-line `name:` value (or prepend one) with `name`."""
if re.search(r"^name:.*$", frontmatter, re.M):
@@ -80,10 +95,8 @@ def main(argv=None):
rows = [] # (dir, status, detail)
created = skipped = manual = warned = 0
for d in sorted(SKILLS.iterdir()):
if not is_skill_dir(d):
continue
name = d.name
for d in discover_skill_dirs():
name = str(d.relative_to(SKILLS))
if (d / "SKILL.md").exists():
rows.append((name, "SKIP", "already has root SKILL.md"))
skipped += 1
@@ -93,7 +106,7 @@ def main(argv=None):
rows.append((name, "MANUAL", "no desktop/ or code/ SKILL.md source (commands/README only)"))
manual += 1
continue
text, issues = build_root_skill(src.read_text(encoding="utf-8"), name)
text, issues = build_root_skill(src.read_text(encoding="utf-8"), d.name)
rel = src.relative_to(d)
if text is None:
rows.append((name, "MANUAL", f"{rel}: {issues[0]}"))