#!/usr/bin/env python3 """ Extract pain points from Reddit comments via llm_client (POOL_ANTHROPIC_KEYS / POOL_GEMINI_API_KEYS). Input: JSON file with an array of objects: [{"comment_id": "", "comment": ""}, ...] Output: JSON file with the structure: { "items": [ { "comment_id": "", "pain_points": [ { "canonical": "", "quotes": ["", "..."] } ] }, ... ] } Usage: python extract_pain_points.py --input comments.json --output pain_points.json --model claude-opus-5 Notes: - Schema is described in the system prompt; llm_client parses the JSON response - Batches comments by token count (tiktoken) if available, else by character count - Retries are handled by llm_client's key-pool rotation and JSON-repair logic """ import os import json import math import argparse from typing import List, Dict, Any import csv import csv import hashlib def gen_id(text): return hashlib.sha256(text.encode("utf-8")).hexdigest() from llm_client import call_llm_chat_json # Optional tokenizer for safer packing try: import tiktoken _ENC = tiktoken.get_encoding("cl100k_base") except Exception: _ENC = None # --------- Config Defaults --------- DEFAULT_MODEL = None # falls back to LLM_MODEL in llm_client.py # Keep some headroom so the model can respond; tweak if you expect many pain points per batch INPUT_TOKEN_BUDGET = 3000 OUTPUT_TOKENS = 1200 TEMP = 0.1 MAX_RETRIES = 4 # --------- Schema (STRICT) --------- PAIN_POINT_SCHEMA = { "name": "pain_points", "schema": { "type":"object", "required":["items"], "properties":{ "items":{ "type":"array", "items":{ "type":"object", "required":["comment_id","pain_points"], "properties":{ "comment_id":{"type":"string"}, "pain_points":{ "type":"array", "items":{ "type":"object", "required":["canonical","quotes"], "properties":{ "canonical":{"type":"string"}, "quotes":{"type":"array","items":{"type":"string"}} }, "additionalProperties":False } } }, "additionalProperties":False } } }, "additionalProperties":False } } # --------- Prompts --------- # GENERIC PROMPT # SYSTEM_PROMPT = """You extract pain points from Reddit comments. # A pain point is a specific problem, frustration with an existing solution, unmet need/desire, workaround, concrete usage scenario where the problem occurs, or negative emotional impact expressed by the commenter. # Your tasks (per comment): # - Identify distinct pain points (0 or more). # - For each pain point, produce: # - canonical: short label (3–8 words) that could serve as a graph node (neutral, no emojis). # - quotes: array of exact verbatim substrings from the comment that best illustrate the pain point. Preserve punctuation, casing, spelling, and slang. # Inclusion rules (extract if any apply): # - Specific problems users are experiencing # - Frustrations with existing solutions # - Unmet needs and desires # - Workarounds they created # - Specific usage scenarios (when/where the problem occurs) # - Emotional impact # Exclusion rules (do NOT extract): # - General/off-topic discussion unrelated to problems or needs # - Advice-seeking with no described problem, or vague complaints without details # - Pure positives unless explicitly contrasting a problem # - Hearsay about others without the commenter describing an encountered issue # Quoting rules: # - Quotes must be verbatim substrings of the provided comment text. # - Prefer 1–3 short excerpts per pain point that clearly evidence the issue. # - Do not paraphrase. # General constraints: # - No ranking, scoring, frequencies, or cross-comment summaries. # - No invented content; only extract what is present. # - If a comment has no valid pain points, return it with an empty pain_points array. # """ SYSTEM_PROMPT = """You extract pain points from Reddit comments. A pain point is a specific problem, frustration with an existing solution, unmet need/desire, workaround, concrete usage scenario where the problem occurs, or negative emotional impact expressed by the commenter. Your tasks (per comment): - Identify distinct pain points (0 or more). - For each pain point, produce: - canonical: short label (3–8 words) that could serve as a graph node (neutral, no emojis). Prefer scenario-based or impact-based labels when possible (e.g., "sleep disruption from late-night scrolling", "anxiety after negative news"). - quotes: array of exact verbatim substrings from the comment that best illustrate the pain point. Preserve punctuation, casing, spelling, and slang. Inclusion rules (extract if any apply): - Specific problems users are experiencing - Frustrations with existing solutions - Unmet needs and desires - Workarounds they created - Specific usage scenarios (when/where the problem occurs) - Emotional impacts (e.g., regret, anxiety, stress, guilt, depression, feeling unproductive, compulsive use) Exclusion rules (do NOT extract): - General/off-topic discussion unrelated to problems or needs - Advice-seeking with no described problem, or vague complaints without details - Pure positives unless explicitly contrasting a problem - Hearsay about others without the commenter describing an encountered issue Quoting rules: - Quotes must be verbatim substrings of the provided comment text. - Prefer 1–3 short excerpts per pain point that clearly evidence the issue. - Do not paraphrase. General constraints: - No ranking, scoring, frequencies, or cross-comment summaries. - No invented content; only extract what is present. - If a comment has no valid pain points, return it with an empty pain_points array. """ USER_INSTRUCTIONS = """Process the following Reddit comments and return ONLY valid JSON matching the schema. For each comment: extract distinct pain points (0+), following the inclusion/exclusion and quoting rules above. Comments (array of objects): {comments_json} """ def save_to_csv(batch_items: list, comments_map: dict, output_path: str, append: bool = False): fieldnames = ["comment_id", "post_url", "author", "upvotes", "comment", "canonical", "quotes"] # dacă avem append și fișierul există -> deschidem cu "a", altfel "w" 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 headerul writer.writeheader() for item in batch_items: if not isinstance(item, dict): continue # skip dacă e string corupt cid = item.get("comment_id", "") original = comments_map.get(cid, {}) # Dacă item are pain_points => e rezultat nou if "pain_points" in item: for pp in item.get("pain_points", []): if not isinstance(pp, dict): print(f"⚠️ Skipping malformed pain_point for {cid}: {pp}") continue writer.writerow({ "comment_id": cid, "post_url": original.get("post_url", ""), "author": original.get("author", ""), "upvotes": original.get("upvotes", ""), "comment": original.get("comment", ""), "canonical": pp.get("canonical", ""), "quotes": " || ".join(pp.get("quotes", [])) }) else: # Item deja formatat (de ex. la resume) writer.writerow({ "comment_id": cid, "post_url": item.get("post_url", ""), "author": item.get("author", ""), "upvotes": item.get("upvotes", ""), "comment": item.get("comment", ""), "canonical": item.get("canonical", ""), "quotes": item.get("quotes", "") }) # --------- Utilities --------- def count_tokens(text: str) -> int: if _ENC is None: # rough fallback: 4 chars per token heuristic 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]]]: """ Packs comments into batches that fit under max_total_tokens (input + expected output). Estimates output size as `expected_output_tokens_per_comment` * number of comments. """ batches = [] current = [] current_input_tokens = 0 overhead = 1000 # for system prompt, formatting, etc. 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 call_llm_batch(model: str, batch: list) -> dict: user = USER_INSTRUCTIONS.format(comments_json=json.dumps(batch, ensure_ascii=False)) schema_hint = json.dumps(PAIN_POINT_SCHEMA["schema"]) system = SYSTEM_PROMPT + f"\n\nReturn ONLY valid JSON matching this schema:\n{schema_hint}" result = call_llm_chat_json( { "temperature": TEMP, "max_tokens": OUTPUT_TOKENS, "retry_on_invalid_json": MAX_RETRIES - 1, }, system_message=system, user_message=user, model=model, ) if result is None: raise RuntimeError("LLM call failed after retries") return result 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 [{comment_id, text}, ...]") ap.add_argument("--model", default=DEFAULT_MODEL, help="Model name (default: LLM_MODEL env var)") ap.add_argument("--batch_tokens", type=int, default=INPUT_TOKEN_BUDGET, help="Approx input token budget per request") args = ap.parse_args() base_name = os.path.splitext(os.path.basename(args.input))[0] sidecar_path = f"{base_name}.processed_ids" comments = [] processed_comments_ids = [] processed_comments = [] with open(args.input, "r", encoding="utf-8") as f: reader = csv.DictReader(f) headers = [h.lower() for h in reader.fieldnames or []] if "comment" not in headers: raise SystemExit("❌ Input file must contain a 'comment' column.") print(f"⚙️ Detected columns: {headers}") for row in reader: txt = str(row.get("comment", "")).strip() if not txt: continue comment_id = gen_id(txt) # Fill missing fields if not present in input comment = { "comment_id": comment_id, "comment": txt, "post_url": str(row.get("post_url", "")).strip() if "post_url" in headers else "", "author": str(row.get("author", "")).strip() if "author" in headers else "", "upvotes": str(row.get("upvotes", "")).strip() if "upvotes" in headers else "" } comments.append(comment) 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} for i, batch in enumerate(batches, 1): print(f"Batch {i}/{len(batches)}: {len(batch)} comments") try: out = call_llm_batch(args.model, batch) save_to_csv(out.get("items", []), comments_map, args.output, append=True) # scrie în sidecar toate comment_id-urile procesate 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}") if __name__ == "__main__": main()