30x Outreach

# Apollo People Search API
APOLLO_API_KEY=your_apollo_api_key_here

# LeadMagic Email Verification API
LEADMAGIC_API_KEY=your_leadmagic_api_key_here

# Instantly Cold Email Platform
INSTANTLY_API_KEY=your_instantly_api_key_here

# Email Sending (for cold-outbound-sender.py)
SENDER_EMAIL=you@yourdomain.com
SENDER_NAME=Your Name
SMTP_HOST=smtp.gmail.com
SMTP_PORT=587
SMTP_USER=you@yourdomain.com
SMTP_PASSWORD=your_app_password_here

# Competitive Monitor (optional)
# Path to a JSON file defining your competitors
# See scripts/competitive-monitor.py for the expected format
COMPETITORS_CONFIG=./competitors.json

# Cross-Signal Detector (optional)
DATA_DIR=./data/agent-outputs
OUTPUT_FILE=./data/cross-signals-latest.json

# Cross-Signal Detector: comma-separated words to exclude from company extraction
# SIGNAL_STOP_WORDS=YourCompany,InternalTool

{
  "business_domain": "yourdomain.com",

  "apis": {
    "apollo": {
      "key": "",
      "note": "Apollo.io People Search API — for lead sourcing"
    },
    "leadmagic": {
      "key": "",
      "note": "LeadMagic — for email verification"
    },
    "instantly": {
      "key": "",
      "note": "Instantly v2 API — for campaign management and sending"
    },
    "brave": {
      "key": "",
      "note": "Brave Search API — for competitive monitoring and Twitter/X scanning"
    }
  },

  "sending": {
    "daily_limit_per_account": 30,
    "warmup_min_score": 80,
    "warmup_min_days": 14,
    "batch_size": 25,
    "smtp": {
      "host": "smtp.gmail.com",
      "port": 587,
      "user": "",
      "password": "",
      "sender_email": "",
      "sender_name": ""
    }
  },

  "competitors": [
    {
      "key": "competitor-a",
      "name": "Competitor A",
      "domain": "competitor-a.com",
      "pricing_url": "https://competitor-a.com/pricing",
      "blog_url": "https://competitor-a.com/blog"
    }
  ],

  "lead_pipeline": {
    "default_volume": 500,
    "exclude_file": null
  },

  "cross_signals": {
    "scan_hours": 48,
    "stop_words": []
  }
}

{
  "company_name": "",
  "domain": "",
  "created_at": "",
  "last_updated": "",

  "positioning": {
    "one_liner": "",
    "value_prop": "",
    "differentiator": "",
    "category": ""
  },

  "product": {
    "what_you_sell": "",
    "primary_offer": "",
    "offer_type": "",
    "price_range": "",
    "pricing_model": "",
    "pricing_url": ""
  },

  "proof": {
    "best_result": "",
    "named_clients": [],
    "testimonials": [],
    "case_study_urls": [],
    "stats": []
  },

  "market": {
    "ideal_buyer_title": "",
    "ideal_buyer_company_type": "",
    "top_competitors": [],
    "top_objection": "",
    "industry_verticals": []
  },

  "brand_voice": {
    "tone": "",
    "style_notes": "",
    "words_to_use": [],
    "words_to_avoid": []
  },

  "sources": {
    "auto_scraped": {
      "homepage": {},
      "pricing_page": {},
      "about_page": {},
      "case_studies": {},
      "linkedin": {},
      "twitter": {}
    },
    "founder_interview": {},
    "conflicts": []
  }
}

{
  "icp_name": "",
  "created_at": "",
  "last_updated": "",
  "business_domain": "",

  "targeting": {
    "titles": {
      "primary": [],
      "secondary": [],
      "never_target": []
    },
    "industries": {
      "primary": [],
      "secondary": [],
      "excluded": []
    },
    "company_size": {
      "employee_min": null,
      "employee_max": null,
      "employee_sweet_spot": "",
      "revenue_min": "",
      "revenue_max": "",
      "funding_stage": ""
    },
    "geography": {
      "primary_markets": [],
      "excluded_regions": []
    }
  },

  "buying_signals": [
    {
      "signal": "",
      "source": "",
      "weight": "high"
    }
  ],

  "anti_icp": {
    "company_traits": [],
    "contact_traits": []
  },

  "offer_fit": {
    "primary_offer": "",
    "why_this_offer": ""
  },

  "objections": [
    {
      "objection": "",
      "neutralizer": "",
      "use_in_step": null
    }
  ],

  "personalization": {
    "data_sources": [],
    "available_fields": [],
    "personalization_strategy": ""
  },

  "technographics": {
    "required_tools": [],
    "competitor_tools": [],
    "complementary_tools": [],
    "data_stack_maturity": ""
  },

  "intent_signals": {
    "bombora_topics": [],
    "g2_categories": [],
    "data_sources": []
  },

  "apollo_search_params": {
    "person_titles": [],
    "q_organization_keyword_tags": [],
    "organization_num_employees_ranges": [],
    "person_locations": [],
    "q_keywords": ""
  },

  "enrichment_fields": {
    "funding_date": true,
    "tech_stack_crm": true,
    "days_in_role": true,
    "note": "These 3 fields have highest conversion lift for trigger-event targeting"
  }
}

requests>=2.28.0

#!/usr/bin/env python3
"""
Cold Outbound Sender — 审核通过后发送邮件

[INPUT]: 依赖 config_loader.py 的 load_config/get_sending_config/write_output
[OUTPUT]: 对外提供邮件发送 + 历史记录
[POS]: scripts/ 的发送脚本，被 SKILL.md Step 8 调用（需人工审批后才运行）
[PROTOCOL]: 变更时更新此头部，然后检查 CLAUDE.md

Usage:
    python3 cold-outbound-sender.py [--dry-run] [--max N]
    python3 cold-outbound-sender.py --approved-file path/to/approved.json

SMTP config: config.json > environment variables
"""

