update sth. at 2025-01-04 23:21:29

.gitignore

@@ -0,0 +1,76 @@
+.externalNativeBuild
+import-summary.txt
+
+#java files
+*.class
+*.dex
+.sync/
+
+#for idea temp file
+*.iws
+*.ipr
+*.iml
+target/
+.idea/
+.gradle/
+release/
+build/
+spoon/
+releasebak/
+
+#mac temp file
+__MACOSX
+.DS_Store
+._.DS_Store
+
+#for eclipse
+.settings/
+local.properties
+*gen/
+*.classpath
+*/bin/
+bin/
+.project
+
+#temp file
+*.bak
+
+*.pmd
+sh.exe.stackdump
+
+.vs/
+.vscode/
+
+*.log
+*.ctxt
+.mtj.tmp/
+
+# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
+hs_err_pid*
+
+# Package Files #
+# *.jar
+*.war
+*.nar
+*.ear
+*.zip
+*.tar.gz
+*.rar
+*.cxx
+*.cfg
+# for nodejs
+node_modules/
+# for python 
+package-lock.json
+.$*
+*.drawio.bkp
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.Python
+env
+.env
+.venv
+pip-log.txt
+

debug.md

@@ -72,7 +72,7 @@ curl -X POST http://localhost:7860/v1/chat/completions \
     "presence_penalty": 0.0
   }'
 
-curl -X POST http://localhost:7860/chat \
+curl -X POST https://sanbo1200-planrun.hf.space/chat \
   -H 'Accept: application/json' \
   -H 'Authorization: Bearer YOUR_API_KEY' \
   -H "Content-Type: application/json" \

m1.py

@@ -1,24 +1,13 @@
 import os
 import uvicorn
 import time
-import logging
 from fastapi import FastAPI, Request, HTTPException
 import tiktoken
 from json.decoder import JSONDecodeError
 
 # 初始化 FastAPI
 app = FastAPI()
-# 配置日志
-logging.basicConfig(level=logging.INFO)
-logger = logging.getLogger(__name__)
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s',
-    handlers=[
-        logging.StreamHandler(),
-        logging.FileHandler('api_server.log', encoding='utf-8')
-    ]
-)
+
 # 获取编码器
 encoding = tiktoken.get_encoding("cl100k_base")
 
@@ -35,10 +24,8 @@ def general_id(zimu=4, num=6):
 # 异常处理函数
 def handle_exception(e):
     if isinstance(e, JSONDecodeError):
-        logger.error("JSON解析错误: %s", str(e))
         raise HTTPException(status_code=400, detail="无效的JSON格式")
     else:
-        logger.error("服务器错误: %s", str(e))
         raise HTTPException(status_code=500, detail="服务器内部错误")
 
 
@@ -52,6 +39,20 @@ def is_chatgpt_format(data):
         pass
     return False
 
+def get_workers_count() -> int:
+    """
+    Calculate optimal number of workers
+    Default: 4, Maximum: 8
+    Formula: min(max(4, (2 * CPU cores) + 1), 8)
+    """
+    try:
+        cpu_cores = multiprocessing.cpu_count()
+        recommended_workers = (2 * cpu_cores) + 1
+        return min(max(4, recommended_workers), 8)
+    except Exception as e:
+        if debug:
+            print(f"Worker count calculation failed: {e}, using default 4")
+        return 4
 
 # 模拟 ChatGPT 响应
 def generate_response(headers, data):
@@ -67,7 +68,7 @@ def generate_response(headers, data):
         if is_chatgpt_format(result):
             response_data = result
         else:
-            # 计算时间戳，13位毫秒时间戳
+            # 计算时间戳
             current_timestamp = int(time.time() * 1000)
 
             # 计算 token 数量
@@ -104,7 +105,7 @@ def generate_response(headers, data):
                 ]
             }
         # 打印最终返回的数据
-        logger.info("Response Data: %s", response_data)
+        print(f"Response Data: {response_data}" )
         return response_data
     except Exception as e:
         handle_exception(e)  # 异常处理函数
@@ -166,24 +167,23 @@ def get_dynamic_routes():
 
 # 注册单个动态路由
 def register_route(path: str):
-    # logger.info("register  route path: %s", path)
+    print(f"register  route path: {path}")
     @app.post(path)
     async def dynamic_chat_endpoint(request: Request):
         try:
             headers = request.headers
             data = request.json()
