Aggregate lightweight doc engagement proxies.
(user_id: str, days: int = 7)
| 1069 | |
| 1070 | |
| 1071 | def get_doc_engagement_stats(user_id: str, days: int = 7) -> Dict[str, Any]: |
| 1072 | """ |
| 1073 | Aggregate lightweight doc engagement proxies. |
| 1074 | """ |
| 1075 | since_timestamp = (datetime.now() - timedelta(days=max(1, int(days)))).isoformat(sep=" ") |
| 1076 | conn = get_connection() |
| 1077 | cursor = conn.cursor() |
| 1078 | cursor.execute( |
| 1079 | """ |
| 1080 | SELECT action, action_type, metadata, timestamp |
| 1081 | FROM behavior_logs |
| 1082 | WHERE user_id = ? |
| 1083 | AND timestamp >= ? |
| 1084 | AND ( |
| 1085 | (action = 'opened_report' AND action_type = 'doc_open') |
| 1086 | OR (action = 'doc_dwell_proxy' AND action_type = 'doc_engagement') |
| 1087 | ) |
| 1088 | ORDER BY timestamp ASC, id ASC |
| 1089 | """, |
| 1090 | (user_id, since_timestamp), |
| 1091 | ) |
| 1092 | rows = cursor.fetchall() |
| 1093 | conn.close() |
| 1094 | |
| 1095 | unique_docs = set() |
| 1096 | total_opens = 0 |
| 1097 | dwell_values: List[float] = [] |
| 1098 | for row in rows: |
| 1099 | metadata = _load_json_metadata(row["metadata"]) |
| 1100 | doc_key = _normalize_identifier(metadata.get("doc_token")) or _normalize_identifier(metadata.get("doc_url")) |
| 1101 | if row["action"] == "opened_report": |
| 1102 | total_opens += 1 |
| 1103 | if doc_key: |
| 1104 | unique_docs.add(doc_key) |
| 1105 | elif row["action"] == "doc_dwell_proxy": |
| 1106 | try: |
| 1107 | dwell_seconds = float(metadata.get("dwell_seconds") or 0.0) |
| 1108 | except (TypeError, ValueError): |
| 1109 | dwell_seconds = 0.0 |
| 1110 | if dwell_seconds > 0: |
| 1111 | dwell_values.append(dwell_seconds) |
| 1112 | |
| 1113 | average_dwell_seconds = round(sum(dwell_values) / len(dwell_values), 2) if dwell_values else 0.0 |
| 1114 | return { |
| 1115 | "total_doc_opens": total_opens, |
| 1116 | "unique_doc_opens": len(unique_docs), |
| 1117 | "avg_dwell_proxy_seconds": average_dwell_seconds, |
| 1118 | "dwell_proxy_count": len(dwell_values), |
| 1119 | } |
| 1120 | |
| 1121 | |
| 1122 | def get_behavior_logs(user_id: str, start_date: str, end_date: str) -> List[Dict]: |
no test coverage detected