#!/usr/bin/env python3

# -*- coding: utf-8 -*-

"""ReadMD —— 轻量级本地 Markdown 阅读器。

特性：

- 本地 127.0.0.1 HTTP 服务 + pywebview 原生窗口，秒开

- 渲染前自动修正常见错误（表格 / 加粗 / 公式 / 标题），只影响显示

- 自动刷新、目录、搜索、主题、字号、最近文件、文件夹浏览、打印

- 全部资源离线（marked + MathJax 已内置），无需联网

用法：

python readmd.py [文件.md] # 打开文件（或空启动）

python readmd.py --browser [文件] # 用默认浏览器打开（无 pywebview 时兜底）

python readmd.py --selftest # 自测（修正器 + 本地服务）

"""

import argparse

import atexit

import base64

import binascii

import collections

import gzip

import hashlib

import json

import logging

import mimetypes

import os

import re

import secrets

import socket

import subprocess

import sys

import tempfile

import time

import threading

import webbrowser

from datetime import datetime, timezone

from email.utils import formatdate, parsedate_to_datetime

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

from pathlib import Path

from urllib.parse import parse_qs, quote, unquote, urlparse

from src.readmd_core import (

normalize_dialog_path,

load_json,

save_json,

read_text,

readmd_fix,

)

from src.readmd_core.file_writer import save_text_atomic

from src.readmd_core.static_assets import resolve_asset

from src.readmd_core.safe_open import safe_external_url, safe_file_target

from src.readmd_core.versioning import compare_versions as _version_compare

from src.readmd_core.versioning import parse_version as _version_parse

from src.readmd_core.versioning import select_update_release as _select_update_release

import src.readmd_modules as RM

from src.readmd_modules.validators import validate_file_path, validate_command, paths_within

from src.readmd_modules.skills import SkillError, SkillRegistry, default_skill_roots

from src.readmd_modules.pet import (

HermesPetBridge,

HermesPetLauncher,

HermesPetPluginInstaller,

PetBatchQueue,

PetController,

foreground_fullscreen,

verify_model_bundle,

get_app_install_dir,

get_default_pet_install_root,

clean_legacy_pet_installations,

check_pet_update,

apply_pet_update,

)

import src.readmd_modules.skill_import as _skill_import

from src.readmd_core.service import ReadMDCoreService

from src.readmd_core import upstream as _upstream_sources

APP_DIR = sys._MEIPASS if getattr(sys, 'frozen', False) else os.path.dirname(os.path.abspath(__file__))

PET_MODEL_DIR = os.path.join(APP_DIR, 'assets', 'pet', 'model')

from src.readmd_core.config import (

DATA_DIR,

SETTINGS_FILE,

RECENT_FILE,

PROMPTS_FILE,

HISTORY_FILE,

LOG_FILE,

IS_MAC,

IS_WIN,

IS_LINUX,

get_system_language,

load_dotenv,

VERSION,

)

load_dotenv()

def native_gui_required():

"""Return whether a Linux/macOS frozen build must use its native backend.

Formal packages fail closed when the native engine is unavailable. The

explicit ``--browser`` mode remains available for development and

diagnostics, but a packaged release must never silently lose the Python

bridge by opening a regular browser window.

"""

if not (IS_MAC or IS_LINUX):

return False

value = os.environ.get('READMD_REQUIRE_NATIVE_GUI')

if value is not None:

return value.strip().lower() in ('1', 'true', 'yes', 'on')

return bool(getattr(sys, 'frozen', False))

MD_EXTS = ('.md', '.markdown', '.mdown', '.mkd', '.mdx', '.txt')

CODE_CONFIG_EXTS = (

'.toml', '.yaml', '.yml', '.json', '.json5', '.jsonc', '.ini', '.cfg',

'.conf', '.config', '.env', '.properties', '.xml', '.plist', '.inf',

'.bat', '.cmd', '.ps1', '.psm1', '.sh', '.bash', '.zsh', '.fish', '.vbs',

'.py', '.js', '.mjs', '.cjs', '.ts', '.tsx', '.jsx', '.c', '.cpp',

'.h', '.hpp', '.cc', '.cxx', '.cs', '.java', '.kt', '.kts', '.rs',

'.go', '.rb', '.php', '.swift', '.lua', '.r', '.m', '.dart', '.sql',

'.dockerfile', '.makefile', '.gradle', '.html', '.htm', '.css', '.scss',

'.sass', '.less', '.vue', '.svelte', '.log', '.out', '.err', '.diff',

'.patch', '.gitignore', '.gitattributes', '.editorconfig', '.npmrc',

'.rst', '.asciidoc', '.adoc', '.bib', '.csv', '.tsv',

)

