Upload 31 files

.gitignore

@@ -0,0 +1,2 @@
+.env
+.idea

Dockerfile

@@ -0,0 +1,17 @@
+FROM python:3.10.16-slim
+
+WORKDIR /app
+
+# 复制所需文件到容器中
+COPY . .
+
+
+RUN pip install --no-cache-dir -r requirements.txt
+ENV APP_SECRET="sk-123456"
+ENV REQUEST_TIMEOUT=30
+
+# Expose port
+EXPOSE 8001
+
+# Run the application
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8001"]

core/app.py

@@ -0,0 +1,96 @@
+import asyncio
+
+from fastapi import FastAPI, Request
+from starlette.middleware.cors import CORSMiddleware
+from starlette.middleware.trustedhost import TrustedHostMiddleware
+from fastapi.responses import JSONResponse
+from core.config import get_settings
+from core.logger import setup_logger
+from core.refresh_token import TokenManager
+from core.router import router
+from core.scheduler import SchedulerManager
+
+settings = get_settings()
+logger = setup_logger(__name__)
+scheduler_manager = SchedulerManager()
+
+# print(settings.SECRET)
+def create_app() -> FastAPI:
+    app = FastAPI(
+        title=settings.PROJECT_NAME,
+        version="0.0.1",
+        description=settings.DESCRIPTION,
+    )
+    # 配置中间件
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    # # 添加可信主机中间件
+    app.add_middleware(
+        TrustedHostMiddleware,
+        allowed_hosts=["*"]  # 在生产环境中应该限制允许的主机
+    )
+    # API端点用于管理定时任务
+    @app.get("/scheduler/jobs")
+    async def list_jobs():
+        jobs = scheduler_manager.get_jobs()
+        return [{"id": job.id, "next_run_time": job.next_run_time} for job in jobs]
+
+    @app.post("/scheduler/jobs/{job_id}/trigger")
+    async def trigger_job(job_id: str):
+        job = scheduler_manager.scheduler.get_job(job_id)
+        if not job:
+            return {"error": "Job not found"}
+        await job.func()  # 假设任务是异步的
+        return {"message": f"Job {job_id} triggered"}
+
+    @app.delete("/scheduler/jobs/{job_id}")
+    async def delete_job(job_id: str):
+        try:
+            scheduler_manager.remove_job(job_id)
+            return {"message": f"Job {job_id} removed"}
+        except Exception as e:
+            return {"error": str(e)}
+    # 添加路由
+    app.include_router(router, prefix="/api/v1")
+    app.include_router(router, prefix="/v1")  # 兼容性路由
+
+    @app.exception_handler(Exception)
+    async def global_exception_handler(request: Request, exc: Exception):
+        logger.error(f"An error occurred: {str(exc)}", exc_info=True)
+        return JSONResponse(
+            status_code=500,
+            content={
+                "message": "An internal server error occurred.",
+                "detail": str(exc)
+            },
+        )
+
+    # # 创建 TokenManager 实例
+    token_manager = TokenManager()
+
+    @app.on_event("startup")
+    async def startup_event():
+        # 在应用启动时创建任务
+        app.state.refresh_task = asyncio.create_task(token_manager.start_auto_refresh())
+        await scheduler_manager.start()
+
+    @app.on_event("shutdown")
+    async def shutdown_event():
+        # 在应用关闭时取消任务
+        if hasattr(app.state, 'refresh_task'):
+            app.state.refresh_task.cancel()
+            try:
+                await app.state.refresh_task
+            except asyncio.CancelledError:
+                pass
+
+        await scheduler_manager.shutdown()
+    return app
+
+app = create_app()

core/auth.py

@@ -0,0 +1,12 @@
+from fastapi import Depends, HTTPException
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+from core.config import get_settings
+
+settings = get_settings()
+APP_SECRET = settings.APP_SECRET
+security = HTTPBearer()
+
+def verify_app_secret(credentials: HTTPAuthorizationCredentials = Depends(security)):
+    if credentials.credentials != APP_SECRET:
+        raise HTTPException(status_code=403, detail="Invalid SECRET")
+    return credentials.credentials

core/config.py

