# 客户端运行日志批量上报 Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** 新增 `POST /api/v1/applog/batch`，把客户端批量上报的运行日志逐条写进独立滚动文件 `logs/app-client.log`，供 Aliyun Logtail 采进独立 SLS logstore。

**Architecture:** 复用两个现成范式——analytics 的「批量入口 + 服务端补 IP」+ `app-server.log` 的「单行 JSON + RotatingFileHandler + Logtail」。核心逻辑在写入层 `app/core/client_log.py`：专用 `client` logger（`propagate=False`，不污染服务日志），每条按「白名单键提顶层 + 其余并入 `data`」封装成一行 JSON，`trace_id` 提到顶层以便跨层检索。端点瘦、fire-and-forget（写失败不 500）。

**Tech Stack:** FastAPI、Pydantic v2、Python `logging.handlers.RotatingFileHandler`、pytest + `TestClient`。

**Spec:** [docs/superpowers/specs/2026-07-19-client-applog-ingest-design.md](../specs/2026-07-19-client-applog-ingest-design.md)

---

## File Structure

| 文件 | 职责 | 动作 |
|---|---|---|
| `app/core/client_log.py` | 专用 logger/handler、滚动配置、白名单+`data` 兜底与信封拼装、`write_records()` | Create |
| `app/schemas/applog.py` | `AppLogBatchIn`（`logs: list[dict]`）/ `AppLogIngestOut` | Create |
| `app/api/v1/applog.py` | 瘦路由：body 上限依赖 → 调 writer → 返回计数 | Create |
| `app/main.py` | 注册 `applog_router` | Modify |
| `tests/test_applog.py` | writer 单测 + 端点集成测试 | Create |

前置约束（写进代码注释）：`RotatingFileHandler` 多进程并发 `doRollover()` 会损坏/丢日志；本方案依赖生产 `--workers 1`（与限流器/SMS 码/SQLite 同一既有假设）。扩 worker 前必须换 `QueueHandler`→单写入者 / 外部 logrotate / 写 stdout 交 journald。

---

## Task 1: 专用落盘 writer `app/core/client_log.py`

**Files:**
- Create: `app/core/client_log.py`
- Test: `tests/test_applog.py`

- [ ] **Step 1: 写失败测试（writer 层）**

创建 `tests/test_applog.py`：

```python
"""客户端运行日志上报:writer 单测 + 端点集成测试。"""
from __future__ import annotations

import json
from pathlib import Path

import pytest

from app.core import client_log


@pytest.fixture()
def client_log_file(tmp_path, monkeypatch):
    """把客户端日志切到临时文件,并重置 writer 单例使其按当时 env 重建。"""
    p = tmp_path / "app-client.log"
    monkeypatch.setenv("CLIENT_LOG_FILE", str(p))
    client_log.reset_client_logger()
    yield p
    client_log.reset_client_logger()


def _read_lines(p: Path) -> list[dict]:
    text = p.read_text(encoding="utf-8").strip()
    return [json.loads(ln) for ln in text.splitlines() if ln]


# ---------------- writer 层 ----------------

def _meta(**kw) -> dict:
    base = {"device_id": "d-1", "user_id": None, "app_ver": None,
            "platform": None, "sent_at": None}
    base.update(kw)
    return base


def test_writer_writes_one_line_per_record(client_log_file):
    recs = [
        {"client_ts": 1737000000000, "level": "info", "msg": "hello"},
        {"client_ts": 1737000000001, "level": "error", "msg": "boom", "tag": "net"},
    ]
    received, dropped = client_log.write_records(
        recs, meta=_meta(user_id=42, app_ver="1.2.3", platform="android"),
        client_ip="1.2.3.4",
    )
    assert (received, dropped) == (2, 0)
    lines = _read_lines(client_log_file)
    assert len(lines) == 2
    assert lines[0]["source"] == "client"
    assert lines[0]["service"] == "app-client"
    assert lines[0]["device_id"] == "d-1"
    assert lines[0]["user_id"] == 42
    assert lines[0]["app_ver"] == "1.2.3"
    assert lines[0]["client_ip"] == "1.2.3.4"
    assert lines[0]["level"] == "INFO"  # 归一化大写
    assert lines[0]["msg"] == "hello"
    assert lines[0]["client_ts"] == 1737000000000
    assert lines[1]["tag"] == "net"


def test_writer_hoists_trace_id_to_top_level(client_log_file):
    client_log.write_records(
        [{"client_ts": 1, "level": "info", "msg": "x", "trace_id": "abc123"}],
        meta=_meta(), client_ip="",
    )
    assert _read_lines(client_log_file)[0]["trace_id"] == "abc123"


def test_writer_sweeps_unknown_keys_into_data(client_log_file):
    client_log.write_records(
        [{"client_ts": 1, "level": "info", "msg": "x",
          "foo": 123, "data": {"bar": "baz"}}],
        meta=_meta(), client_ip="",
    )
    line = _read_lines(client_log_file)[0]
    assert "foo" not in line               # 白名单外不进顶层
    assert line["data"]["foo"] == 123      # 兜底进 data
    assert line["data"]["bar"] == "baz"    # 客户端自带 data 合并进来


def test_writer_truncates_oversize_msg(client_log_file, monkeypatch):
    monkeypatch.setenv("APPLOG_MAX_MSG_BYTES", "10")
    received, dropped = client_log.write_records(
        [{"client_ts": 1, "level": "info", "msg": "x" * 100}],
        meta=_meta(), client_ip="",
    )
    assert (received, dropped) == (1, 0)   # 截断而非丢弃
    line = _read_lines(client_log_file)[0]
    assert line["msg_truncated"] is True
    assert line["msg"].endswith("…[truncated]")


def test_writer_logger_does_not_propagate(client_log_file):
    lg = client_log.get_logger()
    assert lg.name == "client"
    assert lg.propagate is False           # 不冒泡到 root → 不写 app-server.log
```

