ajax/HeurAMS
0
1
Fork 0
Code Issues Pull Requests Actions Releases 14 Wiki Activity

fix: 增加日志

Browse Source
This commit is contained in:
Wang Zhiyu
2025-12-15 15:39:05 +08:00
parent 874494874c
commit 6efd041f72
51 changed files with 635 additions and 1992 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/heurams/interface/__main__.py
View File

@@ -4,6 +4,9 @@ from .screens.dashboard import DashboardScreen
from .screens.nucreator import NucleonCreatorScreen
from .screens.precache import PrecachingScreen
from .screens.about import AboutScreen
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class HeurAMSApp(App):
@@ -34,21 +37,27 @@ class HeurAMSApp(App):
def environment_check():
from pathlib import Path
logger.debug("检查环境路径")
for i in config_var.get()["paths"].values():
i = Path(i)
if not i.exists():
logger.info("创建目录: %s", i)
print(f"创建 {i}")
i.mkdir(exist_ok=True, parents=True)
else:
logger.debug("目录已存在: %s", i)
print(f"找到 {i}")
logger.debug("环境检查完成")
def is_subdir(parent, child):
try:
child.relative_to(parent)
logger.debug("is_subdir: %s 是 %s 的子目录", child, parent)
return 1
except:
logger.debug("is_subdir: %s 不是 %s 的子目录", child, parent)
return 0

5
src/heurams/kernel/algorithms/__init__.py
View File

@@ -1,4 +1,7 @@
from .sm2 import SM2Algorithm
from heurams.services.logger import get_logger
logger = get_logger(__name__)
__all__ = [
"SM2Algorithm",
@@ -8,3 +11,5 @@ algorithms = {
"SM-2": SM2Algorithm,
"supermemo2": SM2Algorithm,
}
logger.debug("算法模块初始化完成，注册的算法: %s", list(algorithms.keys()))

11
src/heurams/kernel/algorithms/base.py
View File

@@ -1,5 +1,8 @@
import heurams.services.timer as timer
from typing import TypedDict
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class BaseAlgorithm:
@@ -30,19 +33,27 @@ class BaseAlgorithm:
cls, algodata: dict, feedback: int = 5, is_new_activation: bool = False
) -> None:
"""迭代记忆数据"""
logger.debug("BaseAlgorithm.revisor 被调用，algodata keys: %s, feedback: %d, is_new_activation: %s",
list(algodata.keys()) if algodata else [], feedback, is_new_activation)
pass
@classmethod
def is_due(cls, algodata) -> int:
"""是否应该复习"""
logger.debug("BaseAlgorithm.is_due 被调用，algodata keys: %s",
list(algodata.keys()) if algodata else [])
return 1
@classmethod
def rate(cls, algodata) -> str:
"""获取评分信息"""
logger.debug("BaseAlgorithm.rate 被调用，algodata keys: %s",
list(algodata.keys()) if algodata else [])
return ""
@classmethod
def nextdate(cls, algodata) -> int:
"""获取下一次记忆时间戳"""
logger.debug("BaseAlgorithm.nextdate 被调用，algodata keys: %s",
list(algodata.keys()) if algodata else [])
return -1

5
src/heurams/kernel/algorithms/fsrs.py
View File

@@ -1 +1,6 @@
# FSRS 算法模块, 尚未就绪
from heurams.services.logger import get_logger
logger = get_logger(__name__)
logger.info("FSRS算法模块尚未实现")

32
src/heurams/kernel/algorithms/sm2.py
View File

