root/MarketingTool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

deploy DOM search v8 update

Browse Source
This commit is contained in:
Marco0300
2026-09-06 00:46:28 +02:00
parent e92ffaa807
commit eeffaf4b6c
5 changed files with 317 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/server.py
+243 -9
View File
@@ -20,16 +20,17 @@ import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from html.parser import HTMLParser
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Any, Callable
from urllib.parse import parse_qs, quote_plus, unquote, urlparse
try:
from intelligence import domain_candidates, scan_website
from intelligence import domain_candidates, scan_website, _safe_url
from browser_investigator import direct_search_candidates, investigate_websites
except ImportError: # tests import the module as api.server
from api.intelligence import domain_candidates, scan_website
from api.intelligence import domain_candidates, scan_website, _safe_url
from api.browser_investigator import direct_search_candidates, investigate_websites
DB_PATH = os.environ.get("PROSPECT_DB", "/data/prospects.db")
@@ -85,6 +86,87 @@ def _saved_settings() -> dict[str, str]:
return {}
COMPANY_PROFILE_KEYS = {
"website_url": "COMPANY_WEBSITE_URL",
"name": "COMPANY_NAME",
"description": "COMPANY_DESCRIPTION",
"services": "COMPANY_SERVICES",
"packages": "COMPANY_PACKAGES",
"tone": "COMPANY_TONE",
"notes": "COMPANY_NOTES",
"website_content": "COMPANY_WEBSITE_CONTENT",
"website_title": "COMPANY_WEBSITE_TITLE",
"website_fetched_at": "COMPANY_WEBSITE_FETCHED_AT",
}
def company_profile_settings(saved: dict[str, str] | None = None) -> dict[str, Any]:
"""Return the owner's reusable company knowledge for AI prompts/UI."""
values = saved if saved is not None else _saved_settings()
profile = {field: str(values.get(key, ""))[:20_000] for field, key in COMPANY_PROFILE_KEYS.items()}
profile["configured"] = bool(profile.get("website_url") or profile.get("description") or profile.get("services") or profile.get("packages"))
return profile
class _CompanyPageParser(HTMLParser):
"""Small bounded text extractor used by Teach AI; no scripts are retained."""
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.title: list[str] = []
self.parts: list[str] = []
self._tag_stack: list[str] = []
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
tag = tag.lower()
self._tag_stack.append(tag)
def handle_endtag(self, tag: str) -> None:
tag = tag.lower()
for index in range(len(self._tag_stack) - 1, -1, -1):
if self._tag_stack[index] == tag:
del self._tag_stack[index:]
break
def handle_data(self, data: str) -> None:
if any(tag in {"script", "style", "noscript", "svg", "template"} for tag in self._tag_stack):
return
clean = re.sub(r"\s+", " ", data).strip()
if not clean:
return
if "title" in self._tag_stack:
self.title.append(clean)
self.parts.append(clean)
def fetch_company_page(url: str) -> dict[str, str]:
"""Fetch a public company page and extract a safe, bounded knowledge snapshot."""
normalized = normalize_url(url)
if not normalized or urlparse(normalized).scheme not in {"http", "https"} or not _safe_url(normalized):
raise ValueError("public_http_website_url_required")
request = urllib.request.Request(normalized, headers={"User-Agent": OUTBOUND_USER_AGENT, "Accept": "text/html,application/xhtml+xml"}, method="GET")
started = time.monotonic()
try:
with urllib.request.urlopen(request, timeout=20) as response:
content_type = str(response.headers.get("Content-Type") or "").lower()
body = response.read(1_000_000)
final_url = str(response.geturl() or normalized)
except urllib.error.HTTPError as exc:
raise RuntimeError(f"company_website_http_{exc.code}") from exc
except urllib.error.URLError as exc:
raise RuntimeError(f"company_website_unreachable:{exc.reason}") from exc
if content_type and "html" not in content_type and "xhtml" not in content_type:
raise RuntimeError("company_website_not_html")
parser = _CompanyPageParser()
parser.feed(body.decode("utf-8", "replace"))
text = " ".join(parser.parts)
# Keep enough context for packages/services while preventing a huge prompt.
text = re.sub(r"\s+", " ", text).strip()[:20_000]
title = " ".join(parser.title).strip()[:300]
record_log("ai", "company_website_learned", details={"url": _safe_log_url(final_url), "title": title, "characters": len(text), "duration_ms": int((time.monotonic() - started) * 1000)})
return {"website_url": final_url, "website_title": title, "website_content": text, "website_fetched_at": now()}
def _setting_int(saved: dict[str, str], key: str, env_key: str, default: int, minimum: int, maximum: int) -> int:
try:
value = int(saved.get(key, os.environ.get(env_key, str(default))))
@@ -331,6 +413,17 @@ def db() -> sqlite3.Connection:
details TEXT NOT NULL DEFAULT '{}',
error TEXT NOT NULL DEFAULT ''
)""")
conn.execute("""CREATE TABLE IF NOT EXISTS promotions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
terms TEXT NOT NULL DEFAULT '',
starts_at TEXT NOT NULL DEFAULT '',
ends_at TEXT NOT NULL DEFAULT '',
active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)""")
# Migrate databases created by v1-v5 without destroying the user's
# catalogue. SQLite permits adding nullable/defaulted columns in place.
@@ -371,6 +464,7 @@ def db() -> sqlite3.Connection:
"ai_unknowns": "TEXT NOT NULL DEFAULT '[]'",
"ai_evidence_ids": "TEXT NOT NULL DEFAULT '[]'",
"ai_follow_up": "TEXT NOT NULL DEFAULT ''",
"ai_promotion_id": "INTEGER",
"opportunity_score": "INTEGER NOT NULL DEFAULT 0",
"raw_json": "TEXT NOT NULL DEFAULT '{}'",
}
@@ -1322,7 +1416,7 @@ def scheduler_loop() -> None:
time.sleep(30)
def ai_generate_message(prospect: dict[str, Any], offer: str, tone: str = "warm", goal: str = "conversation") -> dict[str, Any]:
def ai_generate_message(prospect: dict[str, Any], offer: str, tone: str = "warm", goal: str = "conversation", promotion: dict[str, Any] | None = None) -> dict[str, Any]:
"""Generate an evidence-grounded, human-editable outreach draft.
``tone`` and ``goal`` are deliberately small, validated controls rather
@@ -1356,6 +1450,18 @@ def ai_generate_message(prospect: dict[str, Any], offer: str, tone: str = "warm"
goal_key = "conversation"
tone_instruction = tone_options[tone_key]
goal_instruction = goal_options[goal_key]
profile = company_profile_settings()
profile_for_prompt = {
"name": profile.get("name"),
"website_url": profile.get("website_url"),
"description": profile.get("description"),
"services": profile.get("services"),
"packages": profile.get("packages"),
"tone": profile.get("tone"),
"notes": profile.get("notes"),
"website_title": profile.get("website_title"),
"website_content": profile.get("website_content"),
}
evidence = {
"evidence_ids": evidence_ids,
"business_name": prospect.get("name"),
@@ -1374,12 +1480,18 @@ def ai_generate_message(prospect: dict[str, Any], offer: str, tone: str = "warm"
"public_emails": prospect.get("emails"),
"opportunity_score": prospect.get("opportunity_score"),
"score_reasons": _json_list(prospect.get("score_reasons")),
"company_profile": profile_for_prompt,
"selected_promotion": {
"name": promotion.get("name"), "description": promotion.get("description"),
"terms": promotion.get("terms"), "starts_at": promotion.get("starts_at"), "ends_at": promotion.get("ends_at"),
} if promotion else None,
}
system = (
"You are an expert B2B copywriter for a small South African web-hosting and website-design business. "
"Write one genuinely human first-contact email, not a mass-marketing template. The desired voice is "
f"{tone_instruction}; the goal is to {goal_instruction}. "
"Use only the stored evidence and the user's offer. Never invent a person's name, company activity, "
"Use only the stored prospect evidence, the owner's company profile, the selected promotion, and the user's offer. "
"The company profile describes the sender's business, not the prospect. Never invent a person's name, company activity, "
"price, service, problem, result, testimonial, urgency, authority, or link. Do not mention scraping, "
"scoring, hidden analysis, AI, or this instruction. Do not say you reviewed a website when no website "
"was found. If a scan is uncertain or failed, say that the public check could not confirm it instead of "
@@ -1395,7 +1507,7 @@ def ai_generate_message(prospect: dict[str, Any], offer: str, tone: str = "warm"
"unknowns, evidence_used, follow_up. subject and message are plain text. rationale is concise. "
"recommended_offer is a short summary of the user's offer. unknowns is an array of facts to confirm. "
"evidence_used is an array containing only supplied evidence IDs. follow_up is a short human review note, "
"not part of the email."
"not part of the email. If a selected promotion is supplied, mention it naturally only when it is active and its exact terms are supplied; never invent an expiry, discount, price, or eligibility rule."
)
prompt = json.dumps({"evidence": evidence, "user_offer_or_instruction": offer[:2000], "tone": tone_key, "goal": goal_key}, ensure_ascii=False)
headers = {"Authorization": "Bearer " + ai_api_key} if ai_api_key else {}
@@ -1458,6 +1570,7 @@ def ai_assess_prospect(prospect: dict[str, Any]) -> dict[str, Any]:
raise RuntimeError("nous_api_key_required_for_standalone_app")
if not runtime["base_url"] or not runtime["model"]:
raise RuntimeError("ai_not_configured_set_AI_BASE_URL_and_AI_MODEL")
profile = company_profile_settings()
evidence = {
"business": prospect.get("name"), "category": prospect.get("category"),
"address": prospect.get("address"), "phone": prospect.get("phone"),
@@ -1467,11 +1580,17 @@ def ai_assess_prospect(prospect: dict[str, Any]) -> dict[str, Any]:
"emails": prospect.get("emails"), "rating": prospect.get("rating"),
"review_count": prospect.get("review_count"), "opportunity_score": prospect.get("opportunity_score"),
"score_reasons": _json_list(prospect.get("score_reasons")),
"sender_company_profile": {
"name": profile.get("name"), "description": profile.get("description"),
"services": profile.get("services"), "packages": profile.get("packages"),
"website_url": profile.get("website_url"), "website_content": profile.get("website_content"),
},
}
system = (
"You qualify prospects for a South African web hosting and web design business. "
"Assess only the supplied public evidence. A high probability means the business appears active "
"and has a plausible hosting, domain, website repair, or redesign need. Do not invent facts, "
"and has a plausible need that matches the sender company's services. Use the sender company profile to judge fit, "
"but do not invent facts, "
"intent, budget, decision-maker names, or contact details. Return strict JSON with exactly these "
"keys: probability (integer 0-100), recommendation (one of contact, review, do_not_contact), "
"reasoning (one concise paragraph)."
@@ -1608,6 +1727,7 @@ class Handler(BaseHTTPRequestHandler):
counts["errors"] = error_count
return self.send_json(200, {"items": items, "total": total, "counts": counts})
if parsed.path == "/api/settings":
saved = _saved_settings()
ai_runtime = ai_runtime_settings()
models = nous_models() if ai_runtime["provider"] == "nous" else []
browser_runtime = browser_use_settings()
@@ -1623,6 +1743,7 @@ class Handler(BaseHTTPRequestHandler):
"browser_use": browser_runtime,
"email_collection": {"always_on": True, "public_only": True},
"limits": {"max_results": MAX_RESULTS, "website_scan_timeout_seconds": scan_timeout},
"company_profile": company_profile_settings(saved),
})
if parsed.path == "/api/prospects":
params = parse_qs(parsed.query)
@@ -1649,6 +1770,9 @@ class Handler(BaseHTTPRequestHandler):
return self.send_json(200, {"items": [dict(r) for r in conn.execute("SELECT * FROM suppressions ORDER BY id DESC")]})
if parsed.path == "/api/schedules":
return self.send_json(200, {"items": [dict(r) for r in conn.execute("SELECT * FROM schedules ORDER BY id DESC")]})
if parsed.path == "/api/promotions":
rows = conn.execute("SELECT * FROM promotions ORDER BY active DESC, updated_at DESC, id DESC").fetchall()
return self.send_json(200, {"items": [dict(row) for row in rows]})
match = re.fullmatch(r"/api/prospects/(\d+)/website-candidates", parsed.path)
if match:
prospect_id = int(match.group(1))
@@ -1731,11 +1855,13 @@ class Handler(BaseHTTPRequestHandler):
browser = payload.get("browser_use") if isinstance(payload.get("browser_use"), dict) else {}
maps = payload.get("maps") if isinstance(payload.get("maps"), dict) else {}
limits = payload.get("limits") if isinstance(payload.get("limits"), dict) else {}
company = payload.get("company_profile") if isinstance(payload.get("company_profile"), dict) else {}
conn = db()
saved_keys: list[str] = []
def save_setting(key: str, value: Any) -> None:
text_value = str(value).strip()[:500]
max_length = 20_000 if key.startswith("COMPANY_") else 500
text_value = str(value).strip()[:max_length]
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value,updated_at=excluded.updated_at", (key, text_value, now()))
saved_keys.append(key)
@@ -1766,6 +1892,13 @@ class Handler(BaseHTTPRequestHandler):
if "website_scan_timeout_seconds" in limits:
save_setting("WEBSITE_SCAN_TIMEOUT", max(2, min(60, int(limits.get("website_scan_timeout_seconds")))))
for field, key in COMPANY_PROFILE_KEYS.items():
# Website content/title/timestamp are written by the
# bounded Learn button, but accepting them here makes a
# full profile export/import round-trip cleanly.
if field in company:
save_setting(key, str(company.get(field) or "")[:20_000])
conn.commit()
conn.close()
record_log("action", "settings_updated", details={"keys": saved_keys})
@@ -1777,9 +1910,54 @@ class Handler(BaseHTTPRequestHandler):
"browser_use": browser_use_settings(),
"maps": gmaps_runtime_settings(),
"limits": {"website_scan_timeout_seconds": website_scan_timeout_seconds()},
"company_profile": company_profile_settings(_saved_settings()),
})
except Exception as exc:
return self.send_json(400, {"error": str(exc)})
if urlparse(self.path).path == "/api/company-profile/learn":
conn: sqlite3.Connection | None = None
try:
payload = self.read_json()
learned = fetch_company_page(str(payload.get("url") or "").strip())
conn = db()
for field, value in learned.items():
key = COMPANY_PROFILE_KEYS[field]
conn.execute("INSERT INTO settings(key,value,updated_at) VALUES(?,?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value,updated_at=excluded.updated_at", (key, str(value)[:20_000], now()))
conn.commit()
profile = company_profile_settings(_saved_settings())
return self.send_json(200, {"saved": True, "company_profile": profile})
except (ValueError, TypeError, json.JSONDecodeError) as exc:
return self.send_json(400, {"error": str(exc)})
except RuntimeError as exc:
return self.send_json(502, {"error": str(exc)})
except Exception as exc:
return self.send_json(502, {"error": "company_profile_learn_failed:" + str(exc)[:240]})
finally:
if conn is not None:
conn.close()
if self.path == "/api/promotions":
conn: sqlite3.Connection | None = None
try:
payload = self.read_json()
name = str(payload.get("name") or "").strip()[:160]
description = str(payload.get("description") or "").strip()[:2_000]
if not name or not description:
return self.send_json(400, {"error": "promotion_name_and_description_required"})
active = 0 if str(payload.get("active", True)).strip().lower() in {"0", "false", "no", "off"} else 1
stamp = now()
conn = db()
cur = conn.execute("INSERT INTO promotions(name,description,terms,starts_at,ends_at,active,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?)", (name, description, str(payload.get("terms") or "").strip()[:2_000], str(payload.get("starts_at") or "").strip()[:40], str(payload.get("ends_at") or "").strip()[:40], active, stamp, stamp))
conn.commit()
row = conn.execute("SELECT * FROM promotions WHERE id=?", (cur.lastrowid,)).fetchone()
record_log("action", "promotion_created", details={"promotion_id": int(cur.lastrowid), "name": name})
return self.send_json(201, dict(row))
except (ValueError, TypeError, json.JSONDecodeError) as exc:
return self.send_json(400, {"error": str(exc)})
except Exception as exc:
return self.send_json(400, {"error": str(exc)})
finally:
if conn is not None:
conn.close()
if urlparse(self.path).path == "/api/ai/test":
try:
runtime = ai_runtime_settings()
@@ -1900,8 +2078,23 @@ class Handler(BaseHTTPRequestHandler):
if not row: return self.send_json(404, {"error":"not_found"})
payload = self.read_json(); offer = str(payload.get("offer") or "").strip()
if not offer: return self.send_json(400, {"error":"offer_required"})
draft = ai_generate_message(dict(row), offer, tone=str(payload.get("tone") or "warm"), goal=str(payload.get("goal") or "conversation"))
stamp = now(); conn.execute("UPDATE prospects SET ai_subject=?,ai_message=?,ai_offer=?,ai_reasoning=?,ai_unknowns=?,ai_evidence_ids=?,ai_follow_up=?,ai_checked_at=? WHERE id=?", (draft["subject"],draft["message"],offer,draft["rationale"],json.dumps(draft.get("unknowns") or [], ensure_ascii=False),json.dumps(draft.get("evidence_used") or [], ensure_ascii=False),draft.get("follow_up") or "",stamp,int(message_match.group(1)))); conn.commit()
promotion = None
promotion_id = payload.get("promotion_id")
if promotion_id not in (None, "", 0, "0"):
try:
promotion_id = int(promotion_id)
except (TypeError, ValueError):
return self.send_json(400, {"error": "invalid_promotion_id"})
promotion_row = conn.execute("SELECT * FROM promotions WHERE id=?", (promotion_id,)).fetchone()
if not promotion_row:
return self.send_json(404, {"error": "promotion_not_found"})
if not promotion_row["active"]:
return self.send_json(400, {"error": "promotion_inactive"})
promotion = dict(promotion_row)
else:
promotion_id = None
draft = ai_generate_message(dict(row), offer, tone=str(payload.get("tone") or "warm"), goal=str(payload.get("goal") or "conversation"), promotion=promotion)
stamp = now(); conn.execute("UPDATE prospects SET ai_subject=?,ai_message=?,ai_offer=?,ai_promotion_id=?,ai_reasoning=?,ai_unknowns=?,ai_evidence_ids=?,ai_follow_up=?,ai_checked_at=? WHERE id=?", (draft["subject"],draft["message"],offer,promotion_id,draft["rationale"],json.dumps(draft.get("unknowns") or [], ensure_ascii=False),json.dumps(draft.get("evidence_used") or [], ensure_ascii=False),draft.get("follow_up") or "",stamp,int(message_match.group(1)))); conn.commit()
return self.send_json(200, {**draft,"ai_checked_at":stamp,"prospect_id":int(message_match.group(1))})
except RuntimeError as exc: return self.send_json(503, {"error":str(exc)})
except (ValueError, TypeError, json.JSONDecodeError) as exc: return self.send_json(400, {"error":str(exc)})
@@ -1941,6 +2134,37 @@ class Handler(BaseHTTPRequestHandler):
return self.send_json(400, {"error": str(exc)})
def do_PATCH(self) -> None:
promotion_match = re.fullmatch(r"/api/promotions/(\d+)", urlparse(self.path).path)
if promotion_match:
conn: sqlite3.Connection | None = None
try:
payload = self.read_json()
conn = db()
existing = conn.execute("SELECT * FROM promotions WHERE id=?", (int(promotion_match.group(1)),)).fetchone()
if not existing:
return self.send_json(404, {"error": "not_found"})
fields: dict[str, Any] = {}
for field in ("name", "description", "terms", "starts_at", "ends_at"):
if field in payload:
fields[field] = str(payload.get(field) or "").strip()[:2_000]
if "name" in fields and not fields["name"]:
return self.send_json(400, {"error": "promotion_name_required"})
if "description" in fields and not fields["description"]:
return self.send_json(400, {"error": "promotion_description_required"})
if "active" in payload:
fields["active"] = 0 if str(payload.get("active")).strip().lower() in {"0", "false", "no", "off"} else 1
fields["updated_at"] = now()
assignments = ",".join(f"{key}=?" for key in fields)
conn.execute("UPDATE promotions SET " + assignments + " WHERE id=?", (*fields.values(), int(promotion_match.group(1))))
conn.commit()
row = conn.execute("SELECT * FROM promotions WHERE id=?", (int(promotion_match.group(1)),)).fetchone()
record_log("action", "promotion_updated", details={"promotion_id": int(promotion_match.group(1)), "fields": list(fields)})
return self.send_json(200, dict(row))
except (ValueError, TypeError, json.JSONDecodeError) as exc:
return self.send_json(400, {"error": str(exc)})
finally:
if conn is not None:
conn.close()
candidate_match = re.fullmatch(r"/api/prospects/(\d+)/website-candidates/(\d+)", urlparse(self.path).path)
if candidate_match:
prospect_id, candidate_id = (int(candidate_match.group(1)), int(candidate_match.group(2)))
@@ -2038,6 +2262,16 @@ class Handler(BaseHTTPRequestHandler):
conn.close()
def do_DELETE(self) -> None:
promotion_match = re.fullmatch(r"/api/promotions/(\d+)", urlparse(self.path).path)
if promotion_match:
conn = db()
try:
cur = conn.execute("DELETE FROM promotions WHERE id=?", (int(promotion_match.group(1)),))
conn.commit()
record_log("action", "promotion_deleted", details={"promotion_id": int(promotion_match.group(1))})
return self.send_json(200, {"deleted": bool(cur.rowcount)})
finally:
conn.close()
match = re.fullmatch(r"/api/suppressions/(\d+)", urlparse(self.path).path)
if not match: return self.send_json(404, {"error":"not_found"})
conn=db()