#!/usr/bin/env python3 """ ReviewPilot — the review-response desk for local businesses. Paste or upload Google Business Profile reviews, and ReviewPilot tags each by sentiment, drafts a personalized reply in your tone of voice, and produces a monthly reputation report. $19/mo, Bitcoin via BTCPay, no accounts, no tracking. LLM calls route to nightmare (.128) Ollama — never runs a model on this CT. """ import os, json, time, sqlite3, urllib.request, urllib.parse, ssl, csv, io from flask import Flask, request, render_template_string, jsonify, send_file, redirect # --------------------------------------------------------------------------- # config # --------------------------------------------------------------------------- BASE = "/opt/reviewpilot" DB = os.path.join(BASE, "data", "reviews.db") os.makedirs(os.path.dirname(DB), exist_ok=True) # LLM (nightmare Ollama) LLM = "http://10.30.20.128:11434" MODEL = "qwen3.8long" # BTCPay (store created by Orion pipeline) BTCPAY_API = "https://10.30.20.140" BTCPAY_PUBLIC = "https://btcpay.thetempleofdoom.com" BTCPAY_STORE_ID = "3optSJe1MBS1ykJZ2ym5jj26Edotqq5pTR7R6GdRAqUb" BTCPAY_API_KEY = "653a8866fdbf7c17d3e1c67d50071cfd28c10f2c" PRICE_SATS = 1900 # $19/mo @ ~100k sats/BTC — flat monthly _CTX = ssl.create_default_context() _CTX.check_hostname = False _CTX.verify_mode = ssl.CERT_NONE app = Flask(__name__) # --------------------------------------------------------------------------- # db # --------------------------------------------------------------------------- def db(): c = sqlite3.connect(DB) c.row_factory = sqlite3.Row return c def init_db(): with db() as c: c.execute("""CREATE TABLE IF NOT EXISTS reviews ( id INTEGER PRIMARY KEY AUTOINCREMENT, reviewer TEXT, rating INTEGER, text TEXT, sentiment TEXT, reply TEXT, created_at REAL )""") c.execute("""CREATE TABLE IF NOT EXISTS subs ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT, invoice TEXT, status TEXT, created_at REAL )""") init_db() # --------------------------------------------------------------------------- # LLM helper (nightmare Ollama) # --------------------------------------------------------------------------- def _ollama(prompt, system=""): body = json.dumps({"model": MODEL, "prompt": prompt, "system": system, "stream": False, "options": {"temperature": 0.7}}).encode() req = urllib.request.Request(LLM + "/api/generate", data=body, headers={"Content-Type": "application/json"}) try: r = urllib.request.urlopen(req, timeout=90) return json.loads(r.read()).get("response", "").strip() except Exception as e: return None def _sentiment(text): """LLM sentiment, with a lenient parse + keyword fallback so the app never hangs.""" out = _ollama( 'Classify this customer review as exactly one word: positive, negative, or neutral.\n\n' f'Review: "{text}"\n\nAnswer:') if out: w = out.lower().strip().split()[0] if out.strip() else "" if w.startswith("neg") or "negative" in out.lower(): return "negative" if w.startswith("pos") or "positive" in out.lower(): return "positive" if w.startswith("neu") or "neutral" in out.lower(): return "neutral" # keyword fallback neg = ("terrible", "awful", "bad", "worst", "rude", "slow", "overpriced", "disappointed", "never", "avoid", "scam", "horrible", "poor", "unprofessional", "refund", "wrong", "waited", "late", "mistake", "error", "worse", "unhappy", "waste", "joke", "lied") pos = ("great", "amazing", "love", "best", "excellent", "friendly", "fast", "awesome", "wonderful", "recommend", "perfect", "fantastic", "helpful", "nice", "delicious", "clean", "professional", "quick", "favorite") t = text.lower() if any(w in t for w in neg): return "negative" if any(w in t for w in pos): return "positive" return "neutral" def _reply(reviewer, rating, text, sentiment, tone, business="our business"): sys = ("You are a customer-service copywriter. Write a short, genuine, personalized reply " "to a customer review. 2-3 sentences. No preamble, no sign-off, no emojis unless the " "tone calls for it. Address the reviewer's specific points.") prompt = (f"Reviewer: {reviewer or 'a customer'}\nRating: {rating}/5\nSentiment: {sentiment}\n" f'Tone: {tone}\nReview: "{text}"\n\nReply:') out = _ollama(prompt, system=sys) if out: return out # fallback reply if sentiment == "negative": return f"Thank you for your feedback — we're sorry to hear this and would love to make it right. Please reach out so we can help." if sentiment == "positive": return f"Thank you for the kind words! We're thrilled you had a great experience and hope to see you again soon." return f"Thanks for taking the time to leave a review — we appreciate your feedback." # --------------------------------------------------------------------------- # review parsing # --------------------------------------------------------------------------- def parse_reviews(text): """Accept 'rating|reviewer|text' or 'reviewer|text' or bare text, one per line.""" reviews = [] for line in text.splitlines(): line = line.strip() if not line: continue parts = [p.strip() for p in line.split("|")] if len(parts) >= 3 and parts[0].isdigit(): rating, reviewer, rtext = int(parts[0]), parts[1], "|".join(parts[2:]) elif len(parts) == 2 and parts[0].isdigit(): rating, reviewer, rtext = int(parts[0]), "", parts[1] elif len(parts) >= 2 and not parts[0].isdigit(): rating, reviewer, rtext = 5, parts[0], "|".join(parts[1:]) else: rating, reviewer, rtext = 5, "", parts[0] rating = max(1, min(5, rating)) if rtext: reviews.append({"reviewer": reviewer, "rating": rating, "text": rtext}) return reviews def parse_csv(data): """Accept Google Business Profile CSV (Reviewer name, Star rating, Review text).""" reviews = [] reader = csv.DictReader(io.StringIO(data)) for row in reader: rtext = (row.get("Review text") or row.get("review text") or row.get("text") or "").strip() if not rtext: continue reviewer = (row.get("Reviewer name") or row.get("reviewer") or row.get("Name") or "").strip() try: rating = int(float(row.get("Star rating") or row.get("rating") or row.get("Rating") or 5)) except (ValueError, TypeError): rating = 5 reviews.append({"reviewer": reviewer, "rating": max(1, min(5, rating)), "text": rtext}) return reviews # --------------------------------------------------------------------------- # report # --------------------------------------------------------------------------- def report_stats(reviews): if not reviews: return {} n = len(reviews) avg = sum(r["rating"] for r in reviews) / n sent = {"positive": 0, "negative": 0, "neutral": 0} for r in reviews: sent[r["sentiment"]] += 1 # keyword themes (top repeated words, stopword-filtered) stop = set("the a an and or but to of in for on with we you our your they their i my me it its is are was were be been this that have has had very really just not so".split()) words = {} for r in reviews: for w in r["text"].lower().split(): w = w.strip(".,!?()\"'") if len(w) > 3 and w not in stop: words[w] = words.get(w, 0) + 1 themes = [w for w, _ in sorted(words.items(), key=lambda x: -x[1])[:8]] return { "count": n, "avg": round(avg, 1), "positive": sent["positive"], "negative": sent["negative"], "neutral": sent["neutral"], "themes": themes, "response_rate": round(sum(1 for r in reviews if r["reply"]) / n * 100), } # --------------------------------------------------------------------------- # routes # --------------------------------------------------------------------------- LANDING = r""" ReviewPilot — Reply to every review in minutes
$19/mo · Bitcoin only