import argparse
import json
import os
import smtplib
import subprocess
import sys
from datetime import datetime
from email.mime.text import MIMEText
from pathlib import Path

try:
    from config_loader import load_config, get_sending_config, write_output
except ImportError:
    def load_config(): return {}
    def get_sending_config(c): return {}
    def write_output(m, d, **kw): pass


DEFAULT_MAX_PER_DAY = 10
DEFAULT_APPROVED_FILE = "./data/cold-outbound-approved.json"
DEFAULT_HISTORY_FILE = "./data/cold-outbound-history.json"


def validate_outbound(text):
    """Basic validation for outbound content. Returns (ok, text)."""
    if not text or not isinstance(text, str):
        return False, text
    # Check for common leaked credential patterns
    suspicious_patterns = [
        r'sk-[a-zA-Z0-9]{20,}',       # API keys
        r'Bearer [a-zA-Z0-9\-_.]+',    # Auth headers
        r'/Users/[a-zA-Z]+/',          # Local paths
        r'password\s*[:=]\s*\S+',      # Password patterns
    ]
    import re
    for pattern in suspicious_patterns:
        if re.search(pattern, text, re.IGNORECASE):
            return False, text
    return True, text


def load_history(history_path):
    if os.path.exists(history_path):
        try:
            with open(history_path) as f:
                return json.load(f)
        except Exception:
            pass
    return []


def save_history(history, history_path):
    os.makedirs(os.path.dirname(history_path), exist_ok=True)
    with open(history_path, 'w') as f:
        json.dump(history, f, indent=2)


def count_sent_today(history):
    today = datetime.now().strftime("%Y-%m-%d")
    return sum(1 for h in history if h.get("sent_date", "").startswith(today))


def send_email_smtp(to, subject, body, sender_email, sender_name,
                    smtp_host, smtp_port, smtp_user, smtp_password, dry_run=False):
    """Send via SMTP."""
    ok_subj, subject = validate_outbound(subject)
    ok_body, body = validate_outbound(body)
    if not ok_subj or not ok_body:
        print(f"  🛡️  Email to {to} BLOCKED by validation (suspicious content detected)")
        return False

    if dry_run:
        print(f"  [DRY RUN] Would send to {to}: {subject}")
        return True

    try:
        msg = MIMEText(body, 'plain')
        msg['Subject'] = subject
        msg['From'] = f"{sender_name} <{sender_email}>"
        msg['To'] = to

        with smtplib.SMTP(smtp_host, int(smtp_port)) as server:
            server.starttls()
            server.login(smtp_user, smtp_password)
            server.sendmail(sender_email, [to], msg.as_string())

        print(f"  ✅ Sent to {to}: {subject}")
        return True
    except Exception as e:
        print(f"  ❌ Error sending to {to}: {e}", file=sys.stderr)
        return False


def send_email_cli(to, subject, body, sender_email, sender_name, cli_command, dry_run=False):
    """Send via a CLI tool (e.g., gog, msmtp, mailx)."""
    ok_subj, subject = validate_outbound(subject)
    ok_body, body = validate_outbound(body)
    if not ok_subj or not ok_body:
        print(f"  🛡️  Email to {to} BLOCKED by validation (suspicious content detected)")
        return False

    if dry_run:
        print(f"  [DRY RUN] Would send to {to}: {subject}")
        return True

    try:
        # Default CLI pattern: gog gmail send
        cmd = cli_command.split() + [
            "--to", to,
            "--subject", subject,
            "--body", body,
            "--from", f"{sender_name} <{sender_email}>",
        ]
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=30)
        if result.returncode == 0:
            print(f"  ✅ Sent to {to}: {subject}")
            return True
        else:
            print(f"  ❌ Failed to send to {to}: {result.stderr}", file=sys.stderr)
            return False
    except Exception as e:
        print(f"  ❌ Error sending to {to}: {e}", file=sys.stderr)
        return False


