diff --git a/packages/README.md b/packages/README.md new file mode 100755 index 0000000..31e0c36 --- /dev/null +++ b/packages/README.md @@ -0,0 +1,46 @@ +# `packages/` — 业务域与能力包 + +架构 §3.2:**单体不等于混乱**。这里的每个子目录都是一个边界,越界会被 CI 拦下。 + +## 目录含义 + +| 包 | 类型 | 内容 | +| --- | --- | --- | +| `domain/` | 纯逻辑层 | 领域模型与业务规则。**无 IO、无框架、可纯单测** | +| `shared/` | 基础设施 | 配置、日志、错误码、统一响应、DB、Redis、可观测 | +| `llm/` | AI 能力 | LLM 网关(已实现) | +| `retrieval/` `rules/` `memory/` | AI 能力 | 混合检索 / 规则引擎 / 记忆系统(占位) | +| `compliance/` | 业务域 | 统一合规出口(占位,CR-05/CR-06) | +| `identity/` `billing/` `task/` `agent/` | 业务域 | 已有实现 | +| `profile/` `report/` `tryon/` `kb/` `notify/` `wardrobe/` | 业务域 | 占位 | + +每个业务域包的标准结构: + +``` +packages/<域>/ +├─ __init__.py +├─ README.md 这个域负责什么、不负责什么 +├─ service.py 对外唯一入口。其他域只能调这里 +└─ repo.py SQL。只有本域可以 import +``` + +## 硬性规则(由 `make importlint` 机器校验) + +1. **禁止跨域直接查表**。要用别的域的数据,调它的 `service`,不要 import 它的 + `repo`,更不要在自己的 SQL 里 JOIN 别人的表。表名的域前缀(`id_` / `bl_` / + `tk_` / …)就是为将来拆库预留的边界。 +2. **`packages/domain/` 必须保持纯净**。不许 import `fastapi`、`sqlalchemy`、 + `httpx`、`redis`、`celery`,也不许 import `packages/shared`。放不进去说明 + 它不是领域逻辑,是应用逻辑,应该待在 `service.py`。 +3. **`compliance` 不得依赖任何业务域**。它必须能被任何模块安全调用。 +4. **`identity` 不得依赖任何其他业务域**。登录链路不能被业务拖累。 +5. **AI 能力包(`llm` / `retrieval` / `rules`)不得依赖业务域**,也不得 import + `apps`。它们要能在 V2 原样拆成独立服务。 + +## 加一个新业务域时 + +1. 建目录,写 `README.md` 说明职责与不负责的事。 +2. 在 `pyproject.toml` 的 `[tool.importlinter]` 契约里把它加进相应的 + `layers` / `forbidden_modules` 列表 —— 新域默认没有约束,忘了加就等于没有边界。 +3. 领域规则先写进 `packages/domain/<域>/` 并配单测,再写 `service.py`。 +4. 表加域前缀,走 Alembic,迁移必须向后兼容(NFR-MAINT-06)。 diff --git a/packages/__init__.py b/packages/__init__.py new file mode 100644 index 0000000..096fbb3 --- /dev/null +++ b/packages/__init__.py @@ -0,0 +1 @@ +"""业务域与能力包。域边界由 import-linter 强制(架构文档 §4.2)。""" diff --git a/packages/domain/README.md b/packages/domain/README.md new file mode 100755 index 0000000..fbf1f7f --- /dev/null +++ b/packages/domain/README.md @@ -0,0 +1,39 @@ +# `packages/domain/` — 纯领域逻辑 + +架构 §3.2 硬性规则 2:**领域逻辑放在这里,不依赖 FastAPI、不依赖数据库会话, +可纯单元测试。** + +## 能放什么 + +- 数据类与值对象(配额状态、任务状态、诊断维度…) +- 判定与计算(配额是否够、状态能否流转、置信度冲突消解、色差计算…) +- 纯函数、不可变数据结构 + +## 绝对不能放什么 + +- `import fastapi` / `sqlalchemy` / `httpx` / `redis` / `celery` +- `import packages.shared`(连日志和配置都不行) +- 任何 IO:读文件、发请求、查库、读环境变量 +- `async def` 里真的 await 了 IO 的东西 + +违反会被 import-linter 的 `packages.domain is pure` 契约在 CI 拦下。 + +## 为什么这么严 + +这是 AI 大量写代码时唯一还能快速验证正确性的地方 —— 没有 mock、没有 +fixture、没有容器,几毫秒跑完几百个用例。一旦让 IO 渗进来,这个性质就没了, +后面所有业务规则的测试都会退化成集成测试。 + +## mypy strict + +本目录与 `packages/llm/` 一起,是 mypy **strict** 模式覆盖的范围 +(架构 §16.2)。所有函数必须有完整类型注解。 + +## 现有内容 + +| 模块 | 内容 | +| --- | --- | +| `ids.py` | 雪花 ID 生成(§6.1 原则 2) | +| `errors.py` | 纯领域异常,适配层负责翻译成 `AppError` | +| `billing/quota.py` | 配额三段式的算术与周期键(§7.4) | +| `task/state.py` | `tk_task` 状态机(§6.2) | diff --git a/packages/domain/__init__.py b/packages/domain/__init__.py new file mode 100644 index 0000000..7091c6e --- /dev/null +++ b/packages/domain/__init__.py @@ -0,0 +1 @@ +"""纯领域逻辑:无 IO、无框架、可纯单测(架构文档 §3.2)。""" diff --git a/packages/domain/billing/__init__.py b/packages/domain/billing/__init__.py new file mode 100644 index 0000000..77e3c69 --- /dev/null +++ b/packages/domain/billing/__init__.py @@ -0,0 +1 @@ +"""配额与账本的纯逻辑(架构文档 §7.4)。""" diff --git a/packages/domain/billing/quota.py b/packages/domain/billing/quota.py new file mode 100755 index 0000000..6086803 --- /dev/null +++ b/packages/domain/billing/quota.py @@ -0,0 +1,140 @@ +"""Quota arithmetic — pure logic, no IO (架构文档 §7.4 V1 配额三段式). + +The three-phase protocol lives here as data transformations so it can be unit +tested without a database: + + reserve → 冻结额度并写 bl_ledger(status=reserved) + commit → 确认扣除 (status=committed) + refund → 释放额度并写反向流水 (tx_type=refund) + +V1 doesn't involve money, but 用户对「次数被白扣」的敏感度不比扣钱低,and V1 +has no customer-service refund path — the mechanism is the only guarantee. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from datetime import UTC, datetime, timedelta, timezone +from typing import Literal + +from packages.domain.errors import QuotaExceededError + +PeriodType = Literal["day", "month", "lifetime"] +LedgerStatus = Literal["reserved", "committed", "refunded"] +TxType = Literal["system_grant", "consume", "refund", "purchase", "gift"] + +#: FR-BILL-01: 跨时区按 UTC+8 统一。 +DEFAULT_TZ_OFFSET_HOURS = 8 + + +def period_key( + now: datetime, + period_type: PeriodType, + tz_offset_hours: int = DEFAULT_TZ_OFFSET_HOURS, +) -> str: + """Bucket key that makes quota reset naturally, with no cron job. + + ``day`` → ``d:2026-08-13``; ``month`` → ``m:2026-08``; ``lifetime`` → ``life``. + """ + if period_type == "lifetime": + return "life" + local = now.astimezone(UTC) + timedelta(hours=tz_offset_hours) + if period_type == "day": + return f"d:{local:%Y-%m-%d}" + if period_type == "month": + return f"m:{local:%Y-%m}" + msg = f"unknown period_type: {period_type}" + raise ValueError(msg) + + +def period_reset_at( + now: datetime, + period_type: PeriodType, + tz_offset_hours: int = DEFAULT_TZ_OFFSET_HOURS, +) -> datetime | None: + """When the current period rolls over (used for the `reset_at` hint).""" + if period_type == "lifetime": + return None + tz = timezone(timedelta(hours=tz_offset_hours)) + local = now.astimezone(tz) + if period_type == "day": + nxt = (local + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0) + else: + year, month = (local.year + 1, 1) if local.month == 12 else (local.year, local.month + 1) + nxt = local.replace( + year=year, month=month, day=1, hour=0, minute=0, second=0, microsecond=0 + ) + return nxt.astimezone(UTC) + + +@dataclass(frozen=True, slots=True) +class QuotaState: + """Current period counters for one (user, quota_key).""" + + used: int + base_limit: int + bonus: int = 0 + + @property + def effective_limit(self) -> int: + return self.base_limit + self.bonus + + @property + def remaining(self) -> int: + return max(self.effective_limit - self.used, 0) + + +@dataclass(frozen=True, slots=True) +class QuotaDecision: + allowed: bool + required: int + remaining: int + effective_limit: int + + +def evaluate(state: QuotaState, amount: int) -> QuotaDecision: + """Server-side hard check (FR-BILL-06). Client hints are UX only.""" + if amount <= 0: + msg = "amount must be positive" + raise ValueError(msg) + return QuotaDecision( + allowed=state.used + amount <= state.effective_limit, + required=amount, + remaining=state.remaining, + effective_limit=state.effective_limit, + ) + + +def reserve(state: QuotaState, amount: int) -> QuotaState: + """Freeze ``amount`` units; raises :class:`QuotaExceededError` when over the limit.""" + decision = evaluate(state, amount) + if not decision.allowed: + raise QuotaExceededError(required=amount, remaining=decision.remaining) + return replace(state, used=state.used + amount) + + +def refund(state: QuotaState, amount: int) -> QuotaState: + """Release a previously reserved amount. Never drives ``used`` below zero.""" + if amount <= 0: + msg = "amount must be positive" + raise ValueError(msg) + return replace(state, used=max(state.used - amount, 0)) + + +def balance_after(state: QuotaState) -> int: + """Value written to ``bl_ledger.balance_after``. + + V1 has no wallet balance, so the ledger records *remaining quota* — that is + the number the user actually sees ("今日剩余 N 次"). When V2 introduces + ``bl_wallet`` this becomes the wallet balance without a schema change. + """ + return state.remaining + + +def reserve_idempotency_key(*, user_id: int, quota_key: str, request_key: str) -> str: + return f"rsv:{user_id}:{quota_key}:{request_key}" + + +def refund_idempotency_key(reserve_key: str) -> str: + """FR-BILL-05: 退还必须幂等。Derived, so a retry cannot double-refund.""" + return f"rfd:{reserve_key}" diff --git a/packages/domain/errors.py b/packages/domain/errors.py new file mode 100755 index 0000000..3dda900 --- /dev/null +++ b/packages/domain/errors.py @@ -0,0 +1,26 @@ +"""Pure-domain exceptions. + +This layer must not import :mod:`packages.shared` (enforced by import-linter), +so it defines its own error types. Adapters translate them into +:class:`packages.shared.errors.AppError` at the boundary. +""" + +from __future__ import annotations + + +class DomainError(Exception): + """Base class for all pure-domain rule violations.""" + + +class QuotaExceededError(DomainError): + def __init__(self, *, required: int, remaining: int) -> None: + self.required = required + self.remaining = remaining + super().__init__(f"quota exceeded: required={required} remaining={remaining}") + + +class InvalidTaskTransitionError(DomainError): + def __init__(self, *, current: str, target: str) -> None: + self.current = current + self.target = target + super().__init__(f"invalid task transition: {current} -> {target}") diff --git a/packages/domain/ids.py b/packages/domain/ids.py new file mode 100755 index 0000000..d6efd42 --- /dev/null +++ b/packages/domain/ids.py @@ -0,0 +1,85 @@ +"""Snowflake IDs (架构文档 §6.1 原则 2). + +Auto-increment keys invite enumeration attacks and collide when the database is +later sharded; UUID primary keys have poor index locality. A 64-bit snowflake +avoids both. + +Layout: 41 bits ms since epoch | 10 bits node id | 12 bits sequence. +""" + +from __future__ import annotations + +import threading +import time +from collections.abc import Callable + +#: 2025-01-01T00:00:00Z in milliseconds. Do not change once IDs are issued. +DEFAULT_EPOCH_MS = 1_735_689_600_000 + +_NODE_BITS = 10 +_SEQUENCE_BITS = 12 +_MAX_NODE_ID = (1 << _NODE_BITS) - 1 +_SEQUENCE_MASK = (1 << _SEQUENCE_BITS) - 1 + + +def _wall_clock_ms() -> int: + return int(time.time() * 1000) + + +class SnowflakeGenerator: + """Thread-safe monotonic 64-bit id generator.""" + + __slots__ = ("_clock", "_epoch_ms", "_last_ms", "_lock", "_node_id", "_sequence") + + def __init__( + self, + node_id: int = 1, + epoch_ms: int = DEFAULT_EPOCH_MS, + clock: Callable[[], int] | None = None, + ) -> None: + if not 0 <= node_id <= _MAX_NODE_ID: + msg = f"node_id must be within 0..{_MAX_NODE_ID}" + raise ValueError(msg) + self._node_id = node_id + self._epoch_ms = epoch_ms + self._sequence = 0 + self._last_ms = -1 + self._lock = threading.Lock() + self._clock = clock or _wall_clock_ms + + def _now_ms(self) -> int: + return self._clock() + + def next_id(self) -> int: + with self._lock: + now = self._now_ms() + if now < self._last_ms: + # Clock moved backwards: wait it out rather than issue duplicates. + now = self._last_ms + if now == self._last_ms: + self._sequence = (self._sequence + 1) & _SEQUENCE_MASK + if self._sequence == 0: + while now <= self._last_ms: + now = self._now_ms() + else: + self._sequence = 0 + self._last_ms = now + elapsed = now - self._epoch_ms + return ( + (elapsed << (_NODE_BITS + _SEQUENCE_BITS)) + | (self._node_id << _SEQUENCE_BITS) + | self._sequence + ) + + +_default_generator = SnowflakeGenerator() + + +def next_id() -> int: + """Process-wide default generator.""" + return _default_generator.next_id() + + +def timestamp_ms_of(snowflake_id: int, epoch_ms: int = DEFAULT_EPOCH_MS) -> int: + """Extract the creation timestamp from an id (useful for partition pruning).""" + return (snowflake_id >> (_NODE_BITS + _SEQUENCE_BITS)) + epoch_ms diff --git a/packages/domain/task/__init__.py b/packages/domain/task/__init__.py new file mode 100644 index 0000000..a5598e1 --- /dev/null +++ b/packages/domain/task/__init__.py @@ -0,0 +1 @@ +"""统一任务中心的状态机(架构文档 §6.2)。""" diff --git a/packages/domain/task/state.py b/packages/domain/task/state.py new file mode 100755 index 0000000..2aa4c3e --- /dev/null +++ b/packages/domain/task/state.py @@ -0,0 +1,76 @@ +"""tk_task state machine — pure logic (架构文档 §6.2 task 域, FR-TASK-01). + +统一任务中心 is the only long-running-task view in the system; every long task +(report / tryon / hairstyle / matting / render) moves through these states. +""" + +from __future__ import annotations + +from enum import StrEnum + +from packages.domain.errors import InvalidTaskTransitionError + + +class TaskStatus(StrEnum): + PENDING = "pending" # created, quota reserved, not yet queued + QUEUED = "queued" # handed to Celery + RUNNING = "running" + SUCCEEDED = "succeeded" + FAILED = "failed" + CANCELED = "canceled" + + +class TaskType(StrEnum): + REPORT = "report" + TRYON = "tryon" + HAIRSTYLE = "hairstyle" + MATTING = "matting" + RENDER = "render" + PING = "ping" # scaffold-only: proves the async path end to end + + +TERMINAL_STATUSES: frozenset[TaskStatus] = frozenset( + {TaskStatus.SUCCEEDED, TaskStatus.FAILED, TaskStatus.CANCELED} +) + +#: Refunds are owed for every terminal state other than success (FR-TRY-05). +REFUNDABLE_STATUSES: frozenset[TaskStatus] = frozenset({TaskStatus.FAILED, TaskStatus.CANCELED}) + +ALLOWED_TRANSITIONS: dict[TaskStatus, frozenset[TaskStatus]] = { + TaskStatus.PENDING: frozenset({TaskStatus.QUEUED, TaskStatus.FAILED, TaskStatus.CANCELED}), + TaskStatus.QUEUED: frozenset({TaskStatus.RUNNING, TaskStatus.FAILED, TaskStatus.CANCELED}), + TaskStatus.RUNNING: frozenset( + {TaskStatus.SUCCEEDED, TaskStatus.FAILED, TaskStatus.CANCELED, TaskStatus.QUEUED} + ), + TaskStatus.SUCCEEDED: frozenset(), + TaskStatus.FAILED: frozenset({TaskStatus.QUEUED}), # retry re-queues + TaskStatus.CANCELED: frozenset(), +} + + +def can_transition(current: TaskStatus, target: TaskStatus) -> bool: + return target in ALLOWED_TRANSITIONS[current] + + +def assert_transition(current: TaskStatus, target: TaskStatus) -> None: + if not can_transition(current, target): + raise InvalidTaskTransitionError(current=current.value, target=target.value) + + +def is_terminal(status: TaskStatus) -> bool: + return status in TERMINAL_STATUSES + + +def needs_refund(status: TaskStatus) -> bool: + return status in REFUNDABLE_STATUSES + + +def progress_for(status: TaskStatus) -> int: + return { + TaskStatus.PENDING: 0, + TaskStatus.QUEUED: 5, + TaskStatus.RUNNING: 50, + TaskStatus.SUCCEEDED: 100, + TaskStatus.FAILED: 100, + TaskStatus.CANCELED: 100, + }[status] diff --git a/packages/shared/README.md b/packages/shared/README.md new file mode 100755 index 0000000..9cc5fb0 --- /dev/null +++ b/packages/shared/README.md @@ -0,0 +1,40 @@ +# `packages/shared/` — 横切基础设施 + +配置、日志、错误码、统一响应、数据库、Redis、可观测性。架构 §16.1。 + +| 模块 | 内容 | +| --- | --- | +| `config.py` | pydantic-settings 类型化配置,全部来自环境变量 | +| `errors.py` | 统一错误码(§9.2:4xxxx 客户端 / 5xxxx 服务端 / 6xxxx 外部依赖) | +| `response.py` | 统一响应信封 `{code, message, data, request_id, hint}` | +| `handlers.py` | FastAPI 异常处理器,保证任何异常都返回信封且不泄露堆栈 | +| `middleware.py` | 请求 ID / 追踪 ID、访问日志、HTTP 指标、`/v1` 路径别名 | +| `logging.py` | structlog JSON 日志 + **脱敏处理器** | +| `context.py` | 请求上下文(request_id / trace_id / user_id / task_id) | +| `db.py` | 异步 SQLAlchemy 引擎与会话 | +| `models.py` | ORM 表定义 | +| `redis_client.py` | Redis 连接 | +| `observability.py` | OTel tracing + Prometheus 指标注册表 | + +## 能放什么 + +跨所有域复用、且不含任何业务判断的东西。 + +## 不能放什么 + +- **任何业务规则**。「配额够不够」属于 `packages/domain/billing`, + 「任务能不能取消」属于 `packages/domain/task`。放进来就没人能纯单测了。 +- 对具体业务域的 import。`shared` 在依赖层级的底部,反向依赖会立刻形成环。 + +## 日志脱敏不是可选项 + +`logging.py` 的脱敏是 structlog **处理器**,作用于每一条日志,没有 opt-in。 +新增敏感字段时加进 `SENSITIVE_KEYS`,或用 `LOG_EXTRA_REDACT_KEYS` 环境变量。 +NFR-SEC-03 / 架构 §11.2:手机号、UnionID、图片地址、诊断结论、令牌一律脱敏。 + +## 新增错误码 + +在 `ErrorCode` 加值,**同时**在 `ERROR_META` 加 message 和 hint。 +测试 `test_every_error_code_has_metadata` 会强制这一点。 +FR-BILL-07:配额相关文案里不能出现价格、购买、会员、升级字样, +`test_quota_hint_never_mentions_price_or_purchase` 会扫描全部文案。 diff --git a/packages/shared/__init__.py b/packages/shared/__init__.py new file mode 100644 index 0000000..ab08ba6 --- /dev/null +++ b/packages/shared/__init__.py @@ -0,0 +1 @@ +"""横切基础设施:配置、日志、错误、响应、DB、Redis、可观测。""" diff --git a/packages/shared/config.py b/packages/shared/config.py new file mode 100755 index 0000000..e20532b --- /dev/null +++ b/packages/shared/config.py @@ -0,0 +1,148 @@ +"""Typed application configuration. + +Every value comes from the environment (架构文档 §11.2: 密钥不进代码/镜像). +`.env` is only a developer convenience; dev/staging/prod inject via SSM/KMS. +""" + +from __future__ import annotations + +import json +from decimal import Decimal +from functools import lru_cache +from typing import Any, Literal + +from pydantic import Field, ValidationInfo, field_validator +from pydantic_settings import BaseSettings, SettingsConfigDict + +Environment = Literal["local", "dev", "staging", "prod"] +CachePrefixMode = Literal["strict", "warn", "off"] + +_PLACEHOLDER_JWT_SECRET = "change-me-local-only-do-not-use-outside-local" # noqa: S105 + + +class Settings(BaseSettings): + """Root settings object. Read it through :func:`get_settings`.""" + + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + extra="ignore", + case_sensitive=False, + ) + + # --- Application ------------------------------------------------------- + app_env: Environment = "local" + app_name: str = "meida-backend" + api_prefix: str = "/api/v1" + debug: bool = False + log_level: str = "INFO" + log_json: bool = True + log_extra_redact_keys: str = "" + + # --- Database ---------------------------------------------------------- + database_url: str = "postgresql+asyncpg://meida:meida@localhost:55432/meida" + db_pool_size: int = 10 + db_max_overflow: int = 20 + db_echo: bool = False + + # --- Redis / Celery ---------------------------------------------------- + redis_url: str = "redis://localhost:56379/0" + celery_broker_url: str = "redis://localhost:56379/1" + celery_result_backend: str = "redis://localhost:56379/2" + + # --- Auth -------------------------------------------------------------- + jwt_secret: str = _PLACEHOLDER_JWT_SECRET + jwt_algorithm: str = "HS256" + jwt_access_ttl_seconds: int = 900 + jwt_issuer: str = "meida" + dev_auth_enabled: bool = True + + # --- LLM routing ------------------------------------------------------- + llm_primary_provider: str = "deepseek" + llm_backup_provider: str = "zhipu" + llm_structured_max_retries: int = 2 + llm_circuit_failure_threshold: int = 3 + llm_circuit_cooldown_seconds: float = 30.0 + llm_scene_overrides: dict[str, dict[str, Any]] = Field( + default_factory=dict, alias="LLM_SCENE_OVERRIDES_JSON" + ) + + # --- DeepSeek (primary, OpenAI-compatible) ----------------------------- + deepseek_api_key: str = "" + deepseek_base_url: str = "https://api.deepseek.com/v1" + deepseek_model: str = "deepseek-v4-flash" + deepseek_timeout_seconds: float = 30.0 + + # --- 智谱 GLM (backup slot, not implemented in V1) --------------------- + zhipu_api_key: str = "" + zhipu_base_url: str = "https://open.bigmodel.cn/api/paas/v4" + zhipu_model: str = "glm-4-flash" + zhipu_timeout_seconds: float = 30.0 + + # --- Prompt caching discipline (架构文档 §7.2) ------------------------- + llm_cache_prefix_mode: CachePrefixMode = "strict" + prompts_dir: str = "prompts" + + # --- Cost attribution (架构文档 §13) ----------------------------------- + llm_cost_rates: dict[str, dict[str, Decimal]] = Field( + default_factory=dict, alias="LLM_COST_RATES_JSON" + ) + llm_default_cost_rate: dict[str, Decimal] = Field( + default_factory=lambda: { + "input_cents_per_mtok": Decimal("100"), + "cached_input_cents_per_mtok": Decimal("20"), + "output_cents_per_mtok": Decimal("200"), + }, + alias="LLM_DEFAULT_COST_RATE_JSON", + ) + + # --- Global daily cost circuit breaker --------------------------------- + llm_daily_budget_enabled: bool = True + llm_daily_budget_cents: Decimal = Decimal("100000") + llm_budget_timezone_offset_hours: int = 8 + + # --- Observability ----------------------------------------------------- + otel_enabled: bool = False + otel_service_name: str = "meida-api" + otel_exporter_otlp_endpoint: str = "" + prometheus_enabled: bool = True + + @field_validator( + "llm_scene_overrides", "llm_cost_rates", "llm_default_cost_rate", mode="before" + ) + @classmethod + def _parse_json_object(cls, value: Any) -> Any: + if isinstance(value, str): + stripped = value.strip() + if not stripped: + return {} + return json.loads(stripped) + return value + + @field_validator("jwt_secret") + @classmethod + def _reject_placeholder_secret_outside_local(cls, value: str, info: ValidationInfo) -> str: + env = info.data.get("app_env", "local") + if env in ("staging", "prod") and (value == _PLACEHOLDER_JWT_SECRET or len(value) < 32): + msg = "JWT_SECRET must be set to a >=32 char random value outside local/dev" + raise ValueError(msg) + return value + + @property + def is_local(self) -> bool: + return self.app_env in ("local", "dev") + + @property + def extra_redact_keys(self) -> tuple[str, ...]: + return tuple(k.strip() for k in self.log_extra_redact_keys.split(",") if k.strip()) + + @property + def sync_database_url(self) -> str: + """psycopg-style URL, used by tooling that cannot speak asyncpg.""" + return self.database_url.replace("+asyncpg", "") + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + """Process-wide settings singleton.""" + return Settings() diff --git a/packages/shared/context.py b/packages/shared/context.py new file mode 100755 index 0000000..5bd8623 --- /dev/null +++ b/packages/shared/context.py @@ -0,0 +1,75 @@ +"""Request-scoped context propagated into logs and traces (架构文档 §15).""" + +from __future__ import annotations + +import uuid +from contextvars import ContextVar, Token +from dataclasses import dataclass + +_request_id: ContextVar[str | None] = ContextVar("meida_request_id", default=None) +_trace_id: ContextVar[str | None] = ContextVar("meida_trace_id", default=None) +_user_id: ContextVar[int | None] = ContextVar("meida_user_id", default=None) +_task_id: ContextVar[int | None] = ContextVar("meida_task_id", default=None) + + +@dataclass(slots=True) +class ContextTokens: + request_id: Token[str | None] + trace_id: Token[str | None] + user_id: Token[int | None] + task_id: Token[int | None] + + +def new_request_id() -> str: + return uuid.uuid4().hex + + +def bind_context( + *, + request_id: str | None = None, + trace_id: str | None = None, + user_id: int | None = None, + task_id: int | None = None, +) -> ContextTokens: + return ContextTokens( + request_id=_request_id.set(request_id if request_id is not None else _request_id.get()), + trace_id=_trace_id.set(trace_id if trace_id is not None else _trace_id.get()), + user_id=_user_id.set(user_id if user_id is not None else _user_id.get()), + task_id=_task_id.set(task_id if task_id is not None else _task_id.get()), + ) + + +def reset_context(tokens: ContextTokens) -> None: + _request_id.reset(tokens.request_id) + _trace_id.reset(tokens.trace_id) + _user_id.reset(tokens.user_id) + _task_id.reset(tokens.task_id) + + +def get_request_id() -> str | None: + return _request_id.get() + + +def get_trace_id() -> str | None: + return _trace_id.get() + + +def get_user_id() -> int | None: + return _user_id.get() + + +def get_task_id() -> int | None: + return _task_id.get() + + +def context_snapshot() -> dict[str, str | int]: + snapshot: dict[str, str | int] = {} + for key, value in ( + ("request_id", _request_id.get()), + ("trace_id", _trace_id.get()), + ("user_id", _user_id.get()), + ("task_id", _task_id.get()), + ): + if value is not None: + snapshot[key] = value + return snapshot diff --git a/packages/shared/db.py b/packages/shared/db.py new file mode 100755 index 0000000..babf7e8 --- /dev/null +++ b/packages/shared/db.py @@ -0,0 +1,73 @@ +"""Async SQLAlchemy engine and session management (架构文档 §5, §12.2). + +The engine is created lazily so that importing this module never opens a +connection — important for unit tests and for Celery workers that fork. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from sqlalchemy.ext.asyncio import ( + AsyncEngine, + AsyncSession, + async_sessionmaker, + create_async_engine, +) + +from packages.shared.config import Settings, get_settings + +_engine: AsyncEngine | None = None +_session_factory: async_sessionmaker[AsyncSession] | None = None + + +def create_engine(settings: Settings | None = None) -> AsyncEngine: + cfg = settings or get_settings() + return create_async_engine( + cfg.database_url, + echo=cfg.db_echo, + pool_size=cfg.db_pool_size, + max_overflow=cfg.db_max_overflow, + pool_pre_ping=True, + pool_recycle=1800, + ) + + +def get_engine() -> AsyncEngine: + global _engine + if _engine is None: + _engine = create_engine() + return _engine + + +def get_session_factory() -> async_sessionmaker[AsyncSession]: + global _session_factory + if _session_factory is None: + _session_factory = async_sessionmaker( + bind=get_engine(), + expire_on_commit=False, + autoflush=False, + ) + return _session_factory + + +@asynccontextmanager +async def session_scope() -> AsyncIterator[AsyncSession]: + """Transactional scope: commit on success, rollback on error.""" + factory = get_session_factory() + async with factory() as session: + try: + yield session + await session.commit() + except Exception: + await session.rollback() + raise + + +async def dispose_engine() -> None: + global _engine, _session_factory + if _engine is not None: + await _engine.dispose() + _engine = None + _session_factory = None diff --git a/packages/shared/errors.py b/packages/shared/errors.py new file mode 100755 index 0000000..608567a --- /dev/null +++ b/packages/shared/errors.py @@ -0,0 +1,138 @@ +"""Unified error codes (架构文档 §9.2, FR-SYS-02). + +Segments: + ``4xxxx`` client error, ``5xxxx`` server error, ``6xxxx`` external dependency. + +Splitting 5xxxx from 6xxxx lets monitoring separate "our bug" from "vendor +outage" without parsing messages. + +Every code carries a user-readable ``message`` and an optional ``hint`` telling +the user what to do next — required by FR-SYS-02 and NFR-UX-02. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import IntEnum +from typing import Any + + +class ErrorCode(IntEnum): + OK = 0 + + # --- 4xxxx client ----------------------------------------------------- + BAD_REQUEST = 40000 + VALIDATION_FAILED = 40001 + IDEMPOTENCY_CONFLICT = 40002 + UNAUTHENTICATED = 40100 + TOKEN_EXPIRED = 40101 + TOKEN_INVALID = 40102 + FORBIDDEN = 40300 + QUOTA_EXHAUSTED = 40301 + CONSENT_REQUIRED = 40302 + MINOR_RESTRICTED = 40303 + NOT_FOUND = 40400 + STATE_CONFLICT = 40900 + RATE_LIMITED = 42900 + + # --- 5xxxx server ----------------------------------------------------- + INTERNAL_ERROR = 50000 + DB_ERROR = 50001 + TASK_STATE_INVALID = 50002 + CONFIG_ERROR = 50003 + NOT_IMPLEMENTED = 50100 + + # --- 6xxxx external dependency --------------------------------------- + UPSTREAM_ERROR = 60000 + LLM_ALL_PROVIDERS_FAILED = 60001 + LLM_STRUCTURED_OUTPUT_FAILED = 60002 + LLM_DAILY_BUDGET_EXCEEDED = 60003 + LLM_PROMPT_CACHE_VIOLATION = 60004 + MODERATION_UNAVAILABLE = 60005 + + +@dataclass(frozen=True, slots=True) +class ErrorMeta: + http_status: int + message: str + hint: str | None = None + + +ERROR_META: dict[ErrorCode, ErrorMeta] = { + ErrorCode.OK: ErrorMeta(200, "ok"), + ErrorCode.BAD_REQUEST: ErrorMeta(400, "请求参数有误", "请检查后重试"), + ErrorCode.VALIDATION_FAILED: ErrorMeta(422, "请求参数校验未通过", "请按提示修改后重试"), + ErrorCode.IDEMPOTENCY_CONFLICT: ErrorMeta( + 409, "相同请求正在处理中", "请稍候查看任务中心,不要重复提交" + ), + ErrorCode.UNAUTHENTICATED: ErrorMeta(401, "请先登录", "登录后即可继续"), + ErrorCode.TOKEN_EXPIRED: ErrorMeta(401, "登录状态已过期", "请重新登录"), + ErrorCode.TOKEN_INVALID: ErrorMeta(401, "登录状态无效", "请重新登录"), + ErrorCode.FORBIDDEN: ErrorMeta(403, "没有访问权限"), + ErrorCode.QUOTA_EXHAUSTED: ErrorMeta(403, "今日额度已用完", "明天恢复,或完成任务多换几次"), + ErrorCode.CONSENT_REQUIRED: ErrorMeta(403, "需要先完成授权", "请阅读并同意后继续"), + ErrorCode.MINOR_RESTRICTED: ErrorMeta(403, "该功能未向未成年人开放"), + ErrorCode.NOT_FOUND: ErrorMeta(404, "内容不存在或已删除"), + ErrorCode.STATE_CONFLICT: ErrorMeta(409, "当前状态无法执行该操作"), + ErrorCode.RATE_LIMITED: ErrorMeta(429, "操作太频繁了", "请稍后再试"), + ErrorCode.INTERNAL_ERROR: ErrorMeta(500, "服务开小差了", "请稍后重试"), + ErrorCode.DB_ERROR: ErrorMeta(500, "服务开小差了", "请稍后重试"), + ErrorCode.TASK_STATE_INVALID: ErrorMeta(500, "任务状态异常", "请到任务中心重试"), + ErrorCode.CONFIG_ERROR: ErrorMeta(500, "服务配置异常", "请联系客服"), + ErrorCode.NOT_IMPLEMENTED: ErrorMeta(501, "该能力尚未开放"), + ErrorCode.UPSTREAM_ERROR: ErrorMeta(502, "外部服务暂时不可用", "请稍后重试"), + ErrorCode.LLM_ALL_PROVIDERS_FAILED: ErrorMeta( + 502, "AI 服务暂时不可用", "请稍后重试,额度不会被扣除" + ), + ErrorCode.LLM_STRUCTURED_OUTPUT_FAILED: ErrorMeta( + 502, "AI 返回结果格式异常", "请稍后重试,额度不会被扣除" + ), + ErrorCode.LLM_DAILY_BUDGET_EXCEEDED: ErrorMeta( + 503, "AI 能力今日已达服务上限", "请明天再试,核心功能不受影响" + ), + ErrorCode.LLM_PROMPT_CACHE_VIOLATION: ErrorMeta(500, "服务配置异常", "请联系客服"), + ErrorCode.MODERATION_UNAVAILABLE: ErrorMeta( + 503, "内容安全服务不可用", "为保证安全本次请求已中止,请稍后重试" + ), +} + + +class AppError(Exception): + """Base application error carrying a unified error code. + + Never let a raw traceback reach the client (FR-SYS-02). + """ + + def __init__( + self, + code: ErrorCode, + *, + message: str | None = None, + hint: str | None = None, + data: dict[str, Any] | None = None, + ) -> None: + meta = ERROR_META.get(code, ERROR_META[ErrorCode.INTERNAL_ERROR]) + self.code = code + self.message = message or meta.message + self.hint = hint if hint is not None else meta.hint + self.http_status = meta.http_status + self.data = data or {} + super().__init__(f"[{int(code)}] {self.message}") + + +class QuotaExhaustedError(AppError): + def __init__(self, *, required: int, remaining: int, reset_at: str | None = None) -> None: + super().__init__( + ErrorCode.QUOTA_EXHAUSTED, + data={"required": required, "remaining": remaining, "reset_at": reset_at}, + ) + + +class NotFoundError(AppError): + def __init__(self, message: str | None = None) -> None: + super().__init__(ErrorCode.NOT_FOUND, message=message) + + +class ForbiddenError(AppError): + def __init__(self, message: str | None = None) -> None: + super().__init__(ErrorCode.FORBIDDEN, message=message) diff --git a/packages/shared/handlers.py b/packages/shared/handlers.py new file mode 100755 index 0000000..fecf74b --- /dev/null +++ b/packages/shared/handlers.py @@ -0,0 +1,69 @@ +"""FastAPI exception handlers producing the unified envelope (架构文档 §9.2). + +No handler is allowed to leak a stack trace to the client (FR-SYS-02). +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import FastAPI, Request +from fastapi.exceptions import RequestValidationError +from fastapi.responses import JSONResponse +from starlette.exceptions import HTTPException as StarletteHTTPException + +from packages.shared.errors import ERROR_META, AppError, ErrorCode +from packages.shared.logging import get_logger +from packages.shared.response import fail + +logger = get_logger(__name__) + +_HTTP_STATUS_TO_CODE: dict[int, ErrorCode] = { + 400: ErrorCode.BAD_REQUEST, + 401: ErrorCode.UNAUTHENTICATED, + 403: ErrorCode.FORBIDDEN, + 404: ErrorCode.NOT_FOUND, + 405: ErrorCode.BAD_REQUEST, + 409: ErrorCode.STATE_CONFLICT, + 422: ErrorCode.VALIDATION_FAILED, + 429: ErrorCode.RATE_LIMITED, + 501: ErrorCode.NOT_IMPLEMENTED, + 502: ErrorCode.UPSTREAM_ERROR, + 503: ErrorCode.UPSTREAM_ERROR, +} + + +def _json(status: int, body: dict[str, Any]) -> JSONResponse: + return JSONResponse(status_code=status, content=body) + + +def install_exception_handlers(app: FastAPI) -> None: + @app.exception_handler(AppError) + async def _app_error(_request: Request, exc: AppError) -> JSONResponse: + logger.warning("app_error", code=int(exc.code), error_message=exc.message) + return _json( + exc.http_status, + fail(exc.code, message=exc.message, hint=exc.hint, data=exc.data or None), + ) + + @app.exception_handler(RequestValidationError) + async def _validation_error(_request: Request, exc: RequestValidationError) -> JSONResponse: + details = [ + {"field": ".".join(str(p) for p in err.get("loc", ())), "reason": err.get("msg", "")} + for err in exc.errors() + ] + return _json( + 422, + fail(ErrorCode.VALIDATION_FAILED, data={"errors": details}), + ) + + @app.exception_handler(StarletteHTTPException) + async def _http_error(_request: Request, exc: StarletteHTTPException) -> JSONResponse: + code = _HTTP_STATUS_TO_CODE.get(exc.status_code, ErrorCode.INTERNAL_ERROR) + message = exc.detail if isinstance(exc.detail, str) else ERROR_META[code].message + return _json(exc.status_code, fail(code, message=message)) + + @app.exception_handler(Exception) + async def _unhandled(_request: Request, exc: Exception) -> JSONResponse: + logger.exception("unhandled_error", error_type=type(exc).__name__) + return _json(500, fail(ErrorCode.INTERNAL_ERROR)) diff --git a/packages/shared/logging.py b/packages/shared/logging.py new file mode 100755 index 0000000..ed80e6c --- /dev/null +++ b/packages/shared/logging.py @@ -0,0 +1,165 @@ +"""Structured JSON logging with request/trace propagation and redaction. + +NFR-SEC-03 / 架构文档 §11.2: 手机号、UnionID、图片地址、诊断结论、令牌 +一律脱敏或不记录. Redaction runs as a structlog processor so it applies to +every log line regardless of who wrote it — there is no opt-in to forget. +""" + +from __future__ import annotations + +import logging +import re +import sys +from collections.abc import Iterable, MutableMapping +from typing import Any + +import structlog + +from packages.shared.context import context_snapshot + +REDACTED = "***" + +#: Field names whose values must never be logged in clear text. +SENSITIVE_KEYS: frozenset[str] = frozenset( + { + # credentials + "authorization", + "access_token", + "refresh_token", + "token", + "id_token", + "password", + "secret", + "api_key", + "apikey", + "jwt", + "purchase_token", + "refresh_token_hash", + # personal identifiers + "phone", + "phone_number", + "mobile", + "id_card", + "union_id", + "unionid", + "open_id", + "openid", + "device_fp", + "client_ip", + "nickname", + # face / image material (架构文档 §11.2, CR-02) + "face_image", + "face_image_key", + "face_key", + "avatar_key", + "image_key", + "image_url", + "photo_url", + "cos_key", + "presigned_url", + # diagnosis output is sensitive derived data + "diagnosis", + "dimensions", + "derived_features", + } +) + +_PHONE_RE = re.compile(r"(? str: + value = _SK_RE.sub("sk-" + REDACTED, value) + value = _BEARER_RE.sub("Bearer " + REDACTED, value) + return _PHONE_RE.sub(r"\1****\2", value) + + +def _redact_value(value: Any, sensitive_keys: frozenset[str], depth: int = 0) -> Any: + if depth > 6: + return value + if isinstance(value, str): + return _scrub_text(value) + if isinstance(value, dict): + return { + k: ( + REDACTED + if isinstance(k, str) and k.lower() in sensitive_keys + else _redact_value(v, sensitive_keys, depth + 1) + ) + for k, v in value.items() + } + if isinstance(value, (list, tuple)): + return type(value)(_redact_value(v, sensitive_keys, depth + 1) for v in value) + return value + + +def make_redaction_processor( + extra_keys: Iterable[str] = (), +) -> Any: + sensitive = SENSITIVE_KEYS | {k.lower() for k in extra_keys} + + def processor( + _logger: Any, _name: str, event_dict: MutableMapping[str, Any] + ) -> MutableMapping[str, Any]: + for key in list(event_dict): + if isinstance(key, str) and key.lower() in sensitive: + event_dict[key] = REDACTED + else: + event_dict[key] = _redact_value(event_dict[key], sensitive) + return event_dict + + return processor + + +def _context_processor( + _logger: Any, _name: str, event_dict: MutableMapping[str, Any] +) -> MutableMapping[str, Any]: + for key, value in context_snapshot().items(): + event_dict.setdefault(key, value) + return event_dict + + +def configure_logging( + *, + level: str = "INFO", + json_output: bool = True, + extra_redact_keys: Iterable[str] = (), +) -> None: + """Idempotently configure stdlib logging + structlog.""" + logging.basicConfig( + format="%(message)s", + stream=sys.stdout, + level=getattr(logging, level.upper(), logging.INFO), + force=True, + ) + for noisy in ("uvicorn.access", "httpx", "httpcore"): + logging.getLogger(noisy).setLevel(logging.WARNING) + + renderer: Any = ( + structlog.processors.JSONRenderer(ensure_ascii=False) + if json_output + else structlog.dev.ConsoleRenderer(colors=False) + ) + + structlog.configure( + processors=[ + structlog.contextvars.merge_contextvars, + structlog.processors.add_log_level, + structlog.processors.TimeStamper(fmt="iso", utc=True), + _context_processor, + make_redaction_processor(extra_redact_keys), + structlog.processors.StackInfoRenderer(), + structlog.processors.format_exc_info, + renderer, + ], + wrapper_class=structlog.make_filtering_bound_logger( + getattr(logging, level.upper(), logging.INFO) + ), + logger_factory=structlog.PrintLoggerFactory(file=sys.stdout), + cache_logger_on_first_use=True, + ) + + +def get_logger(name: str | None = None) -> Any: + return structlog.get_logger(name) diff --git a/packages/shared/middleware.py b/packages/shared/middleware.py new file mode 100755 index 0000000..2453885 --- /dev/null +++ b/packages/shared/middleware.py @@ -0,0 +1,74 @@ +"""ASGI middleware: request id, access log, metrics, /v1 path alias.""" + +from __future__ import annotations + +import time +from collections.abc import Awaitable, Callable + +from fastapi import FastAPI, Request, Response +from starlette.middleware.base import BaseHTTPMiddleware + +from packages.shared.context import bind_context, new_request_id, reset_context +from packages.shared.logging import get_logger +from packages.shared.observability import HTTP_LATENCY, HTTP_REQUESTS, current_trace_id + +REQUEST_ID_HEADER = "X-Request-Id" + +logger = get_logger(__name__) + + +class RequestContextMiddleware(BaseHTTPMiddleware): + """Bind request/trace ids, emit the access log, and record HTTP metrics.""" + + async def dispatch( + self, request: Request, call_next: Callable[[Request], Awaitable[Response]] + ) -> Response: + request_id = request.headers.get(REQUEST_ID_HEADER) or new_request_id() + tokens = bind_context(request_id=request_id, trace_id=current_trace_id()) + started = time.perf_counter() + status_code = 500 + try: + response = await call_next(request) + status_code = response.status_code + response.headers[REQUEST_ID_HEADER] = request_id + return response + finally: + elapsed = time.perf_counter() - started + route = _route_template(request) + HTTP_REQUESTS.labels(request.method, route, str(status_code)).inc() + HTTP_LATENCY.labels(request.method, route).observe(elapsed) + logger.info( + "http_request", + method=request.method, + route=route, + status=status_code, + duration_ms=round(elapsed * 1000, 2), + ) + reset_context(tokens) + + +def _route_template(request: Request) -> str: + route = request.scope.get("route") + path = getattr(route, "path", None) + return str(path) if path else request.url.path + + +def install_version_alias(app: FastAPI, api_prefix: str) -> None: + """Accept ``/v1/...`` as an alias of ``/api/v1/...``. + + 架构文档 §9.1 fixes the base path at ``/api/v1``; several internal specs and + smoke scripts refer to ``/v1/...``. Rewriting at the ASGI scope keeps a + single set of route definitions and one OpenAPI document. + """ + legacy_prefix = "/v1" + if not api_prefix.endswith(legacy_prefix): + return + + @app.middleware("http") + async def _rewrite( + request: Request, call_next: Callable[[Request], Awaitable[Response]] + ) -> Response: + path = request.scope.get("path", "") + if path == legacy_prefix or path.startswith(legacy_prefix + "/"): + request.scope["path"] = api_prefix + path[len(legacy_prefix) :] + return await call_next(request) diff --git a/packages/shared/models.py b/packages/shared/models.py new file mode 100755 index 0000000..76defcb --- /dev/null +++ b/packages/shared/models.py @@ -0,0 +1,249 @@ +"""SQLAlchemy 2.0 ORM models. + +Scope of this scaffold: only the tables needed to prove the vertical slice +(架构文档 §6.2 identity / billing / task / 观测与成本). All remaining domains +(pf_ / rp_ / kb_ / ag_ / tr_ / cp_) are intentionally absent — see +packages/*/README.md for what still has to be built. + +Conventions enforced here (架构文档 §6.1): + 1. 表名带域前缀 (`id_` / `bl_` / `tk_` / `ob_`),为将来拆库预留边界。 + 2. 主键统一雪花 ID (bigint),不用自增、不用 UUID。 + 3. 涉及配额的表必须有幂等键与流水。 + 4. JSONB 只用于「结构会演进且不需要索引全部字段」的场景。 +""" + +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import ( + BigInteger, + Boolean, + DateTime, + Index, + Integer, + Numeric, + PrimaryKeyConstraint, + SmallInteger, + String, + Text, + UniqueConstraint, + func, +) +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + + +class Base(DeclarativeBase): + """Declarative base; ``Base.metadata`` is Alembic's ``target_metadata``.""" + + +# --------------------------------------------------------------------------- +# identity 域 +# --------------------------------------------------------------------------- +class User(Base): + __tablename__ = "id_user" + + user_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + status: Mapped[str] = mapped_column(String(16), nullable=False, default="active") + nickname: Mapped[str | None] = mapped_column(String(64)) + avatar_key: Mapped[str | None] = mapped_column(String(255)) + register_source: Mapped[str] = mapped_column(String(32), nullable=False, default="unknown") + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + last_active_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + deleted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + + __table_args__ = (Index("ix_id_user_status", "status"),) + + +class UserIdentity(Base): + """External identity mapping. UNIQUE(provider, union_id) per 架构文档 §6.2.""" + + __tablename__ = "id_user_identity" + + identity_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + provider: Mapped[str] = mapped_column(String(32), nullable=False) + union_id: Mapped[str] = mapped_column(String(128), nullable=False) + open_id: Mapped[str | None] = mapped_column(String(128)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + UniqueConstraint("provider", "union_id", name="uq_id_user_identity_provider_union"), + Index("ix_id_user_identity_user", "user_id"), + ) + + +class Consent(Base): + """CR-01 留证:人脸的单独同意必须独立成行,不得混入总协议。""" + + __tablename__ = "id_consent" + + consent_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + consent_type: Mapped[str] = mapped_column(String(32), nullable=False) + doc_version: Mapped[str] = mapped_column(String(32), nullable=False) + granted_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + client_ip: Mapped[str | None] = mapped_column(String(64)) + ua: Mapped[str | None] = mapped_column(String(512)) + + __table_args__ = (Index("ix_id_consent_user_type", "user_id", "consent_type"),) + + +# --------------------------------------------------------------------------- +# billing 域(V1:配额与计量,无支付) +# --------------------------------------------------------------------------- +class QuotaRule(Base): + """FR-BILL-04 配额规则配置化:改配置不发版。""" + + __tablename__ = "bl_quota_rule" + + quota_key: Mapped[str] = mapped_column(String(64), primary_key=True) + period_type: Mapped[str] = mapped_column(String(16), nullable=False) # day/month/lifetime + base_limit: Mapped[int] = mapped_column(Integer, nullable=False) + task_bonus_max: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + hard_cap: Mapped[int] = mapped_column(Integer, nullable=False) + description: Mapped[str | None] = mapped_column(String(255)) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + +class Quota(Base): + """周期配额计量:按 ``period_key`` 自然重置,无需定时任务清零。""" + + __tablename__ = "bl_quota" + + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + quota_key: Mapped[str] = mapped_column(String(64), nullable=False) + period_key: Mapped[str] = mapped_column(String(32), nullable=False) + used: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + limit_value: Mapped[int] = mapped_column(Integer, nullable=False) + bonus_value: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + PrimaryKeyConstraint("user_id", "quota_key", "period_key", name="pk_bl_quota"), + ) + + +class Ledger(Base): + """流水是真相,余额是缓存 (架构文档 §6.1 原则 4 / §6.2 bl_ledger). + + ``tx_type`` 在 V1 只用 system_grant / consume / refund;V2 增加 + purchase / gift —— 这是 V1→V2 免数据迁移的关键,不要拆表。 + ``status`` 承载 §7.4 的预扣三段式:reserved → committed / refunded。 + """ + + __tablename__ = "bl_ledger" + + tx_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + delta: Mapped[int] = mapped_column(Integer, nullable=False) + tx_type: Mapped[str] = mapped_column(String(24), nullable=False) + status: Mapped[str] = mapped_column(String(16), nullable=False, default="committed") + reason: Mapped[str | None] = mapped_column(String(128)) + ref_type: Mapped[str | None] = mapped_column(String(32)) + ref_id: Mapped[str | None] = mapped_column(String(64)) + quota_key: Mapped[str | None] = mapped_column(String(64)) + period_key: Mapped[str | None] = mapped_column(String(32)) + balance_after: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + idempotency_key: Mapped[str] = mapped_column(String(128), nullable=False) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + UniqueConstraint("idempotency_key", name="uq_bl_ledger_idempotency_key"), + Index("ix_bl_ledger_user_created", "user_id", "created_at"), + ) + + +# --------------------------------------------------------------------------- +# task 域(统一任务中心) +# --------------------------------------------------------------------------- +class Task(Base): + """全系统唯一的长耗时任务视图,前端任务中心直接读它 (架构文档 §6.2)。""" + + __tablename__ = "tk_task" + + task_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + user_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + task_type: Mapped[str] = mapped_column(String(32), nullable=False) + status: Mapped[str] = mapped_column(String(16), nullable=False) + progress: Mapped[int] = mapped_column(SmallInteger, nullable=False, default=0) + stage_text: Mapped[str | None] = mapped_column(String(128)) + idempotency_key: Mapped[str] = mapped_column(String(128), nullable=False) + payload: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False, default=dict) + result: Mapped[dict[str, Any] | None] = mapped_column(JSONB) + error_code: Mapped[int | None] = mapped_column(Integer) + error_message: Mapped[str | None] = mapped_column(String(512)) + retry_count: Mapped[int] = mapped_column(SmallInteger, nullable=False, default=0) + points_reserved: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + reserve_tx_id: Mapped[int | None] = mapped_column(BigInteger) + submitted_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + eta_seconds: Mapped[int | None] = mapped_column(Integer) + + __table_args__ = ( + UniqueConstraint("idempotency_key", name="uq_tk_task_idempotency_key"), + Index("ix_tk_task_user_status", "user_id", "status"), + ) + + +# --------------------------------------------------------------------------- +# 观测与成本 +# --------------------------------------------------------------------------- +class AiCall(Base): + """成本归因的唯一数据源 (架构文档 §13, FR-BILL-08). + + **不允许「先上线后补埋点」** —— V1 不收费,成本是唯一财务变量,且这张表 + 是 V2 定价的唯一数据来源。每一次 provider 调用(含失败与结构化重试)都写 + 一行。生产环境按月分区。 + """ + + __tablename__ = "ob_ai_call" + + call_id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=False) + user_id: Mapped[int | None] = mapped_column(BigInteger) + scene: Mapped[str] = mapped_column(String(48), nullable=False) + provider: Mapped[str] = mapped_column(String(32), nullable=False) + model: Mapped[str] = mapped_column(String(64), nullable=False) + # Prompt 版本:架构文档 §8.1「可复现」要求记录,§6.2 的列清单漏了,这里补上。 + prompt_id: Mapped[str | None] = mapped_column(String(64)) + prompt_version: Mapped[str | None] = mapped_column(String(16)) + tokens_in: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + tokens_out: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + tokens_cached: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + latency_ms: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + cost_cents: Mapped[Decimal] = mapped_column( + Numeric(14, 6), nullable=False, default=Decimal("0") + ) + success: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + error_code: Mapped[str | None] = mapped_column(String(64)) + error_message: Mapped[str | None] = mapped_column(Text) + ref_type: Mapped[str | None] = mapped_column(String(32)) + ref_id: Mapped[str | None] = mapped_column(String(64)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + __table_args__ = ( + Index("ix_ob_ai_call_created", "created_at"), + Index("ix_ob_ai_call_scene_created", "scene", "created_at"), + Index("ix_ob_ai_call_user_created", "user_id", "created_at"), + ) diff --git a/packages/shared/observability.py b/packages/shared/observability.py new file mode 100755 index 0000000..578985a --- /dev/null +++ b/packages/shared/observability.py @@ -0,0 +1,116 @@ +"""OpenTelemetry tracing + Prometheus metrics (架构文档 §15, TD-24). + +The three dashboards required by §15 (业务漏斗 / AI 成本 / 任务健康) are built +from the metrics declared here plus the ``ob_ai_call`` table. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any + +from prometheus_client import CollectorRegistry, Counter, Histogram + +if TYPE_CHECKING: # pragma: no cover + from fastapi import FastAPI + + from packages.shared.config import Settings + +REGISTRY = CollectorRegistry(auto_describe=True) + +HTTP_REQUESTS = Counter( + "meida_http_requests_total", + "HTTP requests handled by the API", + labelnames=("method", "route", "status"), + registry=REGISTRY, +) +HTTP_LATENCY = Histogram( + "meida_http_request_duration_seconds", + "HTTP request latency", + labelnames=("method", "route"), + buckets=(0.01, 0.05, 0.1, 0.2, 0.3, 0.5, 1.0, 2.0, 5.0, 8.0, 15.0), + registry=REGISTRY, +) + +# --- AI 成本看板 --------------------------------------------------------- +LLM_CALLS = Counter( + "meida_llm_calls_total", + "LLM provider calls (one per HTTP attempt, including retries and failures)", + labelnames=("scene", "provider", "model", "success"), + registry=REGISTRY, +) +LLM_TOKENS = Counter( + "meida_llm_tokens_total", + "LLM tokens by kind; `cached` drives the prompt-cache hit-rate panel (NFR-COST-03)", + labelnames=("scene", "provider", "model", "kind"), + registry=REGISTRY, +) +LLM_COST_CENTS = Counter( + "meida_llm_cost_cents_total", + "Accumulated LLM cost in cents, attributable to scene/provider/model", + labelnames=("scene", "provider", "model"), + registry=REGISTRY, +) +LLM_LATENCY = Histogram( + "meida_llm_latency_seconds", + "LLM provider call latency", + labelnames=("scene", "provider", "model"), + buckets=(0.2, 0.5, 1.0, 2.0, 4.0, 8.0, 15.0, 30.0, 60.0), + registry=REGISTRY, +) + +# --- 业务漏斗 / 任务健康 ------------------------------------------------- +QUOTA_EVENTS = Counter( + "meida_quota_events_total", + "Quota three-phase events (reserve/commit/refund) and their outcome", + labelnames=("quota_key", "phase", "result"), + registry=REGISTRY, +) +TASK_TRANSITIONS = Counter( + "meida_task_transitions_total", + "tk_task state machine transitions", + labelnames=("task_type", "status"), + registry=REGISTRY, +) + + +def setup_tracing(app: FastAPI, settings: Settings) -> None: + """Install the OTel tracer provider and instrument FastAPI. + + Disabled by default locally; enable with ``OTEL_ENABLED=true`` and an + OTLP endpoint. + """ + if not settings.otel_enabled: + return + + from opentelemetry import trace + from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter + from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor + from opentelemetry.sdk.resources import Resource + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import BatchSpanProcessor + + resource = Resource.create( + {"service.name": settings.otel_service_name, "deployment.environment": settings.app_env} + ) + provider = TracerProvider(resource=resource) + if settings.otel_exporter_otlp_endpoint: + provider.add_span_processor( + BatchSpanProcessor( + OTLPSpanExporter(endpoint=f"{settings.otel_exporter_otlp_endpoint}/v1/traces") + ) + ) + trace.set_tracer_provider(provider) + FastAPIInstrumentor.instrument_app(app, tracer_provider=provider) + + +def current_trace_id() -> str | None: + """Return the active W3C trace id, if any.""" + try: + from opentelemetry import trace + except ImportError: # pragma: no cover + return None + span = trace.get_current_span() + ctx: Any = span.get_span_context() + if not getattr(ctx, "is_valid", False): + return None + return format(ctx.trace_id, "032x") diff --git a/packages/shared/redis_client.py b/packages/shared/redis_client.py new file mode 100755 index 0000000..76ab303 --- /dev/null +++ b/packages/shared/redis_client.py @@ -0,0 +1,28 @@ +"""Shared Redis connection (cache / quota counters / cost budget counter).""" + +from __future__ import annotations + +from redis.asyncio import Redis, from_url + +from packages.shared.config import Settings, get_settings + +_client: Redis | None = None + + +def create_redis(settings: Settings | None = None) -> Redis: + cfg = settings or get_settings() + return from_url(cfg.redis_url, encoding="utf-8", decode_responses=True) + + +def get_redis() -> Redis: + global _client + if _client is None: + _client = create_redis() + return _client + + +async def close_redis() -> None: + global _client + if _client is not None: + await _client.aclose() + _client = None diff --git a/packages/shared/response.py b/packages/shared/response.py new file mode 100755 index 0000000..34a194a --- /dev/null +++ b/packages/shared/response.py @@ -0,0 +1,49 @@ +"""Unified response envelope (架构文档 §9.2). + + {"code": 0, "message": "ok", "data": {...}, "request_id": "..."} + {"code": 40301, "message": "今日额度已用完", "data": {...}, + "request_id": "...", "hint": "明天恢复,或完成任务多换 3 次"} + +Every endpoint returns this shape. Handlers in :mod:`packages.shared.handlers` +guarantee it even for unhandled exceptions. +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, Field + +from packages.shared.context import get_request_id +from packages.shared.errors import ERROR_META, ErrorCode + + +class Envelope(BaseModel): + code: int = Field(default=int(ErrorCode.OK), description="0 表示成功;4/5/6 开头为错误段") + message: str = Field(default="ok", description="用户可读文案") + data: Any = Field(default=None) + request_id: str | None = Field(default=None) + hint: str | None = Field(default=None, description="告诉用户下一步怎么办") + + +def ok(data: Any = None) -> dict[str, Any]: + return Envelope( + code=int(ErrorCode.OK), message="ok", data=data, request_id=get_request_id() + ).model_dump(exclude_none=True) + + +def fail( + code: ErrorCode, + *, + message: str | None = None, + hint: str | None = None, + data: Any = None, +) -> dict[str, Any]: + meta = ERROR_META.get(code, ERROR_META[ErrorCode.INTERNAL_ERROR]) + return Envelope( + code=int(code), + message=message or meta.message, + data=data, + request_id=get_request_id(), + hint=hint if hint is not None else meta.hint, + ).model_dump(exclude_none=True) diff --git a/packages/shared/security.py b/packages/shared/security.py new file mode 100755 index 0000000..8d29285 --- /dev/null +++ b/packages/shared/security.py @@ -0,0 +1,76 @@ +"""JWT access tokens (架构文档 §9.1: Bearer JWT, 15 分钟). + +The revocable refresh-token half (`id_session`) is FR-ACC-05 and is deliberately +NOT implemented in this scaffold — see packages/identity/README.md. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Any + +import jwt + +from packages.shared.errors import AppError, ErrorCode + + +@dataclass(frozen=True, slots=True) +class AccessTokenClaims: + user_id: int + issued_at: int + expires_at: int + issuer: str + + +def create_access_token( + *, + user_id: int, + secret: str, + algorithm: str = "HS256", + ttl_seconds: int = 900, + issuer: str = "meida", + now: int | None = None, +) -> str: + issued_at = int(now if now is not None else time.time()) + payload: dict[str, Any] = { + "sub": str(user_id), + "iat": issued_at, + "exp": issued_at + ttl_seconds, + "iss": issuer, + "typ": "access", + } + return jwt.encode(payload, secret, algorithm=algorithm) + + +def decode_access_token( + token: str, + *, + secret: str, + algorithm: str = "HS256", + issuer: str = "meida", +) -> AccessTokenClaims: + try: + payload = jwt.decode( + token, + secret, + algorithms=[algorithm], + issuer=issuer, + options={"require": ["exp", "iat", "sub"]}, + ) + except jwt.ExpiredSignatureError as exc: + raise AppError(ErrorCode.TOKEN_EXPIRED) from exc + except jwt.InvalidTokenError as exc: + raise AppError(ErrorCode.TOKEN_INVALID) from exc + + try: + user_id = int(payload["sub"]) + except (KeyError, TypeError, ValueError) as exc: + raise AppError(ErrorCode.TOKEN_INVALID) from exc + + return AccessTokenClaims( + user_id=user_id, + issued_at=int(payload["iat"]), + expires_at=int(payload["exp"]), + issuer=str(payload.get("iss", issuer)), + )