Reply to every review
in minutes, not hours.

Paste your Google Business Profile reviews. ReviewPilot tags sentiment, drafts a personalized reply in your tone of voice, and builds your monthly reputation report.

3×faster replies
100%reviews answered
PDFreputation report
₿no account

Analyze your reviews

🎯 Sentiment auto-tag

Every review tagged positive, negative, or neutral — no manual sorting.

✍️ Personalized replies

Each reply addresses the reviewer's actual words, not a canned template.

🎭 Your tone of voice

Friendly, professional, playful, empathetic, or witty — consistently.

📊 Reputation report

Rating trend, response rate, and keyword themes in a clean monthly PDF.

""" RESULT_TMPL = r""" Replies — ReviewPilot

Your drafted replies

{{ business }} · {{ tone }} tone · {{ count }} reviews · view report →
{% for r in reviews %} {% endfor %}
ReviewerRatingSentimentOriginal reviewDrafted reply
{{ r.reviewer or '—' }} {{ '★' * r.rating }}{{ '☆' * (5 - r.rating) }} {{ r.sentiment }} {{ r.text }}
{{ r.reply }}
Analyze more →
""" REPORT_TMPL = r""" Reputation Report — ReviewPilot

Reputation Report

{{ business }} · generated {{ now }}
{{ s.count }}
total reviews
{{ s.avg }} ★
average rating
{{ s.response_rate }}%
response rate

Sentiment breakdown

Positive — {{ s.positive }}
Neutral — {{ s.neutral }}
Negative — {{ s.negative }}

What customers talk about

{% for t in s.themes %}{{ t }}{% endfor %}
← Back to reviews
""" @app.route("/") def index(): return LANDING @app.route("/health") def health(): return jsonify({"status": "ok", "app": "reviewpilot"}) @app.route("/analyze", methods=["POST"]) def analyze(): business = request.form.get("business", "our business").strip() tone = request.form.get("tone", "Friendly").strip() reviews = parse_reviews(request.form.get("reviews", "")) csv_file = request.files.get("csv") if csv_file and csv_file.filename: reviews += parse_csv(csv_file.read().decode("utf-8", "ignore")) if not reviews: return "No reviews found — paste text or upload a CSV.", 400 results = [] for r in reviews: r["sentiment"] = _sentiment(r["text"]) r["reply"] = _reply(r["reviewer"], r["rating"], r["text"], r["sentiment"], tone, business) with db() as c: c.execute("INSERT INTO reviews (reviewer, rating, text, sentiment, reply, created_at) " "VALUES (?,?,?,?,?,?)", (r["reviewer"], r["rating"], r["text"], r["sentiment"], r["reply"], time.time())) results.append(r) return render_template_string(RESULT_TMPL, reviews=results, business=business, tone=tone, count=len(results)) @app.route("/report") def report(): with db() as c: rows = c.execute("SELECT * FROM reviews ORDER BY created_at DESC LIMIT 500").fetchall() reviews = [dict(r) for r in rows] s = report_stats(reviews) return render_template_string(REPORT_TMPL, s=s, business="your business", now=time.strftime("%B %Y")) @app.route("/upgrade") def upgrade(): return ("

ReviewPilot Pro — $19/mo

Full feature access, unlimited reviews and locations. " "Pay on-chain Bitcoin.

") if __name__ == "__main__": app.run(host="127.0.0.1", port=5000)