ForgeFlow-ERP/docs/superpowers/plans/2026-06-11-miniapp-operation-report-sync.md

# Miniapp Operation Report Sync 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:** Make 工序报工 page refresh fast by separating read-only list refresh from explicit miniapp-to-ERP synchronization, with sync range fixed to automatic catch-up from the last successful sync time.

**Architecture:** Keep the existing list endpoint for display, but make the frontend call it with `sync_to_erp=false`. Add a dedicated backend sync endpoint that reads and locks `MINIAPP_OPERATION_REPORT_LAST_SYNCED_AT` in `sys_system_config`, syncs all miniapp report items changed since that time, and only updates the timestamp after the transaction succeeds. The frontend adds a disabled loading state for the sync button and refreshes the list after sync.

**Tech Stack:** FastAPI, SQLAlchemy, Pydantic, MySQL-compatible row locking, Vue 3 Composition API, Vite, Python `unittest`, Node static UI checks.

---

## File Structure

- Modify: `backend/app/services/system_config.py`
  - Add constants and helpers for `MINIAPP_OPERATION_REPORT_LAST_SYNCED_AT`.
  - Add parse/format helpers for sync timestamps.
  - Add helpers to initialize and lock the sync config row, so first sync also has a database lock target.
- Modify: `backend/app/schemas/operations.py`
  - Add `MiniAppOperationReportSyncRead`.
- Modify: `backend/app/api/routes/production.py`
  - Add helper query for miniapp rows changed since the last successful sync.
  - Add `POST /production/miniapp-operation-reports/sync`.
  - Keep list endpoint behavior backward-compatible, but frontend will call it as read-only.
- Modify: `backend/tests/test_miniapp_operation_report_date_filter.py`
  - Add tests for read-only list refresh and explicit automatic catch-up sync.
- Modify: `frontend/src/views/OperationReportView.vue`
  - Change list refresh to `sync_to_erp=false`.
  - Add sync button, loading state, last-sync display, and result feedback.
- Modify: `frontend/scripts/test-operation-report-date-filter.mjs`
  - Update static checks for read-only refresh.
- Create: `frontend/scripts/test-operation-report-sync.mjs`
  - Static checks for sync button, dedicated endpoint, disabled loading state, and auto-refresh after sync.

---

## Task 1: Backend Sync Timestamp Helpers

**Files:**
- Modify: `backend/app/services/system_config.py`
- Test: `backend/tests/test_miniapp_operation_report_date_filter.py`

- [ ] **Step 1: Add failing helper tests**

Append these imports in `backend/tests/test_miniapp_operation_report_date_filter.py`:

```python
from app.services.system_config import (  # noqa: E402
    MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_CODE,
    format_miniapp_operation_report_last_sync,
    get_miniapp_operation_report_last_sync,
    parse_miniapp_operation_report_last_sync,
    upsert_miniapp_operation_report_last_sync,
)
from app.models.org import SystemConfig  # noqa: E402
from sqlalchemy import select  # noqa: E402
```

Add this test method to `MiniAppOperationReportDateFilterTest`:

```python
    def test_miniapp_operation_report_last_sync_config_round_trips_datetime(self) -> None:
        timestamp = datetime(2026, 6, 11, 10, 30, 15)

        row = upsert_miniapp_operation_report_last_sync(self.db, timestamp, updated_by=1)
        self.db.commit()

        persisted = self.db.scalar(
            select(SystemConfig).where(SystemConfig.config_code == MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_CODE)
        )
        self.assertIsNotNone(persisted)
        self.assertEqual(row.config_value, "2026-06-11 10:30:15")
        self.assertEqual(format_miniapp_operation_report_last_sync(timestamp), "2026-06-11 10:30:15")
        self.assertEqual(parse_miniapp_operation_report_last_sync("2026-06-11 10:30:15"), timestamp)
        self.assertEqual(get_miniapp_operation_report_last_sync(self.db), timestamp)
```

- [ ] **Step 2: Run the failing backend test**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
python -m pytest tests/test_miniapp_operation_report_date_filter.py::MiniAppOperationReportDateFilterTest::test_miniapp_operation_report_last_sync_config_round_trips_datetime -q
```

Expected: FAIL because the sync config helper functions do not exist.

- [ ] **Step 3: Implement timestamp helpers**

In `backend/app/services/system_config.py`, add the import:

```python
from datetime import datetime
```

The file already imports `datetime`; keep one import only.

Add these constants near existing config constants:

```python
MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_CODE = "MINIAPP_OPERATION_REPORT_LAST_SYNCED_AT"
MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_NAME = "小程序报工上次同步成功时间"
MINIAPP_OPERATION_REPORT_LAST_SYNC_FORMAT = "%Y-%m-%d %H:%M:%S"
```

Add these helpers after `get_smart_operation_report_enabled`:

```python
def format_miniapp_operation_report_last_sync(value: datetime) -> str:
    return value.strftime(MINIAPP_OPERATION_REPORT_LAST_SYNC_FORMAT)