@@ -1,6 +1,9 @@
from .base import BaseAlgorithm
import heurams.services.timer as timer
from typing import TypedDict
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class SM2Algorithm(BaseAlgorithm):
@@ -38,7 +41,10 @@ class SM2Algorithm(BaseAlgorithm):
Args:
quality (int): 记忆保留率量化参数
"""
logger.debug("SM2.revisor 开始，feedback: %d, is_new_activation: %s", feedback, is_new_activation)
if feedback == -1:
logger.debug("feedback 为 -1，跳过更新")
return
algodata[cls.algo_name]["efactor"] = algodata[cls.algo_name]["efactor"] + (
@@ -47,42 +53,62 @@ class SM2Algorithm(BaseAlgorithm):
algodata[cls.algo_name]["efactor"] = max(
1.3, algodata[cls.algo_name]["efactor"]
)
logger.debug("更新 efactor: %f", algodata[cls.algo_name]["efactor"])
if feedback < 3:
algodata[cls.algo_name]["rept"] = 0
algodata[cls.algo_name]["interval"] = 0
logger.debug("feedback < 3，重置 rept 和 interval")
else:
algodata[cls.algo_name]["rept"] += 1
logger.debug("递增 rept: %d", algodata[cls.algo_name]["rept"])
algodata[cls.algo_name]["real_rept"] += 1
logger.debug("递增 real_rept: %d", algodata[cls.algo_name]["real_rept"])
if is_new_activation:
algodata[cls.algo_name]["rept"] = 0
algodata[cls.algo_name]["efactor"] = 2.5
logger.debug("新激活，重置 rept 和 efactor")
if algodata[cls.algo_name]["rept"] == 0:
algodata[cls.algo_name]["interval"] = 1
logger.debug("rept=0，设置 interval=1")
elif algodata[cls.algo_name]["rept"] == 1:
algodata[cls.algo_name]["interval"] = 6
logger.debug("rept=1，设置 interval=6")
else:
algodata[cls.algo_name]["interval"] = round(
algodata[cls.algo_name]["interval"] * algodata[cls.algo_name]["efactor"]
)
logger.debug("rept>1，计算 interval: %d", algodata[cls.algo_name]["interval"])
algodata[cls.algo_name]["last_date"] = timer.get_daystamp()
algodata[cls.algo_name]["next_date"] = (
timer.get_daystamp() + algodata[cls.algo_name]["interval"]
)
algodata[cls.algo_name]["last_modify"] = timer.get_timestamp()
logger.debug("更新日期: last_date=%d, next_date=%d, last_modify=%f",
algodata[cls.algo_name]["last_date"],
algodata[cls.algo_name]["next_date"],
algodata[cls.algo_name]["last_modify"])
@classmethod
def is_due(cls, algodata):
return algodata[cls.algo_name]["next_date"] <= timer.get_daystamp()
result = algodata[cls.algo_name]["next_date"] <= timer.get_daystamp()
logger.debug("SM2.is_due: next_date=%d, current_daystamp=%d, result=%s",
algodata[cls.algo_name]["next_date"], timer.get_daystamp(), result)
return result
@classmethod
def rate(cls, algodata):
return str(algodata[cls.algo_name]["efactor"])
efactor = algodata[cls.algo_name]["efactor"]
logger.debug("SM2.rate: efactor=%f", efactor)
return str(efactor)
@classmethod
def nextdate(cls, algodata) -> int:
return algodata[cls.algo_name]["next_date"]
next_date = algodata[cls.algo_name]["next_date"]
logger.debug("SM2.nextdate: %d", next_date)
return next_date

4
src/heurams/kernel/particles/__init__.py
View File

@@ -4,6 +4,10 @@ Particle 模块 - 粒子对象系统
提供闪卡所需对象, 使用物理学粒子的领域驱动设计
"""
from heurams.services.logger import get_logger
logger = get_logger(__name__)
logger.debug("粒子模块已加载")
from .electron import Electron
from .nucleon import Nucleon
from .orbital import Orbital

29
src/heurams/kernel/particles/atom.py
View File

@@ -8,6 +8,9 @@ import toml
import json
import bidict
from heurams.context import config_var
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class AtomRegister(TypedDict):
@@ -33,8 +36,10 @@ class Atom:
"""
def __init__(self, ident=""):
logger.debug("创建 Atom 实例，ident: '%s'", ident)
self.ident = ident
atom_registry[ident] = self
logger.debug("Atom 已注册到全局注册表，当前注册表大小: %d", len(atom_registry))
# self.is_evaled = False
self.registry: AtomRegister = { # type: ignore
"nucleon": None,
@@ -48,12 +53,16 @@ class Atom:
"orbital_fmt": "toml",
}
self.do_eval()
logger.debug("Atom 初始化完成")
def link(self, key, value):
logger.debug("Atom.link: key='%s', value type: %s", key, type(value).__name__)
if key in self.registry.keys():
self.registry[key] = value
logger.debug("键 '%s' 已链接，触发 do_eval", key)
self.do_eval()
else:
logger.error("尝试链接不受支持的键: '%s'", key)
raise ValueError("不受支持的原子元数据链接操作")
def do_eval(self):
@@ -61,6 +70,7 @@ class Atom:
执行并以结果替换当前单元的所有 eval 语句
TODO: 带有限制的 eval, 异步/多线程执行避免堵塞
"""
logger.debug("Atom.do_eval 开始")
# eval 环境设置
def eval_with_env(s: str):
@@ -72,8 +82,10 @@ class Atom:
ret = "尚未链接对象"
try:
ret = str(eval(s))
logger.debug("eval 执行成功: '%s' -> '%s'", s, ret[:50] + '...' if len(ret) > 50 else ret)
except Exception as e:
ret = f"此 eval 实例发生错误: {e}"
logger.warning("eval 执行错误: '%s' -> %s", s, e)
return ret
def traverse(data, modifier):
@@ -90,37 +102,52 @@ class Atom:
else:
if isinstance(data, str):
if data.startswith("eval:"):
logger.debug("发现 eval 表达式: '%s'", data[5:])
return modifier(data[5:])
return data
traverse(self.registry["nucleon"], eval_with_env)
traverse(self.registry["orbital"], eval_with_env)
logger.debug("Atom.do_eval 完成")
def persist(self, key):
logger.debug("Atom.persist: key='%s'", key)
path: pathlib.Path | None = self.registry[key + "_path"]
if isinstance(path, pathlib.Path):
path = typing.cast(pathlib.Path, path)
logger.debug("持久化路径: %s, 格式: %s", path, self.registry[key + "_fmt"])
path.parent.mkdir(parents=True, exist_ok=True)
if self.registry[key + "_fmt"] == "toml":
with open(path, "w") as f:
toml.dump(self.registry[key], f)
logger.debug("TOML 数据已保存到: %s", path)
elif self.registry[key + "_fmt"] == "json":
with open(path, "w") as f:
json.dump(self.registry[key], f)
logger.debug("JSON 数据已保存到: %s", path)
else:
logger.error("不受支持的持久化格式: %s", self.registry[key + "_fmt"])
raise KeyError("不受支持的持久化格式")
else:
logger.error("路径未初始化: %s_path", key)
raise TypeError("对未初始化的路径对象操作")
def __getitem__(self, key):
logger.debug("Atom.__getitem__: key='%s'", key)
if key in self.registry:
return self.registry[key]
value = self.registry[key]
logger.debug("返回 value type: %s", type(value).__name__)
return value
logger.error("不支持的键: '%s'", key)
raise KeyError(f"不支持的键: {key}")
def __setitem__(self, key, value):
logger.debug("Atom.__setitem__: key='%s', value type: %s", key, type(value).__name__)
if key in self.registry:
self.registry[key] = value
logger.debug("键 '%s' 已设置", key)
else:
logger.error("不支持的键: '%s'", key)
raise KeyError(f"不支持的键: {key}")
@staticmethod

