#!/usr/bin/env python3 """ Complaint & Issue Tracker — No More Chores ========================================== Monitors Slack channels for customer complaint keywords and logs them to GoHighLevel CRM. USAGE: python3 scripts/complaint-tracker.py # Poll channels, process new complaints python3 scripts/complaint-tracker.py --digest # Post daily digest to #emilys-team python3 scripts/complaint-tracker.py --resolve ID # Mark a complaint resolved python3 scripts/complaint-tracker.py --status # Print open complaints to stdout STATE FILE: scripts/complaint-state.json SECRETS: ~/.openclaw/secrets/slack-token.txt ~/.openclaw/secrets/ghl-v2-token.txt CHANNELS MONITORED: emilys-team (C02NFPKNZ1U) vadims-team (C01U2947AUS) customer-feedback (C0AJQTVST1B) COMPLAINT KEYWORDS (case-insensitive): complaint, feedback, unhappy, disappointed, reclean, re-clean, discount, refund, upset, angry, poor quality, not satisfied, issue, problem """ import json import os import re import sys import time import hashlib import urllib.request import urllib.parse import urllib.error from datetime import datetime, timezone, timedelta from typing import Optional # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) STATE_FILE = os.path.join(SCRIPT_DIR, "complaint-state.json") SECRETS_DIR = os.path.expanduser("~/.openclaw/secrets") SLACK_TOKEN_FILE = os.path.join(SECRETS_DIR, "slack-token.txt") GHL_TOKEN_FILE = os.path.join(SECRETS_DIR, "ghl-v2-token.txt") GHL_LOCATION_ID = "BvrCS522k5EuY6TbZQsw" GHL_BASE = "https://services.leadconnectorhq.com" CHANNELS = { "emilys-team": "C02NFPKNZ1U", "vadims-team": "C01U2947AUS", "customer-feedback": "C0AJQTVST1B", } FEEDBACK_CHANNEL = "C0AJQTVST1B" DIGEST_CHANNEL = "C02NFPKNZ1U" COMPLAINT_KEYWORDS = [ "complaint", "feedback", "unhappy", "disappointed", "reclean", "re-clean", "discount", "refund", "upset", "angry", "poor quality", "not satisfied", "issue", "problem", ] MAX_MESSAGE_LEN = 500 # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- def log(msg: str) -> None: ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{ts}] {msg}", file=sys.stderr) # --------------------------------------------------------------------------- # Secrets # --------------------------------------------------------------------------- def read_secret(path: str) -> str: try: with open(path) as f: return f.read().strip() except FileNotFoundError: log(f"ERROR: Secret file not found: {path}") sys.exit(1) # --------------------------------------------------------------------------- # State management # --------------------------------------------------------------------------- def load_state() -> dict: if os.path.exists(STATE_FILE): try: with open(STATE_FILE) as f: return json.load(f) except (json.JSONDecodeError, OSError) as e: log(f"WARN: Could not read state file ({e}), starting fresh") return {"last_checked": {}, "complaints": []} def save_state(state: dict) -> None: try: with open(STATE_FILE, "w") as f: json.dump(state, f, indent=2) except OSError as e: log(f"ERROR: Could not save state: {e}") # --------------------------------------------------------------------------- # HTTP helpers # --------------------------------------------------------------------------- _USER_AGENT = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) HarveyBot/1.0" def _do_request(req: urllib.request.Request, retry_on_5xx: bool = True) -> Optional[dict]: """Execute a request, return parsed JSON or None on error.""" # Cloudflare blocks Python-urllib default UA (error 1010) if "User-Agent" not in req.headers: req.add_header("User-Agent", _USER_AGENT) try: with urllib.request.urlopen(req, timeout=15) as resp: return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: body = e.read().decode(errors="replace") if retry_on_5xx and e.code >= 500: log(f"WARN: HTTP {e.code} — retrying in 2s…") time.sleep(2) return _do_request(req, retry_on_5xx=False) log(f"ERROR: HTTP {e.code} for {req.full_url}: {body[:200]}") return None except urllib.error.URLError as e: log(f"ERROR: URL error for {req.full_url}: {e.reason}") return None except Exception as e: log(f"ERROR: Unexpected error: {e}") return None def slack_get(token: str, endpoint: str, params: dict) -> Optional[dict]: qs = urllib.parse.urlencode(params) url = f"https://slack.com/api/{endpoint}?{qs}" req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"}) return _do_request(req) def slack_post(token: str, payload: dict) -> Optional[dict]: data = json.dumps(payload).encode() req = urllib.request.Request( "https://slack.com/api/chat.postMessage", data=data, headers={ "Authorization": f"Bearer {token}", "Content-Type": "application/json", }, ) return _do_request(req) def ghl_get(token: str, path: str, params: dict = {}) -> Optional[dict]: qs = urllib.parse.urlencode(params) url = f"{GHL_BASE}{path}{'?' + qs if qs else ''}" req = urllib.request.Request( url, headers={ "Authorization": f"Bearer {token}", "Version": "2021-07-28", }, ) return _do_request(req) def ghl_post(token: str, path: str, payload: dict) -> Optional[dict]: data = json.dumps(payload).encode() req = urllib.request.Request( f"{GHL_BASE}{path}", data=data, headers={ "Authorization": f"Bearer {token}", "Version": "2021-07-28", "Content-Type": "application/json", }, ) return _do_request(req) # --------------------------------------------------------------------------- # Complaint keyword detection # --------------------------------------------------------------------------- # Phrases that contain keywords but are NOT complaints _FALSE_POSITIVE_PHRASES = [ "no problem", "not a problem", "no issue", "not an issue", "no complaints", "not a complaint", "any feedback", "provide feedback", "leave feedback", "customer feedback —", # our own digest headers ] def find_keyword(text: str) -> Optional[str]: """Return the first matching complaint keyword, or None.""" lower = text.lower() # Check for false positive phrases first for phrase in _FALSE_POSITIVE_PHRASES: if phrase in lower: # Only skip if this is the ONLY keyword context stripped = lower.replace(phrase, "") has_other = any(kw in stripped for kw in COMPLAINT_KEYWORDS) if not has_other: return None for kw in COMPLAINT_KEYWORDS: if kw in lower: return kw return None # --------------------------------------------------------------------------- # Customer name extraction # --------------------------------------------------------------------------- # Words that look like names but aren't — common false positives _NAME_BLACKLIST = { "the", "that", "this", "sure", "not", "our", "their", "they", "she", "her", "his", "him", "are", "was", "has", "had", "been", "all", "any", "who", "how", "what", "when", "just", "very", "some", "more", "also", "only", "back", "here", "there", "about", "from", "with", "will", "would", "could", "should", "please", "thanks", "thank", "sorry", "maybe", "really", "still", "much", "most", "every", "even", "well", "let", "feedback", "complaint", "issue", "problem", "discount", "unknown", "customer", "clean", "cleaning", "reclean", "today", "yesterday", "tomorrow", "monday", "tuesday", "wednesday", "thursday", "friday", "saturday", "sunday", "january", "february", "march", "april", "june", "july", "august", "september", "october", "november", "december", "complaint", "detected", "resolved", "open", "closed", } _NAME_PATTERNS = [ # Slack bold name: *Firstname Lastname* or *Firstname* re.compile(r'\*([A-Z][a-z]+(?: [A-Z][a-z]+)?)\*'), # "customer: Name" or "customer Name" re.compile(r'\bcustomer[:\s]+\*?([A-Z][a-z]+(?: [A-Z][a-z]+)?)\*?', re.IGNORECASE), # "client: Name" or "client Name" re.compile(r'\bclient[:\s]+\*?([A-Z][a-z]+(?: [A-Z][a-z]+)?)\*?', re.IGNORECASE), # "from Name" (as in "from Julie") re.compile(r'\bfrom\s+\*?([A-Z][a-z]+(?: [A-Z][a-z]+)?)\*?'), # "for Name" (as in "reclean for Sarah") re.compile(r'\bfor\s+\*?([A-Z][a-z]+(?: [A-Z][a-z]+)?)\*?'), ] def _is_valid_name(name: str) -> bool: """Filter out false positive name extractions.""" parts = name.lower().split() # Reject if any part is blacklisted if any(p in _NAME_BLACKLIST for p in parts): return False # Name should be 2-40 chars if len(name) < 2 or len(name) > 40: return False return True def extract_customer_name(text: str) -> Optional[str]: for pat in _NAME_PATTERNS: for m in pat.finditer(text): name = m.group(1).strip() if _is_valid_name(name): return name return None # --------------------------------------------------------------------------- # Complaint ID generation # --------------------------------------------------------------------------- def make_complaint_id(channel_id: str, ts: str) -> str: raw = f"{channel_id}:{ts}" return hashlib.md5(raw.encode()).hexdigest()[:8] # --------------------------------------------------------------------------- # GHL helpers # --------------------------------------------------------------------------- def search_ghl_contact(token: str, name: str) -> Optional[dict]: """Search GHL contacts by name; return first hit or None.""" resp = ghl_get(token, "/contacts/", {"locationId": GHL_LOCATION_ID, "query": name}) if not resp: return None contacts = resp.get("contacts", []) if contacts: return contacts[0] return None def create_ghl_note(token: str, contact_id: str, body: str) -> Optional[str]: """Create a note on a GHL contact; return note id or None.""" resp = ghl_post(token, f"/contacts/{contact_id}/notes", {"body": body}) if resp: return resp.get("id") or resp.get("note", {}).get("id") return None def ghl_contact_url(contact_id: str) -> str: return f"https://app.gohighlevel.com/contacts/{contact_id}" # --------------------------------------------------------------------------- # Slack permalink # --------------------------------------------------------------------------- def get_permalink(token: str, channel_id: str, ts: str) -> Optional[str]: resp = slack_get(token, "chat.getPermalink", {"channel": channel_id, "message_ts": ts}) if resp and resp.get("ok"): return resp.get("permalink") return None # --------------------------------------------------------------------------- # Poll mode # --------------------------------------------------------------------------- def poll_channels(slack_token: str, ghl_token: str, state: dict) -> None: log("=== Polling channels for complaints ===") now_ts = str(datetime.now(timezone.utc).timestamp()) for channel_name, channel_id in CHANNELS.items(): oldest = state["last_checked"].get(channel_id, "0") log(f"Checking #{channel_name} (oldest={oldest})") resp = slack_get(slack_token, "conversations.history", { "channel": channel_id, "oldest": oldest, "limit": 100, }) if not resp or not resp.get("ok"): log(f"WARN: Could not fetch #{channel_name}: {resp}") continue messages = resp.get("messages", []) log(f" {len(messages)} new messages") for msg in messages: text = msg.get("text", "") ts = msg.get("ts", "") # Skip bot messages (including our own notifications) if msg.get("bot_id") or msg.get("subtype") == "bot_message": continue keyword = find_keyword(text) if not keyword: continue complaint_id = make_complaint_id(channel_id, ts) # Skip if already tracked if any(c["id"] == complaint_id for c in state["complaints"]): continue log(f" Complaint detected (kw='{keyword}'): {text[:80]}") customer_name = extract_customer_name(text) or "Unknown Customer" permalink = get_permalink(slack_token, channel_id, ts) snippet = text[:MAX_MESSAGE_LEN] created_at = datetime.now(timezone.utc).isoformat() date_str = datetime.now(timezone.utc).strftime("%Y-%m-%d") # GHL: search contact + create note ghl_contact_id = None ghl_note_id = None ghl_link = None if customer_name != "Unknown Customer": contact = search_ghl_contact(ghl_token, customer_name) if contact: ghl_contact_id = contact.get("id") note_body = ( f"Complaint {date_str}\n\n" f"Channel: #{channel_name}\n" f"Keyword matched: {keyword}\n" f"Permalink: {permalink or 'N/A'}\n\n" f"Message:\n{text}" ) ghl_note_id = create_ghl_note(ghl_token, ghl_contact_id, note_body) ghl_link = ghl_contact_url(ghl_contact_id) log(f" GHL note created for contact {ghl_contact_id}") else: log(f" No GHL contact found for '{customer_name}'") # Save to state complaint = { "id": complaint_id, "customer_name": customer_name, "channel": channel_name, "channel_id": channel_id, "slack_ts": ts, "slack_permalink": permalink, "keyword_matched": keyword, "message_text": snippet, "status": "open", "created_at": created_at, "ghl_contact_id": ghl_contact_id, "ghl_note_id": ghl_note_id, } state["complaints"].append(complaint) # Notify #customer-feedback ghl_line = f"\n• GHL: {ghl_link}" if ghl_link else "" permalink_line = f"\n• Slack: {permalink}" if permalink else "" notify_text = ( f":rotating_light: *Complaint Detected* — #{channel_name}\n" f"• Customer: *{customer_name}*\n" f"• Keyword: `{keyword}`\n" f"• ID: `{complaint_id}`\n" f"• Snippet: _{snippet[:200]}_" f"{ghl_line}" f"{permalink_line}" ) slack_post(slack_token, {"channel": FEEDBACK_CHANNEL, "text": notify_text}) # Update last_checked to now state["last_checked"][channel_id] = now_ts save_state(state) log("=== Poll complete ===") # --------------------------------------------------------------------------- # Digest mode # --------------------------------------------------------------------------- def post_digest(slack_token: str, state: dict) -> None: log("=== Posting daily digest ===") open_complaints = [c for c in state["complaints"] if c["status"] == "open"] cutoff = datetime.now(timezone.utc) - timedelta(days=7) recently_resolved = [ c for c in state["complaints"] if c["status"] == "resolved" and datetime.fromisoformat(c.get("resolved_at", c["created_at"])) >= cutoff ] if not open_complaints: text = ":white_check_mark: No open complaints — great job team!" else: lines = [f":rotating_light: *Daily Complaint Digest* — {datetime.now().strftime('%Y-%m-%d')}\n"] lines.append(f"*{len(open_complaints)} open complaint(s):*") for c in open_complaints: date = c["created_at"][:10] snippet = c["message_text"][:120].replace("\n", " ") pl = f" — <{c['slack_permalink']}|view>" if c.get("slack_permalink") else "" lines.append( f"• `{c['id']}` *{c['customer_name']}* | {date} | #{c['channel']}{pl}\n" f" _{snippet}_" ) if recently_resolved: lines.append(f"\n:white_check_mark: *{len(recently_resolved)} resolved* in the last 7 days") text = "\n".join(lines) result = slack_post(slack_token, {"channel": DIGEST_CHANNEL, "text": text}) if result and result.get("ok"): log("Digest posted successfully") else: log(f"WARN: Digest post may have failed: {result}") # --------------------------------------------------------------------------- # Resolve mode # --------------------------------------------------------------------------- def resolve_complaint(slack_token: str, ghl_token: str, state: dict, complaint_id: str) -> None: log(f"=== Resolving complaint {complaint_id} ===") found = None for c in state["complaints"]: if c["id"] == complaint_id: found = c break if not found: log(f"ERROR: Complaint {complaint_id} not found") print(f"Complaint '{complaint_id}' not found in state.", file=sys.stderr) return if found["status"] == "resolved": log(f"WARN: Complaint {complaint_id} is already resolved") found["status"] = "resolved" found["resolved_at"] = datetime.now(timezone.utc).isoformat() # Attempt to append resolution to GHL note if found.get("ghl_contact_id") and found.get("ghl_note_id"): resolved_note = ( f"RESOLVED {found['resolved_at'][:10]}\n\n" f"Original complaint ID: {complaint_id}\n" f"Customer: {found['customer_name']}\n" f"Channel: #{found['channel']}" ) create_ghl_note(ghl_token, found["ghl_contact_id"], resolved_note) log("GHL resolution note created") save_state(state) # Post to #customer-feedback pl_line = f"\n• Slack: {found['slack_permalink']}" if found.get("slack_permalink") else "" text = ( f":white_check_mark: *Complaint Resolved* — `{complaint_id}`\n" f"• Customer: *{found['customer_name']}*\n" f"• Channel: #{found['channel']}\n" f"• Opened: {found['created_at'][:10]}" f"{pl_line}" ) slack_post(slack_token, {"channel": FEEDBACK_CHANNEL, "text": text}) log(f"Complaint {complaint_id} marked resolved") # --------------------------------------------------------------------------- # Status mode # --------------------------------------------------------------------------- def print_status(state: dict) -> None: open_complaints = [c for c in state["complaints"] if c["status"] == "open"] if not open_complaints: print("No open complaints.") return print(f"{len(open_complaints)} open complaint(s):\n") for c in open_complaints: print(f" ID: {c['id']}") print(f" Customer: {c['customer_name']}") print(f" Channel: #{c['channel']}") print(f" Date: {c['created_at'][:10]}") print(f" Keyword: {c['keyword_matched']}") print(f" Snippet: {c['message_text'][:120]}") if c.get("slack_permalink"): print(f" Slack: {c['slack_permalink']}") if c.get("ghl_contact_id"): print(f" GHL: {ghl_contact_url(c['ghl_contact_id'])}") print() # --------------------------------------------------------------------------- # Entrypoint # --------------------------------------------------------------------------- def main() -> None: args = sys.argv[1:] state = load_state() # --status: no API calls needed if "--status" in args: print_status(state) return slack_token = read_secret(SLACK_TOKEN_FILE) ghl_token = read_secret(GHL_TOKEN_FILE) if "--digest" in args: post_digest(slack_token, state) return if "--resolve" in args: idx = args.index("--resolve") if idx + 1 >= len(args): print("Usage: complaint-tracker.py --resolve ", file=sys.stderr) sys.exit(1) complaint_id = args[idx + 1] resolve_complaint(slack_token, ghl_token, state, complaint_id) return # Default: poll poll_channels(slack_token, ghl_token, state) if __name__ == "__main__": main()