def parse_miniapp_operation_report_last_sync(value: str | None) -> datetime | None:
    normalized = str(value or "").strip()
    if not normalized:
        return None
    return datetime.strptime(normalized, MINIAPP_OPERATION_REPORT_LAST_SYNC_FORMAT)


def get_miniapp_operation_report_last_sync(db: Session) -> datetime | None:
    value = get_system_config_value(db, MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_CODE, "")
    return parse_miniapp_operation_report_last_sync(value)


def upsert_miniapp_operation_report_last_sync(
    db: Session,
    timestamp: datetime,
    *,
    updated_by: int | None = None,
) -> SystemConfig:
    return upsert_system_config(
        db,
        config_code=MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_CODE,
        config_name=MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_NAME,
        config_value=format_miniapp_operation_report_last_sync(timestamp),
        remark="工序报工小程序同步 ERP 的上次成功截止时间",
        updated_by=updated_by,
        now=timestamp,
    )


def ensure_miniapp_operation_report_last_sync_config(
    db: Session,
    *,
    updated_by: int | None = None,
) -> SystemConfig:
    row = db.scalar(select(SystemConfig).where(SystemConfig.config_code == MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_CODE))
    if row:
        return row
    timestamp = datetime.now()
    row = SystemConfig(
        config_code=MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_CODE,
        config_name=MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_NAME,
        config_value="",
        remark="工序报工小程序同步 ERP 的上次成功截止时间",
        status="ACTIVE",
        created_by=updated_by,
        updated_by=updated_by,
        created_at=timestamp,
        updated_at=timestamp,
    )
    db.add(row)
    db.flush()
    return row


def lock_miniapp_operation_report_last_sync(db: Session) -> SystemConfig | None:
    ensure_miniapp_operation_report_last_sync_config(db)
    return db.scalar(
        select(SystemConfig)
        .where(
            SystemConfig.config_code == MINIAPP_OPERATION_REPORT_LAST_SYNC_CONFIG_CODE,
            SystemConfig.status == "ACTIVE",
        )
        .with_for_update()
    )
```

- [ ] **Step 4: Run the helper test**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
python -m pytest tests/test_miniapp_operation_report_date_filter.py::MiniAppOperationReportDateFilterTest::test_miniapp_operation_report_last_sync_config_round_trips_datetime -q
```

Expected: PASS.

---

## Task 2: Backend Read-Only List And Explicit Sync Endpoint

**Files:**
- Modify: `backend/app/schemas/operations.py`
- Modify: `backend/app/api/routes/production.py`
- Modify: `backend/tests/test_miniapp_operation_report_date_filter.py`

- [ ] **Step 1: Add failing tests for read-only refresh and automatic catch-up sync**

Add imports in `backend/tests/test_miniapp_operation_report_date_filter.py`:

```python
from app.api.routes import production as production_routes  # noqa: E402
from app.api.routes.production import sync_miniapp_operation_reports  # noqa: E402
from app.models.operations import OperationReport  # noqa: E402
```

Add this test method:

```python
    def test_list_reports_with_sync_disabled_does_not_call_sync(self) -> None:
        calls: list[int] = []
        original_sync = production_routes._sync_miniapp_item_to_erp

        def fake_sync(db, source_item_id, **kwargs):
            calls.append(source_item_id)
            return None

        production_routes._sync_miniapp_item_to_erp = fake_sync
        try:
            rows = list_miniapp_operation_reports(
                limit=100,
                sync_to_erp=False,
                start_date=date(2026, 5, 18),
                end_date=date(2026, 6, 1),
                db=self.db,
            )
        finally:
            production_routes._sync_miniapp_item_to_erp = original_sync

        self.assertEqual(len(rows), 2)
        self.assertEqual(calls, [])
        self.assertEqual(self.db.query(OperationReport).count(), 0)
```

Add this test method:

