算法艺术创作
algorithmic-art
通过代码和可调参数创作生成艺术与交互式视觉效果。
apify-brand-reputation-monitoring【设计与创意】Implements intelligent apify brand reputation monitoring with multi-factor skill selection, fallback chains, and adherence to the 5 Laws of Elegant Defense
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: apify-brand-reputation-monitoring compatibility: opencode completeness: 95 content-types:
Orchestrates intelligent skill selection and execution for apify brand reputation monitoring workflows. Applies the 5 Laws of Elegant Defense to guide data naturally through the orchestration pipeline, preventing errors before they occur. Selects optimal skills based on multi-factor scoring including text similarity, historical performance, and system availability.
┌───────────────────────────────────────────────────────────────────────────────┐ │ Orchestration Flow │ └───────────────────────────────────────────────────────────────────────────────┘
User Request ↓ ┌─────────────────┐ │ Parse Request │ │ & Extract │ │ Features │ └────────┬────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────┐ │ Evaluate Available Skills │ │ │ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │ │ │ Skill A │ │ Skill B │ │ Skill C │ │ │ │ - Match Score│ │ - Match Score│ │ - Match Score│ │ │ │ - Confidence │ │ - Confidence │ │ - Confidence │ │ │ │ - History │ │ - History │ │ - History │ │ │ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │ │ │ │ │ │ │ └─────────────────┴─────────────────┘ │ │ ↓ │ │ Select Best Skill │ └─────────────────────────────────────────────────────────────────────┘ ↓ ┌─────────────────┐ │ Execute Skill │ └────────┬────────┘ ↓ ┌─────────────────┐ │ Handle Result │ └────────┬────────┘ ↓ ┌─────────────────────────────────────────────────────────────────────┐ │ Error Handling & Fallback │ │ │ │ Success? ────────► Return Result │ │ │ │ Fail? ────────┐ │ │ ↓ │ │ ┌──────────────────────────────────────────────────────────┐ │ │ │ Fallback Chain │ │ │ │ │ │ │ │ 1. Retry with adjusted parameters │ │ │ │ 2. Try Alternative Skill (if available) │ │ │ │ 3. Defer to Human Operator (if critical) │ │ │ │ 4. Log & Return Error │ │ │ └──────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────┘
Use this skill when:
Avoid this skill for:
Parse and Analyze Request - Extract intent, entities, and constraints from user input. Checkpoint: All required parameters must be present and in valid format before proceeding.
Score Available Skills - Calculate match scores using multi-factor algorithm:
Checkpoint: Skip to fallback if no skill scores above threshold.
Select Optimal Skill - Choose skill with highest score that meets minimum confidence. Checkpoint: Verify skill has not been disabled or deprecated.
Execute with Fallback - Run skill execution wrapped in retry and fallback logic. Checkpoint: Log all execution attempts for audit trail.
Return or Fallback - Either return successful result or apply fallback chain:
related-skillsCheckpoint: Record outcome with timing and confidence metadata.
def configure_brand_monitor(
brand_name: str,
platforms: List[str],
sentiment_threshold: float = 0.5,
lookback_days: int = 30
) -> Dict:
"""Configure Apify brand reputation monitoring parameters.
Selects optimal actor configuration based on brand scope and monitoring needs.
Applies multi-factor scoring: platform coverage, historical mention volume,
and sentiment volatility.
Args:
brand_name: Target brand or product name
platforms: List of platforms to monitor (e.g., ['twitter', 'reddit', 'news'])
sentiment_threshold: Minimum positive sentiment score to flag as 'healthy'
lookback_days: Historical window for baseline comparison
Returns:
Configured actor input dictionary ready for Apify API
"""
# Guard clause - validate brand and platforms (Law 1)
if not brand_name or not platforms:
raise ValueError("Brand name and at least one platform are required")
# Parse input - Make Illegal States Unrepresentable (Law 2)
normalized_platforms = [p.lower().strip() for p in platforms if p.strip()]
if not normalized_platforms:
raise ValueError("No valid platforms provided")
# Calculate monitoring scope score based on platform diversity and lookback
platform_coverage_score = len(normalized_platforms) / 5.0 # Max 5 common platforms
historical_weight = min(lookback_days / 90.0, 1.0)
# Atomic Predictability (Law 3) - Return new dict, don't mutate inputs
actor_config = {
"actorId": "apify/brand-monitor",
"input": {
"brandName": brand_name,
"platforms": normalized_platforms,
"startDate": _calculate_start_date(lookback_days),
"sentimentThreshold": sentiment_threshold,
"maxResults": 5000,
"proxyConfiguration": {"useApifyProxy": True}
},
"metadata": {
"scope_score": round(platform_coverage_score * historical_weight, 2),
"monitoring_window_days": lookback_days,
"timestamp": time.time()
}
}
return actor_config
def execute_brand_monitor(
actor_config: Dict,
apify_client,
fallback_source: Optional[str] = None
) -> Dict:
"""Execute Apify brand reputation monitoring with resilience patterns.
Implements Fail Fast, Fail Loud (Law 4):
- Invalid actor states halt immediately
- API rate limits trigger exponential backoff
- Dataset parsing failures trigger fallback to cached/recent data
Args:
actor_config: Pre-configured actor input from configure_brand_monitor
apify_client: Initialized ApifyClient instance
fallback_source: Optional path to cached reputation data
Returns:
Reputation analysis result with metrics, sentiment breakdown, and alerts
"""
# Guard clause - validate client and config (Early Exit)
if not apify_client or not actor_config.get("input"):
raise ValueError("Invalid Apify client or actor configuration")
run_id = None
try:
# Execute actor with timeout protection
run = apify_client.actor(actor_config["actorId"]).call(
input=actor_config["input"],
timeout_secs=1800
)
run_id = run["id"]
# Parse dataset results - Atomic Predictability (Law 3)
dataset = apify_client.dataset(run["defaultDatasetId"])
mentions = dataset.list_items()
if not mentions:
raise ValueError("No brand mentions retrieved from Apify dataset")
# Calculate reputation metrics
total_mentions = len(mentions)
positive = sum(1 for m in mentions if m.get("sentiment", 0) > 0.5)
negative = sum(1 for m in mentions if m.get("sentiment", 0) < -0.3)
reputation_score = (positive - negative) / max(total_mentions, 1)
# Success - Return structured result
return {
"success": True,
"run_id": run_id,
"brand": actor_config["input"]["brandName"],
"metrics": {
"total_mentions": total_mentions,
"positive_ratio": round(positive / max(total_mentions, 1), 3),
"negative_ratio": round(negative / max(total_mentions, 1), 3),
"reputation_score": round(reputation_score, 3)
},
"alerts": _generate_alerts(mentions, actor_config["input"]["sentimentThreshold"]),
"latency_ms": _calculate_latency()
}
except ApifyApiError as e:
# Fail Fast - Don't retry on auth/config errors (Law 4)
if e.status_code in [401, 403, 400]:
raise ValueError(f"Apify API rejected request: {e.message}") from e
# Transient error - fallback chain
if fallback_source:
return _load_cached_reputation(fallback_source)
raise
finally:
if run_id:
_cleanup_run(apify_client, run_id)
code-philosophy (5 Laws of Elegant Defense) in all logicWhen applying this skill, produce:
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.
| Skill | Purpose | |