def main():
    parser = argparse.ArgumentParser(description="Cold Outbound Sender")
    parser.add_argument("--dry-run", action="store_true", help="Don't actually send emails")
    parser.add_argument("--max", type=int, default=DEFAULT_MAX_PER_DAY,
                        help=f"Max emails per day (default: {DEFAULT_MAX_PER_DAY})")
    parser.add_argument("--approved-file", default=DEFAULT_APPROVED_FILE,
                        help="Path to approved prospects JSON file")
    parser.add_argument("--history-file", default=DEFAULT_HISTORY_FILE,
                        help="Path to send history JSON file")
    parser.add_argument("--send-method", choices=["smtp", "cli"], default="smtp",
                        help="Send method: smtp or cli (default: smtp)")
    parser.add_argument("--cli-command", default="gog gmail send",
                        help="CLI command for sending (used with --send-method cli)")
    args = parser.parse_args()

    # ── 加载配置 (config.json > env) ──
    config = load_config()
    smtp_cfg = get_sending_config(config).get("smtp", {})

    sender_email = smtp_cfg.get("sender_email") or os.environ.get("SENDER_EMAIL", "")
    sender_name = smtp_cfg.get("sender_name") or os.environ.get("SENDER_NAME", "")
    smtp_host = smtp_cfg.get("host") or os.environ.get("SMTP_HOST", "smtp.gmail.com")
    smtp_port = smtp_cfg.get("port") or os.environ.get("SMTP_PORT", "587")
    smtp_user = smtp_cfg.get("user") or os.environ.get("SMTP_USER", sender_email)
    smtp_password = smtp_cfg.get("password") or os.environ.get("SMTP_PASSWORD", "")

    if not os.path.exists(args.approved_file):
        print(f"No approved prospects file found at {args.approved_file}")
        sys.exit(0)

    with open(args.approved_file) as f:
        approved = json.load(f)

    history = load_history(args.history_file)
    sent_today = count_sent_today(history)
    remaining = args.max - sent_today

    if remaining <= 0:
        print(f"Already sent {sent_today} emails today (max {args.max}). Stopping.")
        sys.exit(0)

    sent_count = 0
    for prospect in approved:
        if sent_count >= remaining:
            break

        email = prospect.get("email")
        if not email or email == "Unknown":
            continue

        # Check if already sent
        if any(h.get("email") == email for h in history):
            print(f"  SKIP {email}: already in history")
            continue

        angle_key = prospect.get("approved_angle", "A")
        drafts = prospect.get("angle_drafts", {})
        draft = drafts.get(angle_key, {})

        subject = draft.get("subject", f"Quick question for {prospect.get('company', 'you')}")
        body = draft.get("body", "")

        if not body:
            print(f"  SKIP {email}: no draft body for angle {angle_key}")
            continue

        if args.send_method == "smtp":
            if not smtp_password and not args.dry_run:
                print("ERROR: SMTP_PASSWORD env var required for smtp sending.")
                sys.exit(1)
            success = send_email_smtp(
                email, subject, body, sender_email, sender_name,
                smtp_host, smtp_port, smtp_user, smtp_password, args.dry_run
            )
        else:
            success = send_email_cli(
                email, subject, body, sender_email, sender_name,
                args.cli_command, args.dry_run
            )

        if success:
            history.append({
                "company": prospect.get("company", ""),
                "contact_name": prospect.get("contact_name", ""),
                "email": email,
                "angle": angle_key,
                "subject": subject,
                "sent_date": datetime.now().isoformat(),
                "score": prospect.get("score", 0),
            })
            sent_count += 1

    if not args.dry_run:
        save_history(history, args.history_file)

    # Remove sent prospects from approved file
    if not args.dry_run and sent_count > 0:
        sent_emails = {h["email"] for h in history}
        remaining_approved = [p for p in approved if p.get("email") not in sent_emails]
        with open(args.approved_file, 'w') as f:
            json.dump(remaining_approved, f, indent=2)

    print(f"\nSent {sent_count} emails ({'dry run' if args.dry_run else 'live'}). Total today: {sent_today + sent_count}")

    # ── 统一格式输出 (30x) ──
    write_output(
        module="cold-outbound-sender",
        summary=f"Sent {sent_count} emails ({'dry run' if args.dry_run else 'live'})",
        data={
            "sent": sent_count,
            "total_today": sent_today + sent_count,
            "dry_run": args.dry_run,
            "max_per_day": args.max,
        },
    )


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Competitive Monitor — 竞情追踪：定价变动、博客动态、招聘信号

[INPUT]: 依赖 config_loader.py 的 load_config/get_competitors/get_api_key/write_output
[OUTPUT]: 对外提供竞情报告，输出到 data/intel/competitive/
[POS]: scripts/ 的情报脚本，被 SKILL.md Step 5 调用
[PROTOCOL]: 变更时更新此头部，然后检查 CLAUDE.md

Usage:
    python3 competitive-monitor.py               # 从 config.json 读竞对列表
    python3 competitive-monitor.py --company acme # 只扫描单个竞对
    python3 competitive-monitor.py --output report.md

