Compare commits
11 Commits
Author | SHA1 | Date | |
---|---|---|---|
5a096e4b4f | |||
65491117d3 | |||
c82eedde82 | |||
697e3b2b8f | |||
11eff7da43 | |||
c93bcdd489 | |||
bb99b0a0b7 | |||
6293b69ef0 | |||
afb7252f71 | |||
5e96fc8138 | |||
e64d1711d0 |
1
.gitignore
vendored
1
.gitignore
vendored
@@ -1,3 +1,4 @@
|
||||
.devflag
|
||||
.vscode
|
||||
.directory
|
||||
__pycache__/
|
||||
|
@@ -1,2 +0,0 @@
|
||||
def playsound(p):
|
||||
print(p)
|
@@ -5,3 +5,16 @@
|
||||
```bash
|
||||
nuitka --clang --jobs=6 --standalone --onefile main.py
|
||||
```
|
||||
|
||||
## 打开调试日志
|
||||
|
||||
分别运行
|
||||
|
||||
```shell
|
||||
textual console -x SYSTEM -x EVENT -x DEBUG -x INFO
|
||||
```
|
||||
|
||||
```shell
|
||||
textual run --dev main.py
|
||||
```
|
||||
|
||||
|
20
README.md
20
README.md
@@ -2,8 +2,22 @@
|
||||
> 形人而我无形,**则我专而敌分**
|
||||
|
||||
## 概述
|
||||
"潜进" (HeurAMS: Heuristic Auxiliary Memorizing Scheduler, 启发式记忆辅助调度器) 是为习题册, 古诗词, 及其他问答/记忆/理解型知识设计的辅助记忆软件, 提供动态规划的优化记忆方案
|
||||
|
||||
"潜进" (HeurAMS) 是为习题册, 古诗词, 及其他问答/记忆/理解型知识设计的辅助记忆软件, 提供动态规划的优化记忆方案
|
||||
## 关于此仓库
|
||||
"潜进" 软件组项目包含多个子项目:
|
||||
- 此仓库包含了 "潜进" 项目的核心和基于 Textual 的基本用户界面的实现
|
||||
- 关于基于 Flutter 的现代用户界面, 请参阅 "潜进-F" (HeurAMS-F) 仓库
|
||||
- 关于数据同步实现, 请参阅 "潜进-S" (HeurSync) 仓库
|
||||
- 关于云端文档源实现, 请参阅 "潜进-R" (HeurRepo) 仓库
|
||||
|
||||
## 开发计划
|
||||
0.1.x: 简易调度器实现与最小原型
|
||||
0.2.x: 使用 Textual 构建 TUI, 项目可行性验证与采用 SM-2 原始算法用户自评估的原型
|
||||
0.3.x (当前): 基本数据结构, 基于 SM-2 改进算法的自动复习测评评估与遵从 IoC 设计的功能实现, 重点设计古诗文记忆理解功能, 以及 TUI 界面实现, 简单的语言模型集成
|
||||
0.4.x: 更新文件格式, 引入动态数据结构(自动内容生成), 深度语言模型集成
|
||||
0.5.x: 引入云同步与文档源
|
||||
0.6.x: 引入其他算法接口, 与跨语言库引入, 使用 Flutter 构建跨平台现代客户端
|
||||
|
||||
## 技术集成与特性
|
||||
|
||||
@@ -22,13 +36,13 @@
|
||||
- 语法分析:接入生成式人工智能, 支持古文结构交互式解析
|
||||
- 自然语音:集成微软神经网络文本转语音 (TTS) 技术
|
||||
|
||||
### 现代用户界面
|
||||
### 实用用户界面
|
||||
|
||||
- 响应式 Textual 框架构建的跨平台 TUI 界面
|
||||
- 支持触屏/鼠标/键盘多操作模式
|
||||
- 简洁直观的复习流程设计
|
||||
|
||||
## 屏幕截图
|
||||
## 屏幕截图 (基本用户界面)
|
||||
|
||||
> 单击图片以放大
|
||||
|
||||
|
10
auxiliary.py
10
auxiliary.py
@@ -2,8 +2,9 @@ import time
|
||||
import pathlib
|
||||
import toml
|
||||
import typing
|
||||
import playsound
|
||||
from playsound import playsound
|
||||
import threading
|
||||
import hashlib
|
||||
import edge_tts as tts
|
||||
|
||||
class ConfigFile:
|
||||
@@ -47,9 +48,9 @@ def action_play_voice(content):
|
||||
"zh-CN-YunjianNeural",
|
||||
)
|
||||
communicate.save_sync(
|
||||
f"./cache/voice/{content}"
|
||||
f"./cache/voice/{content}.wav"
|
||||
)
|
||||
playsound()
|
||||
playsound(f"./cache/voice/{content}.wav")
|
||||
threading.Thread(target=play).start()
|
||||
|
||||
def get_daystamp() -> int:
|
||||
@@ -61,3 +62,6 @@ def get_daystamp() -> int:
|
||||
return int(time_override)
|
||||
|
||||
return int(time.time() // (24 * 3600))
|
||||
|
||||
def get_md5(text):
|
||||
return hashlib.md5(text.encode('utf-8')).hexdigest()
|
||||
|
@@ -0,0 +1,8 @@
|
||||
# [调试] 将更改保存到文件
|
||||
save = 1
|
||||
# [调试] 覆写时间
|
||||
time_override = -1
|
||||
# [调试] 一键通过
|
||||
quick_pass = 0
|
||||
# 对于每个项目的新记忆核子数量
|
||||
tasked_number = 6
|
0
electron/赤壁赋.toml
Normal file
0
electron/赤壁赋.toml
Normal file
0
llmplugin.py
Normal file
0
llmplugin.py
Normal file
247
main.py
247
main.py
@@ -1,253 +1,12 @@
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.widgets import (
|
||||
Header,
|
||||
Footer,
|
||||
ListView,
|
||||
ProgressBar,
|
||||
DirectoryTree,
|
||||
ListItem,
|
||||
Label,
|
||||
Static,
|
||||
Button,
|
||||
)
|
||||
from textual.containers import Container, Horizontal, Center
|
||||
from textual.screen import Screen
|
||||
import pathlib
|
||||
import threading
|
||||
import edge_tts as tts
|
||||
from playsound import playsound
|
||||
import particles as pt
|
||||
from reactor import Reactor, Apparatus
|
||||
import auxiliary as aux
|
||||
import compositions as compo
|
||||
import builtins
|
||||
|
||||
ver = "0.3.2"
|
||||
|
||||
config = aux.ConfigFile("config.toml")
|
||||
|
||||
|
||||
class MemScreen(Screen):
|
||||
BINDINGS = [
|
||||
("d", "toggle_dark", "改变色调"),
|
||||
("q", "pop_screen", "返回主菜单"),
|
||||
("v", "play_voice", "朗读"),
|
||||
]
|
||||
if config.get("quick_pass"):
|
||||
BINDINGS.append(("k", "quick_pass", "快速通过[调试]"))
|
||||
btn = dict()
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
nucleon_file: pt.NucleonUnion,
|
||||
electron_file: pt.ElectronUnion,
|
||||
tasked_num,
|
||||
):
|
||||
super().__init__(name=None, id=None, classes=None)
|
||||
self.reactor = Reactor(nucleon_file, electron_file, self, tasked_num)
|
||||
self.stage = 1
|
||||
self.stage += self.reactor.set_round_templated(self.stage)
|
||||
self.reactor.forward()
|
||||
self.compo = next(self.reactor.current_appar)
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
if type(self.compo).__name__ == "Recognition":
|
||||
self.action_play_voice()
|
||||
yield Header(show_clock=True)
|
||||
with Center():
|
||||
yield Static(
|
||||
f"{len(self.reactor.procession) - self.reactor.index}/{len(self.reactor.procession)}"
|
||||
)
|
||||
yield from self.compo.compose()
|
||||
yield Footer()
|
||||
|
||||
def on_mount(self):
|
||||
pass
|
||||
|
||||
def on_button_pressed(self, event):
|
||||
ret = self.compo.handler(event, "button")
|
||||
self._forward_judge(ret)
|
||||
|
||||
def _forward_judge(self, ret):
|
||||
if ret == -1:
|
||||
return
|
||||
if ret == 0:
|
||||
try:
|
||||
self.compo = next(self.reactor.current_appar)
|
||||
self.refresh_ui()
|
||||
except StopIteration:
|
||||
nxt = self.reactor.forward(1)
|
||||
try:
|
||||
self.compo = next(self.reactor.current_appar)
|
||||
except:
|
||||
pass
|
||||
if nxt == -1:
|
||||
if self.reactor.round_set == 0:
|
||||
if self.stage == 4:
|
||||
if config.get("save"):
|
||||
self.reactor.save()
|
||||
self.compo = compo.Finished(
|
||||
self, None, pt.Atom.placeholder()
|
||||
)
|
||||
self.refresh_ui()
|
||||
else:
|
||||
self.reactor.set_round_templated(self.stage)
|
||||
self.reactor.forward(1)
|
||||
self.stage += 1
|
||||
self.compo = next(self.reactor.current_appar)
|
||||
self.refresh_ui()
|
||||
return
|
||||
return
|
||||
else:
|
||||
self.refresh_ui()
|
||||
return
|
||||
if ret == 1:
|
||||
self.refresh_ui()
|
||||
return
|
||||
|
||||
def refresh_ui(self):
|
||||
self.call_later(self.recompose)
|
||||
print(type(self.compo).__name__)
|
||||
|
||||
def action_play_voice(self):
|
||||
print("VOICE")
|
||||
def play():
|
||||
cache_dir = pathlib.Path(f"./cache/voice/")
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
cache = cache_dir / f"{self.reactor.current_atom[1].content.replace('/','')}.wav"
|
||||
if not cache.exists():
|
||||
communicate = tts.Communicate(
|
||||
self.reactor.current_atom[1].content.replace("/", ""),
|
||||
"zh-CN-YunjianNeural",
|
||||
)
|
||||
communicate.save_sync(
|
||||
f"./cache/voice/{self.reactor.current_atom[1].content.replace('/','')}.wav"
|
||||
)
|
||||
playsound(str(cache))
|
||||
|
||||
threading.Thread(target=play).start()
|
||||
|
||||
def action_quick_pass(self):
|
||||
self.reactor.report(self.reactor.current_atom, 5)
|
||||
self._forward_judge(0)
|
||||
def action_toggle_dark(self):
|
||||
self.app.action_toggle_dark()
|
||||
|
||||
def action_pop_screen(self):
|
||||
self.app.pop_screen()
|
||||
|
||||
|
||||
class PreparationScreen(Screen):
|
||||
BINDINGS = [("q", "go_back", "返回"), ("escape", "quit_app", "退出")]
|
||||
|
||||
def __init__(
|
||||
self, nucleon_file: pt.NucleonUnion, electron_file: pt.ElectronUnion
|
||||
) -> None:
|
||||
super().__init__(name=None, id=None, classes=None)
|
||||
self.nucleon_file = nucleon_file
|
||||
self.electron_file = electron_file
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Header(show_clock=True)
|
||||
with Container(id="learning_screen_container"):
|
||||
yield Label(f"记忆项目: [b]{self.nucleon_file.name}[/b]\n")
|
||||
yield Label(
|
||||
f"核子文件对象: ./nucleon/[b]{self.nucleon_file.name}[/b].toml"
|
||||
)
|
||||
yield Label(
|
||||
f"电子文件对象: ./electron/[b]{self.electron_file.name}[/b].toml"
|
||||
)
|
||||
yield Label(f"核子数量:{len(self.nucleon_file)}")
|
||||
yield Button(
|
||||
"开始记忆",
|
||||
id="start_memorizing_button",
|
||||
variant="primary",
|
||||
classes="start-button",
|
||||
)
|
||||
yield Static(f"\n全文如下:\n")
|
||||
yield Static(self._get_full_content().replace("/", ""), classes="full")
|
||||
yield Footer()
|
||||
|
||||
def _get_full_content(self):
|
||||
content = ""
|
||||
for i in self.nucleon_file.nucleons:
|
||||
content += i["content"]
|
||||
return content
|
||||
|
||||
def action_go_back(self):
|
||||
self.app.pop_screen()
|
||||
|
||||
def action_quit_app(self):
|
||||
self.app.exit()
|
||||
|
||||
def on_button_pressed(self, event: Button.Pressed) -> None:
|
||||
if event.button.id == "start_memorizing_button":
|
||||
newscr = MemScreen(
|
||||
self.nucleon_file, self.electron_file, config.get("tasked_number", 6)
|
||||
)
|
||||
self.app.push_screen(newscr)
|
||||
|
||||
|
||||
class FileSelectorScreen(Screen):
|
||||
global ver
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Header(show_clock=True)
|
||||
yield Container(
|
||||
Label(f'欢迎使用 "潜进" 辅助记忆软件, 版本 {ver}', classes="title-label"),
|
||||
Label("选择要学习的文件:", classes="title-label"),
|
||||
ListView(id="file-list", classes="file-list-view"),
|
||||
)
|
||||
yield Footer()
|
||||
|
||||
def on_mount(self) -> None:
|
||||
file_list_widget = self.query_one("#file-list", ListView)
|
||||
nucleon_path = pathlib.Path("./nucleon")
|
||||
nucleon_files = sorted(
|
||||
[f.name for f in nucleon_path.iterdir() if f.suffix == ".toml"]
|
||||
)
|
||||
|
||||
if nucleon_files:
|
||||
for filename in nucleon_files:
|
||||
file_list_widget.append(ListItem(Label(filename)))
|
||||
else:
|
||||
file_list_widget.append(
|
||||
ListItem(Static("在 ./nucleon/ 中未找到任何核子文件. 请放置文件后重启应用."))
|
||||
)
|
||||
file_list_widget.disabled = True
|
||||
|
||||
def on_list_view_selected(self, event: ListView.Selected) -> None:
|
||||
if not isinstance(event.item, ListItem):
|
||||
return
|
||||
|
||||
selected_label = event.item.query_one(Label)
|
||||
if "未找到任何 .toml 文件" in str(selected_label.renderable):
|
||||
return
|
||||
|
||||
selected_filename = str(selected_label.renderable)
|
||||
nucleon_file = pt.NucleonUnion(
|
||||
pathlib.Path("./nucleon") / selected_filename
|
||||
)
|
||||
electron_file_path = pathlib.Path("./electron") / selected_filename
|
||||
if electron_file_path.exists():
|
||||
pass
|
||||
else:
|
||||
electron_file_path.touch()
|
||||
electron_file = pt.ElectronUnion(
|
||||
pathlib.Path("./electron") / selected_filename
|
||||
)
|
||||
self.app.push_screen(PreparationScreen(nucleon_file, electron_file))
|
||||
|
||||
def action_quit_app(self) -> None:
|
||||
self.app.exit()
|
||||
|
||||
from textual.app import App
|
||||
import screens
|
||||
|
||||
class AppLauncher(App):
|
||||
CSS_PATH = "styles.css"
|
||||
TITLE = "潜进 - 辅助记忆程序"
|
||||
BINDINGS = [("escape", "quit", "退出"), ("d", "toggle_dark", "改变色调")]
|
||||
SCREENS = {
|
||||
"file_selection_screen": FileSelectorScreen,
|
||||
"file_selection_screen": screens.FileSelectorScreen,
|
||||
}
|
||||
|
||||
def on_mount(self) -> None:
|
||||
|
1
metadata.py
Normal file
1
metadata.py
Normal file
@@ -0,0 +1 @@
|
||||
ver = "0.3.3"
|
@@ -2,7 +2,7 @@ import pathlib
|
||||
import toml
|
||||
import time
|
||||
import auxiliary as aux
|
||||
|
||||
from typing import List
|
||||
|
||||
class Electron:
|
||||
"""电子: 记忆分析元数据及算法"""
|
||||
@@ -178,12 +178,14 @@ class NucleonUnion:
|
||||
all = toml.load(f)
|
||||
lst = list()
|
||||
for i in all.keys():
|
||||
if "attr" in i:
|
||||
continue
|
||||
if "data" in i:
|
||||
continue
|
||||
lst.append(Nucleon(i, all[i]))
|
||||
self.keydata = all["keydata"]
|
||||
self.testdata = all["testdata"]
|
||||
self.nucleons = lst
|
||||
self.nucleons: List[Nucleon] = lst
|
||||
self.nucleons_dict = {i.content: i for i in lst}
|
||||
|
||||
def __len__(self):
|
||||
|
29
precache.py
29
precache.py
@@ -1,18 +1,20 @@
|
||||
# 音频预缓存实用程序, 独立于主程序之外, 但依赖 particles 组件
|
||||
# 音频预缓存实用程序, 独立于主程序之外, 但依赖其他组件
|
||||
import particles as pt
|
||||
import auxiliary as aux
|
||||
import edge_tts as tts
|
||||
from pathlib import Path
|
||||
import shutil
|
||||
import time
|
||||
|
||||
|
||||
def precache(text: str):
|
||||
"""预缓存单个文本的音频"""
|
||||
cache_dir = Path("./cache/voice/")
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
cache = cache_dir / f"{text}.wav"
|
||||
cache = cache_dir / f"{aux.get_md5(text)}.wav"
|
||||
if not cache.exists():
|
||||
communicate = tts.Communicate(text, "zh-CN-YunjianNeural")
|
||||
communicate.save_sync(f"./cache/voice/{text}.wav")
|
||||
communicate = tts.Communicate(text, "zh-CN-XiaoxiaoNeural")
|
||||
communicate.save_sync(f"./cache/voice/{aux.get_md5(text)}.wav")
|
||||
|
||||
|
||||
def proc_file(path: Path):
|
||||
@@ -29,14 +31,17 @@ def walk(path_str: str):
|
||||
"""遍历目录处理所有文件"""
|
||||
path = Path(path_str)
|
||||
print(f"正在遍历目录: {path}")
|
||||
|
||||
for item in path.iterdir():
|
||||
if item.is_file() and item.suffix == ".toml":
|
||||
print(f"正预缓存文件: {item.name}")
|
||||
proc_file(item)
|
||||
elif item.is_dir():
|
||||
print(f"进入目录: {item.name}")
|
||||
|
||||
try:
|
||||
for item in path.iterdir():
|
||||
if item.is_file() and item.suffix == ".toml":
|
||||
print(f"正预缓存文件: {item.name}")
|
||||
proc_file(item)
|
||||
elif item.is_dir():
|
||||
print(f"进入目录: {item.name}")
|
||||
except:
|
||||
print("发生一个异常, 于 5 秒后自动重新下载")
|
||||
time.sleep(5)
|
||||
walk(path_str)
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("音频预缓存实用程序")
|
||||
|
@@ -1,11 +1,11 @@
|
||||
import random
|
||||
|
||||
|
||||
class Puzzle:
|
||||
class BasePuzzle:
|
||||
pass
|
||||
|
||||
|
||||
class BlankPuzzle(Puzzle):
|
||||
class BlankPuzzle(BasePuzzle):
|
||||
"""填空题谜题生成器
|
||||
|
||||
Args:
|
||||
@@ -41,7 +41,7 @@ class BlankPuzzle(Puzzle):
|
||||
return f"{self.wording}\n{str(self.answer)}"
|
||||
|
||||
|
||||
class SelectionPuzzle(Puzzle):
|
||||
class SelectionPuzzle(BasePuzzle):
|
||||
"""选择题谜题生成器
|
||||
|
||||
Args:
|
||||
|
@@ -20,6 +20,9 @@ class Apparatus():
|
||||
for i in self.positron["testdata"].keys():
|
||||
if i == "additional_inf":
|
||||
continue
|
||||
if i == "fill_blank_test": # 加深
|
||||
self.procession.append(comps.registry[i](screen, reactor, atom))
|
||||
self.procession.append(comps.registry[i](screen, reactor, atom))
|
||||
self.procession.append(comps.registry[i](screen, reactor, atom))
|
||||
# self.procession.reverse()
|
||||
random.shuffle(self.procession)
|
||||
@@ -139,6 +142,7 @@ class Reactor():
|
||||
e.revisor(5, True)
|
||||
continue
|
||||
e.revisor(q)
|
||||
|
||||
def report(self, atom, quality):
|
||||
"向反应器和最低质量记录汇报"
|
||||
if atom in self.atoms_new:
|
||||
|
76
reactor_k.py
Normal file
76
reactor_k.py
Normal file
@@ -0,0 +1,76 @@
|
||||
import particles as pt
|
||||
import auxiliary as aux
|
||||
import time
|
||||
from typing import List
|
||||
|
||||
class BaseScheduler():
|
||||
"调度器接口"
|
||||
|
||||
def revisor(self, electron: pt.Electron, quality: int):
|
||||
"""由 quality 更新电子的记忆参数"""
|
||||
pass
|
||||
|
||||
def get_atoms_for_review(self, electron_file: pt.ElectronUnion):
|
||||
"""从电子文件中筛选出当前需要复习的所有原子"""
|
||||
pass
|
||||
|
||||
def get_atoms_for_learning(self, nucleon_file: pt.NucleonUnion, limit: int):
|
||||
"""从核子文件中获取待学习的新原子"""
|
||||
pass
|
||||
|
||||
class SM2Scheduler(BaseScheduler):
|
||||
"基于 SM-2 的调度器实现"
|
||||
def revisor(self, electron: pt.Electron, quality: int, is_new_activation):
|
||||
print(f"REVISOR: {quality}, {is_new_activation}")
|
||||
if quality == -1:
|
||||
return -1
|
||||
|
||||
electron.metadata['efactor'] = electron.metadata['efactor'] + (
|
||||
0.1 - (5 - quality) * (0.08 + (5 - quality) * 0.02)
|
||||
)
|
||||
electron.metadata['efactor'] = max(1.3, electron.metadata['efactor'])
|
||||
|
||||
if quality < 3:
|
||||
# 若保留率低于 3,重置重复次数
|
||||
electron.metadata['rept'] = 0
|
||||
electron.metadata['interval'] = 0 # 设为0,以便下面重新计算 I(1)
|
||||
else:
|
||||
electron.metadata['rept'] += 1
|
||||
|
||||
electron.metadata['real_rept'] += 1
|
||||
|
||||
if is_new_activation: # 初次激活
|
||||
electron.metadata['rept'] = 0
|
||||
electron.metadata['efactor'] = 2.5
|
||||
|
||||
if electron.metadata['rept'] == 0: # 刚被重置或初次激活后复习
|
||||
electron.metadata['interval'] = 1 # I(1)
|
||||
elif electron.metadata['rept'] == 1:
|
||||
electron.metadata['interval'] = 6 # I(2) 经验公式
|
||||
else:
|
||||
electron.metadata['interval'] = round(
|
||||
electron.metadata['interval'] * electron.metadata['efactor']
|
||||
)
|
||||
|
||||
electron.metadata['last_date'] = aux.get_daystamp()
|
||||
electron.metadata['next_date'] = aux.get_daystamp() + electron.metadata['interval']
|
||||
electron.metadata['last_modify'] = time.time()
|
||||
|
||||
def get_atoms_for_learning(self, nucleon_file: pt.NucleonUnion, limit: int):
|
||||
def electron_dict_get_fallback(key) -> pt.Electron:
|
||||
value = self.electron_dict.get(key)
|
||||
# 如果值不存在,则设置默认值
|
||||
if value is None:
|
||||
value = pt.Electron(key, {}) # 获取默认值
|
||||
self.electron_dict[key] = value # 将默认值存入字典
|
||||
electron_file.sync()
|
||||
return value # 返回获取的值(可能是默认值)
|
||||
|
||||
for i in nucleon_file.nucleons:
|
||||
if i.metadata
|
||||
|
||||
def get_atoms_for_review(self, electron_file: pt.ElectronUnion):
|
||||
return super().get_atoms_for_review(electron_file)
|
||||
|
||||
class FSRSScheduler():
|
||||
"基于 FSRS 的调度器实现"
|
239
screens.py
Normal file
239
screens.py
Normal file
@@ -0,0 +1,239 @@
|
||||
from textual.app import App, ComposeResult
|
||||
from textual.widgets import (
|
||||
Header,
|
||||
Footer,
|
||||
ListView,
|
||||
ProgressBar,
|
||||
DirectoryTree,
|
||||
ListItem,
|
||||
Label,
|
||||
Static,
|
||||
Button,
|
||||
)
|
||||
from textual.containers import Container, Horizontal, Center
|
||||
from textual.screen import Screen
|
||||
import pathlib
|
||||
import threading
|
||||
import edge_tts as tts
|
||||
from playsound import playsound
|
||||
import particles as pt
|
||||
from reactor import Reactor, Apparatus
|
||||
import auxiliary as aux
|
||||
import compositions as compo
|
||||
import builtins
|
||||
import metadata
|
||||
|
||||
config = aux.ConfigFile("config.toml")
|
||||
|
||||
class MemScreen(Screen):
|
||||
BINDINGS = [
|
||||
("d", "toggle_dark", "改变色调"),
|
||||
("q", "pop_screen", "返回主菜单"),
|
||||
("v", "play_voice", "朗读"),
|
||||
]
|
||||
if config.get("quick_pass"):
|
||||
BINDINGS.append(("k", "quick_pass", "快速通过[调试]"))
|
||||
btn = dict()
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
nucleon_file: pt.NucleonUnion,
|
||||
electron_file: pt.ElectronUnion,
|
||||
tasked_num,
|
||||
):
|
||||
super().__init__(name=None, id=None, classes=None)
|
||||
self.reactor = Reactor(nucleon_file, electron_file, self, tasked_num)
|
||||
self.stage = 1
|
||||
self.stage += self.reactor.set_round_templated(self.stage)
|
||||
self.reactor.forward()
|
||||
self.compo = next(self.reactor.current_appar)
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
if type(self.compo).__name__ == "Recognition":
|
||||
self.action_play_voice()
|
||||
yield Header(show_clock=True)
|
||||
with Center():
|
||||
yield Static(
|
||||
f"{len(self.reactor.procession) - self.reactor.index}/{len(self.reactor.procession)}"
|
||||
)
|
||||
yield from self.compo.compose()
|
||||
yield Footer()
|
||||
|
||||
def on_mount(self):
|
||||
pass
|
||||
|
||||
def on_button_pressed(self, event):
|
||||
ret = self.compo.handler(event, "button")
|
||||
self._forward_judge(ret)
|
||||
|
||||
def _forward_judge(self, ret):
|
||||
if ret == -1:
|
||||
return
|
||||
if ret == 0:
|
||||
try:
|
||||
self.compo = next(self.reactor.current_appar)
|
||||
self.refresh_ui()
|
||||
except StopIteration:
|
||||
nxt = self.reactor.forward(1)
|
||||
try:
|
||||
self.compo = next(self.reactor.current_appar)
|
||||
except:
|
||||
pass
|
||||
if nxt == -1:
|
||||
if self.reactor.round_set == 0:
|
||||
if self.stage == 4:
|
||||
if config.get("save"):
|
||||
self.reactor.save()
|
||||
self.compo = compo.Finished(
|
||||
self, None, pt.Atom.placeholder()
|
||||
)
|
||||
self.refresh_ui()
|
||||
else:
|
||||
self.reactor.set_round_templated(self.stage)
|
||||
self.reactor.forward(1)
|
||||
self.stage += 1
|
||||
self.compo = next(self.reactor.current_appar)
|
||||
self.refresh_ui()
|
||||
return
|
||||
return
|
||||
else:
|
||||
self.refresh_ui()
|
||||
return
|
||||
if ret == 1:
|
||||
self.refresh_ui()
|
||||
return
|
||||
|
||||
def refresh_ui(self):
|
||||
self.call_later(self.recompose)
|
||||
print(type(self.compo).__name__)
|
||||
|
||||
def action_play_voice(self):
|
||||
def play():
|
||||
cache_dir = pathlib.Path(f"./cache/voice/")
|
||||
cache_dir.mkdir(parents=True, exist_ok=True)
|
||||
cache = cache_dir / f"{aux.get_md5(self.reactor.current_atom[1].content.replace('/',''))}.wav"
|
||||
if not cache.exists():
|
||||
communicate = tts.Communicate(
|
||||
self.reactor.current_atom[1].content.replace("/", ""),
|
||||
"zh-CN-XiaoxiaoNeural",
|
||||
)
|
||||
communicate.save_sync(
|
||||
f"./cache/voice/{aux.get_md5(self.reactor.current_atom[1].content.replace('/',''))}.wav"
|
||||
)
|
||||
playsound(str(cache))
|
||||
|
||||
threading.Thread(target=play).start()
|
||||
|
||||
def action_quick_pass(self):
|
||||
self.reactor.report(self.reactor.current_atom, 5)
|
||||
self._forward_judge(0)
|
||||
def action_toggle_dark(self):
|
||||
self.app.action_toggle_dark()
|
||||
|
||||
def action_pop_screen(self):
|
||||
self.app.pop_screen()
|
||||
|
||||
|
||||
class PreparationScreen(Screen):
|
||||
BINDINGS = [("q", "go_back", "返回"), ("escape", "quit_app", "退出")]
|
||||
|
||||
def __init__(
|
||||
self, nucleon_file: pt.NucleonUnion, electron_file: pt.ElectronUnion
|
||||
) -> None:
|
||||
super().__init__(name=None, id=None, classes=None)
|
||||
self.nucleon_file = nucleon_file
|
||||
self.electron_file = electron_file
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Header(show_clock=True)
|
||||
with Container(id="learning_screen_container"):
|
||||
yield Label(f"记忆项目: [b]{self.nucleon_file.name}[/b]\n")
|
||||
yield Label(
|
||||
f"核子文件对象: ./nucleon/[b]{self.nucleon_file.name}[/b].toml"
|
||||
)
|
||||
yield Label(
|
||||
f"电子文件对象: ./electron/[b]{self.electron_file.name}[/b].toml"
|
||||
)
|
||||
yield Label(f"核子数量:{len(self.nucleon_file)}")
|
||||
yield Button(
|
||||
"开始记忆",
|
||||
id="start_memorizing_button",
|
||||
variant="primary",
|
||||
classes="start-button",
|
||||
)
|
||||
yield Static(f"\n全文如下:\n")
|
||||
yield Static(self._get_full_content().replace("/", ""), classes="full")
|
||||
yield Footer()
|
||||
|
||||
def _get_full_content(self):
|
||||
content = ""
|
||||
for i in self.nucleon_file.nucleons:
|
||||
content += i["content"]
|
||||
return content
|
||||
|
||||
def action_go_back(self):
|
||||
self.app.pop_screen()
|
||||
|
||||
def action_quit_app(self):
|
||||
self.app.exit()
|
||||
|
||||
def on_button_pressed(self, event: Button.Pressed) -> None:
|
||||
if event.button.id == "start_memorizing_button":
|
||||
newscr = MemScreen(
|
||||
self.nucleon_file, self.electron_file, config.get("tasked_number", 6)
|
||||
)
|
||||
self.app.push_screen(newscr)
|
||||
|
||||
|
||||
class FileSelectorScreen(Screen):
|
||||
|
||||
def compose(self) -> ComposeResult:
|
||||
yield Header(show_clock=True)
|
||||
yield Container(
|
||||
Label(f'欢迎使用 "潜进" 辅助记忆软件, 版本 {metadata.ver}', classes="title-label"),
|
||||
Label("选择要学习的文件:", classes="title-label"),
|
||||
ListView(id="file-list", classes="file-list-view"),
|
||||
)
|
||||
yield Footer()
|
||||
|
||||
def on_mount(self) -> None:
|
||||
file_list_widget = self.query_one("#file-list", ListView)
|
||||
nucleon_path = pathlib.Path("./nucleon")
|
||||
nucleon_files = sorted(
|
||||
[f.name for f in nucleon_path.iterdir() if f.suffix == ".toml"]
|
||||
)
|
||||
|
||||
if nucleon_files:
|
||||
for filename in nucleon_files:
|
||||
file_list_widget.append(ListItem(Label(filename)))
|
||||
else:
|
||||
file_list_widget.append(
|
||||
ListItem(Static("在 ./nucleon/ 中未找到任何核子文件. 请放置文件后重启应用."))
|
||||
)
|
||||
file_list_widget.disabled = True
|
||||
|
||||
def on_list_view_selected(self, event: ListView.Selected) -> None:
|
||||
if not isinstance(event.item, ListItem):
|
||||
return
|
||||
|
||||
selected_label = event.item.query_one(Label)
|
||||
if "未找到任何 .toml 文件" in str(selected_label.renderable):
|
||||
return
|
||||
|
||||
selected_filename = str(selected_label.renderable)
|
||||
nucleon_file = pt.NucleonUnion(
|
||||
pathlib.Path("./nucleon") / selected_filename
|
||||
)
|
||||
electron_file_path = pathlib.Path("./electron") / selected_filename
|
||||
if electron_file_path.exists():
|
||||
pass
|
||||
else:
|
||||
electron_file_path.touch()
|
||||
electron_file = pt.ElectronUnion(
|
||||
pathlib.Path("./electron") / selected_filename
|
||||
)
|
||||
self.app.push_screen(PreparationScreen(nucleon_file, electron_file))
|
||||
|
||||
def action_quit_app(self) -> None:
|
||||
self.app.exit()
|
||||
|
247
tweak.py
Normal file
247
tweak.py
Normal file
@@ -0,0 +1,247 @@
|
||||
import os
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
def check_dev_flag():
|
||||
"""检查是否存在开发标志文件,如果存在则退出程序"""
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
dev_flag_path = os.path.join(script_dir, '.devflag')
|
||||
|
||||
if os.path.exists(dev_flag_path):
|
||||
print("检测到开发标志文件 (.devflag),不得运行此程序")
|
||||
sys.exit(0)
|
||||
|
||||
def main():
|
||||
# 检查开发标志文件
|
||||
check_dev_flag()
|
||||
|
||||
# 输出标题
|
||||
print("HeurAMS 更新 & 数据管理工具")
|
||||
print("君欲何为?")
|
||||
print("\nR: 全新安装 HeurAMS (删除 nucleon 与 electron 的用户数据, 并从上游同步软件更新)")
|
||||
print("F: 翻新 HeurAMS (保留 nucleon 与 electron 的用户数据, 并从上游同步软件更新)")
|
||||
print("U: 卸载 HeurAMS (删除 HeurAMS 程序文件, 保留用户数据)")
|
||||
print("P: 应用 Termux 音频补丁")
|
||||
|
||||
# 获取用户输入
|
||||
choice = input("\n请输入选择 (R/F/U/P): ").strip().lower()
|
||||
|
||||
# 获取脚本所在目录
|
||||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
os.chdir(script_dir)
|
||||
|
||||
if choice == 'r':
|
||||
# 检查开发标志文件(再次检查,防止在运行时创建)
|
||||
check_dev_flag()
|
||||
|
||||
# 全新安装 - 删除所有文件和文件夹(包括.git)
|
||||
print("正在执行全新安装...")
|
||||
|
||||
# 遍历当前目录下的所有文件和文件夹
|
||||
for item in os.listdir('.'):
|
||||
# 跳过脚本自身(如果存在)和开发标志文件
|
||||
if item == os.path.basename(__file__) or item == '.devflag':
|
||||
continue
|
||||
|
||||
item_path = os.path.join(script_dir, item)
|
||||
|
||||
try:
|
||||
if os.path.isfile(item_path) or os.path.islink(item_path):
|
||||
os.remove(item_path)
|
||||
elif os.path.isdir(item_path):
|
||||
shutil.rmtree(item_path)
|
||||
except Exception as e:
|
||||
print(f"删除 {item} 时出错: {e}")
|
||||
|
||||
# 执行git clone到临时目录,然后移动文件
|
||||
try:
|
||||
temp_dir = os.path.join(script_dir, 'temp_clone')
|
||||
subprocess.run(['git', 'clone', 'https://gitea.imwangzhiyu.xyz/ajax/HeurAMS', temp_dir], check=True)
|
||||
|
||||
# 移动所有文件到当前目录(除了.git目录)
|
||||
for item in os.listdir(temp_dir):
|
||||
if item != '.git':
|
||||
src = os.path.join(temp_dir, item)
|
||||
dst = os.path.join(script_dir, item)
|
||||
if os.path.exists(dst):
|
||||
if os.path.isdir(dst):
|
||||
shutil.rmtree(dst)
|
||||
else:
|
||||
os.remove(dst)
|
||||
shutil.move(src, dst)
|
||||
|
||||
# 删除临时目录
|
||||
shutil.rmtree(temp_dir)
|
||||
print("全新安装完成!")
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
print(f"Git clone 失败: {e}")
|
||||
except Exception as e:
|
||||
print(f"文件操作失败: {e}")
|
||||
|
||||
elif choice == 'f':
|
||||
# 检查开发标志文件(再次检查,防止在运行时创建)
|
||||
check_dev_flag()
|
||||
|
||||
# 翻新安装 - 保留特定目录
|
||||
print("正在执行翻新安装...")
|
||||
|
||||
# 需要保留的目录列表
|
||||
preserve_dirs = ['nucleon', 'electron', 'cache']
|
||||
|
||||
# 备份需要保留的目录到临时位置
|
||||
backup_dir = os.path.join(script_dir, 'temp_backup')
|
||||
os.makedirs(backup_dir, exist_ok=True)
|
||||
|
||||
for dir_name in preserve_dirs:
|
||||
dir_path = os.path.join(script_dir, dir_name)
|
||||
if os.path.exists(dir_path):
|
||||
backup_path = os.path.join(backup_dir, dir_name)
|
||||
if os.path.exists(backup_path):
|
||||
shutil.rmtree(backup_path)
|
||||
shutil.copytree(dir_path, backup_path)
|
||||
|
||||
# 删除所有文件和文件夹(包括.git)
|
||||
for item in os.listdir('.'):
|
||||
# 跳过脚本自身、备份目录和开发标志文件
|
||||
if item == os.path.basename(__file__) or item == 'temp_backup' or item == '.devflag':
|
||||
continue
|
||||
|
||||
item_path = os.path.join(script_dir, item)
|
||||
|
||||
try:
|
||||
if os.path.isfile(item_path) or os.path.islink(item_path):
|
||||
os.remove(item_path)
|
||||
elif os.path.isdir(item_path):
|
||||
shutil.rmtree(item_path)
|
||||
except Exception as e:
|
||||
print(f"删除 {item} 时出错: {e}")
|
||||
|
||||
# 执行git clone到当前目录
|
||||
try:
|
||||
temp_dir = os.path.join(script_dir, 'temp_clone')
|
||||
subprocess.run(['git', 'clone', 'https://gitea.imwangzhiyu.xyz/ajax/HeurAMS', temp_dir], check=True)
|
||||
|
||||
# 移动所有文件到当前目录(除了.git目录)
|
||||
for item in os.listdir(temp_dir):
|
||||
if item != '.git':
|
||||
src = os.path.join(temp_dir, item)
|
||||
dst = os.path.join(script_dir, item)
|
||||
if os.path.exists(dst):
|
||||
if os.path.isdir(dst):
|
||||
shutil.rmtree(dst)
|
||||
else:
|
||||
os.remove(dst)
|
||||
shutil.move(src, dst)
|
||||
|
||||
# 删除临时克隆目录
|
||||
shutil.rmtree(temp_dir)
|
||||
|
||||
# 恢复保留的目录(覆盖git仓库中的同名目录)
|
||||
for dir_name in preserve_dirs:
|
||||
backup_path = os.path.join(backup_dir, dir_name)
|
||||
if os.path.exists(backup_path):
|
||||
target_path = os.path.join(script_dir, dir_name)
|
||||
if os.path.exists(target_path):
|
||||
shutil.rmtree(target_path)
|
||||
shutil.copytree(backup_path, target_path)
|
||||
|
||||
# 删除备份目录
|
||||
shutil.rmtree(backup_dir)
|
||||
print("翻新安装完成!")
|
||||
|
||||
except subprocess.CalledProcessError as e:
|
||||
print(f"Git clone 失败: {e}")
|
||||
except Exception as e:
|
||||
print(f"文件操作失败: {e}")
|
||||
|
||||
elif choice == 'u':
|
||||
# 检查开发标志文件(再次检查,防止在运行时创建)
|
||||
check_dev_flag()
|
||||
|
||||
# 卸载 HeurAMS - 删除程序文件,保留用户数据
|
||||
print("正在卸载 HeurAMS,保留用户数据...")
|
||||
|
||||
# 需要保留的用户数据目录列表
|
||||
preserve_dirs = ['nucleon', 'electron', 'cache']
|
||||
|
||||
# 备份需要保留的目录到临时位置
|
||||
backup_dir = os.path.join(script_dir, 'temp_backup')
|
||||
os.makedirs(backup_dir, exist_ok=True)
|
||||
|
||||
for dir_name in preserve_dirs:
|
||||
dir_path = os.path.join(script_dir, dir_name)
|
||||
if os.path.exists(dir_path):
|
||||
backup_path = os.path.join(backup_dir, dir_name)
|
||||
if os.path.exists(backup_path):
|
||||
shutil.rmtree(backup_path)
|
||||
shutil.copytree(dir_path, backup_path)
|
||||
print(f"已备份用户数据: {dir_name}")
|
||||
|
||||
# 删除所有文件和文件夹(除了脚本自身、备份目录和开发标志文件)
|
||||
for item in os.listdir('.'):
|
||||
# 跳过脚本自身、备份目录和开发标志文件
|
||||
if item == os.path.basename(__file__) or item == 'temp_backup' or item == '.devflag':
|
||||
continue
|
||||
|
||||
item_path = os.path.join(script_dir, item)
|
||||
|
||||
try:
|
||||
if os.path.isfile(item_path) or os.path.islink(item_path):
|
||||
os.remove(item_path)
|
||||
print(f"已删除文件: {item}")
|
||||
elif os.path.isdir(item_path):
|
||||
shutil.rmtree(item_path)
|
||||
print(f"已删除目录: {item}")
|
||||
except Exception as e:
|
||||
print(f"删除 {item} 时出错: {e}")
|
||||
|
||||
# 恢复保留的用户数据目录
|
||||
for dir_name in preserve_dirs:
|
||||
backup_path = os.path.join(backup_dir, dir_name)
|
||||
if os.path.exists(backup_path):
|
||||
target_path = os.path.join(script_dir, dir_name)
|
||||
if os.path.exists(target_path):
|
||||
shutil.rmtree(target_path)
|
||||
shutil.copytree(backup_path, target_path)
|
||||
print(f"已恢复用户数据: {dir_name}")
|
||||
|
||||
# 删除备份目录
|
||||
shutil.rmtree(backup_dir)
|
||||
print("卸载完成!HeurAMS 程序文件已删除,用户数据已保留。")
|
||||
|
||||
elif choice == 'p':
|
||||
# 应用 Termux 音频补丁
|
||||
print("应用 Termux 音频补丁")
|
||||
|
||||
# 询问用户是否使用安卓Termux环境
|
||||
termux_choice = input("是否使用安卓Termux环境? (y/n): ").strip().lower()
|
||||
|
||||
if termux_choice in ['y', 'yes']:
|
||||
# 创建playsound.py文件
|
||||
playsound_content = '''import os
|
||||
def playsound(path):
|
||||
os.system(f"play-audio '{path}'")
|
||||
'''
|
||||
|
||||
playsound_path = os.path.join(script_dir, 'playsound.py')
|
||||
|
||||
try:
|
||||
with open(playsound_path, 'w', encoding='utf-8') as f:
|
||||
f.write(playsound_content)
|
||||
print("已创建 playsound.py 文件")
|
||||
print("Termux 音频补丁应用成功!")
|
||||
print("现在可以使用 play-audio 命令播放音频了。")
|
||||
|
||||
except Exception as e:
|
||||
print(f"创建 playsound.py 文件时出错: {e}")
|
||||
else:
|
||||
print("已取消应用 Termux 音频补丁。")
|
||||
|
||||
else:
|
||||
print("无效的选择,请输入 R、F、U 或 P。")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
Reference in New Issue
Block a user