#!/usr/bin/env python3 """ Extract delighters (positives) from Reddit comments via OpenAI Responses API (Structured Outputs). Input: CSV file with columns: comment_id, post_url, comment Output: CSV file with the structure: comment_id, post_url, comment, canonical, category, quotes Usage: export OPENAI_API_KEY=sk-... python extract_delighters.py --input comments.csv --output delighters.csv --model gpt-4o Notes: - Uses Structured Outputs with json_schema for parseable results - Batches comments by token count (tiktoken) if available, else by character count - Retries transient failures with exponential backoff """ import os import json import time import math import argparse from typing import List, Dict, Any import csv import hashlib import re def gen_id(text): return hashlib.sha256(text.encode("utf-8")).hexdigest() # OpenAI SDK try: from openai import OpenAI except Exception as e: raise SystemExit("Please install the official OpenAI Python SDK: pip install openai") from e # Optional tokenizer try: import tiktoken _ENC = tiktoken.get_encoding("cl100k_base") except Exception: _ENC = None # --------- Config Defaults --------- DEFAULT_MODEL = "gpt-4o" INPUT_TOKEN_BUDGET = 3000 OUTPUT_TOKENS = 1200 TEMP = 0.1 MAX_RETRIES = 4 # --------- Schema (STRICT) --------- DELIGHTER_SCHEMA = { "name": "delighters", "schema": { "type":"object", "required":["items"], "properties":{ "items":{ "type":"array", "items":{ "type":"object", "required":["comment_id","delighters"], "properties":{ "comment_id":{"type":"string"}, "delighters":{ "type":"array", "items":{ "type":"object", "required":["canonical","quotes"], "properties":{ "canonical":{"type":"string"}, "category":{"type":"string"}, "quotes":{"type":"array","items":{"type":"string"}} }, "additionalProperties":False } } }, "additionalProperties":False } } }, "additionalProperties":False } } # --------- Prompts --------- SYSTEM_PROMPT = """You extract *delighters* (positives) from Reddit comments. A delighter is a feature, behavior, or aspect of the app that users explicitly like, appreciate, or value. It may include: favorite features, smooth experiences, surprising benefits, strong reliability, helpful support, or positive emotional impacts. Your tasks (per comment): - Identify distinct delighters (0 or more). - For each delighter, produce: - canonical: short label (3–8 words), neutral phrasing. - category: optional tag (choose one if applicable): UX, Features, Integrations, Performance, Community, Pricing, Support, Other. - quotes: array of verbatim substrings from the comment that best illustrate the delighter. Inclusion rules: - Features users praise or enjoy - Positive emotional reactions - Workflows that are easy or helpful - Aspects users say they "love", "like", "prefer", "use all the time" Exclusion rules: - General positivity with no detail ("great app", "love it") - Off-topic or irrelevant discussion - Pain points or negatives unless contrasted with a positive Quoting rules: - Quotes must be verbatim substrings from the provided comment text. - Prefer 1–3 short excerpts per delighter. - Do not paraphrase. General constraints: - No ranking, scoring, frequencies, or summaries across comments. - If a comment has no valid delighters, return it with an empty delighters array. """ USER_INSTRUCTIONS = """Process the following Reddit comments and return ONLY valid JSON matching the schema. For each comment: extract distinct delighters (0+), following the inclusion/exclusion and quoting rules above. Comments (array of objects): {comments_json} """ # --------- CSV Writer --------- def save_to_csv(batch_items: list, comments_map: dict, output_path: str, append: bool = True): fieldnames = ["comment_id", "post_url", "comment", "canonical", "category", "quotes"] mode = "a" if append and os.path.exists(output_path) else "w" with open(output_path, mode, encoding="utf-8", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) if mode == "w": # prima dată, scriem și header writer.writeheader() for item in batch_items: if not isinstance(item, dict): continue cid = item.get("comment_id", "") original = comments_map.get(cid, {}) if "delighters" in item: for dl in item.get("delighters", []): if not isinstance(dl, dict): print(f"⚠️ Skipping malformed delighter for {cid}: {dl}") continue writer.writerow({ "comment_id": cid, "post_url": original.get("post_url", ""), "comment": original.get("text", ""), "canonical": dl.get("canonical", ""), "category": dl.get("category", ""), "quotes": " || ".join(dl.get("quotes", [])) }) else: writer.writerow({ "comment_id": cid, "post_url": item.get("post_url", ""), "comment": item.get("comment", ""), "canonical": item.get("canonical", ""), "category": item.get("category", ""), "quotes": item.get("quotes", "") }) # --------- Utilities --------- def count_tokens(text: str) -> int: if _ENC is None: return max(1, math.ceil(len(text) / 4)) return len(_ENC.encode(text)) def pack_batches(comments: List[Dict[str, str]], max_total_tokens: int, expected_output_tokens_per_comment: int = 60) -> List[List[Dict[str, str]]]: batches = [] current = [] current_input_tokens = 0 overhead = 1000 for item in comments: s = json.dumps(item, ensure_ascii=False) t = count_tokens(s) est_output = (len(current) + 1) * expected_output_tokens_per_comment if t + current_input_tokens + overhead + est_output > max_total_tokens: if current: batches.append(current) current = [item] current_input_tokens = t else: current.append(item) current_input_tokens += t if current: batches.append(current) return batches def run_with_retries(fn, *args, **kwargs): delay = 2.0 for attempt in range(1, MAX_RETRIES + 1): try: return fn(*args, **kwargs) except Exception as e: if attempt == MAX_RETRIES: raise time.sleep(delay) delay = min(30.0, delay * 1.8) def safe_parse_response(resp): func_args = resp.choices[0].message.function_call.arguments if func_args: try: return json.loads(func_args) except json.JSONDecodeError: try: return json.loads(func_args.encode("utf-8").decode("unicode_escape")) except Exception: pass m = re.search(r'\{.*\}|\[.*\]', func_args, re.DOTALL) if m: try: return json.loads(m.group(0)) except Exception: pass content = resp.choices[0].message.content if content: try: return json.loads(content) except json.JSONDecodeError as e: raise RuntimeError(f"Could not parse model output:\n{content[:500]}...") from e raise RuntimeError("No valid JSON found in model response.") def call_openai_batch(client, model: str, batch: list) -> dict: user = USER_INSTRUCTIONS.format(comments_json=json.dumps(batch, ensure_ascii=False)) fc_schema = { "name": "extract_delighters", "description": "Extract delighters (positives) from a batch of Reddit comments.", "parameters": DELIGHTER_SCHEMA["schema"] } resp = client.chat.completions.create( model=model, messages=[ {"role": "system", "content": SYSTEM_PROMPT}, {"role": "user", "content": user} ], functions=[fc_schema], function_call={"name": "extract_delighters"}, temperature=TEMP, max_tokens=OUTPUT_TOKENS ) return safe_parse_response(resp) def merge_items(into: Dict[str, Any], batch_obj: Dict[str, Any]) -> None: into.setdefault("items", []) if not batch_obj: return for it in batch_obj.get("items", []): into["items"].append(it) def main(): ap = argparse.ArgumentParser() ap.add_argument("--input", required=True, help="Path to input csv with [{comment_id, text}, ...]") ap.add_argument("--output", required=True, help="Path to output csv with delighters") ap.add_argument("--model", default=DEFAULT_MODEL, help=f"Model name (default: {DEFAULT_MODEL})") ap.add_argument("--batch_tokens", type=int, default=INPUT_TOKEN_BUDGET, help="Approx input token budget per request") args = ap.parse_args() api_key = os.getenv("OPENAI_API_KEY") if not api_key: raise SystemExit("Set OPENAI_API_KEY in your environment.") client = OpenAI(api_key=api_key) # ---------- NEW: sidecar path ---------- base_name = os.path.splitext(os.path.basename(args.input))[0] sidecar_path = f"{base_name}.processed_ids" comments = [] with open(args.input, "r", encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: txt = str(row.get("comment", "")).strip() comment_id = gen_id(txt) url = str(row.get("post_url", "")).strip() if txt: comment = { "comment_id": comment_id, "text": txt, "post_url": url } comments.append(comment) # ---------- NEW: load processed IDs ---------- processed_comments_ids = set() if os.path.exists(sidecar_path): with open(sidecar_path, "r", encoding="utf-8") as f: for line in f: processed_comments_ids.add(line.strip()) not_processed_comments = [c for c in comments if c["comment_id"] not in processed_comments_ids] batches = pack_batches(not_processed_comments, args.batch_tokens) print(f"Processing {len(not_processed_comments)} comments in {len(batches)} batches...") comments_map = {c["comment_id"]: c for c in comments} all_items = [] # don't reload from output, we'll just append for i, batch in enumerate(batches, 1): print(f"Batch {i}/{len(batches)}: {len(batch)} comments") try: out = run_with_retries(call_openai_batch, client, args.model, batch) save_to_csv(out.get("items", []), comments_map, args.output, append=True) # scriem și în sidecar with open(sidecar_path, "a", encoding="utf-8") as f: for c in batch: f.write(c["comment_id"] + "\n") except RuntimeError: continue print(f"✅ All results saved to {args.output}") print(f"📝 Progress tracked in {sidecar_path}") if __name__ == "__main__": main()