-            logger.info("Received Request Header: %s\nData: %s", headers, data)
+            print(f"Received Request Header: {headers}\nData: {data}")
             result = generate_response(headers, data)
             return result
         except Exception as e:
-            logger.error("Error: %s", str(e))
             raise HTTPException(status_code=500, detail=str(e))
 
 
 # 动态设置路由
 def setup_dynamic_routes():
     routes = get_dynamic_routes()
-    logger.info("Registering routes: %s", routes)
+    print("Registering routes: {routes}")
     for path in routes:
         register_route(path)
 

more_core.py

@@ -1,147 +1,42 @@
 import os
-import asyncio
 import time
 import multiprocessing
-from typing import Dict, Any, List, Optional
+from typing import Dict, Any, List
 from fastapi import FastAPI, Request, HTTPException
 import uvicorn
 import tiktoken
 from json.decoder import JSONDecodeError
+import random
+import string
 
-# Initialize FastAPI application
 app = FastAPI(
-    title="ChatGPT API Simulation",
-    description="High-performance ChatGPT API simulation service",
-    version="1.0.0"
+    title="ones",
+    description="High-performance API service",
+    version="1.0.0|2025.1.6"
 )
 
-debug = True
-
-
-class ServerConfig:
-    """Server configuration handler with dynamic worker calculation"""
-
-    @staticmethod
-    def get_workers_count() -> int:
-        """
-        Calculate optimal number of workers
-        Default: 4, Maximum: 8
-        Formula: min(max(4, (2 * CPU cores) + 1), 8)
-        """
-        try:
-            cpu_cores = multiprocessing.cpu_count()
-            recommended_workers = (2 * cpu_cores) + 1
-            return min(max(4, recommended_workers), 8)
-        except Exception as e:
-            if debug:
-                print(f"Worker count calculation failed: {e}, using default 4")
-            return 4
-
-    @classmethod
-    def get_uvicorn_config(cls,
-                           app: FastAPI,
-                           host: str = "0.0.0.0",
-                           port: int = 7860) -> uvicorn.Config:
-        """Get optimized Uvicorn configuration"""
-        workers = cls.get_workers_count()
-        if debug:
-            print(f"Configuring server with {workers} workers")
-
-        return uvicorn.Config(
-            app=app,
-            host=host,
-            port=port,
-            workers=workers,
-            loop="uvloop",
-            limit_concurrency=1000,
-            timeout_keep_alive=30,
-            access_log=True,
-            log_level="info",
-            http="httptools"
-        )
-
-
-class TokenProcessor:
-    """Token processing handler"""
-
-    def __init__(self):
-        self.encoding = tiktoken.get_encoding("cl100k_base")
-
-    async def calculate_tokens(self, text: str) -> int:
-        """Calculate tokens for given text"""
-        return len(self.encoding.encode(text))
-
-
-class ChatGPTSimulator:
-    """ChatGPT response simulator"""
-
-    def __init__(self):
-        self.token_processor = TokenProcessor()
-
-    @staticmethod
-    def generate_id(letters: int = 4, numbers: int = 6) -> str:
-        """Generate random chat completion ID"""
-        import random
-        import string
-        letters_str = ''.join(random.choices(string.ascii_lowercase, k=letters))
-        numbers_str = ''.join(random.choices(string.digits, k=numbers))
-        return f"chatcmpl-{letters_str}{numbers_str}"
-
-    async def generate_response(self, headers: Dict[str, str], data: Dict[str, Any]) -> Dict[str, Any]:
-        """Generate simulated ChatGPT response"""
-        try:
-            result = "This is a test result."
-
-            prompt_tokens = await self.token_processor.calculate_tokens(str(data))
-            completion_tokens = await self.token_processor.calculate_tokens(result)
-            total_tokens = prompt_tokens + completion_tokens
-
-            response_data = {
-                "id": self.generate_id(),
-                "object": "chat.completion",
-                "created": int(time.time()),
-                "model": data.get("model", "gpt-3.5-turbo"),
-                "usage": {
-                    "prompt_tokens": prompt_tokens,
-                    "completion_tokens": completion_tokens,
-                    "total_tokens": total_tokens
-                },
-                "choices": [
-                    {
-                        "message": {
-                            "role": "assistant",
-                            "content": result
-                        },
-                        "finish_reason": "stop",
-                        "index": 0
-                    }
-                ]
-            }
-            return response_data
-
-        except Exception as e:
-            if debug:
-                print(f"Response generation error: {e}")
-            raise HTTPException(status_code=500, detail=str(e))
-
-
-class RouteManager:
-    """Route management handler"""
+debug = False
 
