227 lines
11 KiB
Python
227 lines
11 KiB
Python
"""幂等转录本导出，无需覆盖已编辑的笔记。"""
|
||
import asyncio
|
||
import hashlib
|
||
from contextlib import closing
|
||
|
||
from app.config import get_settings
|
||
from app.contracts import Message, MessageRole, ModelRequest, TranscriptNoteRequest
|
||
from app.database.db import connect, transaction
|
||
from app.errors import ApiError
|
||
from app.providers.base import ProviderError
|
||
from app.providers.registry import ProviderNotFoundError
|
||
from app.services import note_service
|
||
from app.services.transcription_service import require_job
|
||
|
||
_locks = {}
|
||
|
||
|
||
async def create_transcript_note(job_id, options):
|
||
identity = (str(get_settings().db_path), job_id)
|
||
lock = _locks.setdefault(identity, asyncio.Lock())
|
||
async with lock:
|
||
job = require_job(job_id)
|
||
if job.status != "completed":
|
||
raise ApiError(409, "TRANSCRIPT_NOT_READY", "Only completed transcripts can become notes.")
|
||
options_hash = hashlib.sha256(options.model_copy(update={"update_existing": False}).model_dump_json(exclude={"update_existing"}).encode()).hexdigest()
|
||
with closing(connect()) as conn:
|
||
conn.execute("CREATE TABLE IF NOT EXISTS media_note_baselines (note_id TEXT PRIMARY KEY, content_hash TEXT NOT NULL)")
|
||
previous = conn.execute("SELECT m.note_id,b.content_hash FROM media_notes m LEFT JOIN media_note_baselines b ON b.note_id=m.note_id WHERE m.job_id=? AND m.options_hash=? ORDER BY m.revision DESC LIMIT 1", (job_id, options_hash)).fetchone()
|
||
row = conn.execute("SELECT note_id FROM media_notes WHERE job_id=? AND revision=? AND options_hash=?",
|
||
(job_id, job.revision, options_hash)).fetchone()
|
||
if row:
|
||
return await note_service.get_note(row[0])
|
||
marker = f"<!-- transcription:{job_id}:{job.revision}:{options_hash} -->"
|
||
title = f"{options.title} · {job_id[-8:]}-r{job.revision}-{options_hash[:6]}"
|
||
lines = [marker, f"# {options.title}", "", f"[源音频](/#/media?job={job_id})", ""]
|
||
if job.segments:
|
||
for segment in job.segments:
|
||
prefix = []
|
||
if options.include_timestamps:
|
||
seconds = segment.start_time
|
||
label = f"{int(seconds // 60):02}:{int(seconds % 60):02}"
|
||
prefix.append(f"[{label}](/#/media?job={job_id}&time={seconds})")
|
||
if options.include_speakers and segment.speaker:
|
||
prefix.append(job.speaker_names.get(segment.speaker, segment.speaker))
|
||
lines.append(" ".join([*prefix, segment.text]))
|
||
lines.append("")
|
||
else:
|
||
lines.append(job.text or "")
|
||
if job.local_only:
|
||
# 保留 Vault 中的索引策略，包括以后的重建。
|
||
lines = ["---", "embedding_local_only: true", "---", "", *lines]
|
||
markdown = "\n".join(lines)
|
||
if options.update_existing:
|
||
if previous is None or previous[1] is None:
|
||
raise ApiError(409, "NOTE_UPDATE_BASELINE_MISSING", "没有可安全更新的导出记录，请先创建新笔记。")
|
||
current = await note_service.get_note(previous[0])
|
||
if current is None:
|
||
raise ApiError(404, "RESOURCE_NOT_FOUND", "已导出笔记不存在。")
|
||
# 如果 Vault 写入后链接失败，则恢复成功更新。
|
||
if current.markdown == markdown:
|
||
note = current
|
||
else:
|
||
note = await note_service.update_note(previous[0], markdown=markdown, expected_content_hash=previous[1])
|
||
else:
|
||
note = await _create_note(title, markdown, options, marker)
|
||
with closing(connect()) as conn, transaction(conn):
|
||
conn.execute("INSERT OR IGNORE INTO media_notes VALUES (?,?,?,?)", (job_id, job.revision, options_hash, note.note_id))
|
||
conn.execute("INSERT OR REPLACE INTO media_note_baselines VALUES (?,?)", (note.note_id, hashlib.sha256(markdown.encode()).hexdigest()))
|
||
return note
|
||
|
||
|
||
def _transcript_text(job) -> str:
|
||
if job.segments:
|
||
rows = []
|
||
for segment in job.segments:
|
||
speaker = job.speaker_names.get(segment.speaker, segment.speaker) if segment.speaker else ""
|
||
stamp = f"{int(segment.start_time // 60):02}:{int(segment.start_time % 60):02}"
|
||
rows.append(f"[{stamp}] {speaker}：{segment.text}" if speaker else f"[{stamp}] {segment.text}")
|
||
return "\n".join(rows)
|
||
return job.text or ""
|
||
|
||
|
||
def _chunks(text: str, limit: int = 12000) -> list[str]:
|
||
"""按段落切分长转录，避免在中间截断句子。"""
|
||
paragraphs = [part.strip() for part in text.splitlines() if part.strip()]
|
||
if not paragraphs:
|
||
return []
|
||
chunks: list[str] = []
|
||
current: list[str] = []
|
||
size = 0
|
||
for paragraph in paragraphs:
|
||
if current and size + len(paragraph) + 1 > limit:
|
||
chunks.append("\n".join(current))
|
||
current, size = [], 0
|
||
if len(paragraph) > limit:
|
||
if current:
|
||
chunks.append("\n".join(current))
|
||
current, size = [], 0
|
||
chunks.extend(paragraph[index:index + limit] for index in range(0, len(paragraph), limit))
|
||
continue
|
||
current.append(paragraph)
|
||
size += len(paragraph) + 1
|
||
if current:
|
||
chunks.append("\n".join(current))
|
||
return chunks
|
||
|
||
|
||
async def _complete(provider_id: str, model: str, system: str, content: str) -> str:
|
||
from app.container import container
|
||
try:
|
||
provider = container.providers.get(provider_id).adapter
|
||
except ProviderNotFoundError as exc:
|
||
raise ApiError(404, "PROVIDER_NOT_FOUND", "所选模型提供商不存在或未启用。",
|
||
{"provider_id": provider_id}) from exc
|
||
try:
|
||
turn = await provider.complete(ModelRequest(
|
||
provider_id=provider_id,
|
||
model=model,
|
||
system=system,
|
||
messages=[Message(role=MessageRole.user, content=content)],
|
||
temperature=0.2,
|
||
))
|
||
except ProviderError as exc:
|
||
raise ApiError(502, exc.code, exc.message, {"provider_id": provider_id}) from exc
|
||
if not turn.text or not turn.text.strip():
|
||
raise ApiError(502, "KNOWLEDGE_NOTE_EMPTY", "模型没有返回知识点笔记。")
|
||
return turn.text.strip().removeprefix("```markdown").removeprefix("```").removesuffix("```").strip()
|
||
|
||
|
||
async def _knowledge_markdown(job, provider_id: str, model: str, title: str) -> str:
|
||
transcript = _transcript_text(job)
|
||
if not transcript.strip():
|
||
raise ApiError(409, "TRANSCRIPT_EMPTY", "转录内容为空，无法提取知识点。")
|
||
system = (
|
||
"你是一名严谨的课程笔记整理助手。只能依据提供的转录内容整理，不补写未出现的事实。"
|
||
"输出中文 Markdown 正文，使用清晰的二级、三级标题；包含课程主题、核心概念、关键论证或步骤、"
|
||
"重要例子、待复习问题。合并口语重复，保留专业术语和必要条件。不要使用代码围栏，也不要写处理说明。"
|
||
)
|
||
parts = _chunks(transcript)
|
||
summaries: list[str] = []
|
||
for index, part in enumerate(parts, 1):
|
||
summaries.append(await _complete(
|
||
provider_id, model, system,
|
||
f"这是课程转录的第 {index}/{len(parts)} 部分。请提取可供最终整合的知识点：\n\n{part}",
|
||
))
|
||
if len(summaries) == 1:
|
||
body = summaries[0]
|
||
else:
|
||
body = await _complete(
|
||
provider_id, model, system,
|
||
"请将以下分段知识点合并成一篇完整课程笔记，消除重复并保持逻辑顺序：\n\n"
|
||
+ "\n\n".join(f"### 分段 {index}\n{summary}" for index, summary in enumerate(summaries, 1)),
|
||
)
|
||
return "\n".join([
|
||
f"<!-- knowledge-note:{job.job_id}:{job.revision}:{provider_id}:{model} -->",
|
||
f"# {title}", "", f"[查看完整转录稿](/#/media?job={job.job_id})", "", body,
|
||
])
|
||
|
||
|
||
async def create_transcript_artifacts(job_id, options):
|
||
"""为完成的转录生成可回听的全文和模型整理的知识点笔记。"""
|
||
transcript_options = TranscriptNoteRequest(
|
||
title=options.title,
|
||
folder=options.folder,
|
||
update_existing=options.update_existing,
|
||
include_timestamps=options.include_timestamps,
|
||
include_speakers=options.include_speakers,
|
||
)
|
||
transcript_note = await create_transcript_note(job_id, transcript_options)
|
||
job = require_job(job_id)
|
||
knowledge_title = options.knowledge_title or f"{options.title} · 知识点"
|
||
identity = (str(get_settings().db_path), job_id, "knowledge")
|
||
lock = _locks.setdefault(identity, asyncio.Lock())
|
||
async with lock:
|
||
signature = "knowledge:" + hashlib.sha256(options.model_copy(update={
|
||
"update_existing": False,
|
||
"knowledge_title": knowledge_title,
|
||
}).model_dump_json(exclude={"update_existing"}).encode()).hexdigest()
|
||
with closing(connect()) as conn:
|
||
row = conn.execute(
|
||
"SELECT note_id FROM media_notes WHERE job_id=? AND revision=? AND options_hash=?",
|
||
(job_id, job.revision, signature),
|
||
).fetchone()
|
||
if row:
|
||
knowledge_note = await note_service.get_note(row[0])
|
||
if knowledge_note is not None:
|
||
return {"transcript": transcript_note, "knowledge_note": knowledge_note}
|
||
markdown = await _knowledge_markdown(job, options.provider_id, options.model, knowledge_title)
|
||
note_title = f"{knowledge_title} · {job_id[-8:]}-r{job.revision}-{signature[-6:]}"
|
||
marker = markdown.splitlines()[0]
|
||
knowledge_note = await _create_note(
|
||
note_title, markdown, options, marker, tags=["课程笔记", "知识点"]
|
||
)
|
||
with closing(connect()) as conn, transaction(conn):
|
||
conn.execute("INSERT OR IGNORE INTO media_notes VALUES (?,?,?,?)",
|
||
(job_id, job.revision, signature, knowledge_note.note_id))
|
||
return {"transcript": transcript_note, "knowledge_note": knowledge_note}
|
||
|
||
|
||
async def _create_note(title, markdown, options, marker, *, tags=None):
|
||
try:
|
||
note = await note_service.create_note(
|
||
title=title, markdown=markdown, folder=options.folder, tags=tags or ["转写"]
|
||
)
|
||
except ApiError as exc:
|
||
if exc.code == "RESOURCE_CONFLICT" and "note_id" in exc.details:
|
||
note = await note_service.get_note(exc.details["note_id"])
|
||
elif exc.code == "REVISION_CONFLICT":
|
||
# Rust Host 已完成写入、但 Core 尚未来得及保存关联时，重试会报告路径冲突。
|
||
# 只恢复标题和不可伪造的任务 marker 都匹配的文件，避免误认用户同名笔记。
|
||
summaries, _ = note_service.list_notes(
|
||
limit=1000, offset=0, folder=options.folder, tag=None
|
||
)
|
||
note = None
|
||
for summary in summaries:
|
||
if summary.title != title:
|
||
continue
|
||
candidate = await note_service.get_note(summary.note_id)
|
||
if candidate is not None and marker in candidate.markdown:
|
||
note = candidate
|
||
break
|
||
else:
|
||
raise
|
||
if note is None or marker not in note.markdown:
|
||
raise
|
||
return note
|