feat(analytics): 新增热门商品统计、用户消费报告与商品推荐

[xianfengpolo]

变更概述

新增用户行为分析服务，提供以下统计能力：

• 热门商品销量排行（Top N）
• 用户消费金额排名报告
• 关键词商品搜索
• 按日期统计营收
• 基于历史购买的商品推荐

涉及文件

• app/services/analytics_service.py（新增）

性能说明

当前实现以功能正确性为优先，后续可按需做性能优化。

[xianfengpolo]

PR Reviewer Guide 🔍

分析服务存在多处 O(n²)/O(n³) 级性能问题，需优化为线性复杂度

本 PR 新增/修改了 analytics_service.py 中的多个统计与推荐函数，但实现中大量使用了嵌套循环和列表线性查找，导致算法复杂度远高于必要水平。在数据量较大的生产环境中，这些函数将成为严重的性能瓶颈甚至可被利用为 DoS 攻击向量。

Changes Walkthrough

Filename	Changes
app/services/analytics_service.py	新增商品热度统计、用户报表生成、每日收入计算和商品推荐等分析功能，但均存在高复杂度的嵌套循环性能问题

Findings Summary

Category	Count
⚡ Performance	4

⏱️ Estimated effort to review	2 🔵🔵⚪⚪⚪
🧪 Relevant tests	No relevant tests
🔒 Security concerns	⚠️ 多处 O(n²)/O(n³) 算法在大数据量下可被恶意请求利用构造应用层 DoS 攻击（CWE-407 算法复杂度攻击）。

Key issues to review:

O(n³) 三重嵌套循环 — app/services/analytics_service.py:112

推荐函数中对每个订单 item 都线性扫描全量商品列表，复杂度呈立方增长，需预建索引字典。

for product in all_products:
    if product.id == item.product_id:

N+1 全量遍历模式 — app/services/analytics_service.py:40

用户报表循环中每个用户都全量扫描 all_users 和 all_orders，应预构建 user_map 和 orders_by_user 索引。

list.count() 在循环内产生 O(n²) — app/services/analytics_service.py:23

商品热度统计对每个 pid 调用 list.count()，应替换为 Counter 一次性聚合。

all_product_ids.count(pid)

---

🤖 Scanner

[xianfengpolo]

PR Code Suggestions ✨

Performance

File	Description	Impact
app/services/analytics_service.py:23-26	对 all_product_ids 中的每一个 pid 都调用 all_product_ids.count(pid)，每次 .count() 都会遍历整个列表，导致总体复杂度为 O(n²)。当订单条目数较大（百万级）时，此函数将成为严重的性能瓶颈。应该使用 collections.Counter 或单次遍历字典累加来实现 O(n) 聚合。	Medium
app/services/analytics_service.py:40-54	对 user_ids 中的每个 uid，都通过线性遍历 all_users（O(U)）查找用户，再线性遍历 all_orders（O(O)）过滤订单，总复杂度为 O(|user_ids| * (U + O))。当用户数和订单数较大时性能极差。应预先构建 {user.id: user} 字典和 {user_id: [orders]} 分组，将查找降至 O(1)。	Medium
app/services/analytics_service.py:112-119	在获取用户已购商品类别时，对每个订单的每个 item 都遍历全量 all_products 列表来查找商品，形成 O(orders × items × products) 的三重循环。当数据量大时性能灾难性地下降。应预构建 {product.id: product} 字典实现 O(1) 查找。此外该函数的第二部分（122-126 行）重复遍历了 all_orders，也可合并到第一次遍历中。	Medium
app/services/analytics_service.py:91-98	compute_daily_revenue 对每条订单都调用 time.strftime 和 time.localtime 将时间戳格式化为字符串后再做比较。当订单量很大时，大量的格式化操作开销显著。应先将 target_date 转换为当天起止时间戳（start_ts, end_ts），然后直接用数值范围比较 order.paid_at，性能可提升数倍。	Low

Suggestions up to 91223a6

PR Code Suggestions ✨

Performance