```python
    def test_sync_reports_uses_last_success_time_not_page_filter(self) -> None:
        upsert_miniapp_operation_report_last_sync(self.db, datetime(2026, 5, 19, 0, 0, 0), updated_by=1)
        self.db.commit()
        calls: list[int] = []
        original_sync = production_routes._sync_miniapp_item_to_erp
        original_now = production_routes.datetime

        class FixedDateTime(datetime):
            @classmethod
            def now(cls, tz=None):
                return cls(2026, 6, 11, 12, 0, 0)

        def fake_sync(db, source_item_id, **kwargs):
            calls.append(source_item_id)
            return None

        production_routes._sync_miniapp_item_to_erp = fake_sync
        production_routes.datetime = FixedDateTime
        try:
            result = sync_miniapp_operation_reports(db=self.db)
        finally:
            production_routes._sync_miniapp_item_to_erp = original_sync
            production_routes.datetime = original_now

        self.assertEqual(calls, [2, 3])
        self.assertEqual(result.total, 2)
        self.assertEqual(result.sync_from, datetime(2026, 5, 19, 0, 0, 0))
        self.assertEqual(result.sync_to, datetime(2026, 6, 11, 12, 0, 0))
        self.assertEqual(get_miniapp_operation_report_last_sync(self.db), datetime(2026, 6, 11, 12, 0, 0))
```

- [ ] **Step 2: Run the failing tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
python -m pytest tests/test_miniapp_operation_report_date_filter.py -q
```

Expected: FAIL because `sync_miniapp_operation_reports` and response schema do not exist yet.

- [ ] **Step 3: Add sync response schema**

In `backend/app/schemas/operations.py`, add after `MiniAppOperationReportRead`:

```python
class MiniAppOperationReportSyncRead(BaseModel):
    sync_from: datetime | None = None
    sync_to: datetime
    last_synced_at: datetime
    total: int = 0
    synced: int = 0
    skipped: int = 0
    failed: int = 0
```

- [ ] **Step 4: Import sync helpers and schema**

In `backend/app/api/routes/production.py`, extend the operations schema import to include:

```python
MiniAppOperationReportSyncRead,
```

Extend the system config imports to include:

```python
get_miniapp_operation_report_last_sync,
lock_miniapp_operation_report_last_sync,
upsert_miniapp_operation_report_last_sync,
```

- [ ] **Step 5: Add changed-since miniapp row helper**

In `backend/app/api/routes/production.py`, add this helper near `_miniapp_report_item_rows`:

```python
def _miniapp_report_item_rows_changed_since(
    db: Session,
    *,
    sync_from: datetime | None,
    sync_to: datetime,
    limit: int = 800,
) -> list[dict]:
    stmt = (
        select(
            MiniAppProductionReport.id.label("source_report_id"),
            MiniAppProductionReportItem.id.label("source_item_id"),
            func.coalesce(
                MiniAppProductionReportItem.attendance_point_name,
                MiniAppProductionReport.attendance_point_name,
            ).label("attendance_point_name"),
            MiniAppProductionReport.employee_phone.label("employee_phone"),
            MiniAppPersonnel.name.label("employee_name"),
            MiniAppProductionReport.report_date.label("report_date"),
            MiniAppProductionReport.start_at.label("start_time"),
            MiniAppProductionReport.end_at.label("end_time"),
            MiniAppProductionReport.duration_minutes.label("duration_minutes"),
            MiniAppProductionReport.effective_minutes.label("effective_minutes"),
            MiniAppProductionReport.status.label("status"),
            MiniAppProductionReport.reviewer_phone.label("reviewer_phone"),
            MiniAppProductionReport.reviewed_at.label("reviewed_at"),
            MiniAppProductionReport.reject_reason.label("reject_reason"),
            MiniAppProductionReportItem.device_no.label("device_no"),
            MiniAppProductionReportItem.project_no.label("project_no"),
            MiniAppProductionReportItem.product_name.label("product_name"),
            MiniAppProductionReportItem.material_code.label("material_code"),
            MiniAppProductionReportItem.material_name.label("material_name"),
            MiniAppProductionReportItem.raw_material_batch_no.label("raw_material_batch_no"),
            MiniAppProductionReportItem.process_name.label("process_name"),
            MiniAppProductionReportItem.stamping_method.label("stamping_method"),
            MiniAppProductionReportItem.standard_beat.label("standard_beat"),
            MiniAppProductionReportItem.standard_workload.label("standard_workload"),
            MiniAppProductionReportItem.good_qty.label("good_qty"),
            MiniAppProductionReportItem.defect_qty.label("defect_qty"),
            MiniAppProductionReportItem.scrap_qty.label("miniapp_scrap_qty"),
            MiniAppProductionReportItem.allocated_minutes.label("allocated_minutes"),
            MiniAppProductionReportItem.started_at.label("started_at"),
        )
        .join(MiniAppProductionReport, MiniAppProductionReport.id == MiniAppProductionReportItem.report_id)
        .outerjoin(MiniAppPersonnel, MiniAppPersonnel.phone == MiniAppProductionReport.employee_phone)
        .where(MiniAppProductionReport.updated_at <= sync_to)
        .order_by(MiniAppProductionReport.updated_at.asc(), MiniAppProductionReportItem.id.asc())
        .limit(limit)
    )
    if sync_from is not None:
        stmt = stmt.where(MiniAppProductionReport.updated_at > sync_from)
    rows = db.execute(stmt).mappings().all()
    return [dict(row) for row in rows]
