From 30381f687e3b284c34fcb12ea36539ac02923029 Mon Sep 17 00:00:00 2001
From: hlohaus <983577+hlohaus@users.noreply.github.com>
Date: Thu, 30 Jan 2025 00:35:51 +0100
Subject: Add Janus_Pro_7B provider Add proof of DeepSeekAPI provider Add
 BackendApi provider from HuggingSpace

---
 g4f/Provider/needs_auth/DeepSeekAPI.py | 105 +++++++++++++++++++++++++++++++++
 1 file changed, 105 insertions(+)
 create mode 100644 g4f/Provider/needs_auth/DeepSeekAPI.py

(limited to 'g4f/Provider/needs_auth/DeepSeekAPI.py')

diff --git a/g4f/Provider/needs_auth/DeepSeekAPI.py b/g4f/Provider/needs_auth/DeepSeekAPI.py
new file mode 100644
index 00000000..8158d2c6
--- /dev/null
+++ b/g4f/Provider/needs_auth/DeepSeekAPI.py
@@ -0,0 +1,105 @@
+from __future__ import annotations
+
+import os
+import json
+import time
+from typing import AsyncIterator
+import asyncio
+
+from ..base_provider import AsyncAuthedProvider
+from ...requests import get_args_from_nodriver
+from ...providers.response import AuthResult, RequestLogin, Reasoning, JsonConversation, FinishReason
+from ...typing import AsyncResult, Messages
+
+try:
+    from curl_cffi import requests
+    from dsk.api import DeepSeekAPI, AuthenticationError, DeepSeekPOW
+except ImportError:
+    pass
+
+class DeepSeekAPIArgs(DeepSeekAPI):
+    def __init__(self, args: dict):
+        args.pop("headers")
+        self.auth_token = args.pop("api_key")
+        if not self.auth_token or not isinstance(self.auth_token, str):
+            raise AuthenticationError("Invalid auth token provided")
+        self.args = args
+        self.pow_solver = DeepSeekPOW()
+
+    def _make_request(self, method: str, endpoint: str, json_data: dict, pow_required: bool = False):
+        url = f"{self.BASE_URL}{endpoint}"
+        headers = self._get_headers()
+        if pow_required:
+            challenge = self._get_pow_challenge()
+            pow_response = self.pow_solver.solve_challenge(challenge)
+            headers = self._get_headers(pow_response)
+            
+        response = requests.request(
+            method=method,
+            url=url,
+            json=json_data, **{
+                "headers":headers,
+                "impersonate":'chrome',
+                "timeout":None,
+                **self.args
+            }
+        )
+        return response.json()
+
+class DeepSeekAPI(AsyncAuthedProvider):
+    url = "https://chat.deepseek.com"
+    working = False
+    needs_auth = True
+    use_nodriver = True
+    _access_token = None
+
+    @classmethod
+    async def on_auth_async(cls, proxy: str = None, **kwargs) -> AsyncIterator:
+        yield RequestLogin(cls.__name__, os.environ.get("G4F_LOGIN_URL") or "")
+        async def callback(page):
+            while True:
+                await asyncio.sleep(1)
+                cls._access_token = json.loads(await page.evaluate("localStorage.getItem('userToken')") or "{}").get("value")
+                if cls._access_token:
+                    break
+        args = await get_args_from_nodriver(cls.url, proxy, callback=callback)
+        yield AuthResult(
+            api_key=cls._access_token,
+            **args
+        )
+
+    @classmethod
+    async def create_authed(
+        cls,
+        model: str,
+        messages: Messages,
+        auth_result: AuthResult,
+        conversation: JsonConversation = None,
+        **kwargs
+    ) -> AsyncResult:
+        # Initialize with your auth token
+        api = DeepSeekAPIArgs(auth_result.get_dict())
+
+        # Create a new chat session
+        if conversation is None:
+            chat_id = api.create_chat_session()
+            conversation = JsonConversation(chat_id=chat_id)
+
+        is_thinking = 0
+        for chunk in api.chat_completion(
+            conversation.chat_id,
+            messages[-1]["content"],
+            thinking_enabled=True
+        ):
+            if chunk['type'] == 'thinking':
+                if not is_thinking:
+                    yield Reasoning(None, "Is thinking...")
+                    is_thinking = time.time()
+                yield Reasoning(chunk['content'])
+            elif chunk['type'] == 'text':
+                if is_thinking:
+                    yield Reasoning(None, f"Thought for {time.time() - is_thinking:.2f}s")
+                    is_thinking = 0
+                yield chunk['content']
+            if chunk['finish_reason']:
+                yield FinishReason(chunk['finish_reason'])
\ No newline at end of file
-- 
cgit v1.2.3