diff --git a/apps/README.md b/apps/README.md new file mode 100755 index 0000000..bf5690b --- /dev/null +++ b/apps/README.md @@ -0,0 +1,33 @@ +# `apps/` — 可部署的进程 + +| 目录 | 进程 | +| --- | --- | +| `api/` | FastAPI 应用:路由、依赖注入、中间件 | +| `worker/` | Celery 任务定义与四队列配置(gpu / cpu / io / beat) | +| `admin/` | 极简运营后台(占位,FR-ADM-01~05) | + +## API 层的职责边界(架构 §3.1 L3) + +**只做编排**:认证鉴权、参数校验、幂等、配额校验与预扣、统一响应、OpenAPI。 + +## 不能放什么 + +- **业务规则**。路由函数里出现 `if` 判断业务条件,几乎总是应该下沉到 + `packages/<域>/service.py` 或 `packages/domain/`。 +- **直接写 SQL 或用 ORM 查表**。走域的 `service`。 +- **直接调用外部 API**。LLM 走 `packages/llm`;未来的审核、视觉、换装同理。 +- 裸 `dict` 返回。所有响应必须经过 `packages.shared.response.ok()` / `fail()`。 + +## 契约先行(NFR-MAINT-01) + +先在 `apps/api/schemas.py` 定义 Pydantic schema,再写实现。客户端类型由 +OpenAPI 生成,schema 一变客户端立刻可见。 + +## Worker 的规矩 + +- 每个长耗时任务都必须落 `tk_task`,走 `packages/task` 的状态机。 + 不要自己维护状态字段。 +- 失败必须退还配额(`billing.refund()`,FR-BILL-05)。 + `apps/worker/tasks.py` 里的 `ping_task` 是形状示例。 +- 队列按资源画像分:GPU 推理进 `gpu`,视觉/渲染进 `cpu`,外部调用进 `io`。 + 混队列会让「换装排队把报告拖慢」。 diff --git a/apps/__init__.py b/apps/__init__.py new file mode 100644 index 0000000..b55802b --- /dev/null +++ b/apps/__init__.py @@ -0,0 +1 @@ +"""可部署进程:API / Celery worker / 运营后台。""" diff --git a/apps/admin/__init__.py b/apps/admin/__init__.py new file mode 100644 index 0000000..3ee4f89 --- /dev/null +++ b/apps/admin/__init__.py @@ -0,0 +1 @@ +"""极简运营后台(占位,FR-ADM-01~05)。""" diff --git a/apps/api/__init__.py b/apps/api/__init__.py new file mode 100644 index 0000000..50669c7 --- /dev/null +++ b/apps/api/__init__.py @@ -0,0 +1 @@ +"""FastAPI 应用(架构文档 §3.1 L3 接入编排层)。""" diff --git a/apps/api/deps.py b/apps/api/deps.py new file mode 100755 index 0000000..a6d7edb --- /dev/null +++ b/apps/api/deps.py @@ -0,0 +1,172 @@ +"""Composition root + FastAPI dependencies. + +This is the only place that wires concrete adapters into the domain packages. +Everything below it takes its collaborators as constructor arguments, which is +what keeps ``packages/domain`` and ``packages/llm`` unit-testable without IO. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from decimal import Decimal +from typing import Annotated + +from fastapi import Depends, Request +from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer +from sqlalchemy.ext.asyncio import AsyncSession + +from packages.agent.service import AgentService +from packages.billing.service import BillingService +from packages.identity.service import IdentityService +from packages.llm.budget import DailyBudgetGuard, RedisCostCounter +from packages.llm.circuit import CircuitBreakerRegistry +from packages.llm.cost import CompositeAICallRecorder, ModelRate, RateTable +from packages.llm.gateway import LLMGateway +from packages.llm.prompts import PromptRegistry +from packages.llm.providers.base import LLMProvider +from packages.llm.providers.deepseek import DeepSeekProvider +from packages.llm.providers.zhipu import ZhipuGLMProvider +from packages.llm.recording import PrometheusAICallRecorder, SqlAlchemyAICallRecorder +from packages.llm.routing import SceneRouter, build_scene_configs +from packages.shared.config import Settings, get_settings +from packages.shared.context import bind_context +from packages.shared.db import get_session_factory +from packages.shared.errors import AppError, ErrorCode +from packages.shared.redis_client import get_redis +from packages.shared.security import decode_access_token +from packages.task.service import TaskService + +bearer_scheme = HTTPBearer(auto_error=False) + + +def build_llm_gateway(settings: Settings) -> LLMGateway: + """Assemble the gateway from configuration (架构文档 §8.1).""" + providers: dict[str, LLMProvider] = { + "deepseek": DeepSeekProvider( + api_key=settings.deepseek_api_key, + base_url=settings.deepseek_base_url, + default_model=settings.deepseek_model, + ), + "zhipu": ZhipuGLMProvider( + api_key=settings.zhipu_api_key, + base_url=settings.zhipu_base_url, + default_model=settings.zhipu_model, + ), + } + + router = SceneRouter( + build_scene_configs( + primary_provider=settings.llm_primary_provider, + backup_provider=settings.llm_backup_provider, + structured_max_retries=settings.llm_structured_max_retries, + overrides=settings.llm_scene_overrides, + ) + ) + + rates = RateTable( + {model: ModelRate.from_mapping(rate) for model, rate in settings.llm_cost_rates.items()}, + default=ModelRate.from_mapping(settings.llm_default_cost_rate), + ) + + recorder = CompositeAICallRecorder( + SqlAlchemyAICallRecorder(get_session_factory()), + PrometheusAICallRecorder(), + ) + + budget = DailyBudgetGuard( + RedisCostCounter(get_redis()), + daily_budget_cents=Decimal(settings.llm_daily_budget_cents), + enabled=settings.llm_daily_budget_enabled, + tz_offset_hours=settings.llm_budget_timezone_offset_hours, + ) + + return LLMGateway( + providers=providers, + router=router, + prompts=PromptRegistry( + settings.prompts_dir, cache_prefix_mode=settings.llm_cache_prefix_mode + ), + rates=rates, + recorder=recorder, + budget=budget, + breakers=CircuitBreakerRegistry( + failure_threshold=settings.llm_circuit_failure_threshold, + cooldown_seconds=settings.llm_circuit_cooldown_seconds, + ), + ) + + +# --- request-scoped dependencies ------------------------------------------- + + +def settings_dep() -> Settings: + return get_settings() + + +async def session_dep() -> AsyncIterator[AsyncSession]: + factory = get_session_factory() + async with factory() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + + +def gateway_dep(request: Request) -> LLMGateway: + gateway = getattr(request.app.state, "llm_gateway", None) + if gateway is None: # pragma: no cover - startup guarantees this + raise AppError(ErrorCode.CONFIG_ERROR, message="LLM 网关未初始化") + return gateway + + +SettingsDep = Annotated[Settings, Depends(settings_dep)] +SessionDep = Annotated[AsyncSession, Depends(session_dep)] +GatewayDep = Annotated[LLMGateway, Depends(gateway_dep)] + + +async def current_user_id( + credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer_scheme)], + settings: SettingsDep, +) -> int: + """架构文档 §9.1: ``Authorization: Bearer ``.""" + if credentials is None or not credentials.credentials: + raise AppError(ErrorCode.UNAUTHENTICATED) + claims = decode_access_token( + credentials.credentials, + secret=settings.jwt_secret, + algorithm=settings.jwt_algorithm, + issuer=settings.jwt_issuer, + ) + bind_context(user_id=claims.user_id) + return claims.user_id + + +CurrentUserId = Annotated[int, Depends(current_user_id)] + + +def identity_service_dep(session: SessionDep) -> IdentityService: + return IdentityService(session) + + +def billing_service_dep(session: SessionDep, settings: SettingsDep) -> BillingService: + return BillingService(session, tz_offset_hours=settings.llm_budget_timezone_offset_hours) + + +def task_service_dep(session: SessionDep) -> TaskService: + return TaskService(session) + + +def agent_service_dep( + session: SessionDep, + gateway: GatewayDep, + billing: Annotated[BillingService, Depends(billing_service_dep)], +) -> AgentService: + return AgentService(session, gateway=gateway, billing=billing) + + +IdentityServiceDep = Annotated[IdentityService, Depends(identity_service_dep)] +BillingServiceDep = Annotated[BillingService, Depends(billing_service_dep)] +TaskServiceDep = Annotated[TaskService, Depends(task_service_dep)] +AgentServiceDep = Annotated[AgentService, Depends(agent_service_dep)] diff --git a/apps/api/main.py b/apps/api/main.py new file mode 100755 index 0000000..0d1f926 --- /dev/null +++ b/apps/api/main.py @@ -0,0 +1,78 @@ +"""FastAPI application factory (架构文档 §3.1 L3 API 层). + +The API layer only orchestrates: auth, validation, idempotency, quota, +unified response, OpenAPI. Business rules live in ``packages/``. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from fastapi import FastAPI + +from apps.api.deps import build_llm_gateway +from apps.api.routers import agent, auth, health, quota, tasks +from packages.shared.config import Settings, get_settings +from packages.shared.db import dispose_engine +from packages.shared.handlers import install_exception_handlers +from packages.shared.logging import configure_logging, get_logger +from packages.shared.middleware import RequestContextMiddleware, install_version_alias +from packages.shared.observability import setup_tracing +from packages.shared.redis_client import close_redis + +logger = get_logger(__name__) + + +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncIterator[None]: + settings: Settings = get_settings() + app.state.llm_gateway = build_llm_gateway(settings) + logger.info( + "api_started", + env=settings.app_env, + api_prefix=settings.api_prefix, + llm_primary=settings.llm_primary_provider, + llm_model=settings.deepseek_model, + ) + try: + yield + finally: + await app.state.llm_gateway.aclose() + await close_redis() + await dispose_engine() + logger.info("api_stopped") + + +def create_app(settings: Settings | None = None) -> FastAPI: + cfg = settings or get_settings() + configure_logging( + level=cfg.log_level, json_output=cfg.log_json, extra_redact_keys=cfg.extra_redact_keys + ) + + app = FastAPI( + title="美搭 backend", + version="0.1.0", + description=( + "AI 穿搭美学助手服务端。统一响应见架构文档 §9.2;" + "所有业务接口在 " + cfg.api_prefix + " 之下(/v1 为等价别名)。" + ), + openapi_url=f"{cfg.api_prefix}/openapi.json", + docs_url=f"{cfg.api_prefix}/docs" if cfg.is_local else None, + redoc_url=None, + lifespan=lifespan, + ) + + install_version_alias(app, cfg.api_prefix) + app.add_middleware(RequestContextMiddleware) + install_exception_handlers(app) + setup_tracing(app, cfg) + + app.include_router(health.router) + for router in (auth.router, agent.router, quota.router, tasks.router): + app.include_router(router, prefix=cfg.api_prefix) + + return app + + +app = create_app() diff --git a/apps/api/routers/__init__.py b/apps/api/routers/__init__.py new file mode 100644 index 0000000..281bc42 --- /dev/null +++ b/apps/api/routers/__init__.py @@ -0,0 +1 @@ +"""HTTP 路由。只做编排,业务规则下沉到 packages/。""" diff --git a/apps/api/routers/agent.py b/apps/api/routers/agent.py new file mode 100755 index 0000000..6c3f5e2 --- /dev/null +++ b/apps/api/routers/agent.py @@ -0,0 +1,64 @@ +"""``POST /agent/chat`` — the vertical slice. + +One request exercises: JWT auth → quota reserve → LLM gateway (routing, prompt +version, cache-prefix check, daily budget) → ``ob_ai_call`` cost row → quota +commit/refund → unified envelope. Streaming (SSE, FR-AGT-01) is intentionally +out of scope for the scaffold. +""" + +from __future__ import annotations + +import uuid + +from fastapi import APIRouter + +from apps.api.deps import AgentServiceDep, CurrentUserId +from apps.api.schemas import ChatRequest, ChatResponse, ChatUsage +from packages.llm.errors import ( + AllProvidersFailedError, + CachePrefixViolationError, + DailyBudgetExceededError, + StructuredOutputError, +) +from packages.shared.errors import AppError, ErrorCode +from packages.shared.response import ok + +router = APIRouter(prefix="/agent", tags=["agent"]) + + +@router.post("/chat", summary="Single-turn styling chat (non-streaming)") +async def chat( + payload: ChatRequest, + user_id: CurrentUserId, + agent: AgentServiceDep, +) -> dict[str, object]: + request_key = payload.request_key or uuid.uuid4().hex + try: + result = await agent.chat(user_id=user_id, message=payload.message, request_key=request_key) + except DailyBudgetExceededError as exc: + raise AppError(ErrorCode.LLM_DAILY_BUDGET_EXCEEDED) from exc + except StructuredOutputError as exc: + raise AppError(ErrorCode.LLM_STRUCTURED_OUTPUT_FAILED) from exc + except AllProvidersFailedError as exc: + raise AppError(ErrorCode.LLM_ALL_PROVIDERS_FAILED) from exc + except CachePrefixViolationError as exc: + raise AppError(ErrorCode.LLM_PROMPT_CACHE_VIOLATION) from exc + + return ok( + ChatResponse( + reply=result.reply, + provider=result.provider, + model=result.model, + prompt_id=result.prompt_id, + prompt_version=result.prompt_version, + call_id=str(result.call_id), + usage=ChatUsage( + tokens_in=result.tokens_in, + tokens_out=result.tokens_out, + tokens_cached=result.tokens_cached, + latency_ms=result.latency_ms, + cost_cents=str(result.cost_cents), + ), + quota_remaining=result.quota_remaining, + ).model_dump() + ) diff --git a/apps/api/routers/auth.py b/apps/api/routers/auth.py new file mode 100755 index 0000000..e36f0d0 --- /dev/null +++ b/apps/api/routers/auth.py @@ -0,0 +1,50 @@ +"""Development-only token issuance. + +**This is not the V1 auth design.** FR-ACC-01/04/05 require 华为账号登录、手机号 +验证码、双令牌与服务端可撤销的 refresh token. This endpoint exists so the +vertical slice can be exercised end to end, and it refuses to run unless +``DEV_AUTH_ENABLED`` is true and the environment is local/dev. +""" + +from __future__ import annotations + +from fastapi import APIRouter + +from apps.api.deps import IdentityServiceDep, SettingsDep +from apps.api.schemas import DevTokenRequest, DevTokenResponse +from packages.identity.service import DEV_PROVIDER +from packages.shared.errors import AppError, ErrorCode +from packages.shared.response import ok +from packages.shared.security import create_access_token + +router = APIRouter(prefix="/auth", tags=["auth"]) + + +@router.post("/dev-token", summary="[dev only] Issue an access token for testing") +async def dev_token( + payload: DevTokenRequest, + settings: SettingsDep, + identity: IdentityServiceDep, +) -> dict[str, object]: + if not (settings.dev_auth_enabled and settings.is_local): + raise AppError(ErrorCode.NOT_FOUND) + + user = await identity.get_or_create_by_external_identity( + provider=DEV_PROVIDER, + union_id=payload.device_key, + register_source="dev", + ) + token = create_access_token( + user_id=user.user_id, + secret=settings.jwt_secret, + algorithm=settings.jwt_algorithm, + ttl_seconds=settings.jwt_access_ttl_seconds, + issuer=settings.jwt_issuer, + ) + return ok( + DevTokenResponse( + access_token=token, + expires_in=settings.jwt_access_ttl_seconds, + user_id=str(user.user_id), + ).model_dump() + ) diff --git a/apps/api/routers/health.py b/apps/api/routers/health.py new file mode 100755 index 0000000..c0be456 --- /dev/null +++ b/apps/api/routers/health.py @@ -0,0 +1,59 @@ +"""Liveness, readiness and metrics. Mounted at the root, outside ``/api/v1``. + +``/healthz`` must never touch a dependency: if it did, a slow database would +make the orchestrator kill healthy processes. ``/readyz`` is the one that checks +dependencies, because a process that cannot reach PostgreSQL should be taken out +of the load-balancer rotation but not restarted. +""" + +from __future__ import annotations + +from fastapi import APIRouter, Response +from prometheus_client import CONTENT_TYPE_LATEST, generate_latest +from sqlalchemy import text + +from packages.shared.config import get_settings +from packages.shared.db import get_session_factory +from packages.shared.errors import AppError, ErrorCode +from packages.shared.logging import get_logger +from packages.shared.observability import REGISTRY +from packages.shared.redis_client import get_redis +from packages.shared.response import ok + +router = APIRouter(tags=["ops"]) +logger = get_logger(__name__) + + +@router.get("/healthz", summary="Liveness probe") +async def healthz() -> dict[str, object]: + settings = get_settings() + return ok({"status": "ok", "app": settings.app_name, "env": settings.app_env}) + + +@router.get("/readyz", summary="Readiness probe (checks PostgreSQL and Redis)") +async def readyz() -> dict[str, object]: + checks: dict[str, str] = {} + + try: + factory = get_session_factory() + async with factory() as session: + await session.execute(text("SELECT 1")) + checks["postgres"] = "ok" + except Exception as exc: + checks["postgres"] = f"error: {type(exc).__name__}" + + try: + await get_redis().ping() + checks["redis"] = "ok" + except Exception as exc: + checks["redis"] = f"error: {type(exc).__name__}" + + if any(v != "ok" for v in checks.values()): + logger.warning("readiness_failed", checks=checks) + raise AppError(ErrorCode.UPSTREAM_ERROR, message="依赖未就绪", data={"checks": checks}) + return ok({"status": "ready", "checks": checks}) + + +@router.get("/metrics", summary="Prometheus scrape endpoint", include_in_schema=False) +async def metrics() -> Response: + return Response(content=generate_latest(REGISTRY), media_type=CONTENT_TYPE_LATEST) diff --git a/apps/api/routers/quota.py b/apps/api/routers/quota.py new file mode 100755 index 0000000..47503ef --- /dev/null +++ b/apps/api/routers/quota.py @@ -0,0 +1,61 @@ +"""``GET /quota`` and ``GET /quota/ledger`` (架构文档 §9.3 配额). + +FR-BILL-07: 额度相关文案中**不得出现任何价格、购买、会员、升级字样**。 +Only two recovery paths may ever be surfaced: 等待恢复 and 做任务换额度. +""" + +from __future__ import annotations + +from fastapi import APIRouter, Query + +from apps.api.deps import BillingServiceDep, CurrentUserId +from apps.api.schemas import LedgerEntry, QuotaResponse +from packages.agent.service import QUOTA_KEY_AGENT_CHAT +from packages.shared.response import ok + +router = APIRouter(prefix="/quota", tags=["quota"]) + + +@router.get("", summary="Current period quota for a feature") +async def get_quota( + user_id: CurrentUserId, + billing: BillingServiceDep, + quota_key: str = Query(default=QUOTA_KEY_AGENT_CHAT, max_length=64), +) -> dict[str, object]: + snapshot = await billing.get_quota(user_id=user_id, quota_key=quota_key) + return ok( + QuotaResponse( + quota_key=snapshot.quota_key, + period_key=snapshot.period_key, + used=snapshot.used, + limit_value=snapshot.limit_value, + bonus_value=snapshot.bonus_value, + remaining=snapshot.remaining, + reset_at=snapshot.reset_at, + ).model_dump() + ) + + +@router.get("/ledger", summary="Quota ledger — 流水是真相,余额是缓存") +async def get_ledger( + user_id: CurrentUserId, + billing: BillingServiceDep, + limit: int = Query(default=50, ge=1, le=200), +) -> dict[str, object]: + rows = await billing.list_ledger(user_id=user_id, limit=limit) + return ok( + [ + LedgerEntry( + tx_id=str(row.tx_id), + delta=row.delta, + tx_type=row.tx_type, + status=row.status, + reason=row.reason, + quota_key=row.quota_key, + period_key=row.period_key, + balance_after=row.balance_after, + created_at=row.created_at, + ).model_dump() + for row in rows + ] + ) diff --git a/apps/api/routers/tasks.py b/apps/api/routers/tasks.py new file mode 100755 index 0000000..4ce43c0 --- /dev/null +++ b/apps/api/routers/tasks.py @@ -0,0 +1,80 @@ +"""任务中心 endpoints + a trivial async task that proves the Celery path. + +The ``ping`` task carries no business value; it exists so that the ``tk_task`` +state machine, the queue routing and the worker deployment are all exercised +before any real long-running task is written (M1 出口条件). +""" + +from __future__ import annotations + +import uuid + +from fastapi import APIRouter, Query + +from apps.api.deps import CurrentUserId, TaskServiceDep +from apps.api.schemas import TaskResponse, TaskSubmitRequest +from packages.domain.task.state import TaskStatus, TaskType +from packages.shared.models import Task +from packages.shared.response import ok + +router = APIRouter(prefix="/tasks", tags=["tasks"]) + + +def _serialize(task: Task) -> dict[str, object]: + return TaskResponse( + task_id=str(task.task_id), + task_type=task.task_type, + status=task.status, + progress=task.progress, + stage_text=task.stage_text, + result=task.result, + error_code=task.error_code, + error_message=task.error_message, + retry_count=task.retry_count, + submitted_at=task.submitted_at, + started_at=task.started_at, + finished_at=task.finished_at, + ).model_dump() + + +@router.post("/ping", summary="Submit the scaffold's trivial async task") +async def submit_ping( + payload: TaskSubmitRequest, + user_id: CurrentUserId, + tasks: TaskServiceDep, +) -> dict[str, object]: + from apps.worker.tasks import ping_task + + idempotency_key = payload.idempotency_key or f"ping:{user_id}:{uuid.uuid4().hex}" + task = await tasks.submit( + user_id=user_id, + task_type=TaskType.PING, + idempotency_key=idempotency_key, + payload={"echo": payload.echo}, + eta_seconds=2, + ) + if task.status == TaskStatus.PENDING.value: + task = await tasks.transition( + task_id=task.task_id, target=TaskStatus.QUEUED, stage_text="排队中" + ) + ping_task.apply_async(args=[task.task_id], queue="cpu") + return ok(_serialize(task)) + + +@router.get("/{task_id}", summary="Task status (ownership checked)") +async def get_task( + task_id: int, + user_id: CurrentUserId, + tasks: TaskServiceDep, +) -> dict[str, object]: + return ok(_serialize(await tasks.get(task_id=task_id, user_id=user_id))) + + +@router.get("", summary="Task centre list") +async def list_tasks( + user_id: CurrentUserId, + tasks: TaskServiceDep, + limit: int = Query(default=20, ge=1, le=100), +) -> dict[str, object]: + rows = await tasks.list_for_user(user_id=user_id, limit=limit) + return ok([_serialize(row) for row in rows]) diff --git a/apps/api/schemas.py b/apps/api/schemas.py new file mode 100755 index 0000000..a937ff5 --- /dev/null +++ b/apps/api/schemas.py @@ -0,0 +1,98 @@ +"""Request/response schemas. + +NFR-MAINT-01 契约先行: schema 先定义,客户端类型由 OpenAPI 生成。 +Wire responses are wrapped by the unified envelope (架构文档 §9.2); these models +describe the ``data`` payload only. +""" + +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, Field + + +class DevTokenRequest(BaseModel): + device_key: str = Field( + default="local-dev", + min_length=1, + max_length=64, + description="任意稳定字符串;同一 device_key 复用同一测试用户", + ) + + +class DevTokenResponse(BaseModel): + access_token: str + token_type: str = "Bearer" # noqa: S105 — a scheme name, not a credential + expires_in: int + user_id: str = Field(description="雪花 ID 以字符串返回,避免 JS 精度丢失") + + +class ChatRequest(BaseModel): + message: str = Field(min_length=1, max_length=2000) + request_key: str | None = Field( + default=None, + max_length=128, + description="幂等键;相同键不会重复扣配额 (架构文档 §9.1)", + ) + + +class ChatUsage(BaseModel): + tokens_in: int + tokens_out: int + tokens_cached: int + latency_ms: int + cost_cents: str + + +class ChatResponse(BaseModel): + reply: str + provider: str + model: str + prompt_id: str | None + prompt_version: str | None + call_id: str + usage: ChatUsage + quota_remaining: int + + +class QuotaResponse(BaseModel): + quota_key: str + period_key: str + used: int + limit_value: int + bonus_value: int + remaining: int + reset_at: datetime | None + + +class LedgerEntry(BaseModel): + tx_id: str + delta: int + tx_type: str + status: str + reason: str | None + quota_key: str | None + period_key: str | None + balance_after: int + created_at: datetime + + +class TaskSubmitRequest(BaseModel): + echo: str = Field(default="ping", max_length=200) + idempotency_key: str | None = Field(default=None, max_length=128) + + +class TaskResponse(BaseModel): + task_id: str + task_type: str + status: str + progress: int + stage_text: str | None + result: dict[str, object] | None + error_code: int | None + error_message: str | None + retry_count: int + submitted_at: datetime + started_at: datetime | None + finished_at: datetime | None diff --git a/apps/worker/__init__.py b/apps/worker/__init__.py new file mode 100644 index 0000000..ff2858b --- /dev/null +++ b/apps/worker/__init__.py @@ -0,0 +1 @@ +"""Celery 任务定义与四队列配置(gpu/cpu/io/beat)。""" diff --git a/apps/worker/celery_app.py b/apps/worker/celery_app.py new file mode 100755 index 0000000..964a987 --- /dev/null +++ b/apps/worker/celery_app.py @@ -0,0 +1,54 @@ +"""Celery application with the four queues from TD-11 / 架构文档 §3.1. + + gpu 换装、发型、抠图 —— 并发上限由 GPU 容量决定,队列即容量 (§7.3) + cpu 报告视觉分析、渲染 + io 外部 API 调用、推送、上传 + beat 定时任务(单实例,防重) + +Queue isolation is the point: 换装排队不能把报告生成拖慢。 +""" + +from __future__ import annotations + +from celery import Celery +from kombu import Queue + +from packages.shared.config import get_settings +from packages.shared.logging import configure_logging + +settings = get_settings() +configure_logging( + level=settings.log_level, + json_output=settings.log_json, + extra_redact_keys=settings.extra_redact_keys, +) + +celery_app = Celery( + "meida", + broker=settings.celery_broker_url, + backend=settings.celery_result_backend, + include=["apps.worker.tasks"], +) + +celery_app.conf.update( + task_default_queue="cpu", + task_queues=( + Queue("gpu"), + Queue("cpu"), + Queue("io"), + Queue("beat"), + ), + task_routes={ + "meida.tryon.*": {"queue": "gpu"}, + "meida.matting.*": {"queue": "gpu"}, + "meida.report.*": {"queue": "cpu"}, + "meida.notify.*": {"queue": "io"}, + }, + task_acks_late=True, + task_reject_on_worker_lost=True, + task_track_started=True, + worker_prefetch_multiplier=1, + broker_connection_retry_on_startup=True, + timezone="Asia/Shanghai", + result_expires=86400, +) diff --git a/apps/worker/tasks.py b/apps/worker/tasks.py new file mode 100755 index 0000000..6f00824 --- /dev/null +++ b/apps/worker/tasks.py @@ -0,0 +1,68 @@ +"""Celery task definitions. + +Only the scaffold's ``ping`` task lives here. It proves the whole async path: +API creates ``tk_task`` → queue → worker transitions running → succeeded → +task centre reflects it. Real long-running tasks (report / tryon / matting / +render) follow exactly this shape and additionally call +``billing.commit()``/``refund()`` and ``compliance.publish_asset()``. +""" + +from __future__ import annotations + +import asyncio +from typing import Any + +from apps.worker.celery_app import celery_app +from packages.domain.task.state import TaskStatus +from packages.shared.context import bind_context +from packages.shared.db import get_session_factory +from packages.shared.logging import get_logger +from packages.task.service import TaskService + +logger = get_logger(__name__) + + +async def _run_ping(task_id: int, echo: str | None = None) -> dict[str, Any]: + factory = get_session_factory() + async with factory() as session: + service = TaskService(session) + task = await service.transition( + task_id=task_id, target=TaskStatus.RUNNING, stage_text="执行中" + ) + payload = task.payload or {} + message = echo or str(payload.get("echo", "pong")) + await session.commit() + + finished = await service.transition( + task_id=task_id, + target=TaskStatus.SUCCEEDED, + stage_text="已完成", + result={"echo": message}, + ) + await session.commit() + return {"task_id": str(finished.task_id), "echo": message} + + +async def _fail_ping(task_id: int, error: str) -> None: + factory = get_session_factory() + async with factory() as session: + service = TaskService(session) + await service.transition( + task_id=task_id, + target=TaskStatus.FAILED, + stage_text="执行失败", + error_message=error, + ) + await session.commit() + + +@celery_app.task(name="meida.task.ping", bind=True, max_retries=2) +def ping_task(self: Any, task_id: int, echo: str | None = None) -> dict[str, Any]: + bind_context(task_id=task_id) + try: + return asyncio.run(_run_ping(task_id, echo)) + except Exception as exc: + logger.exception("ping_task_failed", task_id=task_id) + # Real tasks additionally refund the quota reservation here (FR-BILL-05). + asyncio.run(_fail_ping(task_id, f"{type(exc).__name__}: {exc}")) + raise