Competitor config: config.json > --config flag > COMPETITORS_CONFIG env var
"""

import argparse
import json
import os
import re
import sys
import urllib.request
import urllib.parse
from datetime import datetime, timedelta
from difflib import unified_diff
from typing import Dict, List, Optional
from html.parser import HTMLParser
from urllib.error import URLError, HTTPError


try:
    from config_loader import load_config, get_competitors, get_api_key, write_output, SKILL_ROOT
except ImportError:
    SKILL_ROOT = Path(__file__).resolve().parent.parent if 'Path' in dir() else None
    def load_config(): return {}
    def get_competitors(c): return []
    def get_api_key(c, s): return ""
    def write_output(m, d, **kw): pass


def validate_text(text, max_length=500000):
    """Basic input validation for scraped content."""
    if not text or not isinstance(text, str):
        return text
    # Truncate extremely long content
    if len(text) > max_length:
        text = text[:max_length]
    return text


class BlogExtractor(HTMLParser):
    """Extract blog post titles and dates from HTML."""

    def __init__(self):
        super().__init__()
        self.posts = []
        self.current_title = None
        self.current_date = None
        self.in_title = False
        self.in_date = False
        self.title_tags = ['h1', 'h2', 'h3', 'h4']

    def handle_starttag(self, tag, attrs):
        if tag.lower() in self.title_tags:
            self.in_title = True
        for name, value in attrs:
            if name in ['class', 'id'] and any(
                date_word in value.lower() for date_word in ['date', 'time', 'published']
            ):
                self.in_date = True

    def handle_endtag(self, tag):
        if tag.lower() in self.title_tags:
            self.in_title = False
        self.in_date = False

    def handle_data(self, data):
        if self.in_title and data.strip():
            self.current_title = data.strip()

        if self.in_date and data.strip():
            date_match = re.search(
                r'\b\d{1,2}[/-]\d{1,2}[/-]\d{2,4}\b|\b\w+ \d{1,2},? \d{4}\b', data
            )
            if date_match:
                self.current_date = date_match.group()

        if self.current_title and self.current_date:
            self.posts.append({
                'title': self.current_title,
                'date': self.current_date,
            })
            self.current_title = None
            self.current_date = None


class CompetitiveMonitor:
    """Main competitive monitoring class."""

    # Example competitors for demo. Override with --config or COMPETITORS_CONFIG.
    EXAMPLE_COMPETITORS = {
        'competitor_a': {
            'name': 'Competitor A',
            'domain': 'competitor-a.com',
            'pricing_url': 'https://www.competitor-a.com/pricing',
            'blog_url': 'https://www.competitor-a.com/blog',
            'linkedin_query': 'Competitor A site:linkedin.com',
            'jobs_query': 'Competitor A careers OR jobs',
        },
        'competitor_b': {
            'name': 'Competitor B',
            'domain': 'competitor-b.com',
            'pricing_url': 'https://www.competitor-b.com/pricing',
            'blog_url': 'https://www.competitor-b.com/blog',
            'linkedin_query': 'Competitor B site:linkedin.com',
            'jobs_query': 'Competitor B careers OR jobs',
        },
    }

    def __init__(self, data_dir: str = None, competitors: dict = None):
        self.data_dir = data_dir or os.path.join(os.getcwd(), 'data', 'competitive')
        self.pricing_dir = os.path.join(self.data_dir, 'pricing-snapshots')
        self.history_dir = os.path.join(self.data_dir, 'scan-history')
        self.competitors = competitors or self.EXAMPLE_COMPETITORS

        os.makedirs(self.pricing_dir, exist_ok=True)
        os.makedirs(self.history_dir, exist_ok=True)

    def fetch_url(self, url: str, timeout: int = 10) -> Optional[str]:
        """Fetch URL content with error handling."""
        try:
            headers = {
                'User-Agent': (
                    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) '
                    'AppleWebKit/537.36 (KHTML, like Gecko) '
                    'Chrome/91.0.4472.124 Safari/537.36'
                )
            }
            request = urllib.request.Request(url, headers=headers)

            with urllib.request.urlopen(request, timeout=timeout) as response:
                content = response.read().decode('utf-8', errors='ignore')
                content = validate_text(content)
                return content

        except (URLError, HTTPError, UnicodeDecodeError) as e:
            print(f"❌ Error fetching {url}: {e}")
            return None

    def extract_blog_posts(self, html: str) -> List[Dict]:
        """Extract blog posts from HTML."""
        if not html:
            return []

        extractor = BlogExtractor()
        try:
            extractor.feed(html)
            return extractor.posts
        except Exception as e:
            print(f"Error extracting blog posts: {e}")
            return []

    def is_recent_post(self, date_str: str, days_back: int = 7) -> bool:
        """Check if post is from last N days."""
        if not date_str:
            return False

        formats = [
            '%m/%d/%Y', '%m-%d-%Y', '%Y-%m-%d',
            '%B %d, %Y', '%b %d, %Y', '%B %d %Y', '%b %d %Y',
        ]

        for fmt in formats:
            try:
                post_date = datetime.strptime(date_str, fmt)
                cutoff_date = datetime.now() - timedelta(days=days_back)
                return post_date >= cutoff_date
            except ValueError:
                continue

        return False

    def get_pricing_diff(self, company_key: str, current_content: str) -> Optional[str]:
        """Compare current pricing with previous snapshot."""
        today = datetime.now().strftime('%Y-%m-%d')
        pricing_file = os.path.join(self.pricing_dir, f'{company_key}-{today}.txt')

        with open(pricing_file, 'w', encoding='utf-8') as f:
            f.write(current_content)

        previous_files = [
            f for f in os.listdir(self.pricing_dir)
            if f.startswith(f'{company_key}-') and f != f'{company_key}-{today}.txt'
        ]

        if not previous_files:
            return "🆕 First pricing snapshot saved"

        previous_files.sort(reverse=True)
        previous_file = os.path.join(self.pricing_dir, previous_files[0])

        try:
            with open(previous_file, 'r', encoding='utf-8') as f:
                previous_content = f.read()

            if current_content.strip() == previous_content.strip():
                return None

            current_lines = current_content.splitlines()
            previous_lines = previous_content.splitlines()

            diff = list(unified_diff(
                previous_lines, current_lines,
                fromfile='previous', tofile='current', n=0
            ))

            changes = len([
                line for line in diff
                if line.startswith(('+', '-')) and not line.startswith(('+++', '---'))
            ])

            return f"🔍 {changes} lines changed since last snapshot"

        except Exception as e:
            return f"❌ Error comparing snapshots: {e}"

    def scan_competitor(self, company_key: str) -> Dict:
        """Scan single competitor."""
        company = self.competitors[company_key]
        print(f"\n🔍 Scanning {company['name']}...")

        results = {
            'company': company['name'],
            'domain': company['domain'],
            'scan_time': datetime.now().isoformat(),
            'pricing': {},
            'blog': {},
            'search_queries': {
                'linkedin': company.get('linkedin_query', ''),
                'jobs': company.get('jobs_query', ''),
            },
        }

        # Fetch pricing page
        pricing_url = company.get('pricing_url')
        if pricing_url:
            print(f"  📄 Fetching pricing: {pricing_url}")
            pricing_content = self.fetch_url(pricing_url)

            if pricing_content:
                clean_content = re.sub(r'<[^>]+>', '', pricing_content)
                clean_content = re.sub(r'\s+', ' ', clean_content).strip()

                pricing_diff = self.get_pricing_diff(company_key, clean_content)

                results['pricing'] = {
                    'url': pricing_url,
                    'fetched': True,
                    'content_length': len(clean_content),
                    'diff': pricing_diff,
                }
            else:
                results['pricing'] = {
                    'url': pricing_url,
                    'fetched': False,
                    'error': 'Failed to fetch pricing page',
                }

        # Fetch blog page
        blog_url = company.get('blog_url')
        if blog_url:
            print(f"  📝 Fetching blog: {blog_url}")
            blog_content = self.fetch_url(blog_url)

            recent_posts = []
            if blog_content:
                all_posts = self.extract_blog_posts(blog_content)
                recent_posts = [post for post in all_posts if self.is_recent_post(post['date'])]

            results['blog'] = {
                'url': blog_url,
                'fetched': bool(blog_content),
                'total_posts_found': len(self.extract_blog_posts(blog_content)) if blog_content else 0,
                'recent_posts': recent_posts,
            }

        return results

    def generate_report(self, scan_results: List[Dict], threat_keywords: List[str] = None) -> str:
        """Generate markdown report."""
        today = datetime.now().strftime('%Y-%m-%d')

        # Configurable threat keywords (topics that signal competitive overlap)
        if threat_keywords is None:
            threat_keywords = ['funnel', 'conversion', 'landing page', 'ab test', 'optimize', 'cro']

        report = f"""# 🔍 Competitive Intelligence Report - {today}

