#!/usr/bin/env python """Feedback Analyzer — paste or upload customer feedback in bulk and get, per item: sentiment + intent + confidence + a reason + the evidence phrase — plus a rollup (sentiment/intent mix + top themes) and a downloadable labeled CSV. Reliability: ONE Claude call with forced tool-use returns the whole batch as schema-valid JSON (Ollama fallback uses constrained JSON decoding). pip install flask requests anthropic python app.py # http://127.0.0.1:8500 Project #5 of the "30 Projects in 15 Days" challenge — GritAI. """ import os, io, csv, json from flask import Flask, request, jsonify, Response PORT = int(os.environ.get("PORT", "8500")) ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY") ANTHROPIC_MODEL = os.environ.get("ANTHROPIC_MODEL", "claude-sonnet-5") OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434") OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "qwen2.5:7b") PRESET_INTENTS = ["praise", "complaint", "question", "suggestion", "other"] MAX_ROWS = 40 TEXT_COLS = ("text", "review", "comment", "feedback", "message", "body", "content", "response") def analysis_schema(labels): intents = labels + ["other"] if labels else PRESET_INTENTS item = {"type": "object", "properties": { "index": {"type": "integer"}, "sentiment": {"type": "string", "enum": ["positive", "negative", "neutral", "mixed"]}, "intent": {"type": "string", "enum": intents}, "confidence": {"type": "number"}, "reason": {"type": "string"}, "evidence": {"type": "string"}}, "required": ["index", "sentiment", "intent", "confidence", "reason", "evidence"]} return {"type": "object", "properties": { "results": {"type": "array", "items": item}, "themes": {"type": "array", "items": {"type": "string"}}}, "required": ["results", "themes"]} def analyze_batch(rows, labels): schema = analysis_schema(labels) numbered = "\n".join(f"[{i}] {t[:400]}" for i, t in enumerate(rows)) sysp = ("You analyze customer feedback. For EACH numbered item return: sentiment (overall tone), " "intent (its primary purpose), confidence (0-1, how sure you are), reason (one short sentence), " "and evidence (the exact phrase from THAT item that supports your call). " "Also return 'themes': the 3-6 most common themes across all items, short noun phrases, most frequent first. " "Base everything only on the text; never invent details.") user = "Feedback items:\n" + numbered if ANTHROPIC_KEY: import anthropic client = anthropic.Anthropic(api_key=ANTHROPIC_KEY) r = client.messages.create(model=ANTHROPIC_MODEL, max_tokens=4096, system=sysp, tools=[{"name": "report", "description": "Return the analysis.", "input_schema": schema}], tool_choice={"type": "tool", "name": "report"}, messages=[{"role": "user", "content": user}]) for b in r.content: if b.type == "tool_use": return b.input return {"results": [], "themes": []} import requests r = requests.post(f"{OLLAMA_URL}/api/chat", timeout=180, json={ "model": OLLAMA_MODEL, "stream": False, "format": schema, "messages": [{"role": "system", "content": sysp}, {"role": "user", "content": user}], "options": {"temperature": 0}}) r.raise_for_status() return json.loads(r.json()["message"]["content"]) def parse_rows(raw, is_csv): if is_csv: rows = list(csv.reader(io.StringIO(raw))) if not rows: return [] low = [c.strip().lower() for c in rows[0]] idx = next((i for i, h in enumerate(low) if h in TEXT_COLS), 0) start = 1 if low[idx] in TEXT_COLS else 0 # skip header only if we recognized one return [r[idx].strip() for r in rows[start:] if len(r) > idx and r[idx].strip()] # paste mode: one item per line (predictable). Multi-paragraph reviews -> use CSV, # where csv.reader correctly keeps quoted, newline-containing fields as one item. return [l.strip() for l in raw.splitlines() if l.strip()] app = Flask(__name__) app.config["MAX_CONTENT_LENGTH"] = 5 * 1024 * 1024 @app.route("/") def home(): return Response(PAGE, mimetype="text/html") @app.route("/api/analyze", methods=["POST"]) def api_analyze(): b = request.get_json(force=True) raw = (b.get("input") or "").strip() is_csv = bool(b.get("is_csv")) labels = [l.strip() for l in (b.get("labels") or "").split(",") if l.strip()] rows = parse_rows(raw, is_csv) if not rows: return jsonify(error="No rows found. Paste one item per line, or upload a CSV with a text column."), 400 capped = rows[:MAX_ROWS] try: out = analyze_batch(capped, labels) except Exception as e: import sys; print("analyze error:", type(e).__name__, file=sys.stderr, flush=True) return jsonify(error="Analysis failed — please try again."), 200 # attach original text, order by index, compute counts in code (deterministic) by_idx = {r.get("index", i): r for i, r in enumerate(out.get("results", []))} results, sent, intent = [], {}, {} for i, t in enumerate(capped): r = dict(by_idx.get(i, {"sentiment": "neutral", "intent": "other", "confidence": 0, "reason": "", "evidence": ""})) r["text"] = t results.append(r) sent[r.get("sentiment", "neutral")] = sent.get(r.get("sentiment", "neutral"), 0) + 1 intent[r.get("intent", "other")] = intent.get(r.get("intent", "other"), 0) + 1 return jsonify(results=results, themes=out.get("themes", []), sentiment=sent, intent=intent, total=len(capped), truncated=len(rows) > MAX_ROWS, original=len(rows)) PAGE = """ Feedback Analyzer
GRITAIFeedback Analyzer
▸ LiveSDVOSBLawton OK

Every review, scored for sentiment & intent

Paste or upload customer feedback in bulk — get per-item sentiment, intent, confidence, a reason, and the exact evidence phrase, plus a rollup and a downloadable labeled CSV.

01Feedback
Load sample
›
02Rollup

Sentiment

Top themes

    03Per-item results

    Labeled feedback

    FeedbackSentimentIntentConf.ReasonEvidence
    """ if __name__ == "__main__": backend = "Claude API" if ANTHROPIC_KEY else f"Ollama ({OLLAMA_MODEL} @ {OLLAMA_URL})" print(f"Feedback Analyzer on http://127.0.0.1:{PORT} [backend: {backend}]", flush=True) app.run(host="0.0.0.0", port=PORT, threaded=True)