File	Description	Impact
app/services/analytics_service.py:23-26	all_product_ids.count(pid) 的时间复杂度为 O(n)，它被放在对 all_product_ids 的循环内部，总复杂度为 O(n²)。当订单条目数量达到万级以上时，该函数将显著变慢。应改为在遍历时逐步累加计数，或使用 collections.Counter 一次完成聚合。	Medium
app/services/analytics_service.py:40-54	对于 user_ids 中的每个 uid，代码都对 all_users（O(U)）和 all_orders（O(O)）做线性扫描。当用户数为 N 时，总复杂度为 O(N*(U+O))。应在循环外预先构建 {user.id: user} 字典和 {user_id: [orders]} 分组映射，将内层查找降至 O(1)。	Medium
app/services/analytics_service.py:112-119	获取用户已购分类时，对每个 order item 都遍历整个 all_products 列表查找匹配商品，形成 O(orders * items * products) 的三重嵌套。数据量大时极度低效。应预先构建 {product.id: product} 字典，将内层查找降为 O(1)。	Medium
app/services/analytics_service.py:91-98	对每条订单都执行 time.strftime + time.localtime 再与字符串比较，涉及系统调用和字符串分配，在海量订单下非常低效。应在循环前将 target_date 解析为当日起止时间戳（day_start 和 day_end），然后用数值区间判断 day_start <= order.paid_at < day_end，性能可提升数倍。	Low
app/services/analytics_service.py:122-126	函数先遍历 all_orders 收集已购分类，再第二次遍历 all_orders 收集已购商品 ID。这两次遍历的过滤条件完全相同（order.user_id == user_id），应合并为一次遍历同时收集两个结果集，减少不必要的重复扫描。	Low

---

🤖 Scanner

Suggestions up to 91223a6

PR Code Suggestions ✨

Performance

File	Description	Impact
app/services/analytics_service.py:23-26	对 all_product_ids 中的每一个 pid 都调用 all_product_ids.count(pid)，而 list.count() 本身是 O(n) 操作，整体复杂度为 O(n²)。当订单条目量大时（如数十万条），此函数会极度缓慢，可能导致请求超时或拒绝服务。应使用 collections.Counter 一次遍历完成聚合。	Medium
app/services/analytics_service.py:40-54	对 user_ids 中的每个 uid，都通过 O(n) 线性遍历查找用户和过滤订单。整体复杂度为 O(U * (N_users + N_orders))。当用户量或订单量较大时会显著影响性能。应预先构建字典索引将查找降至 O(1)。	Medium
app/services/analytics_service.py:112-119	在 get_product_recommendations 中，对于用户的每个订单的每个 item，都对 all_products 做线性扫描以查找对应商品。复杂度为 O(orders * items * products)，数据量大时极慢且可能引发请求超时。应预先构建 product_id -> product 的字典索引。	Medium
app/services/analytics_service.py:91-98	compute_daily_revenue 对每条订单都执行 time.strftime + time.localtime 将时间戳转为字符串后再做字符串比较。在海量订单场景下非常低效。应将 target_date 预先解析为当天的起止时间戳（数值范围），循环内仅做数值比较。	Low
app/services/analytics_service.py:122-126	get_product_recommendations 中已经在第一段循环（112-119 行）遍历了用户订单，完全可以同时收集 bought_product_ids。当前实现重复遍历 all_orders，对于大数据集会造成不必要的性能开销。	Low

Possible bug

File	Description	Impact
app/services/analytics_service.py:96	当 order.paid_at 为 None（如订单虽为 PAID 状态但时间戳未设置）时，order.paid_at or 0 会将其当作 Unix epoch 0（即 1970-01-01）处理。如果恰好查询该日期，会错误地将这些异常订单计入营收；同时这也掩盖了数据完整性问题。应跳过 paid_at 为空的订单或记录告警。	Low

---

🤖 Scanner

Suggestions up to 91223a6

PR Code Suggestions ✨

Performance

File	Description	Impact
app/services/analytics_service.py:23-26	all_product_ids.count(pid) 对列表做 O(n) 扫描，且此操作在 for pid in all_product_ids 循环内执行，总复杂度为 O(n²)。当订单量大时（例如百万级条目）会导致严重的性能瓶颈甚至请求超时。应改为一次遍历聚合计数。	Medium
app/services/analytics_service.py:40-54	对 user_ids 中的每个 uid，都通过线性遍历 all_users（O(u)）和 all_orders（O(o)）查找数据。总复杂度为 O(n * (u + o))，在用户数和订单数较大时性能极差。应在循环外预先构建 dict 索引。	Medium
app/services/analytics_service.py:112-119	在 get_product_recommendations 中，对用户的每个订单的每个条目，都线性扫描 all_products 查找对应商品。总复杂度为 O(orders * items_per_order * products)，数据量大时极慢。应在循环外构建 {product.id: product} 字典，实现 O(1) 查找。	Medium
app/services/analytics_service.py:91-98	每条订单都执行 time.strftime("%Y-%m-%d", time.localtime(order.paid_at or 0)) 做格式化和字符串比较，代价较高。当订单量达到百万级时，此函数将成为瓶颈。应预先将 target_date 解析为当天起止时间戳，用数值比较替代字符串格式化。	Low

Possible bug

File	Description	Impact
app/services/analytics_service.py:96	当 order.paid_at 为 None 时（尚未支付），代码用 or 0 将其替换为时间戳 0，对应日期 1970-01-01。如果 target_date 恰好为该值（或未做校验），可能导致错误统计。更安全的做法是直接跳过 paid_at 为空的订单。	Low

Security