+class APIServer:
+    """High-performance API server implementation"""
+    
     def __init__(self, app: FastAPI):
         self.app = app
-        self.simulator = ChatGPTSimulator()
-
-    def setup_routes(self):
-        """Setup API routes"""
-        routes = self._get_dynamic_routes()
-        if debug:
-            print(f"Registering routes: {routes}")
+        self.encoding = tiktoken.get_encoding("cl100k_base")
+        self._setup_routes()
+    
+    def _setup_routes(self) -> None:
+        """Initialize API routes"""
+        routes = self._get_routes()
         for path in routes:
             self._register_route(path)
-
-    def _get_dynamic_routes(self) -> List[str]:
-        """Get dynamic routes based on environment variables"""
+        
+        @self.app.get("/")
+        async def health_check() -> str:
+            return "你好"
+    
+    def _get_routes(self) -> List[str]:
+        """Get configured API routes"""
         default_path = "/v1/chat/completions"
         replace_chat = os.getenv("REPLACE_CHAT", "")
         prefix_chat = os.getenv("PREFIX_CHAT", "")
@@ -152,57 +47,120 @@ class RouteManager:
 
         routes = []
         if prefix_chat:
-            routes.extend(f"{prefix.rstrip('/')}{default_path}"
-                          for prefix in prefix_chat.split(","))
+            routes.extend(f"{prefix.rstrip('/')}{default_path}" 
+                        for prefix in prefix_chat.split(","))
             return routes
 
         if append_chat:
             append_paths = [path.strip() for path in append_chat.split(",") if path.strip()]
             routes = [default_path] + append_paths
+            return routes
 
-        return routes or [default_path]
-
-    def _register_route(self, path: str):
-        """Register a single route"""
+        return [default_path]
 
-        @self.app.post(path)
+    def _register_route(self, path: str) -> None:
+        """Register a single API route"""
         async def chat_endpoint(request: Request) -> Dict[str, Any]:
             try:
                 headers = dict(request.headers)
                 data = await request.json()
                 if debug:
                     print(f"Request received...\r\n\tHeaders: {headers},\r\n\tData: {data}")
-                return await self.simulator.generate_response(headers, data)
+                return await self._generate_response(headers, data)
             except JSONDecodeError as e:
                 if debug:
                     print(f"JSON decode error: {e}")
-                raise HTTPException(status_code=400, detail="Invalid JSON format")
+                raise HTTPException(status_code=400, detail="Invalid JSON format") from e
             except Exception as e:
                 if debug:
                     print(f"Request processing error: {e}")
-                raise HTTPException(status_code=500, detail="Internal server error")
+                raise HTTPException(status_code=500, detail="Internal server error") from e
+        
+        self.app.post(path)(chat_endpoint)
 
+    def _calculate_tokens(self, text: str) -> int:
+        """Calculate token count for text"""
+        return len(self.encoding.encode(text))
+
+    def _generate_id(self, letters: int = 4, numbers: int = 6) -> str:
+        """Generate unique chat completion ID"""
+        letters_str = ''.join(random.choices(string.ascii_lowercase, k=letters))
+        numbers_str = ''.join(random.choices(string.digits, k=numbers))
+        return f"chatcmpl-{letters_str}{numbers_str}"
+
+    async def _generate_response(self, headers: Dict[str, str], data: Dict[str, Any]) -> Dict[str, Any]:
+        """Generate API response"""
+        try:
+            result = "This is a test result."
+            prompt_tokens = self._calculate_tokens(str(data))
+            completion_tokens = self._calculate_tokens(result)
+            total_tokens = prompt_tokens + completion_tokens
+
+            return {
+                "id": self._generate_id(),
+                "object": "chat.completion",
+                "created": int(time.time()),
+                "model": data.get("model", "gpt-3.5-turbo"),
+                "usage": {
+                    "prompt_tokens": prompt_tokens,
+                    "completion_tokens": completion_tokens,
+                    "total_tokens": total_tokens
+                },
+                "choices": [{
+                    "message": {
+                        "role": "assistant",
+                        "content": result
+                    },
+                    "finish_reason": "stop",
+                    "index": 0
+                }]
+            }
+        except Exception as e:
+            if debug:
+                print(f"Response generation error: {e}")
+            raise HTTPException(status_code=500, detail=str(e)) from e
 
