如何通过搜索验证人工智能生成的品牌提及

import os, re, requests

SCAVIO_KEY = os.environ['SCAVIO_API_KEY']
H = {'x-api-key': SCAVIO_KEY, 'Content-Type': 'application/json'}

def extract_claims(text: str) -> list:
    """Extract verifiable claims from AI-generated text."""
    claims = []
    # Pricing claims: "costs $X", "$X/month", "starts at $X"
    for m in re.finditer(r'(\w[\w\s]+?)\s+(?:costs?|priced? at|starts? at|for)\s+(\$[\d,.]+(?:/\w+)?)', text):
        claims.append({'type': 'pricing', 'brand': m.group(1).strip(), 'claim': m.group(2)})
    # Feature claims: "X offers Y", "X includes Y", "X supports Y"
    for m in re.finditer(r'(\w[\w\s]+?)\s+(?:offers?|includes?|supports?|provides?|features?)\s+(.+?)[\.!,]', text):
        claims.append({'type': 'feature', 'brand': m.group(1).strip(), 'claim': m.group(2).strip()})
    # Comparison claims: "X is better than Y", "X outperforms Y"
    for m in re.finditer(r'(\w+)\s+(?:is better than|outperforms|beats|surpasses)\s+(\w+)', text):
        claims.append({'type': 'comparison', 'brand': m.group(1), 'claim': f'better than {m.group(2)}'})
    return claims

# Example AI-generated text
ai_text = """Notion costs $10/month for the Pro plan and offers real-time collaboration.
Obsidian starts at $50/year for commercial use and supports plugin extensions.
Notion is better than Obsidian for team collaboration."""

claims = extract_claims(ai_text)
for c in claims:
    print(f'[{c["type"]}] {c["brand"]}: {c["claim"]}')

import time

def verify_claim(claim: dict) -> dict:
    """Verify a single claim against search data."""
    brand = claim['brand']
    claim_text = claim['claim']
    # Build verification query
    if claim['type'] == 'pricing':
        query = f'{brand} pricing plans 2026'
    elif claim['type'] == 'feature':
        query = f'{brand} {claim_text}'
    else:
        query = f'{brand} vs {claim_text.replace("better than ", "")}'
    resp = requests.post('https://api.scavio.dev/api/v1/search', headers=H,
        json={'query': query, 'country_code': 'us', 'num_results': 5})
    results = resp.json().get('organic_results', [])
    all_text = ' '.join(f"{r.get('title','')} {r.get('snippet','')}" for r in results).lower()
    # Check verification
    if claim['type'] == 'pricing':
        price_val = re.search(r'\$([\d,.]+)', claim_text)
        if price_val:
            found = price_val.group(1) in all_text or price_val.group(0) in all_text
            return {**claim, 'verified': found,
                    'status': 'VERIFIED' if found else 'UNVERIFIED',
                    'evidence': all_text[:200]}
    elif claim['type'] == 'feature':
        key_terms = [w for w in claim_text.lower().split() if len(w) > 3]
        matches = sum(1 for t in key_terms if t in all_text)
        coverage = matches / len(key_terms) if key_terms else 0
        return {**claim, 'verified': coverage > 0.5,
                'status': 'VERIFIED' if coverage > 0.5 else 'UNVERIFIED',
                'evidence': all_text[:200]}
    return {**claim, 'verified': False, 'status': 'CHECK MANUALLY', 'evidence': all_text[:200]}

for claim in claims:
    result = verify_claim(claim)
    print(f'[{result["status"]}] {result["brand"]}: {result["claim"]}')
    time.sleep(0.3)

def validate_content(text: str) -> dict:
    claims = extract_claims(text)
    if not claims:
        return {'status': 'NO_CLAIMS', 'message': 'No verifiable claims found'}
    results = []
    for claim in claims:
        result = verify_claim(claim)
        results.append(result)
        time.sleep(0.3)
    verified = sum(1 for r in results if r['status'] == 'VERIFIED')
    unverified = sum(1 for r in results if r['status'] == 'UNVERIFIED')
    manual = sum(1 for r in results if r['status'] == 'CHECK MANUALLY')
    total = len(results)
    accuracy = verified / total if total else 0
    if accuracy >= 0.8:
        overall = 'PASS'
    elif accuracy >= 0.5:
        overall = 'REVIEW'
    else:
        overall = 'FAIL'
    report = {
        'overall': overall,
        'accuracy': accuracy,
        'verified': verified,
        'unverified': unverified,
        'manual_check': manual,
        'total_claims': total,
        'results': results,
        'cost': total * 0.005,
    }
    print(f'Content Validation: {overall}')
    print(f'Claims: {verified} verified, {unverified} unverified, {manual} manual check')
    print(f'Accuracy: {accuracy:.0%}')
    print(f'Cost: ${report["cost"]:.3f}')
    for r in results:
        icon = 'v' if r['status'] == 'VERIFIED' else 'x' if r['status'] == 'UNVERIFIED' else '?'
        print(f'  [{icon}] {r["brand"]}: {r["claim"]}')
    return report

validate_content(ai_text)

def pre_publish_check(content: str, min_accuracy: float = 0.7) -> dict:
    """Run before publishing AI-generated content."""
    report = validate_content(content)
    if report.get('status') == 'NO_CLAIMS':
        return {'action': 'PUBLISH', 'reason': 'No brand claims to verify'}
    if report['accuracy'] >= min_accuracy:
        return {
            'action': 'PUBLISH',
            'reason': f'{report["accuracy"]:.0%} accuracy meets threshold',
            'warnings': [r for r in report['results'] if r['status'] != 'VERIFIED']
        }
    return {
        'action': 'HOLD',
        'reason': f'{report["accuracy"]:.0%} accuracy below {min_accuracy:.0%} threshold',
        'unverified_claims': [r for r in report['results'] if r['status'] != 'VERIFIED'],
        'suggestion': 'Review and correct unverified claims before publishing'
    }

# Test the pipeline
result = pre_publish_check(ai_text)
print(f'\nAction: {result["action"]}')
print(f'Reason: {result["reason"]}')
if result.get('warnings'):
    print('Warnings:')
    for w in result['warnings']:
        print(f'  - {w["brand"]}: {w["claim"]} ({w["status"]})')