ALL_TEXT_EXTS = MD_EXTS + CODE_CONFIG_EXTS

MEDIA_EXTS = ('.mp3', '.wav', '.m4a', '.mp4', '.flac', '.ogg', '.webm', '.aac', '.wma', '.mkv', '.mov', '.avi')

CONVERT_EXTS = ('.docx', '.doc', '.pptx', '.ppt', '.xlsx', '.xls', '.pdf', '.html', '.htm',

'.txt', '.csv', '.json', '.xml', '.zip', '.eml', '.msg', '.rtf', '.odt', '.epub',

'.tex', '.latex') + MEDIA_EXTS + CODE_CONFIG_EXTS

WIN7_CONVERT_EXTS = ('.docx', '.pdf')

WIN7_UNAVAILABLE = '该功能在 Win7 版暂不支持（本版本仅保留 docx / pdf 转 MD 与导出功能）'

# ------------------------------------------------------------------ 升级推送（静默）

_UPGRADE_RELEASE_URL = 'https://api.github.com/repos/Natsummerance/readMD/releases/latest'

_UPGRADE_RELEASES_URL = 'https://api.github.com/repos/Natsummerance/readMD/releases?per_page=100'

_UPGRADE_CACHE = {'done': False, 'result': None}

def _parse_version(value):

return _version_parse(value)

def _compare_versions(left, right):

return _version_compare(left, right)

def check_latest_release():

"""查询 GitHub 最新 Release；失败/超时静默返回 None，结果进程内缓存。"""

if _UPGRADE_CACHE['done']:

return _UPGRADE_CACHE['result']

result = None

try:

import urllib.request as _urlreq

parsed_current = _parse_version(VERSION)

current_is_prerelease = bool(parsed_current and parsed_current[1] == 0)

url = _UPGRADE_RELEASES_URL if current_is_prerelease else _UPGRADE_RELEASE_URL

req = _urlreq.Request(url, headers={

'User-Agent': 'ReadMD/%s' % VERSION,

'Accept': 'application/vnd.github+json',

})

with _urlreq.urlopen(req, timeout=4) as resp:

payload = json.loads(resp.read(1024 * 1024).decode('utf-8'))

releases = payload if isinstance(payload, list) else [payload]

latest_release = _select_update_release(VERSION, releases)

tag = str(latest_release.get('tag_name') or '') if latest_release else ''

current = _parse_version(VERSION)

if latest_release and current and (_compare_versions(tag, VERSION) or 0) > 0:

result = {

'latest': tag,

'url': str(latest_release.get('html_url') or _UPGRADE_RELEASE_URL),

}

except Exception:

logging.debug('upgrade check failed (silent)', exc_info=True)

_UPGRADE_CACHE['done'] = True

_UPGRADE_CACHE['result'] = result

return result

CONTROL_PORT = 26891

INSTANCE_FILE = os.path.join(DATA_DIR, 'instance.json')

_CONVERT_JOBS = {}

_CONVERT_JOB_SEQ = [0]

_CONVERT_LOCK = threading.Lock()

_T0 = time.time()

_BOOT_LOCK = threading.Lock()

_BOOT_MILESTONES = {}

_STARTUP_PROBE = {'enabled': False, 'timeout': 20.0, 'json_path': '',

'window': None, 'finished': False, 'timed_out': False,

'timer': None}

def is_win7():

"""Win7 检测：驱动功能裁剪与内置固定版 WebView2 109 运行时。"""

if os.environ.get('READMD_FORCE_WIN7') == '1':

return True

try:

import platform

return platform.system() == 'Windows' and platform.release() == '7'

except Exception:

return False

def setup_win7_webview2_env():

"""Win7：把内置固定版 WebView2 109 运行时目录与嵌入式 user-data 目录注入环境变量，

win7 构建里打过补丁的 pywebview edgechromium 会读取这两个变量。"""

if not is_win7():

return

try:

if getattr(sys, 'frozen', False):

base = os.path.dirname(os.path.abspath(sys.executable))

else:

base = APP_DIR

rt = os.path.join(base, 'webview2_runtime')

if os.path.isdir(rt):

os.environ['READMD_WEBVIEW2_RUNTIME'] = rt

