Phase 1: bulk-upload endpoints (bulk-crops, screen-types, anchors)

README.md

@@ -35,10 +35,10 @@ See the source repo for the full API reference. Main surface:
 - `GET /health` — service status
 - `GET /knowledge` — merged icon knowledge base (phash → name)
 - `GET /model/version` — latest trained model metadata
-- `POST /contribute` — single crop + label
-- `POST /contribute/bulk` — batch of crops (planned, Phase 1)
-- `POST /upload/screen-types` — screen-type screenshot (planned, Phase 1)
-- `POST /upload/anchors` — anchor grid (planned, Phase 1)
+- `POST /contribute` — single crop + label (legacy, → `sets-sto/warp-knowledge`)
+- `POST /contribute/bulk-crops` — batch of confirmed crops + annotations (→ `sets-sto/sto-icon-dataset`)
+- `POST /upload/screen-types` — batch of screen-type screenshots (→ `sets-sto/sto-icon-dataset`)
+- `POST /upload/anchors` — batch of anchor grids (→ `sets-sto/sto-icon-dataset`)
 - `POST /admin/merge` — admin-only knowledge merge
 
 ## Secrets
@@ -48,6 +48,7 @@ Configured in Space Settings → Secrets:
 - `HF_TOKEN` — fine-grained write token, scoped to both
   `datasets/sets-sto/warp-knowledge` (current `/contribute`) and
   `datasets/sets-sto/sto-icon-dataset` (Phase 1 bulk endpoints)
-- `HF_REPO_ID` — e.g. `sets-sto/warp-knowledge`
+- `HF_REPO_ID` — e.g. `sets-sto/warp-knowledge` (target for `/contribute`, `/admin/merge`)
+- `HF_ICONS_REPO_ID` — e.g. `sets-sto/sto-icon-dataset` (target for Phase 1 bulk endpoints; defaults to that value if unset)
 - `ADMIN_KEY` — guards `/admin/*` endpoints
 - `MAX_REQ_PER_IP` — per-IP daily rate cap

main.py

@@ -64,10 +64,30 @@ def _load_env():
 _load_env()
 
 # ── Config from environment ────────────────────────────────────────────────────
-HF_TOKEN       = os.environ.get('HF_TOKEN', '')
-HF_REPO_ID     = os.environ.get('HF_REPO_ID', 'sets-sto/warp-knowledge')
-ADMIN_KEY      = os.environ.get('ADMIN_KEY', '')
-MAX_REQ_PER_IP = int(os.environ.get('MAX_REQ_PER_IP', '500'))
+HF_TOKEN          = os.environ.get('HF_TOKEN', '')
+HF_REPO_ID        = os.environ.get('HF_REPO_ID', 'sets-sto/warp-knowledge')
+# Icon-dataset target for bulk-crops, screen-types and anchor-grid uploads
+# (Phase 1 of backend-proxy migration). Separate from HF_REPO_ID because
+# the icon training data lives in its own dataset.
+HF_ICONS_REPO_ID  = os.environ.get('HF_ICONS_REPO_ID', 'sets-sto/sto-icon-dataset')
+ADMIN_KEY         = os.environ.get('ADMIN_KEY', '')
+MAX_REQ_PER_IP    = int(os.environ.get('MAX_REQ_PER_IP', '500'))
+
+# Bulk-endpoint payload limits — guards against accidental floods and HF
+# commit-size limits. Per-item caps mirror client-side validation in
+# warp/trainer/sync.py.
+MAX_BULK_CROPS         = 50
+MAX_BULK_SCREEN_TYPES  = 20
+MAX_BULK_ANCHOR_GRIDS  = 20
+MAX_CROP_PNG_BYTES     = 150_000     # icon crop
+MAX_SCREEN_PNG_BYTES   = 2_500_000   # full screenshot
+MIN_CROP_PX            = 16
+MIN_TEXT_CROP_H        = 10
+MIN_TEXT_CROP_W        = 50
+MAX_NAME_LEN           = 120
+_TEXT_CROP_PREFIXES    = ('ship_type_', 'ship_tier_')
+_INSTALL_ID_RE         = re.compile(r'^[a-zA-Z0-9_-]{8,64}$')
+_SCREEN_TYPE_RE        = re.compile(r'^[a-zA-Z0-9_-]{1,40}$')
 
 # GitHub Config for automated training triggers
 GH_TOKEN = os.environ.get('GH_TOKEN', '')
