Files
999/scripts/g4f_local_server.py
T

130 lines
4.7 KiB
Python

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Local G4F API Server for Oracle 999
Provides an OpenAI-compatible /v1/chat/completions endpoint on port 1337
Routing specifically through verified live providers:
- Gemini
- CohereForAI_C4AI_Command
- HuggingSpace
- Yqcloud
"""
import json
from http.server import HTTPServer, BaseHTTPRequestHandler
import g4f
from g4f.client import Client
from g4f.Provider import (
Gemini,
CohereForAI_C4AI_Command,
HuggingSpace,
Yqcloud,
)
client = Client()
PRIORITY_PROVIDERS = [
('Gemini', Gemini),
('CohereForAI', CohereForAI_C4AI_Command),
('HuggingSpace', HuggingSpace),
('Yqcloud', Yqcloud),
]
class G4FHandler(BaseHTTPRequestHandler):
def do_OPTIONS(self):
self.send_response(200)
self.send_header('Access-Control-Allow-Origin', '*')
self.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
self.send_header('Access-Control-Allow-Headers', 'Content-Type, Authorization')
self.end_headers()
def do_GET(self):
self.send_response(200)
self.send_header('Content-Type', 'application/json')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps({'status': 'running', 'service': 'Oracle 999 G4F Server', 'providers': ['Gemini', 'CohereForAI', 'HuggingSpace', 'Yqcloud']}).encode('utf-8'))
def do_POST(self):
content_length = int(self.headers.get('Content-Length', 0))
post_data = self.rfile.read(content_length).decode('utf-8')
try:
req_json = json.loads(post_data)
messages = req_json.get('messages', [])
print(f"[G4F Server] Processing request ({len(messages)} messages)...")
response_text = None
for p_name, prov in PRIORITY_PROVIDERS:
try:
chat_res = client.chat.completions.create(
model=g4f.models.default,
provider=prov,
messages=messages,
timeout=12,
)
content = chat_res.choices[0].message.content
if content and len(content.strip()) > 15:
response_text = content.strip()
print(f"[G4F Server] Generated successfully via provider: {p_name}!")
break
except Exception as e:
print(f"[G4F Server] Provider {p_name} failed ({e}), trying next...")
if not response_text or len(response_text.strip()) == 0:
# Ultimate fallback to default client
try:
chat_res = client.chat.completions.create(
model=g4f.models.default,
messages=messages,
)
response_text = chat_res.choices[0].message.content
except Exception:
pass
if not response_text:
response_text = "Оракул открывает знаки: путь разворачивается перед вами. Сосредоточьтесь на ключевом символе и следуйте зову сердца."
resp_payload = {
"id": "chatcmpl-oracle999",
"object": "chat.completion",
"choices": [{
"index": 0,
"message": {
"role": "assistant",
"content": response_text
},
"finish_reason": "stop"
}]
}
self.send_response(200)
self.send_header('Content-Type', 'application/json; charset=utf-8')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps(resp_payload, ensure_ascii=False).encode('utf-8'))
except Exception as err:
print(f"[G4F Server] Handler error: {err}")
self.send_response(500)
self.send_header('Content-Type', 'application/json')
self.send_header('Access-Control-Allow-Origin', '*')
self.end_headers()
self.wfile.write(json.dumps({'error': str(err)}).encode('utf-8'))
def log_message(self, format, *args):
return
def run(port=1337):
server_address = ('127.0.0.1', port)
httpd = HTTPServer(server_address, G4FHandler)
print(f"🔮 [Oracle 999] High-Speed G4F Server is LIVE on http://127.0.0.1:{port}/v1/chat/completions")
try:
httpd.serve_forever()
except KeyboardInterrupt:
pass
httpd.server_close()
if __name__ == '__main__':
run()