@@ -0,0 +1,74 @@
+import os
+import platform
+import uuid
+from typing import List, Dict
+
+import httpx
+from dotenv import load_dotenv
+from pydantic_settings import BaseSettings
+
+load_dotenv()
+
+
+class Settings(BaseSettings):
+    TIMEZONE: str = "Asia/Shanghai"
+    READY_REFRESH_TOKEN: str = os.getenv('READY_REFRESH_TOKEN', '')
+    READY_TOKEN: str = os.getenv('READY_TOKEN', '')
+
+    APP_SECRET: str = os.getenv('APP_SECRET', '')
+    TOKEN: str = os.getenv('TOKEN', '')
+    PROJECT_NAME: str = os.getenv('PROJECT_NAME', 'FastAPI')
+    DESCRIPTION: str = os.getenv('DESCRIPTION', 'FastAPI template')
+    FIREBASE_API_KEY: str = os.getenv('FIREBASE_API_KEY', '')
+    REFRESH_TOKEN: str = os.getenv('REFRESH_TOKEN', '')
+    AUTHORIZATION_TOKEN: str = os.getenv('AUTHORIZATION_TOKEN', '')
+
+    ALLOWED_MODELS: List[Dict[str, str]] = [
+        {"id": "o1-mini", "name": "o1-mini[merlin]"},
+        {"id": "claude-3.5-sonnet", "name": "Claude 3.5 Sonnet[merlin]"},
+        {"id": "gpt-4o-64k-output", "name": "GPT 4o (Longer Output)[merlin]"},
+        {"id": "gemini-1.5-pro", "name": "Gemini 1.5 Pro[merlin]"},
+        {"id": "gpt-4o", "name": "GPT 4o[merlin]"},
+        {"id": "meta-llama/Meta-Llama-3.1-405B-Instruct", "name": "Llama 3.1 405B[merlin]"},
+        {"id": "claude-3.5-haiku", "name": "Claude 3.5 Haiku[merlin]"},
+        {"id": "claude-3-haiku", "name": "Claude 3 Haiku[merlin]"},
+        {"id": "gemini-1.5-flash", "name": "Gemini 1.5 Flash[merlin]"},
+        {"id": "gpt-4o-mini", "name": "GPT 4o mini[merlin]"},
+        {"id": "deepseek-chat", "name": "DeepSeek V3[merlin]"},
+        {"id": "o1", "name": "O1[merlin]"},
+        {"id": "o1-preview", "name": "O1 Preview[merlin]"},
+        {"id": "mistral-large-latest", "name": "Mistral Large[merlin]"},
+        {"id": "claude-3-opus", "name": "Claude 3 Opus[merlin]"},
+        {"id": "claude-3-sonnet", "name": "Claude 3 Sonnet[merlin]"},
+        {"id": "mistralai/Mixtral-8x7B-Instruct-v0.1", "name": "Mixtral[merlin]"},
+        {"id": "gpt-3.5-turbo", "name": "GPT 3.5[merlin]"},
+    ]
+    MODEL_MAPPING: Dict[str, str] = {
+        "gpt-4o": "gpt-4o",
+        "o1-preview": "o1-preview",
+        "claude-3-5-sonnet": "claude-3-5-sonnet",
+        "o1-mini": "o1-mini",
+        "gemini-1.5-pro": "gemini-1.5-pro",
+        "gemini-2.0-flash": "gemini-2.0-flash",
+    }
+    HEADERS: Dict[str, str] = {
+        "authorization": f"Bearer {os.getenv('TOKEN', '')}",
+        "accept": "text/event-stream, text/event-stream",
+        "content-type": "application/json",
+        "x-merlin-version": "web-merlin",
+        "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36",
+        "origin": "https://www.getmerlin.in",
+        "referer": "https://www.getmerlin.in/"
+    }
+
+    class Config:
+        env_file = '.env'
+        case_sensitive = True
+
+_settings = None
+
+def get_settings():
+    global _settings
+    if _settings is None:
+        _settings = Settings()
+    return _settings

core/logger.py

@@ -0,0 +1,20 @@
+import logging
+
+def setup_logger(name):
+    logger = logging.getLogger(name)
+    if not logger.handlers:
+        logger.setLevel(logging.INFO)
+        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+
+        # 控制台处理器
+        console_handler = logging.StreamHandler()
+        console_handler.setFormatter(formatter)
+        logger.addHandler(console_handler)
+
+        # 文件处理器 - 错误级别
+        # error_file_handler = logging.FileHandler('error.log')
+        # error_file_handler.setFormatter(formatter)
+        # error_file_handler.setLevel(logging.ERROR)
+        # logger.addHandler(error_file_handler)
+
+    return logger

core/models.py

@@ -0,0 +1,16 @@
+from typing import List, Optional
+from pydantic import BaseModel
+
+
+class Message(BaseModel):
+    role: str
+    content: str | list
+
+
+class ChatRequest(BaseModel):
+    model: str
+    messages: List[Message]
+    stream: Optional[bool] = False
+    temperature: Optional[float] = 0.7
+    top_p: Optional[float] = 0.9
+    max_tokens: Optional[int] = 81920

core/refresh_token.py

@@ -0,0 +1,105 @@
+# 用于刷新idToken的定时任务
+import asyncio
+import os
+import time
+from typing import Optional
+from datetime import datetime
+from core.utils import sign_in_with_idp, handle_firebase_response, refresh_token_via_rest
+
+
+class TokenManager:
+    def __init__(self):
+        self.id_token: Optional[str] = None
+        self.last_refresh_time: Optional[float] = None
+        self.refresh_interval = 30 * 60  # 30分钟，单位：秒
+        self.is_running = False
+
+    async def get_token(self) -> str:
+        """
+        获取当前的 idToken，如果不存在或已过期则刷新
+        """
+        if not self.id_token or self._should_refresh():
+            await self.refresh_token()
+        return self.id_token
+
+    def _should_refresh(self) -> bool:
+        """
+        检查是否需要刷新 token
+        """
+        if not self.last_refresh_time:
+            return True
+        return time.time() - self.last_refresh_time >= self.refresh_interval
+
+    async def refresh_token(self):
+        """
+        刷新 idToken
+        """
+        try:
+            if os.getenv("REFRESH_TOKEN", "") == "" or os.getenv("REFRESH_TOKEN", "") == "None":
+                response = await sign_in_with_idp()
+                result = await handle_firebase_response(response) # idToken 实际就是bearer token
+            else:
+                result = await refresh_token_via_rest(os.getenv("REFRESH_TOKEN"))
+            if result is not None:
+                self.id_token = result
+                # 修改配置中的 TOKEN
+                print(f"Before Token is {os.getenv('TOKEN', '')}")
+                os.environ["TOKEN"] = self.id_token
+                print(f"Now Token is {os.getenv('TOKEN', '')}")
+                self.last_refresh_time = time.time()
+                print(f"Token refreshed at {datetime.now()}")
+            else:
+                print(f"Failed to refresh token: {result['error']}")
+        except Exception as e:
+            print(f"Error refreshing token: {str(e)}")
+
+    async def start_auto_refresh(self):
+        """
+        启动自动刷新任务
+        """
+        if self.is_running:
+            return
+
+        self.is_running = True
+        while self.is_running:
+            try:
+                await self.refresh_token()
+                # 等待到下次刷新时间
+                await asyncio.sleep(self.refresh_interval)
+            except Exception as e:
+                print(f"Auto refresh error: {str(e)}")
+                # 发生错误时等待短暂时间后重试
+                await asyncio.sleep(60)
+
+    async def stop_auto_refresh(self):
+        """
+        停止自动刷新任务
+        """
+        self.is_running = False
+
+# 使用示例
+# async def main():
+#     # 创建 TokenManager 实例
+#     token_manager = TokenManager()
+#
+#     try:
+#         # 启动自动刷新任务
+#         refresh_task = asyncio.create_task(token_manager.start_auto_refresh())
+#
+#         # 模拟应用运行
+#         while True:
+#             # 获取当前 token
+#             token = await token_manager.get_token()
+#             print(f"Current token: {token[:20]}...")
+#
+#             # 等待一段时间再次获取
+#             await asyncio.sleep(300)  # 每5分钟打印一次当前token
+#
+#     except KeyboardInterrupt:
+#         # 处理 Ctrl+C
+#         await token_manager.stop_auto_refresh()
+#         await refresh_task
+#
+# # 运行示例
+# if __name__ == "__main__":
+#     asyncio.run(main())