os.environ['READMD_WEBVIEW2_USERDATA'] = os.path.join(base, 'webview2_userdata')

except Exception:

pass

def milestone(group, name):

"""启动里程碑打点：写入 readmd.log，用于验证“秒开”。"""

elapsed = int((time.time() - _T0) * 1000)

if group == 'boot':

with _BOOT_LOCK:

_BOOT_MILESTONES.setdefault(name, elapsed)

try:

logging.info('[%s] %dms %s', group, elapsed, name)

except Exception:

pass

def startup_probe_summary(milestones=None, timed_out=False):

"""Build a privacy-safe startup report; deliberately contains no document data."""

milestones = dict(_BOOT_MILESTONES if milestones is None else milestones)

names = ('server_up', 'window_created', 'window_loaded', 'page_loaded',

'first_document')

return {'version': VERSION, 'timed_out': bool(timed_out),

'milestones_ms': {name: milestones.get(name) for name in names}}

def write_startup_probe(path='', timed_out=False):

"""Print and optionally atomically persist a startup probe report."""

report = startup_probe_summary(timed_out=timed_out)

encoded = json.dumps(report, ensure_ascii=False, sort_keys=True)

safe_print(encoded)

if path:

directory = os.path.dirname(os.path.abspath(path))

if directory:

os.makedirs(directory, exist_ok=True)

tmp = os.path.join(directory, '.%s.%s.tmp' %

(os.path.basename(path), os.getpid()))

try:

with open(tmp, 'w', encoding='utf-8') as handle:

handle.write(encoded + '\n')

os.replace(tmp, path)

except Exception:

try:

if os.path.exists(tmp):

os.remove(tmp)

except Exception:

pass

raise

return report

def _finish_startup_probe(timed_out=False):

"""End a probe run without persisting document paths or document content."""

with _BOOT_LOCK:

if not _STARTUP_PROBE.get('enabled') or _STARTUP_PROBE.get('finished'):

return

_STARTUP_PROBE['finished'] = True

_STARTUP_PROBE['timed_out'] = bool(timed_out)

timer = _STARTUP_PROBE.get('timer')

if timer is not None:

try:

timer.cancel()

except Exception:

pass

window = _STARTUP_PROBE.get('window')

if window is not None:

try:

window.destroy()

except Exception:

pass

# ---------------------------------------------------------------- 单实例常驻

# 固定控制端口 + instance.json（端口/随机 token）。新进程先 ping 已有实例，

# 命中则把要打开的文件 POST 过去后立即退出，实现“双击 .md 秒开”。

_CONTROL = {'queue': [], 'pet_batches': [], 'pet_menus': 0, 'window': None, 'ready': False}

_control_lock = threading.Lock()

def _read_instance():

return load_json(INSTANCE_FILE, {})

def _write_instance(port, token):

save_json(INSTANCE_FILE, {'port': port, 'token': token,

'pid': os.getpid(), 'started': time.time()})

def _clear_instance():

try:

if os.path.isfile(INSTANCE_FILE):

os.remove(INSTANCE_FILE)

except Exception:

pass

def _ping_instance(port, token, timeout=0.8):

try:

import urllib.request

req = urllib.request.Request(

'http://127.0.0.1:%d/api/ping?t=%s' % (port, token))

with urllib.request.urlopen(req, timeout=timeout) as r:

return bool(json.loads(r.read().decode('utf-8')).get('ok'))

except Exception:

return False

def instance_alive():

"""存在可用的常驻实例则返回 (port, token)，否则 None。"""

d = _read_instance()

port = d.get('port')

token = d.get('token')

if not port or not token:

return None

return (port, token) if _ping_instance(port, token) else None

def forward_open(port, token, path):

"""把文件转发给常驻实例并唤起窗口；成功返回 True。"""

import urllib.request

payload = json.dumps({'token': token, 'file': path or ''}).encode('utf-8')

req = urllib.request.Request(

'http://127.0.0.1:%d/api/control/open' % port,

data=payload, headers={'Content-Type': 'application/json'}, method='POST')

try:

with urllib.request.urlopen(req, timeout=3) as r:

return bool(json.loads(r.read().decode('utf-8')).get('ok'))

except Exception:

return False

def push_control(path):

"""控制请求入队；窗口就绪时立即推送并显示（秒开路径）。"""

with _control_lock:

_CONTROL['queue'].append(path or '')

win = _CONTROL.get('window')

ready = _CONTROL.get('ready')

if win is not None and ready:

try:

win.evaluate_js('window.openExternalFile(%s);' % json.dumps(path or ''))

except Exception:

pass

try:

win.show()

win.restore()

except Exception:

pass

def pop_control():

with _control_lock:

if _CONTROL['queue']:

return _CONTROL['queue'].pop(0)

return None

def push_pet_batch(paths):

"""Request a user-confirmed batch through the already-loaded ReadMD UI."""

safe_paths = [path for path in (paths or ()) if isinstance(path, str) and path]

if not safe_paths:

return False

with _control_lock:

_CONTROL['pet_batches'].append(safe_paths)

win = _CONTROL.get('window')

ready = _CONTROL.get('ready')

if win is not None and ready:

try:

win.evaluate_js('window.receivePetBatch && window.receivePetBatch(%s);' %

json.dumps(safe_paths))

with _control_lock:

if _CONTROL['pet_batches'] and _CONTROL['pet_batches'][0] == safe_paths:

_CONTROL['pet_batches'].pop(0)

except Exception:

pass

try:

win.show()

win.restore()

except Exception:

pass

return True

def pop_pet_batch():

with _control_lock:

if _CONTROL['pet_batches']:

return _CONTROL['pet_batches'].pop(0)

return None

def push_pet_menu():

"""Open the existing More menu from the copied Hermes single-click event."""

with _control_lock:

_CONTROL['pet_menus'] += 1

win = _CONTROL.get('window')

ready = _CONTROL.get('ready')

if win is not None and ready:

try:

win.evaluate_js('window.openPetQuickMenu && window.openPetQuickMenu();')

with _control_lock:

if _CONTROL['pet_menus']:

_CONTROL['pet_menus'] -= 1

except Exception:

pass

try:

win.show()

win.restore()

except Exception:

pass

return True

def pop_pet_menu():

with _control_lock:

if _CONTROL['pet_menus']:

_CONTROL['pet_menus'] -= 1

return True

return False

_ACTIVE_PET_LAUNCHER = None

def register_active_pet_launcher(launcher):

global _ACTIVE_PET_LAUNCHER

_ACTIVE_PET_LAUNCHER = launcher

def stop_active_pet():

global _ACTIVE_PET_LAUNCHER

if _ACTIVE_PET_LAUNCHER is not None:

try:

_ACTIVE_PET_LAUNCHER.stop()

except Exception:

pass

_ACTIVE_PET_LAUNCHER = None

atexit.register(stop_active_pet)

def quit_app():

"""托盘“退出 ReadMD”：清理单实例文件、关闭桌宠后结束进程。"""

try:

stop_active_pet()

except Exception:

pass

try:

_clear_instance()

except Exception:

pass

try:

stop_lan_server()

except Exception:

pass

os._exit(0)

def safe_print(*args, **kwargs):

try:

if sys.stdout is not None:

print(*args, **kwargs)

except Exception:

pass

def setup_logging():

try:

os.makedirs(DATA_DIR, exist_ok=True)

logging.basicConfig(

filename=LOG_FILE, level=logging.INFO, encoding='utf-8',

format='%(asctime)s %(levelname)s %(message)s')

except Exception:

pass

_WINDOWS_RESERVED_NAMES = {

'CON', 'PRN', 'AUX', 'NUL',

*('COM%d' % i for i in range(1, 10)),

*('LPT%d' % i for i in range(1, 10)),

}

def _validate_rename_stem(stem, extension):

stem = str(stem or '')

if not stem or stem != stem.strip():

raise ValueError('文件名不能为空或以空格开头、结尾')

if stem.endswith('.') or any(ord(ch) < 32 for ch in stem):

raise ValueError('文件名包含无效字符')

if any(ch in stem for ch in '<>:"/\\|?*'):

raise ValueError('文件名不能包含 < > : " / \\ | ? *')

if stem.split('.', 1)[0].upper() in _WINDOWS_RESERVED_NAMES:

raise ValueError('该名称是 Windows 系统保留名')

filename = stem + extension

if len(filename) > 255:

raise ValueError('文件名过长')

return stem

def _paths_equal(left, right):

return os.path.normcase(os.path.abspath(left)) == os.path.normcase(os.path.abspath(right))

def _same_file_target(left, right):

"""Handle case-only names on case-insensitive macOS/Windows volumes."""

if _paths_equal(left, right):

return True

try:

return os.path.exists(left) and os.path.exists(right) and os.path.samefile(left, right)

except (OSError, ValueError):

return False