## Executive Summary

Monitored {len(scan_results)} competitors for pricing changes, recent blog activity, and market signals.

"""

        threats = []
        interesting = []
        opportunities = []
        search_queries = []

        for result in scan_results:
            company = result['company']

            pricing = result.get('pricing', {})
            if pricing.get('diff') and '🔍' in str(pricing['diff']):
                interesting.append(
                    f"**{company}**: {pricing['diff']} → *Monitor for pricing strategy shifts*"
                )
            elif pricing.get('diff') and '🆕' in str(pricing['diff']):
                interesting.append(
                    f"**{company}**: {pricing['diff']} → *Baseline established for future tracking*"
                )

            blog = result.get('blog', {})
            recent_posts = blog.get('recent_posts', [])

            if recent_posts:
                post_titles = [
                    post['title'][:80] + '...' if len(post['title']) > 80 else post['title']
                    for post in recent_posts[:3]
                ]
                content_lower = ' '.join(post_titles).lower()

                if any(keyword in content_lower for keyword in threat_keywords):
                    threats.append(
                        f"**{company}**: {len(recent_posts)} recent posts, potential feature overlap → *Review competitive positioning*"
                    )
                else:
                    interesting.append(
                        f"**{company}**: {len(recent_posts)} recent posts → *{', '.join(post_titles[:2])}*"
                    )
            else:
                opportunities.append(
                    f"**{company}**: No recent blog content → *Content marketing gap you can exploit*"
                )

            sq = result.get('search_queries', {})
            if sq.get('linkedin'):
                search_queries.append(f"LinkedIn search: {sq['linkedin']}")
            if sq.get('jobs'):
                search_queries.append(f"Jobs search: {sq['jobs']}")

        if threats:
            report += "## 🔴 THREATS\n\n"
            for threat in threats:
                report += f"- {threat}\n"
            report += "\n"

        if interesting:
            report += "## 🟡 INTERESTING\n\n"
            for item in interesting:
                report += f"- {item}\n"
            report += "\n"

        if opportunities:
            report += "## 🟢 OPPORTUNITIES\n\n"
            for opp in opportunities:
                report += f"- {opp}\n"
            report += "\n"

        if search_queries:
            report += "## 🔎 LinkedIn/Jobs Search Queries\n\n"
            report += "Run these queries for social/hiring signals:\n\n"
            for query in search_queries:
                report += f"- `{query}`\n"
            report += "\n"

        report += "## 📊 Technical Summary\n\n"
        for result in scan_results:
            company = result['company']
            pricing = result.get('pricing', {})
            blog = result.get('blog', {})

            report += f"**{company}:**\n"
            report += f"- Pricing: {'✅' if pricing.get('fetched') else '❌'} {pricing.get('diff', 'No changes')}\n"
            report += f"- Blog: {'✅' if blog.get('fetched') else '❌'} {len(blog.get('recent_posts', []))} recent posts\n\n"

        return report

    def save_results(self, scan_results: List[Dict]) -> str:
        """Save scan results to files."""
        today = datetime.now().strftime('%Y-%m-%d')

        latest_file = os.path.join(self.data_dir, 'latest-scan.json')
        with open(latest_file, 'w') as f:
            json.dump(scan_results, f, indent=2)

        history_file = os.path.join(self.history_dir, f'{today}.json')
        with open(history_file, 'w') as f:
            json.dump(scan_results, f, indent=2)

        return latest_file

    def run(self, company_filter: Optional[str] = None) -> str:
        """Run competitive monitoring scan."""
        print("🚀 Starting competitive monitoring scan...")

        companies_to_scan = (
            [company_filter] if company_filter else list(self.competitors.keys())
        )

        if company_filter and company_filter not in self.competitors:
            print(f"❌ Unknown company: {company_filter}")
            print(f"Available companies: {', '.join(self.competitors.keys())}")
            return ""

        scan_results = []
        for company_key in companies_to_scan:
            try:
                result = self.scan_competitor(company_key)
                scan_results.append(result)
            except Exception as e:
                print(f"❌ Error scanning {company_key}: {e}")

        self.save_results(scan_results)
        report = self.generate_report(scan_results)

        print(f"\n✅ Scan complete! Results for {len(scan_results)} companies.")
        return report


def load_competitors_config(config_path: str) -> dict:
    """Load competitors from a JSON config file.

    Expected format:
    {
        "competitor_key": {
            "name": "Competitor Name",
            "domain": "competitor.com",
            "pricing_url": "https://competitor.com/pricing",
            "blog_url": "https://competitor.com/blog",
            "linkedin_query": "Competitor Name site:linkedin.com",
            "jobs_query": "Competitor Name careers OR jobs"
        }
    }
    """
    with open(config_path, 'r') as f:
        return json.load(f)


def main():
    parser = argparse.ArgumentParser(description='Competitive Monitoring Scraper')
    parser.add_argument('--company', help='Scan specific company only (by key)')
    parser.add_argument('--output', '-o', help='Save report to file')
    parser.add_argument('--config', help='Path to competitors JSON config file')
    parser.add_argument('--data-dir', help='Directory for storing scan data')
    parser.add_argument('--threat-keywords', nargs='*',
                        help='Keywords that signal competitive overlap (space-separated)')

    args = parser.parse_args()

    # ── 加载竞对配置 (config.json > --config > env > 内置示例) ──
    config = load_config()
    competitors = None

    # 优先从 config.json 读取
    config_competitors = get_competitors(config)
    if config_competitors:
        competitors = {}
        for comp in config_competitors:
            key = comp.get("key", comp.get("domain", "").replace(".", "-"))
            competitors[key] = {
                "name": comp.get("name", key),
                "domain": comp.get("domain", ""),
                "pricing_url": comp.get("pricing_url", f"https://{comp.get('domain', '')}/pricing"),
                "blog_url": comp.get("blog_url", f"https://{comp.get('domain', '')}/blog"),
                "linkedin_query": f"{comp.get('name', key)} site:linkedin.com",
                "jobs_query": f"{comp.get('name', key)} careers OR jobs",
            }
        print(f"📋 Loaded {len(competitors)} competitors from config.json")

    # 其次从 --config 或 env 读取
    if not competitors:
        config_path = args.config or os.environ.get('COMPETITORS_CONFIG')
        if config_path:
            try:
                competitors = load_competitors_config(config_path)
                print(f"📋 Loaded {len(competitors)} competitors from {config_path}")
            except Exception as e:
                print(f"❌ Error loading config: {e}")
                sys.exit(1)

    # 设置数据目录到 30x 标准路径
    data_dir = args.data_dir
    if not data_dir:
        try:
            data_dir = str(SKILL_ROOT / "data" / "intel" / "competitive")
        except:
            data_dir = None

    monitor = CompetitiveMonitor(
        data_dir=data_dir,
        competitors=competitors,
    )

    report = monitor.run(args.company)

    if report:
        print("\n" + "=" * 60)
        print(report)
        print("=" * 60)

        if args.output:
            with open(args.output, 'w') as f:
                f.write(report)
            print(f"\n📁 Report saved to: {args.output}")

        # ── 统一格式输出 (30x) ──
        write_output(
            module="competitive-monitor",
            summary=f"Scanned {len(monitor.competitors)} competitors",
            data={"report": report[:2000], "competitors_scanned": list(monitor.competitors.keys())},
        )


if __name__ == '__main__':
    main()

#!/usr/bin/env python3
"""
config_loader.py — 统一配置加载器