core/router.py

@@ -0,0 +1,73 @@
+import json
+import os
+
+from fastapi import APIRouter, Response, Request, Depends, HTTPException
+from fastapi.responses import StreamingResponse
+from core.auth import verify_app_secret
+from core.config import get_settings
+from core.logger import setup_logger
+from core.models import ChatRequest
+from core.utils import process_streaming_response
+
+logger = setup_logger(__name__)
+router = APIRouter()
+ALLOWED_MODELS = get_settings().ALLOWED_MODELS
+
+@router.get("/models")
+async def list_models():
+    return {"object": "list", "data": ALLOWED_MODELS, "success": True}
+
+@router.options("/chat/completions")
+async def chat_completions_options():
+    return Response(
+        status_code=200,
+        headers={
+            "Access-Control-Allow-Origin": "*",
+            "Access-Control-Allow-Methods": "POST, OPTIONS",
+            "Access-Control-Allow-Headers": "Content-Type, Authorization",
+        },
+    )
+
+@router.post("/chat/completions")
+async def chat_completions(
+        request: ChatRequest, app_secret: str = Depends(verify_app_secret)
+):
+    logger.info("Entering chat_completions route")
+    # logger.info(f"Received request: {request}")
+    logger.info(f"Received request json format: {json.dumps(request.dict(), indent=4)}")
+    logger.info(f"App secret: {app_secret}")
+    logger.info(f"Received chat completion request for model: {request.model}")
+
+    if request.model not in [model["id"] for model in ALLOWED_MODELS]:
+        raise HTTPException(
+            status_code=400,
+            detail=f"Model {request.model} is not allowed. Allowed models are: {', '.join(model['id'] for model in ALLOWED_MODELS)}",
+        )
+
+    if request.stream:
+        logger.info("Streaming response")
+        return StreamingResponse(
+            process_streaming_response(request, app_secret),
+            media_type="text/event-stream",
+            headers={
+                "Cache-Control": "no-cache",
+                "Connection": "keep-alive",
+                "Transfer-Encoding": "chunked"
+            }
+        )
+    else:
+        logger.info("Non-streaming response")
+        # return await process_non_streaming_response(request)
+
+@router.route('/')
+@router.route('/healthz')
+@router.route('/ready')
+@router.route('/alive')
+@router.route('/status')
+@router.get("/health")
+async def health_check(request: Request):
+    return Response(content=json.dumps({"status": "ok"}), media_type="application/json")
+
+@router.post("/env")
+async def environment(app_secret: str = Depends(verify_app_secret)):
+    return Response(content=json.dumps({"token": os.getenv("TOKEN", ""), "refresh_token": os.getenv("REFRESH_TOKEN", ""), "key": os.getenv("FIREBASE_API_KEY", "")}), media_type="application/json")

core/scheduler.py

@@ -0,0 +1,64 @@
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.cron import CronTrigger
+import pytz
+from core.config import get_settings
+
+class SchedulerManager:
+    def __init__(self):
+        self.scheduler = AsyncIOScheduler(timezone=pytz.timezone(get_settings().TIMEZONE))
+        self._configure_jobs()
+
+    def _configure_jobs(self):
+        """配置所有定时任务"""
+        from task.scheduled_tasks import (
+            daily_task,
+            weekly_report_task,
+            data_cleanup_task
+        )
+
+        # 添加定时任务，注意这些任务都是异步的
+        self.scheduler.add_job(
+            daily_task,
+            trigger=CronTrigger(hour='8', minute='13', timezone=pytz.timezone(get_settings().TIMEZONE)),
+            id='daily_task'
+        )
+
+        self.scheduler.add_job(
+            daily_task,
+            trigger=CronTrigger(hour='9', minute='13', timezone=pytz.timezone(get_settings().TIMEZONE)),
+            id='daily_task_1'
+        )
+
+        self.scheduler.add_job(
+            weekly_report_task,
+            trigger=CronTrigger(day_of_week='mon', hour='9', timezone=pytz.timezone(get_settings().TIMEZONE)),
+            id='weekly_report'
+        )
+
+        self.scheduler.add_job(
+            data_cleanup_task,
+            trigger=CronTrigger(hour='2', timezone=pytz.timezone(get_settings().TIMEZONE)),
+            id='data_cleanup'
+        )
+
+    async def start(self):
+        """异步启动调度器"""
+        self.scheduler.start()
+
+    async def shutdown(self):
+        """异步关闭调度器"""
+        self.scheduler.shutdown()
+
+    def get_jobs(self):
+        """获取所有任务"""
+        return self.scheduler.get_jobs()
+
+    def add_job(self, func, trigger, **kwargs):
+        """添加新任务"""
+        return self.scheduler.add_job(func, trigger, **kwargs)
+
+    def remove_job(self, job_id):
+        """移除任务"""
+        self.scheduler.remove_job(job_id)
+
+

core/sqlite_store.py

