whatpull/neuronface

1"""Cloudflare D1 영속 저장 (커뮤니티 weight / exemplar 집계).2 3D1 단일 source of truth — SQLite fallback 없음. CF env 미설정 시 community 기능은4graceful skip (ok=False, reason='no-d1') 하며 학습/추론 자체에는 영향 없음.5 6ENV (필수, 미설정 시 community routes no-op):7  CF_ACCOUNT_ID, CF_API_TOKEN, CF_DATABASE_ID8 9테이블 (`nf_` prefix 로 namespacing):10  nf_community_weights (11    id, synapse_key, weight REAL, contributor_id, network_preset, created_at12  )13  nf_community_exemplars (14    id, out_key, feature_json, label, contributor_id, created_at15  )16  nf_meta (key TEXT PK, value TEXT)17"""18from __future__ import annotations19 20import json21import os22import time23from typing import Any, Dict, List, Optional, Tuple24 25try:26    import httpx  # type: ignore27except Exception:  # pragma: no cover28    httpx = None  # type: ignore[assignment]29 30CF_ACCOUNT_ID = os.getenv("CF_ACCOUNT_ID")31CF_API_TOKEN = os.getenv("CF_API_TOKEN")32CF_DATABASE_ID = os.getenv("CF_DATABASE_ID")33SCHEMA_VERSION = "1"34 35CF_D1_URL = (36    f"https://api.cloudflare.com/client/v4/accounts/{CF_ACCOUNT_ID}"37    f"/d1/database/{CF_DATABASE_ID}/query"38    if CF_ACCOUNT_ID and CF_DATABASE_ID39    else None40)41 42 43def _has_cf() -> bool:44    return bool(CF_ACCOUNT_ID and CF_API_TOKEN and CF_DATABASE_ID and httpx is not None)45 46 47def _no_cf_result(reason: str = "no-d1") -> Dict[str, Any]:48    return {"ok": False, "reason": reason}49 50 51# ---------------------------------------------------------------------------52# Cloudflare D1 (HTTP).53# ---------------------------------------------------------------------------54 55def _cf_query(sql: str, params: Optional[List[Any]] = None) -> Optional[Dict[str, Any]]:56    if not _has_cf():57        return None58    headers = {59        "Authorization": f"Bearer {CF_API_TOKEN}",60        "Content-Type": "application/json",61    }62    payload: Dict[str, Any] = {"sql": sql}63    if params:64        payload["params"] = params65    try:66        with httpx.Client(timeout=30.0) as client:  # type: ignore[union-attr]67            r = client.post(CF_D1_URL, headers=headers, json=payload)68            data = r.json()69            if not data.get("success"):70                print(f"[d1] query failed: {data}")71                return None72            results = data.get("result", [])73            return results[0] if results else None74    except Exception as e:  # pragma: no cover75        print(f"[d1] HTTP error: {e}")76        return None77 78 79# ---------------------------------------------------------------------------80# Schema.81# ---------------------------------------------------------------------------82 83SCHEMA_STATEMENTS: List[str] = [84    """CREATE TABLE IF NOT EXISTS nf_community_weights (85        id INTEGER PRIMARY KEY AUTOINCREMENT,86        synapse_key TEXT NOT NULL,87        weight REAL NOT NULL,88        contributor_id TEXT,89        network_preset TEXT,90        created_at TEXT NOT NULL91    )""",92    "CREATE INDEX IF NOT EXISTS idx_nf_weights_key ON nf_community_weights(synapse_key)",93    "CREATE INDEX IF NOT EXISTS idx_nf_weights_preset ON nf_community_weights(network_preset)",94    """CREATE TABLE IF NOT EXISTS nf_community_exemplars (95        id INTEGER PRIMARY KEY AUTOINCREMENT,96        out_key TEXT NOT NULL,97        feature_json TEXT NOT NULL,98        label TEXT,99        contributor_id TEXT,100        created_at TEXT NOT NULL101    )""",102    "CREATE INDEX IF NOT EXISTS idx_nf_exemplars_out ON nf_community_exemplars(out_key)",103    """CREATE TABLE IF NOT EXISTS nf_meta (104        key TEXT PRIMARY KEY,105        value TEXT NOT NULL106    )""",107]108 109 110def init_schema() -> Dict[str, Any]:111    """앱 부팅 시 1회 — D1 스키마 보강. CF env 없으면 no-op."""112    if not _has_cf():113        return {"ok": False, "backend": "none", "reason": "no-d1-env"}114    for sql in SCHEMA_STATEMENTS:115        _cf_query(sql)116    _cf_query(117        "INSERT OR REPLACE INTO nf_meta (key, value) VALUES (?, ?)",118        ["schema_version", SCHEMA_VERSION],119    )120    return {"ok": True, "backend": "cloudflare-d1", "schema_version": SCHEMA_VERSION}121 122 123def list_tables() -> List[str]:124    """D1 의 모든 테이블 — 운영자 cleanup 검토용."""125    if not _has_cf():126        return []127    r = _cf_query("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")128    if not r or "results" not in r:129        return []130    return [row["name"] for row in r["results"] if not row["name"].startswith("sqlite_")]131 132 133def drop_table(name: str) -> Dict[str, Any]:134    """미사용 테이블 정리용."""135    if not _has_cf():136        return _no_cf_result()137    if not name or "'" in name or '"' in name or ";" in name:138        return {"ok": False, "reason": "invalid table name"}139    r = _cf_query(f"DROP TABLE IF EXISTS {name}")140    return {"ok": r is not None, "table": name}141 142 143# ---------------------------------------------------------------------------144# Weight contribute / aggregate.145# ---------------------------------------------------------------------------146 147# 단일 multi-row INSERT 의 row 한도 — D1 SQL/param 한도 회피 (5 param × 200 = 1000).148BATCH_ROWS = 200149 150 151def contribute_weights(152    synapses: List[Dict[str, Any]],153    contributor_id: Optional[str] = None,154    network_preset: str = "feature16",155) -> Dict[str, Any]:156    """List[{pre, post, weight}] → nf_community_weights batch insert.157 158    Multi-row VALUES 청크 분할로 시냅스당 1 query 패턴 회피 (D1 free tier159    100k writes/day 빠른 소진 방지).160    """161    if not _has_cf():162        return _no_cf_result()163    if not synapses:164        return {"ok": True, "inserted": 0}165 166    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())167    rows: List[Tuple[Any, ...]] = []168    for s in synapses:169        if "pre" not in s or "post" not in s or "weight" not in s:170            continue171        rows.append((172            f"{s['pre']}|{s['post']}",173            float(s["weight"]),174            contributor_id,175            network_preset,176            now,177        ))178    if not rows:179        return {"ok": True, "inserted": 0}180 181    inserted = 0182    for i in range(0, len(rows), BATCH_ROWS):183        chunk = rows[i:i + BATCH_ROWS]184        placeholders = ",".join(["(?, ?, ?, ?, ?)"] * len(chunk))185        sql = (186            "INSERT INTO nf_community_weights "187            "(synapse_key, weight, contributor_id, network_preset, created_at) "188            f"VALUES {placeholders}"189        )190        params: List[Any] = []191        for row in chunk:192            params.extend(row)193        if _cf_query(sql, params) is None:194            return {"ok": False, "inserted": inserted, "reason": "cf-batch-failed"}195        inserted += len(chunk)196    return {"ok": True, "inserted": inserted}197 198 199def aggregate_weights(200    network_preset: str = "feature16",201    min_contributors: int = 1,202) -> Dict[str, Any]:203    """동일 synapse_key 의 평균 weight + 기여자 수 — preset 필터링."""204    if not _has_cf():205        return {"ok": False, "synapses": [], "count": 0, "reason": "no-d1"}206    sql = (207        "SELECT synapse_key, AVG(weight) AS avg_weight, "208        "COUNT(DISTINCT contributor_id) AS contributors, COUNT(*) AS samples "209        "FROM nf_community_weights "210        "WHERE network_preset = ? "211        "GROUP BY synapse_key "212        "HAVING COUNT(DISTINCT contributor_id) >= ?"213    )214    r = _cf_query(sql, [network_preset, min_contributors])215    rows = (r or {}).get("results", []) if r else []216    synapses: List[Dict[str, Any]] = []217    for row in rows:218        key = row["synapse_key"]219        if "|" not in key:220            continue221        pre, post = key.split("|", 1)222        synapses.append({223            "pre": pre,224            "post": post,225            "weight": round(float(row["avg_weight"]), 4),226            "contributors": int(row["contributors"]),227            "samples": int(row["samples"]),228        })229    return {"ok": True, "synapses": synapses, "count": len(synapses)}230 231 232# ---------------------------------------------------------------------------233# Exemplar contribute / list.234# ---------------------------------------------------------------------------235 236def contribute_exemplar(237    out_key: str,238    feature: List[float],239    label: Optional[str] = None,240    contributor_id: Optional[str] = None,241) -> Dict[str, Any]:242    if not _has_cf():243        return _no_cf_result()244    if not out_key:245        return {"ok": False, "reason": "out_key required"}246    if not isinstance(feature, list) or not feature:247        return {"ok": False, "reason": "feature must be non-empty list"}248    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())249    sql = (250        "INSERT INTO nf_community_exemplars "251        "(out_key, feature_json, label, contributor_id, created_at) "252        "VALUES (?, ?, ?, ?, ?)"253    )254    params = [out_key, json.dumps(feature), label, contributor_id, now]255    r = _cf_query(sql, params)256    return {"ok": r is not None}257 258 259def list_exemplars(out_key: Optional[str] = None, limit: int = 100) -> Dict[str, Any]:260    if not _has_cf():261        return {"ok": False, "exemplars": [], "count": 0, "reason": "no-d1"}262    if out_key:263        sql = (264            "SELECT out_key, feature_json, label, contributor_id, created_at "265            "FROM nf_community_exemplars WHERE out_key = ? "266            "ORDER BY created_at DESC LIMIT ?"267        )268        params = [out_key, int(limit)]269    else:270        sql = (271            "SELECT out_key, feature_json, label, contributor_id, created_at "272            "FROM nf_community_exemplars "273            "ORDER BY created_at DESC LIMIT ?"274        )275        params = [int(limit)]276    r = _cf_query(sql, params)277    rows = (r or {}).get("results", []) if r else []278    out: List[Dict[str, Any]] = []279    for row in rows:280        try:281            feat = json.loads(row["feature_json"])282        except Exception:283            feat = []284        out.append({285            "out_key": row["out_key"],286            "feature": feat,287            "label": row.get("label"),288            "contributor_id": row.get("contributor_id"),289            "created_at": row.get("created_at"),290        })291    return {"ok": True, "exemplars": out, "count": len(out)}292 293 294def stats() -> Dict[str, Any]:295    """대시보드용 — 테이블 row count + contributor 수."""296    if not _has_cf():297        return {"backend": "none", "reason": "no-d1"}298    out: Dict[str, Any] = {"backend": "cloudflare-d1"}299    queries = [300        ("weights_total", "SELECT COUNT(*) AS c FROM nf_community_weights"),301        ("weights_contributors", "SELECT COUNT(DISTINCT contributor_id) AS c FROM nf_community_weights"),302        ("exemplars_total", "SELECT COUNT(*) AS c FROM nf_community_exemplars"),303        ("exemplars_per_out",304         "SELECT out_key, COUNT(*) AS c FROM nf_community_exemplars GROUP BY out_key"),305    ]306    for label, sql in queries:307        r = _cf_query(sql) or {}308        rows = r.get("results", [])309        if label == "exemplars_per_out":310            out[label] = {row["out_key"]: int(row["c"]) for row in rows}311        else:312            out[label] = int(rows[0]["c"]) if rows else 0313    return out314 315 316# ---------------------------------------------------------------------------317# PatternKey — pk_ prefix 테이블 (nf_ 와 분리).318# ---------------------------------------------------------------------------319 320PK_SCHEMA_STATEMENTS: List[str] = [321    """CREATE TABLE IF NOT EXISTS pk_api_keys (322        id TEXT PRIMARY KEY,323        key_hash TEXT NOT NULL UNIQUE,324        key_prefix TEXT NOT NULL,325        owner_email TEXT,326        plan TEXT NOT NULL DEFAULT 'free',327        monthly_limit INTEGER NOT NULL DEFAULT 999999,328        allowed_origins TEXT DEFAULT '*',329        created_at TEXT NOT NULL DEFAULT (datetime('now'))330    )""",331    "CREATE INDEX IF NOT EXISTS idx_pk_api_keys_hash ON pk_api_keys(key_hash)",332    """CREATE TABLE IF NOT EXISTS pk_sessions (333        id TEXT PRIMARY KEY,334        api_key_id TEXT NOT NULL,335        token TEXT NOT NULL UNIQUE,336        expires_at TEXT NOT NULL,337        used INTEGER NOT NULL DEFAULT 0,338        created_at TEXT NOT NULL DEFAULT (datetime('now'))339    )""",340    "CREATE INDEX IF NOT EXISTS idx_pk_sessions_token ON pk_sessions(token)",341    """CREATE TABLE IF NOT EXISTS pk_usage_logs (342        id TEXT PRIMARY KEY,343        api_key_id TEXT NOT NULL,344        action TEXT NOT NULL,345        result TEXT NOT NULL,346        created_at TEXT NOT NULL DEFAULT (datetime('now'))347    )""",348    "CREATE INDEX IF NOT EXISTS idx_pk_usage_logs_key ON pk_usage_logs(api_key_id)",349    "CREATE INDEX IF NOT EXISTS idx_pk_usage_logs_created ON pk_usage_logs(created_at)",350]351 352 353def pk_init_schema() -> bool:354    """PatternKey 테이블 초기화 (idempotent). CF env 없으면 False."""355    if not _has_cf():356        return False357    for sql in PK_SCHEMA_STATEMENTS:358        if _cf_query(sql) is None:359            print(f"[d1/pk] schema statement failed: {sql[:60]}")360    return True361 362 363def pk_get_api_key(key_hash: str) -> Optional[Dict[str, Any]]:364    """key_hash 로 pk_api_keys 행 조회. 없으면 None."""365    if not _has_cf():366        return None367    r = _cf_query(368        "SELECT id, key_hash, key_prefix, owner_email, plan, monthly_limit, "369        "allowed_origins, created_at FROM pk_api_keys WHERE key_hash = ?",370        [key_hash],371    )372    rows = (r or {}).get("results", [])373    return rows[0] if rows else None374 375 376async def pk_create_api_key(377    key_id: str,378    key_hash: str,379    key_prefix: str,380    owner_email: Optional[str] = None,381    plan: str = "free",382    monthly_limit: int = 999999,383    allowed_origins: str = "*",384) -> bool:385    """pk_api_keys INSERT. 성공 시 True."""386    if not _has_cf():387        return False388    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())389    r = _cf_query(390        "INSERT INTO pk_api_keys "391        "(id, key_hash, key_prefix, owner_email, plan, monthly_limit, allowed_origins, created_at) "392        "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",393        [key_id, key_hash, key_prefix, owner_email, plan, monthly_limit, allowed_origins, now],394    )395    return r is not None396 397 398def pk_create_session(api_key_id: str, token: str, expires_at: str) -> bool:399    """pk_sessions INSERT (1회용 세션). 성공 시 True."""400    if not _has_cf():401        return False402    import uuid403    session_id = str(uuid.uuid4())404    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())405    r = _cf_query(406        "INSERT INTO pk_sessions (id, api_key_id, token, expires_at, used, created_at) "407        "VALUES (?, ?, ?, ?, 0, ?)",408        [session_id, api_key_id, token, expires_at, now],409    )410    return r is not None411 412 413def pk_verify_session(token: str) -> Optional[Dict[str, Any]]:414    """세션 검증 + 1회용 소진 (used=1 업데이트).415 416    반환값: {'api_key_id': ..., 'expires_at': ...} 또는 None (무효/만료/소진).417    """418    if not _has_cf():419        return None420    r = _cf_query(421        "SELECT id, api_key_id, expires_at, used FROM pk_sessions WHERE token = ?",422        [token],423    )424    rows = (r or {}).get("results", [])425    if not rows:426        return None427    row = rows[0]428    if row.get("used", 1):429        return None430    # TTL 확인: expires_at 이 현재 UTC 보다 이전이면 만료.431    now_str = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())432    if row["expires_at"] < now_str:433        return None434    # used=1 마킹.435    _cf_query(436        "UPDATE pk_sessions SET used = 1 WHERE id = ?",437        [row["id"]],438    )439    return {"api_key_id": row["api_key_id"], "expires_at": row["expires_at"]}440 441 442def pk_log_usage(api_key_id: str, action: str, result: str) -> None:443    """pk_usage_logs INSERT (fire-and-forget)."""444    if not _has_cf():445        return446    import uuid447    log_id = str(uuid.uuid4())448    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())449    _cf_query(450        "INSERT INTO pk_usage_logs (id, api_key_id, action, result, created_at) "451        "VALUES (?, ?, ?, ?, ?)",452        [log_id, api_key_id, action, result, now],453    )454 455 456def pk_monthly_count(api_key_id: str) -> int:457    """당월 pk_usage_logs 건수 (action='session_create' 기준)."""458    if not _has_cf():459        return 0460    year_month = time.strftime("%Y-%m", time.gmtime())461    r = _cf_query(462        "SELECT COUNT(*) AS c FROM pk_usage_logs "463        "WHERE api_key_id = ? AND action = 'session_create' "464        "AND created_at LIKE ?",465        [api_key_id, f"{year_month}%"],466    )467    rows = (r or {}).get("results", [])468    return int(rows[0]["c"]) if rows else 0469 470 471def pk_update_plan(key_id: str, plan: str, monthly_limit: int) -> bool:472    """pk_api_keys 의 plan + monthly_limit 업데이트. 성공 시 True."""473    if not _has_cf():474        return False475    r = _cf_query(476        "UPDATE pk_api_keys SET plan=?, monthly_limit=? WHERE id=?",477        [plan, monthly_limit, key_id],478    )479    return r is not None480 481 482async def pk_list_api_keys() -> List[Dict[str, Any]]:483    """전체 API 키 목록 (key_hash 제외, 관리자용)."""484    if not _has_cf():485        return []486    r = _cf_query(487        "SELECT id, key_prefix, owner_email, plan, monthly_limit, "488        "allowed_origins, created_at FROM pk_api_keys ORDER BY created_at DESC"489    )490    return (r or {}).get("results", []) if r else []491 492 493async def pk_list_logs(page: int = 1, page_size: int = 100) -> Dict[str, Any]:494    """사용량 로그 목록 (페이지네이션, 관리자용)."""495    if not _has_cf():496        return {"items": [], "total": 0, "page": page, "page_size": page_size}497    offset = (page - 1) * page_size498    r_rows = _cf_query(499        "SELECT id, api_key_id, action, result, created_at "500        "FROM pk_usage_logs ORDER BY created_at DESC LIMIT ? OFFSET ?",501        [page_size, offset],502    )503    r_count = _cf_query("SELECT COUNT(*) AS cnt FROM pk_usage_logs")504    rows = (r_rows or {}).get("results", []) if r_rows else []505    count_rows = (r_count or {}).get("results", []) if r_count else []506    total = int(count_rows[0]["cnt"]) if count_rows else 0507    return {"items": rows, "total": total, "page": page, "page_size": page_size}508 509 510async def pk_delete_api_key(key_id: str) -> bool:511    """pk_api_keys DELETE. 성공 시 True."""512    return _cf_query("DELETE FROM pk_api_keys WHERE id=?", [key_id]) is not None513 514 515# ---------------------------------------------------------------------------516# PatternKey Beta Signup — pk_beta_requests 테이블.517# ---------------------------------------------------------------------------518 519PK_BETA_SCHEMA_STATEMENTS: List[str] = [520    """CREATE TABLE IF NOT EXISTS pk_beta_requests (521        id TEXT PRIMARY KEY,522        name TEXT NOT NULL,523        email TEXT NOT NULL UNIQUE,524        company TEXT,525        use_case TEXT,526        status TEXT NOT NULL DEFAULT 'pending',527        api_key_id TEXT,528        created_at TEXT NOT NULL DEFAULT (datetime('now')),529        reviewed_at TEXT530    )""",531    "CREATE INDEX IF NOT EXISTS idx_beta_status ON pk_beta_requests(status)",532]533 534 535def pk_beta_init_schema() -> bool:536    """pk_beta_requests 테이블 초기화 (idempotent). CF env 없으면 False."""537    if not _has_cf():538        return False539    for sql in PK_BETA_SCHEMA_STATEMENTS:540        if _cf_query(sql) is None:541            print(f"[d1/pk-beta] schema statement failed: {sql[:60]}")542    return True543 544 545def pk_beta_signup(546    id: str,547    name: str,548    email: str,549    company: Optional[str] = None,550    use_case: Optional[str] = None,551) -> bool:552    """베타 신청 INSERT. 성공 시 True. UNIQUE 충돌(중복 email) 시 False."""553    if not _has_cf():554        return False555    r = _cf_query(556        "INSERT INTO pk_beta_requests (id, name, email, company, use_case) "557        "VALUES (?, ?, ?, ?, ?)",558        [id, name, email, company, use_case],559    )560    return r is not None561 562 563def pk_beta_list(status: Optional[str] = None) -> List[Dict[str, Any]]:564    """관리자용 베타 신청 목록 조회. status 필터 옵션."""565    if not _has_cf():566        return []567    if status:568        r = _cf_query(569            "SELECT id, name, email, company, use_case, status, api_key_id, "570            "created_at, reviewed_at FROM pk_beta_requests WHERE status = ? "571            "ORDER BY created_at DESC",572            [status],573        )574    else:575        r = _cf_query(576            "SELECT id, name, email, company, use_case, status, api_key_id, "577            "created_at, reviewed_at FROM pk_beta_requests "578            "ORDER BY created_at DESC",579        )580    return (r or {}).get("results", []) if r else []581 582 583def pk_beta_approve(request_id: str, api_key_id: str) -> bool:584    """베타 신청 승인 — status='approved', api_key_id 저장, reviewed_at 기록. 성공 시 True."""585    if not _has_cf():586        return False587    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())588    r = _cf_query(589        "UPDATE pk_beta_requests SET status='approved', api_key_id=?, reviewed_at=? "590        "WHERE id=?",591        [api_key_id, now, request_id],592    )593    return r is not None594 595 596def pk_beta_reject(request_id: str) -> bool:597    """베타 신청 거절 — status='rejected', reviewed_at 기록. 성공 시 True."""598    if not _has_cf():599        return False600    now = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())601    r = _cf_query(602        "UPDATE pk_beta_requests SET status='rejected', reviewed_at=? WHERE id=?",603        [now, request_id],604    )605    return r is not None606 607 608async def pk_beta_delete(request_id: str) -> bool:609    """베타 신청 DELETE. 성공 시 True."""610    return _cf_query(611        "DELETE FROM pk_beta_requests WHERE id=?", [request_id]612    ) is not None613 614 615def pk_beta_get_by_email(email: str) -> Optional[Dict[str, Any]]:616    """이메일로 베타 신청 조회 (중복 체크용). 없으면 None."""617    if not _has_cf():618        return None619    r = _cf_query(620        "SELECT id, name, email, company, use_case, status, api_key_id, "621        "created_at, reviewed_at FROM pk_beta_requests WHERE email = ?",622        [email],623    )624    rows = (r or {}).get("results", [])625    return rows[0] if rows else None626 627 628# ---------------------------------------------------------------------------629# HandFace Pattern Backup — hf_pattern_backups 테이블.630# ---------------------------------------------------------------------------631 632HF_BACKUP_SCHEMA = """633CREATE TABLE IF NOT EXISTS hf_pattern_backups (634  device_id TEXT PRIMARY KEY,635  exemplars TEXT NOT NULL DEFAULT '{}',636  pattern_count INTEGER NOT NULL DEFAULT 0,637  created_at TEXT NOT NULL DEFAULT (datetime('now')),638  updated_at TEXT NOT NULL DEFAULT (datetime('now'))639);640"""641 642 643def hf_backup_init_schema() -> bool:644    """hf_pattern_backups 테이블 초기화 (idempotent). CF env 없으면 False."""645    if not _has_cf():646        return False647    if _cf_query(HF_BACKUP_SCHEMA.strip()) is None:648        print("[d1/hf-backup] schema init failed")649    return True650 651 652async def hf_backup_save(device_id: str, exemplars_json: str, pattern_count: int) -> bool:653    """device_id 기반 UPSERT. 성공 시 True."""654    if not _has_cf():655        return False656    r = _cf_query(657        """INSERT INTO hf_pattern_backups (device_id, exemplars, pattern_count)658           VALUES (?, ?, ?)659           ON CONFLICT(device_id) DO UPDATE SET660             exemplars=excluded.exemplars,661             pattern_count=excluded.pattern_count,662             updated_at=datetime('now')""",663        [device_id, exemplars_json, pattern_count],664    )665    return r is not None666 667 668async def hf_backup_load(device_id: str) -> Optional[Dict[str, Any]]:669    """device_id 로 백업 행 조회. 없으면 None."""670    if not _has_cf():671        return None672    r = _cf_query(673        "SELECT device_id, exemplars, pattern_count, updated_at "674        "FROM hf_pattern_backups WHERE device_id=?",675        [device_id],676    )677    rows = (r or {}).get("results", [])678    return rows[0] if rows else None679 680 681async def hf_backup_delete(device_id: str) -> bool:682    """device_id 백업 행 DELETE. 성공 시 True."""683    if not _has_cf():684        return False685    r = _cf_query(686        "DELETE FROM hf_pattern_backups WHERE device_id=?",687        [device_id],688    )689    return r is not None690