[INPUT]: 依赖 config.json 或 config.example.json
[OUTPUT]: 对外提供 load_config(), get_api_key(), get_sending_config(), get_competitors(), write_output()
[POS]: scripts/ 的基础设施，被所有其他脚本依赖
[PROTOCOL]: 变更时更新此头部，然后检查 CLAUDE.md
"""

import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path


# ─────────────────────────────────────────────
# 配置文件路径解析
# ─────────────────────────────────────────────

def _find_skill_root():
    """向上查找包含 SKILL.md 的目录作为 skill root"""
    current = Path(__file__).resolve().parent
    for _ in range(5):
        if (current / "SKILL.md").exists():
            return current
        current = current.parent
    return Path(__file__).resolve().parent.parent


SKILL_ROOT = _find_skill_root()


def load_config():
    """加载 config.json，不存在则尝试 config.example.json"""
    config_path = SKILL_ROOT / "config.json"
    example_path = SKILL_ROOT / "config.example.json"

    if config_path.exists():
        with open(config_path) as f:
            return json.load(f)

    if example_path.exists():
        print(f"  ⚠️  No config.json found. Using config.example.json (API keys will be empty)")
        with open(example_path) as f:
            return json.load(f)

    print("  ❌ No config.json or config.example.json found.")
    sys.exit(1)


def get_api_key(config, service):
    """从 config 或环境变量获取 API key"""
    env_map = {
        "apollo": "APOLLO_API_KEY",
        "leadmagic": "LEADMAGIC_API_KEY",
        "instantly": "INSTANTLY_API_KEY",
        "brave": "BRAVE_API_KEY",
    }

    env_var = env_map.get(service, "")
    env_val = os.environ.get(env_var, "")
    if env_val:
        return env_val

    return config.get("apis", {}).get(service, {}).get("key", "")


def get_sending_config(config):
    """获取发送配置"""
    return config.get("sending", {})


def get_competitors(config):
    """获取竞对列表"""
    return config.get("competitors", [])


# ─────────────────────────────────────────────
# 统一输出格式
# ─────────────────────────────────────────────

def write_output(module, data, summary="", icp=None):
    """写入统一格式的 JSON 输出到 data/output/

    统一 schema:
    {
        "module": "instantly-audit",
        "timestamp": "2026-04-02T10:30:00Z",
        "icp": "saas-vp-marketing",
        "status": "success",
        "data": { ... },
        "summary": "一句话总结"
    }
    """
    output_dir = SKILL_ROOT / "data" / "output"
    output_dir.mkdir(parents=True, exist_ok=True)

    today = datetime.now().strftime("%Y-%m-%d")
    output_path = output_dir / f"{today}-{module}.json"

    envelope = {
        "module": module,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "icp": icp,
        "status": "success",
        "data": data,
        "summary": summary,
    }

    with open(output_path, "w") as f:
        json.dump(envelope, f, indent=2, default=str)

    print(f"  💾 Output: {output_path}")
    return output_path


def load_icp_profile(icp_name):
    """加载指定 ICP 的 profile.json"""
    icp_path = SKILL_ROOT / "data" / "icps" / icp_name / "profile.json"
    if not icp_path.exists():
        print(f"  ❌ ICP not found: {icp_path}")
        print(f"     Run /30x-outreach {icp_name} to create it first.")
        return None

    with open(icp_path) as f:
        return json.load(f)


def load_business_profile():
    """加载业务 profile.json"""
    biz_path = SKILL_ROOT / "data" / "business" / "profile.json"
    if not biz_path.exists():
        return None

    with open(biz_path) as f:
        return json.load(f)

#!/usr/bin/env python3
"""
Cross-Signal Detector — 跨源信号交叉检测