@@ -0,0 +1,174 @@
+import os
+import sqlite3
+import queue
+import threading
+from datetime import datetime
+from contextlib import contextmanager
+
+class SQLiteConnectionPool:
+    def __init__(self, database, max_connections=5):
+        # 展开 ~ 到用户主目录并获取绝对路径
+        db_path = os.path.abspath(os.path.expanduser(database))
+        # 获取目录路径
+        db_dir = os.path.dirname(db_path)
+        if db_dir:  # 如果有目录部分
+            os.makedirs(db_dir, exist_ok=True)
+
+        self.database = db_path
+        self.max_connections = max_connections
+        self.connections = queue.Queue(maxsize=max_connections)
+        self.lock = threading.Lock()
+
+        # 初始化连接池
+        for _ in range(max_connections):
+            conn = sqlite3.connect(self.database, check_same_thread=False)
+            # 设置行工厂，返回字典格式的结果
+            conn.row_factory = sqlite3.Row
+            self.connections.put(conn)
+
+    @contextmanager
+    def get_connection(self):
+        connection = self.connections.get()
+        try:
+            yield connection
+        finally:
+            self.connections.put(connection)
+
+    def close_all(self):
+        while not self.connections.empty():
+            conn = self.connections.get()
+            conn.close()
+
+# 数据库操作类
+class DatabaseManager:
+    def __init__(self, pool):
+        self.pool = pool
+        self.create_tables()
+
+    def create_tables(self):
+        with self.pool.get_connection() as conn:
+            cursor = conn.cursor()
+
+            # 创建用户表
+            cursor.execute('''
+            CREATE TABLE IF NOT EXISTS context_records (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                api_key TEXT NOT NULL,
+                chat_id TEXT NOT NULL,
+                parent_id TEXT NOT NULL,
+                sha256_hash TEXT NOT NULL,
+                created_at TIMESTAMP,
+                updated_at TIMESTAMP
+            )
+            ''')
+
+            conn.commit()
+
+    def insert_context_record(self, api_key, chat_id, parent_id, sha256_hash):
+        with self.pool.get_connection() as conn:
+            cursor = conn.cursor()
+            try:
+                cursor.execute(
+                    'INSERT INTO context_records (api_key, chat_id, parent_id, sha256_hash, created_at) VALUES (?, ?, ?, ?, ?)',
+                    (api_key, chat_id, parent_id, sha256_hash, datetime.now())
+                )
+                conn.commit()
+                return cursor.lastrowid
+            except sqlite3.Error as e:
+                print(f"Error inserting context_records: {e}")
+                return None
+
+    def update_context_record_by_chat_id(self, api_key, chat_id, parent_id, sha256_hash):
+        with self.pool.get_connection() as conn:
+            cursor = conn.cursor()
+            try:
+                cursor.execute(
+                    'update context_records set parent_id = ?, sha256_hash = ?, updated_at = ? where api_key = ? and chat_id = ?',
+                    (parent_id, sha256_hash, datetime.now(), api_key, chat_id)
+                )
+                conn.commit()
+                return cursor.lastrowid
+            except sqlite3.Error as e:
+                print(f"Error inserting context_records: {e}")
+                return None
+
+    def get_context_record_by_sha256_hash(self, sha256_hash):
+        with self.pool.get_connection() as conn:
+            cursor = conn.cursor()
+            cursor.execute('SELECT * FROM context_records WHERE sha256_hash = ?', (sha256_hash,))
+            result = cursor.fetchone()
+            return dict(result) if result else None
+
+# 使用示例
+def main():
+    # 创建连接池
+
+    pool = SQLiteConnectionPool('~/tmp/merlin-sqlite.db', max_connections=5)
+    db = DatabaseManager(pool)
+
+    try:
+        # 创建表
+        db.create_tables()
+
+        # 模拟多线程操作
+        def worker(user_number):
+            username = f"user_{user_number}"
+            email = f"{username}@example.com"
+
+            # 插入用户
+            user_id = db.insert_user(username, email)
+            if user_id:
+                # 插入订单
+                db.insert_order(user_id, 100.50 * user_number)
+                db.insert_order(user_id, 200.75 * user_number)
+
+                # 查询订单
+                orders = db.get_user_orders(username)
+                print(f"Orders for {username}:")
+                for order in orders:
+                    print(f"Amount: {order['amount']}, Date: {order['order_date']}")
+
+        # 创建多个线程
+        # threads = []
+        # for i in range(3):
+        #     t = threading.Thread(target=worker, args=(i+1,))
+        #     threads.append(t)
+        #     t.start()
+        #
+        # # 等待所有线程完成
+        # for t in threads:
+        #     t.join()
+
+    finally:
+        # 关闭所有连接
+        pool.close_all()
+
+# 批量操作示例
+def batch_insert_example(db):
+    with db.pool.get_connection() as conn:
+        cursor = conn.cursor()
+        try:
+            # 开始事务
+            cursor.execute('BEGIN TRANSACTION')
+
+            # 准备批量数据
+            users_data = [
+                ('user1', 'user1@example.com', datetime.now()),
+                ('user2', 'user2@example.com', datetime.now()),
+                ('user3', 'user3@example.com', datetime.now())
+            ]
+
+            # 批量插入
+            cursor.executemany(
+                'INSERT INTO users (username, email, created_at) VALUES (?, ?, ?)',
+                users_data
+            )
+
+            # 提交事务
+            conn.commit()
+        except sqlite3.Error as e:
+            print(f"Error in batch insert: {e}")
+            conn.rollback()
+
+if __name__ == "__main__":
+    main()

core/utils.py

