diff --git a/.env.example b/.env.example index a30f8af..87f42fb 100644 --- a/.env.example +++ b/.env.example @@ -1,29 +1,43 @@ -# DeepSeek provider: one or more bearer tokens, comma-separated +# TOKENS ========================== +# Open Developer -> Network settings; make sure logged in; +# - Deepseek: find a '/chat/completion' URL call and get the Authorization: Bearer XXXXXXXXXX) +# - Qwen: find a 'chats/' URL call and get the cookie (extract token=eyJhb... or paste the whole cookie string) +# DeepSeek/Qwen tokens (multiple = comma-separated) DEEPSEEK_TOKENS= - -# Qwen provider: one or more bearer tokens, comma-separated QWEN_TOKENS= -# address the API server binds to + +# CONNECTION ====================== +# address the API server binds (alt HOST, PORT, PROXY, API_KEY) DANYAPI_HOST=0.0.0.0 -# port the API server listens on DANYAPI_PORT=8000 +# (optional) outgoing proxy for upstream communication +# (e.g. socks5://127.0.0.1:1080 or http://127.0.0.1:8080, docker: socks5://host.docker.internal:1080) +DANYAPI_PROXY= +# (optional) if enforcing USER Bearer token on user-api calls +DANYAPI_API_KEY= + + +# SETTINGS ======================== # upstream request timeout in seconds DANYAPI_TIMEOUT=60 # human-like delay before each upstream request (seconds) DANYAPI_HUMAN_DELAY_MIN=0.5 DANYAPI_HUMAN_DELAY_MAX=3.0 # seconds to wait for a free account before returning 429 (empty = wait forever) -DANYAPI_ACQUIRE_TIMEOUT= -# max server-side chats cached per provider (LRU) for stateless session reuse +DANYAPI_ACQUIRE_TIMEOUT=600 + +# SESSION: session_id -> maps to arbitrary user value (ex. session_id: "george") +# max server-side session-maps chats cached per provider (LRU) for stateless session reuse DANYAPI_SESSION_CACHE_SIZE=128 -# seconds an unused session/context stays reusable (0 = never expire) -DANYAPI_SESSION_TTL_SECONDS=3600 +# seconds an unused session/context stays reusable (0 = never expire, currently 7 days) +DANYAPI_SESSION_TTL_SECONDS=604800 # directory for on-disk session cache (default = system temp dir, e.g. %TEMP%\danyapi) DANYAPI_CACHE_DIR= # set to 1/true/yes to disable on-disk cache entirely DANYAPI_CACHE_DISABLED= -# log level: DEBUG, INFO, WARNING, ERROR + +# LOGGING: level -> DEBUG, INFO, WARNING, ERROR DANYAPI_LOG_LEVEL=INFO # file path for persistent logs (empty = console only) DANYAPI_LOG_FILE= @@ -31,9 +45,13 @@ DANYAPI_LOG_FILE= DANYAPI_LOG_MAX_BYTES=10485760 # number of rotated log files to keep DANYAPI_LOG_BACKUP_COUNT=3 + # set to 0/false/no/off to disable usage tracking (token counter stats) DANYAPI_USAGE_ENABLED=1 # max recent usage records kept in memory for /v1/usage DANYAPI_USAGE_MAX_RECORDS=1000 # auto-update to the latest GitHub release on each start (0 disables) -DANYAPI_AUTO_UPDATE=1 \ No newline at end of file +DANYAPI_AUTO_UPDATE=1 + +# custom User-Agent header for upstream requests to DeepSeek/Qwen (empty = default modern Chrome) +DANYAPI_USERAGENT=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 800407c..8bcc31f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -13,38 +13,7 @@ concurrency: cancel-in-progress: true jobs: - test: - runs-on: ubuntu-latest - timeout-minutes: 60 - strategy: - fail-fast: false - matrix: - python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"] - steps: - - uses: actions/checkout@v7 - - - uses: actions/setup-python@v7 - with: - python-version: ${{ matrix.python-version }} - cache: pip - cache-dependency-path: requirements-dev.txt - - - name: Install system tools - run: | - sudo apt-get update - sudo apt-get install -y --no-install-recommends clang clang-format clang-tidy nodejs - - - name: Install dependencies - run: pip install -r requirements-dev.txt - - - name: Build native PoW solver - run: clang -O3 -pthread -funroll-loops -flto -march=native -mtune=native -o danyapi/deepseek/pow_solver danyapi/deepseek/pow_solver.c - - - name: Run tests - run: python tests.py -j 2 - docker-build: - needs: test runs-on: ubuntu-latest timeout-minutes: 60 permissions: diff --git a/.gitignore b/.gitignore index 7f84673..2ee494c 100644 --- a/.gitignore +++ b/.gitignore @@ -23,3 +23,4 @@ pow_solver .DS_Store Thumbs.db references +*.txt diff --git a/Dockerfile b/Dockerfile index 8b10ffb..8b4e500 100644 --- a/Dockerfile +++ b/Dockerfile @@ -8,12 +8,13 @@ RUN pip install --no-cache-dir -r requirements.txt COPY danyapi ./danyapi RUN apt-get update \ - && apt-get install -y --no-install-recommends gcc libc6-dev nodejs \ + && apt-get install -y --no-install-recommends gcc libc6-dev nodejs curl \ && gcc -O3 -pthread -funroll-loops -flto -fomit-frame-pointer -o danyapi/deepseek/pow_solver danyapi/deepseek/pow_solver.c \ && apt-get purge -y gcc libc6-dev \ && apt-get autoremove -y \ && rm -rf /var/lib/apt/lists/* +ENV PYTHONUNBUFFERED=1 ENV DANYAPI_HOST=0.0.0.0 ENV DANYAPI_PORT=8000 diff --git a/app.py b/app.py index cc34840..0a87456 100644 --- a/app.py +++ b/app.py @@ -42,9 +42,10 @@ def main() -> None: import uvicorn from danyapi.config import settings - from danyapi.logging import uvicorn_log_config + from danyapi.logging import log_startup_info, uvicorn_log_config print(f"DanyAPI starting on {settings.host}:{settings.port}") + log_startup_info() uvicorn.run( "danyapi.api.openai:app", host=settings.host, diff --git a/danyapi/__main__.py b/danyapi/__main__.py index ef4ce44..168bdda 100644 --- a/danyapi/__main__.py +++ b/danyapi/__main__.py @@ -13,7 +13,9 @@ def main() -> None: ) sys.exit(1) from danyapi.config import settings - from danyapi.logging import uvicorn_log_config + from danyapi.logging import log_startup_info, uvicorn_log_config + + log_startup_info() uvicorn.run( "danyapi.api.openai:app", diff --git a/danyapi/api/openai.py b/danyapi/api/openai.py index 1042855..62a1230 100644 --- a/danyapi/api/openai.py +++ b/danyapi/api/openai.py @@ -7,17 +7,20 @@ import logging import random import re +import secrets import time import uuid from contextlib import asynccontextmanager from dataclasses import dataclass +from email import policy +from email.parser import BytesParser from pathlib import Path from typing import Any import httpx from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import HTMLResponse, StreamingResponse +from fastapi.responses import HTMLResponse, JSONResponse, StreamingResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field, field_validator @@ -125,7 +128,7 @@ class FileSpec(BaseModel): class ChatCompletionRequest(BaseModel): - model: str = Field(default="deepseek-v4-flash") + model: str | None = None messages: list[ChatMessage] = Field(default_factory=list) stream: bool = False temperature: float | None = None @@ -135,6 +138,7 @@ class ChatCompletionRequest(BaseModel): session_id: str | None = None user: str | None = None files: list[FileSpec] | None = None + file_ids: list[str] | None = None tools: list[Any] | None = None tool_choice: Any = None parallel_tool_calls: bool | None = None @@ -227,7 +231,15 @@ async def lifespan(app: FastAPI): stable_id=_token_stable_id(token), ) ) - log.info("deepseek accounts ready: %d", len(accounts)) + if accounts: + ds_models = list(MODEL_TYPE_BY_NAME.keys()) + log.info( + "deepseek accounts ready: %d (%s)", + len(accounts), + ", ".join(ds_models), + ) + else: + log.info("deepseek accounts ready: 0") if settings.qwen_tokens: for i, token in enumerate(settings.qwen_tokens): qw_client = QwenClient(token=token, timeout=settings.timeout) @@ -245,7 +257,6 @@ async def lifespan(app: FastAPI): stable_id=_token_stable_id(token), ) ) - log.info("qwen accounts ready: %d", len(qwen_accounts)) if accounts: app.state.pool = AccountPool( accounts, @@ -266,11 +277,18 @@ async def lifespan(app: FastAPI): affinity_store=qwen_affinity_store, ) app.state.qwen_models = await _fetch_qwen_models(qwen_accounts[0].client) + qw_models = list(dict.fromkeys(m["id"] for m in app.state.qwen_models if isinstance(m, dict) and m.get("id"))) + models_str = f" ({', '.join(qw_models)})" if qw_models else "" + log.info("qwen accounts ready: %d%s", len(qwen_accounts), models_str) else: app.state.qwen_pool = None app.state.qwen_models = [] + if settings.qwen_tokens: + log.info("qwen accounts ready: 0") if not accounts and not qwen_accounts: raise RuntimeError("no valid credentials: set DEEPSEEK_TOKENS or QWEN_TOKENS") + app.state.default_model = _determine_default_model(app) + log.info("default model: %s", app.state.default_model) yield finally: for ds_acct in accounts: @@ -318,6 +336,22 @@ async def _fetch_qwen_models(client: QwenClient) -> list[dict]: return models +def _determine_default_model(app_obj: FastAPI | None = None) -> str: + target = app_obj or app + pool = getattr(target.state, "pool", None) + qwen_pool = getattr(target.state, "qwen_pool", None) + qwen_models: list[dict] = getattr(target.state, "qwen_models", []) + + if pool and pool.accounts: + return next(iter(MODEL_TYPE_BY_NAME), "deepseek-v4-flash") + if qwen_pool and qwen_pool.accounts and qwen_models: + chat_models = [m["id"] for m in qwen_models if m.get("model_type") == "chat"] + if chat_models: + return chat_models[0] + return qwen_models[0]["id"] + return "deepseek-v4-flash" + + app = FastAPI(title="DanyAPI", lifespan=lifespan) app.add_middleware( @@ -490,6 +524,9 @@ async def add_tokens(tokens: dict) -> dict: _write_env_tokens(merged_ds, merged_qw) settings.deepseek_tokens = merged_ds settings.qwen_tokens = merged_qw + if added_ds or added_qw: + app.state.default_model = _determine_default_model(app) + log.info("default model updated: %s", app.state.default_model) parts = [] if added_ds: @@ -565,12 +602,21 @@ def _request_details(request: Request, payload: dict[str, Any]) -> str: return " ".join(parts) -def _log_request_failure(request: Request, payload: dict[str, Any], duration: float, status: int | None = None, exc: Exception | None = None) -> None: +def _log_request_failure( + request: Request, + payload: dict[str, Any], + duration: float, + status: int | None = None, + exc: Exception | None = None, + detail: str | None = None, +) -> None: details = _request_details(request, payload) details_part = f" {details}" if details else "" ip = _request_client_ip(request) if status is not None: reason = f"status={status}" + if detail: + reason += f" detail={detail}" else: reason = f"error={str(exc) if exc else 'unknown'}" log.warning( @@ -610,21 +656,198 @@ async def _log_requests(request: Request, call_next): payload, (time.monotonic() - started) * 1000, exc=exc, + detail=getattr(exc, "detail", None), ) raise duration = (time.monotonic() - started) * 1000 if response.status_code >= 400: + detail = getattr(getattr(request, "state", None), "error_detail", None) _log_request_failure( request, payload, duration, status=response.status_code, + detail=detail, ) else: _log_request_success(request, payload, duration) return response +PUBLIC_PATHS = {"/", "/favicon.ico", "/health"} + + +@app.middleware("http") +async def _authenticate_request(request: Request, call_next): + if settings.api_key and request.method != "OPTIONS": + path = request.url.path.rstrip("/") or "/" + if path not in PUBLIC_PATHS and not path.startswith("/docs"): + auth_header = request.headers.get("Authorization", "") + extracted_auth = auth_header[7:].strip() if auth_header.startswith("Bearer ") else auth_header.strip() + if not extracted_auth: + extracted_auth = request.headers.get("x-api-key", "").strip() + + if not extracted_auth or not secrets.compare_digest(extracted_auth, settings.api_key): + return JSONResponse( + status_code=401, + content={ + "error": { + "message": "Incorrect API key provided or missing authorization token.", + "type": "invalid_request_error", + "param": None, + "code": "invalid_api_key", + } + }, + ) + return await call_next(request) + + +class FileStagingManager: + """Tracks uploaded file IDs staged for automatic attachment to chat completions.""" + + def __init__(self, ttl: float = 3600.0) -> None: + self._lock = asyncio.Lock() + self._by_session: dict[str, list[dict]] = {} + self._by_client: dict[str, list[dict]] = {} + self._ttl = max(0.0, ttl) + + async def stage(self, file_record: dict, session_id: str | None = None, client_key: str | None = None) -> None: + now = time.monotonic() + record = dict(file_record) + record["staged_at"] = now + async with self._lock: + if session_id: + self._by_session.setdefault(session_id, []).append(record) + elif client_key: + self._by_client.setdefault(client_key, []).append(record) + + async def consume_records(self, session_id: str | None = None, client_key: str | None = None) -> list[dict]: + now = time.monotonic() + records_out: list[dict] = [] + async with self._lock: + if session_id and session_id in self._by_session: + records = self._by_session.pop(session_id, []) + for r in records: + if self._ttl <= 0 or (now - r.get("staged_at", now) < self._ttl): + records_out.append(r) + if client_key and client_key in self._by_client: + records = self._by_client.pop(client_key, []) + for r in records: + if self._ttl <= 0 or (now - r.get("staged_at", now) < self._ttl): + records_out.append(r) + return records_out + + async def consume(self, session_id: str | None = None, client_key: str | None = None) -> list[str]: + records = await self.consume_records(session_id=session_id, client_key=client_key) + return [r["id"] for r in records if "id" in r] + + async def find(self, file_id: str) -> dict | None: + async with self._lock: + for recs in list(self._by_session.values()) + list(self._by_client.values()): + for r in recs: + if r.get("id") == file_id: + return r + return None + + +file_staging = FileStagingManager(ttl=getattr(settings, "session_ttl", 3600.0) or 3600.0) + + +def _get_client_key(request: Request | None) -> str: + if request is None: + return "default" + auth = request.headers.get("authorization", "") + if auth.startswith("Bearer "): + token = auth[7:].strip() + if token: + return f"auth:{hashlib.sha256(token.encode()).hexdigest()[:16]}" + if request.client and request.client.host: + return f"ip:{request.client.host}" + return "default" + + +async def _extract_file_upload(request: Request) -> tuple[bytes, str, str, str | None, str, str | None]: + content_type_header = request.headers.get("content-type", "") + + if "application/json" in content_type_header: + body = await request.json() + raw_b64 = body.get("file") or body.get("content") + if not raw_b64: + raise HTTPException(400, "JSON file upload missing 'file' or 'content' base64 field") + try: + data = base64.b64decode(raw_b64) + except Exception as exc: + raise HTTPException(400, f"Invalid base64 payload: {exc}") from exc + filename = str(body.get("filename") or body.get("name") or "upload.bin") + ctype = str(body.get("content_type") or "application/octet-stream") + session_id = str(body.get("session_id")) if body.get("session_id") else None + purpose = str(body.get("purpose") or "assistants") + model = str(body.get("model")) if body.get("model") else None + return data, filename, ctype, session_id, purpose, model + + if "multipart/form-data" in content_type_header: + data_bytes: bytes | None = None + filename = "upload.bin" + ctype = "application/octet-stream" + session_id = None + purpose = "assistants" + model = None + try: + form = await request.form() + file_field = form.get("file") + if file_field is not None and hasattr(file_field, "read"): + read_res = await file_field.read() # type: ignore[union-attr] + if isinstance(read_res, (bytes, bytearray)): + data_bytes = bytes(read_res) + filename = str(getattr(file_field, "filename", "upload.bin") or "upload.bin") + ctype = str(getattr(file_field, "content_type", "application/octet-stream") or "application/octet-stream") + session_id = str(form.get("session_id")) if form.get("session_id") else None + purpose = str(form.get("purpose") or "assistants") + model = str(form.get("model")) if form.get("model") else None + except Exception: + body_bytes = await request.body() + header_bytes = b"Content-Type: " + content_type_header.encode("latin1", "replace") + b"\r\n\r\n" + msg = BytesParser(policy=policy.default).parsebytes(header_bytes + body_bytes) + for part in msg.iter_parts(): + cd = part.get_param("name", header="content-disposition") + if cd == "file": + payload = part.get_payload(decode=True) + if isinstance(payload, (bytes, bytearray)): + data_bytes = bytes(payload) + fn = part.get_filename() + if fn: + filename = str(fn) + ct = part.get_content_type() + if ct and ct != "application/octet-stream": + ctype = str(ct) + elif cd == "session_id": + val = part.get_payload(decode=True) + if isinstance(val, (bytes, bytearray)): + session_id = val.decode("utf-8", errors="replace").strip() or None + elif cd == "purpose": + val = part.get_payload(decode=True) + if isinstance(val, (bytes, bytearray)): + purpose = val.decode("utf-8", errors="replace").strip() or "assistants" + elif cd == "model": + val = part.get_payload(decode=True) + if isinstance(val, (bytes, bytearray)): + model = val.decode("utf-8", errors="replace").strip() or None + + if data_bytes is None: + raise HTTPException(400, "Multipart form missing 'file' field") + return data_bytes, filename, ctype, session_id, purpose, model + + body_bytes = await request.body() + if not body_bytes: + raise HTTPException(400, "Empty request body") + filename = str(request.query_params.get("filename", "upload.bin")) + ctype = str(request.headers.get("content-type", "application/octet-stream")) + session_id = str(request.query_params.get("session_id")) if request.query_params.get("session_id") else None + purpose = str(request.query_params.get("purpose", "assistants")) + model = str(request.query_params.get("model")) if request.query_params.get("model") else None + return body_bytes, filename, ctype, session_id, purpose, model + + MAX_FILES_PER_REQUEST = 50 MAX_FILE_SIZE = 100 * 1024 * 1024 @@ -693,6 +916,8 @@ def _validate_attachments(attachments: list[Attachment], model_type: str) -> Non raise HTTPException(400, "deepseek-v4-pro does not support file attachments") if model_type == "vision" and any(not att.is_image for att in attachments): raise HTTPException(400, "deepseek-v4-vision accepts images only") + if model_type == "default" and any(att.is_image for att in attachments): + raise HTTPException(400, "deepseek-v4-flash does not support image attachments; use deepseek-v4-vision") async def _fresh_pow_upload_headers(account) -> dict: @@ -860,11 +1085,14 @@ def _resolve_provider(model: str) -> str: @app.post("/v1/chat/completions") -async def chat_completions(req: ChatCompletionRequest) -> Any: - provider = _resolve_provider(req.model) +async def chat_completions(req: ChatCompletionRequest, request: Request) -> Any: + user_specified_model = bool(req.model) + model_name = req.model or str(getattr(app.state, "default_model", "deepseek-v4-flash")) + req.model = model_name + provider = _resolve_provider(model_name) if provider == "qwen": - return await _chat_completions_qwen(req) - return await _chat_completions_deepseek(req) + return await _chat_completions_qwen(req, request=request) + return await _chat_completions_deepseek(req, request=request, user_specified_model=user_specified_model) @app.post("/v1/images/generations") @@ -898,7 +1126,12 @@ async def image_generations(req: ImageGenerationRequest) -> dict: data.append({"url": url}) continue try: - async with httpx.AsyncClient(follow_redirects=True, timeout=30) as hc: + async with httpx.AsyncClient( + headers={"User-Agent": settings.user_agent}, + follow_redirects=True, + timeout=30, + proxy=settings.proxy, + ) as hc: img_resp = await hc.get(url) if img_resp.status_code != 200: log.warning("image download failed (%s) for %s, returning url", img_resp.status_code, url) @@ -921,6 +1154,338 @@ async def image_generations(req: ImageGenerationRequest) -> dict: } +@app.post("/v1/files") +async def upload_file_endpoint(request: Request) -> dict: + data, filename, ctype, session_id, purpose, model = await _extract_file_upload(request) + + if len(data) > MAX_FILE_SIZE: + raise HTTPException(400, f"file {filename} exceeds {MAX_FILE_SIZE // (1024*1024)} MB limit") + + target_provider = None + if model: + try: + target_provider = _resolve_provider(model) + except HTTPException: + target_provider = None + + ds_pool: AccountPool | None = getattr(app.state, "pool", None) + qw_pool: AccountPool | None = getattr(app.state, "qwen_pool", None) + + if target_provider is None and session_id: + if qw_pool and qw_pool.account_for_session(session_id) is not None: + target_provider = "qwen" + elif ds_pool and ds_pool.account_for_session(session_id) is not None: + target_provider = "deepseek" + + use_qwen = (target_provider == "qwen") or ( + target_provider is None and not (ds_pool and ds_pool.healthy) and (qw_pool and qw_pool.healthy) + ) + + if use_qwen: + if qw_pool is None or not qw_pool.healthy: + raise HTTPException(503, "qwen provider is not configured or no healthy accounts available") + + account = None + if session_id: + account = qw_pool.account_for_session(session_id) + if account is None: + try: + account, _ = await qw_pool.acquire(session_id) + except AccountPoolBusy: + raise HTTPException(429, "all accounts are busy, please retry shortly") from None + if session_id: + qw_pool.register(account.index, session_id) + + try: + q_att = await account.client.upload_file( + data=data, + filename=filename, + content_type=ctype, + ) + except Exception as exc: + raise HTTPException(502, f"qwen file upload failed: {exc}") from exc + + file_id = q_att.get("id") + if not file_id: + raise HTTPException(502, "file upload failed: no file id returned from Qwen") + + is_image = ctype.startswith("image/") or filename.lower().endswith((".jpg", ".jpeg", ".png", ".webp", ".gif")) + client_key = _get_client_key(request) + file_record = { + "id": file_id, + "object": "file", + "bytes": len(data), + "created_at": int(time.time()), + "filename": filename, + "purpose": purpose, + "session_id": session_id, + "status": "processed", + "content_type": ctype, + "is_image": is_image, + "provider": "qwen", + "qwen_attachment": q_att, + } + await file_staging.stage(file_record, session_id=session_id, client_key=client_key) + log.info("uploaded and staged qwen file %s (%s, %d bytes) for session=%s", file_id, filename, len(data), session_id) + return file_record + + pool: AccountPool = ds_pool + if pool is None or not pool.healthy: + raise HTTPException(503, "deepseek provider is not configured or no healthy accounts available") + + # Resolve account with session affinity if session_id is given + account = None + if session_id: + account = pool.account_for_session(session_id) + if account is None: + try: + account, _ = await pool.acquire(session_id) + except AccountPoolBusy: + raise HTTPException(429, "all accounts are busy, please retry shortly") from None + if session_id: + pool.register(account.index, session_id) + + model_type = "vision" if ctype.startswith("image/") else "default" + if model: + resolved_type = MODEL_TYPE_BY_NAME.get(model) + if resolved_type: + model_type = resolved_type + + pow_headers = await _fresh_pow_upload_headers(account) + try: + info = await account.client.upload_file( + data=data, + filename=filename, + content_type=ctype, + model_type=model_type, + thinking_enabled=False, + pow_headers=pow_headers, + ) + except DeepSeekError as exc: + _handle_account_error(account, exc) + raise HTTPException(_deepseek_status(exc), f"file upload failed: {exc}") from exc + + file_id = info.get("id") + if not file_id: + raise HTTPException(502, "file upload failed: no file id returned from DeepSeek") + + is_image = ctype.startswith("image/") or filename.lower().endswith((".jpg", ".jpeg", ".png", ".webp", ".gif")) + client_key = _get_client_key(request) + file_record = { + "id": file_id, + "object": "file", + "bytes": len(data), + "created_at": int(time.time()), + "filename": filename, + "purpose": purpose, + "session_id": session_id, + "status": "processed", + "content_type": ctype, + "model_type": model_type, + "is_image": is_image, + } + await file_staging.stage(file_record, session_id=session_id, client_key=client_key) + log.info("uploaded and staged file %s (%s, %d bytes) for session=%s", file_id, filename, len(data), session_id) + return file_record + + +@app.get("/v1/files/{file_id}") +async def get_file_endpoint(file_id: str) -> dict: + staged = await file_staging.find(file_id) + if staged: + return { + "id": staged.get("id", file_id), + "object": "file", + "bytes": staged.get("bytes", 0), + "created_at": staged.get("created_at", int(time.time())), + "filename": staged.get("filename", ""), + "purpose": staged.get("purpose", "assistants"), + "status": staged.get("status", "processed"), + "raw": staged, + } + + pool: AccountPool | None = getattr(app.state, "pool", None) + if pool is None or not pool.healthy: + raise HTTPException(503, "deepseek provider is not configured or no healthy accounts available") + + for acct in pool.healthy: + try: + files = await acct.client.fetch_files([file_id]) + if files: + f = files[0] + return { + "id": f.get("id", file_id), + "object": "file", + "bytes": f.get("file_size", 0), + "created_at": f.get("created_at", int(time.time())), + "filename": f.get("file_name", ""), + "purpose": "assistants", + "status": "processed", + "raw": f, + } + except Exception: + continue + raise HTTPException(404, f"file {file_id} not found") + + +@app.get("/v1/sessions") +async def list_sessions( + account: int | None = None, + pinned: bool = False, + count: int = 20, +) -> dict: + pool: AccountPool | None = getattr(app.state, "pool", None) + if pool is None or not pool.healthy: + raise HTTPException(503, "deepseek provider is not configured or no healthy accounts available") + + target_accounts: list[Any] = [] + if account is not None: + if 0 <= account < len(pool.accounts): + acct = pool.accounts[account] + if acct.broken: + raise HTTPException(400, f"Account #{account} is currently marked broken") + target_accounts = [acct] + else: + raise HTTPException(400, f"Invalid account index: {account}. Valid: 0..{len(pool.accounts)-1}") + else: + target_accounts = pool.healthy + + count = max(1, min(count, 100)) + all_sessions: list[dict] = [] + for acct in target_accounts: + try: + items = await acct.client.fetch_page(pinned=pinned, count=count) + for item in items: + sid = item.get("id") + if sid: + pool.register(acct.index, sid) + item["account_index"] = acct.index + all_sessions.append(item) + except Exception as exc: + log.warning("failed to fetch sessions for account #%d: %s", acct.index, exc) + + return { + "object": "list", + "data": all_sessions, + } + + +def _resolve_session_uuid(pool: AccountPool, session_id: str) -> str: + target_account = pool.account_for_session(session_id) + if target_account is not None: + sess = target_account.sessions.get(session_id) + if sess is not None and getattr(sess, "id", None): + return sess.id + for acct in pool.accounts: + sess = acct.sessions.get(session_id) + if sess is not None and getattr(sess, "id", None): + return sess.id + return session_id + + +@app.get("/v1/sessions/{session_id}") +async def get_session(session_id: str, account: int | None = None) -> dict: + pool: AccountPool | None = getattr(app.state, "pool", None) + if pool is None or not pool.healthy: + raise HTTPException(503, "deepseek provider is not configured or no healthy accounts available") + + upstream_id = _resolve_session_uuid(pool, session_id) + + target_account = None + if account is not None: + if 0 <= account < len(pool.accounts): + target_account = pool.accounts[account] + else: + raise HTTPException(400, f"Invalid account index: {account}") + else: + target_account = pool.account_for_session(session_id) or pool.account_for_session(upstream_id) + + if target_account is not None and not target_account.broken: + try: + messages = await target_account.client.history_messages(upstream_id) + pool.register(target_account.index, session_id) + if upstream_id != session_id: + pool.register(target_account.index, upstream_id) + return { + "id": session_id, + "object": "chat.session", + "account_index": target_account.index, + "messages": messages, + } + except DeepSeekError as exc: + if exc.biz_code in (404, 40004): + pass + else: + raise HTTPException(_deepseek_status(exc), f"DeepSeek error: {exc}") from exc + except HTTPException: + raise + except Exception as exc: + log.warning("failed to fetch session %s from account #%d: %s", session_id, target_account.index, exc) + + # Search across healthy accounts + for acct in pool.healthy: + if target_account is not None and acct.index == target_account.index: + continue + try: + messages = await acct.client.history_messages(upstream_id) + pool.register(acct.index, session_id) + if upstream_id != session_id: + pool.register(acct.index, upstream_id) + return { + "id": session_id, + "object": "chat.session", + "account_index": acct.index, + "messages": messages, + } + except Exception: + continue + + raise HTTPException(404, f"Session {session_id} not found") + + +@app.delete("/v1/sessions/{session_id}") +async def delete_session_endpoint(session_id: str, account: int | None = None) -> dict: + pool: AccountPool | None = getattr(app.state, "pool", None) + if pool is None or not pool.healthy: + raise HTTPException(503, "deepseek provider is not configured or no healthy accounts available") + + upstream_id = _resolve_session_uuid(pool, session_id) + + target_account = None + if account is not None and 0 <= account < len(pool.accounts): + target_account = pool.accounts[account] + else: + target_account = pool.account_for_session(session_id) or pool.account_for_session(upstream_id) + + if target_account is None: + for acct in pool.healthy: + try: + msgs = await acct.client.history_messages(upstream_id) + if msgs is not None: + target_account = acct + break + except Exception: + continue + + if target_account is None: + raise HTTPException(404, f"Session {session_id} not found") + + try: + await target_account.client.delete_session(upstream_id) + except DeepSeekError as exc: + raise HTTPException(_deepseek_status(exc), f"Delete session failed: {exc}") from exc + + target_account.sessions.forget(session_id) + target_account.sessions.forget(upstream_id) + pool.forget(session_id) + pool.forget(upstream_id) + return { + "id": session_id, + "object": "chat.session", + "deleted": True, + } + + async def _acquire_account(pool: AccountPool, session_id: str | None): try: return await pool.acquire(session_id, settings.acquire_timeout) @@ -1034,22 +1599,74 @@ async def _stream_guard(gen, model: str): yield line -async def _chat_completions_deepseek(req: ChatCompletionRequest) -> Any: +async def _chat_completions_deepseek( + req: ChatCompletionRequest, + request: Request | None = None, + user_specified_model: bool = True, +) -> Any: pool: AccountPool = app.state.pool if pool is None: raise HTTPException(503, "deepseek provider is not configured") - model_type = _resolve_model(req.model) - thinking = req.thinking if req.thinking is not None else _is_reasoning_model(req.model) - search = bool(req.search) and model_type == "default" - account, existing_sid, context_seq, prompt, tool_mode = await _acquire_and_build(pool, req) attachments = _collect_attachments(req) + client_key = _get_client_key(request) + staged_records = await file_staging.consume_records(session_id=existing_sid or req.session_id, client_key=client_key) + + has_image = any(att.is_image for att in attachments) or any( + r.get("is_image") or r.get("model_type") == "vision" or str(r.get("filename", "")).lower().endswith((".jpg", ".jpeg", ".png", ".webp", ".gif")) + for r in staged_records + ) + + req_file_ids = getattr(req, "file_ids", None) + if not has_image and req_file_ids and not user_specified_model: + try: + file_meta = await account.client.fetch_files(req_file_ids[:5]) + if any(f.get("is_image") or f.get("model_kind") == "VISION" for f in file_meta): + has_image = True + except Exception as exc: + log.debug("failed to fetch files metadata: %s", exc) + + default_model = getattr(app.state, "default_model", "deepseek-v4-flash") + if has_image and (not user_specified_model or req.model == default_model): + req.model = "deepseek-v4-vision" + log.info("auto-selected deepseek-v4-vision for image attachments (session=%s)", existing_sid or req.session_id) + + model_name: str = req.model or default_model + req.model = model_name + + model_type = _resolve_model(model_name) + thinking = req.thinking if req.thinking is not None else _is_reasoning_model(model_name) + search = bool(req.search) and model_type == "default" + _validate_attachments(attachments, model_type) - ref_file_ids = None + ref_file_ids_list: list[str] = [] + + # 1. Any explicitly passed file IDs + if req_file_ids: + ref_file_ids_list.extend(req_file_ids) + + # 2. Any auto-staged files from POST /v1/files + for r in staged_records: + if "id" in r: + ref_file_ids_list.append(r["id"]) + + # 3. Any inline attachments uploaded on the fly if attachments: - ref_file_ids = await _upload_attachments(account, attachments, model_type, thinking) + inline_ids = await _upload_attachments(account, attachments, model_type, thinking) + ref_file_ids_list.extend(inline_ids) + + ref_file_ids: list[str] | None = None + if ref_file_ids_list: + seen: set[str] = set() + deduped: list[str] = [] + for fid in ref_file_ids_list: + if fid not in seen: + seen.add(fid) + deduped.append(fid) + ref_file_ids = deduped + log.info("attached %d file(s) to completion: %s", len(ref_file_ids), ref_file_ids) common = { "account": account, @@ -1076,7 +1693,7 @@ async def _chat_completions_deepseek(req: ChatCompletionRequest) -> Any: } if req.stream: return StreamingResponse( - _stream_guard(_stream_openai(lock=account.sem, **common), req.model), + _stream_guard(_stream_openai(lock=account.sem, **common), model_name), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) @@ -1087,7 +1704,7 @@ async def _chat_completions_deepseek(req: ChatCompletionRequest) -> Any: raise HTTPException(429, "all accounts are busy, try again later") from None -async def _chat_completions_qwen(req: ChatCompletionRequest) -> Any: +async def _chat_completions_qwen(req: ChatCompletionRequest, request: Request | None = None) -> Any: pool: AccountPool = app.state.qwen_pool if pool is None: raise HTTPException(503, "qwen provider is not configured") @@ -1095,15 +1712,35 @@ async def _chat_completions_qwen(req: ChatCompletionRequest) -> Any: thinking = req.thinking if req.thinking is not None else True search = bool(req.search) - account, existing_sid, context_seq, prompt, tool_mode = await _acquire_and_build(pool, req, {"model": req.model}) + client_key = _get_client_key(request) + staged_records = await file_staging.consume_records(session_id=req.session_id, client_key=client_key) + qwen_files: list[dict] = [] + for r in staged_records: + q_att = r.get("qwen_attachment") + if q_att: + qwen_files.append(q_att) + + model_name: str = req.model or "qwen3.7-plus" + req.model = model_name + + account, existing_sid, context_seq, prompt, tool_mode = await _acquire_and_build(pool, req, {"model": model_name}) + + inline_attachments = _collect_attachments(req) + if inline_attachments: + for att in inline_attachments: + q_att = await account.client.upload_file(att.data, att.name, att.content_type) + qwen_files.append(q_att) + + if qwen_files: + log.info("attached %d file(s) to qwen completion", len(qwen_files)) common = { "account": account, "pool": pool, "existing_sid": existing_sid, "prompt": prompt, - "model": req.model, - "model_id": req.model, + "model": model_name, + "model_id": model_name, "thinking": thinking, "search": search, "tool_schemas": toolemu.tool_schema_map(getattr(req, "tools", None)), @@ -1115,10 +1752,11 @@ async def _chat_completions_qwen(req: ChatCompletionRequest) -> Any: "tool_choice": getattr(req, "tool_choice", None), "response_format": getattr(req, "response_format", None), "user": getattr(req, "user", None), + "files": qwen_files or None, } if req.stream: return StreamingResponse( - _stream_guard(qwen_api.stream_openai(lock=account.sem, **common), req.model), + _stream_guard(qwen_api.stream_openai(lock=account.sem, **common), model_name), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}, ) @@ -1161,6 +1799,14 @@ async def _send_completion( search, ref_file_ids=None, ): + log.debug( + "deepseek completion request: session=%s parent=%s model_type=%s prompt_len=%d files=%s", + session_id, + parent_message_id, + model_type, + len(prompt), + ref_file_ids, + ) try: resp = await client.completion( chat_session_id=session_id, @@ -1180,28 +1826,37 @@ async def _send_completion( if resp.status_code != 200: body = await resp.aread() await resp.aclose() - raise HTTPException(resp.status_code, body[:500].decode("utf-8", errors="replace")) + err_msg = body[:500].decode("utf-8", errors="replace") + log.warning("deepseek upstream error (%s): %s", resp.status_code, err_msg) + raise HTTPException(resp.status_code, err_msg) content_type = resp.headers.get("content-type", "") if "text/event-stream" not in content_type: body = await resp.aread() await resp.aclose() + raw_body = body[:500].decode("utf-8", errors="replace") try: payload = json.loads(body) except json.JSONDecodeError as exc: - raise HTTPException(502, body[:500].decode("utf-8", errors="replace")) from exc + log.warning("deepseek non-stream non-json response (%s): %s", resp.status_code, raw_body) + raise HTTPException(502, raw_body) from exc data = payload.get("data") or {} if data.get("biz_code"): code = data["biz_code"] + msg = data.get("biz_msg") or "" status = 401 if code in DEEPSEEK_AUTH_ERROR_CODES else 502 - raise HTTPException(status, f"DeepSeek error {code}: {data.get('biz_msg')}") + log.warning("deepseek upstream error %s: %s", code, msg) + raise HTTPException(status, f"DeepSeek error {code}: {msg}") if payload.get("code"): code = payload["code"] + msg = payload.get("msg") or payload.get("message") or "" status = 401 if code in DEEPSEEK_AUTH_ERROR_CODES else 502 + log.warning("deepseek upstream error %s: %s", code, msg) raise HTTPException( status, - f"DeepSeek error {code}: {payload.get('msg') or payload.get('message')}", + f"DeepSeek error {code}: {msg}", ) + log.warning("deepseek unexpected non-stream response: %s", raw_body) raise HTTPException(502, "unexpected non-stream response") return resp @@ -1213,6 +1868,27 @@ def _is_retryable_hint(rec: MessageReconstructor) -> bool: RETRYABLE_HTTP_STATUSES = {408, 425, 429, 500, 502, 503, 504} STALE_SESSION_STATUSES = {400, 404} +DEEPSEEK_STALE_SESSION_CODES = {26, 40004, 40005, 40006, 40007, 40011, 40018} + + +def _is_stale_session_error(exc: HTTPException) -> bool: + if exc.status_code in STALE_SESSION_STATUSES: + return True + detail = str(getattr(exc, "detail", "") or "").lower() + if any(phrase in detail for phrase in ( + "invalid message id", + "chat session not found", + "parent message", + "message not found", + "session not found", + "session closed", + "session expired", + )): + return True + for code in DEEPSEEK_STALE_SESSION_CODES: + if f"error {code}:" in detail or f"error {code}" in detail or f"biz error {code}" in detail: + return True + return False def _retry_delay(attempt: int) -> float: @@ -1613,7 +2289,7 @@ async def _collect_non_stream( rec = MessageReconstructor() rec.hint_error = input_hint break - if exc.status_code in STALE_SESSION_STATUSES and had_cached_session and not stale_rebuilt and messages is not None: + if _is_stale_session_error(exc) and had_cached_session and not stale_rebuilt and messages is not None: stale_rebuilt = True _drop_session(pool, account, session_key) try: @@ -1621,7 +2297,7 @@ async def _collect_non_stream( tool_schemas = toolemu.tool_schema_map(tools) except (ValueError, TypeError, AttributeError) as build_exc: raise exc from build_exc - log.warning("deepseek session %s is stale (%s), rebuilt full history into a fresh chat", session_key, exc.status_code) + log.warning("deepseek session %s is stale (%s - %s), rebuilt full history into a fresh chat", session_key, exc.status_code, exc.detail) session, session_key, parent_message_id = await _prepare_session(account, pool, existing_sid, context_seq) stop_message_id = None response_message_id = None @@ -1630,8 +2306,9 @@ async def _collect_non_stream( attempt += 1 delay = _retry_delay(attempt) log.warning( - "deepseek provider error (%s), retry %d/%d in %.1fs", + "deepseek provider error (%s - %s), retry %d/%d in %.1fs", exc.status_code, + exc.detail, attempt, MAX_RETRIES, delay, @@ -1694,7 +2371,8 @@ async def _collect_non_stream( raise HTTPException(429, _busy_error_body(rec)) request_tokens = _advance_session_usage(session, rec.accumulated_tokens) usage = _deepseek_usage(request_tokens, prompt) - account.sessions.touch_last_message(session_key, rec.id or response_message_id) + if content or reasoning: + account.sessions.touch_last_message(session_key, rec.id or response_message_id) record_usage( "deepseek", model, @@ -1787,7 +2465,7 @@ async def _stream_openai( rec = MessageReconstructor() rec.hint_error = input_hint break - if exc.status_code in STALE_SESSION_STATUSES and had_cached_session and not stale_rebuilt and messages is not None: + if _is_stale_session_error(exc) and had_cached_session and not stale_rebuilt and messages is not None: stale_rebuilt = True _drop_session(pool, account, session_key) try: @@ -1798,7 +2476,7 @@ async def _stream_openai( for line in _stream_error_sse(chunk_id, created, model, detail, session_key): yield line return - log.warning("deepseek session %s is stale (%s), rebuilt full history into a fresh chat", session_key, exc.status_code) + log.warning("deepseek session %s is stale (%s - %s), rebuilt full history into a fresh chat", session_key, exc.status_code, exc.detail) session, session_key, parent_message_id = await _prepare_session(account, pool, existing_sid, context_seq) stop_message_id = None response_message_id = None @@ -1807,8 +2485,9 @@ async def _stream_openai( attempt += 1 delay = _retry_delay(attempt) log.warning( - "deepseek provider error (%s), retry %d/%d in %.1fs", + "deepseek provider error (%s - %s), retry %d/%d in %.1fs", exc.status_code, + exc.detail, attempt, MAX_RETRIES, delay, diff --git a/danyapi/config.py b/danyapi/config.py index 31fc57c..761512a 100644 --- a/danyapi/config.py +++ b/danyapi/config.py @@ -45,8 +45,11 @@ def _env_str(key: str, default: str = "") -> str: class Settings: def __init__(self) -> None: - self.host = os.environ.get("DANYAPI_HOST", "0.0.0.0") - self.port = _env_int("DANYAPI_PORT", 8000) + self.host = _env_str("DANYAPI_HOST") or _env_str("HOST") or "0.0.0.0" + self.port = _env_int("DANYAPI_PORT", 0) or _env_int("PORT", 0) or 8000 + self.proxy = _env_str("DANYAPI_PROXY") or _env_str("PROXY") or None + self.api_key = _env_str("DANYAPI_API_KEY") or _env_str("API_KEY") or None + self.user_agent = _env_str("DANYAPI_USERAGENT") or _env_str("USER_AGENT") tokens = [t.strip() for t in os.environ.get("DEEPSEEK_TOKENS", "").split(",") if t.strip()] self.deepseek_tokens = tokens qwen_tokens = [t.strip() for t in os.environ.get("QWEN_TOKENS", "").split(",") if t.strip()] @@ -55,8 +58,8 @@ def __init__(self) -> None: self.acquire_timeout = _env_float_opt("DANYAPI_ACQUIRE_TIMEOUT") self.session_cache_size = _env_int("DANYAPI_SESSION_CACHE_SIZE", 128) self.session_ttl = _env_float("DANYAPI_SESSION_TTL_SECONDS", 3600.0) - self.log_level = _env_str("DANYAPI_LOG_LEVEL", "INFO") or "INFO" - self.log_file = _env_str("DANYAPI_LOG_FILE") + self.log_level = _env_str("DANYAPI_LOG_LEVEL") or _env_str("LOG_LEVEL") or "INFO" + self.log_file = _env_str("DANYAPI_LOG_FILE") or _env_str("LOG_FILE") self.log_max_bytes = _env_int("DANYAPI_LOG_MAX_BYTES", 10 * 1024 * 1024) self.log_backup_count = _env_int("DANYAPI_LOG_BACKUP_COUNT", 3) self.cache_dir = _env_str("DANYAPI_CACHE_DIR") diff --git a/danyapi/deepseek/client.py b/danyapi/deepseek/client.py index 9be759b..0c74554 100644 --- a/danyapi/deepseek/client.py +++ b/danyapi/deepseek/client.py @@ -7,6 +7,8 @@ import httpx +from ..config import settings + log = logging.getLogger("danyapi.deepseek") BASE_URL = "https://chat.deepseek.com" @@ -19,7 +21,7 @@ "x-client-timezone-offset": "0", } -USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36" +USER_AGENT = settings.user_agent def new_device_id() -> str: @@ -48,16 +50,20 @@ def __init__( token: str | None = None, device_id: str | None = None, timeout: float = 60.0, + proxy: str | None = None, + user_agent: str | None = None, ) -> None: self.token = token self.device_id = device_id or new_device_id() + ua = user_agent or settings.user_agent or USER_AGENT headers = { - "User-Agent": USER_AGENT, "Referer": "https://chat.deepseek.com/", "Origin": "https://chat.deepseek.com", "Accept": "*/*", **CLIENT_HEADERS, } + if ua: + headers["User-Agent"] = ua if token: headers["Authorization"] = f"Bearer {token}" self.http = httpx.AsyncClient( @@ -65,6 +71,7 @@ def __init__( headers=headers, timeout=httpx.Timeout(timeout), follow_redirects=True, + proxy=proxy or settings.proxy, ) async def aclose(self) -> None: @@ -101,6 +108,9 @@ async def check_auth(self) -> bool: params={"did": self.device_id, "scope": "main"}, ) return resp.json().get("code") == 0 + except (httpx.ConnectError, httpx.ProxyError, httpx.TimeoutException) as exc: + log.error("upstream connection failed (%s): check network/proxy settings", exc) + return False except (httpx.HTTPError, ValueError): return False diff --git a/danyapi/logging.py b/danyapi/logging.py index 0d26c85..d22bc13 100644 --- a/danyapi/logging.py +++ b/danyapi/logging.py @@ -1,5 +1,6 @@ from __future__ import annotations +import ipaddress import logging import re import sys @@ -156,7 +157,7 @@ def configure() -> None: root.setLevel(level) if not _has_handler(root, CONSOLE_HANDLER_NAME): - console = logging.StreamHandler() + console = logging.StreamHandler(sys.stdout) console.name = CONSOLE_HANDLER_NAME console.setLevel(level) console.setFormatter(_ColorFormatter()) @@ -195,3 +196,104 @@ def uvicorn_log_config() -> dict: "uvicorn.access": {"handlers": [], "level": level, "propagate": True}, }, } + + +IP_CHECK_ENDPOINTS = [ + "https://ifconfig.me/ip", + "https://api.ipify.org", + "https://icanhazip.com", +] + + +def _is_valid_ip(text: str) -> bool: + try: + ipaddress.ip_address(text.strip()) + return True + except ValueError: + return False + + +def get_outgoing_ip(proxy: str | None = None, timeout: float = 4.0) -> tuple[str | None, str | None]: + proxy_url = proxy if (isinstance(proxy, str) and proxy.strip()) else None + last_err: str | None = None + + try: + from .config import settings + + ua = getattr(settings, "user_agent", "curl/7.88.1") + except Exception: + ua = "curl/7.88.1" + + # 1. Try httpx + try: + import httpx + + with httpx.Client(headers={"User-Agent": ua}, proxy=proxy_url, timeout=timeout) as client: + for url in IP_CHECK_ENDPOINTS: + try: + resp = client.get(url) + if resp.status_code == 200: + candidate = resp.text.strip() + if candidate and _is_valid_ip(candidate): + return candidate, None + except Exception as exc: + last_err = f"{type(exc).__name__}: {exc}" + continue + except Exception as exc: + last_err = f"{type(exc).__name__}: {exc}" + + # 2. Fallback to curl if available + try: + import shutil + import subprocess + + curl_path = shutil.which("curl") + if curl_path: + for url in IP_CHECK_ENDPOINTS: + cmd = [curl_path, "-s", "-A", ua, "--max-time", str(int(timeout))] + if proxy_url: + if proxy_url.startswith("socks5://") or proxy_url.startswith("socks5h://"): + socks_addr = proxy_url.split("://", 1)[1] + cmd.extend(["--socks5-hostname", socks_addr]) + else: + cmd.extend(["-x", proxy_url]) + cmd.append(url) + try: + res = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 1, check=False) + candidate = res.stdout.strip() + if res.returncode == 0 and candidate and _is_valid_ip(candidate): + return candidate, None + except Exception: + continue + except Exception as exc: + last_err = str(exc) + + return None, last_err + + +def log_startup_info() -> None: + from danyapi.config import settings + + log = logging.getLogger("danyapi") + + raw_proxy = getattr(settings, "proxy", None) + proxy = raw_proxy if (isinstance(raw_proxy, str) and raw_proxy.strip()) else None + proxy_desc = f"via proxy {proxy}" if proxy else "direct, no proxy" + + try: + ip, err = get_outgoing_ip(proxy=proxy, timeout=4.0) + if ip: + log.info("outgoing IP: %s (%s)", ip, proxy_desc) + elif err: + log.warning("could not determine outgoing IP (%s): %s", proxy_desc, err) + else: + log.warning("could not determine outgoing IP (%s)", proxy_desc) + except Exception as exc: + log.warning("could not determine outgoing IP (%s): %s", proxy_desc, exc) + + raw_api_key = getattr(settings, "api_key", None) + api_key = raw_api_key if (isinstance(raw_api_key, str) and raw_api_key.strip()) else None + if api_key: + log.info("authentication: Bearer token required") + else: + log.info("authentication: open (no API_KEY set)") diff --git a/danyapi/qwen/api.py b/danyapi/qwen/api.py index 95b577c..fc028af 100644 --- a/danyapi/qwen/api.py +++ b/danyapi/qwen/api.py @@ -119,7 +119,16 @@ async def _prepare_session(account, pool, existing_sid: str | None, model_id: st return session, session_key -async def _send_completion(client: QwenClient, session, prompt: str, model_id: str, thinking: bool, search: bool, chat_type: str = "t2t"): +async def _send_completion( + client: QwenClient, + session, + prompt: str, + model_id: str, + thinking: bool, + search: bool, + chat_type: str = "t2t", + files: list[dict] | None = None, +): try: resp = await client.completion( chat_session_id=session.id, @@ -129,6 +138,7 @@ async def _send_completion(client: QwenClient, session, prompt: str, model_id: s thinking=thinking, search=search, chat_type=chat_type, + files=files, ) except httpx.HTTPStatusError as exc: raise HTTPException(exc.response.status_code, exc.response.text[:500]) from exc @@ -283,6 +293,7 @@ async def _collect_response( had_cached_session, tool_mode, tool_schemas, + files=None, ): stop_response_id: str | None = None stale_rebuilt = False @@ -291,7 +302,7 @@ async def _collect_response( try: while True: try: - resp = await _send_completion(account.client, session, prompt, model_id, thinking, search, chat_type) + resp = await _send_completion(account.client, session, prompt, model_id, thinking, search, chat_type, files=files) except ContextLimitError: _drop_session(pool, account, session_key) raise HTTPException(400, "context length exceeded: conversation too long, start a new conversation") from None @@ -381,6 +392,7 @@ async def collect_non_stream( tool_choice=None, response_format=None, user=None, + files=None, ): await _human_delay() async with account_lock(lock, settings.acquire_timeout): @@ -411,6 +423,7 @@ async def collect_non_stream( had_cached_session, tool_mode, tool_schemas, + files=files, ) if _is_context_limit(rec) and not rec.has_content: @@ -483,6 +496,7 @@ async def stream_openai( tool_choice=None, response_format=None, user=None, + files=None, ): chunk_id = f"chatcmpl-{uuid.uuid4().hex}" created = int(time.time()) @@ -512,7 +526,7 @@ async def stream_openai( attempt = 0 while True: try: - resp = await _send_completion(account.client, session, prompt, model_id, thinking, search) + resp = await _send_completion(account.client, session, prompt, model_id, thinking, search, files=files) except ContextLimitError: _drop_session(pool, account, session_key) for line in _stream_context_limit_lines(chunk_id, created, model, session_key): diff --git a/danyapi/qwen/client.py b/danyapi/qwen/client.py index 1595ce6..8761667 100644 --- a/danyapi/qwen/client.py +++ b/danyapi/qwen/client.py @@ -8,13 +8,16 @@ import httpx +from ..config import settings + log = logging.getLogger("danyapi.qwen") BASE_URL = "https://chat.qwen.ai" WEB_VERSION = "0.2.83" -USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/151.0.0.0 Safari/537.36" +USER_AGENT = settings.user_agent + COMMON_HEADERS = { "Accept": "application/json, text/plain, */*", @@ -62,22 +65,43 @@ def __init__( self, token: str | None = None, timeout: float = 60.0, + proxy: str | None = None, + user_agent: str | None = None, ) -> None: - self.token = token + raw_token = token + extracted_token = token + aux_cookies: dict[str, str] = {} + if raw_token and (";" in raw_token or "token=" in raw_token): + for item in raw_token.split(";"): + part = item.strip() + if "=" in part: + k, v = part.split("=", 1) + k, v = k.strip(), v.strip() + if k == "token": + extracted_token = v + elif k: + aux_cookies[k] = v + + self.token = extracted_token + ua = user_agent or settings.user_agent or USER_AGENT headers = { - "User-Agent": USER_AGENT, **COMMON_HEADERS, } - if token: - headers["Authorization"] = f"Bearer {token}" + if ua: + headers["User-Agent"] = ua + if extracted_token: + headers["Authorization"] = f"Bearer {extracted_token}" self.http = httpx.AsyncClient( base_url=BASE_URL, headers=headers, timeout=httpx.Timeout(timeout), follow_redirects=True, + proxy=proxy or settings.proxy, ) - if token: - self.http.cookies.set("token", token, domain="chat.qwen.ai", path="/") + if extracted_token: + self.http.cookies.set("token", extracted_token, domain="chat.qwen.ai", path="/") + for k, v in aux_cookies.items(): + self.http.cookies.set(k, v, domain="chat.qwen.ai", path="/") async def aclose(self) -> None: await self.http.aclose() @@ -132,6 +156,9 @@ async def check_auth(self) -> bool: if isinstance(nested, dict) and nested.get("id"): return True return bool(payload.get("id")) + except (httpx.ConnectError, httpx.ProxyError, httpx.TimeoutException) as exc: + log.error("upstream connection failed (%s): check network/proxy settings", exc) + return False except (httpx.HTTPError, ValueError): return False @@ -157,6 +184,45 @@ async def create_chat(self, model: str, chat_mode: str = "normal", chat_type: st log.info("qwen create chat success (%.0fms)", (time.monotonic() - started) * 1000) return chat_id + async def upload_file( + self, + data: bytes, + filename: str, + content_type: str = "image/jpeg", + ) -> dict: + """Uploads a file directly to Alibaba Cloud OSS and returns Qwen file attachment metadata. + + Ref: https://github.com/youssefvdel/qwengate/blob/dev/src/services/qwenFileUpload.ts + """ + from .upload import build_qwen_file_attachment, parse_and_poll, upload_to_oss + + is_image = content_type.startswith("image/") + filetype = "image" if is_image else "file" + body = { + "filename": filename, + "filesize": str(len(data)), + "filetype": filetype, + } + res = await self._post("/api/v2/files/getstsToken", json_body=body) + sts = res.get("data") if isinstance(res, dict) and "data" in res else res + if not isinstance(sts, dict) or not sts.get("file_id"): + raise QwenError(-1, f"getstsToken failed: invalid STS response: {res}") + + await upload_to_oss(self.http, sts, data, content_type) + + att = build_qwen_file_attachment( + sts=sts, + filename=filename, + filesize=len(data), + content_type=content_type, + attachment_type="image" if is_image else "file", + ) + + if not is_image: + await parse_and_poll(self, sts["file_id"]) + + return att + async def completion( self, chat_session_id: str, @@ -166,6 +232,7 @@ async def completion( thinking: bool = False, search: bool = False, chat_type: str = "t2t", + files: list[dict] | None = None, ) -> httpx.Response: log.debug("qwen completion start session=%s model=%s chat_type=%s", chat_session_id, model, chat_type) ts = int(datetime.datetime.now().timestamp()) @@ -188,7 +255,7 @@ async def completion( "role": "user", "content": prompt, "user_action": "chat", - "files": [], + "files": files or [], "timestamp": ts, "models": [model], "model": "", diff --git a/danyapi/qwen/upload.py b/danyapi/qwen/upload.py new file mode 100644 index 0000000..8681e7d --- /dev/null +++ b/danyapi/qwen/upload.py @@ -0,0 +1,218 @@ +# Source reference: +# https://github.com/youssefvdel/qwengate/blob/dev/src/services/qwenFileUpload.ts +# Implements Qwen web UI direct-to-Alibaba OSS file upload and attachment flow. + +from __future__ import annotations + +import asyncio +import base64 +import email.utils +import hmac +import logging +import time +import uuid +from typing import TYPE_CHECKING, Any + +import httpx + +if TYPE_CHECKING: + from .client import QwenClient + +__all__ = [ + "build_oss_canonical_request", + "build_qwen_file_attachment", + "hmac_sha1_base64", + "parse_and_poll", + "upload_to_oss", +] + +log = logging.getLogger("danyapi.qwen.upload") + + +def hmac_sha1_base64(key: str, message: str) -> str: + """HMAC-SHA1 Base64 digest for Alibaba OSS authorization.""" + sig = hmac.new(key.encode("utf-8"), message.encode("utf-8"), "sha1").digest() + return base64.b64encode(sig).decode("utf-8") + + +def build_oss_canonical_request( + method: str, + content_type: str, + date_str: str, + security_token: str, + bucket: str, + key: str, +) -> str: + """Builds CanonicalizedOSSHeaders and CanonicalizedResource string. + + Format: + VERB + "\\n" + + Content-MD5 + "\\n" + + Content-Type + "\\n" + + Date + "\\n" + + CanonicalizedOSSHeaders + + CanonicalizedResource + """ + return "\n".join( + [ + method, + "", # Content-MD5 (empty) + content_type, + date_str, + f"x-oss-security-token:{security_token}", + f"/{bucket}/{key}", + ] + ) + + +async def upload_to_oss( + http_client: httpx.AsyncClient, + sts: dict[str, Any], + file_bytes: bytes, + content_type: str, +) -> str: + """Uploads raw binary bytes to Alibaba Cloud OSS bucket using STS credentials. + + Ref: qwengate/src/services/qwenFileUpload.ts:uploadToOss + """ + date_str = email.utils.formatdate(usegmt=True) + key = sts.get("file_path", "") + bucket = sts.get("bucketname", "qwen-webui-prod") + + bucket_prefix = f"{bucket}/" + object_key = key.removeprefix(bucket_prefix) + + canonical_req = build_oss_canonical_request( + method="PUT", + content_type=content_type, + date_str=date_str, + security_token=sts["security_token"], + bucket=bucket, + key=object_key, + ) + + signature = hmac_sha1_base64(sts["access_key_secret"], canonical_req) + auth_header = f"OSS {sts['access_key_id']}:{signature}" + + endpoint = sts.get("endpoint", "").rstrip("/") + if bucket not in endpoint: + clean_endpoint = endpoint.replace("https://", "").replace("http://", "") + endpoint = f"https://{bucket}.{clean_endpoint}" + upload_url = f"{endpoint}/{object_key}" + + headers = { + "Content-Type": content_type, + "Date": date_str, + "Authorization": auth_header, + "x-oss-security-token": sts["security_token"], + } + + log.debug("uploading %d bytes to OSS: %s", len(file_bytes), upload_url) + resp = await http_client.put(upload_url, content=file_bytes, headers=headers, timeout=60.0) + if resp.status_code not in (200, 204): + body_sample = resp.text[:300] if hasattr(resp, "text") else "" + raise RuntimeError(f"OSS upload failed ({resp.status_code}): {body_sample}") + + return sts.get("file_url", upload_url) + + +def build_qwen_file_attachment( + sts: dict[str, Any], + filename: str, + filesize: int, + content_type: str, + attachment_type: str = "file", +) -> dict[str, Any]: + """Builds nested file descriptor matching Qwen web UI messages[].files[] format. + + Ref: qwengate/src/services/qwenFileUpload.ts:buildQwenFileAttachment + """ + file_path = sts.get("file_path", "") + user_id = file_path.split("/")[0] if "/" in file_path else "" + file_id = sts.get("file_id", "") + file_url = sts.get("file_url", "") + now_ms = int(time.time() * 1000) + + is_image = attachment_type == "image" or content_type.startswith("image/") + att_type = "image" if is_image else "file" + file_class = "vision" if is_image else "document" + show_type = "image" if is_image else "file" + + meta: dict[str, Any] = { + "name": filename, + "size": filesize, + "content_type": content_type, + } + if not is_image: + meta["parse_meta"] = {"parse_status": "success"} + + return { + "type": att_type, + "file": { + "created_at": now_ms, + "data": {}, + "filename": filename, + "hash": None, + "id": file_id, + "user_id": user_id, + "meta": meta, + "update_at": now_ms, + "lastModified": now_ms, + "name": filename, + "webkitRelativePath": "", + "size": filesize, + "type": content_type, + }, + "id": file_id, + "url": file_url, + "name": filename, + "collection_name": "", + "progress": 0, + "status": "uploaded", + "greenNet": "success", + "size": filesize, + "error": "", + "itemId": str(uuid.uuid4()), + "file_type": content_type, + "showType": show_type, + "file_class": file_class, + "uploadTaskId": str(uuid.uuid4()), + } + + +async def parse_and_poll( + client: QwenClient, + file_id: str, + max_wait_sec: float = 5.0, +) -> None: + """Triggers server-side text/doc parsing and polls until complete. + + Images do not require this step — Qwen vision processes them directly from OSS. + Ref: qwengate/src/services/qwenFileUpload.ts:parseFile & pollParseStatus + """ + try: + await client._post("/api/v2/files/parse", json_body={"file_id": file_id}) + except Exception as exc: + log.warning("file parse trigger failed for %s: %s", file_id, exc) + return + + start_time = time.monotonic() + while time.monotonic() - start_time < max_wait_sec: + try: + resp = await client._post( + "/api/v2/files/parse/status", + json_body={"file_id_list": [file_id]}, + ) + # Response: {"data": [{"file_id": "...", "status": "success"}]} + items = resp if isinstance(resp, list) else resp.get("data", []) + if isinstance(items, list) and items: + st = items[0].get("status") + if st == "success": + log.debug("parse complete for %s in %.2fs", file_id, time.monotonic() - start_time) + return + if st == "failed": + log.warning("file parsing failed for %s", file_id) + return + except Exception as exc: + log.debug("error polling parse status for %s: %s", file_id, exc) + await asyncio.sleep(1.0) diff --git a/danyapi/reg/captcha.py b/danyapi/reg/captcha.py index 12468e0..bafb6f6 100644 --- a/danyapi/reg/captcha.py +++ b/danyapi/reg/captcha.py @@ -6,6 +6,8 @@ import httpx +from ..config import settings + HCAPTCHA_SITEKEY = "352e5376-f2cc-43fe-a744-e51640449610" HCAPTCHA_PAGE_URL = "https://chat.deepseek.com/sign_up" @@ -59,7 +61,7 @@ def __init__(self, api_key: str, timeout: float = 180.0, poll_interval: float = self.poll_interval = poll_interval async def solve(self) -> str: - async with httpx.AsyncClient(timeout=30.0) as http: + async with httpx.AsyncClient(headers={"User-Agent": settings.user_agent}, timeout=30.0, proxy=settings.proxy) as http: try: resp = await http.get( "https://2captcha.com/in.php", @@ -105,7 +107,7 @@ def __init__(self, api_key: str, timeout: float = 180.0, poll_interval: float = self.poll_interval = poll_interval async def solve(self) -> str: - async with httpx.AsyncClient(timeout=30.0) as http: + async with httpx.AsyncClient(headers={"User-Agent": settings.user_agent}, timeout=30.0, proxy=settings.proxy) as http: try: resp = await http.post( "https://api.capsolver.com/createTask", diff --git a/danyapi/reg/deepseek.py b/danyapi/reg/deepseek.py index 4f86ab6..c8cc671 100644 --- a/danyapi/reg/deepseek.py +++ b/danyapi/reg/deepseek.py @@ -7,6 +7,7 @@ import httpx +from ..config import settings from ..pow import solve_challenge log = logging.getLogger("danyapi.reg.deepseek") @@ -75,13 +76,25 @@ def guest_pow_header(salt: str, answer: int) -> dict[str, str]: class DeepSeekRegistrar: - def __init__(self, device_id: str | None = None, timeout: float = 60.0, waf_token: str | None = None) -> None: + def __init__( + self, + device_id: str | None = None, + timeout: float = 60.0, + waf_token: str | None = None, + proxy: str | None = None, + user_agent: str | None = None, + ) -> None: self.device_id = device_id or new_device_id() + ua = user_agent or settings.user_agent or USER_AGENT + headers = {**CLIENT_HEADERS} + if ua: + headers["User-Agent"] = ua self.http = httpx.AsyncClient( base_url=BASE_URL, - headers={"User-Agent": USER_AGENT, **CLIENT_HEADERS}, + headers=headers, timeout=httpx.Timeout(timeout), follow_redirects=True, + proxy=proxy or settings.proxy, ) if waf_token: self.http.cookies.set("aws-waf-token", waf_token, domain="chat.deepseek.com", path="/") diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..2320221 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,15 @@ +services: + danyapi: + # you can build from source: docker compose up --build + build: . + # or comment above out, and load latest from repo. + # image: ghcr.io/mindflowgo/danyapi:latest + container_name: danyapi + restart: unless-stopped + ports: + - "8000:8000" + env_file: + - ./.env + volumes: + #- ./web:/app/web:ro + - /etc/localtime:/etc/localtime:ro diff --git a/docs/DOCS.md b/docs/DOCS.md new file mode 100644 index 0000000..0e481af --- /dev/null +++ b/docs/DOCS.md @@ -0,0 +1,255 @@ +# DANYAPI DOCS +-------------- +The DanyAPI provides a server-side dispatcher to communicate with Deepseek/Qwen. + +# CONFIGURATION +The config has 3 main areas: tokens, connections and settings (session, logging). + +### Tokens +Edit the *.env* file with the tokens. + +- **DeepSeek**: Open `chat.deepseek.com`, open DevTools Network tab, look for a request with `/chat/completion`, and copy the token from `Authorization: Bearer XXXXXXXXXXX`. +- **Qwen**: Open `chat.qwen.ai`, open DevTools Application/Storage -> Cookies (or Network tab), and copy the value of the `token` cookie (the JWT starting with `eyJhbGci...`). You can also paste your entire browser cookie header string into `QWEN_TOKENS`. + +Put that token into `DEEPSEEK_TOKENS` or `QWEN_TOKENS` in your `.env`. Multiple accounts can be comma-separated. Don't let untrusted users access this API. Treat this system as a privilege to use. + +### Connections +By default it binds to all IPs on the system at port 8000. If you want to run it through a proxy (ex. to make it appear to have a residential IP), you would set the proxy in DANYAPI_PROXY. + +If you want to force incoming requests to you to have a Bearer-token, you can set that in DANYAPI_API_KEY. If it's not set, all requests are accepted (only run it locally then!) + +You can customize the upstream User-Agent header sent to DeepSeek and Qwen by setting DANYAPI_USERAGENT (defaults to a modern Chrome browser string). + +### Settings +The 'session' is like a chat conversation. So each call you make can be grouped together in the same session (which saves lots of context tokens, as it remembers prior conversation). You can set the session_id to whatever, ex a name, etc. It saves 128 by default for 7 days, but examine the .env.example and change. + +### Logging & more +By default it logs everything to the docker log, or stdout but you can choose what to log, or even if a file. + +You can log tokens used, and auto-update. + + +# RUNNING +You can run it on your system, or more easily you can run it from docker: + +``` +docker compose up +``` + +It will show log output like: + ✔ Image danyapi-danyapi Built 149.6s + ✔ Container danyapi Recreated 0.5s +Attaching to danyapi + +danyapi | (21:37:33) outgoing IP: 99.88.77.66 (via proxy socks5://host.docker.internal:1080) +danyapi | (21:37:33) authentication: open (no API_KEY set) +danyapi | (21:37:35) deepseek accounts ready: 1 (deepseek-v4-flash, deepseek-v4-pro, deepseek-v4-vision) +danyapi | (21:37:35) default model: deepseek-v4-flash +danyapi | (21:37:35) DanyAPI running on http://0.0.0.0:8000 (Press CTRL+C to quit) +danyapi | (21:42:35) deepseek create session success (1520ms) +danyapi | (21:42:49) deepseek completion success (14234ms) +danyapi | (21:42:49) POST /v1/chat/completions success (17800ms) +danyapi | (21:59:23) deepseek completion success (2910ms) +danyapi | (21:59:23) POST /v1/chat/completions success (3650ms) + + +# ENDPOINTS + +#### LLM Endpoints (OpenAI-Compatible) +- POST **/v1/chat/completions** - Chat, reasoning (thinking), search, tools, sessions (session_id), file attachments, and streaming. + * messages (array, required): List of message objects (role: system | user | assistant, content: string). + * optional: model (string), stream (boolean, default: false), thinking (boolean), search (boolean), session_id (string): continue conversation, tools (array), file_ids (array): explicit file IDs to attach. Uploaded files for this session are also automatically attached. +- POST **/v1/images/generations** - Text-to-image generation powered by Qwen. + * prompt (string, required): Text description of the image to generate. + * optional: model (string, default: qwen-image-gen), n (integer, default: 1), size (string, default: 1024x1024), response_format (string, default: url). + +- _Note: If Bearer Auth given, that will be required in calls_. + +### File & Image Uploads (DeepSeek & Qwen) +- POST **/v1/files** - Upload files or images with automatic provider routing: + * **DeepSeek**: Uploads directly using dynamic Proof-of-Work (PoW) challenge solving. + * **Qwen**: Uploads directly to Alibaba Cloud OSS using STS temporary credentials. + * Files are streamed in-memory (no local disk storage) and automatically staged to attach to your next chat completion for that `session_id`. + * Parameters: `file` (binary / multipart, or base64 JSON string, required), optional: `session_id` (string), `model` (string, e.g. `qwen3.7-plus` or `deepseek-v4-vision`), `purpose` (string, default: `assistants`). +- GET **/v1/files/{file_id}** - Retrieve status and metadata for an uploaded or staged file. + +### Session Management (DeepSeek) +- GET **/v1/sessions** - List active chat sessions stored on DeepSeek across accounts. + * optional: account (integer): filter by account index, pinned (boolean, default: false), count (integer, default: 20, max: 100). +- GET **/v1/sessions/{session_id}** - Retrieve full conversation history and messages for a session. +- DELETE **/v1/sessions/{session_id}** - Delete a chat session from DeepSeek and evict from cache. + +### Models & Token Management +- GET **/v1/models** - List of all available DeepSeek and Qwen models. +- GET **/v1/usage** - Real-time request and token consumption statistics. +- POST **/v1/tokens** - Hot-add new tokens without restarting the server. + +### Health & Diagnostics +- GET **/health** - Provider readiness, active accounts, and cache metrics. + +### Web UI & Documentation +- GET **/** - Interactive token management and usage dashboard. +- GET **/docs/** - Landing page, playground, and guides. +- GET **/openapi.json** & **GET /redoc** - OpenAPI specs and ReDoc interactive viewer. + + +# EXAMPLES +See the models available: +``` +curl -s http://localhost:8000/v1/models +``` + +By default the first model with a valid token is selected if none specified (so if deepseek token available, deepseek-v4-flash is selected). + +We will use curl to show the same prompts: +``` +curl -s http://localhost:8000/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "messages": [ + {"role": "system", "content": "You are a helpful assistant that summarizes news articles."}, + {"role": "user", "content": "Please summarize the results of this poll: https://slashdot.org/poll/3284/how-much-of-your-coding-is-done-by-ai-coding-agents-these-days"} + ], + "session_id": "george"}' +``` + +Then follow-up chats woulds keep the same session_id, ex. +``` +curl -s http://localhost:8000/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "messages": [ + {"role": "system", "content": "You are a helpful assistant that summarizes news articles."}, + {"role": "user", "content": "What was the top pick?"} + ], + "session_id": "george"}' +``` + +### Image Generation (Qwen) + +If you have a Qwen token in this, you can generate images: +``` +curl -s http://localhost:8000/v1/images/generations \ + -H "Content-Type: application/json" \ + -d '{ + "prompt": "Futuristic skyline at dusk, cyberpunk style, digital art", + "size": "1024x1024" + }' +``` + +### Usage Tracking +You can use the webpage (if it's enabled to see from a browser), or: + +```bash +curl -s http://localhost:8000/v1/usage +``` + +### File & Image Uploads (DeepSeek & Qwen) +It can attempt to figure out the model (ex. if session already exists), but it's safest to send it with the file upload request, include the session too. + +Alternatively sending a URL will cause our backend to download it, and attach it (see #2). + +```bash +curl -s -X POST http://localhost:8000/v1/files \ + -F "file=@person.jpg" \ + -F "session_id=george-qwen" \ + -F "model=qwen3.7-plus" +``` + +Response: +```json +{ + "id": "b3224714-5c92-4c85-b3d7-992d546d14fc", + "object": "file", + "bytes": 25559, + "created_at": 1788810429, + "filename": "person.jpg", + "purpose": "assistants", + "session_id": "george-qwen", + "status": "processed", + "provider": "qwen" +} +``` + +Now query Qwen with the same `session_id` - the uploaded image is **automatically attached**: +```bash +curl -s http://localhost:8000/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen3.7-plus", + "session_id": "george-qwen", + "messages": [ + {"role": "user", "content": "Describe who is in the picture and what they are doing."} + ] + }' +``` + +--- + +#### 2. Direct Inline Multimodal Vision (No separate upload step) +You can also pass images directly into `POST /v1/chat/completions` using the standard OpenAI multimodal format with data URIs: +```bash +curl -s http://localhost:8000/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "qwen3.7-plus", + "messages": [ + { + "role": "user", + "content": [ + {"type": "text", "text": "Describe what is in this image:"}, + {"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,'$(base64 -i person.jpg)'"}} + ] + } + ] + }' +``` +DanyAPI automatically detects the target model (`qwen3.7-plus`), uploads the inline base64 image directly to Alibaba OSS, and sends it to Qwen on the fly. + +If you have the file_id from uploading previously, you can reference that again, ex. + +#### Explicit Attachment by File ID +You can reuse previously uploaded files across calls by passing `file_ids`: +```bash +curl -s http://localhost:8000/v1/chat/completions \ + -H "Content-Type: application/json" \ + -d '{ + "model": "deepseek-v4-flash", + "session_id": "george", + "file_ids": ["file-69723d3c-8b90-4b6e-9241-c9a08804e834"], + "messages": [ + {"role": "user", "content": "What are the main findings in this file?"} + ] + }' +``` + +--- + +#### 6. Inspect File Status +```bash +curl -s http://localhost:8000/v1/files/b3224714-5c92-4c85-b3d7-992d546d14fc +``` + +### Session Management + +Inspect, retrieve message history, or delete server-side sessions on DeepSeek: + +**List active sessions:** +``` +curl -s "http://localhost:8000/v1/sessions?count=10" +``` + +**Get session details and message contents:** +``` +curl -s http://localhost:8000/v1/sessions/george +``` + +**Delete a session:** +``` +curl -s -X DELETE http://localhost:8000/v1/sessions/george +``` + + + +# CONTRIBUTIONS +Qwen Upload Sourced From: https://github.com/youssefvdel/qwengate/tree/dev diff --git a/requirements.txt b/requirements.txt index 526f6ac..91cd0e4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,7 @@ fastapi uvicorn -httpx +httpx[socks] pydantic python-dotenv pillow +python-multipart diff --git a/tests/test_qwen_upload.py b/tests/test_qwen_upload.py new file mode 100644 index 0000000..a61ee56 --- /dev/null +++ b/tests/test_qwen_upload.py @@ -0,0 +1,89 @@ +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from danyapi.qwen.client import QwenClient +from danyapi.qwen.upload import ( + build_oss_canonical_request, + build_qwen_file_attachment, + hmac_sha1_base64, +) + + +def test_hmac_sha1_base64(): + key = "my-secret-key" + message = "test-message-to-sign" + sig = hmac_sha1_base64(key, message) + assert isinstance(sig, str) + assert len(sig) > 0 + # Deterministic check + assert sig == hmac_sha1_base64(key, message) + + +def test_build_oss_canonical_request(): + req = build_oss_canonical_request( + method="PUT", + content_type="image/jpeg", + date_str="Mon, 07 Sep 2026 19:42:56 GMT", + security_token="tok123", + bucket="qwen-webui-prod", + key="user1/file1_test.jpg", + ) + expected = "PUT\n\nimage/jpeg\nMon, 07 Sep 2026 19:42:56 GMT\nx-oss-security-token:tok123\n/qwen-webui-prod/user1/file1_test.jpg" + assert req == expected + + +def test_build_qwen_file_attachment(): + sts = { + "file_id": "fid-123", + "file_url": "https://oss.example.com/u1/fid-123_pic.jpg", + "file_path": "u1/fid-123_pic.jpg", + } + att = build_qwen_file_attachment(sts, "pic.jpg", 1024, "image/jpeg", "image") + assert att["type"] == "image" + assert att["id"] == "fid-123" + assert att["file"]["id"] == "fid-123" + assert att["file"]["user_id"] == "u1" + assert att["file"]["size"] == 1024 + assert att["showType"] == "image" + assert att["file_class"] == "vision" + + +def test_qwen_client_cookie_parsing(): + raw_cookie = "cna=test_cna; _bl_uid=uid123; token=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjEyMyJ9.sig; atpsida=atp123; isg=isg123" + client = QwenClient(token=raw_cookie) + assert client.token == "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpZCI6IjEyMyJ9.sig" + assert client.http.headers.get("Authorization") == f"Bearer {client.token}" + assert client.http.cookies.get("token") == client.token + assert client.http.cookies.get("cna") == "test_cna" + assert client.http.cookies.get("_bl_uid") == "uid123" + assert client.http.cookies.get("atpsida") == "atp123" + assert client.http.cookies.get("isg") == "isg123" + + +@pytest.mark.asyncio +async def test_qwen_upload_file_flow(): + client = QwenClient(token="jwt-token") + mock_sts = { + "file_id": "file-qwen-999", + "file_url": "https://qwen-webui-prod.oss-accelerate.aliyuncs.com/u1/file-qwen-999_test.jpg", + "file_path": "u1/file-qwen-999_test.jpg", + "bucketname": "qwen-webui-prod", + "endpoint": "https://oss-accelerate.aliyuncs.com", + "access_key_id": "STS.KEY", + "access_key_secret": "SECRET", + "security_token": "SEC_TOK", + } + client._post = AsyncMock(return_value={"success": True, "data": mock_sts}) + client.http.put = AsyncMock(return_value=MagicMock(status_code=200)) + + att = await client.upload_file(b"image-data", "test.jpg", "image/jpeg") + + assert att["id"] == "file-qwen-999" + assert att["type"] == "image" + assert att["file"]["size"] == 10 + client._post.assert_awaited_once_with( + "/api/v2/files/getstsToken", + json_body={"filename": "test.jpg", "filesize": "10", "filetype": "image"}, + ) + client.http.put.assert_awaited_once()