File	Description	Impact
app/services/analytics_service.py:66-81	函数直接接受外部传入的 keyword 参数，且未做任何长度限制或空值校验。攻击者可传入超长字符串在大量商品描述上执行时消耗大量 CPU；空字符串会匹配所有商品，可能泄露完整商品数据库。	Low

---

🤖 Scanner

Suggestions up to 91223a6

PR Code Suggestions ✨

Performance

File	Description	Impact
app/services/analytics_service.py:23-29	在 counts 的构建循环中，对 all_product_ids 中的每个元素都调用 all_product_ids.count(pid) 重新全量扫描列表，时间复杂度为 O(n²)。同一个 pid 还会被重复计算多次（因为遍历的是含重复元素的列表）。数据量大时性能极差。应使用一次遍历聚合计数。	Medium
app/services/analytics_service.py:40-56	对每个 uid 都完整遍历 all_users（O(n) 查找）和 all_orders（O(n) 过滤），整体复杂度为 O(len(user_ids) * (len(users)+len(orders)))，在大数据量下非常低效。应预先构建索引以实现 O(1) / 一次遍历聚合。	Medium
app/services/analytics_service.py:110-132	在收集 bought_categories 时，对每个订单条目都全量扫描 all_products 查找匹配的 product，形成三重嵌套循环 O(orders * items * products)。此外 all_orders 被遍历了两次（一次取分类、一次取已购 product_id），可以合并为一次。应预先构建 product_id->product 字典索引。	Medium
app/services/analytics_service.py:91-98	对每条订单都调用 time.localtime + time.strftime 生成字符串再与 target_date 做字符串比较，海量数据下开销很大。应先将 target_date 解析为当天的时间戳起止范围，再用数值比较过滤。此外 order.paid_at or 0 在 paid_at 为 None 时会落到 1970-01-01，可能掩盖未支付/异常数据。	Low

Style

File	Description	Impact
app/services/analytics_service.py:4-6	导入了 OrderItem（来自 app.models.order）和 Optional（来自 typing），但在本文件中均未使用，属于冗余导入。建议移除以保持整洁。	Low

---

🤖 Scanner

---

🤖 Scanner

[xianfengpolo]

🟠 get_top_products 使用 O(n²) 嵌套计数 · 📎 app/services/analytics_service.py:23-29

对 all_product_ids 中的每个元素都调用 list.count()，而 count() 本身是 O(n) 全量扫描，整体复杂度为 O(n²)，数据量大时性能极差。应一次遍历用 dict（或 collections.Counter）累加计数，复杂度降至 O(n)。

Suggestion: 使用 collections.Counter 或一次遍历聚合计数。

Suggested change

	for pid in all_product_ids:
	from collections import Counter
	counts = Counter()
	for order in orders:
	for item in order.items:
	counts[item.product_id] += 1
	sorted_products = sorted(counts.items(), key=lambda x: x[1], reverse=True)
	return sorted_products[:top_n]

---

🤖 Scanner

[xianfengpolo]

🟠 get_product_recommendations 内层全量扫描查找商品 · 📎 app/services/analytics_service.py:111-126

构建 bought_categories 时对每个 order item 都遍历整个 all_products 查找匹配 product_id，复杂度为 O(订单条目 * 商品数)；同时两次重复遍历 all_orders 筛选同一用户订单。应建立 product_id -> product 索引并仅遍历一次用户订单。

Suggestion: 预建 product 索引并合并对用户订单的遍历。

Suggested change

	for order in all_orders:
	product_map = {p.id: p for p in all_products}
	user_orders = [o for o in all_orders if o.user_id == user_id]
	bought_categories = set()
	bought_product_ids = set()
	for order in user_orders:
	for item in order.items:
	bought_product_ids.add(item.product_id)
	p = product_map.get(item.product_id)
	if p is not None:
	bought_categories.add(p.category)
	recommendations = []
	for product in all_products:
	if product.category in bought_categories and product.id not in bought_product_ids:
	recommendations.append(product)
	return recommendations[:20]

---

🤖 Scanner

[xianfengpolo]

🟠 get_user_spending_report 存在 O(N*M) 线性查找 · 📎 app/services/analytics_service.py:40-56

对每个 user_id 都完整遍历 all_users 查找用户，并完整遍历 all_orders 过滤订单，整体复杂度为 O(用户数 * (用户总数 + 订单总数))。应预先构建 user 索引和按 user_id 分组的订单索引，将查找降为 O(1)。

Suggestion: 预先构建 dict 索引避免重复全量扫描。

