git clone https://github.com/killsnake01/China-Marketing-Copilot-Skill.git
cd China-Marketing-Copilot-Skill

pip install pandas
python scripts/preprocess.py --input raw_data.csv --output knowledge-base/

docs/
  templates/          # Output formats (creative, insights, risk)
  references/         # KOL database, platform rules, personas
  ecosystem/          # Industry terminology, memes to avoid
knowledge-base/
  mobile/            # Phone brand matrix, pricing tiers
  headphones/        # Earphone reviews, clip-on comparison
  laptops/           # Notebook selection guide (笔吧)
  wearables/         # Smartwatch/band market data
  smart-home/        # Robot vacuum reviews, case studies
scripts/
  preprocess.py      # Data cleaning utility

undefined

**De-duplication:**
Check `docs/templates/used-ideas.md` before generation. Append new ideas to prevent repetition.

import os
import re

def load_category_data(category: str) -> dict:
    """Load brand matrix and evaluation data for a category."""
    base_path = f"knowledge-base/{category}/"
    index_file = os.path.join(base_path, "_index.md")
    
    with open(index_file, 'r', encoding='utf-8') as f:
        content = f.read()
    
    # Extract brand matrix table
    brands = re.findall(r'\| (.+?) \| (.+?) \| (.+?) \|', content)
    
    return {
        'category': category,
        'brands': brands,
        'raw_content': content
    }

**Output format:**
Uses `docs/templates/insight-output.md` with required sections:
- **Market Position** (price tier, feature positioning)
- **Competitive Threats** (direct overlap products)
- **Differentiation Gaps** (where competitor is weak)
- **Data Sources** (must cite: KOL name + platform + date)

PERSONAS = {
    "参数党": "Spec sheet warriors - will fact-check every number",
    "解构找茬": "Deconstruction trolls - expose marketing tricks",
    "真实体验派": "Real user experience advocates - hate hype",
    "品牌信仰": "Brand loyalists - defend their tribe",
    "吃瓜群众": "Casual observers - amplify drama"
}

def simulate_comment_section(campaign_text: str, personas: dict) -> list:
    """Generate predicted negative comments from each persona."""
    comments = []
    for persona_type, description in personas.items():
        prompt = f"As a {persona_type} ({description}), critique: {campaign_text}"
        # Feed to AI agent
        comments.append(generate_critique(prompt))
    return comments

def generate_perception_reset_campaign(product_data: dict) -> dict:
    """
    Find human limit benchmark → Product breaks it → Visualize data
    
    Example: DJI ROMO cleaning 500㎡ villa in 2 hours
    (Human baseline: 4+ hours)
    """
    return {
        "human_baseline": "Find existing perception (e.g., 'villa cleaning takes half a day')",
        "product_performance": product_data['key_metric'],
        "visualization": "Side-by-side time-lapse video",
        "data_source": "Must cite: test environment + conditions"
    }

**Method 5: 专业信任纪录片法 (Professional Trust Documentary)**
- Recruit real professionals (not actors)
- 6-month field test period
- Minimalist documentary style (no hype language)
- Case study: DJI agriculture drones with actual farmers

undefined

Validation rules:
- Numbers must have units
- Ratings must have scale (e.g., 5/5)
- Price must include currency and date
- Source must include: KOL name + platform + publish date
"""
df = pd.read_csv(input_file)

# Validation
required_columns = ['product_name', 'metric', 'value', 'unit', 'source', 'date']
for col in required_columns:
    if col not in df.columns:
        raise ValueError(f"Missing required column: {col}")

# Check source format
for idx, row in df.iterrows():
    source = row['source']
    if not ('|' in source and len(source.split('|')) >= 3):
        print(f"Warning: Invalid source format at row {idx}: {source}")
        print("Expected: KOL名|平台|日期 (e.g., 爱否科技|Bilibili|2025-03)")

# Export to knowledge base
category = df['category'].iloc[0]
output_path = f"{output_dir}/{category}/_data.json"

df.to_json(output_path, orient='records', force_ascii=False, indent=2)
print(f"Processed {len(df)} records to {output_path}")

**Run preprocessing:**

```bash
python scripts/preprocess.py earphone_reviews.csv knowledge-base/

{
  "brand": "your_brand_name",
  "category": "mobile",
  "target_price_range": "2000-3000",
  "competitors": ["Redmi", "Realme", "iQOO"],
  "platforms": ["Xiaohongshu", "Douyin", "Bilibili"],
  "risk_tolerance": "conservative",
  "kol_preferences": ["爱否科技", "ZEALER"]
}

undefined

Load in Python:

```python
import os
from dotenv import load_dotenv

load_dotenv()
api_key = os.getenv('XIAOHONGSHU_API_KEY')

import os
import json

prompt = f"""
Knowledge Base:
{kb_data}

Template:
{template}

Already Used Hooks (avoid these):
{used_hooks}

