ai-backend/server.py

import base64
import binascii
import datetime
from io import BytesIO
import os
import re
import shutil
import sys

from flask import Flask, jsonify, request, send_from_directory
from flask_cors import CORS
from PIL import Image
import pytesseract

from model_service import DarkPatternModelService


if hasattr(sys.stdout, "reconfigure"):
    try:
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stderr.reconfigure(encoding="utf-8")
    except Exception:
        pass

if shutil.which("tesseract") is None:
    windows_tesseract = r"C:\Program Files\Tesseract-OCR\tesseract.exe"
    if os.name == "nt" and os.path.exists(windows_tesseract):
        pytesseract.pytesseract.tesseract_cmd = windows_tesseract


SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
DIST_DIR = os.path.abspath(os.path.join(SCRIPT_DIR, "../dist"))
DATASET_PATH = os.path.join(SCRIPT_DIR, "dataset.csv")
SAMPLES_DIR = os.path.join(SCRIPT_DIR, "samples")

MAX_IMAGE_BYTES = 10 * 1024 * 1024
MAX_IMAGE_PIXELS = 25_000_000
MAX_BATCH_ITEMS = 300
MAX_TEXT_LENGTH = 2_000
OCR_MIN_CONFIDENCE = float(os.environ.get("OCR_MIN_CONFIDENCE", "35"))
SCREENSHOT_MIN_MODEL_SCORE = float(
    os.environ.get("SCREENSHOT_MIN_MODEL_SCORE", "45")
)

Image.MAX_IMAGE_PIXELS = MAX_IMAGE_PIXELS

if os.path.exists(DIST_DIR):
    app = Flask(__name__, static_folder=DIST_DIR, static_url_path="/")
else:
    app = Flask(__name__)

app.config["MAX_CONTENT_LENGTH"] = 12 * 1024 * 1024
CORS(
    app,
    resources={
        r"/api/*": {
            "origins": [
                re.compile(r"^chrome-extension://[a-p]{32}$"),
                "http://127.0.0.1:5173",
                "http://localhost:5173",
                "http://127.0.0.1:8000",
                "http://localhost:8000",
            ]
        }
    },
)

print("Loading grouped and calibrated NLP model...")
model_service = DarkPatternModelService(DATASET_PATH)
print(
    "NLP model ready: "
    f"accuracy={model_service.metrics['accuracy']}, "
    f"macro-F1={model_service.metrics['macroF1']}, "
    f"group-overlap={model_service.metrics['groupOverlap']}"
)


def get_severity(prediction):
    severity_map = {
        "Urgency": "high",
        "Scarcity": "medium",
        "Social Proof": "low",
        "Misdirection": "high",
        "Obstruction": "critical",
        "Sneaking": "critical",
        "Forced Action": "critical",
    }
    return severity_map.get(prediction, "medium")


def get_compliance_metadata(prediction, text):
    metadata = {
        "Urgency": (
            "Artificial urgency may pressure users into immediate decisions and "
            "can contribute to a deceptive-practices finding.",
            f"Remove or substantiate urgency language such as '{text}'.",
        ),
        "Scarcity": (
            "Unverified scarcity claims can mislead consumers about availability.",
            f"Verify '{text}' against live inventory or remove the claim.",
        ),
        "Social Proof": (
            "Unverified social proof can misrepresent genuine user activity.",
            f"Document the source of '{text}' or remove the notification.",
        ),
        "Misdirection": (
            "Guilt-inducing or biased language can interfere with neutral choice.",
            f"Rewrite '{text}' using neutral and symmetric option labels.",
        ),
        "Obstruction": (
            "Unnecessarily difficult cancellation or opt-out flows may obstruct "
            "consumer choice.",
            f"Simplify the exit path related to '{text}'.",
        ),
        "Sneaking": (
            "Hidden charges or preselected additions can obtain payment without "
            "clear, active consent.",
            f"Require explicit opt-in for any addition related to '{text}'.",
        ),
        "Forced Action": (
            "Requiring unrelated consent or account actions may undermine freely "
            "given consumer choice.",
            f"Allow users to continue without the unrelated requirement in '{text}'.",
        ),
    }
    return metadata.get(
        prediction,
        (
            "Potentially deceptive language requires human review.",
            "Review the copy for transparency, neutrality, and informed consent.",
        ),
    )


def add_policy_metadata(result):
    enriched = dict(result)
    if result["isDarkPattern"]:
        enriched["severity"] = get_severity(result["prediction"])
        violation, recommendation = get_compliance_metadata(
            result["prediction"], result["text"]
        )
        enriched["cfpbViolation"] = violation
        enriched["recommendation"] = recommendation
    else:
        enriched["severity"] = "none"
    return enriched