@@ -123,6 +143,52 @@ class ContributeRequest(BaseModel):
         return re.sub(r'[^a-zA-Z0-9\-_]', '', v)[:64]
 
 
+# ── Phase 1: bulk-upload request models ───────────────────────────────────────
+#
+# These mirror the payloads previously built by warp/trainer/sync.py on the
+# client side. The client used to call HfApi.create_commit directly with a
+# write-scoped HF token; in Phase 2 it will POST these payloads to the
+# backend instead, so the token never leaves the server.
+
+class _BulkCropItem(BaseModel):
+    slot:         str = Field(..., min_length=1, max_length=80)
+    name:         str = Field(..., min_length=1, max_length=MAX_NAME_LEN)
+    crop_png_b64: str = Field(..., min_length=100, max_length=200_000)
+    ml_name:      str = Field('', max_length=MAX_NAME_LEN)
+
+    @field_validator('name', 'ml_name', 'slot')
+    @classmethod
+    def _strip_ctrl(cls, v: str) -> str:
+        return re.sub(r'[\x00-\x1f\x7f]', '', v).strip()
+
+
+class BulkCropsRequest(BaseModel):
+    install_id: str             = Field(..., min_length=8, max_length=64)
+    items:      list[_BulkCropItem] = Field(..., min_length=1, max_length=MAX_BULK_CROPS)
+
+
+class _ScreenTypeItem(BaseModel):
+    png_b64: str = Field(..., min_length=100, max_length=4_000_000)
+
+
+class ScreenTypesRequest(BaseModel):
+    install_id:  str = Field(..., min_length=8, max_length=64)
+    screen_type: str = Field(..., min_length=1, max_length=40)
+    items:       list[_ScreenTypeItem] = Field(..., min_length=1, max_length=MAX_BULK_SCREEN_TYPES)
+
+
+class _AnchorGrid(BaseModel):
+    build_type: str            = Field(..., min_length=1, max_length=40)
+    aspect:     str | None     = Field(None, max_length=16)
+    resolution: str            = Field('', max_length=16)
+    slots:      dict[str, list[float]] = Field(..., min_length=3)
+
+
+class AnchorsRequest(BaseModel):
+    install_id: str               = Field(..., min_length=8, max_length=64)
+    grids:      list[_AnchorGrid] = Field(..., min_length=1, max_length=MAX_BULK_ANCHOR_GRIDS)
+
+
 # ── Endpoints ──────────────────────────────────────────────────────────────────
 
 @app.get('/health')
@@ -222,6 +288,225 @@ async def contribute(req: ContributeRequest, request: Request):
     return {'ok': True, 'contribution_id': contrib_id}
 
 