38
src/heurams/kernel/particles/electron.py
View File

@@ -1,6 +1,9 @@
import heurams.services.timer as timer
from heurams.context import config_var
from heurams.kernel.algorithms import algorithms
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class Electron:
@@ -14,45 +17,69 @@ class Electron:
algodata: 算法数据字典, 包含算法的各项参数和设置
algo: 使用的算法模块标识
"""
logger.debug("创建 Electron 实例，ident: '%s', algo_name: '%s'", ident, algo_name)
self.algodata = algodata
self.ident = ident
self.algo = algorithms[algo_name]
logger.debug("使用的算法类: %s", self.algo.__name__)
if self.algo not in self.algodata.keys():
self.algodata[self.algo] = {}
logger.debug("算法键 '%s' 不存在，已创建空字典", self.algo)
if not self.algodata[self.algo]:
logger.debug("算法数据为空，使用默认值初始化")
self._default_init(self.algo.defaults)
else:
logger.debug("算法数据已存在，跳过默认初始化")
logger.debug("Electron 初始化完成，algodata keys: %s", list(self.algodata.keys()))
def _default_init(self, defaults: dict):
"""默认初始化包装"""
logger.debug("Electron._default_init: 使用默认值，keys: %s", list(defaults.keys()))
self.algodata[self.algo] = defaults.copy()
def activate(self):
"""激活此电子"""
logger.debug("Electron.activate: 激活 ident='%s'", self.ident)
self.algodata[self.algo]["is_activated"] = 1
self.algodata[self.algo]["last_modify"] = timer.get_timestamp()
logger.debug("电子已激活，is_activated=1")
def modify(self, var: str, value):
"""修改 algodata[algo] 中子字典数据"""
logger.debug("Electron.modify: var='%s', value=%s", var, value)
if var in self.algodata[self.algo]:
self.algodata[self.algo][var] = value
self.algodata[self.algo]["last_modify"] = timer.get_timestamp()
logger.debug("变量 '%s' 已修改，更新 last_modify", var)
else:
logger.warning("'%s' 非已知元数据字段", var)
print(f"警告: '{var}' 非已知元数据字段")
def is_due(self):
"""是否应该复习"""
return self.algo.is_due(self.algodata)
logger.debug("Electron.is_due: 检查 ident='%s'", self.ident)
result = self.algo.is_due(self.algodata)
logger.debug("is_due 结果: %s", result)
return result
def is_activated(self):
return self.algodata[self.algo]["is_activated"]
result = self.algodata[self.algo]["is_activated"]
logger.debug("Electron.is_activated: ident='%s', 结果: %d", self.ident, result)
return result
def rate(self):
"评价"
return self.algo.rate(self.algodata)
logger.debug("Electron.rate: ident='%s'", self.ident)
result = self.algo.rate(self.algodata)
logger.debug("rate 结果: %s", result)
return result
def nextdate(self) -> int:
return self.algo.nextdate(self.algodata)
logger.debug("Electron.nextdate: ident='%s'", self.ident)
result = self.algo.nextdate(self.algodata)
logger.debug("nextdate 结果: %d", result)
return result
def revisor(self, quality: int = 5, is_new_activation: bool = False):
"""算法迭代决策机制实现
@@ -61,7 +88,10 @@ class Electron:
quality (int): 记忆保留率量化参数 (0-5)
is_new_activation (bool): 是否为初次激活
"""
logger.debug("Electron.revisor: ident='%s', quality=%d, is_new_activation=%s",
self.ident, quality, is_new_activation)
self.algo.revisor(self.algodata, quality, is_new_activation)
logger.debug("revisor 完成，更新后的 algodata: %s", self.algodata.get(self.algo, {}))
def __str__(self):
return (

13
src/heurams/kernel/particles/loader.py
View File

@@ -5,12 +5,17 @@ import pathlib
import toml
import json
from copy import deepcopy
from heurams.services.logger import get_logger
logger = get_logger(__name__)
def load_nucleon(path: pathlib.Path, fmt="toml"):
logger.debug("load_nucleon: 加载文件 %s, 格式: %s", path, fmt)
with open(path, "r") as f:
dictdata = dict()
dictdata = toml.load(f) # type: ignore
logger.debug("TOML 解析成功，keys: %s", list(dictdata.keys()))
lst = list()
nested_data = dict()
# 修正 toml 解析器的不管嵌套行为
@@ -24,12 +29,15 @@ def load_nucleon(path: pathlib.Path, fmt="toml"):
current[part] = {}
current = current[part]
current[parts[-1]] = value
logger.debug("处理元数据键: %s", key)
else:
nested_data[key] = value
logger.debug("嵌套数据处理完成，keys: %s", list(nested_data.keys()))
# print(nested_data)
for item, attr in nested_data.items():
if item == "__metadata__":
continue
logger.debug("处理项目: %s", item)
lst.append(
(
Nucleon(
@@ -38,6 +46,7 @@ def load_nucleon(path: pathlib.Path, fmt="toml"):
deepcopy(nested_data["__metadata__"]["orbital"]),
)
)
logger.debug("load_nucleon 完成，加载了 %d 个 Nucleon 对象", len(lst))
return lst
@@ -51,10 +60,14 @@ def load_electron(path: pathlib.Path, fmt="json") -> dict:
Returns:
dict: 键名是电子对象名称, 值是电子对象
"""
logger.debug("load_electron: 加载文件 %s, 格式: %s", path, fmt)
with open(path, "r") as f:
dictdata = dict()
dictdata = json.load(f) # type: ignore
logger.debug("JSON 解析成功，keys: %s", list(dictdata.keys()))
dic = dict()
for item, attr in dictdata.items():
logger.debug("处理电子项目: %s", item)
dic[item] = Electron(hasher.hash(item), attr)
logger.debug("load_electron 完成，加载了 %d 个 Electron 对象", len(dic))
return dic

