碎片时间微学习系统一、实际应用场景描述小王是一名全栈开发工程师每天通勤需要乘坐地铁。他发现自己在等车和排队进站时有大量碎片化时间每次约3-8分钟但这些零散时间要么被浪费刷短视频要么因为不知道学什么而白白流失。他希望能够开发一个智能系统实时识别这些空闲时段自动推送适合短时学习的编程知识点卡片帮助他在一年内积累1000个编程知识点。二、引入痛点1. 时间碎片化通勤等待时间不连续难以系统化学习2. 内容选择困难短时间不知该学什么打开APP又耗时关闭3. 缺乏连续性学习内容零散难以形成知识体系4. 手动操作繁琐掏出手机→打开APP→搜索内容→学习流程太长导致放弃三、核心逻辑讲解系统架构图┌─────────────────────────────────────────────────────────────┐│ 碎片时间微学习系统 │├─────────────────────────────────────────────────────────────┤│ 数据采集层 │ 智能决策层 │ 执行层 ││ ┌──────────────┐ │ ┌──────────────┐ │ ┌──────────┐ ││ │ 位置传感器 │───▶│ │ 空闲检测引擎 │──▶│ │ 推送模块 │ ││ │ 计时器 │ │ │ 难度评估器 │ │ │ 提醒模块 │ ││ │ 状态管理器 │ │ │ 兴趣匹配器 │ │ │ 记录模块 │ ││ └──────────────┘ │ └──────────────┘ │ └──────────┘ │├─────────────────────────────────────────────────────────────┤│ 数据存储层 ││ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ││ │ 知识点库 │ │ 学习记录库 │ │ 用户画像库 │ ││ └──────────────┘ └──────────────┘ └──────────────┘ │└─────────────────────────────────────────────────────────────┘核心算法逻辑空闲时段识别算法IF (位置 地铁站) AND (运动状态 静止) AND (持续时间 60秒) THEN判定为空闲时段计算可用时长 min(持续时间, 300秒)推送对应时长的学习内容END IF智能推荐算法知识点得分 (相关性权重 × 兴趣匹配度) (时效性权重 × 新鲜度) (难度权重 × 适配度)推送优先级 知识点得分 × 随机因子(防止信息茧房)四、代码模块化实现项目结构micro_learning_system/│├── main.py # 程序入口├── config.py # 配置文件├── README.md # 说明文档│├── core/ # 核心模块│ ├── __init__.py│ ├── idle_detector.py # 空闲检测器│ ├── knowledge_engine.py # 知识引擎│ └── recommendation.py # 推荐算法│├── data/ # 数据模块│ ├── __init__.py│ ├── knowledge_base.json # 知识点库│ ├── learning_records.json # 学习记录│ └── user_profile.json # 用户画像│├── utils/ # 工具模块│ ├── __init__.py│ ├── logger.py # 日志工具│ └── helpers.py # 辅助函数│└── tests/ # 测试模块├── __init__.py└── test_idle_detector.py核心代码文件config.py - 配置文件配置文件模块存储系统运行所需的配置参数from dataclasses import dataclassfrom typing import List, Dictdataclassclass SystemConfig:系统配置类# 空闲检测阈值秒IDLE_THRESHOLD: int 60# 最大学习时间秒- 5分钟MAX_LEARNING_TIME: int 300# 最小学习时间秒- 1分钟MIN_LEARNING_TIME: int 60# 知识库文件路径KNOWLEDGE_BASE_PATH: str data/knowledge_base.json# 学习记录路径RECORDS_PATH: str data/learning_records.json# 用户画像路径PROFILE_PATH: str data/user_profile.json# 推送冷却时间秒- 防止频繁推送PUSH_COOLDOWN: int 1800 # 30分钟dataclassclass RecommendationWeights:推荐算法权重配置RELEVANCE_WEIGHT: float 0.4 # 相关性权重FRESHNESS_WEIGHT: float 0.3 # 新鲜度权重DIFFICULTY_WEIGHT: float 0.3 # 难度适配权重# 预设的兴趣标签INTEREST_TAGS [Python基础, 数据结构, 算法, 设计模式,Web框架, 数据库, DevOps, 微服务,前端技术, 人工智能, 系统设计]# 知识点难度等级DIFFICULTY_LEVELS [入门, 初级, 中级, 高级]# 学习时长分类DURATION_CATEGORIES {short: (60, 120), # 1-2分钟medium: (121, 240), # 2-4分钟long: (241, 300) # 4-5分钟}utils/logger.py - 日志工具日志工具模块提供统一的日志记录功能import loggingimport sysfrom datetime import datetimefrom pathlib import Pathdef setup_logger(log_level: int logging.INFO) - logging.Logger:设置并返回日志记录器Args:log_level: 日志级别Returns:logging.Logger: 配置好的日志记录器# 创建logger实例logger logging.getLogger(MicroLearning)logger.setLevel(log_level)# 清除已有的处理器避免重复日志logger.handlers.clear()# 创建控制台处理器console_handler logging.StreamHandler(sys.stdout)console_handler.setLevel(log_level)# 创建文件处理器log_dir Path(logs)log_dir.mkdir(exist_okTrue)file_handler logging.FileHandler(log_dir / fmicro_learning_{datetime.now().strftime(%Y%m%d)}.log,encodingutf-8)file_handler.setLevel(logging.DEBUG)# 定义日志格式formatter logging.Formatter(%(asctime)s | %(name)s | %(levelname)s | %(message)s,datefmt%Y-%m-%d %H:%M:%S)console_handler.setFormatter(formatter)file_handler.setFormatter(formatter)# 添加处理器到loggerlogger.addHandler(console_handler)logger.addHandler(file_handler)return logger# 全局logger实例logger setup_logger()utils/helpers.py - 辅助函数辅助函数模块提供通用工具函数from typing import List, Dict, Anyfrom datetime import datetime, timedeltaimport randomimport jsonfrom pathlib import Pathdef load_json_file(filepath: str) - Dict[str, Any]:加载JSON文件Args:filepath: JSON文件路径Returns:Dict: 解析后的JSON数据try:with open(filepath, r, encodingutf-8) as f:return json.load(f)except FileNotFoundError:return {}except json.JSONDecodeError as e:raise ValueError(fJSON文件格式错误: {e})def save_json_file(filepath: str, data: Dict[str, Any]) - None:保存数据到JSON文件Args:filepath: JSON文件路径data: 要保存的数据# 确保目录存在Path(filepath).parent.mkdir(parentsTrue, exist_okTrue)with open(filepath, w, encodingutf-8) as f:json.dump(data, f, ensure_asciiFalse, indent2)def calculate_time_category(seconds: int) - str:根据秒数计算时长类别Args:seconds: 时长秒Returns:str: 时长类别 (short, medium, long)if 60 seconds 120:return shortelif 121 seconds 240:return mediumelif 241 seconds 300:return longelse:return unknowndef generate_unique_id() - str:生成唯一IDReturns:str: 唯一标识符timestamp datetime.now().strftime(%Y%m%d%H%M%S)random_suffix random.randint(1000, 9999)return f{timestamp}_{random_suffix}def get_current_timestamp() - str:获取当前时间戳字符串Returns:str: ISO格式的时间戳return datetime.now().isoformat()def is_cooldown_passed(last_push_time: str, cooldown_seconds: int) - bool:检查冷却时间是否已过Args:last_push_time: 上次推送时间cooldown_seconds: 冷却时间秒Returns:bool: 是否可以再次推送if not last_push_time:return Truelast_time datetime.fromisoformat(last_push_time)current_time datetime.now()elapsed (current_time - last_time).total_seconds()return elapsed cooldown_secondsdata/knowledge_base.json - 知识点库{version: 1.0.0,last_updated: 2026-03-23T00:00:00,knowledge_points: [{id: kp_001,title: Python列表推导式,content: 列表推导式是Python中创建列表的简洁方式。语法[表达式 for 变量 in 可迭代对象 if 条件]。例squares [x**2 for x in range(10)],tags: [Python基础, 数据结构],difficulty: 入门,estimated_time: 90,category: short,prerequisites: []},{id: kp_002,title: 装饰器原理与应用,content: 装饰器是修改函数行为的函数。本质是高阶函数闭包。decorator语法糖。例timer装饰器可测量函数执行时间。,tags: [Python基础, 设计模式],difficulty: 中级,estimated_time: 150,category: medium,prerequisites: [函数基础]},{id: kp_003,title: 二分查找算法,content: 二分查找O(log n)时间复杂度。前提有序数组。核心取中点比较缩小搜索范围。递归/迭代实现。,tags: [算法, 数据结构],difficulty: 初级,estimated_time: 120,category: medium,prerequisites: [数组基础]},{id: kp_004,title: 单例模式实现,content: 单例确保类只有一个实例。Python实现__new__方法、装饰器、元类。注意线程安全。适用场景配置管理、连接池。,tags: [设计模式, Python基础],difficulty: 中级,estimated_time: 180,category: medium,prerequisites: [类与对象, 装饰器]},{id: kp_005,title: HTTP状态码速记,content: 1xx信息 2xx成功(200OK) 3xx重定向(301永久,302临时) 4xx客户端错(404未找到,401未授权) 5xx服务端错(500内部错误),tags: [Web框架, 网络协议],difficulty: 入门,estimated_time: 60,category: short,prerequisites: []},{id: kp_006,title: SQL索引优化,content: 索引加速查询但增删改变慢。B树结构。最左前缀原则。避免SELECT *。EXPLAIN分析执行计划。,tags: [数据库, 性能优化],difficulty: 中级,estimated_time: 200,category: long,prerequisites: [SQL基础]},{id: kp_007,title: Git rebase vs merge,content: merge保留分支历史rebase线性整洁。rebase黄金法则只对自己分支操作。解决冲突后git rebase --continue。,tags: [DevOps, 版本控制],difficulty: 初级,estimated_time: 110,category: medium,prerequisites: [Git基础]},{id: kp_008,title: Redis缓存策略,content: 缓存穿透(布隆过滤器) 缓存击穿(互斥锁) 缓存雪崩(过期时间随机)。淘汰策略LRU/LFU。持久化RDB/AOF。,tags: [数据库, 缓存, 微服务],difficulty: 高级,estimated_time: 250,category: long,prerequisites: [NoSQL基础, 缓存概念]}]}core/idle_detector.py - 空闲检测器空闲检测器模块负责识别用户的空闲时段import timeimport threadingfrom dataclasses import dataclass, fieldfrom enum import Enum, autofrom typing import Optional, Callablefrom datetime import datetimeimport randomfrom utils.logger import loggerfrom config import SystemConfigclass UserState(Enum):用户状态枚举UNKNOWN auto() # 未知状态MOVING auto() # 移动中WAITING auto() # 等待中静止LEARNING auto() # 学习中INTERRUPTED auto() # 被中断dataclassclass StateTransition:状态转换记录from_state: UserStateto_state: UserStatetimestamp: strduration: float # 原状态持续时长秒dataclassclass IdleSession:空闲会话记录session_id: strstart_time: strend_time: Optional[str] Nonemax_duration: int 300 # 最大学习时长is_active: bool Truepropertydef duration(self) - float:计算已持续时间if self.end_time:end datetime.fromisoformat(self.end_time)else:end datetime.now()start datetime.fromisoformat(self.start_time)return (end - start).total_seconds()propertydef remaining_time(self) - int:剩余可用时间elapsed self.durationreturn max(0, min(self.max_duration - int(elapsed), 300))class IdleDetector:空闲检测器类通过模拟位置变化检测用户是否进入空闲状态核心逻辑1. 持续监测用户状态位置、运动2. 当检测到静止状态超过阈值时触发空闲事件3. 计算可用学习时长4. 支持状态回调通知def __init__(self, config: SystemConfig None):初始化空闲检测器Args:config: 系统配置对象self.config config or SystemConfig()self.current_state UserState.UNKNOWNself.state_start_time time.time()self.transitions: List[StateTransition] []# 回调函数self.on_idle_start: Optional[Callable[[IdleSession], None]] Noneself.on_idle_end: Optional[Callable[[IdleSession], None]] Noneself.on_learning_start: Optional[Callable[[str], None]] None# 控制标志self._running Falseself._thread: Optional[threading.Thread] None# 当前空闲会话self.current_session: Optional[IdleSession] None# 模拟位置实际项目中应使用GPS/传感器APIself._simulated_location homeself._simulated_moving Falselogger.info(空闲检测器初始化完成)def set_callbacks(self,on_idle_start: Callable None,on_idle_end: Callable None,on_learning_start: Callable None) - None:设置状态变化回调函数Args:on_idle_start: 进入空闲状态时的回调on_idle_end: 离开空闲状态时的回调on_learning_start: 开始学习时的回调self.on_idle_start on_idle_startself.on_idle_end on_idle_endself.on_learning_start on_learning_startdef _change_state(self, new_state: UserState) - None:内部方法改变用户状态Args:new_state: 新状态now time.time()duration now - self.state_start_time# 记录状态转换transition StateTransition(from_stateself.current_state,to_statenew_state,timestampdatetime.now().isoformat(),durationduration)self.transitions.append(transition)logger.debug(f状态转换: {self.current_state.name} - {new_state.name}, f持续{int(duration)}秒)# 处理特定状态转换if self.current_state ! UserState.WAITING and new_state UserState.WAITING:self._handle_waiting_start()elif self.current_state UserState.WAITING and new_state ! UserState.WAITING:self._handle_waiting_end()self.current_state new_stateself.state_start_time nowdef _handle_waiting_start(self) - None:处理进入等待状态的事件检查是否满足空闲条件创建空闲会话logger.info(检测到用户进入等待状态)# 模拟等待时长判断实际项目中应根据真实停留时间wait_duration random.uniform(45, 180) # 模拟45-180秒等待# 延迟检查模拟真实场景def delayed_check():time.sleep(wait_duration)if self.current_state UserState.WAITING:self._activate_idle_session()check_thread threading.Thread(targetdelayed_check, daemonTrue)check_thread.start()def _activate_idle_session(self) - None:激活空闲会话创建空闲会话并触发回调session_id fidle_{int(time.time())}self.current_session IdleSession(session_idsession_id,start_timedatetime.now().isoformat(),max_durationself.config.MAX_LEARNING_TIME)logger.info(f空闲会话激活: {session_id}, f预估可用时长: {self.current_session.remaining_time}秒)# 触发空闲开始回调if self.on_idle_start:self.on_idle_start(self.current_session)def _handle_waiting_end(self) - None:处理结束等待状态的事件结束当前空闲会话logger.info(用户离开等待状态)if self.current_session and self.current_session.is_active:self.current_session.end_time datetime.now().isoformat()self.current_session.is_active Falseremaining self.current_session.remaining_timelogger.info(f空闲会话结束: {self.current_session.session_id}, f实际持续: {int(self.current_session.duration)}秒, f剩余时间: {remaining}秒)# 触发空闲结束回调if self.on_idle_end:self.on_idle_end(self.current_session)self.current_session Nonedef simulate_location_change(self, location: str, is_moving: bool) - None:模拟位置变化用于测试Args:location: 位置名称is_moving: 是否在移动self._simulated_location locationself._simulated_moving is_movingif is_moving:if self.current_state UserState.WAITING:self._change_state(UserState.MOVING)else:if self.current_state UserState.MOVING:self._change_state(UserState.WAITING)def get_status(self) - dict:获取当前检测器状态Returns:dict: 状态信息字典return {current_state: self.current_state.name,state_duration: int(time.time() - self.state_start_time),is_idle_session_active: (self.current_session is not None andself.current_session.is_active),current_session: ({session_id: self.current_session.session_id,duration: int(self.current_session.duration),remaining_time: self.current_session.remaining_time} if self.current_session else None),total_transitions: len(self.transitions)}def start_monitoring(self) - None:启动后台监控线程实际项目中应接入真实的位置/传感器服务if self._running:logger.warning(检测器已在运行中)returnself._running Trueself._thread threading.Thread(targetself._monitor_loop, daemonTrue)self._thread.start()logger.info(空闲检测器监控已启动)def stop_monitoring(self) - None:停止后台监控self._running Falseif self._thread:self._thread.join(timeout2)logger.info(空闲检测器监控已停止)def _monitor_loop(self) - None:监控循环后台线程模拟真实环境中的状态检测locations_cycle [home, subway_station, office, cafe]current_loc_index 0while self._running:# 模拟位置变化实际项目中应使用GPS APIif self._simulated_moving:time.sleep(1)continue# 模拟通勤场景if self._simulated_location subway_station:# 在地铁站有概率变为静止等待if random.random() 0.3: # 30%概率进入等待self.simulate_location_change(subway_station, False)time.sleep(5) # 每5秒检查一次def cleanup_old_transitions(self, keep_count: int 100) - None:清理旧的状态转换记录Args:keep_count: 保留的记录数量if len(self.transitions) keep_count:removed len(self.transitions) - keep_countself.transitions self.transitions[-keep_count:]logger.debug(f清理了{removed}条旧的状态转换记录)# 使用示例和测试if __name__ __main__:# 创建检测器实例detector IdleDetector()# 设置回调函数def on_idle(session):print(f 检测到空闲时段会话ID: {session.session_id})print(f 可用学习时长: {session.remaining_time}秒)def on_idle_end(session):print(f⏹️ 空闲时段结束会话ID: {session.session_id})print(f 实际学习时长: {int(session.duration)}秒)detector.set_callbacks(on_idle_starton_idle, on_idle_endon_idle_end)# 启动监控detector.start_monitoring()# 模拟场景用户在地铁站等待print( 模拟场景开始 )detector.simulate_location_change(subway_station, False) # 到达地铁站并停下# 等待检测结果time.sleep(5)# 查看状态status detector.get_status()print(f\n当前状态: {status})# 模拟离开time.sleep(3)detector.simulate_location_change(subway_station, True) # 开始移动# 停止监控time.sleep(2)detector.stop_monitoring()print(\n 模拟场景结束 )core/knowledge_engine.py - 知识引擎知识引擎模块管理知识点库和用户学习记录import jsonfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Setfrom datetime import datetime, timedeltaimport randomfrom pathlib import Pathfrom utils.logger import loggerfrom utils.helpers import load_json_file, save_json_file, calculate_time_categoryfrom config import SystemConfig, INTEREST_TAGS, DIFFICULTY_LEVELS, DURATION_CATEGORIESdataclassclass KnowledgePoint:知识点数据类id: strtitle: strcontent: strtags: List[str]difficulty: strestimated_time: intcategory: strprerequisites: List[str] field(default_factorylist)created_at: str field(default_factorylambda: datetime.now().isoformat())view_count: int 0completion_利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
碎片化时间利用程序，识别等车排队空档，推荐微学习，积少成多，提升自己。
发布时间：2026/7/11 8:32:21
碎片时间微学习系统一、实际应用场景描述小王是一名全栈开发工程师每天通勤需要乘坐地铁。他发现自己在等车和排队进站时有大量碎片化时间每次约3-8分钟但这些零散时间要么被浪费刷短视频要么因为不知道学什么而白白流失。他希望能够开发一个智能系统实时识别这些空闲时段自动推送适合短时学习的编程知识点卡片帮助他在一年内积累1000个编程知识点。二、引入痛点1. 时间碎片化通勤等待时间不连续难以系统化学习2. 内容选择困难短时间不知该学什么打开APP又耗时关闭3. 缺乏连续性学习内容零散难以形成知识体系4. 手动操作繁琐掏出手机→打开APP→搜索内容→学习流程太长导致放弃三、核心逻辑讲解系统架构图┌─────────────────────────────────────────────────────────────┐│ 碎片时间微学习系统 │├─────────────────────────────────────────────────────────────┤│ 数据采集层 │ 智能决策层 │ 执行层 ││ ┌──────────────┐ │ ┌──────────────┐ │ ┌──────────┐ ││ │ 位置传感器 │───▶│ │ 空闲检测引擎 │──▶│ │ 推送模块 │ ││ │ 计时器 │ │ │ 难度评估器 │ │ │ 提醒模块 │ ││ │ 状态管理器 │ │ │ 兴趣匹配器 │ │ │ 记录模块 │ ││ └──────────────┘ │ └──────────────┘ │ └──────────┘ │├─────────────────────────────────────────────────────────────┤│ 数据存储层 ││ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ ││ │ 知识点库 │ │ 学习记录库 │ │ 用户画像库 │ ││ └──────────────┘ └──────────────┘ └──────────────┘ │└─────────────────────────────────────────────────────────────┘核心算法逻辑空闲时段识别算法IF (位置 地铁站) AND (运动状态 静止) AND (持续时间 60秒) THEN判定为空闲时段计算可用时长 min(持续时间, 300秒)推送对应时长的学习内容END IF智能推荐算法知识点得分 (相关性权重 × 兴趣匹配度) (时效性权重 × 新鲜度) (难度权重 × 适配度)推送优先级 知识点得分 × 随机因子(防止信息茧房)四、代码模块化实现项目结构micro_learning_system/│├── main.py # 程序入口├── config.py # 配置文件├── README.md # 说明文档│├── core/ # 核心模块│ ├── __init__.py│ ├── idle_detector.py # 空闲检测器│ ├── knowledge_engine.py # 知识引擎│ └── recommendation.py # 推荐算法│├── data/ # 数据模块│ ├── __init__.py│ ├── knowledge_base.json # 知识点库│ ├── learning_records.json # 学习记录│ └── user_profile.json # 用户画像│├── utils/ # 工具模块│ ├── __init__.py│ ├── logger.py # 日志工具│ └── helpers.py # 辅助函数│└── tests/ # 测试模块├── __init__.py└── test_idle_detector.py核心代码文件config.py - 配置文件配置文件模块存储系统运行所需的配置参数from dataclasses import dataclassfrom typing import List, Dictdataclassclass SystemConfig:系统配置类# 空闲检测阈值秒IDLE_THRESHOLD: int 60# 最大学习时间秒- 5分钟MAX_LEARNING_TIME: int 300# 最小学习时间秒- 1分钟MIN_LEARNING_TIME: int 60# 知识库文件路径KNOWLEDGE_BASE_PATH: str data/knowledge_base.json# 学习记录路径RECORDS_PATH: str data/learning_records.json# 用户画像路径PROFILE_PATH: str data/user_profile.json# 推送冷却时间秒- 防止频繁推送PUSH_COOLDOWN: int 1800 # 30分钟dataclassclass RecommendationWeights:推荐算法权重配置RELEVANCE_WEIGHT: float 0.4 # 相关性权重FRESHNESS_WEIGHT: float 0.3 # 新鲜度权重DIFFICULTY_WEIGHT: float 0.3 # 难度适配权重# 预设的兴趣标签INTEREST_TAGS [Python基础, 数据结构, 算法, 设计模式,Web框架, 数据库, DevOps, 微服务,前端技术, 人工智能, 系统设计]# 知识点难度等级DIFFICULTY_LEVELS [入门, 初级, 中级, 高级]# 学习时长分类DURATION_CATEGORIES {short: (60, 120), # 1-2分钟medium: (121, 240), # 2-4分钟long: (241, 300) # 4-5分钟}utils/logger.py - 日志工具日志工具模块提供统一的日志记录功能import loggingimport sysfrom datetime import datetimefrom pathlib import Pathdef setup_logger(log_level: int logging.INFO) - logging.Logger:设置并返回日志记录器Args:log_level: 日志级别Returns:logging.Logger: 配置好的日志记录器# 创建logger实例logger logging.getLogger(MicroLearning)logger.setLevel(log_level)# 清除已有的处理器避免重复日志logger.handlers.clear()# 创建控制台处理器console_handler logging.StreamHandler(sys.stdout)console_handler.setLevel(log_level)# 创建文件处理器log_dir Path(logs)log_dir.mkdir(exist_okTrue)file_handler logging.FileHandler(log_dir / fmicro_learning_{datetime.now().strftime(%Y%m%d)}.log,encodingutf-8)file_handler.setLevel(logging.DEBUG)# 定义日志格式formatter logging.Formatter(%(asctime)s | %(name)s | %(levelname)s | %(message)s,datefmt%Y-%m-%d %H:%M:%S)console_handler.setFormatter(formatter)file_handler.setFormatter(formatter)# 添加处理器到loggerlogger.addHandler(console_handler)logger.addHandler(file_handler)return logger# 全局logger实例logger setup_logger()utils/helpers.py - 辅助函数辅助函数模块提供通用工具函数from typing import List, Dict, Anyfrom datetime import datetime, timedeltaimport randomimport jsonfrom pathlib import Pathdef load_json_file(filepath: str) - Dict[str, Any]:加载JSON文件Args:filepath: JSON文件路径Returns:Dict: 解析后的JSON数据try:with open(filepath, r, encodingutf-8) as f:return json.load(f)except FileNotFoundError:return {}except json.JSONDecodeError as e:raise ValueError(fJSON文件格式错误: {e})def save_json_file(filepath: str, data: Dict[str, Any]) - None:保存数据到JSON文件Args:filepath: JSON文件路径data: 要保存的数据# 确保目录存在Path(filepath).parent.mkdir(parentsTrue, exist_okTrue)with open(filepath, w, encodingutf-8) as f:json.dump(data, f, ensure_asciiFalse, indent2)def calculate_time_category(seconds: int) - str:根据秒数计算时长类别Args:seconds: 时长秒Returns:str: 时长类别 (short, medium, long)if 60 seconds 120:return shortelif 121 seconds 240:return mediumelif 241 seconds 300:return longelse:return unknowndef generate_unique_id() - str:生成唯一IDReturns:str: 唯一标识符timestamp datetime.now().strftime(%Y%m%d%H%M%S)random_suffix random.randint(1000, 9999)return f{timestamp}_{random_suffix}def get_current_timestamp() - str:获取当前时间戳字符串Returns:str: ISO格式的时间戳return datetime.now().isoformat()def is_cooldown_passed(last_push_time: str, cooldown_seconds: int) - bool:检查冷却时间是否已过Args:last_push_time: 上次推送时间cooldown_seconds: 冷却时间秒Returns:bool: 是否可以再次推送if not last_push_time:return Truelast_time datetime.fromisoformat(last_push_time)current_time datetime.now()elapsed (current_time - last_time).total_seconds()return elapsed cooldown_secondsdata/knowledge_base.json - 知识点库{version: 1.0.0,last_updated: 2026-03-23T00:00:00,knowledge_points: [{id: kp_001,title: Python列表推导式,content: 列表推导式是Python中创建列表的简洁方式。语法[表达式 for 变量 in 可迭代对象 if 条件]。例squares [x**2 for x in range(10)],tags: [Python基础, 数据结构],difficulty: 入门,estimated_time: 90,category: short,prerequisites: []},{id: kp_002,title: 装饰器原理与应用,content: 装饰器是修改函数行为的函数。本质是高阶函数闭包。decorator语法糖。例timer装饰器可测量函数执行时间。,tags: [Python基础, 设计模式],difficulty: 中级,estimated_time: 150,category: medium,prerequisites: [函数基础]},{id: kp_003,title: 二分查找算法,content: 二分查找O(log n)时间复杂度。前提有序数组。核心取中点比较缩小搜索范围。递归/迭代实现。,tags: [算法, 数据结构],difficulty: 初级,estimated_time: 120,category: medium,prerequisites: [数组基础]},{id: kp_004,title: 单例模式实现,content: 单例确保类只有一个实例。Python实现__new__方法、装饰器、元类。注意线程安全。适用场景配置管理、连接池。,tags: [设计模式, Python基础],difficulty: 中级,estimated_time: 180,category: medium,prerequisites: [类与对象, 装饰器]},{id: kp_005,title: HTTP状态码速记,content: 1xx信息 2xx成功(200OK) 3xx重定向(301永久,302临时) 4xx客户端错(404未找到,401未授权) 5xx服务端错(500内部错误),tags: [Web框架, 网络协议],difficulty: 入门,estimated_time: 60,category: short,prerequisites: []},{id: kp_006,title: SQL索引优化,content: 索引加速查询但增删改变慢。B树结构。最左前缀原则。避免SELECT *。EXPLAIN分析执行计划。,tags: [数据库, 性能优化],difficulty: 中级,estimated_time: 200,category: long,prerequisites: [SQL基础]},{id: kp_007,title: Git rebase vs merge,content: merge保留分支历史rebase线性整洁。rebase黄金法则只对自己分支操作。解决冲突后git rebase --continue。,tags: [DevOps, 版本控制],difficulty: 初级,estimated_time: 110,category: medium,prerequisites: [Git基础]},{id: kp_008,title: Redis缓存策略,content: 缓存穿透(布隆过滤器) 缓存击穿(互斥锁) 缓存雪崩(过期时间随机)。淘汰策略LRU/LFU。持久化RDB/AOF。,tags: [数据库, 缓存, 微服务],difficulty: 高级,estimated_time: 250,category: long,prerequisites: [NoSQL基础, 缓存概念]}]}core/idle_detector.py - 空闲检测器空闲检测器模块负责识别用户的空闲时段import timeimport threadingfrom dataclasses import dataclass, fieldfrom enum import Enum, autofrom typing import Optional, Callablefrom datetime import datetimeimport randomfrom utils.logger import loggerfrom config import SystemConfigclass UserState(Enum):用户状态枚举UNKNOWN auto() # 未知状态MOVING auto() # 移动中WAITING auto() # 等待中静止LEARNING auto() # 学习中INTERRUPTED auto() # 被中断dataclassclass StateTransition:状态转换记录from_state: UserStateto_state: UserStatetimestamp: strduration: float # 原状态持续时长秒dataclassclass IdleSession:空闲会话记录session_id: strstart_time: strend_time: Optional[str] Nonemax_duration: int 300 # 最大学习时长is_active: bool Truepropertydef duration(self) - float:计算已持续时间if self.end_time:end datetime.fromisoformat(self.end_time)else:end datetime.now()start datetime.fromisoformat(self.start_time)return (end - start).total_seconds()propertydef remaining_time(self) - int:剩余可用时间elapsed self.durationreturn max(0, min(self.max_duration - int(elapsed), 300))class IdleDetector:空闲检测器类通过模拟位置变化检测用户是否进入空闲状态核心逻辑1. 持续监测用户状态位置、运动2. 当检测到静止状态超过阈值时触发空闲事件3. 计算可用学习时长4. 支持状态回调通知def __init__(self, config: SystemConfig None):初始化空闲检测器Args:config: 系统配置对象self.config config or SystemConfig()self.current_state UserState.UNKNOWNself.state_start_time time.time()self.transitions: List[StateTransition] []# 回调函数self.on_idle_start: Optional[Callable[[IdleSession], None]] Noneself.on_idle_end: Optional[Callable[[IdleSession], None]] Noneself.on_learning_start: Optional[Callable[[str], None]] None# 控制标志self._running Falseself._thread: Optional[threading.Thread] None# 当前空闲会话self.current_session: Optional[IdleSession] None# 模拟位置实际项目中应使用GPS/传感器APIself._simulated_location homeself._simulated_moving Falselogger.info(空闲检测器初始化完成)def set_callbacks(self,on_idle_start: Callable None,on_idle_end: Callable None,on_learning_start: Callable None) - None:设置状态变化回调函数Args:on_idle_start: 进入空闲状态时的回调on_idle_end: 离开空闲状态时的回调on_learning_start: 开始学习时的回调self.on_idle_start on_idle_startself.on_idle_end on_idle_endself.on_learning_start on_learning_startdef _change_state(self, new_state: UserState) - None:内部方法改变用户状态Args:new_state: 新状态now time.time()duration now - self.state_start_time# 记录状态转换transition StateTransition(from_stateself.current_state,to_statenew_state,timestampdatetime.now().isoformat(),durationduration)self.transitions.append(transition)logger.debug(f状态转换: {self.current_state.name} - {new_state.name}, f持续{int(duration)}秒)# 处理特定状态转换if self.current_state ! UserState.WAITING and new_state UserState.WAITING:self._handle_waiting_start()elif self.current_state UserState.WAITING and new_state ! UserState.WAITING:self._handle_waiting_end()self.current_state new_stateself.state_start_time nowdef _handle_waiting_start(self) - None:处理进入等待状态的事件检查是否满足空闲条件创建空闲会话logger.info(检测到用户进入等待状态)# 模拟等待时长判断实际项目中应根据真实停留时间wait_duration random.uniform(45, 180) # 模拟45-180秒等待# 延迟检查模拟真实场景def delayed_check():time.sleep(wait_duration)if self.current_state UserState.WAITING:self._activate_idle_session()check_thread threading.Thread(targetdelayed_check, daemonTrue)check_thread.start()def _activate_idle_session(self) - None:激活空闲会话创建空闲会话并触发回调session_id fidle_{int(time.time())}self.current_session IdleSession(session_idsession_id,start_timedatetime.now().isoformat(),max_durationself.config.MAX_LEARNING_TIME)logger.info(f空闲会话激活: {session_id}, f预估可用时长: {self.current_session.remaining_time}秒)# 触发空闲开始回调if self.on_idle_start:self.on_idle_start(self.current_session)def _handle_waiting_end(self) - None:处理结束等待状态的事件结束当前空闲会话logger.info(用户离开等待状态)if self.current_session and self.current_session.is_active:self.current_session.end_time datetime.now().isoformat()self.current_session.is_active Falseremaining self.current_session.remaining_timelogger.info(f空闲会话结束: {self.current_session.session_id}, f实际持续: {int(self.current_session.duration)}秒, f剩余时间: {remaining}秒)# 触发空闲结束回调if self.on_idle_end:self.on_idle_end(self.current_session)self.current_session Nonedef simulate_location_change(self, location: str, is_moving: bool) - None:模拟位置变化用于测试Args:location: 位置名称is_moving: 是否在移动self._simulated_location locationself._simulated_moving is_movingif is_moving:if self.current_state UserState.WAITING:self._change_state(UserState.MOVING)else:if self.current_state UserState.MOVING:self._change_state(UserState.WAITING)def get_status(self) - dict:获取当前检测器状态Returns:dict: 状态信息字典return {current_state: self.current_state.name,state_duration: int(time.time() - self.state_start_time),is_idle_session_active: (self.current_session is not None andself.current_session.is_active),current_session: ({session_id: self.current_session.session_id,duration: int(self.current_session.duration),remaining_time: self.current_session.remaining_time} if self.current_session else None),total_transitions: len(self.transitions)}def start_monitoring(self) - None:启动后台监控线程实际项目中应接入真实的位置/传感器服务if self._running:logger.warning(检测器已在运行中)returnself._running Trueself._thread threading.Thread(targetself._monitor_loop, daemonTrue)self._thread.start()logger.info(空闲检测器监控已启动)def stop_monitoring(self) - None:停止后台监控self._running Falseif self._thread:self._thread.join(timeout2)logger.info(空闲检测器监控已停止)def _monitor_loop(self) - None:监控循环后台线程模拟真实环境中的状态检测locations_cycle [home, subway_station, office, cafe]current_loc_index 0while self._running:# 模拟位置变化实际项目中应使用GPS APIif self._simulated_moving:time.sleep(1)continue# 模拟通勤场景if self._simulated_location subway_station:# 在地铁站有概率变为静止等待if random.random() 0.3: # 30%概率进入等待self.simulate_location_change(subway_station, False)time.sleep(5) # 每5秒检查一次def cleanup_old_transitions(self, keep_count: int 100) - None:清理旧的状态转换记录Args:keep_count: 保留的记录数量if len(self.transitions) keep_count:removed len(self.transitions) - keep_countself.transitions self.transitions[-keep_count:]logger.debug(f清理了{removed}条旧的状态转换记录)# 使用示例和测试if __name__ __main__:# 创建检测器实例detector IdleDetector()# 设置回调函数def on_idle(session):print(f 检测到空闲时段会话ID: {session.session_id})print(f 可用学习时长: {session.remaining_time}秒)def on_idle_end(session):print(f⏹️ 空闲时段结束会话ID: {session.session_id})print(f 实际学习时长: {int(session.duration)}秒)detector.set_callbacks(on_idle_starton_idle, on_idle_endon_idle_end)# 启动监控detector.start_monitoring()# 模拟场景用户在地铁站等待print( 模拟场景开始 )detector.simulate_location_change(subway_station, False) # 到达地铁站并停下# 等待检测结果time.sleep(5)# 查看状态status detector.get_status()print(f\n当前状态: {status})# 模拟离开time.sleep(3)detector.simulate_location_change(subway_station, True) # 开始移动# 停止监控time.sleep(2)detector.stop_monitoring()print(\n 模拟场景结束 )core/knowledge_engine.py - 知识引擎知识引擎模块管理知识点库和用户学习记录import jsonfrom dataclasses import dataclass, fieldfrom typing import List, Dict, Optional, Setfrom datetime import datetime, timedeltaimport randomfrom pathlib import Pathfrom utils.logger import loggerfrom utils.helpers import load_json_file, save_json_file, calculate_time_categoryfrom config import SystemConfig, INTEREST_TAGS, DIFFICULTY_LEVELS, DURATION_CATEGORIESdataclassclass KnowledgePoint:知识点数据类id: strtitle: strcontent: strtags: List[str]difficulty: strestimated_time: intcategory: strprerequisites: List[str] field(default_factorylist)created_at: str field(default_factorylambda: datetime.now().isoformat())view_count: int 0completion_利用AI解决实际问题如果你觉得这个工具好用欢迎关注长安牧笛