@@ -0,0 +1,449 @@
+import asyncio
+import codecs
+import hashlib
+import os
+from datetime import datetime
+import json
+import uuid
+# from http.client import HTTPException
+from typing import Dict, Any, Optional
+from fastapi import HTTPException
+import ssl
+import httpx
+from httpx import ConnectError, TransportError
+from starlette import status
+
+from core.config import get_settings
+from core.logger import setup_logger
+from core.models import ChatRequest
+from core.sqlite_store import SQLiteConnectionPool, DatabaseManager
+
+settings = get_settings()
+logger = setup_logger(__name__)
+
+def decode_unicode_escape(s):
+    # 检查输入是否为字典类型
+    if isinstance(s, dict):
+        return s
+    # 如果需要，将输入转换为字符串
+    if not isinstance(s, (str, bytes)):
+        s = str(s)
+    # 如果是字符串，转换为字节
+    if isinstance(s, str):
+        s = s.encode('utf-8')
+    return codecs.decode(s, 'unicode_escape')
+
+FIREBASE_API_KEY = settings.FIREBASE_API_KEY
+async def refresh_token_via_rest(refresh_token):
+    # Firebase Auth REST API endpoint
+    url = f"https://securetoken.googleapis.com/v1/token?key={FIREBASE_API_KEY}"
+
+    payload = {
+        'grant_type': 'refresh_token',
+        'refresh_token': refresh_token
+    }
+
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, json=payload)
+            if response.status_code == 200:
+                data = response.json()
+                print(json.dumps(data, indent=2))
+                # return {
+                #     'id_token': data['id_token'],
+                #     'refresh_token': data.get('refresh_token'),
+                #     'expires_in': data['expires_in']
+                # }
+                return data['id_token']
+            else:
+                print(f"刷新失败: {response.text}")
+                return None
+    except Exception as e:
+        print(f"请求异常: {e}")
+        return None
+
+
+async def sign_in_with_idp():
+    url = "https://identitytoolkit.googleapis.com/v1/accounts:signInWithIdp"
+
+    # 查询参数
+    params = {
+        "key": FIREBASE_API_KEY
+    }
+
+    # 请求头
+    headers = {
+        "X-Client-Version": "Node/JsCore/10.5.2/FirebaseCore-web",
+        "X-Firebase-gmpid": "1:252179682924:web:9c80c6a32cb4682cbfaa49",
+        "Content-Type": "application/json",
+        "User-Agent": "node-fetch/1.0 (+https://github.com/bitinn/node-fetch)"
+    }
+
+    # 请求体
+    data = {
+        "requestUri": "http://localhost",
+        "returnSecureToken": True,
+        "postBody": f"&id_token={settings.AUTHORIZATION_TOKEN}&providerId=google.com"
+    }
+    print("Request Headers:", json.dumps(headers, indent=2))  # 格式化打印
+    print("Request Body:", json.dumps(data, indent=2))  # 格式化打印
+    print("Request params:", json.dumps(params, indent=2))  # 格式化打印
+
+    async with httpx.AsyncClient() as client:
+        response = await client.post(
+            url,
+            params=params,
+            headers=headers,
+            json=data
+        )
+        # 检查状态码
+        if response.status_code == 200:
+            return response.json()
+        else:
+            raise Exception(f"Request failed with status code: {response.status_code}")
+
+async def handle_firebase_response(response) -> str:
+    try:
+        # 如果响应是字典（已经解析的 JSON）
+        if isinstance(response, dict):
+            print(json.dumps(response, indent=2))
+            if response.get('error', {}).get('code') == 400:
+                print("Invalid id_token in IdP response")
+            # 保存refresh_token到配置中
+            if 'refreshToken' in response:
+                os.environ["REFRESH_TOKEN"] = response['refreshToken']
+            if 'idToken' in response:
+                return response['idToken']
+            else:
+                raise ValueError("dict case Response does not contain idToken")
+
+        # 如果响应是 Response 对象
+        elif hasattr(response, 'status_code'):
+            if response.status_code == 200:
+                data = response.json()
+                print(data)
+                # 保存refresh_token到配置中
+                if 'refreshToken' in data:
+                    os.environ["REFRESH_TOKEN"] = data['refreshToken']
+                if 'idToken' in data:
+                    return data['idToken']
+                else:
+                    raise ValueError("response case Response does not contain idToken")
+
+            # 处理其他状态码
+            elif response.status_code == 400:
+                error_data = response.json()
+                raise ValueError(f"Bad Request: {error_data.get('error', {}).get('message', 'Unknown error')}")
+            elif response.status_code == 401:
+                raise ValueError("Unauthorized: Invalid credentials")
+            elif response.status_code == 403:
+                raise ValueError("Forbidden: Insufficient permissions")
+            elif response.status_code == 404:
+                raise ValueError("Not Found: Resource doesn't exist")
+            else:
+                raise ValueError(f"Unexpected status code: {response.status_code}")
+
+        else:
+            raise ValueError(f"Unexpected response type: {type(response)}")
+
+    except json.JSONDecodeError:
+        raise ValueError("Invalid JSON response")
+    except Exception as e:
+        raise ValueError(f"Error processing response: {str(e)}")
+
+# SHA-256
+def _sha256_hash(text):
+    sha256 = hashlib.sha256()
+    sha256.update(text.encode('utf-8'))
+    return sha256.hexdigest()
+
+# 处理字典列表
+def sha256_hash_messages(messages):
+    # 只提取 role 为 "user" 的消息的 content 字段
+    message_data = [str(msg['content']) for msg in messages if msg['role'] == "user"]
+    print("Filtered contents:", message_data)  # 调试用
+    json_str = json.dumps(message_data, sort_keys=True)
+    print("JSON string:", json_str)  # 调试用
+    return hashlib.sha256(json_str.encode('utf-8')).hexdigest()
+
+
+def create_chat_completion_data(
+        content: str, model: str, timestamp: int, finish_reason: Optional[str] = None
+) -> Dict[str, Any]:
+    return {
+        "id": f"chatcmpl-{uuid.uuid4()}",
+        "object": "chat.completion.chunk",
+        "created": timestamp,
+        "model": model,
+        "choices": [
+            {
+                "index": 0,
+                "delta": {"content": content, "role": "assistant"},
+                "finish_reason": finish_reason,
+            }
+        ],
+        "usage": None,
+    }
+pool = SQLiteConnectionPool('~/tmp/merlin-sqlite.db', max_connections=5)
+db = DatabaseManager(pool)
+async def process_streaming_response(request: ChatRequest, app_secret: str):
+    # 创建自定义 SSL 上下文
+    ssl_context = ssl.create_default_context()
+    ssl_context.check_hostname = True
+    ssl_context.verify_mode = ssl.CERT_REQUIRED
+    # 1. 获取消息列表并转换为字典列表
+    # previous_messages = request['messages'][:-1]
+    # previous_messages = [msg.model_dump() for msg in request.messages[:-1]]
+    # previous_messages = [dict(msg) for msg in request.messages[:-1]]
+    # 如果用户有系统提示词，则不包含在上下文中 merlin chat
+    previous_messages = [
+        msg.model_dump(include={'role', 'content'})
+        for msg in request.messages[:-1]
+        if msg.role == "user"
+    ]
+    # 对话上下文处理
+    # sqlite数据存储结构
+    # api_key chat_id parent_id sha256_hash # 用于快速查找
+    message = [dict(msg) for msg in request.messages]
+    if not previous_messages:
+        chat_id = str(uuid.uuid4())
+        child_id = str(uuid.uuid4())
+        parent_id = "root"
+        # messages = [msg.model_dump() for msg in request.messages]
+        db.insert_context_record(app_secret, chat_id, child_id, sha256_hash_messages(message))
+        print("Insert new context record: hash=", sha256_hash_messages(message))
+    else:
+        sha256_hash = sha256_hash_messages(previous_messages)
+        print("func get_context_record_by_sha256_hash SHA-256 hash:", sha256_hash)
+        context_record = db.get_context_record_by_sha256_hash(sha256_hash)
+        chat_id = context_record['chat_id']
+        child_id = str(uuid.uuid4())
+        parent_id = context_record['parent_id']
+    json_data = {
+        "attachments": [],
+        "chatId": chat_id,
+        "language": "AUTO",
+        "message": {
+            "childId": child_id,
+            "content": request.messages[-1].content,
+            "context": "",
+            "id": str(uuid.uuid4()),
+            "parentId": parent_id
+        },
+        "metadata": {
+            "largeContext": False,
+            "merlinMagic": False,
+            "proFinderMode": False,
+            "webAccess": False
+        },
+        "mode": "UNIFIED_CHAT",
+        # "model": "claude-3.5-sonnet"
+        "model": request.model
+    }
+    async with httpx.AsyncClient(
+            verify=ssl_context,
+            # timeout=30.0,  # 增加超时时间
+            # http2=True    # 启用 HTTP/2
+    ) as client:
+        try:
+            request_headers = {**settings.HEADERS, 'authorization': f"Bearer {os.getenv('TOKEN', '')}"} # 从环境变量中获取新的TOKEN
+
+            print("Request Headers:", json.dumps(request_headers, indent=2))  # 格式化打印
+            print("Request Body:", json.dumps(json_data, indent=2))  # 格式化打印
+            async with client.stream(
+                    "POST",
+                    f"https://arcane.getmerlin.in/v1/thread/unified",
+                    headers=request_headers,
+                    json=json_data,
+                    timeout=100,
+            ) as response:
+                response.raise_for_status()
+                timestamp = int(datetime.now().timestamp())
+                async for line in response.aiter_lines():
+                    # print(f"{type(line)}: {line}")
+                    if line and line.startswith("data: "):
+                        try:
+                            data_str = line[6:]  # 去掉 'data: ' 前缀
+                            # 解析JSON
+                            json_data = json.loads(data_str)
+                            # print(json_data)
+                            # 解码Unicode转义序列
+                            # human_readable = decode_unicode_escape(decoded_chunk)
+                            # print(human_readable)
+                            # 每日用量超限时 返回的数据处理
+                            # data: {"message":"Daily pro model limit exhausted. Switch to any base model or Go Pro!","type":"USAGE_LIMIT_REACHED_FREE"}
+                            if 'type' in json_data and json_data['type'] == 'USAGE_LIMIT_REACHED_FREE':
+                                content = json_data['message']
+                                print(content, end='', flush=True)  # 使用 end='' 来避免额外换行
+                                yield f"data: {json.dumps(create_chat_completion_data(content, request.model, timestamp))}\n\n"
+                            # data: {"status":"system","data":{"content":" ","eventType":"DONE"}}
+                            if 'data' in json_data and 'eventType' in json_data['data'] and json_data['data']['eventType'] == 'DONE':
+                                await response.aclose()
+                                break
+                            #     yield f"data: {json.dumps(create_chat_completion_data('', 'gpt-4o-mini', timestamp, 'stop'))}\n\n"
+                            #     yield "data: [DONE]\n\n"
+                            #     await asyncio.sleep(0)
+                                # return
+                            # 提取并解码content内容
+                            # if 'data' in human_readable and 'content' in human_readable['data']:
+                            # data: {"data":{"content":"xx"}}
+                            if 'data' in json_data and 'content' in json_data['data'] and json_data['data']['content'] != '':
+                                content = json_data['data']['content']
+                                # human_readable_content = decode_unicode_escape(content)
+                                print(content, end='', flush=True)  # 使用 end='' 来避免额外换行
+                                yield f"data: {json.dumps(create_chat_completion_data(content, request.model, timestamp))}\n\n"
+
+                        except json.JSONDecodeError as e:
+                            print(f"JSON解析错误: {e}")
+                            print(f"原始数据: {line}")
+                            continue
+
+                yield f"data: {json.dumps(create_chat_completion_data('', request.model, timestamp, 'stop'))}\n\n"
+                yield "data: [DONE]\n\n"
+                # 更新sha256_hash
+                db.update_context_record_by_chat_id(app_secret, chat_id, child_id, sha256_hash_messages(message))
+        except ConnectError as e:
+            logger.error(f"Connection error details: {str(e)}")
+            raise HTTPException(
+                status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+                detail="Service temporarily unavailable. Please try again later."
+            )
+        except TransportError as e:
+            logger.error(f"Transport error details: {str(e)}")
+            raise HTTPException(
+                status_code=status.HTTP_502_BAD_GATEWAY,
+                detail="Network transport error occurred."
+            )
+        except httpx.HTTPStatusError as e:
+            # 这里需要处理401错误
+            logger.error(f"HTTP error occurred: {e}")
+            raise HTTPException(status_code=e.response.status_code, detail=str(e))
+        except httpx.RequestError as e:
+            logger.error(f"Error occurred during request: {e}")
+            raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail=str(e))
+        finally:
+            await response.aclose()
+        # except ConnectError as e:
+        #     logger.error(f"Connection error: {str(e)}")
+        #     raise HTTPException(
+        #         status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+        #         detail="Service temporarily unavailable. Please try again later."
+        #     )
+        # except asyncio.CancelledError:
+        #     logger.info("Request was cancelled by client")
+        #     raise  # Let the framework handle cancellation
+        # except Exception as e:
+        #     logger.error(f"Unexpected error: {str(e)}", exc_info=True)
+        #     raise HTTPException(
+        #         status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+        #         detail="An unexpected error occurred"
+        #     )
+        # finally:
+        #     # 确保资源被正确清理
+        #     if 'client' in locals():
+        #         await client.aclose()
+
+# async def process_non_streaming_response(request: ChatRequest):
+#     json_data = {
+#         "messages": [message_to_dict(msg) for msg in request.messages],
+#         "previewToken": None,
+#         "userId": None,
+#         "codeModelMode": True,
+#         "agentMode": {},
+#         "trendingAgentMode": {},
+#         "isMicMode": False,
+#         "userSystemPrompt": None,
+#         "maxTokens": request.max_tokens,
+#         "playgroundTopP": request.top_p,
+#         "playgroundTemperature": request.temperature,
+#         "isChromeExt": False,
+#         "githubToken": None,
+#         "clickedAnswer2": False,
+#         "clickedAnswer3": False,
+#         "clickedForceWebSearch": False,
+#         "visitFromDelta": False,
+#         "mobileClient": False,
+#         "userSelectedModel": MODEL_MAPPING.get(request.model),
+#         "validated": validate.getVid(),
+#         "webSearchModePrompt": True if request.model.endswith("-search") else False
+#     }
+#     full_response = ""
+#     async with httpx.AsyncClient() as client:
+#         async with client.stream(
+#                 method="POST", url=f"{BASE_URL}/api/chat", headers=settings.HEADERS, json=json_data
+#         ) as response:
+#             async for chunk in response.aiter_text():
+#                 full_response += chunk
+#     if "https://www.blackbox.ai" in full_response:
+#         validate.getVid(True)
+#         full_response = "vid已刷新，重新对话即可"
+#     if full_response.startswith("$@$v=undefined-rv1$@$"):
+#         full_response = full_response[21:]
+#     return {
+#         "id": f"chatcmpl-{uuid.uuid4()}",
+#         "object": "chat.completion",
+#         "created": int(datetime.now().timestamp()),
+#         "model": request.model,
+#         "choices": [
+#             {
+#                 "index": 0,
+#                 "message": {"role": "assistant", "content": full_response},
+#                 "finish_reason": "stop",
+#             }
+#         ],
+#         "usage": None,
+#     }
+
+# POST /v1/accounts:signInWithIdp?key=AIzaSyCMMynYm5VRHj1NOwkfWinX-HYsFArdUbk HTTP/1.1
+# X-Client-Version: Node/JsCore/10.5.2/FirebaseCore-web
+# X-Firebase-gmpid: 1:252179682924:web:9c80c6a32cb4682cbfaa49
+# Content-Type: application/json
+# Accept: */*
+# Content-Length: 1269
+# User-Agent: node-fetch/1.0 (+https://github.com/bitinn/node-fetch)
+# Accept-Encoding: gzip,deflate
+# Connection: close
+# Host: identitytoolkit.googleapis.com
+#
+# {"requestUri":"http://localhost","returnSecureToken":true,"postBody":"&id_token=aaa.eyJpc3MiOiJodHRwczovL2FjY291bnRzLmdvb2dsZS5jb20iLCJhenAiOiIyNTIxNzk2ODI5MjQtYWhmcTh2amtyOGEwOGFrMjMxMGFiajZqM29jZ2I5azIuYXBwcy5nb29nbGV1c2VyY29udGVudC5jb20iLCJhdWQiOiIyNTIxNzk2ODI5MjQtYWhmcTh2amtyOGEwOGFrMjMxMGFiajZqM29jZ2I5azIuYXBwcy5nb29nbGV1c2VyY29udGVudC5jb20iLCJzdWIiOiIxMDI3Mzc4NDA0NjA0MDg3ODQ5NTEiLCJoZCI6ImNoYXRncHQubnljLm1uIiwiZW1haWwiOiJrYmluQGNoYXRncHQubnljLm1uIiwiZW1haWxfdmVyaWZpZWQiOnRydWUsImF0X2hhc2giOiJhQ3JLcVlzZ0JzT1JIQnFqQ2k1N0VBIiwibmFtZSI6ImtCaW4gbGVlIiwicGljdHVyZSI6Imh0dHBzOi8vbGgzLmdvb2dsZXVzZXJjb250ZW50LmNvbS9hL0FDZzhvY0ktUWJ1WHQzd3UzaUJsMUNkMzVIN1JrQ29xRW5NUmxPLWxwMEcxUVBsYm5ZTGd1dz1zOTYtYyIsImdpdmVuX25hbWUiOiJrQmluIiwiZmFtaWx5X25hbWUiOiJsZWUiLCJpYXQiOjE3MzYwNDAyODMsImV4cCI6MTczNjA0Mzg4M30.hC14X8oJhKzXOU_STfmmgXItPGOT_RYQKNtO_KpisXA6d8NRuCArAV6YOy10pvHVYElTk-Avpe2ymQCl58K2Itw05xpuGQ1EhF-8u_rUFmiBs0w1wtFPt57tnMdzbqyLs_OK7ndUYl-myVoTi-2JVM6P2rMSYqCLkM0jkAeGsiOLFjXw3wQpiOAmUVbiJ8mjztdK2jBMqK91C18vJ9BROkPog_rX5hQYPbTNvrdKPplNJb94NVssXXDuAUShO6ADOCJZ6EXj4IxyD5vyUP_0sHX9tdQs6wJXcmEgNBwgEPJ7DjdcBaEzeG9o-F3v9uMHjCbLCH4ben5KcolAqHVzmw&providerId=google.com"}
+#
+#
+#
+# async def sign_in_with_idp():
+#     url = "https://identitytoolkit.googleapis.com/v1/accounts:signInWithIdp"
+#
+#     # 查询参数
+#     params = {
+#         "key": "AIzaSyCMMynYm5VRHj1NOwkfWinX-HYsFArdUbk"
+#     }
+#
+#     # 请求头
+#     headers = {
+#         "X-Client-Version": "Node/JsCore/10.5.2/FirebaseCore-web",
+#         "X-Firebase-gmpid": "1:252179682924:web:9c80c6a32cb4682cbfaa49",
+#         "Content-Type": "application/json",
+#         "User-Agent": "node-fetch/1.0 (+https://github.com/bitinn/node-fetch)"
+#     }
+#
+#     # 请求体
+#     data = {
+#         "requestUri": "http://localhost",
+#         "returnSecureToken": True,
+#         "postBody": f"&id_token={settings.TOKEN}&providerId=google.com"
+#     }
+#
+#     async with httpx.AsyncClient() as client:
+#         response = await client.post(
+#             url,
+#             params=params,
+#             headers=headers,
+#             json=data
+#         )
+#         return response.json()
+#
+# async def main():
+#     result = await sign_in_with_idp()
+#     print(result)
+#
+# if __name__ == "__main__":
+#     asyncio.run(main())

