MarketingTool/apps/api/app/main.py

from __future__ import annotations

import argparse
import json
import os
import sqlite3
import sys
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse

if __package__ in (None, ""):
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    from app.domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, score_business
else:
    from .domain import deduplication_key, deduplicate_businesses, is_suppressed, normalize_business, score_business

ORGANIZATION_ID = "demo-tenant"
SCHEMA = Path(__file__).resolve().parents[1] / "schema.sql"


def connect(db_path: str) -> sqlite3.Connection:
    db = sqlite3.connect(db_path)
    db.row_factory = sqlite3.Row
    db.execute("PRAGMA foreign_keys = ON")
    db.executescript(SCHEMA.read_text())
    return db


def row_json(row: sqlite3.Row) -> dict:
    result = dict(row)
    result["score_factors"] = json.loads(result.pop("score_factors", "[]"))
    return result


def existing_businesses(db: sqlite3.Connection) -> list[dict]:
    return [row_json(r) for r in db.execute("SELECT * FROM businesses WHERE organization_id = ? ORDER BY id", (ORGANIZATION_ID,))]


def suppression_rows(db: sqlite3.Connection) -> list[dict]:
    return [dict(r) for r in db.execute("SELECT kind, value FROM suppressions WHERE organization_id = ?", (ORGANIZATION_ID,))]


