Create app.py

app.py

@@ -0,0 +1,376 @@
+"""
+API OpenAI-compatible wrappant aifreeforever.com
+Endpoints :
+  GET  /v1/models
+  POST /v1/chat/completions          (stream=false & stream=true)
+"""
+
+import asyncio, json, time, uuid
+from typing import Optional
+from fastapi import FastAPI, HTTPException
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel
+from playwright.async_api import async_playwright, Page, TimeoutError as PWTimeout
+
+# ════════════════════════════════════════
+# FastAPI
+# ════════════════════════════════════════
+app = FastAPI(title="Free Chat API")
+
+# Limite 1 requête Playwright à la fois (free tier = peu de RAM)
+SEM = asyncio.Semaphore(1)
+
+MODEL_NAME = "aifreeforever"
+
+# ════════════════════════════════════════
+# Modèles Pydantic (format OpenAI)
+# ════════════════════════════════════════
+class Message(BaseModel):
+    role: str
+    content: str
+
+class ChatRequest(BaseModel):
+    model: Optional[str] = MODEL_NAME
+    messages: list[Message]
+    stream: Optional[bool] = False
+    temperature: Optional[float] = None
+    max_tokens: Optional[int] = None
+
+# ════════════════════════════════════════
+# CONFIG PLAYWRIGHT
+# ════════════════════════════════════════
+URL         = "https://aifreeforever.com/tools/free-chatgpt-no-login"
+HEADLESS    = True
+MAX_RETRIES = 5
+
+SEL_TEXTAREA    = 'textarea[placeholder*="Ask anything"]'
+SEL_SEND_BTN    = 'button.absolute.right-3.top-4.w-10.h-10'
+SEL_BOT_MSG     = '.flex.justify-start .rounded-2xl.shadow-sm.bg-white.border.border-gray-100'
+SEL_BOT_CONTENT = '.markdown-content'
+SEL_ACCEPT_BTN  = 'button:has-text("Accept & Continue")'
+SEL_AGE_BTN     = 'button:has-text("13 and Over")'
+SEL_COPY_BTN    = (
+    'button.text-xs.text-gray-500.flex.items-center.gap-1'
+    '.px-2.py-1.rounded-md.transition-colors:has-text("Copy")'
+)
+SEL_ERROR_MSG   = 'p.text-red-600:has-text("Failed to send message")'
+
+COOKIE_SELECTORS = [
+    'button.unic-agree-all-button',
+    'button:has-text("Accepter et continuer")',
+    'button:has-text("Accepter tout")',
+    'button:has-text("Accept all")',
+    'button:has-text("Accept All")',
+    'button:has-text("I agree")',
+]
+
+AD_SELECTORS = [
+    'button:has(path[fill="#1D1D1B"][opacity="0.7"])',
+    'button:has(path[style*="stroke: rgb(255, 255, 255)"][style*="stroke-width: 6.353"])',
+    'button[aria-label*="close" i]',
+    'button[aria-label*="fermer" i]',
+    'button[title*="close" i]',
+    'button:has-text("×")',
+    'button:has-text("✕")',
+    'button:has-text("Close")',
+    '[class*="close-btn"]',
+]
+
+# ════════════════════════════════════════
+# HELPERS PLAYWRIGHT (identiques au script)
+# ════════════════════════════════════════
+
+async def _try_click(page: Page, selector: str, timeout: int) -> bool:
+    try:
+        btn = page.locator(selector).first
+        await btn.wait_for(state="visible", timeout=timeout)
+        await btn.click()
+        return True
+    except PWTimeout:
+        return False
+
+async def click_first_visible(page: Page, selectors: list[str], timeout: int = 4000):
+    tasks   = [_try_click(page, sel, timeout) for sel in selectors]
+    results = await asyncio.gather(*tasks, return_exceptions=True)
+    return any(r is True for r in results)
+
+async def click_if_visible(page: Page, selector: str, timeout: int = 8000) -> bool:
+    return await _try_click(page, selector, timeout)
+
+async def send_with_retry(page: Page, prompt: str) -> bool:
+    for attempt in range(1, MAX_RETRIES + 1):
+        if attempt > 1:
+            await page.locator(SEL_TEXTAREA).fill(prompt)
+        await page.locator(SEL_SEND_BTN).click(timeout=15_000)
+        await asyncio.sleep(2)
+        if await page.locator(SEL_ERROR_MSG).count() == 0:
+            return True
+    return False
+
+async def try_copy_button(page: Page) -> str | None:
+    try:
+        msgs  = page.locator(SEL_BOT_MSG)
+        count = await msgs.count()
+        if count == 0:
+            return None
+        last_msg = None
+        for i in range(count - 1, -1, -1):
+            msg = msgs.nth(i)
+            if await msg.locator(SEL_BOT_CONTENT).count() > 0:
+                last_msg = msg
+                break
+        if last_msg is None:
+            last_msg = msgs.nth(count - 1)
+        copy_btn = last_msg.locator(SEL_COPY_BTN)
+        await copy_btn.wait_for(state="visible", timeout=3000)
+        await copy_btn.click()
+        await asyncio.sleep(0.8)
+        text = await page.evaluate("async () => navigator.clipboard.readText()")
+        if text and len(text.strip()) > 20:
+            return text.strip()
+    except Exception:
+        pass
+    return None
+
+async def scrape_last_bot_message(page: Page) -> str | None:
+    try:
+        msgs  = page.locator(SEL_BOT_MSG)
+        count = await msgs.count()
+        for i in range(count - 1, -1, -1):
+            msg        = msgs.nth(i)
+            content_el = msg.locator(SEL_BOT_CONTENT)
+            if await content_el.count() > 0:
+                text = (await content_el.first.inner_text()).strip()
+                if len(text) > 20:
+                    return text
+            else:
+                text = (await msg.inner_text()).strip()
+                if len(text) > 80 and "Accept & Continue" not in text:
+                    return text
+    except Exception:
+        pass
+    return None
+
+async def get_response(page: Page) -> str:
+    text = await try_copy_button(page)
+    if text:
+        return text
+    return await scrape_last_bot_message(page) or ""
+
+async def wait_for_stream_end(page: Page, timeout_s: int = 120) -> None:
+    prev_text = ""
+    stable    = 0
+    elapsed   = 0.0
+    interval  = 0.6
+    while elapsed < timeout_s:
+        await asyncio.sleep(interval)
+        elapsed += interval
+        msgs  = page.locator(SEL_BOT_MSG)
+        count = await msgs.count()
+        if count == 0:
+            continue
+        current = ""
+        for i in range(count - 1, -1, -1):
+            content_el = msgs.nth(i).locator(SEL_BOT_CONTENT)
+            if await content_el.count() > 0:
+                t = (await content_el.first.inner_text()).strip()
+                if len(t) > 20:
+                    current = t
+                    break
+        if not current:
+            continue
+        if current != prev_text:
+            stable = 0
+        else:
+            stable += 1
+            if stable >= 3:
+                try:
+                    await msgs.nth(count - 1).locator(SEL_COPY_BTN).wait_for(
+                        state="visible", timeout=1500
+                    )
+                    return
+                except PWTimeout:
+                    stable = 0
+        prev_text = current
+
+# ════════════════════════════════════════
+# CŒUR : envoie un prompt, récupère la réponse
+# ════════════════════════════════════════
+
+async def chat(prompt: str) -> str:
+    async with async_playwright() as p:
+        browser = await p.chromium.launch(
+            headless=HEADLESS,
+            args=[
+                "--disable-blink-features=AutomationControlled",
+                "--disable-extensions", "--disable-default-apps",
+                "--no-first-run", "--no-sandbox", "--disable-gpu",
+                "--disable-dev-shm-usage",
+                "--window-size=1280,720",
+            ],
+        )
+        context = await browser.new_context(
+            user_agent=(
+                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+                "AppleWebKit/537.36 (KHTML, like Gecko) "
+                "Chrome/124.0.0.0 Safari/537.36"
+            ),
+            viewport={"width": 1280, "height": 720},
+            permissions=["clipboard-read", "clipboard-write"],
+            java_script_enabled=True,
+        )
+        await context.route(
+            "**/*",
+            lambda route: route.abort()
+            if route.request.resource_type in ("image", "media", "font")
+            else route.continue_(),
+        )
+        page = await context.new_page()
+
+        try:
+            await page.goto(URL, wait_until="domcontentloaded", timeout=60_000)
+
+            await asyncio.gather(
+                click_first_visible(page, AD_SELECTORS, timeout=3000),
+                click_first_visible(page, COOKIE_SELECTORS, timeout=5000),
+            )
+
+            await page.wait_for_selector(SEL_TEXTAREA, timeout=30_000)
+            await page.locator(SEL_TEXTAREA).fill(prompt)
+
+            if not await send_with_retry(page, prompt):
+                return ""
+
+            await asyncio.gather(
+                click_if_visible(page, SEL_AGE_BTN, timeout=8000),
+                click_if_visible(page, SEL_ACCEPT_BTN, timeout=8000),
+            )
+
+            await wait_for_stream_end(page, timeout_s=120)
+            response = await get_response(page)
+        finally:
+            await browser.close()
+
+        return response
+
+# ════════════════════════════════════════
+# HELPERS FORMAT OPENAI
+# ════════════════════════════════════════
+
+def _make_id():
+    return f"chatcmpl-{uuid.uuid4().hex[:29]}"
+
+def _completion_response(content: str, model: str) -> dict:
+    return {
+        "id": _make_id(),
+        "object": "chat.completion",
+        "created": int(time.time()),
+        "model": model,
+        "choices": [
+            {
+                "index": 0,
+                "message": {"role": "assistant", "content": content},
+                "finish_reason": "stop",
+            }
+        ],
+        "usage": {
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0,
+        },
+    }
+
+async def _stream_chunks(content: str, model: str):
+    """Génère des SSE au format OpenAI streaming."""
+    cid = _make_id()
+    created = int(time.time())
+
+    # Premier chunk (role)
+    chunk = {
+        "id": cid, "object": "chat.completion.chunk",
+        "created": created, "model": model,
+        "choices": [{"index": 0, "delta": {"role": "assistant", "content": ""}, "finish_reason": None}],
+    }
+    yield f"data: {json.dumps(chunk)}\n\n"
+
+    # Contenu découpé mot par mot
+    words = content.split(" ")
+    for i, word in enumerate(words):
+        token = word if i == 0 else f" {word}"
+        chunk = {
+            "id": cid, "object": "chat.completion.chunk",
+            "created": created, "model": model,
+            "choices": [{"index": 0, "delta": {"content": token}, "finish_reason": None}],
+        }
+        yield f"data: {json.dumps(chunk)}\n\n"
+        await asyncio.sleep(0.02)
+
+    # Dernier chunk
+    chunk = {
+        "id": cid, "object": "chat.completion.chunk",
+        "created": created, "model": model,
+        "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
+    }
+    yield f"data: {json.dumps(chunk)}\n\n"
+    yield "data: [DONE]\n\n"
+
+# ════════════════════════════════════════
+# ENDPOINTS
+# ════════════════════════════════════════
+
+@app.get("/")
+async def root():
+    return {"status": "ok", "message": "OpenAI-compatible API. Use /v1/chat/completions"}
+
+@app.get("/v1/models")
+async def list_models():
+    return {
+        "object": "list",
+        "data": [
+            {
+                "id": MODEL_NAME,
+                "object": "model",
+                "created": 1700000000,
+                "owned_by": "aifreeforever",
+            }
+        ],
+    }
+
+@app.post("/v1/chat/completions")
+async def chat_completions(req: ChatRequest):
+    # Construire le prompt à partir des messages
+    # On prend le dernier message user, ou on concatène tout
+    parts = []
+    for m in req.messages:
+        if m.role == "system":
+            parts.append(f"[System] {m.content}")
+        elif m.role == "user":
+            parts.append(f"{m.content}")
+        elif m.role == "assistant":
+            parts.append(f"[Assistant] {m.content}")
+    prompt = "\n\n".join(parts)
+
+    if not prompt.strip():
+        raise HTTPException(status_code=400, detail="Empty prompt")
+
+    # Sémaphore : 1 seule requête Playwright à la fois
+    async with SEM:
+        try:
+            content = await chat(prompt)
+        except Exception as e:
+            raise HTTPException(status_code=500, detail=str(e))
+
+    if not content:
+        raise HTTPException(status_code=502, detail="No response from upstream")
+
+    model = req.model or MODEL_NAME
+
+    # Streaming
+    if req.stream:
+        return StreamingResponse(
+            _stream_chunks(content, model),
+            media_type="text/event-stream",
+        )
+
+    # Non-streaming
+    return _completion_response(content, model)