feat(memory-graph): 完成 Phase 2 - 记忆构建与工具接口
Phase 2 实现内容: - 时间解析器 (utils/time_parser.py): 支持自然语言时间表达式 - 记忆提取器 (core/extractor.py): 参数验证和标准化 - 记忆构建器 (core/builder.py): 自动构造记忆子图，支持节点去重和关联 - 嵌入生成器 (utils/embeddings.py): API 优先策略，降低本地负载 - LLM 工具接口 (tools/memory_tools.py): create_memory, link_memories, search_memories 关键修复: - VectorStore: 支持 ChromaDB 列表元数据的 JSON 序列化 - 测试数据同步: 确保向量存储和图存储数据一致性 测试结果: 时间解析器: 6/6 通过 记忆提取器: 3 个测试用例通过 记忆构建器: 构建记忆子图成功 端到端流程: 成功创建 3 条记忆 记忆关联: 建立因果关系成功 记忆搜索: 语义搜索返回正确结果 工具 Schema: 3 个工具定义完整 下一步: Phase 3 - 管理层实现
This commit is contained in:
Windpicker-owo
8
src/memory_graph/utils/__init__.py
Normal file
8
src/memory_graph/utils/__init__.py
Normal file
@@ -0,0 +1,8 @@
|
||||
"""
|
||||
工具模块
|
||||
"""
|
||||
|
||||
from src.memory_graph.utils.embeddings import EmbeddingGenerator, get_embedding_generator
|
||||
from src.memory_graph.utils.time_parser import TimeParser
|
||||
|
||||
__all__ = ["TimeParser", "EmbeddingGenerator", "get_embedding_generator"]
|
||||
299
src/memory_graph/utils/embeddings.py
Normal file
299
src/memory_graph/utils/embeddings.py
Normal file
@@ -0,0 +1,299 @@
|
||||
"""
|
||||
嵌入向量生成器：优先使用配置的 embedding API，sentence-transformers 作为备选
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from functools import lru_cache
|
||||
from typing import List, Optional
|
||||
|
||||
import numpy as np
|
||||
|
||||
from src.common.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class EmbeddingGenerator:
|
||||
"""
|
||||
嵌入向量生成器
|
||||
|
||||
策略：
|
||||
1. 优先使用配置的 embedding API（通过 LLMRequest）
|
||||
2. 如果 API 不可用，回退到本地 sentence-transformers
|
||||
3. 如果 sentence-transformers 未安装，使用随机向量（仅测试）
|
||||
|
||||
优点：
|
||||
- 降低本地运算负载
|
||||
- 即使未安装 sentence-transformers 也可正常运行
|
||||
- 保持与现有系统的一致性
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
use_api: bool = True,
|
||||
fallback_model_name: str = "paraphrase-multilingual-MiniLM-L12-v2",
|
||||
):
|
||||
"""
|
||||
初始化嵌入生成器
|
||||
|
||||
Args:
|
||||
use_api: 是否优先使用 API（默认 True）
|
||||
fallback_model_name: 回退本地模型名称
|
||||
"""
|
||||
self.use_api = use_api
|
||||
self.fallback_model_name = fallback_model_name
|
||||
|
||||
# API 相关
|
||||
self._llm_request = None
|
||||
self._api_available = False
|
||||
self._api_dimension = None
|
||||
|
||||
# 本地模型相关
|
||||
self._local_model = None
|
||||
self._local_model_loaded = False
|
||||
|
||||
async def _initialize_api(self):
|
||||
"""初始化 embedding API"""
|
||||
if self._api_available:
|
||||
return
|
||||
|
||||
try:
|
||||
from src.config.config import model_config
|
||||
from src.llm_models.utils_model import LLMRequest
|
||||
|
||||
embedding_config = model_config.model_task_config.embedding
|
||||
self._llm_request = LLMRequest(
|
||||
model_set=embedding_config,
|
||||
request_type="memory_graph.embedding"
|
||||
)
|
||||
|
||||
# 获取嵌入维度
|
||||
if hasattr(embedding_config, "embedding_dimension") and embedding_config.embedding_dimension:
|
||||
self._api_dimension = embedding_config.embedding_dimension
|
||||
|
||||
self._api_available = True
|
||||
logger.info(f"✅ Embedding API 初始化成功 (维度: {self._api_dimension})")
|
||||
|
||||
except Exception as e:
|
||||
logger.warning(f"⚠️ Embedding API 初始化失败: {e}")
|
||||
self._api_available = False
|
||||
|
||||
def _load_local_model(self):
|
||||
"""延迟加载本地模型"""
|
||||
if not self._local_model_loaded:
|
||||
try:
|
||||
from sentence_transformers import SentenceTransformer
|
||||
|
||||
logger.info(f"📦 加载本地嵌入模型: {self.fallback_model_name}")
|
||||
self._local_model = SentenceTransformer(self.fallback_model_name)
|
||||
self._local_model_loaded = True
|
||||
logger.info("✅ 本地嵌入模型加载成功")
|
||||
except ImportError:
|
||||
logger.warning(
|
||||
"⚠️ sentence-transformers 未安装，将使用随机向量（仅测试用）\n"
|
||||
" 安装方法: pip install sentence-transformers"
|
||||
)
|
||||
self._local_model_loaded = False
|
||||
except Exception as e:
|
||||
logger.warning(f"⚠️ 本地模型加载失败: {e}")
|
||||
self._local_model_loaded = False
|
||||
|
||||
async def generate(self, text: str) -> np.ndarray:
|
||||
"""
|
||||
生成单个文本的嵌入向量
|
||||
|
||||
策略：
|
||||
1. 优先使用 API
|
||||
2. API 失败则使用本地模型
|
||||
3. 本地模型不可用则使用随机向量
|
||||
|
||||
Args:
|
||||
text: 输入文本
|
||||
|
||||
Returns:
|
||||
嵌入向量
|
||||
"""
|
||||
if not text or not text.strip():
|
||||
logger.warning("输入文本为空，返回零向量")
|
||||
dim = self._get_dimension()
|
||||
return np.zeros(dim, dtype=np.float32)
|
||||
|
||||
try:
|
||||
# 策略 1: 使用 API
|
||||
if self.use_api:
|
||||
embedding = await self._generate_with_api(text)
|
||||
if embedding is not None:
|
||||
return embedding
|
||||
|
||||
# 策略 2: 使用本地模型
|
||||
embedding = await self._generate_with_local_model(text)
|
||||
if embedding is not None:
|
||||
return embedding
|
||||
|
||||
# 策略 3: 随机向量（仅测试）
|
||||
logger.warning(f"⚠️ 所有嵌入策略失败，使用随机向量: {text[:30]}...")
|
||||
dim = self._get_dimension()
|
||||
return np.random.rand(dim).astype(np.float32)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"❌ 嵌入生成失败: {e}", exc_info=True)
|
||||
dim = self._get_dimension()
|
||||
return np.random.rand(dim).astype(np.float32)
|
||||
|
||||
async def _generate_with_api(self, text: str) -> Optional[np.ndarray]:
|
||||
"""使用 API 生成嵌入"""
|
||||
try:
|
||||
# 初始化 API
|
||||
if not self._api_available:
|
||||
await self._initialize_api()
|
||||
|
||||
if not self._api_available or not self._llm_request:
|
||||
return None
|
||||
|
||||
# 调用 API
|
||||
embedding_list, model_name = await self._llm_request.get_embedding(text)
|
||||
|
||||
if embedding_list and len(embedding_list) > 0:
|
||||
embedding = np.array(embedding_list, dtype=np.float32)
|
||||
logger.debug(f"🌐 API 生成嵌入: {text[:30]}... -> {len(embedding)}维 (模型: {model_name})")
|
||||
return embedding
|
||||
|
||||
return None
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"API 嵌入生成失败: {e}")
|
||||
return None
|
||||
|
||||
async def _generate_with_local_model(self, text: str) -> Optional[np.ndarray]:
|
||||
"""使用本地模型生成嵌入"""
|
||||
try:
|
||||
# 加载本地模型
|
||||
if not self._local_model_loaded:
|
||||
self._load_local_model()
|
||||
|
||||
if not self._local_model_loaded or not self._local_model:
|
||||
return None
|
||||
|
||||
# 在线程池中运行
|
||||
loop = asyncio.get_event_loop()
|
||||
embedding = await loop.run_in_executor(None, self._encode_single_local, text)
|
||||
|
||||
logger.debug(f"💻 本地生成嵌入: {text[:30]}... -> {len(embedding)}维")
|
||||
return embedding
|
||||
|
||||
except Exception as e:
|
||||
logger.debug(f"本地模型嵌入生成失败: {e}")
|
||||
return None
|
||||
|
||||
def _encode_single_local(self, text: str) -> np.ndarray:
|
||||
"""同步编码单个文本（本地模型）"""
|
||||
if self._local_model is None:
|
||||
raise RuntimeError("本地模型未加载")
|
||||
embedding = self._local_model.encode(text, convert_to_numpy=True) # type: ignore
|
||||
return embedding.astype(np.float32)
|
||||
|
||||
def _get_dimension(self) -> int:
|
||||
"""获取嵌入维度"""
|
||||
# 优先使用 API 维度
|
||||
if self._api_dimension:
|
||||
return self._api_dimension
|
||||
|
||||
# 其次使用本地模型维度
|
||||
if self._local_model_loaded and self._local_model:
|
||||
try:
|
||||
return self._local_model.get_sentence_embedding_dimension()
|
||||
except:
|
||||
pass
|
||||
|
||||
# 默认 384（sentence-transformers 常用维度）
|
||||
return 384
|
||||
|
||||
async def generate_batch(self, texts: List[str]) -> List[np.ndarray]:
|
||||
"""
|
||||
批量生成嵌入向量
|
||||
|
||||
Args:
|
||||
texts: 文本列表
|
||||
|
||||
Returns:
|
||||
嵌入向量列表
|
||||
"""
|
||||
if not texts:
|
||||
return []
|
||||
|
||||
try:
|
||||
# 过滤空文本
|
||||
valid_texts = [t for t in texts if t and t.strip()]
|
||||
if not valid_texts:
|
||||
logger.warning("所有文本为空，返回零向量列表")
|
||||
dim = self._get_dimension()
|
||||
return [np.zeros(dim, dtype=np.float32) for _ in texts]
|
||||
|
||||
# 使用 API 批量生成（如果可用）
|
||||
if self.use_api:
|
||||
results = await self._generate_batch_with_api(valid_texts)
|
||||
if results:
|
||||
return results
|
||||
|
||||
# 回退到逐个生成
|
||||
results = []
|
||||
for text in valid_texts:
|
||||
embedding = await self.generate(text)
|
||||
results.append(embedding)
|
||||
|
||||
logger.info(f"✅ 批量生成嵌入: {len(texts)} 个文本")
|
||||
return results
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"❌ 批量嵌入生成失败: {e}", exc_info=True)
|
||||
dim = self._get_dimension()
|
||||
return [np.random.rand(dim).astype(np.float32) for _ in texts]
|
||||
|
||||
async def _generate_batch_with_api(self, texts: List[str]) -> Optional[List[np.ndarray]]:
|
||||
"""使用 API 批量生成"""
|
||||
try:
|
||||
# 对于大多数 API，批量调用就是多次单独调用
|
||||
# 这里保持简单，逐个调用
|
||||
results = []
|
||||
for text in texts:
|
||||
embedding = await self._generate_with_api(text)
|
||||
if embedding is None:
|
||||
return None # 如果任何一个失败，返回 None 触发回退
|
||||
results.append(embedding)
|
||||
return results
|
||||
except Exception as e:
|
||||
logger.debug(f"API 批量生成失败: {e}")
|
||||
return None
|
||||
|
||||
def get_embedding_dimension(self) -> int:
|
||||
"""获取嵌入向量维度"""
|
||||
return self._get_dimension()
|
||||
|
||||
|
||||
# 全局单例
|
||||
_global_generator: Optional[EmbeddingGenerator] = None
|
||||
|
||||
|
||||
def get_embedding_generator(
|
||||
use_api: bool = True,
|
||||
fallback_model_name: str = "paraphrase-multilingual-MiniLM-L12-v2",
|
||||
) -> EmbeddingGenerator:
|
||||
"""
|
||||
获取全局嵌入生成器单例
|
||||
|
||||
Args:
|
||||
use_api: 是否优先使用 API
|
||||
fallback_model_name: 回退本地模型名称
|
||||
|
||||
Returns:
|
||||
EmbeddingGenerator 实例
|
||||
"""
|
||||
global _global_generator
|
||||
if _global_generator is None:
|
||||
_global_generator = EmbeddingGenerator(
|
||||
use_api=use_api,
|
||||
fallback_model_name=fallback_model_name
|
||||
)
|
||||
return _global_generator
|
||||
391
src/memory_graph/utils/time_parser.py
Normal file
391
src/memory_graph/utils/time_parser.py
Normal file
@@ -0,0 +1,391 @@
|
||||
"""
|
||||
时间解析器：将相对时间转换为绝对时间
|
||||
|
||||
支持的时间表达：
|
||||
- 今天、明天、昨天、前天、后天
|
||||
- X天前、X天后
|
||||
- X小时前、X小时后
|
||||
- 上周、上个月、去年
|
||||
- 具体日期：2025-11-05, 11月5日
|
||||
- 时间点：早上8点、下午3点、晚上9点
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from datetime import datetime, timedelta
|
||||
from typing import Optional, Tuple
|
||||
|
||||
from src.common.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class TimeParser:
|
||||
"""
|
||||
时间解析器
|
||||
|
||||
负责将自然语言时间表达转换为标准化的绝对时间
|
||||
"""
|
||||
|
||||
def __init__(self, reference_time: Optional[datetime] = None):
|
||||
"""
|
||||
初始化时间解析器
|
||||
|
||||
Args:
|
||||
reference_time: 参考时间（通常是当前时间）
|
||||
"""
|
||||
self.reference_time = reference_time or datetime.now()
|
||||
|
||||
def parse(self, time_str: str) -> Optional[datetime]:
|
||||
"""
|
||||
解析时间字符串
|
||||
|
||||
Args:
|
||||
time_str: 时间字符串
|
||||
|
||||
Returns:
|
||||
标准化的datetime对象，如果解析失败则返回None
|
||||
"""
|
||||
if not time_str or not isinstance(time_str, str):
|
||||
return None
|
||||
|
||||
time_str = time_str.strip()
|
||||
|
||||
# 尝试各种解析方法
|
||||
parsers = [
|
||||
self._parse_relative_day,
|
||||
self._parse_days_ago,
|
||||
self._parse_hours_ago,
|
||||
self._parse_week_month_year,
|
||||
self._parse_specific_date,
|
||||
self._parse_time_of_day,
|
||||
]
|
||||
|
||||
for parser in parsers:
|
||||
try:
|
||||
result = parser(time_str)
|
||||
if result:
|
||||
logger.debug(f"时间解析: '{time_str}' → {result.isoformat()}")
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.debug(f"解析器 {parser.__name__} 失败: {e}")
|
||||
continue
|
||||
|
||||
logger.warning(f"无法解析时间: '{time_str}'，使用当前时间")
|
||||
return self.reference_time
|
||||
|
||||
def _parse_relative_day(self, time_str: str) -> Optional[datetime]:
|
||||
"""
|
||||
解析相对日期：今天、明天、昨天、前天、后天
|
||||
"""
|
||||
relative_days = {
|
||||
"今天": 0,
|
||||
"今日": 0,
|
||||
"明天": 1,
|
||||
"明日": 1,
|
||||
"昨天": -1,
|
||||
"昨日": -1,
|
||||
"前天": -2,
|
||||
"前日": -2,
|
||||
"后天": 2,
|
||||
"后日": 2,
|
||||
"大前天": -3,
|
||||
"大后天": 3,
|
||||
}
|
||||
|
||||
for keyword, days in relative_days.items():
|
||||
if keyword in time_str:
|
||||
result = self.reference_time + timedelta(days=days)
|
||||
# 保留原有时间，只改变日期
|
||||
return result.replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
|
||||
return None
|
||||
|
||||
def _parse_days_ago(self, time_str: str) -> Optional[datetime]:
|
||||
"""
|
||||
解析 X天前/X天后
|
||||
"""
|
||||
# 匹配：3天前、5天后、一天前
|
||||
pattern = r"([一二三四五六七八九十\d]+)天(前|后)"
|
||||
match = re.search(pattern, time_str)
|
||||
|
||||
if match:
|
||||
num_str, direction = match.groups()
|
||||
num = self._chinese_num_to_int(num_str)
|
||||
|
||||
if direction == "前":
|
||||
num = -num
|
||||
|
||||
result = self.reference_time + timedelta(days=num)
|
||||
return result.replace(hour=0, minute=0, second=0, microsecond=0)
|
||||
|
||||
return None
|
||||
|
||||
def _parse_hours_ago(self, time_str: str) -> Optional[datetime]:
|
||||
"""
|
||||
解析 X小时前/X小时后、X分钟前/X分钟后
|
||||
"""
|
||||
# 小时
|
||||
pattern_hour = r"([一二三四五六七八九十\d]+)小?时(前|后)"
|
||||
match = re.search(pattern_hour, time_str)
|
||||
|
||||
if match:
|
||||
num_str, direction = match.groups()
|
||||
num = self._chinese_num_to_int(num_str)
|
||||
|
||||
if direction == "前":
|
||||
num = -num
|
||||
|
||||
return self.reference_time + timedelta(hours=num)
|
||||
|
||||
# 分钟
|
||||
pattern_minute = r"([一二三四五六七八九十\d]+)分钟(前|后)"
|
||||
match = re.search(pattern_minute, time_str)
|
||||
|
||||
if match:
|
||||
num_str, direction = match.groups()
|
||||
num = self._chinese_num_to_int(num_str)
|
||||
|
||||
if direction == "前":
|
||||
num = -num
|
||||
|
||||
return self.reference_time + timedelta(minutes=num)
|
||||
|
||||
return None
|
||||
|
||||
def _parse_week_month_year(self, time_str: str) -> Optional[datetime]:
|
||||
"""
|
||||
解析：上周、上个月、去年、本周、本月、今年
|
||||
"""
|
||||
now = self.reference_time
|
||||
|
||||
if "上周" in time_str or "上星期" in time_str:
|
||||
return now - timedelta(days=7)
|
||||
|
||||
if "上个月" in time_str or "上月" in time_str:
|
||||
# 简单处理：减30天
|
||||
return now - timedelta(days=30)
|
||||
|
||||
if "去年" in time_str or "上年" in time_str:
|
||||
return now.replace(year=now.year - 1)
|
||||
|
||||
if "本周" in time_str or "这周" in time_str:
|
||||
# 返回本周一
|
||||
return now - timedelta(days=now.weekday())
|
||||
|
||||
if "本月" in time_str or "这个月" in time_str:
|
||||
return now.replace(day=1)
|
||||
|
||||
if "今年" in time_str or "这年" in time_str:
|
||||
return now.replace(month=1, day=1)
|
||||
|
||||
return None
|
||||
|
||||
def _parse_specific_date(self, time_str: str) -> Optional[datetime]:
|
||||
"""
|
||||
解析具体日期：
|
||||
- 2025-11-05
|
||||
- 2025/11/05
|
||||
- 11月5日
|
||||
- 11-05
|
||||
"""
|
||||
# ISO 格式：2025-11-05
|
||||
pattern_iso = r"(\d{4})[-/](\d{1,2})[-/](\d{1,2})"
|
||||
match = re.search(pattern_iso, time_str)
|
||||
if match:
|
||||
year, month, day = map(int, match.groups())
|
||||
return datetime(year, month, day)
|
||||
|
||||
# 中文格式：11月5日、11月5号
|
||||
pattern_cn = r"(\d{1,2})月(\d{1,2})[日号]"
|
||||
match = re.search(pattern_cn, time_str)
|
||||
if match:
|
||||
month, day = map(int, match.groups())
|
||||
# 使用参考时间的年份
|
||||
year = self.reference_time.year
|
||||
return datetime(year, month, day)
|
||||
|
||||
# 短格式：11-05（使用当前年份）
|
||||
pattern_short = r"(\d{1,2})[-/](\d{1,2})"
|
||||
match = re.search(pattern_short, time_str)
|
||||
if match:
|
||||
month, day = map(int, match.groups())
|
||||
year = self.reference_time.year
|
||||
return datetime(year, month, day)
|
||||
|
||||
return None
|
||||
|
||||
def _parse_time_of_day(self, time_str: str) -> Optional[datetime]:
|
||||
"""
|
||||
解析一天中的时间：
|
||||
- 早上、上午、中午、下午、晚上、深夜
|
||||
- 早上8点、下午3点
|
||||
- 8点、15点
|
||||
"""
|
||||
now = self.reference_time
|
||||
result = now.replace(minute=0, second=0, microsecond=0)
|
||||
|
||||
# 时间段映射
|
||||
time_periods = {
|
||||
"早上": 8,
|
||||
"早晨": 8,
|
||||
"上午": 10,
|
||||
"中午": 12,
|
||||
"下午": 15,
|
||||
"傍晚": 18,
|
||||
"晚上": 20,
|
||||
"深夜": 23,
|
||||
"凌晨": 2,
|
||||
}
|
||||
|
||||
# 先检查是否有具体时间点：早上8点、下午3点
|
||||
for period, default_hour in time_periods.items():
|
||||
pattern = rf"{period}(\d{{1,2}})点?"
|
||||
match = re.search(pattern, time_str)
|
||||
if match:
|
||||
hour = int(match.group(1))
|
||||
# 下午时间需要+12
|
||||
if period in ["下午", "晚上"] and hour < 12:
|
||||
hour += 12
|
||||
return result.replace(hour=hour)
|
||||
|
||||
# 检查时间段关键词
|
||||
for period, hour in time_periods.items():
|
||||
if period in time_str:
|
||||
return result.replace(hour=hour)
|
||||
|
||||
# 直接的时间点：8点、15点
|
||||
pattern = r"(\d{1,2})点"
|
||||
match = re.search(pattern, time_str)
|
||||
if match:
|
||||
hour = int(match.group(1))
|
||||
return result.replace(hour=hour)
|
||||
|
||||
return None
|
||||
|
||||
def _chinese_num_to_int(self, num_str: str) -> int:
|
||||
"""
|
||||
将中文数字转换为阿拉伯数字
|
||||
|
||||
Args:
|
||||
num_str: 中文数字字符串（如："一"、"十"、"3"）
|
||||
|
||||
Returns:
|
||||
整数
|
||||
"""
|
||||
# 如果已经是数字，直接返回
|
||||
if num_str.isdigit():
|
||||
return int(num_str)
|
||||
|
||||
# 中文数字映射
|
||||
chinese_nums = {
|
||||
"一": 1,
|
||||
"二": 2,
|
||||
"三": 3,
|
||||
"四": 4,
|
||||
"五": 5,
|
||||
"六": 6,
|
||||
"七": 7,
|
||||
"八": 8,
|
||||
"九": 9,
|
||||
"十": 10,
|
||||
"零": 0,
|
||||
}
|
||||
|
||||
if num_str in chinese_nums:
|
||||
return chinese_nums[num_str]
|
||||
|
||||
# 处理 "十X" 的情况（如"十五"=15）
|
||||
if num_str.startswith("十"):
|
||||
if len(num_str) == 1:
|
||||
return 10
|
||||
return 10 + chinese_nums.get(num_str[1], 0)
|
||||
|
||||
# 处理 "X十" 的情况（如"三十"=30）
|
||||
if "十" in num_str:
|
||||
parts = num_str.split("十")
|
||||
tens = chinese_nums.get(parts[0], 1) * 10
|
||||
ones = chinese_nums.get(parts[1], 0) if len(parts) > 1 and parts[1] else 0
|
||||
return tens + ones
|
||||
|
||||
# 默认返回1
|
||||
return 1
|
||||
|
||||
def format_time(self, dt: datetime, format_type: str = "iso") -> str:
|
||||
"""
|
||||
格式化时间
|
||||
|
||||
Args:
|
||||
dt: datetime对象
|
||||
format_type: 格式类型 ("iso", "cn", "relative")
|
||||
|
||||
Returns:
|
||||
格式化的时间字符串
|
||||
"""
|
||||
if format_type == "iso":
|
||||
return dt.isoformat()
|
||||
|
||||
elif format_type == "cn":
|
||||
return dt.strftime("%Y年%m月%d日 %H:%M:%S")
|
||||
|
||||
elif format_type == "relative":
|
||||
# 相对时间表达
|
||||
diff = self.reference_time - dt
|
||||
days = diff.days
|
||||
|
||||
if days == 0:
|
||||
hours = diff.seconds // 3600
|
||||
if hours == 0:
|
||||
minutes = diff.seconds // 60
|
||||
return f"{minutes}分钟前" if minutes > 0 else "刚刚"
|
||||
return f"{hours}小时前"
|
||||
elif days == 1:
|
||||
return "昨天"
|
||||
elif days == 2:
|
||||
return "前天"
|
||||
elif days < 7:
|
||||
return f"{days}天前"
|
||||
elif days < 30:
|
||||
weeks = days // 7
|
||||
return f"{weeks}周前"
|
||||
elif days < 365:
|
||||
months = days // 30
|
||||
return f"{months}个月前"
|
||||
else:
|
||||
years = days // 365
|
||||
return f"{years}年前"
|
||||
|
||||
return str(dt)
|
||||
|
||||
def parse_time_range(self, time_str: str) -> Tuple[Optional[datetime], Optional[datetime]]:
|
||||
"""
|
||||
解析时间范围：最近一周、最近3天
|
||||
|
||||
Args:
|
||||
time_str: 时间范围字符串
|
||||
|
||||
Returns:
|
||||
(start_time, end_time)
|
||||
"""
|
||||
pattern = r"最近(\d+)(天|周|月|年)"
|
||||
match = re.search(pattern, time_str)
|
||||
|
||||
if match:
|
||||
num, unit = match.groups()
|
||||
num = int(num)
|
||||
|
||||
unit_map = {"天": "days", "周": "weeks", "月": "days", "年": "days"}
|
||||
if unit == "周":
|
||||
num *= 7
|
||||
elif unit == "月":
|
||||
num *= 30
|
||||
elif unit == "年":
|
||||
num *= 365
|
||||
|
||||
end_time = self.reference_time
|
||||
start_time = end_time - timedelta(**{unit_map[unit]: num})
|
||||
|
||||
return (start_time, end_time)
|
||||
|
||||
return (None, None)
|
||||
Reference in New Issue
Block a user