21
src/heurams/kernel/particles/nucleon.py
View File

@@ -1,3 +1,8 @@
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class Nucleon:
"""原子核: 材料元数据"""
@@ -9,16 +14,24 @@ class Nucleon:
payload: 记忆内容信息
metadata: 可选元数据信息
"""
logger.debug("创建 Nucleon 实例，ident: '%s', payload keys: %s, metadata keys: %s",
ident, list(payload.keys()) if payload else [], list(metadata.keys()) if metadata else [])
self.metadata = metadata
self.payload = payload
self.ident = ident
logger.debug("Nucleon 初始化完成")
def __getitem__(self, key):
logger.debug("Nucleon.__getitem__: key='%s'", key)
if key == "ident":
logger.debug("返回 ident: '%s'", self.ident)
return self.ident
if key in self.payload:
return self.payload[key]
value = self.payload[key]
logger.debug("返回 payload['%s'], value type: %s", key, type(value).__name__)
return value
else:
logger.error("键 '%s' 未在 payload 中找到", key)
raise KeyError(f"Key '{key}' not found in payload.")
def __iter__(self):
@@ -35,14 +48,17 @@ class Nucleon:
执行并以结果替换当前单元的所有 eval 语句
TODO: 带有限制的 eval, 异步/多线程执行避免堵塞
"""
logger.debug("Nucleon.do_eval 开始")
# eval 环境设置
def eval_with_env(s: str):
try:
nucleon = self
ret = str(eval(s))
logger.debug("eval 执行成功: '%s' -> '%s'", s, ret[:50] + '...' if len(ret) > 50 else ret)
except Exception as e:
ret = f"此 eval 实例发生错误: {e}"
logger.warning("eval 执行错误: '%s' -> %s", s, e)
return ret
def traverse(data, modifier):
@@ -59,13 +75,16 @@ class Nucleon:
else:
if isinstance(data, str):
if data.startswith("eval:"):
logger.debug("发现 eval 表达式: '%s'", data[5:])
return modifier(data[5:])
return data
traverse(self.payload, eval_with_env)
traverse(self.metadata, eval_with_env)
logger.debug("Nucleon.do_eval 完成")
@staticmethod
def placeholder():
"""生成一个占位原子核"""
logger.debug("创建 Nucleon 占位符")
return Nucleon("核子对象样例内容", {})