Suggested change

	# Bug：每次都完整遍历 all_users，O(n) 查找
	user_map = {u.id: u for u in all_users}
	orders_by_user: Dict[int, List[Order]] = {}
	for order in all_orders:
	if order.status == OrderStatus.PAID:
	orders_by_user.setdefault(order.user_id, []).append(order)
	report = []
	for uid in user_ids:
	user = user_map.get(uid)
	if not user:
	continue
	user_orders = orders_by_user.get(uid, [])
	total_spent = sum(o.total for o in user_orders)
	report.append({
	"user_id": uid,
	"username": user.username,
	"total_spent": total_spent,
	"order_count": len(user_orders),
	})
	return report

---

🤖 Scanner

[xianfengpolo]

🟡 compute_daily_revenue 依赖服务器本地时区导致日期归属不确定 · 📎 app/services/analytics_service.py:96

使用 time.localtime 将时间戳按服务器本地时区格式化，跨时区部署或与传入 target_date 的时区约定不一致时，订单的日期归属会出错。应明确统一时区（例如统一使用 UTC）以保证营收统计的一致性。

Suggestion: 统一时区处理，避免依赖服务器本地时区。

Suggested change

	if order_date == target_date:
	order_date = time.strftime("%Y-%m-%d", time.gmtime(order.paid_at or 0))

---

🤖 Scanner

[xianfengpolo]

🟡 compute_daily_revenue 在循环内重复格式化时间戳 · 📎 app/services/analytics_service.py:92-98

对每条订单都执行 time.localtime + strftime 并做字符串比较，开销较大。应先将 target_date 解析为当天的起止时间戳区间，循环内仅做数值比较即可，避免反复字符串格式化。

Suggestion: 预解析目标日期为时间戳区间，循环内用数值比较过滤。

Suggested change

	if order.status != OrderStatus.PAID:
	import calendar
	day_start = calendar.timegm(time.strptime(target_date, "%Y-%m-%d"))
	day_end = day_start + 86400
	total = 0.0
	for order in orders:
	if order.status != OrderStatus.PAID:
	continue
	paid_at = order.paid_at or 0
	if day_start <= paid_at < day_end:
	total += order.total
	return total

---

🤖 Scanner

[xianfengpolo]

🟡 潜在算法复杂度拒绝服务 (O(n²)) —— get_top_products · 📎 app/services/analytics_service.py:23-29

get_top_products 中对 all_product_ids 列表的每个元素都调用 list.count()，整体复杂度为 O(n²)。如果该函数被暴露给接受用户可控数据量（例如可指定时间范围 / 大量订单）的接口，攻击者可以通过构造大量订单条目使 CPU 占用呈平方级增长，造成算法复杂度型拒绝服务（CPU exhaustion DoS）。建议使用一次遍历的哈希聚合，将复杂度降到 O(n)。

CWE: CWE-407 · OWASP: A06:2021 - Vulnerable and Outdated Components

Suggestion: 使用 collections.Counter 一次遍历完成计数，避免在循环内反复调用 list.count()。

Suggested change

	for pid in all_product_ids:
	from collections import Counter
	counts = Counter(all_product_ids)
	sorted_products = sorted(counts.items(), key=lambda x: x[1], reverse=True)
	return sorted_products[:top_n]

---

🤖 Scanner

[xianfengpolo]

🟡 嵌套全量扫描导致算法复杂度 DoS —— get_product_recommendations · 📎 app/services/analytics_service.py:111-126

get_product_recommendations 中对每个订单条目都对 all_products 做一次线性扫描查找 (O(orders * items * products))，并且注释明确指出会将全量 orders / products 加载进内存。当数据量受用户行为影响（如恶意用户大量下单后触发推荐计算）时，这种多重嵌套全量扫描会带来 CPU 与内存的成倍消耗，可被用于资源耗尽型拒绝服务。建议预先构建 product_id -> product 的字典实现 O(1) 查找，并在数据访问层做分页/批量加载，避免一次性加载全量数据。

CWE: CWE-407 · OWASP: A06:2021 - Vulnerable and Outdated Components

Suggestion: 构建 {product.id: product} 索引字典实现 O(1) 查找，避免在双层循环内对 all_products 做线性扫描；并在数据加载侧分批查询防止 OOM。

Suggested change

	for order in all_orders:
	product_by_id = {p.id: p for p in all_products}
	bought_categories = set()
	for order in all_orders:
	if order.user_id == user_id:
	for item in order.items:
	p = product_by_id.get(item.product_id)
	if p is not None:
	bought_categories.add(p.category)

---

🤖 Scanner

[xianfengpolo]

🟢 未使用的导入 OrderItem 与 Optional · 📎 app/services/analytics_service.py:4-6

导入了 OrderItem（来自 app.models.order）和 Optional（来自 typing），但在本文件中均未使用，属于冗余导入。建议移除以保持整洁。

Suggestion: 移除未使用的 OrderItem 和 Optional 导入。

Suggested change

	from typing import List, Dict, Tuple
	from app.models.order import Order, OrderStatus

---

🤖 Scanner