Skip to content

沈秋雨 / import_hk_songs

Go to a project
Toggle navigation
Toggle navigation pinning
Commit cbe31001 cbe3100183b9d29e7568b7b5591baf7f12205960 by 沈秋雨
Options

feat(review): 新增审核编辑暗号及会话登录功能 

- 在 .env.example 添加 REVIEW_ACCESS_CODE 和 REVIEW_SESSION_TTL_SECONDS 配置项
- 在 l2_review_dashboard.html 增加审核登录弹窗界面及样式
- 实现异步登录流程，支持昵称和暗号输入，完成服务端认证
- 使用本地存储和会话存储管理审核者昵称和认证令牌
- 对所有编辑相关的 API 请求附加认证信息，实现服务端权限校验
- 服务端新增审核登录和会话验证接口，支持暗号验证和会话续期
- 审核操作接口添加会话有效性检查，非法或过期会话返回 401
- 新增 ReviewAuthError 自定义异常用于认证错误处理
- 修改入库脚本执行错误时返回更友好信息
- 相关前后端代码统一处理认证状态，支持登录失效自动刷新提醒
- README.md 中新增审核登录暗号配置及使用说明
- 添加后端认证逻辑的单元测试，验证正确登录和错误拒绝情况
1 parent 96799bc7
Inline Side-by-side
Showing 5 changed files with 267 additions and 35 deletions
......@@ -14,6 +14,11 @@ TARGET_DB_NAME=
TARGET_TABLE_NAME=hk_songs_test
TARGET_TABLE_NAME_TMP=hk_songs_import_staging
# L2 审核页面编辑暗号；未配置时所有写操作禁用
REVIEW_ACCESS_CODE=
# 编辑会话有效期（秒），默认 12 小时
REVIEW_SESSION_TTL_SECONDS=43200
# run_etl.py 默认读取并回写的导入状态表：yinyan_song_records 或 yinyan_song_records2
YINYAN_IMPORT_TABLE=yinyan_song_records
......
......@@ -144,6 +144,14 @@ python -m pytest test_dedup.py::TestL2Benchmark::test_l2_recall_topk_efficiency_
### 7. 启动复核前端
先在 `.env` 配置审核编辑暗号（支持中文或英文）：
```dotenv
REVIEW_ACCESS_CODE=请替换成你的暗号
```
暗号只在服务端校验。验证成功后，当前浏览器标签页会获得默认有效期 12 小时的临时编辑会话；未配置暗号或验证失败时，领取、审核、撤销、删除和入库接口均不可调用。
首次启用多人审核时，先让服务补齐领取和乐观锁字段（已有审核结果不会被修改）：
```bash
......
......@@ -547,6 +547,34 @@
#reviewerInput { width: 130px; }
.login-overlay {
position: fixed;
inset: 0;
z-index: 1000;
display: flex;
align-items: center;
justify-content: center;
background: rgba(245, 247, 250, .96);
}
.login-overlay.hidden { display: none; }
.login-panel {
width: min(380px, calc(100vw - 32px));
padding: 24px;
border: 1px solid var(--line);
border-radius: 12px;
background: #fff;
box-shadow: 0 12px 36px rgba(30, 48, 70, .16);
}
.login-panel h2 { margin: 0 0 8px; }
.login-panel p { margin: 0 0 18px; color: var(--muted); font-size: 13px; }
.login-panel label { display: block; margin: 10px 0 5px; font-size: 13px; }
.login-panel input { width: 100%; box-sizing: border-box; }
.login-panel button { width: 100%; margin-top: 16px; }
.login-error { min-height: 20px; margin-top: 10px; color: #b42318; font-size: 13px; }
@media (max-width: 1100px) {
main { grid-template-columns: 1fr; height: auto; }
aside { border-right: 0; border-bottom: 1px solid var(--line); max-height: 420px; }
......@@ -558,6 +586,18 @@
</style>
</head>
<body>
<div id="loginOverlay" class="login-overlay">
<form id="loginForm" class="login-panel">
<h2>进入审核</h2>
<p>昵称用于记录审核人；暗号仅发送到服务端验证。</p>
<label for="loginReviewer">昵称</label>
<input id="loginReviewer" maxlength="64" autocomplete="username" required>
<label for="loginAccessCode">暗号</label>
<input id="loginAccessCode" type="password" autocomplete="current-password" required>
<button id="loginBtn" type="submit">进入编辑</button>
<div id="loginError" class="login-error"></div>
</form>
</div>
<header>
<h1>歌词召回复核</h1>
<div class="toolbar">
......@@ -646,6 +686,7 @@
lyricsCache: new Map(),
reviewer: '',
clientToken: '',
authToken: '',
syncBusy: false,
detailGeneration: 0,
queryLoadGeneration: 0,
......@@ -671,7 +712,13 @@
queryList: document.getElementById('queryList'),
detail: document.getElementById('scrollDetail'),
stickyInfo: document.getElementById('stickyInfo'),
syncNotice: document.getElementById('syncNotice')
syncNotice: document.getElementById('syncNotice'),
loginOverlay: document.getElementById('loginOverlay'),
loginForm: document.getElementById('loginForm'),
loginReviewer: document.getElementById('loginReviewer'),
loginAccessCode: document.getElementById('loginAccessCode'),
loginBtn: document.getElementById('loginBtn'),
loginError: document.getElementById('loginError')
};
function newClientToken() {
......@@ -679,16 +726,65 @@
return `${Date.now()}-${Math.random().toString(16).slice(2)}`;
}
function initReviewerIdentity() {
function showReviewLogin(prefillReviewer = '') {
els.loginReviewer.value = prefillReviewer;
els.loginAccessCode.value = '';
els.loginError.textContent = '';
els.loginOverlay.classList.remove('hidden');
window.setTimeout(() => (prefillReviewer ? els.loginAccessCode : els.loginReviewer).focus(), 0);
return new Promise(resolve => {
els.loginForm.onsubmit = async event => {
event.preventDefault();
const reviewer = els.loginReviewer.value.trim();
const accessCode = els.loginAccessCode.value;
els.loginBtn.disabled = true;
els.loginError.textContent = '';
try {
const session = await postJSON('/api/review-login', {
reviewer,
access_code: accessCode
});
state.reviewer = session.reviewer;
state.authToken = session.session_token;
localStorage.setItem('l2ReviewerName', state.reviewer);
sessionStorage.setItem('l2ReviewAuthToken', state.authToken);
els.reviewerInput.value = state.reviewer;
els.loginAccessCode.value = '';
els.loginOverlay.classList.add('hidden');
resolve();
} catch (error) {
els.loginError.textContent = error.message;
} finally {
els.loginBtn.disabled = false;
}
};
});
}
async function initReviewerIdentity() {
state.reviewer = (localStorage.getItem('l2ReviewerName') || '').trim();
if (!state.reviewer) {
state.reviewer = (window.prompt('请输入你的昵称') || '').trim();
}
if (!state.reviewer) throw new Error('必须填写你的昵称后才能开始审核');
localStorage.setItem('l2ReviewerName', state.reviewer);
els.reviewerInput.value = state.reviewer;
state.clientToken = sessionStorage.getItem('l2ReviewClientToken') || newClientToken();
sessionStorage.setItem('l2ReviewClientToken', state.clientToken);
state.authToken = sessionStorage.getItem('l2ReviewAuthToken') || '';
els.loginReviewer.value = state.reviewer;
if (state.reviewer && state.authToken) {
try {
const session = await postJSON('/api/review-login', {
reviewer: state.reviewer,
session_token: state.authToken
});
state.authToken = session.session_token;
els.reviewerInput.value = state.reviewer;
els.loginOverlay.classList.add('hidden');
return;
} catch {
sessionStorage.removeItem('l2ReviewAuthToken');
state.authToken = '';
}
}
await showReviewLogin(state.reviewer);
}
function showSyncNotice(message) {
......@@ -807,6 +903,24 @@
return payload;
}
async function reviewPostJSON(url, body = {}) {
try {
return await postJSON(url, {
...body,
reviewer: state.reviewer,
session_token: state.authToken
});
} catch (error) {
if (error.status === 401) {
sessionStorage.removeItem('l2ReviewAuthToken');
state.authToken = '';
alert(`${error.message}\n页面将刷新，请重新输入暗号。`);
window.location.reload();
}
throw error;
}
}
function applyReviewSnapshot(row, record) {
if (!row || !record) return;
row.review_status = record.biz_review_status ?? row.review_status;
......@@ -847,7 +961,7 @@
// new / merge / skip 等记录只浏览，不应触发领取请求。
if (!reviewable) return true;
try {
const payload = await postJSON('/api/review-claim', {
const payload = await reviewPostJSON('/api/review-claim', {
// bigint 必须按字符串传输，不能转成会丢精度的 JavaScript Number。
staging_id: String(query.staging_id),
reviewer: state.reviewer,
......@@ -994,7 +1108,7 @@
}
async function updateStagingReview(query, decision, note) {
return postJSON('/api/staging-review', {
return reviewPostJSON('/api/staging-review', {
staging_id: String(query.staging_id),
expected_version: Number(query.review_version || 0),
client_token: state.clientToken,
......@@ -1344,7 +1458,7 @@
els.importReviewedBtn.disabled = true;
els.importReviewedBtn.textContent = '入库中...';
try {
const payload = await postJSON('/api/import-reviewed', { reviewer: state.reviewer });
const payload = await reviewPostJSON('/api/import-reviewed');
alert(`入库完成：${payload.inserted_count} 条。结果文件：${payload.review_csv}`);
await loadGroups();
await reloadSummary();
......@@ -1468,7 +1582,7 @@
if (state.run?.type !== 'staging' || !query?.staging_id || !query.review_claimed_by) return;
const heartbeatStagingId = String(query.staging_id);
try {
const payload = await postJSON('/api/review-heartbeat', {
const payload = await reviewPostJSON('/api/review-heartbeat', {
staging_id: String(query.staging_id),
client_token: state.clientToken
});
......@@ -1528,9 +1642,11 @@
els.reviewerInput.value = state.reviewer;
return;
}
state.reviewer = nextReviewer;
els.reviewerInput.value = state.reviewer;
sessionStorage.removeItem('l2ReviewAuthToken');
state.authToken = '';
await showReviewLogin(nextReviewer);
state.clientToken = newClientToken();
localStorage.setItem('l2ReviewerName', state.reviewer);
sessionStorage.setItem('l2ReviewClientToken', state.clientToken);
await loadGroups();
renderAll();
......@@ -1589,23 +1705,21 @@
renderAll();
});
try {
initReviewerIdentity();
} catch (err) {
els.detail.innerHTML = `<div class="empty">${esc(err.message)}</div>`;
throw err;
}
window.setInterval(() => {
if (!document.hidden) syncVisibleReviewStatuses();
}, 15000);
window.setInterval(() => {
if (!document.hidden) reloadSummary().catch(console.error);
}, 60000);
window.setInterval(() => {
if (!document.hidden) heartbeatCurrentReview();
}, 60000);
loadRuns().catch(err => {
async function bootstrap() {
await initReviewerIdentity();
window.setInterval(() => {
if (!document.hidden) syncVisibleReviewStatuses();
}, 15000);
window.setInterval(() => {
if (!document.hidden) reloadSummary().catch(console.error);
}, 60000);
window.setInterval(() => {
if (!document.hidden) heartbeatCurrentReview();
}, 60000);
await loadRuns();
}
bootstrap().catch(err => {
els.detail.innerHTML = `<div class="empty">${esc(err.message)}</div>`;
});
</script>
......
......@@ -8,8 +8,10 @@ import csv
import json
import mimetypes
import os
import secrets
import subprocess
import sys
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
......@@ -27,6 +29,10 @@ REPORT_DIR.mkdir(parents=True, exist_ok=True)
GROUP_INDEX_CACHE: dict[tuple[str, int, int, str], list[dict[str, object]]] = {}
load_dotenv(ROOT / ".env")
REVIEW_CLAIM_TTL_SECONDS = max(60, int(os.getenv("REVIEW_CLAIM_TTL_SECONDS", "600")))
REVIEW_ACCESS_CODE = os.getenv("REVIEW_ACCESS_CODE", "")
REVIEW_SESSION_TTL_SECONDS = max(300, int(os.getenv("REVIEW_SESSION_TTL_SECONDS", "43200")))
REVIEW_SESSIONS: dict[str, tuple[str, float]] = {}
REVIEW_SESSIONS_LOCK = threading.Lock()
TARGET_DB_CONFIG = {
"host": os.getenv("TARGET_DB_HOST"),
......@@ -70,6 +76,52 @@ class ReviewConflictError(Exception):
self.current_record = current_record or {}
class ReviewAuthError(Exception):
"""The caller does not have a valid review editing session."""
def _review_login(
reviewer: str,
access_code: str = "",
session_token: str = "",
) -> dict[str, object]:
reviewer = reviewer.strip()
if not reviewer:
raise ReviewAuthError("请输入昵称")
now = time.time()
with REVIEW_SESSIONS_LOCK:
expired = [token for token, (_name, expires_at) in REVIEW_SESSIONS.items() if expires_at <= now]
for token in expired:
REVIEW_SESSIONS.pop(token, None)
existing = REVIEW_SESSIONS.get(session_token) if session_token else None
if existing and existing[0] == reviewer and existing[1] > now:
REVIEW_SESSIONS[session_token] = (reviewer, now + REVIEW_SESSION_TTL_SECONDS)
return {"reviewer": reviewer, "session_token": session_token, "expires_in": REVIEW_SESSION_TTL_SECONDS}
if not REVIEW_ACCESS_CODE:
raise ReviewAuthError("服务端尚未配置 REVIEW_ACCESS_CODE，编辑功能已禁用")
if not secrets.compare_digest(access_code.encode("utf-8"), REVIEW_ACCESS_CODE.encode("utf-8")):
raise ReviewAuthError("暗号错误")
token = secrets.token_urlsafe(32)
with REVIEW_SESSIONS_LOCK:
REVIEW_SESSIONS[token] = (reviewer, now + REVIEW_SESSION_TTL_SECONDS)
return {"reviewer": reviewer, "session_token": token, "expires_in": REVIEW_SESSION_TTL_SECONDS}
def _require_review_session(payload: dict[str, object]) -> str:
reviewer = str(payload.get("reviewer") or "").strip()
session_token = str(payload.get("session_token") or "").strip()
now = time.time()
with REVIEW_SESSIONS_LOCK:
session = REVIEW_SESSIONS.get(session_token)
if not session or session[1] <= now:
REVIEW_SESSIONS.pop(session_token, None)
raise ReviewAuthError("编辑会话无效或已过期，请重新输入暗号")
if session[0] != reviewer:
raise ReviewAuthError("编辑会话与当前昵称不匹配")
REVIEW_SESSIONS[session_token] = (reviewer, now + REVIEW_SESSION_TTL_SECONDS)
return reviewer
def _conflict_response(handler: BaseHTTPRequestHandler, exc: ReviewConflictError) -> None:
_json_response(
handler,
......@@ -512,7 +564,7 @@ def _claim_staging_review(staging_id: int, reviewer: str, client_token: str) ->
review_claim_expires_at = DATE_ADD(NOW(), INTERVAL %s SECOND),
review_version = review_version + 1
WHERE staging_id = %s
AND staging_status <> 'imported'
AND staging_status NOT IN ('imported', 'deleted')
AND (
biz_review_status IN ('pending', 'unsure')
OR biz_review_status = 'not_required'
......@@ -533,6 +585,8 @@ def _claim_staging_review(staging_id: int, reviewer: str, client_token: str) ->
message = f"找不到 staging_id={staging_id}，请刷新页面后重试"
elif current.get("staging_status") == "imported":
message = "该记录已经入库，只能浏览，不能再领取审核"
elif current.get("staging_status") == "deleted":
message = "该记录已被删除，不能领取审核"
elif current.get("review_claimed_by"):
message = f"该记录已由 {current['review_claimed_by']} 领取"
else:
......@@ -988,10 +1042,15 @@ def _run_import(review_csv: Path) -> dict[str, object]:
stderr=subprocess.STDOUT,
check=False,
)
output_tail = result.stdout[-12000:]
if result.returncode != 0:
raise RuntimeError(
f"入库脚本执行失败 (exit={result.returncode}):\n{output_tail}"
)
return {
"command": " ".join(cmd),
"returncode": result.returncode,
"output": result.stdout[-12000:],
"output": output_tail,
}
......@@ -1093,22 +1152,39 @@ class Handler(BaseHTTPRequestHandler):
def do_POST(self) -> None:
parsed = urlparse(self.path)
if parsed.path == "/api/review-login":
try:
length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
result = _review_login(
str(payload.get("reviewer") or ""),
str(payload.get("access_code") or ""),
str(payload.get("session_token") or ""),
)
_json_response(self, result)
except ReviewAuthError as exc:
_error(self, str(exc), status=401)
except Exception as exc: # noqa: BLE001
_error(self, str(exc), status=400)
return
if parsed.path in {"/api/review-claim", "/api/review-heartbeat"}:
try:
length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
reviewer = _require_review_session(payload)
staging_id = int(payload.get("staging_id") or 0)
client_token = str(payload.get("client_token") or "").strip()
if not staging_id:
raise ValueError("staging_id is required")
if parsed.path == "/api/review-claim":
reviewer = str(payload.get("reviewer") or "").strip()
current = _claim_staging_review(staging_id, reviewer, client_token)
else:
current = _heartbeat_staging_review(staging_id, client_token)
_json_response(self, {"record": current})
except ReviewConflictError as exc:
_conflict_response(self, exc)
except ReviewAuthError as exc:
_error(self, str(exc), status=401)
except Exception as exc: # noqa: BLE001
_error(self, str(exc), status=400)
return
......@@ -1116,10 +1192,10 @@ class Handler(BaseHTTPRequestHandler):
try:
length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
reviewer = _require_review_session(payload)
staging_id = int(payload.get("staging_id") or 0)
decision = str(payload.get("decision", "")).strip()
note = str(payload.get("note", "")).strip()
reviewer = str(payload.get("reviewer") or "").strip()
client_token = str(payload.get("client_token") or "").strip()
expected_version = int(payload.get("expected_version", -1))
if not staging_id:
......@@ -1134,6 +1210,8 @@ class Handler(BaseHTTPRequestHandler):
_json_response(self, {"record": result})
except ReviewConflictError as exc:
_conflict_response(self, exc)
except ReviewAuthError as exc:
_error(self, str(exc), status=401)
except Exception as exc: # noqa: BLE001
_error(self, str(exc), status=400)
return
......@@ -1143,6 +1221,7 @@ class Handler(BaseHTTPRequestHandler):
try:
length = int(self.headers.get("Content-Length", "0"))
payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
_require_review_session(payload)
source_ids = payload.get("source_ids")
if source_ids is not None and not isinstance(source_ids, list):
raise ValueError("source_ids must be a list")
......@@ -1157,6 +1236,8 @@ class Handler(BaseHTTPRequestHandler):
]
review_csv = _write_review_import_csv(approved)
_json_response(self, {"review_csv": str(review_csv), **result})
except ReviewAuthError as exc:
_error(self, str(exc), status=401)
except Exception as exc: # noqa: BLE001 - local diagnostic API
_error(self, str(exc), status=400)
......
......@@ -194,3 +194,27 @@ def test_dashboard_distinguishes_reviewed_from_submitted():
assert '<option value="reviewed">已审核</option>' in html
assert '<option value="submitted">已提交</option>' in html
assert "hasFinalReview && query.staging_status !== 'imported'" in html
def test_review_access_code_issues_and_validates_edit_session():
dashboard.REVIEW_SESSIONS.clear()
with patch.object(dashboard, "REVIEW_ACCESS_CODE", "中文暗号"):
login = dashboard._review_login("alice", "中文暗号")
reviewer = dashboard._require_review_session(
{"reviewer": "alice", "session_token": login["session_token"]}
)
assert reviewer == "alice"
assert login["session_token"]
def test_wrong_access_code_and_mismatched_reviewer_are_rejected():
dashboard.REVIEW_SESSIONS.clear()
with patch.object(dashboard, "REVIEW_ACCESS_CODE", "correct"):
with pytest.raises(dashboard.ReviewAuthError, match="暗号错误"):
dashboard._review_login("alice", "wrong")
login = dashboard._review_login("alice", "correct")
with pytest.raises(dashboard.ReviewAuthError, match="昵称不匹配"):
dashboard._require_review_session(
{"reviewer": "bob", "session_token": login["session_token"]}
)
......