- [ ] **Step 2: 跑测试确认失败**

Run: `pytest tests/test_applog.py -q`
Expected: FAIL —— `AttributeError: module 'app.core.client_log' has no attribute 'reset_client_logger'`（模块尚不存在）。

- [ ] **Step 3: 实现 `app/core/client_log.py`**

```python
"""客户端运行日志专用落盘 writer(独立于服务端 app-server.log)。

- 独占 logger "client" + 自己的 RotatingFileHandler,propagate=False → 不污染 app-server.log。
- 每条按「白名单键(client_ts/level/trace_id/tag/msg)提顶层 + 其余并入 data」封装,再
  json.dumps 成一行写出(钉死 SLS 索引列;见 spec §5)。formatter 用 %(message)s——行本身
  已是 JSON,不能再过 JsonFormatter 二次编码。
- 滚动 20MB×10(env 可调),与服务日志同机制。
  ⚠️ 依赖 --workers 1:RotatingFileHandler 多进程并发 doRollover 会损坏/丢日志;扩 worker
  前换 QueueHandler→单写入者 / 外部 logrotate(copytruncate) / 写 stdout 交 journald。

服务端补的字段(time/source/service/client_ip/device_id/...)是「事实」,与客户端自述分开。
`time` 用服务端接收时间作 SLS 主时间(客户端时钟不可信),client_ts 另存为可查字段。
"""
from __future__ import annotations

import json
import logging
import os
from datetime import datetime
from logging.handlers import RotatingFileHandler
from pathlib import Path

# 仅这些客户端键提到输出行顶层;其余(含客户端自带 data)一律并入 data,防 SLS 索引列爆炸
_TOP_LEVEL_KEYS = ("client_ts", "level", "trace_id", "tag", "msg")

_logger: logging.Logger | None = None


def _max_msg_bytes() -> int:
    return int(os.getenv("APPLOG_MAX_MSG_BYTES", "8192"))


def _build_logger() -> logging.Logger:
    lg = logging.getLogger("client")
    lg.setLevel(logging.INFO)
    lg.propagate = False  # 不冒泡到 root → 不写进 app-server.log
    log_file = os.getenv("CLIENT_LOG_FILE") or str(
        Path(os.getenv("LOG_DIR", "logs")) / "app-client.log"
    )
    Path(log_file).parent.mkdir(parents=True, exist_ok=True)
    handler = RotatingFileHandler(
        log_file,
        maxBytes=int(os.getenv("CLIENT_LOG_MAX_BYTES", str(20 * 1024 * 1024))),
        backupCount=int(os.getenv("CLIENT_LOG_BACKUP_COUNT", "10")),
        encoding="utf-8",
    )
    handler.setFormatter(logging.Formatter("%(message)s"))  # 行已是 JSON,不再包装
    lg.handlers = [handler]
    return lg


def get_logger() -> logging.Logger:
    global _logger
    if _logger is None:
        _logger = _build_logger()
    return _logger


def reset_client_logger() -> None:
    """测试用:关闭并丢弃当前 logger,使下次 get_logger 按当时 env 重建(切临时文件)。"""
    global _logger
    if _logger is not None:
        for h in list(_logger.handlers):
            h.close()
        _logger.handlers = []
    _logger = None


def _truncate_msg(msg: str) -> tuple[str, bool]:
    raw = msg.encode("utf-8")
    limit = _max_msg_bytes()
    if len(raw) <= limit:
        return msg, False
    # 按字节截断后解码,忽略截断处半个多字节字符
    return raw[:limit].decode("utf-8", "ignore") + "…[truncated]", True


def _build_line(
    record: dict, *, meta: dict, client_ip: str, service: str, now_iso: str
) -> str:
    out: dict = {
        "time": now_iso,
        "source": "client",
        "service": service,
        "client_ip": client_ip,
    }
    # 批级公共字段(非空才带)
    for k in ("device_id", "user_id", "app_ver", "platform", "sent_at"):
        v = meta.get(k)
        if v is not None:
            out[k] = v
    # 白名单键提顶层
    if record.get("level") is not None:
        out["level"] = str(record["level"]).upper()
    if record.get("trace_id"):
        out["trace_id"] = record["trace_id"]
    if record.get("tag"):
        out["tag"] = record["tag"]
    if record.get("client_ts") is not None:
        out["client_ts"] = record["client_ts"]
    if record.get("msg") is not None:
        msg, truncated = _truncate_msg(str(record["msg"]))
        out["msg"] = msg
        if truncated:
            out["msg_truncated"] = True
    # 其余键(含客户端自带 data)并入 data
    data: dict = {}
    client_data = record.get("data")
    if isinstance(client_data, dict):
        data.update(client_data)
    for k, v in record.items():
        if k in _TOP_LEVEL_KEYS or k == "data":
            continue
        data[k] = v
    if data:
        out["data"] = data
    return json.dumps(out, ensure_ascii=False, default=str)


def write_records(
    records: list[dict], *, meta: dict, client_ip: str
) -> tuple[int, int]:
    """把一批客户端日志逐条写入专用文件。返回 (received, dropped)。

    尽力而为(fire-and-forget):logger 初始化或单条写入失败只跳过并计 dropped,
    不抛给上层——端点因此永不因写日志而 500。
    """
    try:
        lg = get_logger()
    except Exception:  # noqa: BLE001 — 初始化失败也不能让端点 500
        logging.getLogger("shagua.applog").exception("client log writer init failed")
        return 0, len(records)
    service = os.getenv("CLIENT_LOG_SERVICE_NAME", "app-client")
    now_iso = datetime.now().strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3]
    received = dropped = 0
    for rec in records:
        try:
            line = _build_line(
                rec, meta=meta, client_ip=client_ip, service=service, now_iso=now_iso
            )
            lg.info(line)
            received += 1
        except Exception:  # noqa: BLE001 — 坏条跳过,不影响其余
            dropped += 1
    return received, dropped
```