-@app.get("/")
-async def health_check() -> str:
-    """Health check endpoint"""
-    return "Service is running..."
+    def _get_workers_count(self) -> int:
+        """Calculate optimal worker count"""
+        try:
+            cpu_cores = multiprocessing.cpu_count()
+            recommended_workers = (2 * cpu_cores) + 1
+            return min(max(4, recommended_workers), 8)
+        except Exception as e:
+            if debug:
+                print(f"Worker count calculation failed: {e}, using default 4")
+            return 4
 
+    def get_server_config(self, host: str = "0.0.0.0", port: int = 7860) -> uvicorn.Config:
+        """Get server configuration"""
+        workers = self._get_workers_count()
+        if debug:
+            print(f"Configuring server with {workers} workers")
 
-def run_server(host: str = "0.0.0.0", port: int = 7860):
-    """Run server with optimized configuration"""
-    config = ServerConfig.get_uvicorn_config(
-        app=app,
-        host=host,
-        port=port
-    )
+        return uvicorn.Config(
+            app=self.app,
+            host=host,
+            port=port,
+            workers=workers,
+            loop="uvloop",
+            limit_concurrency=1000,
+            timeout_keep_alive=30,
+            access_log=True,
+            log_level="info",
+            http="httptools"
+        )
 
-    server = uvicorn.Server(config)
-    server.run()
+    def run(self, host: str = "0.0.0.0", port: int = 7860) -> None:
+        """Run the API server"""
+        config = self.get_server_config(host, port)
+        server = uvicorn.Server(config)
+        server.run()
 
+def create_server() -> APIServer:
+    """Factory function to create server instance"""
+    return APIServer(app)
 
 if __name__ == "__main__":
-    route_manager = RouteManager(app)
-    route_manager.setup_routes()
     port = int(os.getenv("PORT", "7860"))
-    run_server(port=port)
+    server = create_server()
+    server.run(port=port)

more_core_bksss.py