# ---------------------------------------------------------------- AI 模板 / 历史会话

# Built-in actions are metadata only; instruction text lives in assets/skills.

_BUILTIN_ACTIONS = (

("quick_read", "快速阅读", "quick_read", "readmd-quick-read"),

("polish", "润色文稿", "polish", "readmd-polish"),

("proofread", "语法纠错", "proofread", "readmd-proofread"),

("to_english", "翻译为英文", "translate_en", "readmd-translate"),

("to_chinese", "翻译为中文", "translate_zh", "readmd-translate"),

("action_items", "提取待办", "todo", "readmd-todo"),

("continue", "续写内容", "continue", "readmd-continue"),

("ask", "自由提问", "ask", "readmd-ask"),

("summary", "总结要点", "summary", "readmd-summary"),

("outline", "生成大纲", "outline", "readmd-outline"),

("weekly", "生成周报", "weekly", "readmd-weekly"),

("code_review", "代码审查", "code_review", "readmd-code-review"),

("fix_format", "修正格式", "modify", "readmd-format-fix"),

)

def _skill_registry(project_dir=None):

"""Return the canonical shared Skill registry used by every client."""

if project_dir:

return ReadMDCoreService(project_dir).skills

global _DESKTOP_CORE_SERVICE

if _DESKTOP_CORE_SERVICE is None:

_DESKTOP_CORE_SERVICE = ReadMDCoreService()

else:

_DESKTOP_CORE_SERVICE.reload()

return _DESKTOP_CORE_SERVICE.skills

_DESKTOP_CORE_SERVICE = None

def _builtin_prompts():

registry = _skill_registry()

result = []

for template_id, name, action, skill_id in _BUILTIN_ACTIONS:

skill = registry.get(skill_id)

if not skill:

continue

result.append({

"id": template_id,

"skill_id": skill_id,

"name": name,

"action": action,

"system": skill.instructions,

"user": "{doc}\n\n{prompt}",

"builtin": True,

})

return result

BUILTIN_PROMPTS = _builtin_prompts()

def load_prompts():

"""内置 + 自定义模板合并；自定义可覆盖同名内置。"""

d = load_json(PROMPTS_FILE, {})

customs = d.get('templates', [])

by_id = {t.get('id'): t for t in customs}

merged = []

seen = set()

for b in BUILTIN_PROMPTS:

bid = b.get('id')

seen.add(bid)

merged.append(dict(by_id.get(bid, b), builtin=True))

for c in customs:

cid = c.get('id')

if cid in seen:

continue

merged.append(dict(c, builtin=False))

return {'templates': merged}

def _public_skill(skill, include_instructions=False):

# Never return an absolute user path to a renderer, extension or MCP

# client. It is both unnecessary for the workbench and a local privacy

# leak in exported history/screenshots.

try:

safe_path = os.path.relpath(skill.path, skill.root).replace('\\', '/')

except (TypeError, ValueError):

safe_path = skill.id

data = {

'id': skill.id,

'name': skill.name,

'description': skill.description,

'scope': skill.scope,

'variables': skill.variables,

'path': safe_path,

'metadata': dict(skill.metadata),

}

provenance = data['metadata'].get('provenance')

if isinstance(provenance, dict):

data['provenance'] = dict(provenance)

data.setdefault('source_files', data['metadata'].get('source_files', []))

data.setdefault('license', data['metadata'].get('license', ''))

data.setdefault('adaptation_notes', data['metadata'].get('adaptation_notes', []))

if include_instructions:

data['instructions'] = skill.instructions

return data

def load_skills(project_dir=None):

"""List Skills from builtin, user and optional project roots."""

# The workbench needs a local preview without another round trip. This is

# still instruction text (never credentials) and follows the same registry

# precedence and path checks as the read endpoint.

return [_public_skill(skill, include_instructions=True)

for skill in _skill_registry(project_dir).list(include_disabled=True)]

def _user_skill_folder(skill_id):

if not re.fullmatch(r'[a-z0-9][a-z0-9-]{0,63}', str(skill_id or '')):

raise SkillError('Skill id must be lowercase kebab-case')

folder = os.path.realpath(os.path.join(DATA_DIR, 'skills', str(skill_id)))

root = os.path.realpath(os.path.join(DATA_DIR, 'skills'))

if not paths_within(folder, root) or folder == root:

raise SkillError('Skill path is outside the user Skill directory')

return folder

def validate_skill_document(skill_id, content, metadata=None):