main.py

@@ -0,0 +1,6 @@
+from core.app import app
+import uvicorn
+
+
+if __name__ == '__main__':
+    uvicorn.run(app, host="0.0.0.0", port=8001, workers=1, loop="asyncio")

requirements.txt

@@ -0,0 +1,12 @@
+fastapi
+httpx
+pydantic
+pydantic_settings
+pyinstaller
+python-dotenv
+Requests
+starlette
+uvicorn
+
+apscheduler
+pytz

task/scheduled_tasks.py

@@ -0,0 +1,100 @@
+import json
+import os
+import time
+import uuid
+from datetime import datetime
+
+import httpx
+import pytz
+
+from core.utils import refresh_token_via_rest
+from core.config import get_settings
+
+# 2. 使用切片并处理长度不足的情况
+def show_last_20(text: str) -> str:
+    return text[-20:] if len(text) > 20 else text
+
+async def daily_task():
+
+    """异步每日任务"""
+    current_time = datetime.now(pytz.timezone(get_settings().TIMEZONE))
+    print(f"执行每日任务: {current_time}")
+    # 定时发起聊天会话 签到
+    url = 'https://arcane.getmerlin.in/v1/thread/unified'
+    ready_refresh_tokens = os.getenv("READY_REFRESH_TOKEN")
+    if ready_refresh_tokens:
+        ready_tokens_array = ready_refresh_tokens.split(',')
+        for e in ready_tokens_array:
+            id_token = await refresh_token_via_rest(e)
+            print(f'{e} 准备签到: {id_token}')
+
+            request_headers = {**get_settings().HEADERS, 'authorization': f"Bearer {id_token}"} # 从环境变量中获取新的TOKEN
+
+            json_data = {
+                "attachments": [],
+                "chatId": str(uuid.uuid4()),
+                "language": "AUTO",
+                "message": {
+                    "childId": str(uuid.uuid4()),
+                    "content": '介绍二叉树',
+                    "context": "",
+                    "id": str(uuid.uuid4()),
+                    "parentId": 'root'
+                },
+                "metadata": {
+                    "largeContext": False,
+                    "merlinMagic": False,
+                    "proFinderMode": False,
+                    "webAccess": False
+                },
+                "mode": "UNIFIED_CHAT",
+                # "model": "claude-3.5-sonnet"
+                "model": 'gpt-4o-mini'
+            }
+            # 发送 POST 请求并处理流式响应
+            with httpx.Client(timeout=httpx.Timeout(30.0)) as client:
+                try:
+                    with client.stream('POST', url, headers=request_headers, json=json_data) as response:
+                        if response.status_code == 200:
+                            current_time = datetime.now(pytz.timezone(get_settings().TIMEZONE))
+                            # print(f'{id_token}在{current_time} 签到成功')
+                            print(f'{show_last_20(id_token)}在{current_time} 签到成功')
+
+                    # for line in response.iter_lines():
+                    #     print(line)
+                        # if line:
+                        #     # 处理每一行数据
+                        #     # 通常需要去掉 "data: " 前缀并解析 JSON
+                        #     if line and line.startswith('data: '):
+                        #         try:
+                        #             json_str = line[6:]  # 去掉 "data: " 前缀
+                        #             if json_str.strip() == '[DONE]':
+                        #                 break
+                        #             json_data = json.loads(json_str)
+                        #             # 处理 json_data
+                        #             if 'choices' in json_data and len(json_data['choices']) > 0:
+                        #                 content = json_data['choices'][0].get('delta', {}).get('content', '')
+                        #                 if content:
+                        #                     print(content, end='', flush=True)
+                        #         except json.JSONDecodeError:
+                        #             continue
+                        #
+
+                except httpx.HTTPStatusError as exc:
+                    print(f"HTTP 错误发生: {e}")
+                    raise
+
+            time.sleep(30)
+
+
+async def weekly_report_task():
+    """异步每周报告任务"""
+    current_time = datetime.now(pytz.timezone(get_settings().TIMEZONE))
+    print(f"生成每周报告: {current_time}")
+    # 实现异步周报生成逻辑
+
+async def data_cleanup_task():
+    """异步数据清理任务"""
+    current_time = datetime.now(pytz.timezone(get_settings().TIMEZONE))
+    print(f"执行数据清理: {current_time}")
+    # 实现异步数据清理逻辑