[INPUT]: 依赖 config_loader.py，扫描 data/output/ 下所有模块输出
[OUTPUT]: 对外提供跨信号检测报告，输出到 data/intel/signals/
[POS]: scripts/ 的情报脚本，被 SKILL.md Step 5 调用
[PROTOCOL]: 变更时更新此头部，然后检查 CLAUDE.md

Usage:
    python3 cross-signal-detector.py                      # 扫描 data/output/
    python3 cross-signal-detector.py --data-dir ./custom  # 自定义目录
    python3 cross-signal-detector.py --hours 48           # 回溯时间窗口
"""

import argparse
import json
import os
import re
import glob
from datetime import datetime, timedelta, timezone
from collections import defaultdict
from pathlib import Path

try:
    from config_loader import load_config, write_output, SKILL_ROOT
except ImportError:
    SKILL_ROOT = Path(__file__).resolve().parent.parent
    def load_config(): return {}
    def write_output(m, d, **kw): pass


# Words to exclude from company name extraction (common English words that look like names)
STOP_WORDS = {
    'The', 'This', 'That', 'What', 'How', 'Why', 'When', 'Where',
    'For', 'From', 'With', 'About', 'Into', 'Over', 'After',
    'Before', 'Between', 'Under', 'During', 'Through',
    'Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday',
    'Saturday', 'Sunday', 'January', 'February', 'March',
    'April', 'May', 'June', 'July', 'August', 'September',
    'October', 'November', 'December',
    'None', 'True', 'False', 'Error', 'Warning',
}

# Configurable: add your own team names / internal terms to exclude
CUSTOM_STOP_WORDS = set(os.environ.get('SIGNAL_STOP_WORDS', '').split(',')) if os.environ.get('SIGNAL_STOP_WORDS') else set()


def get_recent_files(directory, hours=24):
    """Get files modified in the last N hours."""
    cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
    recent = []
    if not os.path.isdir(directory):
        return recent
    for f in glob.glob(os.path.join(directory, "*")):
        if os.path.isfile(f):
            mtime = datetime.fromtimestamp(os.path.getmtime(f), tz=timezone.utc)
            if mtime > cutoff:
                recent.append(f)
    return recent


def extract_companies(text):
    """Extract company names (capitalized words, common patterns)."""
    companies = set()
    all_stop = STOP_WORDS | CUSTOM_STOP_WORDS
    for match in re.findall(
        r'\b([A-Z][a-zA-Z]+(?:\.[a-zA-Z]+)?(?:\s+(?:AI|Inc|Corp|Labs|Tech|io))?)\b',
        text
    ):
        if len(match) > 2 and match not in all_stop:
            companies.add(match)
    return companies


def extract_keywords(text):
    """Extract keyword themes from marketing/business text."""
    keywords = set()
    patterns = [
        r'(?:ai|artificial intelligence)\s+(?:marketing|agent|tool|saas|automation)',
        r'(?:seo|content|digital)\s+(?:marketing|strategy|optimization|growth)',
        r'(?:b2b|saas|enterprise)\s+(?:marketing|growth|sales)',
        r'(?:social media|linkedin|twitter|youtube)\s+(?:marketing|growth|strategy)',
        r'(?:email|outbound|cold)\s+(?:marketing|outreach|campaign)',
        r'(?:paid|ppc|google)\s+(?:ads|advertising|media)',
    ]
    text_lower = text.lower()
    for p in patterns:
        match = re.search(p, text_lower)
        if match:
            keywords.add(match.group())
    return keywords


def extract_verticals(text):
    """Extract industry verticals."""
    verticals = set()
    vertical_keywords = {
        'fintech': ['fintech', 'financial', 'banking', 'payments'],
        'healthtech': ['healthtech', 'health tech', 'healthcare', 'medical'],
        'edtech': ['edtech', 'education', 'learning platform'],
        'ai_saas': ['ai saas', 'ai tool', 'ai agent', 'ai platform', 'artificial intelligence'],
        'ecommerce': ['ecommerce', 'e-commerce', 'shopify', 'dtc', 'd2c'],
        'cybersecurity': ['cybersecurity', 'security', 'infosec'],
        'martech': ['martech', 'marketing tech', 'marketing tool'],
        'hr_tech': ['hr tech', 'hiring', 'recruiting', 'talent'],
    }
    text_lower = text.lower()
    for vertical, kws in vertical_keywords.items():
        if any(kw in text_lower for kw in kws):
            verticals.add(vertical)
    return verticals


def read_file_safe(filepath):
    """Read file content safely."""
    try:
        with open(filepath) as f:
            return f.read()
    except Exception:
        return ""


def categorize_file(filepath, agent_patterns=None):
    """Categorize a file by agent/source based on filename patterns.

    Override with agent_patterns dict: {"pattern": "agent_name"}
    """
    basename = os.path.basename(filepath).lower()

    # Default patterns — customize these for your setup
    default_patterns = {
        'seo': 'seo',
        'oracle': 'seo',
        'content': 'content',
        'flash': 'content',
        'trend': 'content',
        'deal': 'deal',
        'cold': 'cold_outbound',
        'outbound': 'cold_outbound',
        'recruit': 'recruiting',
        'hiring': 'recruiting',
    }

    patterns = agent_patterns or default_patterns

    for pattern, agent in patterns.items():
        if pattern in basename:
            return agent

    return 'other'


def detect_signals(data_dir, additional_data_dirs=None, hours=48, agent_patterns=None):
    """Main detection logic.

    Args:
        data_dir: Primary directory to scan for agent output files
        additional_data_dirs: Dict of {"agent_name": "glob_pattern"} for extra data
        hours: How far back to look for files
        agent_patterns: Dict of {"filename_pattern": "agent_name"} for categorization
    """
    recent_files = get_recent_files(data_dir, hours=hours)
    if not recent_files:
        # Fallback to 7 days
        recent_files = get_recent_files(data_dir, hours=168)

    # Categorize by agent/source
    agent_data = defaultdict(lambda: {
        "files": [], "companies": set(), "keywords": set(), "verticals": set(), "text": ""
    })

    for f in recent_files:
        agent = categorize_file(f, agent_patterns)
        text = read_file_safe(f)

        agent_data[agent]["files"].append(f)
        agent_data[agent]["companies"].update(extract_companies(text))
        agent_data[agent]["keywords"].update(extract_keywords(text))
        agent_data[agent]["verticals"].update(extract_verticals(text))
        agent_data[agent]["text"] += text + "\n"

    # Scan additional data directories
    if additional_data_dirs:
        for agent, pattern in additional_data_dirs.items():
            files = sorted(glob.glob(pattern))[-1:]  # latest only
            for f in files:
                text = read_file_safe(f)
                agent_data[agent]["companies"].update(extract_companies(text))
                agent_data[agent]["keywords"].update(extract_keywords(text))
                agent_data[agent]["verticals"].update(extract_verticals(text))

    # Find overlaps
    signals = []
    agents_list = list(agent_data.keys())

    # 1. Company overlap
    for i, a1 in enumerate(agents_list):
        for a2 in agents_list[i + 1:]:
            common_companies = agent_data[a1]["companies"] & agent_data[a2]["companies"]
            if common_companies:
                confidence = min(95, 60 + len(common_companies) * 10)
                signals.append({
                    "confidence": confidence,
                    "type": "company_overlap",
                    "agents": [a1, a2],
                    "signal": f"Company overlap: {', '.join(list(common_companies)[:5])} appearing in both {a1} and {a2}",
                    "recommended_play": f"Cross-reference {a1} and {a2} data for these companies — coordinate outreach/content",
                    "entities": list(common_companies)[:10],
                })

    # 2. Vertical overlap
    for i, a1 in enumerate(agents_list):
        for a2 in agents_list[i + 1:]:
            common_verticals = agent_data[a1]["verticals"] & agent_data[a2]["verticals"]
            if common_verticals:
                confidence = min(90, 50 + len(common_verticals) * 15)
                signals.append({
                    "confidence": confidence,
                    "type": "vertical_alignment",
                    "agents": [a1, a2],
                    "signal": f"Vertical alignment: {', '.join(common_verticals)} trending across {a1} + {a2}",
                    "recommended_play": f"Coordinated push into {', '.join(common_verticals)}: content + outbound + SEO",
                    "entities": list(common_verticals),
                })

    # 3. Keyword cluster overlap
    for i, a1 in enumerate(agents_list):
        for a2 in agents_list[i + 1:]:
            common_kw = agent_data[a1]["keywords"] & agent_data[a2]["keywords"]
            if common_kw:
                confidence = min(88, 55 + len(common_kw) * 12)
                signals.append({
                    "confidence": confidence,
                    "type": "keyword_cluster",
                    "agents": [a1, a2],
                    "signal": f"Keyword cluster overlap: {', '.join(list(common_kw)[:3])}",
                    "recommended_play": "Target these keywords in content and outbound simultaneously",
                    "entities": list(common_kw),
                })

    # Deduplicate and sort by confidence
    signals.sort(key=lambda x: x["confidence"], reverse=True)

    output = {
        "date": datetime.now().strftime("%Y-%m-%d"),
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "agents_analyzed": list(agent_data.keys()),
        "files_scanned": sum(len(d["files"]) for d in agent_data.values()),
        "signals": signals[:20],  # top 20
    }

    return output


def main():
    parser = argparse.ArgumentParser(
        description='Cross-Signal Detector — find overlapping signals across data sources'
    )
    default_data_dir = str(SKILL_ROOT / "data" / "output") if SKILL_ROOT else "./data/output"
    default_output = str(SKILL_ROOT / "data" / "intel" / "signals" / "latest.json") if SKILL_ROOT else "./data/cross-signals-latest.json"

    parser.add_argument('--data-dir', default=os.environ.get('DATA_DIR', default_data_dir),
                        help='Directory containing module output files')
    parser.add_argument('--output', default=os.environ.get('OUTPUT_FILE', default_output),
                        help='Output file path')
    parser.add_argument('--hours', type=int, default=48,
                        help='How far back to look for files (default: 48)')
    args = parser.parse_args()

    output = detect_signals(data_dir=args.data_dir, hours=args.hours)

    os.makedirs(os.path.dirname(args.output) or '.', exist_ok=True)
    with open(args.output, "w") as f:
        json.dump(output, f, indent=2)

    signals = output.get("signals", [])
    print(f"Cross-signal detection complete: {len(signals)} signals found")
    print(f"Output: {args.output}")
    if signals:
        print(f"Top signal (confidence {signals[0]['confidence']}): {signals[0]['signal'][:100]}")


if __name__ == "__main__":
    main()