TaskManager-HJY

🚀 完全独立的任务管理包 - 基于新包架构，无旧包依赖

版本: 0.4.0
作者: hjy

📋 简介

TaskManager-HJY 是一个完全独立的任务管理包，基于新的包架构设计，不依赖任何旧包（如 ai_runner_hjy 或 keymaster_hjy）。该包提供了完整的任务创建、执行、监控和管理功能。

✨ 特性

• 🆕 完全独立: 不依赖任何旧包，基于新包架构
• 🚀 异步支持: 支持异步任务执行
• 📊 任务监控: 实时任务状态和进度跟踪
• 🎯 多种任务类型: 支持音频分析、数据处理、AI推理等
• 🔧 灵活配置: 支持任务优先级和自定义参数
• 📈 统计功能: 提供详细的任务统计信息
• 🛠️ CLI工具: 内置命令行工具，方便使用

📦 安装

pip install taskmanager-hjy==0.4.0

🔧 依赖

• pydantic>=2.0.0 - 数据验证
• loguru>=0.7.0 - 日志记录
• redis>=4.5.0 - 缓存支持
• rq>=1.15.0 - 任务队列
• pyyaml>=6.0 - 配置文件支持
• typing-extensions>=4.0.0 - 类型提示扩展

新包依赖（可选）

• configmanager-hjy>=0.3.0 - 配置管理
• datamanager-hjy>=0.3.0 - 数据管理
• aimanager-hjy>=0.3.0 - AI管理
• keymanager-hjy>=0.3.0 - 密钥管理

🚀 快速开始

基本使用

import asyncio
from taskmanager_hjy import TaskManager, TaskType, TaskPriority

# 创建任务管理器
manager = TaskManager()

# 创建任务
task = manager.create_task(
    name="音频分析任务",
    task_type=TaskType.AUDIO_ANALYSIS,
    parameters={"file_path": "/path/to/audio.wav"},
    priority=TaskPriority.HIGH
)

# 执行任务
async def main():
    success = await manager.execute_task(task.id)
    if success:
        print("任务执行成功!")
    
    # 获取任务信息
    task_info = manager.get_task(task.id)
    print(f"任务状态: {task_info.status}")
    print(f"任务结果: {task_info.result}")

asyncio.run(main())

便捷函数使用

import asyncio
from taskmanager_hjy import create_task, execute_task, get_task, list_tasks

# 创建任务
task = create_task("数据处理任务", TaskType.DATA_PROCESSING)

# 执行任务
async def main():
    await execute_task(task.id)
    
    # 列出所有任务
    all_tasks = list_tasks()
    print(f"总任务数: {len(all_tasks)}")
    
    # 列出待执行任务
    pending_tasks = list_tasks(status=TaskStatus.PENDING)
    print(f"待执行任务数: {len(pending_tasks)}")

asyncio.run(main())

🛠️ 命令行工具

TaskManager-HJY 提供了完整的命令行工具：

# 创建任务
python -m taskmanager_hjy create "测试任务" audio_analysis

# 列出任务
python -m taskmanager_hjy list pending

# 执行任务
python -m taskmanager_hjy execute task_20241201_12345678_abc12345

# 获取统计信息
python -m taskmanager_hjy stats

# 查看帮助
python -m taskmanager_hjy help

📊 任务类型

类型	描述	示例
AUDIO_ANALYSIS	音频分析	语音识别、音频特征提取
DATA_PROCESSING	数据处理	数据清洗、格式转换
AI_INFERENCE	AI推理	模型预测、图像识别
SYSTEM_MAINTENANCE	系统维护	日志清理、缓存更新
CUSTOM	自定义任务	用户自定义逻辑

🎯 任务状态

状态	描述
PENDING	待执行
RUNNING	执行中
COMPLETED	已完成
FAILED	失败
CANCELLED	已取消

🔝 任务优先级

优先级	描述
LOW	低优先级
NORMAL	普通优先级
HIGH	高优先级
URGENT	紧急优先级

📈 统计功能

from taskmanager_hjy import get_task_statistics

stats = get_task_statistics()
print(f"总任务数: {stats['total_tasks']}")
print(f"完成率: {stats['success_rate']:.2f}%")
print(f"运行中任务: {stats['running_tasks']}")

🔧 高级功能

任务取消

from taskmanager_hjy import cancel_task

success = cancel_task(task_id)
if success:
    print("任务取消成功")

批量操作

from taskmanager_hjy import TaskManager, TaskStatus

manager = TaskManager()

# 批量创建任务
for i in range(5):
    manager.create_task(f"批量任务 {i+1}", TaskType.DATA_PROCESSING)

# 批量执行待执行任务
pending_tasks = manager.list_tasks(status=TaskStatus.PENDING)
for task in pending_tasks:
    await manager.execute_task(task.id)

🏗️ 架构设计

TaskManager-HJY 采用模块化设计：

taskmanager_hjy/
├── __init__.py          # 包入口
├── core/
│   ├── models.py        # 数据模型
│   ├── manager.py       # 核心管理器
│   └── utils.py         # 工具函数
└── cli.py              # 命令行工具

🔄 版本历史

• 0.4.0 - 完全独立版本，移除所有旧包依赖
• 0.3.0 - 基础版本（已废弃）

📝 许可证

MIT License

🤝 贡献

欢迎提交 Issue 和 Pull Request！

📞 支持

如有问题，请通过以下方式联系：

• 提交 GitHub Issue
• 发送邮件至: hjy@example.com

---

注意: 此版本完全独立，不依赖任何旧包，确保与现有系统的兼容性。