def decode_image(image_url):
    if image_url.startswith("data:image/"):
        match = re.match(
            r"^data:image/(?:png|jpeg|jpg|webp|gif);base64,(.*)$",
            image_url,
            flags=re.IGNORECASE | re.DOTALL,
        )
        if not match:
            raise ValueError("Unsupported or invalid image data URL")
        image_bytes = base64.b64decode(match.group(1), validate=True)
        if len(image_bytes) > MAX_IMAGE_BYTES:
            raise ValueError("Image exceeds the 10 MB limit")
        return Image.open(BytesIO(image_bytes)).convert("RGB")

    if "/api/samples/" in image_url:
        filename = os.path.basename(image_url.split("?")[0])
        sample_path = os.path.join(SAMPLES_DIR, filename)
        if not os.path.isfile(sample_path):
            raise ValueError("Sample image not found")
        return Image.open(sample_path).convert("RGB")

    raise ValueError(
        "Remote image URLs are disabled. Upload an image or use a bundled sample."
    )


def extract_ocr_lines(image):
    ocr_data = pytesseract.image_to_data(
        image, output_type=pytesseract.Output.DICT
    )
    lines = {}

    for index, raw_text in enumerate(ocr_data["text"]):
        text = raw_text.strip()
        if not text:
            continue

        try:
            confidence = float(ocr_data["conf"][index])
        except (TypeError, ValueError):
            confidence = -1
        if confidence < OCR_MIN_CONFIDENCE:
            continue

        key = (
            ocr_data["block_num"][index],
            ocr_data["par_num"][index],
            ocr_data["line_num"][index],
        )
        left = ocr_data["left"][index]
        top = ocr_data["top"][index]
        width = ocr_data["width"][index]
        height = ocr_data["height"][index]

        line = lines.setdefault(
            key,
            {
                "words": [],
                "confidences": [],
                "left": left,
                "top": top,
                "right": left + width,
                "bottom": top + height,
            },
        )
        line["words"].append(text)
        line["confidences"].append(confidence)
        line["left"] = min(line["left"], left)
        line["top"] = min(line["top"], top)
        line["right"] = max(line["right"], left + width)
        line["bottom"] = max(line["bottom"], top + height)

    extracted_lines = []
    for line in lines.values():
        text = " ".join(line["words"]).strip()
        if len(text) < 3:
            continue
        line["text"] = text
        line["ocrConfidence"] = round(
            sum(line["confidences"]) / len(line["confidences"]), 1
        )
        extracted_lines.append(line)
    return extracted_lines


@app.route("/api/health", methods=["GET"])
def health():
    return jsonify(
        {
            "status": "ok",
            "modelReady": model_service is not None,
            "calibrated": model_service.metrics["calibrated"],
        }
    )


@app.route("/api/metrics", methods=["GET"])
def get_metrics():
    return jsonify(model_service.metrics)


@app.route("/api/analyze-text", methods=["POST", "OPTIONS"])
def analyze_text():
    if request.method == "OPTIONS":
        return jsonify({}), 200

    data = request.get_json(silent=True) or {}
    text = " ".join(str(data.get("text") or "").split())
    if len(text) < 3:
        return jsonify({"error": "Text must contain at least 3 characters"}), 400
    if len(text) > MAX_TEXT_LENGTH:
        return jsonify({"error": "Text exceeds the 2,000 character limit"}), 400

    return jsonify(add_policy_metadata(model_service.classify(text)))


@app.route("/api/analyze-texts", methods=["POST", "OPTIONS"])
def analyze_texts():
    if request.method == "OPTIONS":
        return jsonify({}), 200

    data = request.get_json(silent=True) or {}
    items = data.get("items")
    if not isinstance(items, list):
        return jsonify({"error": "items must be an array"}), 400
    if len(items) > MAX_BATCH_ITEMS:
        return jsonify(
            {"error": f"Batch exceeds the {MAX_BATCH_ITEMS} item limit"}
        ), 400

    normalized = []
    for index, item in enumerate(items):
        if isinstance(item, str):
            item_id = str(index)
            text = item
        elif isinstance(item, dict):
            item_id = str(item.get("id", index))
            text = item.get("text", "")
        else:
            continue

        cleaned_text = " ".join(str(text).split())
        if 3 <= len(cleaned_text) <= MAX_TEXT_LENGTH:
            normalized.append({"id": item_id, "text": cleaned_text})

    predictions = model_service.classify_many(
        [item["text"] for item in normalized]
    )
    results = []
    for item, prediction in zip(normalized, predictions):
        enriched = add_policy_metadata(prediction)
        enriched["id"] = item["id"]
        results.append(enriched)

    return jsonify(
        {
            "status": "success",
            "received": len(items),
            "analyzed": len(results),
            "results": results,
        }
    )


