lead-scorer-pro（首席得分手专业版）

Lead Scorer Skill

Takes a completed audit dict → applies scoring rubric → assigns tier → writes row to Google Sheet.

Scoring Rubric (0–100 scale)

Higher score = hotter lead = more likely to need a new website.

SCORING_RUBRIC = [
    # Signal                            Condition                           Points
    ("dead_site",        lambda a: a.get("status_code") in ("DEAD","TIMEOUT","SSL_ERROR"), 40),
    ("copyright_5plus",  lambda a: (a.get("years_outdated") or 0) >= 5,    35),
    ("pagespeed_low",    lambda a: 0 < (a.get("pagespeed_mobile") or 101) < 50, 30),
    ("copyright_3to4",   lambda a: 3 <= (a.get("years_outdated") or 0) < 5, 25),
    ("no_ssl",           lambda a: not a.get("has_ssl"),                    25),
    ("not_mobile",       lambda a: not a.get("is_mobile_friendly"),         20),
    ("outdated_cms",     lambda a: a.get("has_outdated_cms"),               20),
    ("no_contact_found", lambda a: not a.get("primary_email") and not a.get("primary_phone"), 20),
    ("pagespeed_mid",    lambda a: 50 <= (a.get("pagespeed_mobile") or 101) < 70, 15),
    ("table_layout",     lambda a: "table_layout" in (a.get("design_signals") or []), 15),
    ("copyright_1to2",   lambda a: 1 <= (a.get("years_outdated") or 0) < 3, 10),
    ("no_open_graph",    lambda a: "no_open_graph" in (a.get("design_signals") or []), 10),
    ("flash_detected",   lambda a: "flash_detected" in (a.get("design_signals") or []), 20),
    ("uses_frames",      lambda a: "uses_frames" in (a.get("design_signals") or []), 20),
    ("no_meta_desc",     lambda a: "no_meta_description" in (a.get("design_signals") or []), 5),
    ("no_favicon",       lambda a: "no_favicon" in (a.get("design_signals") or []), 5),
    ("font_tags",        lambda a: "font_tags" in (a.get("design_signals") or []), 10),
    ("heavy_inline",     lambda a: "heavy_inline_styles" in (a.get("design_signals") or []), 8),
]

Scoring Function

def score_lead(audit: dict) -> dict:
    """
    Apply rubric to audit dict.
    Returns audit dict + score + tier + issues list + tier emoji.
    """
    score = 0
    issues = []
    matched_rules = []
    
    for rule_name, condition, points in SCORING_RUBRIC:
        try:
            if condition(audit):
                score += points
                matched_rules.append(f"{rule_name} (+{points})")
                issues.append(ISSUE_LABELS.get(rule_name, rule_name))
        except Exception:
            pass
    
    # Cap at 100
    score = min(score, 100)
    
    # Assign tier
    if score >= 80:
        tier = "🔥 Hot"
        tier_code = "hot"
    elif score >= 50:
        tier = "🟡 Warm"
        tier_code = "warm"
    elif score >= 25:
        tier = "🔵 Lukewarm"
        tier_code = "lukewarm"
    else:
        tier = "⚪ Cold"
        tier_code = "cold"
    
    return {
        **audit,
        "lead_score": score,
        "tier": tier,
        "tier_code": tier_code,
        "issues": issues,
        "score_breakdown": matched_rules
    }

# Human-readable issue labels
ISSUE_LABELS = {
    "dead_site":        "Site is dead/unreachable",
    "copyright_5plus":  f"Copyright {'{years}'} years old",
    "copyright_3to4":   "Copyright 3–4 years old",
    "copyright_1to2":   "Copyright 1–2 years old",
    "pagespeed_low":    "PageSpeed score under 50 (mobile)",
    "pagespeed_mid":    "PageSpeed score 50–69 (mobile)",
    "no_ssl":           "No SSL / HTTPS",
    "not_mobile":       "Not mobile responsive",
    "outdated_cms":     "Outdated CMS or tech stack",
    "no_contact_found": "No contact info found on site",
    "table_layout":     "Table-based layout (pre-2010 design)",
    "no_open_graph":    "No social meta tags",
    "flash_detected":   "Flash / Silverlight detected",
    "uses_frames":      "Uses HTML frames",
    "no_meta_desc":     "Missing meta description",
    "no_favicon":       "No favicon",
    "font_tags":        "Font tags detected (ancient HTML)",
    "heavy_inline":     "Heavy inline CSS styles",
}

Google Sheets Integration

import gspread
from google.oauth2.service_account import Credentials
from datetime import datetime
import os

SHEET_COLUMNS = [
    "Date Found", "Business Name", "URL", "Lead Score", "Tier",
    "Primary Email", "All Emails", "Phone", "Copyright Year",
    "Tech Stack", "PageSpeed (Mobile)", "Has SSL", "Mobile Friendly",
    "Issues Found", "Score Breakdown", "Status", "Notes"
]

def get_sheet(sheet_name: str = None, creds_file: str = None):
    """Authenticate and return the target Google Sheet."""
    sheet_name = sheet_name or os.environ.get("GOOGLE_SHEET_NAME", "Website Leads")
    creds_file = creds_file or os.environ.get("GOOGLE_CREDS_FILE", "credentials.json")
    
    scopes = [
        "https://www.googleapis.com/auth/spreadsheets",
        "https://www.googleapis.com/auth/drive"
    ]
    creds = Credentials.from_service_account_file(creds_file, scopes=scopes)
    client = gspread.authorize(creds)
    
    try:
        sheet = client.open(sheet_name).sheet1
    except gspread.SpreadsheetNotFound:
        # Create the sheet if it doesn't exist
        spreadsheet = client.create(sheet_name)
        spreadsheet.share(None, perm_type="anyone", role="writer")
        sheet = spreadsheet.sheet1
        # Write headers
        sheet.append_row(SHEET_COLUMNS)
        print(f"Created new sheet: {sheet_name}")
    
    return sheet