class ApiHandler(BaseHTTPRequestHandler):
    server_version = "ProspectPlatform/0.1"

    def send_json(self, status: int, payload: dict | list):
        body = json.dumps(payload, sort_keys=True).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Access-Control-Allow-Origin", os.environ.get("CORS_ORIGINS", "http://localhost:8080"))
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def read_json(self) -> dict:
        try:
            length = int(self.headers.get("Content-Length", "0"))
            value = json.loads(self.rfile.read(length) or b"{}")
            return value if isinstance(value, dict) else {}
        except (ValueError, json.JSONDecodeError):
            return {}

    def db(self):
        return connect(self.server.db_path)

    def do_OPTIONS(self):
        self.send_response(204)
        self.send_header("Access-Control-Allow-Origin", os.environ.get("CORS_ORIGINS", "http://localhost:8080"))
        self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
        self.send_header("Access-Control-Allow-Headers", "Content-Type")
        self.end_headers()

    def do_GET(self):
        parsed = urlparse(self.path)
        path = parsed.path.rstrip("/")
        if path == "/api/v1/health/live":
            return self.send_json(200, {"status": "ok", "organization_id": ORGANIZATION_ID})
        db = self.db()
        try:
            if path == "/api/v1/dashboard/summary":
                row = db.execute("SELECT COUNT(*) AS businesses, COALESCE(AVG(score), 0) AS average_score FROM businesses WHERE organization_id = ?", (ORGANIZATION_ID,)).fetchone()
                return self.send_json(200, {"organization_id": ORGANIZATION_ID, "businesses": row["businesses"], "average_score": round(row["average_score"], 2), "suppressed": db.execute("SELECT COUNT(*) FROM suppressions WHERE organization_id = ?", (ORGANIZATION_ID,)).fetchone()[0]})
            if path == "/api/v1/businesses":
                query = parse_qs(parsed.query).get("q", [""])[0].strip()
                if query:
                    like = f"%{query}%"
                    rows = db.execute("SELECT * FROM businesses WHERE organization_id = ? AND (name LIKE ? OR website_domain LIKE ? OR email LIKE ?) ORDER BY score DESC, id", (ORGANIZATION_ID, like, like, like)).fetchall()
                else:
                    rows = db.execute("SELECT * FROM businesses WHERE organization_id = ? ORDER BY score DESC, id", (ORGANIZATION_ID,)).fetchall()
                return self.send_json(200, {"organization_id": ORGANIZATION_ID, "items": [row_json(r) for r in rows]})
            if path.startswith("/api/v1/businesses/"):
                ident = path.rsplit("/", 1)[1]
                if not ident.isdigit(): return self.send_json(404, {"error": "not_found"})
                row = db.execute("SELECT * FROM businesses WHERE id = ? AND organization_id = ?", (int(ident), ORGANIZATION_ID)).fetchone()
                return self.send_json(200, row_json(row)) if row else self.send_json(404, {"error": "not_found"})
            return self.send_json(404, {"error": "not_found"})
        finally:
            db.close()

    def do_POST(self):
        path = urlparse(self.path).path.rstrip("/")
        payload = self.read_json()
        if path == "/api/v1/businesses": return self.create_business(payload)
        if path == "/api/v1/suppressions": return self.create_suppression(payload)
        if path == "/api/v1/imports/preview": return self.preview_import(payload)
        return self.send_json(404, {"error": "not_found"})

    def create_business(self, payload):
        if not str(payload.get("name", "")).strip(): return self.send_json(400, {"error": "name_required"})
        business = normalize_business(payload)
        db = self.db()
        try:
            if is_suppressed(business, suppression_rows(db)): return self.send_json(409, {"error": "suppressed"})
            identity_fields = [("website_domain", business["website_domain"]), ("email", business["email"]), ("phone", business["phone"])]
            identity_fields = [(column, value) for column, value in identity_fields if value]
            if identity_fields:
                predicates = " OR ".join(f"{column} = ?" for column, _ in identity_fields)
                values = [value for _, value in identity_fields]
                if db.execute(f"SELECT id FROM businesses WHERE organization_id = ? AND ({predicates})", [ORGANIZATION_ID, *values]).fetchone():
                    return self.send_json(409, {"error": "duplicate"})
            scored = score_business(business)
            cur = db.execute("INSERT INTO businesses (organization_id,name,website,website_domain,email,phone,description,score,score_version,score_factors,website_class) VALUES (?,?,?,?,?,?,?,?,?,?,?)", (ORGANIZATION_ID, business["name"], business["website"], business["website_domain"], business["email"], business["phone"], str(business.get("description", "")), scored["score"], scored["score_version"], json.dumps(scored["factors"]), scored["website_class"]))
            db.commit()
            return self.send_json(201, row_json(db.execute("SELECT * FROM businesses WHERE id = ?", (cur.lastrowid,)).fetchone()))
        finally: db.close()

    def create_suppression(self, payload):
        kind, value = payload.get("kind"), str(payload.get("value", "")).strip().lower()
        if kind not in {"email", "domain", "phone"} or not value: return self.send_json(400, {"error": "invalid_suppression"})
        db = self.db()
        try:
            try:
                db.execute("INSERT INTO suppressions (organization_id,kind,value) VALUES (?,?,?)", (ORGANIZATION_ID, kind, value)); db.commit()
            except sqlite3.IntegrityError: pass
            row = db.execute("SELECT * FROM suppressions WHERE organization_id = ? AND kind = ? AND value = ?", (ORGANIZATION_ID, kind, value)).fetchone()
            return self.send_json(201, dict(row))
        finally: db.close()

    def preview_import(self, payload):
        rows = payload.get("rows", [])
        if not isinstance(rows, list): return self.send_json(400, {"error": "rows_required"})
        normalized = deduplicate_businesses([r for r in rows if isinstance(r, dict) and str(r.get("name", "")).strip()])
        db = self.db()
        try:
            suppressions = suppression_rows(db); existing = existing_businesses(db); seen = set()
            accepted, duplicate, suppressed = [], 0, 0
            for b in normalized:
                key = deduplication_key(b)
                if is_suppressed(b, suppressions): suppressed += 1
                elif (key in {deduplication_key(x) for x in existing} or key in seen): duplicate += 1
                else: seen.add(key); accepted.append(b)
            return self.send_json(200, {"accepted": len(accepted), "duplicates": duplicate + len(rows) - len(normalized), "suppressed": suppressed, "rows": accepted})
        finally: db.close()

    def log_message(self, *_): pass


def create_server(host="127.0.0.1", port=8000, db_path="prospects.db"):
    server = ThreadingHTTPServer((host, port), ApiHandler)
    server.db_path = db_path
    connect(db_path).close()
    return server


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Prospect Platform API")
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=int(os.environ.get("PROSPECT_API_PORT", "8000")))
    parser.add_argument("--db", default=os.environ.get("PROSPECT_API_DB", "prospects.db"))
    args = parser.parse_args()
    server = create_server(args.host, args.port, args.db)
    print(f"Prospect API listening on http://{args.host}:{args.port}", flush=True)
    try: server.serve_forever()
    except KeyboardInterrupt: pass
    finally: server.server_close()