4
src/heurams/kernel/particles/orbital.py
View File

@@ -1,4 +1,8 @@
from typing import TypedDict
from heurams.services.logger import get_logger
logger = get_logger(__name__)
logger.debug("Orbital 类型定义模块已加载")
class OrbitalSchedule(TypedDict):

16
src/heurams/kernel/particles/probe.py
View File

@@ -1,17 +1,26 @@
from heurams.context import config_var
import pathlib
from heurams.services.logger import get_logger
logger = get_logger(__name__)
def probe_by_filename(filename):
"""探测指定文件 (无扩展名) 的所有信息"""
logger.debug("probe_by_filename: 探测文件 '%s'", filename)
paths: dict = config_var.get().get("paths")
logger.debug("配置路径: %s", paths)
formats = ["toml", "json"]
result = {}
for item, attr in paths.items():
for i in formats:
attr: pathlib.Path = pathlib.Path(attr) / filename + "." + i
if attr.exists():
logger.debug("找到文件: %s", attr)
result[item.replace("_dir", "")] = str(attr)
else:
logger.debug("文件不存在: %s", attr)
logger.debug("probe_by_filename 结果: %s", result)
return result
@@ -24,17 +33,24 @@ def probe_all(is_stem=1):
Returns:
dict: 有三项, 每一项的键名都是文件组类型, 值都是文件组列表, 只包含文件名
"""
logger.debug("probe_all: 开始探测，is_stem=%d", is_stem)
paths: dict = config_var.get().get("paths")
logger.debug("配置路径: %s", paths)
result = {}
for item, attr in paths.items():
attr: pathlib.Path = pathlib.Path(attr)
result[item.replace("_dir", "")] = list()
logger.debug("扫描目录: %s", attr)
file_count = 0
for i in attr.iterdir():
if not i.is_dir():
file_count += 1
if is_stem:
result[item.replace("_dir", "")].append(str(i.stem))
else:
result[item.replace("_dir", "")].append(str(i.name))
logger.debug("目录 %s 中找到 %d 个文件", attr, file_count)
logger.debug("probe_all 完成，结果 keys: %s", list(result.keys()))
return result

4
src/heurams/kernel/puzzles/__init__.py
View File

@@ -4,6 +4,9 @@ Puzzle 模块 - 谜题生成系统
提供多种类型的谜题生成器，支持从字符串、字典等数据源导入题目
"""
from heurams.services.logger import get_logger
logger = get_logger(__name__)
from .base import BasePuzzle
from .cloze import ClozePuzzle
from .mcq import MCQPuzzle
@@ -38,6 +41,7 @@ def create_by_dict(config_dict: dict) -> BasePuzzle:
Raises:
ValueError: 当配置无效时抛出
"""
logger.debug("puzzles.create_by_dict: config_dict keys=%s", list(config_dict.keys()))
puzzle_type = config_dict.get("type")
if puzzle_type == "cloze":

7
src/heurams/kernel/puzzles/base.py
View File

@@ -1,9 +1,16 @@
# base.py
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class BasePuzzle:
"""谜题基类"""
def refresh(self):
logger.debug("BasePuzzle.refresh 被调用（未实现）")
raise NotImplementedError("谜题对象未实现 refresh 方法")
def __str__(self):
logger.debug("BasePuzzle.__str__ 被调用")
return f"谜题: {type(self).__name__}"

12
src/heurams/kernel/puzzles/cloze.py
View File

@@ -1,5 +1,8 @@
from .base import BasePuzzle
import random
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class ClozePuzzle(BasePuzzle):
@@ -11,20 +14,27 @@ class ClozePuzzle(BasePuzzle):
"""
def __init__(self, text: str, min_denominator: int, delimiter: str = "/"):
logger.debug("ClozePuzzle.__init__: text length=%d, min_denominator=%d, delimiter='%s'",
len(text), min_denominator, delimiter)
self.text = text
self.min_denominator = min_denominator
self.wording = "填空题 - 尚未刷新谜题"
self.answer = ["填空题 - 尚未刷新谜题"]
self.delimiter = delimiter
logger.debug("ClozePuzzle 初始化完成")
def refresh(self): # 刷新谜题
logger.debug("ClozePuzzle.refresh 开始")
placeholder = "___SLASH___"
tmp_text = self.text.replace(self.delimiter, placeholder)
words = tmp_text.split(placeholder)
if not words:
logger.warning("ClozePuzzle.refresh: 无单词可处理")
return
words = [word for word in words if word]
logger.debug("ClozePuzzle.refresh: 分割出 %d 个单词", len(words))
num_blanks = min(max(1, len(words) // self.min_denominator), len(words))
logger.debug("ClozePuzzle.refresh: 需要生成 %d 个填空", num_blanks)
indices_to_blank = random.sample(range(len(words)), num_blanks)
indices_to_blank.sort()
blanked_words = list(words)
@@ -34,6 +44,8 @@ class ClozePuzzle(BasePuzzle):
answer.append(words[index])
self.answer = answer
self.wording = "".join(blanked_words)
logger.debug("ClozePuzzle.refresh 完成，生成 %d 个填空", len(answer))
def __str__(self):
logger.debug("ClozePuzzle.__str__ 被调用")
return f"{self.wording}\n{str(self.answer)}"

5
src/heurams/kernel/puzzles/mcq.py
View File

@@ -2,6 +2,9 @@
from .base import BasePuzzle
import random
from typing import List, Dict, Optional, Union
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class MCQPuzzle(BasePuzzle):
@@ -35,6 +38,7 @@ class MCQPuzzle(BasePuzzle):
max_riddles_num: 每次生成的最大题目数量, 范围限制在1-5之间
prefix: 题目前缀文本, 会显示在每个题目之前
"""
logger.debug("MCQPuzzle.__init__: mapping size=%d, jammer size=%d, max_riddles_num=%d", len(mapping), len(jammer), max_riddles_num)
self.prefix = prefix
self.mapping = mapping
self.max_riddles_num = max(1, min(max_riddles_num, 5))
@@ -79,6 +83,7 @@ class MCQPuzzle(BasePuzzle):
Raises:
ValueError: 当mapping为空时不会抛出异常, 但会设置空谜题状态
"""
logger.debug("MCQPuzzle.refresh 开始，mapping size=%d", len(self.mapping))
if not self.mapping:
self._set_empty_puzzle()
return

5
src/heurams/kernel/puzzles/recognition.py
View File

@@ -1,13 +1,18 @@
# mcq.py
from .base import BasePuzzle
import random
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class RecognitionPuzzle(BasePuzzle):
"""识别占位符"""
def __init__(self) -> None:
logger.debug("RecognitionPuzzle.__init__")
super().__init__()
def refresh(self):
logger.debug("RecognitionPuzzle.refresh（空实现）")
pass

5
src/heurams/kernel/reactor/__init__.py
View File

@@ -2,5 +2,10 @@ from .states import PhaserState, ProcessionState
from .procession import Procession
from .fission import Fission
from .phaser import Phaser
from heurams.services.logger import get_logger
logger = get_logger(__name__)
__all__ = ["PhaserState", "ProcessionState", "Procession", "Fission", "Phaser"]
logger.debug("反应堆模块已加载")

12
src/heurams/kernel/reactor/phaser.py
View File

@@ -3,12 +3,16 @@
import heurams.kernel.particles as pt
from .states import PhaserState, ProcessionState
from .procession import Procession
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class Phaser:
"""移相器: 全局调度阶段管理器"""
def __init__(self, atoms: list[pt.Atom]) -> None:
logger.debug("Phaser.__init__: 原子数量=%d", len(atoms))
new_atoms = list()
old_atoms = list()
self.state = PhaserState.UNSURE
@@ -17,22 +21,30 @@ class Phaser:
new_atoms.append(i)
else:
old_atoms.append(i)
logger.debug("新原子数量=%d, 旧原子数量=%d", len(new_atoms), len(old_atoms))
self.processions = list()
if len(old_atoms):
self.processions.append(
Procession(old_atoms, PhaserState.QUICK_REVIEW, "初始复习")
)
logger.debug("创建初始复习 Procession")
if len(new_atoms):
self.processions.append(
Procession(new_atoms, PhaserState.RECOGNITION, "新记忆")
)
logger.debug("创建新记忆 Procession")
self.processions.append(Procession(atoms, PhaserState.FINAL_REVIEW, "总体复习"))
logger.debug("创建总体复习 Procession")
logger.debug("Phaser 初始化完成，processions 数量=%d", len(self.processions))
def current_procession(self):
logger.debug("Phaser.current_procession 被调用")
for i in self.processions:
i: Procession
if not i.state == ProcessionState.FINISHED:
self.state = i.phase
logger.debug("找到未完成的 Procession: phase=%s", i.phase)
return i
self.state = PhaserState.FINISHED
logger.debug("所有 Procession 已完成，状态设置为 FINISHED")
return 0

31
src/heurams/kernel/reactor/procession.py
View File

@@ -1,11 +1,16 @@
import heurams.kernel.particles as pt
from .states import PhaserState, ProcessionState
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class Procession:
"""队列: 标识单次记忆流程"""
def __init__(self, atoms: list, phase: PhaserState, name: str = ""):
logger.debug("Procession.__init__: 原子数量=%d, phase=%s, name='%s'",
len(atoms), phase.value, name)
self.atoms = atoms
self.queue = atoms.copy()
self.current_atom = atoms[0]
@@ -13,36 +18,52 @@ class Procession:
self.name = name
self.phase = phase
self.state: ProcessionState = ProcessionState.RUNNING
logger.debug("Procession 初始化完成，队列长度=%d", len(self.queue))
def forward(self, step=1):
logger.debug("Procession.forward: step=%d, 当前 cursor=%d", step, self.cursor)
self.cursor += step
if self.cursor == len(self.queue):
self.state = ProcessionState.FINISHED
logger.debug("Procession 已完成")
else:
self.state = ProcessionState.RUNNING
try:
print(self.cursor)
logger.debug("cursor 更新为: %d", self.cursor)
self.current_atom = self.queue[self.cursor]
logger.debug("当前原子更新为: %s", self.current_atom.ident)
return 1 # 成功
except IndexError as e:
#print(f"{e}")
logger.debug("IndexError: %s", e)
self.state = ProcessionState.FINISHED
logger.debug("Procession 因索引错误而完成")
return 0
def append(self, atom=None):
if atom == None:
atom = self.current_atom
logger.debug("Procession.append: atom=%s", atom.ident if atom else "None")
if self.queue[len(self.queue) - 1] != atom or len(self) <= 1:
self.queue.append(atom)
logger.debug("原子已追加到队列，新队列长度=%d", len(self.queue))
else:
logger.debug("原子未追加（重复或队列长度<=1）")
def __len__(self):
return len(self.queue) - self.cursor
length = len(self.queue) - self.cursor
logger.debug("Procession.__len__: 剩余长度=%d", length)
return length
def process(self):
logger.debug("Procession.process: cursor=%d", self.cursor)
return self.cursor
def total_length(self):
return len(self.queue)
total = len(self.queue)
logger.debug("Procession.total_length: %d", total)
return total
def is_empty(self):
return len(self.queue)
empty = len(self.queue)
logger.debug("Procession.is_empty: %d", empty)
return empty

5
src/heurams/kernel/reactor/states.py
View File

@@ -1,4 +1,7 @@
from enum import Enum, auto
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class PhaserState(Enum):
@@ -12,3 +15,5 @@ class PhaserState(Enum):
class ProcessionState(Enum):
RUNNING = auto()
FINISHED = auto()
logger.debug("状态枚举定义已加载")

4
src/heurams/providers/audio/__init__.py
View File

@@ -1,6 +1,9 @@
# 音频播放器, 必须基于文件操作
from . import termux_audio
from . import playsound_audio
from heurams.services.logger import get_logger
logger = get_logger(__name__)
__all__ = [
"termux_audio",
@@ -8,3 +11,4 @@ __all__ = [
]
providers = {"termux": termux_audio, "playsound": playsound_audio}
logger.debug("音频 providers 已注册: %s", list(providers.keys()))

11
src/heurams/providers/audio/playsound_audio.py
View File

@@ -6,7 +6,16 @@
import os
import pathlib
import playsound
from heurams.services.logger import get_logger
logger = get_logger(__name__)
def play_by_path(path: pathlib.Path):
playsound.playsound(str(path))
logger.debug("playsound_audio.play_by_path: 开始播放 %s", path)
try:
playsound.playsound(str(path))
logger.debug("播放完成: %s", path)
except Exception as e:
logger.error("播放失败: %s, 错误: %s", path, e)
raise

6
src/heurams/providers/audio/protocol.py
View File

@@ -1,6 +1,12 @@
from typing import Protocol
import pathlib
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class PlayFunctionProtocol(Protocol):
def __call__(self, path: pathlib.Path) -> None: ...
logger.debug("音频协议模块已加载")

11
src/heurams/providers/audio/termux_audio.py
View File

@@ -5,9 +5,18 @@
import os
import pathlib
from heurams.services.logger import get_logger
logger = get_logger(__name__)
# from .protocol import PlayFunctionProtocol
def play_by_path(path: pathlib.Path):
os.system(f"play-audio {path}")
logger.debug("termux_audio.play_by_path: 开始播放 %s", path)
try:
os.system(f"play-audio {path}")
logger.debug("播放命令已执行: %s", path)
except Exception as e:
logger.error("播放失败: %s, 错误: %s", path, e)
raise

5
src/heurams/providers/llm/__init__.py
View File

@@ -1 +1,6 @@
# 大语言模型
from heurams.services.logger import get_logger
logger = get_logger(__name__)
logger.debug("LLM providers 模块已加载")

5
src/heurams/providers/llm/base.py
View File

@@ -0,0 +1,5 @@
from heurams.services.logger import get_logger
logger = get_logger(__name__)
logger.debug("LLM 基类模块已加载")

5
src/heurams/providers/llm/openai.py
View File

@@ -0,0 +1,5 @@
from heurams.services.logger import get_logger
logger = get_logger(__name__)
logger.debug("OpenAI provider 模块已加载（未实现）")

5
src/heurams/providers/tts/__init__.py
View File

@@ -1,5 +1,8 @@
from .base import BaseTTS
from .edge_tts import EdgeTTS
from heurams.services.logger import get_logger
logger = get_logger(__name__)
__all__ = [
"BaseTTS",
@@ -10,3 +13,5 @@ providers = {
"basetts": BaseTTS,
"edgetts": EdgeTTS,
}
logger.debug("TTS providers 已注册: %s", list(providers.keys()))

5
src/heurams/providers/tts/base.py
View File

@@ -1,4 +1,7 @@
import pathlib
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class BaseTTS:
@@ -7,4 +10,6 @@ class BaseTTS:
@classmethod
def convert(cls, text: str, path: pathlib.Path | str = "") -> pathlib.Path:
"""path 是可选参数, 不填则自动返回生成文件路径"""
logger.debug("BaseTTS.convert: text length=%d, path=%s", len(text), path)
logger.warning("BaseTTS.convert 是基类方法，未实现具体功能")
return path # type: ignore

22
src/heurams/providers/tts/edge_tts.py
View File

@@ -1,6 +1,9 @@
from .base import BaseTTS
import pathlib
import edge_tts
from heurams.services.logger import get_logger
logger = get_logger(__name__)
class EdgeTTS(BaseTTS):
@@ -8,9 +11,16 @@ class EdgeTTS(BaseTTS):
@classmethod
def convert(cls, text, path: pathlib.Path | str = "") -> pathlib.Path:
communicate = edge_tts.Communicate(
text,
"zh-CN-YunjianNeural",
)
communicate.save_sync(str(path))
return path # type: ignore
logger.debug("EdgeTTS.convert: text length=%d, path=%s", len(text), path)
try:
communicate = edge_tts.Communicate(
text,
"zh-CN-YunjianNeural",
)
logger.debug("EdgeTTS 通信对象创建成功，正在保存音频")
communicate.save_sync(str(path))
logger.debug("EdgeTTS 音频已保存到: %s", path)
return path # type: ignore
except Exception as e:
logger.error("EdgeTTS.convert 失败: %s", e)
raise

4
src/heurams/services/audio_service.py
View File

@@ -2,5 +2,9 @@
from heurams.context import config_var
from heurams.providers.audio import providers as prov
from typing import Callable
from heurams.services.logger import get_logger
logger = get_logger(__name__)
play_by_path: Callable = prov[config_var.get()["services"]["audio"]].play_by_path
logger.debug("音频服务初始化完成，使用 provider: %s", config_var.get()["services"]["audio"])

13
src/heurams/services/hasher.py
View File

@@ -1,10 +1,19 @@
# 哈希服务
import hashlib
from heurams.services.logger import get_logger
logger = get_logger(__name__)
def get_md5(text):
return hashlib.md5(text.encode("utf-8")).hexdigest()
logger.debug("计算MD5哈希，输入长度: %d", len(text))
result = hashlib.md5(text.encode("utf-8")).hexdigest()
logger.debug("MD5哈希结果: %s...", result[:8])
return result
def hash(text):
return hashlib.md5(text.encode("utf-8")).hexdigest()
logger.debug("计算哈希，输入长度: %d", len(text))
result = hashlib.md5(text.encode("utf-8")).hexdigest()
logger.debug("哈希结果: %s...", result[:8])
return result

13
src/heurams/services/timer.py
View File

@@ -1,15 +1,21 @@
# 时间服务
from heurams.context import config_var
import time
from heurams.services.logger import get_logger
logger = get_logger(__name__)
def get_daystamp() -> int:
"""获取当前日戳(以天为单位的整数时间戳)"""
time_override = config_var.get().get("daystamp_override", -1)
if time_override != -1:
logger.debug("使用覆盖的日戳: %d", time_override)
return int(time_override)
return int((time.time() + config_var.get().get("timezone_offset")) // (24 * 3600))
result = int((time.time() + config_var.get().get("timezone_offset")) // (24 * 3600))
logger.debug("计算日戳: %d", result)
return result
def get_timestamp() -> float:
@@ -17,6 +23,9 @@ def get_timestamp() -> float:
# 搞这个类的原因是要支持可复现操作
time_override = config_var.get().get("timestamp_override", -1)
if time_override != -1:
logger.debug("使用覆盖的时间戳: %f", time_override)
return float(time_override)
return time.time()
result = time.time()
logger.debug("获取当前时间戳: %f", result)
return result

4
src/heurams/services/tts_service.py
View File

@@ -2,5 +2,9 @@
from heurams.context import config_var
from heurams.providers.tts import TTSs
from typing import Callable
from heurams.services.logger import get_logger
logger = get_logger(__name__)
convert: Callable = TTSs[config_var.get().get("tts_provider")]
logger.debug("TTS服务初始化完成，使用 provider: %s", config_var.get().get("tts_provider"))

5
src/heurams/services/version.py
View File

@@ -1,5 +1,10 @@
# 版本控制集成服务
from heurams.services.logger import get_logger
logger = get_logger(__name__)
ver = "0.4.0"
stage = "prototype"
codename = "fledge" # 雏鸟, 0.4.x 版本
logger.info("HeurAMS 版本: %s (%s), 阶段: %s", ver, codename, stage)