from flask import Flask, request, jsonify, render_template import os import uuid import time import threading import tiktoken import requests import base64 from datetime import datetime, timedelta from collections import defaultdict, deque from detoxify import Detoxify import logging logging.basicConfig(level=logging.INFO) app = Flask(__name__, static_folder='static', template_folder='templates') print("Loading Detoxify model... This may take a moment.") detoxify_model = Detoxify('multilingual') print("Detoxify model loaded successfully.") API_KEY = os.getenv('API_KEY', 'your-api-key-here') request_durations = deque(maxlen=100) request_timestamps = deque(maxlen=1000) daily_requests = defaultdict(int) daily_tokens = defaultdict(int) concurrent_requests = 0 concurrent_requests_lock = threading.Lock() encoding = tiktoken.get_encoding("cl100k_base") NSFW_MODEL_URL = "https://teachablemachine.withgoogle.com/models/gJOADmf_u/" def count_tokens(text): return len(encoding.encode(text)) def classify_image_nsfw(image_data): try: response = requests.post( NSFW_MODEL_URL, json={"instances": [{"image_bytes": {"b64": image_data}}]}, timeout=10 ) if response.status_code == 200: predictions = response.json().get('predictions', [{}])[0] nsfw_score = predictions.get('nsfw', 0.0) return nsfw_score > 0.5 return False except: return False def transform_predictions(prediction_dict): category_keys = [ "toxicity", "severe_toxicity", "obscene", "threat", "insult", "identity_attack", "sexual_explicit" ] scores = {} for key in category_keys: scores[key] = float(prediction_dict.get(key, 0.0)) threshold = 0.5 bool_categories = {key: (scores[key] > threshold) for key in category_keys} flagged = any(bool_categories.values()) return flagged, bool_categories, scores def track_request_metrics(start_time, tokens_count): end_time = time.time() duration = end_time - start_time app.logger.info(f"Server-side processing for moderation request took {duration * 1000:.2f} ms.") request_durations.append(duration) request_timestamps.append(datetime.now()) today = datetime.now().strftime("%Y-%m-%d") daily_requests[today] += 1 daily_tokens[today] += tokens_count def get_performance_metrics(): global concurrent_requests with concurrent_requests_lock: current_concurrent = concurrent_requests if not request_durations: avg_request_time = 0 peak_request_time = 0 else: avg_request_time = sum(request_durations) / len(request_durations) peak_request_time = max(request_durations) now = datetime.now() one_minute_ago = now - timedelta(seconds=60) requests_last_minute = sum(1 for ts in request_timestamps if ts > one_minute_ago) today = now.strftime("%Y-%m-%d") today_requests = daily_requests.get(today, 0) today_tokens = daily_tokens.get(today, 0) last_7_days = [] for i in range(7): date = (now - timedelta(days=i)).strftime("%Y-%m-%d") last_7_days.append({ "date": date, "requests": daily_requests.get(date, 0), "tokens": daily_tokens.get(date, 0) }) return { "avg_request_time_ms": avg_request_time * 1000, "peak_request_time_ms": peak_request_time * 1000, "requests_per_minute": requests_last_minute, "concurrent_requests": current_concurrent, "today_requests": today_requests, "today_tokens": today_tokens, "last_7_days": last_7_days } @app.route('/') def home(): return render_template('index.html') @app.route('/v1/moderations', methods=['POST']) def moderations(): global concurrent_requests with concurrent_requests_lock: concurrent_requests += 1 start_time = time.time() total_tokens = 0 response = None try: auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith("Bearer "): response = jsonify({"error": "Unauthorized"}), 401 return response provided_api_key = auth_header.split(" ")[1] if provided_api_key != API_KEY: response = jsonify({"error": "Unauthorized"}), 401 return response data = request.get_json() raw_input = data.get('input') if raw_input is None: response = jsonify({"error": "Invalid input, 'input' field is required"}), 400 return response if isinstance(raw_input, str): texts = [raw_input] elif isinstance(raw_input, list): texts = raw_input else: response = jsonify({"error": "Invalid input format, expected string or list of strings"}), 400 return response if not texts: response = jsonify({"error": "Input list cannot be empty"}), 400 return response if len(texts) > 10: response = jsonify({"error": "Too many input items. Maximum 10 allowed."}), 400 return response results = [] for text in texts: if not isinstance(text, str): if isinstance(text, dict) and 'type' in text and text['type'] == 'image_url': image_data = text.get('image_url', {}).get('url', '') if image_data.startswith('data:image'): image_b64 = image_data.split(',')[1] is_nsfw = classify_image_nsfw(image_b64) results.append({ "flagged": is_nsfw, "categories": { "sexual": is_nsfw, "sexual_explicit": is_nsfw }, "category_scores": { "sexual": 0.9 if is_nsfw else 0.1, "sexual_explicit": 0.9 if is_nsfw else 0.1 } }) continue response = jsonify({"error": "Each input item must be a string or image object"}), 400 return response if len(text.encode('utf-8')) > 300000: response = jsonify({"error": "Each input item must have a maximum of 300k bytes."}), 400 return response total_tokens += count_tokens(text) predictions = detoxify_model.predict([text]) single_prediction = {key: value[0] for key, value in predictions.items()} flagged, bool_categories, scores = transform_predictions(single_prediction) results.append({ "flagged": flagged, "categories": bool_categories, "category_scores": scores, }) response_data = { "id": "modr-" + uuid.uuid4().hex[:24], "model": "text-moderation-detoxify-multilingual", "results": results } response = jsonify(response_data) return response except Exception as e: app.logger.error(f"An error occurred: {e}", exc_info=True) response = jsonify({"error": "An internal server error occurred."}), 500 return response finally: if response and (response[1] < 400 if isinstance(response, tuple) else response.status_code < 400): track_request_metrics(start_time, total_tokens) with concurrent_requests_lock: concurrent_requests -= 1 @app.route('/v1/metrics', methods=['GET']) def metrics(): auth_header = request.headers.get('Authorization') if not auth_header or not auth_header.startswith("Bearer "): return jsonify({"error": "Unauthorized"}), 401 provided_api_key = auth_header.split(" ")[1] if provided_api_key != API_KEY: return jsonify({"error": "Unauthorized"}), 401 return jsonify(get_performance_metrics()) # Create the HTML template with Tailwind CSS with open("templates/index.html", "w") as f: f.write(""" AI Content Moderator

