题目重点:先签解析与聚合口径,再按 (IP,machine) 计数并稳定排序,处理同 IP 多设备和缺失字段。
题意与假设
这道练习题保留了面经中值得训练的核心约束,并将需要澄清的条件单独列出。现场拿到变体时,先和面试官确认这些条件。
- 代码接收已经规范化的事件 `{ip,machine,outcome}`;原始日志解析器取决于真实行格式,不能从题目描述编造分隔符或正则。
- 训练版以 `(IP,machine)` 为计数键,`outcome=='failed'` 为失败;同一 IP 在两台机器上分开计。输出按次数降序、IP 和 machine 字典序打破并列。
- 若要求按 IP 汇总并附机器集合,或按登录目标账户聚合,应另改键与输出;缺必需字段直接报错而不静默丢弃。
解题路线
- 先与面试官确认日志字段、失败值、时间窗口、IP 规范化和机器来源;解析阶段把格式错误行单列错误计数。
- 对每个规范化事件,只将 failed 项计入 `(ip,machine)` 频数表;成功事件不影响失败计数。
- 把频数项按 `(-count,ip,machine)` 排序;若总事件量大但只需 Top K,可改维护 K 小根堆。
- 训练实现以独立嵌套字典参考模型验证 37,449 条短事件序列,覆盖重复、同 IP 多机器、成功事件与并列。
手动走一遍
失败事件 (1.1.1.1,A)、(1.1.1.1,B)、(1.1.1.1,A) 与成功事件 (2.2.2.2,C),结果为 [(1.1.1.1,A,2),(1.1.1.1,B,1)]。若改为只按 IP 聚合,结果会是一个计数 3 的 IP,这不是同一查询口径。
Python 参考实现
下面的代码按上文假设实现,适合本地运行并追加自己的测试用例。
代码区可横向滚动;使用键盘时,先聚焦代码区再按方向键查看。
from collections import Counter
def failed_login_report(events: list[dict[str, str]]) -> list[tuple[str, str, int]]:
"""Aggregate normalized login events by (IP, machine)."""
failures: Counter[tuple[str, str]] = Counter()
for event in events:
if not {"ip", "machine", "outcome"}.issubset(event):
raise ValueError("missing normalized event field")
if event["outcome"] == "failed":
failures[(event["ip"], event["machine"])] += 1
return [
(ip, machine, count)
for (ip, machine), count in sorted(
failures.items(), key=lambda item: (-item[1], item[0][0], item[0][1])
)
]
复杂度
N 条规范化日志、一共 U 个不同 (IP,machine) 键,计数平均 O(N) 时间、O(U) 空间;全排序 O(U log U)。需要流式持续时间窗时还要存窗口事件/时间桶,不能只维护终身总数。
易错点与追问
容易失分的边界
- 假设题目描述没给的日志分隔符并写一个看似原题的 parser。
- 把 IP→单 machine 映射覆盖,丢失同 IP 多机器情况。
- 把成功登录也计为失败。
- 未定义并列顺序,测试和输出不稳定。
- IP 字符串规范化和 IPv6 等价形式未处理却声称真实日志已去重。
继续深挖
- 若要最近一小时失败最多的 IP,如何按时间桶过期?
- 如何识别 NAT 后很多用户共享一个 IP,减少误报警?
- 若一行字段缺失/损坏,输出是否包含错误报告?
- 分布式日志流如何做幂等去重与全局 Top K?