- [ ] **Step 4: 跑测试确认通过**

Run: `pytest tests/test_applog.py -q`
Expected: PASS（5 个 writer 测试全绿）。

- [ ] **Step 5: ruff**

Run: `ruff check app/core/client_log.py tests/test_applog.py`
Expected: 无错误（如有 import 排序等自动可修问题：`ruff check --fix`）。

- [ ] **Step 6: 提交**

```bash
git add app/core/client_log.py tests/test_applog.py
git commit -m "feat(applog): 客户端日志专用落盘 writer(白名单+data 兜底, propagate=False)"
```

---

## Task 2: 端点 + schema + 注册

**Files:**
- Create: `app/schemas/applog.py`
- Create: `app/api/v1/applog.py`
- Modify: `app/main.py`（import + `include_router`）
- Test: `tests/test_applog.py`（追加端点用例）

- [ ] **Step 1: 追加失败测试（端点层）**

在 `tests/test_applog.py` 末尾追加：

```python
# ---------------- 端点层 ----------------

def _post(client, body):
    return client.post("/api/v1/applog/batch", json=body)


def test_endpoint_happy_path(client, client_log_file):
    body = {
        "device_id": "d-1", "user_id": 42, "app_ver": "1.2.3", "platform": "android",
        "logs": [
            {"client_ts": 1, "level": "info", "msg": "a", "trace_id": "t1"},
            {"client_ts": 2, "level": "warn", "msg": "b"},
        ],
    }
    resp = _post(client, body)
    assert resp.status_code == 200
    assert resp.json() == {"ok": True, "received": 2, "dropped": 0}
    lines = _read_lines(client_log_file)
    assert len(lines) == 2
    assert lines[0]["trace_id"] == "t1"       # 端到端:trace_id 落到文件顶层
    assert lines[0]["client_ip"]              # 服务端补了 IP


def test_endpoint_rejects_batch_over_max(client, client_log_file):
    body = {"device_id": "d-1",
            "logs": [{"client_ts": i, "level": "info", "msg": str(i)} for i in range(501)]}
    resp = _post(client, body)
    assert resp.status_code == 422           # Pydantic max_length=500


def test_endpoint_rejects_body_over_limit(client, client_log_file, monkeypatch):
    monkeypatch.setenv("APPLOG_MAX_BODY_BYTES", "50")
    body = {"device_id": "d-1", "logs": [{"client_ts": 1, "level": "info", "msg": "x" * 500}]}
    resp = _post(client, body)
    assert resp.status_code == 413           # 依赖查 Content-Length,body 校验前拦截


def test_endpoint_returns_200_on_write_failure(client, client_log_file, monkeypatch):
    class _BoomLogger:
        name = "client"
        propagate = False
        handlers: list = []

        def info(self, *a, **k):
            raise RuntimeError("disk full")

    monkeypatch.setattr(client_log, "get_logger", lambda: _BoomLogger())
    body = {"device_id": "d-1", "logs": [
        {"client_ts": 1, "level": "info", "msg": "a"},
        {"client_ts": 2, "level": "info", "msg": "b"}]}
    resp = _post(client, body)
    assert resp.status_code == 200           # fire-and-forget:写失败不 500
    assert resp.json() == {"ok": True, "received": 0, "dropped": 2}


def test_endpoint_missing_device_id_is_422(client, client_log_file):
    resp = _post(client, {"logs": [{"client_ts": 1, "level": "info", "msg": "a"}]})
    assert resp.status_code == 422           # device_id 必填
```