Product Specs:
{json.dumps(product_spec, ensure_ascii=False)}

Target Platform: {platform}

Generate 3 creative campaign ideas following the template.
Ensure:
1. Every number has a source citation
2. No AI clichés (no "值得注意的是", "首先其次")
3. Hooks are not in used_hooks list
4. Include risk assessment for each idea
"""

# Feed to your AI agent (Claude, GPT, etc.)
return prompt

undefined

def assess_campaign_risk(campaign_text: str) -> dict:
    """
    Run multi-layer risk check on campaign content.
    """
    # Load personas
    personas_content = load_template('comment-personas')
    
    # Load historical failures
    kb_smart_home = load_knowledge_base('smart-home')
    failures = extract_case_studies(kb_smart_home)
    
    prompt = f"""
    Campaign Text:
    {campaign_text}
    
    Comment Personas (simulate negative reactions):
    {personas_content}
    
    Historical Failures (check for similar patterns):
    {failures}
    
    Assess risks in these dimensions:
    1. Technical credibility (can 参数党 debunk it?)
    2. Perception tricks (will 解构找茬 expose methods?)
    3. Platform compliance (content guidelines)
    4. Similar historical failures
    
    Output in risk-assessment.md template format.
    """
    
    return prompt

undefined

def cross_category_comparison(product_a: dict, product_b: dict) -> str:
    """
    Compare products from different categories (e.g., phone vs tablet).
    Must ensure data sources are compatible.
    """
    kb_a = load_knowledge_base(product_a['category'])
    kb_b = load_knowledge_base(product_b['category'])
    
    # Verify data source compatibility
    source_a = extract_sources(kb_a, product_a['name'])
    source_b = extract_sources(kb_b, product_b['name'])
    
    if not sources_compatible(source_a, source_b):
        return "Error: Cannot compare - data sources not aligned (different test methodologies)"
    
    # Proceed with comparison...

def select_kols_for_category(category: str, budget: int) -> list:
    """
    Select KOLs based on category expertise and budget.
    """
    kol_db_path = "docs/ecosystem/kols.md"
    with open(kol_db_path, 'r', encoding='utf-8') as f:
        kol_data = f.read()
    
    # Extract KOLs for category
    import re
    kol_pattern = rf'\| (.+?) \| {category} \| (.+?) \| (.+?) \|'
    matches = re.findall(kol_pattern, kol_data)
    
    # matches = [(name, followers, estimated_cost), ...]
    affordable_kols = [k for k in matches if parse_cost(k[2]) <= budget]
    
    return affordable_kols

def validate_campaign_language(text: str) -> list:
    """
    Check campaign text against industry meme database to avoid backfires.
    """
    meme_db = load_template('industry-memes')
    warnings = []
    
    # Load risky phrases
    risky_phrases = extract_risky_phrases(meme_db)
    
    for phrase in risky_phrases:
        if phrase in text:
            context = get_meme_context(meme_db, phrase)
            warnings.append({
                'phrase': phrase,
                'risk': context['risk_level'],
                'explanation': context['why_risky']
            })
    
    return warnings

undefined

undefined

undefined

undefined

if len(numbers) > len(sources):
    raise ValueError(f"Found {len(numbers)} claims but only {len(sources)} sources")
return True

undefined

AI_CLICHES = [
    "值得注意的是", "首先其次最后", "我们可以发现",
    "不是A而是B", "在...背景下", "赋能", "生态"
]

def check_ai_language(text: str) -> list:
    """Detect AI clichés."""
    found = [phrase for phrase in AI_CLICHES if phrase in text]
    if found:
        print(f"Warning: AI clichés detected: {found}")
        print("Rewrite using colloquial language or data-driven statements")
    return found

undefined

undefined

def enforce_risk_check(campaign: str) -> str:
    """Mandatory risk assessment before finalizing campaign."""
    personas = load_template('comment-personas')
    
    if "[风险评估]" not in campaign:
        risk_section = generate_risk_assessment(campaign, personas)
        campaign += f"\n\n## 风险评估\n{risk_section}"
    
    return campaign

QUALITY_CHECKS = {
    "数据纪律": [
        "✓ Every number has source citation",
        "✓ No fabricated statistics",
        "✓ Speculation labeled as [推测]",
        "✓ Competitive data from same source"
    ],
    "去AI化": [
        "✓ No '值得注意的是' phrases",
        "✓ No '首先其次最后' structures",
        "✓ No generic corporate jargon"
    ],
    "风险覆盖": [
        "✓ 5 personas simulated",
        "✓ Historical case studies checked",
        "✓ Platform compliance verified"
    ]
}

def run_quality_check(content: str) -> dict:
    """Run full quality validation."""
    results = {}
    for category, checks in QUALITY_CHECKS.items():
        results[category] = []
        for check in checks:
            # Implement validation logic for each check
            pass
    return results

undefined

**FactChecker Sub-Agent:**

```python

undefined