@app.route("/api/analyze", methods=["POST", "OPTIONS"])
def analyze_image():
    if request.method == "OPTIONS":
        return jsonify({}), 200

    data = request.get_json(silent=True) or {}
    image_url = str(data.get("imageUrl") or "")
    if not image_url:
        return jsonify({"error": "imageUrl is required"}), 400

    try:
        image = decode_image(image_url)
        width, height = image.size
        ocr_lines = extract_ocr_lines(image)
        classifications = model_service.classify_many(
            [line["text"] for line in ocr_lines]
        )

        dark_patterns = []
        for line, classification in zip(ocr_lines, classifications):
            if not classification["isDarkPattern"]:
                continue
            if classification["confidence"] < SCREENSHOT_MIN_MODEL_SCORE:
                continue

            violation, recommendation = get_compliance_metadata(
                classification["prediction"], line["text"]
            )
            dark_patterns.append(
                {
                    "id": str(len(dark_patterns) + 1),
                    "type": classification["prediction"],
                    "severity": get_severity(classification["prediction"]),
                    "description": (
                        "Language classified as a potential "
                        f"{classification['prediction']} dark pattern."
                    ),
                    "confidence": classification["confidence"],
                    "confidenceBand": classification["confidenceBand"],
                    "calibrated": True,
                    "ocrConfidence": line["ocrConfidence"],
                    "location": {
                        "x": round((line["left"] / width) * 100, 2),
                        "y": round((line["top"] / height) * 100, 2),
                        "width": round(
                            ((line["right"] - line["left"]) / width) * 100, 2
                        ),
                        "height": round(
                            ((line["bottom"] - line["top"]) / height) * 100, 2
                        ),
                    },
                    "cfpbViolation": violation,
                    "recommendation": recommendation,
                    "evidence": line["text"],
                    "explanation": classification["explanation"],
                }
            )

        deductions = {
            "critical": 25,
            "high": 15,
            "medium": 10,
            "low": 5,
        }
        score_deduction = sum(
            deductions.get(pattern["severity"], 10)
            for pattern in dark_patterns
        )
        overall_score = max(5, 100 - score_deduction)
        if not dark_patterns:
            overall_score = 98

        if overall_score >= 80:
            risk_level = "low"
        elif overall_score >= 60:
            risk_level = "medium"
        elif overall_score >= 45:
            risk_level = "high"
        else:
            risk_level = "critical"

        return jsonify(
            {
                "imageUrl": image_url,
                "extractedText": "\n".join(
                    line["text"] for line in ocr_lines
                )
                or "No reliable text detected in screenshot.",
                "ocrMinimumConfidence": OCR_MIN_CONFIDENCE,
                "overallScore": overall_score,
                "riskLevel": risk_level,
                "darkPatterns": dark_patterns,
                "complianceReport": {
                    "cfpbAlignment": overall_score,
                    "issues": list(
                        dict.fromkeys(
                            pattern["cfpbViolation"]
                            for pattern in dark_patterns
                        )
                    ),
                    "recommendations": list(
                        dict.fromkeys(
                            pattern["recommendation"]
                            for pattern in dark_patterns
                        )
                    ),
                },
                "timestamp": datetime.datetime.now(
                    datetime.timezone.utc
                ).isoformat(),
            }
        )
    except (ValueError, binascii.Error) as error:
        return jsonify({"error": str(error)}), 400
    except Exception:
        app.logger.exception("Screenshot analysis failed")
        return jsonify({"error": "Screenshot analysis failed"}), 500


@app.route("/api/samples/<path:filename>", methods=["GET"])
def get_sample(filename):
    return send_from_directory(SAMPLES_DIR, filename)


@app.route("/api/dataset", methods=["GET"])
def get_dataset():
    query = request.args.get("q", "").strip()
    category = request.args.get("category", "").strip()
    try:
        limit = min(max(int(request.args.get("limit", 50)), 1), 100)
        offset = max(int(request.args.get("offset", 0)), 0)
    except ValueError:
        return jsonify({"error": "limit and offset must be integers"}), 400

    filtered = model_service.dataset
    if query:
        filtered = filtered[
            filtered["text"].str.contains(
                query, case=False, na=False, regex=False
            )
        ]
    if category:
        filtered = filtered[
            filtered["Pattern Category"].str.casefold()
            == category.casefold()
        ]

    return jsonify(
        {
            "status": "success",
            "total": int(len(filtered)),
            "limit": limit,
            "offset": offset,
            "records": filtered.iloc[offset : offset + limit].to_dict(
                orient="records"
            ),
            "categoryCounts": model_service.metrics["classDistribution"],
        }
    )


if os.path.exists(DIST_DIR):

    @app.route("/", defaults={"path": ""})
    @app.route("/<path:path>")
    def serve(path):
        target = os.path.join(app.static_folder, path)
        if path and os.path.isfile(target):
            return app.send_static_file(path)
        return app.send_static_file("index.html")


if __name__ == "__main__":
    app.run(
        host="0.0.0.0",
        port=int(os.environ.get("PORT", 8000)),
        debug=False,
    )