+# ── Phase 1: bulk-upload endpoints ────────────────────────────────────────────
+#
+# These replace the direct HF writes that warp/trainer/sync.py used to
+# perform with a client-side write token. Each endpoint accepts a batch,
+# validates per-item, and produces a single HF commit so we stay well
+# inside HF API rate limits.
+
+@app.post('/contribute/bulk-crops')
+async def contribute_bulk_crops(req: BulkCropsRequest, request: Request):
+    """Accept a batch of confirmed crops + annotations.
+
+    Mirrors warp/trainer/sync.py:_upload(): writes PNGs to
+    staging/<install_id>/crops/<sha>.png and rewrites
+    staging/<install_id>/annotations.jsonl with last-wins dedup per sha.
+    All writes happen in a single HF commit.
+    """
+    client_ip = _get_client_ip(request)
+    if not await _check_and_increment_rate_limit(client_ip):
+        raise HTTPException(429, 'Rate limit exceeded. Try again tomorrow.')
+
+    install_id = req.install_id.strip()
+    if not _INSTALL_ID_RE.match(install_id):
+        raise HTTPException(400, 'Invalid install_id format')
+
+    today = date.today().isoformat()
+    staging_dir   = f'staging/{install_id}'
+    staging_crop  = f'{staging_dir}/crops'
+    staging_anno  = f'{staging_dir}/annotations.jsonl'
+
+    files_to_upload: dict[str, bytes] = {}
+    new_entries:    list[dict]        = []
+    accepted   = 0
+    rejected   = 0
+    reasons:   list[str] = []
+
+    for item in req.items:
+        name = item.name.strip()
+        slot = item.slot.strip()
+        if not name or not slot:
+            rejected += 1
+            reasons.append('empty name/slot')
+            continue
+        if not name.isprintable():
+            rejected += 1
+            reasons.append('non-printable name')
+            continue
+        if name.startswith('__') or name == 'Test Item Name':
+            rejected += 1
+            reasons.append(f'poison label {name!r}')
+            continue
+
+        try:
+            png_bytes = base64.b64decode(item.crop_png_b64)
+        except Exception:
+            rejected += 1
+            reasons.append('b64 decode failed')
+            continue
+        if not png_bytes.startswith(b'\x89PNG'):
+            rejected += 1
+            reasons.append('not a PNG')
+            continue
+        if len(png_bytes) > MAX_CROP_PNG_BYTES:
+            rejected += 1
+            reasons.append(f'PNG too large ({len(png_bytes)} B)')
+            continue
+
+        err = _check_crop_dims(png_bytes, slot)
+        if err:
+            rejected += 1
+            reasons.append(err)
+            continue
+
+        sha = hashlib.sha256(png_bytes).hexdigest()[:32]
+        crop_path = f'{staging_crop}/{sha}.png'
+        files_to_upload[crop_path] = png_bytes
+
+        entry = {
+            'slot':        slot,
+            'name':        name,
+            'crop_sha256': sha,
+            'date':        today,
+        }
+        if item.ml_name:
+            entry['ml_name'] = item.ml_name
+        new_entries.append(entry)
+        accepted += 1
+
+    if not new_entries:
+        raise HTTPException(400, f'All {rejected} items rejected: {reasons[:3]}')
+
+    # Merge annotations.jsonl with last-wins per sha (matches client logic).
+    merged_jsonl = _merge_annotations_jsonl(install_id, new_entries)
+    files_to_upload[staging_anno] = merged_jsonl
+
+    ok = _hf_upload_files(
+        files_to_upload,
+        message=f'WARP bulk: {accepted} crops + annotations ({today})',
+        repo_id=HF_ICONS_REPO_ID,
+    )
+    if not ok:
+        raise HTTPException(503, 'Storage unavailable, please try later')
+
+    log.info(f'Bulk crops accepted: install={install_id[:8]} accepted={accepted} rejected={rejected}')
+    return {'ok': True, 'accepted': accepted, 'rejected': rejected,
+            'rejected_reasons': reasons[:10] if rejected else []}
+
+
+@app.post('/upload/screen-types')
+async def upload_screen_types(req: ScreenTypesRequest, request: Request):
+    """Accept a batch of screen-type screenshots for one screen_type label."""
+    client_ip = _get_client_ip(request)
+    if not await _check_and_increment_rate_limit(client_ip):
+        raise HTTPException(429, 'Rate limit exceeded. Try again tomorrow.')
+
+    install_id  = req.install_id.strip()
+    if not _INSTALL_ID_RE.match(install_id):
+        raise HTTPException(400, 'Invalid install_id format')
+    stype = req.screen_type.strip()
+    if not _SCREEN_TYPE_RE.match(stype):
+        raise HTTPException(400, 'Invalid screen_type')
+
+    base_dir = f'staging/{install_id}/screen_types/{stype}'
+    files_to_upload: dict[str, bytes] = {}
+    accepted = 0
+    rejected = 0
+    reasons: list[str] = []
+
+    for item in req.items:
+        try:
+            png_bytes = base64.b64decode(item.png_b64)
+        except Exception:
+            rejected += 1
+            reasons.append('b64 decode failed')
+            continue
+        if not png_bytes.startswith(b'\x89PNG'):
+            rejected += 1
+            reasons.append('not a PNG')
+            continue
+        if len(png_bytes) > MAX_SCREEN_PNG_BYTES:
+            rejected += 1
+            reasons.append(f'PNG too large ({len(png_bytes)} B)')
+            continue
+
+        sha = hashlib.sha256(png_bytes).hexdigest()[:32]
+        files_to_upload[f'{base_dir}/{sha}.png'] = png_bytes
+        accepted += 1
+
+    if not files_to_upload:
+        raise HTTPException(400, f'All {rejected} items rejected: {reasons[:3]}')
+
+    ok = _hf_upload_files(
+        files_to_upload,
+        message=f'WARP screen types: {accepted} {stype} screenshots',
+        repo_id=HF_ICONS_REPO_ID,
+    )
+    if not ok:
+        raise HTTPException(503, 'Storage unavailable, please try later')
+
+    log.info(f'Screen types accepted: install={install_id[:8]} type={stype} accepted={accepted} rejected={rejected}')
+    return {'ok': True, 'accepted': accepted, 'rejected': rejected}
+
+
+@app.post('/upload/anchors')
+async def upload_anchors(req: AnchorsRequest, request: Request):
+    """Accept a batch of anchor grids (one file per grid, keyed by sha8)."""
+    client_ip = _get_client_ip(request)
+    if not await _check_and_increment_rate_limit(client_ip):
+        raise HTTPException(429, 'Rate limit exceeded. Try again tomorrow.')
+
+    install_id = req.install_id.strip()
+    if not _INSTALL_ID_RE.match(install_id):
+        raise HTTPException(400, 'Invalid install_id format')
+
+    base_dir = f'staging/{install_id}'
+    files_to_upload: dict[str, bytes] = {}
+    accepted = 0
+    rejected = 0
+    reasons: list[str] = []
+
+    for grid in req.grids:
+        slots = grid.slots
+        if len(slots) < 3:
+            rejected += 1
+            reasons.append('fewer than 3 slots')
+            continue
+        # Each bbox must be [x, y, w, h] (4 floats); guard against malformed payloads.
+        if any(len(v) != 4 for v in slots.values()):
+            rejected += 1
+            reasons.append('bbox not 4-tuple')
+            continue
+
+        payload = {
+            'build_type': grid.build_type,
+            'aspect':     grid.aspect,
+            'resolution': grid.resolution,
+            'slots':      slots,
+        }
+        # sort_keys=True must match the client's canonical form so hashes
+        # line up and we don't duplicate the same grid as a different file.
+        payload_json = json.dumps(payload, sort_keys=True, ensure_ascii=False)
+        sha8 = hashlib.sha256(payload_json.encode()).hexdigest()[:8]
+        files_to_upload[f'{base_dir}/anchors_grid_{sha8}.json'] = payload_json.encode('utf-8')
+        accepted += 1
+
+    if not files_to_upload:
+        raise HTTPException(400, f'All {rejected} grids rejected: {reasons[:3]}')
+
+    ok = _hf_upload_files(
+        files_to_upload,
+        message=f'WARP anchors: {accepted} grid entries',
+        repo_id=HF_ICONS_REPO_ID,
+    )
+    if not ok:
+        raise HTTPException(503, 'Storage unavailable, please try later')
+
+    log.info(f'Anchors accepted: install={install_id[:8]} accepted={accepted} rejected={rejected}')
+    return {'ok': True, 'accepted': accepted, 'rejected': rejected}
+
+
 @app.post('/webhooks/hf-dataset')
 async def hf_dataset_webhook(request: Request):
     """
@@ -319,6 +604,62 @@ async def admin_merge(
 
 # ── Validation helpers ─────────────────────────────────────────────────────────
 
+def _check_crop_dims(png_bytes: bytes, slot: str) -> str | None:
+    """Validate crop image dimensions. Returns None if OK, else error message.
+
+    Text crops (ship_type_/ship_tier_) are wide horizontal bands and use
+    relaxed height/width minimums instead of the square icon minimum.
+    """
+    try:
+        nparr = np.frombuffer(png_bytes, np.uint8)
+        img = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
+        if img is None:
+            return 'unreadable image'
+        h, w = img.shape[:2]
+        if any(slot.startswith(p) for p in _TEXT_CROP_PREFIXES):
+            if h < MIN_TEXT_CROP_H or w < MIN_TEXT_CROP_W:
+                return f'too small ({w}x{h})'
+        else:
+            if h < MIN_CROP_PX or w < MIN_CROP_PX:
+                return f'too small ({w}x{h})'
+        return None
+    except Exception as e:
+        return str(e)
+
+
+def _merge_annotations_jsonl(install_id: str, new_entries: list[dict]) -> bytes:
+    """Fetch existing staging annotations.jsonl, merge with last-wins per sha.
+
+    Mirrors warp/trainer/sync.py:_append_staging_annotations_to_ops so that
+    the central pipeline sees the same per-sha dedup behaviour regardless of
+    whether the client uploaded directly (legacy) or via this backend.
+    """
+    existing_lines = _fetch_staging_annotations(install_id)
+
+    override_shas = {e.get('crop_sha256', '') for e in new_entries if e.get('crop_sha256')}
+    kept: list[str] = []
+    for line in existing_lines:
+        try:
+            sha = json.loads(line).get('crop_sha256', '')
+        except Exception:
+            kept.append(line)
+            continue
+        if sha and sha in override_shas:
+            continue
+        kept.append(line)
+
+    # Dedup within new_entries themselves (last wins) so a single batch
+    # with duplicate sha doesn't write two conflicting labels.
+    seen: dict[str, dict] = {}
+    for e in new_entries:
+        sha = e.get('crop_sha256', '')
+        if sha:
+            seen[sha] = e
+
+    combined = kept + [json.dumps(e, ensure_ascii=False) for e in seen.values()]
+    return '\n'.join(combined).encode('utf-8')
+
+
 def is_valid_crop(png_bytes: bytes) -> bool:
     """Checks if crop is valid (not garbage/too uniform)."""
     try:
@@ -334,10 +675,20 @@ def is_valid_crop(png_bytes: bytes) -> bool:
 
 # ── HF Dataset helpers ────────────────────���────────────────────────────────────
 
-def _hf_upload_files(files: dict[str, bytes], message: str = 'WARP auto-upload') -> bool:
-    """Upload multiple files to HF Dataset repo atomically."""
-    if not HF_TOKEN or not HF_REPO_ID:
-        log.error('HF_TOKEN or HF_REPO_ID not set')
+def _hf_upload_files(
+    files: dict[str, bytes],
+    message: str = 'WARP auto-upload',
+    repo_id: str | None = None,
+) -> bool:
+    """Upload multiple files to HF Dataset repo atomically.
+
+    `repo_id` defaults to HF_REPO_ID (knowledge dataset). Pass
+    HF_ICONS_REPO_ID for icon-dataset uploads (bulk crops, screen types,
+    anchor grids).
+    """
+    target = repo_id or HF_REPO_ID
+    if not HF_TOKEN or not target:
+        log.error('HF_TOKEN or repo_id not set')
         return False
     try:
         from huggingface_hub import HfApi, CommitOperationAdd
@@ -347,17 +698,42 @@ def _hf_upload_files(files: dict[str, bytes], message: str = 'WARP auto-upload')
             for path, content in files.items()
         ]
         api.create_commit(
-            repo_id=HF_REPO_ID,
+            repo_id=target,
             repo_type='dataset',
             operations=operations,
             commit_message=message,
         )
         return True
     except Exception as e:
-        log.error(f'HF atomic upload failed: {e}')
+        log.error(f'HF atomic upload failed (repo={target}): {e}')
         return False
 
 
+def _fetch_staging_annotations(install_id: str) -> list[str]:
+    """Fetch existing staging/<install_id>/annotations.jsonl lines from HF.
+
+    Returns the raw JSON lines (already stripped of trailing whitespace),
+    or [] if the file doesn't exist yet or download fails.
+    """
+    try:
+        from huggingface_hub import hf_hub_download
+        local = hf_hub_download(
+            repo_id=HF_ICONS_REPO_ID,
+            filename=f'staging/{install_id}/annotations.jsonl',
+            repo_type='dataset',
+            token=HF_TOKEN or None,
+        )
+        out: list[str] = []
+        with open(local, encoding='utf-8') as f:
+            for line in f:
+                line = line.strip()
+                if line:
+                    out.append(line)
+        return out
+    except Exception:
+        return []
+
+
 def _load_model_version_from_hf() -> dict:
     """Download models/model_version.json from HF."""
     if not HF_REPO_ID: