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

# In your AI agent prompt:
context = load_knowledge_base("knowledge-base/mobile/")
prompt = f"""
Using this data: {context}
Generate 3 creative campaign ideas for a mid-range phone (2000-3000 RMB)
with Snapdragon 8 Gen 3 chip, focusing on Xiaohongshu platform.
Follow creative-output.md template.
"""

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
    }

# Usage
laptop_data = load_category_data('laptops')
# Feed to AI agent for competitive threat analysis

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"
    }

# Usage for smart home robot vacuum
romo_data = {
    'key_metric': '2小时清洁500㎡别墅 (2 hours for 500㎡ villa)',
    'comparison': '人工清洁需4小时+ (Manual: 4+ hours)'
}
campaign = generate_perception_reset_campaign(romo_data)

# scripts/preprocess.py
import pandas as pd
import json
import sys

def clean_review_data(input_file: str, output_dir: str):
    """
    Clean and validate review data before adding to knowledge base.
    
    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}")

# Usage
if __name__ == "__main__":
    clean_review_data(sys.argv[1], sys.argv[2])

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"]
}

# .env file
XIAOHONGSHU_API_KEY=your_key_here
BILIBILI_API_KEY=your_key_here
MARKET_DATA_SOURCE=https://your-data-api.com

import os
from dotenv import load_dotenv

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

import os
import json

# 1. Load knowledge base
def load_knowledge_base(category: str) -> str:
    path = f"knowledge-base/{category}/_index.md"
    with open(path, 'r', encoding='utf-8') as f:
        return f.read()

# 2. Load templates
def load_template(template_name: str) -> str:
    path = f"docs/templates/{template_name}.md"
    with open(path, 'r', encoding='utf-8') as f:
        return f.read()

# 3. Check used ideas for deduplication
def load_used_ideas() -> list:
    path = "docs/templates/used-ideas.md"
    with open(path, 'r', encoding='utf-8') as f:
        content = f.read()
        # Extract hooks from markdown list
        import re
        hooks = re.findall(r'- \*\*(.+?)\*\*', content)
        return hooks

# 4. Generate campaign
def generate_campaign(product_spec: dict, platform: str):
    kb_data = load_knowledge_base(product_spec['category'])
    template = load_template('creative-output')
    used_hooks = load_used_ideas()
    
    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

# 5. Usage
product = {
    'category': 'mobile',
    'name': 'SuperPhone X1',
    'chipset': 'Snapdragon 8 Gen 3',
    'price': 2499,
    'key_features': ['120Hz OLED', '5000mAh battery', '67W charging']
}

campaign_prompt = generate_campaign(product, 'Xiaohongshu')
# Send campaign_prompt to AI agent

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

# Usage
risk_prompt = assess_campaign_risk(campaign_output)

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

# Usage
warnings = validate_campaign_language("我们的产品遥遥领先")
# Output: [{'phrase': '遥遥领先', 'risk': 'high', 'explanation': 'Huawei meme - used ironically'}]

# Add data via preprocessing
python scripts/preprocess.py new_product_data.csv knowledge-base/

# Or manually create markdown file
# knowledge-base/mobile/new-brand.md

# Add validation step
def validate_sources(content: str) -> bool:
    """Check if every claim has a source."""
    import re
    # Numbers without sources
    numbers = re.findall(r'\d+(?:\.\d+)?[%㎡mAh元]', content)
    sources = re.findall(r'\[来源:.+?\]', content)
    
    if len(numbers) > len(sources):
        raise ValueError(f"Found {len(numbers)} claims but only {len(sources)} sources")
    return True

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

# Wrong: Comparing battery life from different test conditions
laptop_a_battery = "10 hours (PCMark 10 test)"
laptop_b_battery = "12 hours (video playback)"

# Fix: Flag as incomparable
def compare_metrics(metric_a: dict, metric_b: dict):
    if metric_a['test_method'] != metric_b['test_method']:
        return {
            'comparable': False,
            'reason': f"Different test methods: {metric_a['test_method']} vs {metric_b['test_method']}"
        }

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

# Triggered by: "处理新数据"
def activate_data_processor(raw_data: str):
    """
    Sub-agent for data cleaning and validation.
    See: docs/references/subagent-dataprocessor.md
    """
    steps = [
        "1. 纠错 - Fix encoding, formatting",
        "2. 判断类型 - Classify data type (review/spec/comment)",
        "3. 清洗 - Remove ads, extract facts",
        "4. 提取 - Structured data extraction",
        "5. 更新索引 - Update knowledge base index"
    ]
    # Execute preprocessing pipeline
    return process_pipeline(raw_data, steps)

# Triggered by: "帮我检查"
def activate_fact_checker(content: str):
    """
    Adversarial audit for generated content.
    See: docs/references/subagent-factchecker.md
    """
    checks = {
        "数据核验": verify_data_sources(content),
        "遗漏检测": check_missing_citations(content),
        "幻觉扫描": detect_fabrications(content),
        "逻辑一致性": validate_logic_chain(content)
    }
    return checks