```

- [ ] **Step 6: Add explicit sync endpoint**

In `backend/app/api/routes/production.py`, add before the existing `@router.get("/miniapp-operation-reports"` route:

```python
@router.post("/miniapp-operation-reports/sync", response_model=MiniAppOperationReportSyncRead)
def sync_miniapp_operation_reports(
    limit: int = Query(default=800, ge=1, le=800),
    db: Session = Depends(get_db),
) -> MiniAppOperationReportSyncRead:
    lock_miniapp_operation_report_last_sync(db)
    sync_from = get_miniapp_operation_report_last_sync(db)
    sync_to = datetime.now()
    rows = _miniapp_report_item_rows_changed_since(db, sync_from=sync_from, sync_to=sync_to, limit=limit)
    cache = _new_miniapp_resolve_cache()
    report_nos = [_miniapp_report_no(int(row["source_report_id"]), int(row["source_item_id"])) for row in rows]
    if report_nos:
        existing_reports = db.scalars(select(OperationReport).where(OperationReport.report_no.in_(report_nos))).all()
        cache["report"].update({report.report_no: report for report in existing_reports})

    touched_work_order_ids: set[int] = set()
    synced = 0
    skipped = 0
    for row in rows:
        synced_report = _sync_miniapp_item_to_erp(
            db,
            int(row["source_item_id"]),
            required=False,
            row=row,
            cache=cache,
            sync_status=False,
        )
        if synced_report:
            synced += 1
            if synced_report.work_order_id:
                touched_work_order_ids.add(int(synced_report.work_order_id))
        else:
            skipped += 1

    for work_order_id in touched_work_order_ids:
        sync_work_order_status(db, work_order_id)
    row = upsert_miniapp_operation_report_last_sync(db, sync_to)
    db.commit()
    db.refresh(row)
    return MiniAppOperationReportSyncRead(
        sync_from=sync_from,
        sync_to=sync_to,
        last_synced_at=sync_to,
        total=len(rows),
        synced=synced,
        skipped=skipped,
        failed=0,
    )
```

- [ ] **Step 7: Run backend tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
python -m pytest tests/test_miniapp_operation_report_date_filter.py -q
```

Expected: PASS.

---

## Task 3: Frontend Read-Only Refresh And Sync Button

**Files:**
- Modify: `frontend/src/views/OperationReportView.vue`
- Modify: `frontend/scripts/test-operation-report-date-filter.mjs`
- Create: `frontend/scripts/test-operation-report-sync.mjs`

- [ ] **Step 1: Add failing frontend static test**

Create `frontend/scripts/test-operation-report-sync.mjs`:

```javascript
import assert from "node:assert/strict";
import { readFileSync } from "node:fs";
import path from "node:path";

const root = process.cwd();
const source = readFileSync(path.join(root, "src/views/OperationReportView.vue"), "utf8");

assert.match(source, /syncReportsToErp/, "operation report page should define an explicit ERP sync action");
assert.match(source, /postResource\("\/production\/miniapp-operation-reports\/sync"/, "sync action should call the dedicated sync endpoint");
assert.match(source, /syncingReports\.value\s*=\s*true/, "sync action should enable loading state");
assert.match(source, /:disabled="syncingReports"/, "sync button should be disabled while syncing");
assert.match(source, /同步中/, "sync button should show a syncing label");
assert.match(source, /lastSyncedAt/, "page should display the last successful sync time");
assert.match(source, /await loadReports\(\)/, "page should refresh the list after sync succeeds");
assert.doesNotMatch(source, /sync_to_erp:\s*"true"/, "list refresh should not sync ERP implicitly");
assert.match(source, /sync_to_erp:\s*"false"/, "list refresh should explicitly request read-only data");

console.log("operation report sync checks passed");
```

- [ ] **Step 2: Run failing frontend test**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-operation-report-sync.mjs
```

Expected: FAIL because the sync action and button do not exist.

- [ ] **Step 3: Update imports and state**

In `frontend/src/views/OperationReportView.vue`, change:

```javascript
import { fetchResource } from "../services/api";
```

to:

```javascript
import { fetchResource, postResource } from "../services/api";
```

Add after `const reports = ref([]);`:

```javascript
const syncingReports = ref(false);
const lastSyncedAt = ref("");
```

- [ ] **Step 4: Make list refresh read-only**

In `loadReports()`, change:

```javascript
const params = new URLSearchParams({
  limit: "800",
  sync_to_erp: "true"
});
```

to:

```javascript
const params = new URLSearchParams({
  limit: "800",
  sync_to_erp: "false"
});
```

Change success text from:

```javascript
feedbackMessage.value = "小程序报工数据已刷新并同步 ERP 生产核算";
```

to:

```javascript
feedbackMessage.value = "小程序报工列表已刷新";
```

- [ ] **Step 5: Add explicit sync action**

Add after `loadReports()`:

```javascript
async function syncReportsToErp() {
  if (syncingReports.value) {
    return;
  }
  feedbackMessage.value = "";
  errorMessage.value = "";
  syncingReports.value = true;
  try {
    const result = await postResource("/production/miniapp-operation-reports/sync", {});
    lastSyncedAt.value = result.last_synced_at || result.sync_to || "";
    feedbackMessage.value = [
      `同步完成：共 ${result.total || 0} 条`,
      `成功 ${result.synced || 0} 条`,
      `跳过 ${result.skipped || 0} 条`,
      `失败 ${result.failed || 0} 条`,
      `同步范围：${result.sync_from || "首次同步"} 至 ${result.sync_to || "-"}`
    ].join("，");
    await loadReports();
  } catch (error) {
    errorMessage.value = error.message || "同步小程序报工到 ERP 失败，上次同步成功时间未更新，请稍后重试";
  } finally {
    syncingReports.value = false;
  }
}
```

- [ ] **Step 6: Add sync UI**

In the template, inside `.operation-filter-ribbon` after the refresh button, add:

```vue
<button class="primary-button" type="button" :disabled="syncingReports" @click="syncReportsToErp">
  {{ syncingReports ? "同步中..." : "同步到 ERP" }}
</button>
<span class="operation-sync-hint">上次同步：{{ lastSyncedAt || "暂无" }}</span>
```

- [ ] **Step 7: Update existing date-filter static test**

In `frontend/scripts/test-operation-report-date-filter.mjs`, replace the old negative check:

```javascript
assert.doesNotMatch(
  source,
  /fetchResource\("\/production\/miniapp-operation-reports\?limit=800&sync_to_erp=true"/,
  "operation report page must not use the old all-row request"
);
```

with:

```javascript
assert.match(source, /sync_to_erp:\s*"false"/, "operation report refresh should be read-only by default");
assert.doesNotMatch(source, /sync_to_erp:\s*"true"/, "operation report refresh must not sync ERP implicitly");
```

- [ ] **Step 8: Run frontend static tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-operation-report-date-filter.mjs
node scripts/test-operation-report-sync.mjs
```

Expected: both PASS.

---

## Task 4: Final Verification

**Files:**
- Backend and frontend files from prior tasks.

- [ ] **Step 1: Run backend focused tests**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/backend
python -m pytest tests/test_miniapp_operation_report_date_filter.py -q
```

Expected: PASS.

- [ ] **Step 2: Run frontend static checks**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
node scripts/test-operation-report-date-filter.mjs
node scripts/test-operation-report-sync.mjs
```

Expected: PASS.

- [ ] **Step 3: Run frontend build**

Run:

```bash
cd /Users/souplearn/Gitlab/py/ForgeFlow-ERP/frontend
npm run build
```

Expected: Vite build succeeds. Chunk-size warning is acceptable if it matches the existing project warning.

- [ ] **Step 4: Manual browser verification**

Open `http://127.0.0.1:5173/operation-reports` and verify:

- Page loads the list without waiting for ERP sync.
- `刷新` only refreshes the current visible list.
- `同步到 ERP` changes to `同步中...` and cannot be clicked again while syncing.
- Sync success message shows total, success, skipped, failed, and sync range.
- The list refreshes after sync completes.
- Page date filters do not affect the backend sync range.

---

## Self-Review

- Spec coverage: The plan covers read-only refresh, automatic catch-up sync, no user sync date range, loading button, last-sync persistence, idempotent existing sync reuse, and verification.
- Placeholder scan: No unfinished placeholders or vague “add tests” instructions remain.
- Type consistency: The plan consistently uses `MiniAppOperationReportSyncRead`, `sync_miniapp_operation_reports`, `MINIAPP_OPERATION_REPORT_LAST_SYNCED_AT`, `syncingReports`, and `lastSyncedAt`.