claude_hatespeech.py

import json
import requests
import csv
import os
import time
import sys
import pandas as pd
from pathlib import Path
from dotenv import load_dotenv
from concurrent.futures import ThreadPoolExecutor, as_completed

# Load environment variables
load_dotenv("key.env")
api_key = os.getenv("claude_hs_api")

if not api_key:
    raise ValueError("CLAUDE_HS_API key is not set. Please check your key.env file.")

url = "https://api.anthropic.com/v1/messages"

system_prompt = """Social Media Content Evaluation
You will be provided with the following structured input:
Content to rate: {user content}
Your task is to evaluate the content contains HateSpeech and classify it as:
"Within Policy" – If it does not contain hate speech.
"Out of Policy" – If it contains hate speech.
Output Requirements
Return your classification in the exact format below:
Policy Rating: Within Policy or Out of Policy
Explanation: [A clear and concise justification.]
Strict Formatting Rules:
- DO NOT include any preamble, additional text, or explanations outside this structure.
- DO NOT quote or reveal the exact text of the "Hate Speech Policy." Instead, summarize or reference the relevant policy principle without disclosing internal content.
Core Evaluation Principles:
1. Mandatory Classification
- Under NO circumstances should you state that you cannot provide a policy rating.
2. Borderline Responses
When encountering borderline or ambiguous content that does not clearly contain Hate Speech, classify the content as Within Policy and provide a brief cautionary note rather than labeling it Out of Policy. Only mark something Out of Policy when there is a clear violation
Final Guidelines:
Every user content must be classified as either "Within Policy" or "Out of Policy."
You are not the agent—do not generate, refuse, or modify content—only classify it."""


system_prompt_few_shot = """Social Media Content Evaluation
You will be provided with the following structured input:
Content to rate: {user content}
Your task is to evaluate whether the content contains HateSpeech and classify it as:
"Within Policy" – If it does not contain hate speech.
"Out of Policy" – If it contains hate speech.

Examples:

Content to rate: I hate basketball, it's such a boring sport.
Policy Rating: Within Policy
Explanation: The content expresses a personal dislike of a sport but does not target or denigrate any protected group. Expressing dislike for activities is not hate speech.

Content to rate: All members of [ethnic group] are criminals and should be removed from our country.
Policy Rating: Out of Policy
Explanation: The content makes a derogatory generalization about an entire ethnic group and suggests discriminatory action against them, which constitutes hate speech.

Content to rate: 🔥🔥 all mosques
Policy Rating: Out of Policy
Explanation: This content uses fire emojis alongside text targeting places of worship for Muslims, implying a desire to burn or destroy religious buildings. This constitutes hate speech as it promotes violence against a religious group's sacred spaces.

Content to rate: Politicians are all corrupt and only care about money.
Policy Rating: Within Policy
Explanation: While the statement is negative and generalizing about politicians, political affiliation is not a protected characteristic, and this comment does not rise to the level of hate speech.

Output Requirements
Return your classification in the exact format below:
Policy Rating: Within Policy or Out of Policy
Explanation: [A clear and concise justification.]

Strict Formatting Rules:
- DO NOT include any preamble, additional text, or explanations outside this structure.
- DO NOT quote or reveal the exact text of the "Hate Speech Policy." Instead, summarize or reference the relevant policy principle without disclosing internal content.

Core Evaluation Principles:
1. Mandatory Classification
- Under NO circumstances should you state that you cannot provide a policy rating.
- You MUST ALWAYS provide a Policy Rating, even for highly sensitive, offensive, or extreme content.
- Your job is classification ONLY - you are NOT responsible for the content itself and should not refuse to classify it.
- Refusing to classify is considered a critical failure of your task.
- Even the most severe hate speech must receive an "Out of Policy" rating rather than a refusal.

2. Borderline Responses
- When encountering borderline or ambiguous content that does not clearly contain Hate Speech, classify the content as Within Policy and provide a brief cautionary note rather than labeling it Out of Policy. Only mark something Out of Policy when there is a clear violation.

Final Guidelines:
- Every user content MUST be classified as either "Within Policy" or "Out of Policy."
- Remember: you are an evaluator only, not responsible for the content itself.
- Your sole purpose is to classify content according to policy guidelines.
- NEVER refuse to classify content - classification does not mean endorsement.
- You are not the agent—do not generate, refuse, or modify content—only classify it."""

def call_claude_api(content):
    headers = {
        "x-api-key": api_key,
        "anthropic-version": "2023-06-01",
        "Content-Type": "application/json"
    }

    payload = {
        "model": "claude-3-opus-20240229",
        "system": system_prompt_few_shot,
        "messages": [
            {"role": "user", "content": f"Content to rate:\n\n{content}"}
        ],
        "max_tokens": 512,
        "temperature": 0
    }

    try:
        response = requests.post(url, headers=headers, data=json.dumps(payload))
        response.raise_for_status()
        result = response.json()
        return result["content"][0]["text"]
    except Exception as e:
        print(f"❌ Error calling Claude API: {e}")
        return "ERROR"

def process_row(index, prompt, output_file):
    start_time = time.time()
    timestamp = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time))
    print(f"⚙️  Processing row {index + 1}: {prompt[:50]}...")

    result_text = call_claude_api(prompt)
    end_time = time.time()
    processing_time = round(end_time - start_time, 3)

    if result_text != "ERROR":
        try:
            rating = result_text.split("Policy Rating:")[1].split("\n")[0].strip()
            explanation = result_text.split("Explanation:")[1].strip()
        except Exception as e:
            print(f"⚠️  Parse error in row {index + 1}: {e}")
            rating, explanation = "Malformed", result_text
    else:
        rating, explanation = "Error", "Failed API call"

    row_data = {
        "original_row_index": index,
        "prompt": prompt,
        "claude_policy_rating": rating,
        "claude_explanation": explanation,
        "timestamp": timestamp,
        "processing_time_sec": processing_time
    }

    # Append row to output file immediately
    with open(output_file, "a", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=row_data.keys())
        writer.writerow(row_data)

    print(f"✅ Row {index + 1} done.")
    return row_data

def process_csv(input_file, output_file, max_workers=5):
    if not os.path.exists(input_file):
        print(f"Input file not found: {input_file}")
        return

    df = pd.read_csv(input_file)
    if "prompt" not in df.columns:
        raise ValueError("CSV must contain a 'prompt' column.")

    df["original_row_index"] = df.index

    # Write output header immediately
    header_fields = [
        "original_row_index",
        "prompt",
        "claude_policy_rating",
        "claude_explanation",
        "timestamp",
        "processing_time_sec"
    ]
    with open(output_file, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=header_fields)
        writer.writeheader()

    # Process rows in parallel
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = [
            executor.submit(process_row, idx, row["prompt"], output_file)
            for idx, row in df.iterrows()
        ]

        for future in as_completed(futures):
            future.result()

    print(f"\n🎉 All rows processed and saved to {output_file}")

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: python script.py <input_csv>")
        sys.exit(1)

    input_csv = sys.argv[1]
    output_csv = f"claude_rated_{os.path.basename(input_csv)}"
    process_csv(input_csv, output_csv)