- [ ] **Step 2: 跑测试确认失败**

Run: `pytest tests/test_applog.py -q`
Expected: FAIL —— 端点未注册，`POST /api/v1/applog/batch` 返回 404（happy-path 断言 200 失败）。

- [ ] **Step 3: 实现 schema `app/schemas/applog.py`**

```python
"""客户端运行日志批量上报 schema。

批级公共字段(device_id/user_id/app_ver/platform/sent_at)发一次;logs 为原始 dict 列表,
**不强类型**——尽力而为的日志链路,单条内容异常不该让整批 422。每条的「白名单键 + data
兜底」拆分在写入层 [app.core.client_log] 做(见 spec §4/§5)。
"""
from __future__ import annotations

import os
from typing import Any

from pydantic import BaseModel, Field

_MAX_BATCH = int(os.getenv("APPLOG_MAX_BATCH", "500"))


class AppLogBatchIn(BaseModel):
    device_id: str = Field(max_length=64)
    user_id: int | None = None
    app_ver: str | None = Field(default=None, max_length=32)
    platform: str | None = Field(default=None, max_length=16)
    sent_at: int | None = None
    logs: list[dict[str, Any]] = Field(min_length=1, max_length=_MAX_BATCH)


class AppLogIngestOut(BaseModel):
    ok: bool = True
    received: int
    dropped: int = 0
```

- [ ] **Step 4: 实现端点 `app/api/v1/applog.py`**

```python
"""客户端运行日志批量上报接口。

POST /api/v1/applog/batch — 批量接收客户端运行日志,逐条写专用滚动文件 logs/app-client.log
(供 Logtail 采进独立 SLS logstore)。鉴权同 analytics(不强制登录,user_id 可选在 body)。
fire-and-forget:写失败也不 500(避免客户端重试风暴);超批 422、超体积 413、msg 超限截断。
"""
from __future__ import annotations

import os

from fastapi import APIRouter, Depends, HTTPException, Request

from app.core.client_log import write_records
from app.core.ratelimit import rate_limit
from app.schemas.applog import AppLogBatchIn, AppLogIngestOut

router = APIRouter(prefix="/api/v1/applog", tags=["applog"])


def _client_ip(request: Request) -> str:
    """取客户端 IP:生产经 nginx 反代优先 X-Forwarded-For 首段,否则直连 IP(同 analytics)。"""
    xff = request.headers.get("x-forwarded-for")
    if xff:
        return xff.split(",")[0].strip()
    return request.client.host if request.client else ""


def _enforce_body_limit(request: Request) -> None:
    """依赖:body 声明过大直接 413(在 body 校验前拦截)。缺 Content-Length 由 nginx 兜底。"""
    max_bytes = int(os.getenv("APPLOG_MAX_BODY_BYTES", str(1024 * 1024 * 2)))
    cl = request.headers.get("content-length")
    if cl is not None and cl.isdigit() and int(cl) > max_bytes:
        raise HTTPException(status_code=413, detail="日志批量过大")


@router.post(
    "/batch",
    response_model=AppLogIngestOut,
    summary="批量上报客户端运行日志",
    dependencies=[
        Depends(rate_limit(120, 60, "applog-batch")),
        Depends(_enforce_body_limit),
    ],
)
def ingest_logs(batch: AppLogBatchIn, request: Request) -> AppLogIngestOut:
    received, dropped = write_records(
        batch.logs,
        meta={
            "device_id": batch.device_id,
            "user_id": batch.user_id,
            "app_ver": batch.app_ver,
            "platform": batch.platform,
            "sent_at": batch.sent_at,
        },
        client_ip=_client_ip(request),
    )
    return AppLogIngestOut(received=received, dropped=dropped)
```

