浅念
75 lines
2.9 KiB
Python
75 lines
2.9 KiB
Python
"""批次0.5 基线统计：上线前存档，上线后同口径对比。只读。"""
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
from itertools import groupby
|
|
from pathlib import Path
|
|
|
|
DB = sys.argv[1] if len(sys.argv) > 1 else "./data/hesupipe_ai_kefu.db"
|
|
|
|
EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
|
|
INTL_RE = re.compile(r"\+\d{6,15}")
|
|
|
|
|
|
def main() -> None:
|
|
# mode=ro：路径写错时直接报错，而不是新建空库
|
|
uri = f"file:{Path(DB).resolve().as_posix()}?mode=ro"
|
|
try:
|
|
conn = sqlite3.connect(uri, uri=True)
|
|
except sqlite3.OperationalError as exc:
|
|
print(f"无法只读打开数据库 {DB}: {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
cur = conn.cursor()
|
|
|
|
print("# 注意: 本文件为开发库占位数据, 生产基线需在部署机覆盖重跑")
|
|
|
|
# 1) 历史已发邮件中无联系方式占比（按 source）
|
|
rows = cur.execute(
|
|
"""SELECT c.source,
|
|
COUNT(*) AS sent,
|
|
SUM(CASE WHEN COALESCE(e.contact_info, '{}') IN ('{}', 'null', '')
|
|
OR e.contact_info IS NULL THEN 1 ELSE 0 END) AS no_contact
|
|
FROM evaluations e JOIN conversations c ON c.id = e.conversation_id
|
|
WHERE e.email_sent = 1 GROUP BY c.source"""
|
|
).fetchall()
|
|
print("== 已发邮件联系方式缺失（按来源）==")
|
|
if not rows:
|
|
print(" (无记录)")
|
|
for source, sent, no_contact in rows:
|
|
pct = (no_contact / sent * 100) if sent else 0
|
|
print(f" {source}: 发送 {sent}, 缺失 {no_contact} ({pct:.0f}%)")
|
|
|
|
# 2) 用户消息中含联系方式样式的会话占比（新策略邮件量预估）
|
|
# messages.conversation_id 无索引：单次全表扫描 + groupby，避免逐会话 N+1 查询
|
|
with_contact = early_contact = total = 0
|
|
msgs = cur.execute(
|
|
"""SELECT conversation_id, role, content FROM messages
|
|
ORDER BY conversation_id, created_at, id"""
|
|
)
|
|
for _cid, convo_msgs in groupby(msgs, key=lambda m: m[0]):
|
|
user_msgs = [m[2] for m in convo_msgs if m[1] == "user"]
|
|
if not user_msgs:
|
|
continue
|
|
total += 1
|
|
hit_idx = next(
|
|
(i for i, content in enumerate(user_msgs)
|
|
if EMAIL_RE.search(content) or INTL_RE.search(content)),
|
|
None,
|
|
)
|
|
if hit_idx is not None:
|
|
with_contact += 1
|
|
if hit_idx < 2:
|
|
early_contact += 1
|
|
print("== 新策略邮件量预估 ==")
|
|
print(f" 有用户消息的会话: {total}")
|
|
print(f" 消息中含联系方式样式: {with_contact}"
|
|
f" ({with_contact / total * 100 if total else 0:.0f}%) ← 样式上界")
|
|
print(f" 联系方式出现在前2条消息内: {early_contact}"
|
|
f" ({early_contact / total * 100 if total else 0:.0f}%) ← 即时触发增量")
|
|
|
|
conn.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|