"""Validate a Skill document in memory before it is persisted."""

if not re.fullmatch(r'[a-z0-9][a-z0-9-]{0,63}', str(skill_id or '')):

raise SkillError('Skill id must be lowercase kebab-case')

if not isinstance(content, str) or not content.strip() or len(content.encode('utf-8')) > 512 * 1024:

raise SkillError('Skill content is empty or too large')

import tempfile

with tempfile.TemporaryDirectory(prefix='readmd-skill-') as tmp:

folder = os.path.join(tmp, str(skill_id))

os.makedirs(folder, exist_ok=True)

with open(os.path.join(folder, 'SKILL.md'), 'w', encoding='utf-8', newline='\n') as handle:

handle.write(content)

if metadata is not None:

with open(os.path.join(folder, 'readmd.skill.json'), 'w', encoding='utf-8', newline='\n') as handle:

json.dump(metadata, handle, ensure_ascii=False, indent=2)

skill = SkillRegistry([tmp]).get(str(skill_id))

if skill is None:

raise SkillError('Skill was not discovered after validation')

return _public_skill(skill, include_instructions=True)

def save_user_skill(skill_id, content, metadata=None):

"""Atomically publish a validated user Skill; scripts are never enabled."""

validated = validate_skill_document(skill_id, content, metadata)

folder = _user_skill_folder(skill_id)

# Keep a local rollback snapshot before replacing an existing user Skill.

if os.path.isfile(os.path.join(folder, 'SKILL.md')):

import shutil

stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')

version_dir = os.path.join(os.path.dirname(folder), '.versions', str(skill_id), stamp)

os.makedirs(version_dir, exist_ok=True)

shutil.copy2(os.path.join(folder, 'SKILL.md'), os.path.join(version_dir, 'SKILL.md'))

old_meta = os.path.join(folder, 'readmd.skill.json')

if os.path.isfile(old_meta):

shutil.copy2(old_meta, os.path.join(version_dir, 'readmd.skill.json'))

os.makedirs(folder, exist_ok=True)

save_text_atomic(os.path.join(folder, 'SKILL.md'), content.strip() + '\n')

if metadata is not None:

safe_meta = dict(metadata)

safe_meta['scripts_allowed'] = False

save_text_atomic(os.path.join(folder, 'readmd.skill.json'), json.dumps(safe_meta, ensure_ascii=False, indent=2) + '\n')

return validated

def _skill_versions(skill_id):

"""Return rollback snapshots for one user Skill, newest first."""

_user_skill_folder(skill_id)

root = os.path.join(DATA_DIR, 'skills', '.versions', str(skill_id))

if not os.path.isdir(root):

return []

return [name for name in sorted(os.listdir(root), reverse=True)

if re.fullmatch(r'\d{8}T\d{6}Z', name)

and os.path.isfile(os.path.join(root, name, 'SKILL.md'))]

def save_prompt(template):

"""新增 / 更新模板。id 为空时自动生成；内置 id 表示覆盖内置模板。"""

t = dict(template or {})

if not t.get('id'):

t['id'] = 't_%d' % int(time.time() * 1000)

if not t.get('name'):

t['name'] = '未命名模板'

t.pop('builtin', None)

# 兼容旧版自定义模板：第一次保存时将 system 指令迁移为用户 Skill。

# 迁移只写入 DATA_DIR/skills，且由同一套 Skill 校验器检查，避免继续

# 在 prompts.json 中维护第二份可执行 Prompt 实现。

if not t.get('skill_id') and str(t.get('system') or '').strip():

raw_id = re.sub(r'[^a-z0-9-]+', '-', str(t.get('id') or '').lower()).strip('-')

skill_id = ('prompt-' + raw_id)[:64].rstrip('-') or ('prompt-%d' % int(time.time() * 1000))

system = str(t.get('system') or '').strip()

skill_doc = (

'---\n'

'name: %s\n'

'description: Use when running the custom ReadMD prompt %s.\n'

'---\n\n%s\n' % (skill_id, str(t.get('name') or skill_id), system)

)

try:

save_user_skill(skill_id, skill_doc, {

'id': skill_id,

'source': 'legacy-prompt-migration',

'version': 1,

'scripts_allowed': False,

'legacy_template_id': t.get('id'),

})

t['skill_id'] = skill_id

# Keep user text template only as a presentation/compatibility field;

# the system instruction is now owned by SKILL.md.

t.pop('system', None)

except SkillError:

# Do not persist an unvalidated legacy Prompt. The caller receives a

# normal validation error instead of silently retaining raw Prompt code.

raise

d = load_json(PROMPTS_FILE, {})

customs = [c for c in d.get('templates', []) if c.get('id') != t.get('id')]

customs.append(t)

save_json(PROMPTS_FILE, {'templates': customs})

return t

def delete_prompt(prompt_id):

d = load_json(PROMPTS_FILE, {})

removed = [t for t in d.get('templates', []) if t.get('id') == prompt_id]

d['templates'] = [t for t in d.get('templates', []) if t.get('id') != prompt_id]

save_json(PROMPTS_FILE, d)

# Remove only the exact migrated user Skill, never a builtin/project Skill.

for template in removed:

skill_id = str(template.get('skill_id') or '')

if skill_id.startswith('prompt-'):

try:

folder = _user_skill_folder(skill_id)

if os.path.isdir(folder):

import shutil

shutil.rmtree(folder)

except (OSError, SkillError):

logging.debug('legacy prompt skill cleanup failed', exc_info=True)

return True

def load_history(limit=50):

d = load_json(HISTORY_FILE, {'sessions': []})

return d.get('sessions', [])[:limit]

def save_session(session):

"""新增 / 更新会话（按 id upsert），限制会话 50 个、消息 60 条。"""

s = dict(session or {})

now = time.time()

if not s.get('id'):

s['id'] = 'h_%d' % int(now * 1000)

s['created'] = s.get('created') or now

s['updated'] = now

msgs = (s.get('messages') or [])[-60:]

s['messages'] = msgs

s['msgCount'] = len(msgs)

sessions = [x for x in load_history(500) if x.get('id') != s['id']]

sessions.insert(0, s)

save_json(HISTORY_FILE, {'sessions': sessions[:50]})

return s

def delete_session(session_id):

sessions = [x for x in load_history(500) if x.get('id') != session_id]

save_json(HISTORY_FILE, {'sessions': sessions})

return True

def _md_output_path(src):

"""转换输出路径：源文件同目录同名 .md。"""

d = os.path.dirname(os.path.abspath(src))

base = os.path.splitext(os.path.basename(src))[0]

return os.path.join(d, base + '.md')

def _is_upload_path(src):

"""Check whether a file path resides within the uploads temporary directory."""

if not src:

return False

upload_dir = os.path.realpath(os.path.join(DATA_DIR, 'uploads'))

src_real = os.path.realpath(os.path.abspath(src))

return src_real.startswith(upload_dir + os.sep) or src_real == upload_dir

def _batch_output_paths(paths):

"""Plan collision-free Markdown targets without touching source files."""

planned, used = {}, set()

seen_sources = set()

for src in paths:

source_key = os.path.normcase(os.path.realpath(os.path.abspath(src)))

if source_key in seen_sources:

continue

seen_sources.add(source_key)

candidate = _md_output_path(src)

key = os.path.normcase(os.path.abspath(candidate))

if key not in used:

planned[src] = candidate

used.add(key)

continue

try:

with open(src, 'rb') as handle:

digest = hashlib.sha256(handle.read()).hexdigest()[:8]

except OSError:

digest = hashlib.sha256(os.path.abspath(src).encode('utf-8', errors='replace')).hexdigest()[:8]

stem, ext = os.path.splitext(candidate)

candidate = '%s-%s%s' % (stem, digest, ext)

suffix = 2

while os.path.normcase(os.path.abspath(candidate)) in used:

candidate = '%s-%s-%d%s' % (stem, digest, suffix, ext)

suffix += 1

planned[src] = candidate

used.add(os.path.normcase(os.path.abspath(candidate)))

return planned

def _write_md(path, content):

with open(path, 'w', encoding='utf-8', newline='\n') as f:

f.write(content)

return True

def _safe_export_target(path, suffix):

"""Validate a caller-supplied export target without requiring it to exist."""

raw = os.fspath(path or '')

if not raw or '\x00' in raw or any(ord(ch) < 32 for ch in raw):

raise ValueError('invalid_output_path')

candidate = os.path.realpath(os.path.abspath(raw))

if not candidate.lower().endswith(str(suffix).lower()):

raise ValueError('invalid_output_extension')

parent = os.path.dirname(candidate)

if not os.path.isdir(parent):

raise ValueError('output_directory_not_found')

if os.path.lexists(candidate) and not os.path.isfile(candidate):

raise ValueError('output_target_not_regular')

return candidate