def ensure_headers(sheet):
    """Make sure the header row exists."""
    first_row = sheet.row_values(1)
    if not first_row or first_row[0] != "Date Found":
        sheet.insert_row(SHEET_COLUMNS, index=1)

def lead_to_row(lead: dict) -> list:
    """Convert a scored lead dict to a sheet row."""
    return [
        datetime.now().strftime("%Y-%m-%d"),
        lead.get("business_name", ""),
        lead.get("url", ""),
        lead.get("lead_score", 0),
        lead.get("tier", ""),
        lead.get("primary_email", ""),
        ", ".join(lead.get("emails", [])),
        lead.get("primary_phone", ""),
        lead.get("copyright_year", ""),
        ", ".join(lead.get("tech_stack", [])[:3]),          # Top 3 techs
        lead.get("pagespeed_mobile", ""),
        "✅" if lead.get("has_ssl") else "❌",
        "✅" if lead.get("is_mobile_friendly") else "❌",
        " | ".join(lead.get("issues", [])[:5]),             # Top 5 issues
        " | ".join(lead.get("score_breakdown", [])[:5]),
        "New",
        ""
    ]

def write_lead_to_sheet(lead: dict, sheet=None) -> bool:
    """Score the lead and write it to the Google Sheet."""
    if sheet is None:
        sheet = get_sheet()
    ensure_headers(sheet)
    
    scored = score_lead(lead)
    row = lead_to_row(scored)
    
    try:
        sheet.append_row(row, value_input_option="USER_ENTERED")
        return True
    except Exception as e:
        print(f"Sheet write error: {e}")
        return False

def write_leads_batch(leads: list[dict], sheet=None) -> int:
    """Write multiple leads at once. Returns count written."""
    if sheet is None:
        sheet = get_sheet()
    ensure_headers(sheet)
    
    rows = []
    for lead in leads:
        scored = score_lead(lead)
        if scored.get("lead_score", 0) >= 0:  # Always write (filter upstream)
            rows.append(lead_to_row(scored))
    
    if rows:
        sheet.append_rows(rows, value_input_option="USER_ENTERED")
    
    return len(rows)

Apply Conditional Formatting (Color by Tier)

def apply_conditional_formatting(spreadsheet_id: str, creds_file: str = None):
    """
    Apply color-coding to the sheet based on Lead Score in column D.
    Requires google-api-python-client (pip install google-api-python-client).
    """
    from googleapiclient.discovery import build
    from google.oauth2.service_account import Credentials
    
    creds_file = creds_file or os.environ.get("GOOGLE_CREDS_FILE", "credentials.json")
    scopes = ["https://www.googleapis.com/auth/spreadsheets"]
    creds = Credentials.from_service_account_file(creds_file, scopes=scopes)
    service = build("sheets", "v4", credentials=creds)
    
    rules = [
        # 🔥 Hot (≥80) — red background
        {"ranges": [{"sheetId": 0, "startRowIndex": 1, "endRowIndex": 1000,
                     "startColumnIndex": 0, "endColumnIndex": 17}],
         "booleanRule": {
             "condition": {"type": "NUMBER_GREATER_THAN_EQ",
                           "values": [{"userEnteredValue": "80"}]},
             "format": {"backgroundColor": {"red": 1.0, "green": 0.8, "blue": 0.8}}
         }},
        # 🟡 Warm (50–79) — yellow background
        {"ranges": [{"sheetId": 0, "startRowIndex": 1, "endRowIndex": 1000,
                     "startColumnIndex": 0, "endColumnIndex": 17}],
         "booleanRule": {
             "condition": {"type": "NUMBER_BETWEEN",
                           "values": [{"userEnteredValue": "50"}, {"userEnteredValue": "79"}]},
             "format": {"backgroundColor": {"red": 1.0, "green": 0.95, "blue": 0.7}}
         }},
        # 🔵 Lukewarm (25–49) — blue background
        {"ranges": [{"sheetId": 0, "startRowIndex": 1, "endRowIndex": 1000,
                     "startColumnIndex": 0, "endColumnIndex": 17}],
         "booleanRule": {
             "condition": {"type": "NUMBER_BETWEEN",
                           "values": [{"userEnteredValue": "25"}, {"userEnteredValue": "49"}]},
             "format": {"backgroundColor": {"red": 0.8, "green": 0.9, "blue": 1.0}}
         }},
    ]
    
    # The "D" column (index 3) is used for conditional formatting range
    body = {"requests": [{"addConditionalFormatRule": {"rule": r, "index": i}}
                         for i, r in enumerate(rules)]}
    service.spreadsheets().batchUpdate(spreadsheetId=spreadsheet_id, body=body).execute()
    print("Conditional formatting applied ✅")

Full Pipeline Usage Example

# Putting it all together
from lead_list_builder import run_pipeline

results = run_pipeline(
    niche="landscaping",
    city="Portland OR",
    limit=25,
    min_score=30,
    sheet_name="Portland Landscaping Leads"
)

# Output:
# ✅ Scan complete.
# URLs scanned: 38
# Leads written: 25
# 🔥 Hot: 8 | 🟡 Warm: 11 | 🔵 Lukewarm: 6 | ⚪ Cold: 13

Score Interpretation Guide