上下文管理器与进一步移植

This commit is contained in:
2025-10-13 00:28:15 +08:00
parent 6dac9d5a7c
commit aa99aa7686
67 changed files with 438 additions and 4415 deletions

31
src/heurams/context.py Normal file
View File

@@ -0,0 +1,31 @@
"""
全局上下文管理模块
"""
from contextvars import ContextVar
from typing import Optional
from heurams.services.config import ConfigFile
config_var: ContextVar[ConfigFile] = ContextVar('config_var', default=ConfigFile("")) # 配置文件
runtime_var: ContextVar = ContextVar('runtime_var', default=None) # 运行时共享数据
class ConfigContext:
"""
功能完备的上下文管理器
用于临时切换配置的作用域, 支持嵌套使用
Example:
>>> with ConfigContext(test_config):
... get_daystamp() # 使用 test_config
>>> get_daystamp() # 恢复原配置
"""
def __init__(self, config_provider: ConfigFile):
self.config_provider = config_provider
self._token = None
def __enter__(self):
self._token = config_var.set(self.config_provider)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
config_var.reset(self._token) # type: ignore

View File

@@ -0,0 +1,126 @@
#!/usr/bin/env python3
import pathlib
import toml
import time
import heurams.services.timer as timer
class Electron:
"""电子: 记忆分析元数据及算法"""
algorithm = "SM-2" # 暂时使用 SM-2 算法进行记忆拟合, 考虑 SM-15 替代
def __init__(self, content: str, metadata: dict):
self.content = content
self.metadata = metadata
if metadata == {}:
# print("NULL")
self._default_init()
def _default_init(self):
defaults = {
'efactor': 2.5, # 易度系数, 越大越简单, 最大为5
'real_rept': 0, # (实际)重复次数
'rept': 0, # (有效)重复次数
'interval': 0, # 最佳间隔
'last_date': 0, # 上一次复习的时间戳
'next_date': 0, # 将要复习的时间戳
'is_activated': 0, # 激活状态
# *NOTE: 此处"时间戳"是以天为单位的整数, 即 UNIX 时间戳除以一天的秒数取整
'last_modify': time.time() # 最后修改时间戳(此处是UNIX时间戳)
}
self.metadata = defaults
def activate(self):
self.metadata['is_activated'] = 1
self.metadata['last_modify'] = time.time()
def modify(self, var: str, value):
if var in self.metadata:
self.metadata[var] = value
self.metadata['last_modify'] = time.time()
else:
print(f"警告: '{var}' 非已知元数据字段")
def revisor(self, quality: int = 5, is_new_activation: bool = False):
"""SM-2 算法迭代决策机制实现
根据 quality(0 ~ 5) 进行参数迭代最佳间隔
quality 由主程序评估
Args:
quality (int): 记忆保留率量化参数
"""
print(f"REVISOR: {quality}, {is_new_activation}")
if quality == -1:
return -1
self.metadata['efactor'] = self.metadata['efactor'] + (
0.1 - (5 - quality) * (0.08 + (5 - quality) * 0.02)
)
self.metadata['efactor'] = max(1.3, self.metadata['efactor'])
if quality < 3:
# 若保留率低于 3重置重复次数
self.metadata['rept'] = 0
self.metadata['interval'] = 0 # 设为0以便下面重新计算 I(1)
else:
self.metadata['rept'] += 1
self.metadata['real_rept'] += 1
if is_new_activation: # 初次激活
self.metadata['rept'] = 0
self.metadata['efactor'] = 2.5
if self.metadata['rept'] == 0: # 刚被重置或初次激活后复习
self.metadata['interval'] = 1 # I(1)
elif self.metadata['rept'] == 1:
self.metadata['interval'] = 6 # I(2) 经验公式
else:
self.metadata['interval'] = round(
self.metadata['interval'] * self.metadata['efactor']
)
self.metadata['last_date'] = timer.get_daystamp()
self.metadata['next_date'] = timer.get_daystamp() + self.metadata['interval']
self.metadata['last_modify'] = time.time()
def __str__(self):
return (
f"记忆单元预览 \n"
f"内容: '{self.content}' \n"
f"易度系数: {self.metadata['efactor']:.2f} \n"
f"已经重复的次数: {self.metadata['rept']} \n"
f"下次间隔: {self.metadata['interval']}\n"
f"下次复习日期时间戳: {self.metadata['next_date']}"
)
def __eq__(self, other):
if self.content == other.content:
return True
return False
def __hash__(self):
return hash(self.content)
def __getitem__(self, key):
if key == "content":
return self.content
if key in self.metadata:
return self.metadata[key]
else:
raise KeyError(f"Key '{key}' not found in metadata.")
def __setitem__(self, key, value):
if key == "content":
raise AttributeError("content 应为只读")
self.metadata[key] = value
self.metadata['last_modify'] = time.time()
def __iter__(self):
yield from self.metadata.keys()
def __len__(self):
return len(self.metadata)
@staticmethod
def placeholder():
return Electron("电子对象样例内容", {})

View File

View File

View File

View File

@@ -0,0 +1 @@
# 音频服务

View File

@@ -0,0 +1 @@
# 缓存服务

View File

@@ -0,0 +1,35 @@
# 配置文件服务
import pathlib
import toml
import typing
class ConfigFile:
def __init__(self, path: str):
self.path = pathlib.Path(path)
if not self.path.exists():
self.path.touch()
self.data = dict()
self._load()
def _load(self):
"""从文件加载配置数据"""
with open(self.path, 'r') as f:
try:
self.data = toml.load(f)
except toml.TomlDecodeError:
self.data = {}
def modify(self, key: str, value: typing.Any):
"""修改配置值并保存"""
self.data[key] = value
self.save()
def save(self, path: typing.Union[str, pathlib.Path] = ""):
"""保存配置到文件"""
save_path = pathlib.Path(path) if path else self.path
with open(save_path, 'w') as f:
toml.dump(self.data, f)
def get(self, key: str, default: typing.Any = None) -> typing.Any:
"""获取配置值,如果不存在返回默认值"""
return self.data.get(key, default)

View File

@@ -0,0 +1,5 @@
# 哈希服务
import hashlib
def get_md5(text):
return hashlib.md5(text.encode('utf-8')).hexdigest()

View File

@@ -0,0 +1,20 @@
# 时间服务
from heurams.context import config_var
import time
def get_daystamp() -> int:
"""获取当前日戳(以天为单位的整数时间戳)"""
time_override = config_var.get().get("daystamp_override", -1)
if time_override != -1:
return int(time_override)
return int((time.time() + config_var.get().get("timezone_offset")) // (24 * 3600))
def get_timestamp() -> float:
"""获取 UNIX 时间戳"""
# 搞这个类的原因是要支持可复现操作
time_override = config_var.get().get("timestamp_override", -1)
if time_override != -1:
return float(time_override)
return time.time()

View File

@@ -0,0 +1 @@
# 文本转语音服务

View File

@@ -0,0 +1 @@
# 版本控制集成服务