Smart Moderator

Advanced, multilingual and multimodal content moderation API

API Configuration

API Usage

POST /v1/moderations
Site: https://nixaut-codelabs-smart-moderator.hf.space
curl -X POST -H "Authorization: Bearer YOUR_API_KEY" -H "Content-Type: application/json" -d '{"input":"text to moderate"}' /v1/moderations

Content Analysis

Example Prompts

"Hello, how are you today? I hope you're having a wonderful time!"

"I hate you and I will find you and hurt you badly."

"C'est une belle journée pour apprendre la programmation et l'intelligence artificielle."

"I can't take this anymore. I want to end everything and disappear forever."

"¡Hola! Me encanta aprender nuevos idiomas y conocer diferentes culturas."

"You're absolutely worthless and nobody will ever love someone like you."

About This Tool

Multilingual

Supports content analysis in multiple languages with high accuracy.

Fast Processing

Optimized model for quick content analysis with real-time results.

Secure

API key authentication ensures your requests remain secure and private.

""") def create_directories_and_files(): os.makedirs('templates', exist_ok=True) os.makedirs('static', exist_ok=True) index_path = os.path.join('templates', 'index.html') if not os.path.exists(index_path): with open(index_path, 'w', encoding='utf-8') as f: f.write(''' Smart Moderator

Smart Moderator

Performance Metrics

Avg. Response (last 100)

0ms

Requests / Minute

0

Peak Response (last 100)

0ms

Today's Requests

0

Last 7 Days Activity

API Tester

Maximum 10 text inputs allowed

API Documentation

Endpoint

POST /v1/moderations

Site URL

https://nixaut-codelabs-smart-moderator.hf.space

Request Body

{
  "input": "Text to moderate" 
}

Response

{
  "id": "modr-1234567890abcdef",
  "model": "text-moderation-detoxify-multilingual",
  "results": [
    {
      "flagged": true,
      "categories": {
        "toxicity": true,
        "severe_toxicity": false,
        /* ... other categories */
      },
      "category_scores": {
        "toxicity": 0.95,
        "severe_toxicity": 0.1,
        /* ... other scores */
      }
    }
  ]
}
''') if __name__ == '__main__': create_directories_and_files() port = int(os.getenv('PORT', 7860)) app.run(host='0.0.0.0', port=port, debug=True, use_reloader=False)