@@ -0,0 +1,208 @@
+import os
+import asyncio
+import time
+import multiprocessing
+from typing import Dict, Any, List, Optional
+from fastapi import FastAPI, Request, HTTPException
+import uvicorn
+import tiktoken
+from json.decoder import JSONDecodeError
+
+# Initialize FastAPI application
+app = FastAPI(
+    title="ones",
+    description="High-performance  API  service",
+    version="1.0.0|2025.1.6"
+)
+# debug
+debug = False
+
+
+class ServerConfig:
+    """Server configuration handler with dynamic worker calculation"""
+
+    @staticmethod
+    def get_workers_count() -> int:
+        """
+        Calculate optimal number of workers
+        Default: 4, Maximum: 8
+        Formula: min(max(4, (2 * CPU cores) + 1), 8)
+        """
+        try:
+            cpu_cores = multiprocessing.cpu_count()
+            recommended_workers = (2 * cpu_cores) + 1
+            return min(max(4, recommended_workers), 8)
+        except Exception as e:
+            if debug:
+                print(f"Worker count calculation failed: {e}, using default 4")
+            return 4
+
+    @classmethod
+    def get_uvicorn_config(cls,
+                           app: FastAPI,
+                           host: str = "0.0.0.0",
+                           port: int = 7860) -> uvicorn.Config:
+        """Get optimized Uvicorn configuration"""
+        workers = cls.get_workers_count()
+        if debug:
+            print(f"Configuring server with {workers} workers")
+
+        return uvicorn.Config(
+            app=app,
+            host=host,
+            port=port,
+            workers=workers,
+            loop="uvloop",
+            limit_concurrency=1000,
+            timeout_keep_alive=30,
+            access_log=True,
+            log_level="info",
+            http="httptools"
+        )
+
+
+class TokenProcessor:
+    """Token processing handler"""
+
+    def __init__(self):
+        self.encoding = tiktoken.get_encoding("cl100k_base")
+
+    async def calculate_tokens(self, text: str) -> int:
+        """Calculate tokens for given text"""
+        return len(self.encoding.encode(text))
+
+
+class ChatGPTSimulator:
+    """ChatGPT response simulator"""
+
+    def __init__(self):
+        self.token_processor = TokenProcessor()
+
+    @staticmethod
+    def generate_id(letters: int = 4, numbers: int = 6) -> str:
+        """Generate random chat completion ID"""
+        import random
+        import string
+        letters_str = ''.join(random.choices(string.ascii_lowercase, k=letters))
+        numbers_str = ''.join(random.choices(string.digits, k=numbers))
+        return f"chatcmpl-{letters_str}{numbers_str}"
+
+    async def generate_response(self, headers: Dict[str, str], data: Dict[str, Any]) -> Dict[str, Any]:
+        """Generate simulated ChatGPT response"""
+        try:
+            result = "This is a test result."
+
+            prompt_tokens = await self.token_processor.calculate_tokens(str(data))
+            completion_tokens = await self.token_processor.calculate_tokens(result)
+            total_tokens = prompt_tokens + completion_tokens
+
+            response_data = {
+                "id": self.generate_id(),
+                "object": "chat.completion",
+                "created": int(time.time()),
+                "model": data.get("model", "gpt-3.5-turbo"),
+                "usage": {
+                    "prompt_tokens": prompt_tokens,
+                    "completion_tokens": completion_tokens,
+                    "total_tokens": total_tokens
+                },
+                "choices": [
+                    {
+                        "message": {
+                            "role": "assistant",
+                            "content": result
+                        },
+                        "finish_reason": "stop",
+                        "index": 0
+                    }
+                ]
+            }
+            return response_data
+
+        except Exception as e:
+            if debug:
+                print(f"Response generation error: {e}")
+            raise HTTPException(status_code=500, detail=str(e))
+
+
+class RouteManager:
+    """Route management handler"""
+
+    def __init__(self, app: FastAPI):
+        self.app = app
+        self.simulator = ChatGPTSimulator()
+
+    def setup_routes(self):
+        """Setup API routes"""
+        routes = self._get_dynamic_routes()
+        if debug:
+            print(f"Registering routes: {routes}")
+        for path in routes:
+            self._register_route(path)
+
+    def _get_dynamic_routes(self) -> List[str]:
+        """Get dynamic routes based on environment variables"""
+        default_path = "/v1/chat/completions"
+        replace_chat = os.getenv("REPLACE_CHAT", "")
+        prefix_chat = os.getenv("PREFIX_CHAT", "")
+        append_chat = os.getenv("APPEND_CHAT", "")
+
+        if replace_chat:
+            return [path.strip() for path in replace_chat.split(",") if path.strip()]
+
+        routes = []
+        if prefix_chat:
+            routes.extend(f"{prefix.rstrip('/')}{default_path}"
+                          for prefix in prefix_chat.split(","))
+            return routes
+
+        if append_chat:
+            append_paths = [path.strip() for path in append_chat.split(",") if path.strip()]
+            routes = [default_path] + append_paths
+
+        return routes or [default_path]
+
+    def _register_route(self, path: str):
+        """Register a single route"""
+
+        @self.app.post(path)
+        async def chat_endpoint(request: Request) -> Dict[str, Any]:
+            try:
+                headers = dict(request.headers)
+                data = await request.json()
+                if debug:
+                    print(f"Request received...\r\n\tHeaders: {headers},\r\n\tData: {data}")
+                return await self.simulator.generate_response(headers, data)
+            except JSONDecodeError as e:
+                if debug:
+                    print(f"JSON decode error: {e}")
+                raise HTTPException(status_code=400, detail="Invalid JSON format")
+            except Exception as e:
+                if debug:
+                    print(f"Request processing error: {e}")
+                raise HTTPException(status_code=500, detail="Internal server error")
+
+
+@app.get("/")
+async def health_check() -> str:
+    """Health check endpoint"""
+    return "Service is running..."
+
+
+def run_server(host: str = "0.0.0.0", port: int = 7860):
+    """Run server with optimized configuration"""
+    config = ServerConfig.get_uvicorn_config(
+        app=app,
+        host=host,
+        port=port
+    )
+
+    server = uvicorn.Server(config)
+    server.run()
+
+
+if __name__ == "__main__":
+    route_manager = RouteManager(app)
+    route_manager.setup_routes()
+    port = int(os.getenv("PORT", "7860"))
+    run_server(port=port)