def _convert_worker(job):

items = job['items']

for it in items:

if job.get('cancel'):

it['status'] = 'canceled'

it['done'] = True

continue

it['status'] = 'running'

try:

mod = RM.get('convert')

text, engine, err = mod.convert_verbose(it['src'])

if err and not text:

it['status'] = 'error'

it['error'] = err

it['error_code'] = 'conversion_failed'

it['done'] = True

continue

if not text.strip():

it['status'] = 'error'

it['error'] = '未提取到文字（可尝试 OCR）'

it['error_code'] = 'empty_output'

it['done'] = True

continue

import src.readmd_modules.mdcheck as MDC

fixed, warns = MDC.check(text, os.path.dirname(os.path.abspath(it['src'])))

out = it.get('planned_out') or _md_output_path(it['src'])

it['out'] = out

it['engine'] = engine

it['warns'] = warns

allow_overwrite = bool(job.get('overwrite')) or _is_upload_path(it['src'])

if os.path.exists(out) and not allow_overwrite:

it['status'] = 'skipped'

it['error_code'] = 'output_exists'

else:

try:

_write_md(out, fixed)

it['status'] = 'ok'

except Exception as e: # noqa: BLE001

it['status'] = 'error'

it['error'] = '写入失败：%s' % e

it['error_code'] = 'write_failed'

except Exception as e: # noqa: BLE001

logging.exception('batch convert failed: %s', it.get('src'))

it['status'] = 'error'

it['error'] = str(e)

it['error_code'] = 'conversion_failed'

it['done'] = True

job['running'] = False

job['finished'] = True

def _start_convert_job(paths, overwrite):

with _CONVERT_LOCK:

_CONVERT_JOB_SEQ[0] += 1

jid = 'c%d' % _CONVERT_JOB_SEQ[0]

outputs = _batch_output_paths(paths)

job = {'id': jid, 'overwrite': bool(overwrite), 'running': True,

'finished': False, 'cancel': False,

'items': [{'src': p, 'planned_out': outputs.get(p),

'status': 'queued', 'done': False} for p in paths]}

_CONVERT_JOBS[jid] = job

threading.Thread(target=_convert_worker, args=(job,), daemon=True,

name='convert-batch-%s' % jid).start()

return jid

def read_text(path):

"""按编码优先级读取文本文件（UTF-8 / GB18030 / Big5 / Latin-1）。"""

with open(path, 'rb') as f:

data = f.read()

if data.startswith(b'\xef\xbb\xbf'):

return data.decode('utf-8-sig'), 'utf-8-sig'

for enc in ('utf-8', 'gb18030', 'big5', 'latin-1'):

try:

return data.decode(enc), enc

except (UnicodeDecodeError, LookupError):

continue

return data.decode('utf-8', errors='replace'), 'utf-8'

# ---------------------------------------------------------------- HTTP 服务

SAVE_EXTENSIONS = frozenset(('.md', '.markdown', '.mdown', '.mkd', '.mdx', '.txt'))

class ReadMDHTTPServer(ThreadingHTTPServer):

daemon_threads = True

request_queue_size = 128

def __init__(self, *args, **kwargs):

super().__init__(*args, **kwargs)

self.app_token = secrets.token_urlsafe(24)

self.authorized_save_paths = set()

_SKILL_EVALUATION_TOKENS = {}

_SKILL_EVALUATION_TTL = 10 * 60

def _skill_content_digest(skill_id, content):

payload = ('%s\0%s' % (skill_id, content)).encode('utf-8')

return hashlib.sha256(payload).hexdigest()

def _issue_skill_evaluation_token(skill_id, content):

token = secrets.token_urlsafe(32)

_SKILL_EVALUATION_TOKENS[token] = {

'digest': _skill_content_digest(skill_id, content),

'expires': time.time() + _SKILL_EVALUATION_TTL,

}

# Keep the in-memory registry bounded even if a client abandons drafts.

now = time.time()

for key, value in list(_SKILL_EVALUATION_TOKENS.items()):

if value.get('expires', 0) <= now:

_SKILL_EVALUATION_TOKENS.pop(key, None)

return token

def _consume_skill_evaluation_token(token, skill_id, content):

if not isinstance(token, str) or not token:

return False

record = _SKILL_EVALUATION_TOKENS.pop(token, None)

if not record or record.get('expires', 0) <= time.time():

return False

return secrets.compare_digest(

record.get('digest', ''), _skill_content_digest(skill_id, content))