Neo
d269ee6401
1. docs/ai/app2a_finance_area_system_prompt_20260422_v1.md (新建 · v1.2 生产版): - 基于 app2_finance V5.1 派生 - 板块 C 改"业态收入结构" · 板块 E 改"业态定位与对比" - 新增 H7 硬约束：业态特征引用必须紧跟 payload 真实数据 - H6 扩展区域级 6 类字段缺失降级（储值卡/分渠道现金流/现金流出/会员占比/按星期/日异常） - 经 3 次修正：v1"稀疏" → v1.1 纠正为业务真实 0/非 0 → v1.2 纠正为字段存在/整块缺失 - 已同步百炼控制台 APP ID 0ae965029bc54706bcff44f511ac716b 2. docs/ai/app2_finance_multi_app_design.md (新建 · v2 定稿): - 6 章 + 3 附录 · Q1-Q7 全部决策 · 6 阶段 28 项 checklist - 72 组合数据源支持度三档梳理（必须 / 业务级全店 / 字段存在 vs 整块缺失） - 2 套 prompt 拼接方案 · 2 个派生百炼 APP 策略 3. docs/audit/changes/2026-04-23__app2a_finance_area_integrated.md (新建): - 完整审计记录 · 13 高风险文件逐项注解 - 数据库变更 + 风险与回滚 + 验证方式 + 合规检查 4. docs/audit/audit_dashboard.md (刷新 · 135 条记录) 5. scripts/ab_test_app2a_area.py (新建): - 8 业态 × 3 轮 = 24 次采样评估含金量 - 自动检测 H1/H2/H3/H7 硬约束通过率 + seq11 三色灯分布 6. scripts/ab_to_cache.py (新建): - 复用 A/B 结果直接写 ai_cache · 绕开百炼预算验证 UI 端到端 A/B 实测 24/24 成功 · 12 条齐整率 100% · H1/H3/H7 100% · 达生产级。 Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
82 lines
2.6 KiB
Python
82 lines
2.6 KiB
Python
"""把 A/B 采样结果（export/ai-ab-test/round_v1_app2a_area/*.json）直接写入 ai_cache。
|
||
|
||
用途：E4 小程序 E2E 验证时，不重复消耗百炼预算即可填充 app2a 缓存。
|
||
|
||
用法：
|
||
# 默认：每业态取 round1 结果，共 8 个组合写入 cache
|
||
PYTHONIOENCODING=utf-8 .venv/Scripts/python.exe scripts/ab_to_cache.py
|
||
|
||
# 只写某个业态：
|
||
PYTHONIOENCODING=utf-8 .venv/Scripts/python.exe scripts/ab_to_cache.py --area vip
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, 'apps/backend')
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
|
||
from app.ai.cache_service import AICacheService
|
||
from app.ai.schemas import CacheTypeEnum
|
||
|
||
|
||
SITE_ID = 2790685415443269
|
||
TIME_DIMENSION = 'this_month'
|
||
AB_DIR = Path('export/ai-ab-test/round_v1_app2a_area')
|
||
AREAS = ('hall', 'hallA', 'hallB', 'hallC', 'vip', 'snooker', 'mahjong', 'ktv')
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument('--area', type=str, default=None, help='只写单个业态，不指定则全部')
|
||
ap.add_argument('--round', type=int, default=1, help='使用第几轮结果（默认 round 1）')
|
||
args = ap.parse_args()
|
||
|
||
target_areas = [args.area] if args.area else list(AREAS)
|
||
cache_svc = AICacheService()
|
||
ok = 0
|
||
skipped = 0
|
||
|
||
for area in target_areas:
|
||
path = AB_DIR / f'{area}_round{args.round}.json'
|
||
if not path.exists():
|
||
print(f'[SKIP] {area}: 文件不存在（{path}），可能 A/B 尚未跑到此业态')
|
||
skipped += 1
|
||
continue
|
||
with open(path, 'r', encoding='utf-8') as f:
|
||
result = json.load(f)
|
||
parsed = result.get('parsed')
|
||
if not isinstance(parsed, dict) and not isinstance(parsed, list):
|
||
print(f'[SKIP] {area}: parsed 字段异常')
|
||
skipped += 1
|
||
continue
|
||
# 标准化：result_json 应为 {insights: [...]} 或直接 [...]
|
||
if isinstance(parsed, list):
|
||
result_json = {'insights': parsed}
|
||
else:
|
||
result_json = parsed
|
||
|
||
target_id = f'{TIME_DIMENSION}__{area}'
|
||
cache_svc.write_cache(
|
||
cache_type=CacheTypeEnum.APP2A_FINANCE_AREA.value,
|
||
site_id=SITE_ID,
|
||
target_id=target_id,
|
||
result_json=result_json,
|
||
triggered_by='ab_replay',
|
||
score=None,
|
||
)
|
||
ok += 1
|
||
print(f'[OK] 写入 app2a_finance_area · {target_id} · {len(result_json.get("insights", []))} 条')
|
||
|
||
print()
|
||
print(f'=== 完成：{ok} 个写入 · {skipped} 个跳过 ===')
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|