- [ ] **Step 5: 注册路由 `app/main.py`**

在 import 区（analytics_router 之后，约 [app/main.py:23](../../../app/main.py#L23)）加：

```python
from app.api.v1.applog import router as applog_router
```

在 `include_router` 区（`app.include_router(analytics_router)` 之后，约 [app/main.py:125](../../../app/main.py#L125)）加：

```python
app.include_router(applog_router)
```

- [ ] **Step 6: 跑测试确认通过**

Run: `pytest tests/test_applog.py -q`
Expected: PASS（writer 5 + 端点 5，共 10 个）。

- [ ] **Step 7: 全量测试 + ruff**

Run: `pytest -q && ruff check app/api/v1/applog.py app/schemas/applog.py app/main.py`
Expected: 全绿、无 lint 错误。

- [ ] **Step 8: 提交**

```bash
git add app/api/v1/applog.py app/schemas/applog.py app/main.py tests/test_applog.py
git commit -m "feat(applog): POST /api/v1/applog/batch 批量上报端点(限流+体积上限+fire-and-forget)"
```

---

## Task 3: `.env.example` 文档化新 env（可选但推荐）

**Files:**
- Modify: `.env.example`

- [ ] **Step 1: 追加 env 说明**

在 `.env.example` 末尾（或日志相关区块）追加，让运维知道可调项：

```bash
# 客户端运行日志上报(POST /api/v1/applog/batch)→ 落 logs/app-client.log 供 Logtail 采集
# 独立于服务日志 app-server.log;滚动机制同服务日志。默认值见 app/core/client_log.py。
# CLIENT_LOG_FILE=logs/app-client.log
# CLIENT_LOG_MAX_BYTES=20971520      # 单文件 20MB 滚动
# CLIENT_LOG_BACKUP_COUNT=10         # 保留 10 个 → ~200MB 缓冲
# CLIENT_LOG_SERVICE_NAME=app-client # 输出行 service 字段
# APPLOG_MAX_BATCH=500               # 单批最大条数(超 → 422)
# APPLOG_MAX_BODY_BYTES=2097152      # 请求体上限 2MB(超 → 413)
# APPLOG_MAX_MSG_BYTES=8192          # 单条 msg 超限截断
```

- [ ] **Step 2: 提交**

```bash
git add .env.example
git commit -m "docs(applog): .env.example 补充客户端日志上报可调 env"
```

> 若仓库无 `.env.example`（以 `git ls-files .env.example` 确认），跳过本任务。

---

## Definition of Done

- [ ] `pytest -q` 全绿（含新增 `tests/test_applog.py` 10 用例）。
- [ ] `ruff check .` 无错误。
- [ ] `POST /api/v1/applog/batch` 手动冒烟：发一批含 `trace_id` 的日志，确认 `logs/app-client.log` 出现对应单行 JSON、`trace_id` 在顶层、未知键落在 `data`，且 `logs/app-server.log` **未**被写入客户端记录。
- [ ] Spec 的运维项（独立 logstore、Logtail JSON 模式采 `app-client.log`、给 `trace_id` 建索引、nginx `client_max_body_size` 对齐）已同步给运维（不在本仓代码内，见 spec §8）。

## Self-Review 结论（作者已核对）

- **Spec 覆盖**：§4 端点/限额 → Task 2；§5 白名单+data 兜底 → Task 1 `_build_line`；§6 落盘/滚动/单 worker 注释 → Task 1；§7 trace_id 顶层 → Task 1 + 端点测试；§9 fire-and-forget → Task 1 `write_records` + 端点测试；§13 env → Task 1/2 读取 + Task 3 文档化。§8 为纯运维配置，列入 Definition of Done。
- **无占位符**：所有步骤含完整代码/命令/预期。
- **类型/命名一致**：`write_records(records, *, meta, client_ip) -> (received, dropped)`、`get_logger()`、`reset_client_logger()`、`_TOP_LEVEL_KEYS`、`AppLogBatchIn/AppLogIngestOut` 在 Task 1/2 间一致引用。
