feat: Add Chatwoot Agent Bot prototype and FAQ knowledge base

- Created chatwoot-agent-bot/ with Node.js webhook server
- Bot detects intent (greeting, billing, technical, features, account)
- Auto-responds from FAQ knowledge base or escalates to human
- FAQ-KB.md: Living knowledge base that grows with customer questions
- CHATWOOT-SETUP.md: Complete deployment and configuration guide
- Supports Telegram notifications on escalation
- Bot runs on port 3001, ready for Chatwoot webhook integration
This commit is contained in:
2026-04-01 16:26:05 -04:00
parent 7ba19752de
commit 5319bcd30b
1074 changed files with 456376 additions and 0 deletions

View File

@@ -0,0 +1,146 @@
#!/usr/bin/env python3
"""Google Analytics 4 - Direct JWT Authentication (No gcloud required)"""
import json
import urllib.request
from datetime import datetime, timedelta
from pathlib import Path
import subprocess
CONFIG_DIR = Path(__file__).parent.parent / "config"
GA_CREDENTIALS = CONFIG_DIR / "ga-credentials.json"
GA_PROPERTY_ID = "526394825"
def load_credentials():
"""Load service account credentials"""
with open(GA_CREDENTIALS) as f:
return json.load(f)
def get_jwt_token(creds):
"""Create and sign JWT for OAuth"""
import base64
import hashlib
# Check for PyJWT
try:
import jwt
from cryptography.hazmat.primitives import serialization
now = datetime.utcnow()
claims = {
"iss": creds['client_email'],
"sub": creds['client_email'],
"scope": "https://www.googleapis.com/auth/analytics.readonly",
"aud": creds['token_uri'],
"iat": now,
"exp": now + timedelta(hours=1)
}
private_key = creds['private_key']
token = jwt.encode(claims, private_key, algorithm="RS256")
return token
except ImportError:
return None
def get_access_token_with_jwt(creds):
"""Get OAuth token using JWT"""
jwt_token = get_jwt_token(creds)
if not jwt_token:
return None
body = {
"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
"assertion": jwt_token
}
req = urllib.request.Request(
creds['token_uri'],
data=json.dumps(body).encode(),
headers={"Content-Type": "application/json"},
method="POST"
)
try:
with urllib.request.urlopen(req, timeout=30) as r:
data = json.loads(r.read().decode())
return data.get('access_token')
except Exception as e:
print(f"Token error: {e}")
return None
def get_access_token_with_curl(creds):
"""Get token using curl"""
try:
result = subprocess.run(
[
"curl", "-s", "-X", "POST",
creds['token_uri'],
"-H", "Content-Type: application/x-www-form-urlencoded",
"-d", f"grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer",
"--data-urlencode", f"assertion=<(echo 'JWT_PLACEHOLDER')"
],
capture_output=True,
text=True,
timeout=10
)
return None # Complex JWT signing needed
except:
return None
def query_ga4_direct():
"""Query GA4 using Python requests if available"""
try:
creds = load_credentials()
# Method using google-analytics-data library
try:
from google.analytics.data import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import RunReportRequest, DateRange, Metric, Dimension
client = BetaAnalyticsDataClient.from_service_account_json(str(GA_CREDENTIALS))
request = RunReportRequest(
property=f"properties/{GA_PROPERTY_ID}",
date_ranges=[DateRange(start_date="1daysAgo", end_date="today")],
metrics=[
Metric(name="sessions"),
Metric(name="activeUsers"),
Metric(name="newUsers")
]
)
response = client.run_report(request)
total_sessions = sum(int(r.metric_values[0].value) for r in response.rows)
total_users = sum(int(r.metric_values[1].value) for r in response.rows)
new_users = sum(int(r.metric_values[2].value) for r in response.rows)
return {
"sessions": total_sessions,
"activeUsers": total_users,
"newUsers": new_users,
"success": True
}
except ImportError:
return {"error": "google-analytics-data library required", "install": "pip install google-analytics-data", "success": False}
except Exception as e:
return {"error": str(e), "success": False}
if __name__ == "__main__":
print("🚀 Testing GA4 Direct Connection...")
result = query_ga4_direct()
if result.get('success'):
print(f"""
📊 GA4 Traffic Data (Last 24h):
✅ Sessions: {result.get('sessions', 'N/A'):,}
✅ Active Users: {result.get('activeUsers', 'N/A'):,}
✅ New Users: {result.get('newUsers', 'N/A'):,}
""")
else:
print(f"❌ Error: {result.get('error')}")
print(f"📦 Install: {result.get('install', 'N/A')}")
print("")
print("Quick fix:")
print(" pip install google-analytics-data")