Metadata-Version: 2.4
Name: huace-aigc-auth-client
Version: 1.1.55
Summary: 华策AIGC Auth Client - 提供 Token 验证、用户信息获取、权限检查、旧系统接入等功能
Author-email: Huace <support@huace.com>
License: MIT
Project-URL: Homepage, https://github.com/huace/huace-aigc-auth-client
Project-URL: Repository, https://github.com/huace/huace-aigc-auth-client
Keywords: aigc,auth,huace,sdk,authentication
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.7
Description-Content-Type: text/markdown
Requires-Dist: requests>=2.20.0

# AIGC Auth Python SDK

[![PyPI version](https://badge.fury.io/py/huace-aigc-auth-client.svg)](https://pypi.org/project/huace-aigc-auth-client/)
[![Python Version](https://img.shields.io/pypi/pyversions/huace-aigc-auth-client.svg)](https://pypi.org/project/huace-aigc-auth-client/)

Python 后端服务接入华策 AIGC 鉴权中心的 SDK 工具包，提供 9 大接入能力。

## 安装

```bash
pip install huace-aigc-auth-client
```

### 环境变量（必填）

```bash
AIGC_AUTH_APP_ID=your_app_id
AIGC_AUTH_APP_SECRET=your_app_secret
AIGC_AUTH_BASE_URL=https://your-auth-api-url
```

> 在鉴权中心创建应用后获取 APP_ID 和 APP_SECRET。

### Nginx 代理（可选）

```nginx
location /aigc-auth/ {
    proxy_pass https://aigc-auth.huacemedia.com/aigc-auth/;
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
}
```

---

## SKILL 1: 鉴权中间件

> 为你的 FastAPI / Flask 应用加上统一鉴权拦截，未登录请求自动 401。

### FastAPI — 中间件模式（推荐）

```python
from fastapi import FastAPI, Request
from huace_aigc_auth_client import AigcAuthClient, AuthMiddleware

app = FastAPI()
client = AigcAuthClient()
auth_mw = AuthMiddleware(client, exclude_paths=["/health", "/docs", "/openapi.json"])

@app.middleware("http")
async def auth(request: Request, call_next):
    return await auth_mw.fastapi_middleware(request, call_next)

@app.get("/me")
async def me(request: Request):
    user = request.state.user_info
    return {"username": user.username, "roles": user.roles}
```

**排除路径配置：**

| 参数 | 说明 | 示例 |
|------|------|------|
| `exclude_paths` | 精确匹配，跳过鉴权 | `["/health", "/docs"]` |
| `exclude_prefixes` | 前缀匹配，跳过鉴权 | `["/public/", "/static/"]` |

### FastAPI — 依赖注入

```python
from fastapi import FastAPI, Depends
from huace_aigc_auth_client import AigcAuthClient, create_fastapi_auth_dependency, UserInfo

app = FastAPI()
client = AigcAuthClient()
get_user = create_fastapi_auth_dependency(client)

@app.get("/me")
async def me(user: UserInfo = Depends(get_user)):
    return {"username": user.username}

# 带权限校验
def require_perm(perm: str):
    async def check(user: UserInfo = Depends(get_user)):
        if not user.has_permission(perm):
            from fastapi import HTTPException
            raise HTTPException(403, "权限不足")
        return user
    return check

@app.get("/admin")
async def admin(user: UserInfo = Depends(require_perm("admin:access"))):
    return {"admin": True}
```

### FastAPI — 装饰器

```python
from huace_aigc_auth_client import AigcAuthClient, require_auth, UserInfo

client = AigcAuthClient()

@app.get("/protected")
@require_auth(client)
async def route(request: Request, user_info: UserInfo):
    return {"user": user_info.username}

@app.get("/editor")
@require_auth(client, permissions=["article:write", "article:edit"], any_permission=True)
async def editor(request: Request, user_info: UserInfo):
    return {"editor": True}
```

### Flask

```python
from flask import Flask, g, jsonify
from huace_aigc_auth_client import AigcAuthClient, AuthMiddleware

app = Flask(__name__)
client = AigcAuthClient()
auth_mw = AuthMiddleware(client, exclude_paths=["/health", "/login"])

@app.before_request
def before():
    return auth_mw.flask_before_request()

@app.route("/me")
def me():
    user = g.user_info
    return jsonify({"username": user.username, "roles": user.roles})
```

---

## SKILL 2: Token 验证与用户信息

> 主动验证 Token 或 Access Key，获取用户详情和权限。

### 验证 Token

```python
from huace_aigc_auth_client import AigcAuthClient

client = AigcAuthClient()
result = client.verify_token("user-token")

if result.valid:
    print(result.user_id, result.username, result.expires_at)
```

### 获取用户信息（含角色、权限、用户组）

```python
user = client.get_user_info(token)

print(user.username, user.nickname, user.email)
print(user.roles)         # ["admin", "editor"]
print(user.permissions)   # ["user:read", "user:write"]
print(user.groups)        # [{"id": 1, "name": "技术部", "code": "tech"}]

if user.is_admin or user.has_role("admin"):
    print("管理员")

if user.has_permission("user:write"):
    print("有写权限")

if user.has_any_permission(["user:read", "user:write"]):
    print("至少一个权限")
```

### 验证 Access Key

```python
result = client.verify_access_key("ak-xxxxx")
if result.valid:
    print(result.key_type, result.username, result.is_admin)
    # key_type: "app" / "global" / "user"
```

### UserInfo 字段

| 属性 | 类型 | 说明 |
|------|------|------|
| `id` | int | 用户 ID |
| `username` | str | 用户名 |
| `nickname` | str | 昵称 |
| `email` | str | 邮箱 |
| `phone` | str | 手机号 |
| `avatar` | str | 头像 URL |
| `roles` | List[str] | 角色代码列表 |
| `permissions` | List[str] | 权限代码列表 |
| `groups` | List[Dict] | 用户组 `[{"id", "name", "code"}]` |
| `department` | str | 部门 |
| `company` | str | 公司 |
| `is_admin` | bool | 是否管理员 |
| `status` | str | 状态 |

| 方法 | 说明 |
|------|------|
| `has_role(role)` | 检查角色 |
| `has_permission(perm)` | 检查权限 |
| `has_any_permission(perms)` | 拥有任意一个 |
| `has_all_permissions(perms)` | 拥有全部 |

---

## SKILL 3: 用户查询与用户组

> 批量查询用户、分页搜索、获取用户组清单。

### 用户列表（分页 + 搜索）

```python
# 第一页
result = client.list_users()
print(result["total"], result["items"])

# 搜索
result = client.list_users(keyword="张", page=1, page_size=10)
```

### 批量查询

```python
# 按用户名
users = client.batch_get_users(["zhangsan", "lisi"])

# 按用户 ID
users = client.batch_get_users_by_ids([1, 2, 999])
```

### 获取用户组清单

```python
groups = client.list_groups()
# [{"id": 1, "name": "技术部", "code": "tech", "description": "..."}, ...]
```

---

## SKILL 4: 服务注册与配置拉取

> 服务启动时注册 IP + 端口，定时心跳上报，从鉴权中心拉取全局配置。

### 方式一：环境变量自动注册（推荐）

```bash
# .env — 设置 AIGC_SERVICE_NAME 即启用自动注册
AIGC_SERVICE_NAME=my-service
AIGC_SERVICE_TYPE=website        # website / plugin / mcp / other
AIGC_SERVICE_ENV=pro             # test / pre / pro
AIGC_SERVICE_PORT=8000
AIGC_SERVICE_TAGS=core,api
AIGC_SERVICE_DESCRIPTION=用户中心主服务
```

```python
# 一行搞定：注册 + 心跳 + 配置 Pull
client = AigcAuthClient()
```

### 方式二：显式调用

```python
client = AigcAuthClient(auto_register_service=False)

result = client.start_service(
    name="my-service",
    service_type="website",
    environment="pro",
    port=8000,
    tags=["core", "api"],
    description="用户中心主服务",
    heartbeat_interval=60,
    on_config_change=lambda cfg, ver: print(f"配置更新 v{ver}:", cfg),
)
print(f"service_id={result.service_id}, instance_id={result.instance_id}")

# 退出时注销
client.stop_service("my-service", "pro")
```

### 读取配置

```python
# 从心跳缓存读取，无需额外请求
# 不传参时自动从环境变量 AIGC_SERVICE_NAME / AIGC_SERVICE_ENV 匹配
config = client.get_service_config()
db_url = config.get("DATABASE_URL")

# 多服务场景显式指定
config = client.get_service_config("my-service", "pro")
```

### 服务注册环境变量

| 变量 | 说明 | 默认值 |
|------|------|--------|
| `AIGC_SERVICE_NAME` | 服务名（设置即启用自动注册） | — |
| `AIGC_SERVICE_TYPE` | 类型：website / plugin / mcp / other | `other` |
| `AIGC_SERVICE_ENV` | 环境：test / pre / pro | `pro` |
| `AIGC_SERVICE_PORT` | 对外端口 | — |
| `AIGC_SERVICE_IPS` | IP 列表（逗号分隔） | 自动检测 |
| `AIGC_SERVICE_TAGS` | 标签（逗号分隔） | — |
| `AIGC_SERVICE_DESCRIPTION` | 描述 | — |
| `AIGC_SERVICE_INSTANCE_ID` | 实例 ID | `hostname:port` |
| `AIGC_SERVICE_HEARTBEAT_INTERVAL` | 心跳间隔（秒） | `60` |

### Docker / 容器注意

容器内自动检测的 IP 可能是内网地址。两种应对：

1. 环境变量注入宿主机 IP：`AIGC_SERVICE_IPS=203.0.113.10`
2. 显式传入：`start_service(ips=["203.0.113.10"], ...)`

容器/k8s 建议设置 `AIGC_SERVICE_INSTANCE_ID`（如 Pod 名 + 端口），保证跨重启稳定。

### 服务唯一性模型

| 层级 | 唯一键 | 谁改什么 |
|------|--------|----------|
| Service（逻辑服务） | `app_id + name + environment` | 平台：昵称、描述、类型、标签、配置、状态 |
| ServiceInstance（进程实例） | `service_id + instance_id` | SDK：IP、端口、心跳、metadata |

注册是幂等的：服务已存在时不覆盖平台已编辑的字段，仅更新本实例记录。

---

## SKILL 5: Webhook 接收

> 接收鉴权中心推送的用户变更事件（创建、更新、删除）。

### FastAPI

```python
from fastapi import APIRouter
from huace_aigc_auth_client import register_webhook_router

router = APIRouter()

async def handle_webhook(data: dict) -> dict:
    event = data.get("event")
    print(f"收到事件: {event}")
    return {"status": "ok"}

register_webhook_router(
    router,
    handler=handle_webhook,
    prefix="/webhook",
    secret_env_key="AIGC_AUTH_WEBHOOK_SECRET",
)
# 端点: POST /webhook/auth
```

### Flask

```python
from huace_aigc_auth_client import register_flask_webhook_routes

def handle_webhook(data: dict) -> dict:
    return {"status": "ok"}

register_flask_webhook_routes(app, handler=handle_webhook)
# 端点: POST /api/webhook/auth
```

### 事件类型

| 事件 | 说明 |
|------|------|
| `user.created` | 用户创建 |
| `user.updated` | 用户更新 |
| `user.deleted` | 用户删除 |
| `user.login` | 用户登录 |
| `user.init_sync_auth` | 初始化同步请求 |

### 签名验证

SDK 自动处理：从请求头 `X-Webhook-Signature` 读取签名，用环境变量 `AIGC_AUTH_WEBHOOK_SECRET` 的密钥做 HMAC-SHA256 验证。签名不匹配返回 401。

也可手动验证：

```python
from huace_aigc_auth_client import verify_webhook_signature

is_valid = verify_webhook_signature(payload_bytes, signature_header, secret)
```

---

## SKILL 6: API 统计与计费

> 自动收集 API 调用数据并上报到鉴权中心，支持动态计费规则。

### 自动采集

使用鉴权中间件（SKILL 1）时，默认启用 API 统计收集，无需额外配置。

### 手动初始化

```python
from huace_aigc_auth_client import init_api_stats_collector

init_api_stats_collector(
    api_url="https://auth.example.com/sdk",
    app_id="your-app-id",
    app_secret="your-app-secret",
    batch_size=10,       # 批量提交大小
    flush_interval=5.0,  # 刷新间隔（秒）
    enabled=True,
)
```

### 手动上报

```python
from huace_aigc_auth_client import collect_api_stat

collect_api_stat(
    api_path="/api/chat/completions",
    api_method="POST",
    status_code=200,
    response_time=1.2,
    token="user-token",
    error_message=None,
    request_params={"headers": {"X-Match-Username": "zhangsan"}},
    response_headers={
        "X-Match-Dynamic": "1",
        "X-Match-Cost": "0.0035",
    },
)
```

### 动态计费响应头

上游 API 通过响应头告诉鉴权中心如何计费：

| 响应头 | 说明 | 示例 |
|--------|------|------|
| `X-Match-Dynamic` | 值为 `1` 时启用动态计费 | `1` |
| `X-Match-Cost` | 单次调用费用（元） | `0.0035` |
| `X-Match-Path` | 规范化路径（可选） | `/api/chat/completions` |
| `X-Match-Rule` | 规则名称（可选） | `GPT-4o 对话` |

规则：`X-Match-Dynamic=1` → 创建动态计费规则；不设或非 `1` → 静态规则（费用为 0）。

### X-Match-Username

异步任务场景下，通过 `request_params.headers` 关联用户：

```python
collect_api_stat(
    api_path="/api/task/run",
    api_method="POST",
    status_code=200,
    response_time=1.5,
    token=token,
    request_params={"headers": {"X-Match-Username": "zhangsan"}},
)
```

> 使用 `auth_request`（SKILL 7）会自动从上下文注入 `X-Match-Username`。

---

## SKILL 7: 认证请求封装

> 发起 HTTP 请求时自动注入认证信息（App ID/Secret、Token、IP、Trace ID）并上报统计。

### 同步请求（requests）

```python
from huace_aigc_auth_client import auth_request

response = auth_request("GET", "https://api.example.com/data")
response = auth_request("POST", "https://api.example.com/users",
                        json={"name": "John"})
```

**快捷方法：**

```python
from huace_aigc_auth_client import auth_request_get, auth_request_post

r = auth_request_get("https://api.example.com/users", params={"page": 1})
r = auth_request_post("https://api.example.com/users", json={"name": "John"})
```

**会话模式（连接池复用）：**

```python
from huace_aigc_auth_client import AuthSession

session = AuthSession()
response = session.get("https://api.example.com/data")
```

### 异步请求（httpx）

> 需安装 httpx：`pip install httpx`

```python
from huace_aigc_auth_client import AsyncAuthClient

async def example():
    async with AsyncAuthClient(timeout=10.0) as c:
        r = await c.post("https://api.example.com/users", json={"name": "John"})
        r = await c.get("https://api.example.com/users", params={"page": 1})
```

**快捷函数：**

```python
from huace_aigc_auth_client import async_auth_httpx_request

r = await async_auth_httpx_request("GET", "https://api.example.com/data")
```

### 自动注入的请求头

| Header | 来源 |
|--------|------|
| `X-App-ID` | 环境变量 / set_request_context |
| `X-App-Secret` | 环境变量 / set_request_context |
| `Authorization` | Bearer {token} |
| `x-real-ip` | 请求上下文 |
| `user-agent` | 请求上下文 |
| `X-Trace-ID` | 请求上下文 |
| `X-Match-Username` | 自动从 user_context 注入 |

---

## SKILL 8: 用户上下文

> 在任意代码位置（Service、Utility）获取当前请求的用户信息，无需层层传参。

> 前提：必须先注册 AuthMiddleware（SKILL 1）。

### 获取当前用户

```python
from huace_aigc_auth_client import get_current_user, get_current_user_id, get_current_username, is_current_user_admin

def some_business_logic():
    user = get_current_user()
    if user:
        print(f"用户: {get_current_username()} (ID: {get_current_user_id()})")

    if is_current_user_admin():
        print("管理员操作")
```

### 请求上下文

```python
from huace_aigc_auth_client import (
    set_request_context, get_client_ip, get_trace_id, get_request_token,
)

set_request_context(
    ip_address="192.168.1.1",
    user_agent="Mozilla/5.0...",
    trace_id="trace-123",
    token="user-token",
)

ip = get_client_ip()
trace = get_trace_id()
token = get_request_token()
```

### 在装饰器中使用

```python
from functools import wraps
from huace_aigc_auth_client import is_current_user_admin

def require_admin(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        if not is_current_user_admin():
            raise HTTPException(403, "需要管理员权限")
        return await func(*args, **kwargs)
    return wrapper
```

### 可用函数

| 函数 | 说明 |
|------|------|
| `get_current_user()` | 获取当前用户（dict） |
| `get_current_user_id()` | 当前用户 ID |
| `get_current_username()` | 当前用户名 |
| `get_current_app_id()` | 当前应用 ID |
| `get_current_app_code()` | 当前应用 code |
| `is_current_user_admin()` | 是否管理员 |
| `get_client_ip()` | 客户端 IP |
| `get_user_agent()` | User Agent |
| `get_trace_id()` | 追踪 ID |
| `get_request_token()` | 请求 Token |
| `get_request_id()` | 请求 ID |

> 线程安全 + 异步安全（`threading.local` + `contextvars.ContextVar`）。

---

## SKILL 9: 旧系统用户同步

> 已有用户表的旧系统低成本接入，无需修改历史代码和表结构。

### 原理

```
┌─────────────────┐     ┌─────────────────┐     ┌─────────────────┐
│   aigc-auth     │────>│   SDK 同步层     │────>│   旧系统         │
│   (鉴权中心)     │     │   (字段映射)     │     │   (用户表)       │
└─────────────────┘     └─────────────────┘     └─────────────────┘
        |                       |                       |
        |   1. 初始化同步        |                       |
        |<──────────────────────────────────────────────|
        |   (批量同步旧用户到 auth)                       |
        |                       |                       |
        |   2. 增量同步          |                       |
        |──────────────────────>|──────────────────────>|
        |   (auth 新用户自动同步到旧系统)                  |
        |                       |                       |
        |   3. Webhook 推送      |                       |
        |──────────────────────────────────────────────>|
        |   (用户变更主动通知)                            |
```

### 字段映射

```python
from huace_aigc_auth_client import FieldMapping, create_sync_config, PasswordMode

field_mappings = [
    FieldMapping(auth_field="username", legacy_field="username", required=True),
    FieldMapping(auth_field="email", legacy_field="email"),
    FieldMapping(auth_field="nickname", legacy_field="nickname"),
    # 状态转换
    FieldMapping(
        auth_field="status", legacy_field="is_active",
        transform_to_legacy=lambda s: s == "active",
        transform_to_auth=lambda a: "active" if a else "disabled",
    ),
    # 角色转换
    FieldMapping(
        auth_field="roles", legacy_field="role",
        transform_to_legacy=lambda r: r[0] if r else "viewer",
        transform_to_auth=lambda r: [r] if r else ["viewer"],
    ),
]

sync_config = create_sync_config(
    field_mappings=field_mappings,
    password_mode=PasswordMode.UNIFIED,
    unified_password="Abc@123456",
    webhook_url="https://your-domain.com/api/v1/webhook/auth",
)
```

### 实现 LegacySystemAdapter

```python
from huace_aigc_auth_client import LegacySystemAdapter, LegacyUserData
from typing import Optional, List, Any, Dict

class MyLegacyAdapter(LegacySystemAdapter):

    def __init__(self, db, sync_config, auth_client=None):
        super().__init__(sync_config, auth_client)
        self.db = db

    async def get_user_by_username_async(self, username: str) -> Optional[LegacyUserData]:
        user = await self.db.execute(select(User).where(User.username == username))
        user = user.scalars().first()
        if not user:
            return None
        return LegacyUserData({"id": user.id, "username": user.username, "email": user.email})

    async def _create_user_async(self, user_data: Dict[str, Any]) -> Optional[Any]:
        user = User(username=user_data["username"], email=user_data.get("email"))
        self.db.add(user)
        await self.db.commit()
        return user.id

    async def _update_user_async(self, username: str, user_data: Dict[str, Any]) -> bool:
        user = await self.db.execute(select(User).where(User.username == username))
        user = user.scalars().first()
        if user:
            for k, v in user_data.items():
                setattr(user, k, v)
            await self.db.commit()
            return True
        return False

    async def _delete_user_async(self, username: str) -> bool:
        user = await self.db.execute(select(User).where(User.username == username))
        user = user.scalars().first()
        if user:
            await self.db.delete(user)
            await self.db.commit()
            return True
        return False

    async def get_all_users_async(self) -> List[LegacyUserData]:
        result = await self.db.execute(select(User))
        return [LegacyUserData({"username": u.username}) for u in result.scalars().all()]
```

### 集成到登录流程

```python
from huace_aigc_auth_client import UserSyncService

adapter = MyLegacyAdapter(db, sync_config, client)
sync_service = UserSyncService(client, adapter)

# 在中间件中同步
@app.middleware("http")
async def auth_and_sync(request, call_next):
    response = await auth_mw.fastapi_middleware(request, call_next)
    if hasattr(request.state, "user_info"):
        await sync_service.sync_on_login_async(request.state.user_info)
    return response
```

### Webhook 接收

使用 SKILL 5 注册 webhook 路由，handler 中调用 adapter：

```python
async def handle_webhook(data: dict) -> dict:
    return await adapter.handle_webhook(data.get("event"), data.get("data", {}))

register_webhook_router(router, handler=handle_webhook)
```

### 初始化同步（一次性脚本）

```python
async def init_sync():
    adapter = MyLegacyAdapter(db, sync_config, client)
    users = await adapter.get_all_users_async()
    for user in users:
        auth_data = adapter.transform_legacy_to_auth(user)
        password, is_hashed = adapter.get_password_for_sync(user)
        if is_hashed:
            auth_data["password_hashed"] = password
        else:
            auth_data["password"] = password
        client.sync_user_to_auth(auth_data)

asyncio.run(init_sync())
```

### 密码处理策略

| 模式 | 说明 | 适用场景 |
|------|------|----------|
| `PasswordMode.UNIFIED` | 统一初始密码 | 新系统接入 |
| `PasswordMode.CUSTOM_MAPPING` | 自定义映射函数 | 需保留原密码 |
| `CUSTOM_MAPPING` + `password_is_hashed=True` | 直接同步已加密密码 | 两系统同加密方式（推荐） |

```python
# 推荐方式：直接同步 bcrypt 哈希
sync_config = create_sync_config(
    password_mode=PasswordMode.CUSTOM_MAPPING,
    password_mapper=lambda u: u.get("hashed_password"),
    password_is_hashed=True,
)
```

---

## 异常处理

```python
from huace_aigc_auth_client import AigcAuthError

try:
    user = client.get_user_info(token)
except AigcAuthError as e:
    print(f"错误码: {e.code}, 信息: {e.message}")
```

| 错误码 | 说明 |
|--------|------|
| 401 | Token 无效或已过期 |
| 403 | 用户被禁用或权限不足 |
| 404 | 用户不存在 |
| -1 | 网络请求失败 |

---

## 环境变量完整参考

### 鉴权（必填）

| 变量 | 说明 |
|------|------|
| `AIGC_AUTH_APP_ID` | 应用 ID |
| `AIGC_AUTH_APP_SECRET` | 应用密钥 |
| `AIGC_AUTH_BASE_URL` | 鉴权服务地址 |

### 服务注册（设置 NAME 即启用）

| 变量 | 说明 | 默认值 |
|------|------|--------|
| `AIGC_SERVICE_NAME` | 服务名 | — |
| `AIGC_SERVICE_TYPE` | website / plugin / mcp / other | `other` |
| `AIGC_SERVICE_ENV` | test / pre / pro | `pro` |
| `AIGC_SERVICE_PORT` | 对外端口 | — |
| `AIGC_SERVICE_IPS` | IP 列表（逗号分隔） | 自动检测 |
| `AIGC_SERVICE_TAGS` | 标签（逗号分隔） | — |
| `AIGC_SERVICE_DESCRIPTION` | 描述 | — |
| `AIGC_SERVICE_INSTANCE_ID` | 实例 ID | `hostname:port` |
| `AIGC_SERVICE_HEARTBEAT_INTERVAL` | 心跳间隔（秒） | `60` |

### 旧系统同步

| 变量 | 说明 |
|------|------|
| `AIGC_AUTH_SYNC_ENABLED` | 是否启用同步 |
| `AIGC_AUTH_SYNC_PASSWORD` | 统一初始密码 |
| `AIGC_AUTH_WEBHOOK_URL` | Webhook 接收地址 |
| `AIGC_AUTH_WEBHOOK_SECRET` | Webhook 签名密钥 |

---

## 导出清单

```python
from huace_aigc_auth_client import (
    # 核心类
    AigcAuthClient,
    AuthMiddleware,
    UserInfo,
    TokenVerifyResult,
    AccessKeyVerifyResult,
    AigcAuthError,
    ServiceRegistrationResult,

    # FastAPI 鉴权
    require_auth,
    create_fastapi_auth_dependency,

    # Webhook 接收
    register_webhook_router,
    verify_webhook_signature,

    # API 统计
    init_api_stats_collector,
    get_api_stats_collector,
    stop_api_stats_collector,
    collect_api_stat,

    # 认证请求封装
    auth_request,
    auth_request_get,
    auth_request_post,
    AuthSession,
    async_auth_httpx_request,
    async_auth_httpx_request_get,
    async_auth_httpx_request_post,
    AsyncAuthClient,

    # 用户上下文
    set_current_user,
    get_current_user,
    get_current_user_id,
    get_current_username,
    get_current_app_id,
    get_current_app_code,
    is_current_user_admin,
    clear_current_user,
    set_request_context,
    add_to_request_context,
    get_request_context,
    get_client_ip,
    get_user_agent,
    get_request_id,
    get_trace_id,
    get_request_token,
    get_request_app_id,
    get_request_app_secret,
    clear_request_context,

    # 旧系统接入
    LegacySystemAdapter,
    LegacyUserData,
    SyncConfig,
    SyncDirection,
    PasswordMode,
    FieldMapping,
    UserSyncService,
    WebhookSender,
    SyncResult,
    create_sync_config,
    create_default_field_mappings,

    # Flask Webhook
    create_flask_webhook_blueprint,
    register_flask_webhook_routes,

    # 工具
    setLogger,
)
```

---

## API 变更日志

### v1.1.54 (2026-06-03)

- SDK Client 拆分为 `client/` 包目录（Mixin 模式），API 不变
- 新增 `list_groups()` 获取用户组清单
- `UserInfo` 新增 `groups` 字段
- 用户信息接口返回 `groups`（含 id/name/code）

### v1.1.52 (2026-06-02)

- 新增服务注册中心（详见 SKILL 4）
- 环境变量自动注册 + 心跳 + 配置 Pull
- `get_service_config()` 从心跳缓存读取配置

### v1.1.50 (2026-06-01)

- 新增 Webhook 接收路由注册（详见 SKILL 5）
- `register_webhook_router()` / `verify_webhook_signature()`

### v1.1.49 (2026-05-25)

- 新增 `list_users()` 分页搜索（详见 SKILL 3）

### v1.1.47 (2026-05-18)

- `collect_api_stat()` 新增 `response_headers` 参数（详见 SKILL 6）
- 动态计费规则 `X-Match-*` 响应头
- `X-Match-Username` 用户标识

### v1.1.42 (2026-04-28)

- 新增 `batch_get_users()` / `batch_get_users_by_ids()`（详见 SKILL 3）

### v1.1.0 (2026-01-13)

- 支持直接同步已加密密码（详见 SKILL 9 密码策略）
- `_request` 支持自定义 headers

---

## 许可证

MIT License
