Skip to content

沈秋雨 / import_hk_songs

Go to a project
Toggle navigation
Toggle navigation pinning
Commit 6cce9eaa 6cce9eaa10ceeca9c2d7280f60f884e189f639f8 by 沈秋雨
Options

feat(review): 实现多人协作审核及状态同步功能 

- 新增审核人身份录入和本地缓存功能，支持切换审核人并生成客户端令牌
- 引入审核领取机制，支持防止多人同时修改同一条记录
- 实现审核状态的乐观锁版本控制，防止并发更新冲突
- 后端新增审核相关字段和索引，支持多审核人操作和审核状态管理
- 在审核列表和详情页中展示审核状态、领取状态和审核人信息
- 支持撤销审核、删除标注和入库操作，增加详细错误提示和冲突提示
- 定时同步审核状态，自动刷新审核列表和详情，保持界面数据一致
- 优化审核筛选条件，新增待审核、已审核、已提交过滤项
- 前端增加同步通知提示，展示审核冲突和同步错误信息
- 修改导出和导入逻辑，确保入库操作仅在 staging-db 环境进行
1 parent 7d33be0f
Inline Side-by-side
Showing 3 changed files with 958 additions and 193 deletions
...@@ -534,6 +534,19 @@ ...@@ -534,6 +534,19 @@
534 min-width: 220px; 534 min-width: 220px;
535 } 535 }
536 536
537 .sync-notice {
538 display: none;
539 padding: 8px 12px;
540 background: #fff4d6;
541 border-bottom: 1px solid #e8cf8a;
542 color: #725500;
543 font-size: 13px;
544 }
545
546 .sync-notice.show { display: block; }
547
548 #reviewerInput { width: 130px; }
549
537 @media (max-width: 1100px) { 550 @media (max-width: 1100px) {
538 main { grid-template-columns: 1fr; height: auto; } 551 main { grid-template-columns: 1fr; height: auto; }
539 aside { border-right: 0; border-bottom: 1px solid var(--line); max-height: 420px; } 552 aside { border-right: 0; border-bottom: 1px solid var(--line); max-height: 420px; }
...@@ -554,13 +567,14 @@ ...@@ -554,13 +567,14 @@
554 <select id="topKSelect"></select> 567 <select id="topKSelect"></select>
555 <label for="decisionFilter">类型</label> 568 <label for="decisionFilter">类型</label>
556 <select id="decisionFilter"> 569 <select id="decisionFilter">
570 <option value="pending" selected>待审核</option>
557 <option value="">全部</option> 571 <option value="">全部</option>
558 <option value="l1_hit">L1 命中</option> 572 <option value="l1_hit">L1 命中</option>
559 <option value="l2_duplicate">L2 重复</option>
560 <option value="l2_review">L2 审核</option>
561 <option value="conflict">L1/L2 冲突</option> 573 <option value="conflict">L1/L2 冲突</option>
562 <option value="new">新歌</option> 574 <option value="new">新歌</option>
563 <option value="skip">已跳过</option> 575 <option value="skip">已跳过</option>
576 <option value="reviewed">已审核</option>
577 <option value="submitted">已提交</option>
564 </select> 578 </select>
565 <input id="searchInput" type="search" placeholder="搜索歌名 / ID"> 579 <input id="searchInput" type="search" placeholder="搜索歌名 / ID">
566 <label for="pageSizeSelect">每页</label> 580 <label for="pageSizeSelect">每页</label>
...@@ -572,9 +586,12 @@ ...@@ -572,9 +586,12 @@
572 </select> 586 </select>
573 <button id="exportBtn">导出标注</button> 587 <button id="exportBtn">导出标注</button>
574 <button id="importReviewedBtn">入库审核通过</button> 588 <button id="importReviewedBtn">入库审核通过</button>
589 <label for="reviewerInput">审核人</label>
590 <input id="reviewerInput" maxlength="64" placeholder="姓名或工号">
575 <button id="reloadBtn" class="secondary">刷新</button> 591 <button id="reloadBtn" class="secondary">刷新</button>
576 </div> 592 </div>
577 </header> 593 </header>
594 <div id="syncNotice" class="sync-notice"></div>
578 595
579 <main> 596 <main>
580 <aside> 597 <aside>
...@@ -623,10 +640,16 @@ ...@@ -623,10 +640,16 @@
623 pageSize: 50, 640 pageSize: 50,
624 mode: 'retrieval', 641 mode: 'retrieval',
625 topK: '', 642 topK: '',
626 decisionFilter: '', 643 decisionFilter: 'pending',
627 queryId: '', 644 queryId: '',
628 candidateId: '', 645 candidateId: '',
629 lyricsCache: new Map() 646 lyricsCache: new Map(),
647 reviewer: '',
648 clientToken: '',
649 syncBusy: false,
650 detailGeneration: 0,
651 queryLoadGeneration: 0,
652 searchTimer: null
630 }; 653 };
631 654
632 const els = { 655 const els = {
...@@ -638,6 +661,7 @@ ...@@ -638,6 +661,7 @@
638 pageSizeSelect: document.getElementById('pageSizeSelect'), 661 pageSizeSelect: document.getElementById('pageSizeSelect'),
639 exportBtn: document.getElementById('exportBtn'), 662 exportBtn: document.getElementById('exportBtn'),
640 importReviewedBtn: document.getElementById('importReviewedBtn'), 663 importReviewedBtn: document.getElementById('importReviewedBtn'),
664 reviewerInput: document.getElementById('reviewerInput'),
641 reloadBtn: document.getElementById('reloadBtn'), 665 reloadBtn: document.getElementById('reloadBtn'),
642 prevPageBtn: document.getElementById('prevPageBtn'), 666 prevPageBtn: document.getElementById('prevPageBtn'),
643 nextPageBtn: document.getElementById('nextPageBtn'), 667 nextPageBtn: document.getElementById('nextPageBtn'),
...@@ -646,9 +670,32 @@ ...@@ -646,9 +670,32 @@
646 sampleCount: document.getElementById('sampleCount'), 670 sampleCount: document.getElementById('sampleCount'),
647 queryList: document.getElementById('queryList'), 671 queryList: document.getElementById('queryList'),
648 detail: document.getElementById('scrollDetail'), 672 detail: document.getElementById('scrollDetail'),
649 stickyInfo: document.getElementById('stickyInfo') 673 stickyInfo: document.getElementById('stickyInfo'),
674 syncNotice: document.getElementById('syncNotice')
650 }; 675 };
651 676
677 function newClientToken() {
678 if (window.crypto?.randomUUID) return window.crypto.randomUUID();
679 return `${Date.now()}-${Math.random().toString(16).slice(2)}`;
680 }
681
682 function initReviewerIdentity() {
683 state.reviewer = (localStorage.getItem('l2ReviewerName') || '').trim();
684 if (!state.reviewer) {
685 state.reviewer = (window.prompt('请输入审核人姓名或工号') || '').trim();
686 }
687 if (!state.reviewer) throw new Error('必须填写审核人姓名或工号后才能开始审核');
688 localStorage.setItem('l2ReviewerName', state.reviewer);
689 els.reviewerInput.value = state.reviewer;
690 state.clientToken = sessionStorage.getItem('l2ReviewClientToken') || newClientToken();
691 sessionStorage.setItem('l2ReviewClientToken', state.clientToken);
692 }
693
694 function showSyncNotice(message) {
695 els.syncNotice.textContent = message || '';
696 els.syncNotice.classList.toggle('show', Boolean(message));
697 }
698
652 function esc(value) { 699 function esc(value) {
653 return String(value ?? '').replace(/[&<>"']/g, ch => ({ 700 return String(value ?? '').replace(/[&<>"']/g, ch => ({
654 '&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": '&#39;' 701 '&': '&amp;', '<': '&lt;', '>': '&gt;', '"': '&quot;', "'": '&#39;'
...@@ -744,6 +791,83 @@ ...@@ -744,6 +791,83 @@
744 }); 791 });
745 } 792 }
746 793
794 async function postJSON(url, body) {
795 const res = await fetch(url, {
796 method: 'POST',
797 headers: { 'Content-Type': 'application/json' },
798 body: JSON.stringify(body)
799 });
800 const payload = await res.json().catch(() => ({}));
801 if (!res.ok) {
802 const error = new Error(payload.error || `${res.status} ${res.statusText}`);
803 error.status = res.status;
804 error.payload = payload;
805 throw error;
806 }
807 return payload;
808 }
809
810 function applyReviewSnapshot(row, record) {
811 if (!row || !record) return;
812 row.review_status = record.biz_review_status ?? row.review_status;
813 row.review_note = record.biz_review_note ?? row.review_note;
814 row.reviewed_by = record.reviewed_by ?? '';
815 row.reviewed_at = record.reviewed_at ?? '';
816 row.review_version = String(record.review_version ?? row.review_version ?? 0);
817 row.review_claimed_by = record.review_claimed_by ?? '';
818 row.review_claim_expires_at = record.review_claim_expires_at ?? '';
819 row.staging_status = record.staging_status ?? row.staging_status;
820 row.claimed_by_me = Boolean(record.claimed_by_me) || row.review_claimed_by === state.reviewer;
821 }
822
823 function applySnapshotToCurrentGroup(record) {
824 const group = state.groups.find(item => item.id === state.queryId);
825 if (!group || !record) return;
826 group.review_version = Number(record.review_version || 0);
827 group.review_claimed_by = record.review_claimed_by || '';
828 group.review_claim_expires_at = record.review_claim_expires_at || '';
829 group.claimed_by_me = group.review_claimed_by === state.reviewer;
830 group.reviewed = ['approved_import', 'rejected_duplicate', 'deleted'].includes(record.biz_review_status);
831 group.submitted = group.reviewed && record.staging_status === 'imported';
832 }
833
834 function currentQuery() {
835 return state.queryRows[0] || null;
836 }
837
838 async function claimCurrentReview({ allowReviewed = false, allowNonReview = false } = {}) {
839 const query = currentQuery();
840 if (!query || state.run?.type !== 'staging' || !query.staging_id) return true;
841 const status = query.review_status || '';
842 const isPendingReview = rowDecision(query) === 'review' && ['pending', 'unsure', ''].includes(status);
843 const isOwnReviewedRecord = allowReviewed && query.reviewed_by === state.reviewer;
844 const isNonReviewRecord = allowNonReview && status === 'not_required';
845 const reviewable = isPendingReview || isOwnReviewedRecord || isNonReviewRecord;
846 // new / merge / skip 等记录只浏览，不应触发领取请求。
847 if (!reviewable) return true;
848 try {
849 const payload = await postJSON('/api/review-claim', {
850 // bigint 必须按字符串传输，不能转成会丢精度的 JavaScript Number。
851 staging_id: String(query.staging_id),
852 reviewer: state.reviewer,
853 client_token: state.clientToken
854 });
855 for (const row of state.queryRows) applyReviewSnapshot(row, payload.record);
856 applySnapshotToCurrentGroup(payload.record);
857 showSyncNotice('');
858 return true;
859 } catch (error) {
860 if (error.status === 409) {
861 for (const row of state.queryRows) applyReviewSnapshot(row, error.payload?.current_record);
862 applySnapshotToCurrentGroup(error.payload?.current_record);
863 const owner = error.payload?.current_record?.review_claimed_by;
864 showSyncNotice(owner ? `该记录已由 ${owner} 领取；你仍可浏览，但不能覆盖对方的审核结果。` : error.message);
865 return false;
866 }
867 throw error;
868 }
869 }
870
747 function dataPath() { 871 function dataPath() {
748 if (!state.run) return ''; 872 if (!state.run) return '';
749 return state.run.retrieval; 873 return state.run.retrieval;
...@@ -757,13 +881,14 @@ ...@@ -757,13 +881,14 @@
757 const row = state.summary.find(item => String(item.top_k) === String(state.topK)) || {}; 881 const row = state.summary.find(item => String(item.top_k) === String(state.topK)) || {};
758 els.metrics.innerHTML = [ 882 els.metrics.innerHTML = [
759 metric('总数', row.total_count || '-'), 883 metric('总数', row.total_count || '-'),
760 metric('已入库 new', row.new_count || '-'), 884 metric('待审核', row.pending_review_count || '0'),
761 metric('耗时(s)', row.elapsed_seconds || '-'), 885 metric('已审核', row.reviewed_count || '0'),
762 metric('吞吐(条/s)', row.throughput_per_second || '-'), 886 metric('已提交', row.submitted_count || '0'),
763 metric('平均召回', row.avg_recalled_candidates || '-'), 887 metric('确认不重复', row.approved_count || '0'),
764 metric('命中数', row.hit_count || '0'), 888 metric('确认重复', row.rejected_count || '0'),
889 metric('已删除', row.deleted_count || '0'),
890 metric('new', row.new_count || '-'),
765 metric('merge', row.duplicate_count || '0'), 891 metric('merge', row.duplicate_count || '0'),
766 metric('review', row.review_count || '0'),
767 metric('skip', row.skip_count || '0') 892 metric('skip', row.skip_count || '0')
768 ].join(''); 893 ].join('');
769 } 894 }
...@@ -786,7 +911,12 @@ ...@@ -786,7 +911,12 @@
786 const active = item.id === state.queryId ? ' active' : ''; 911 const active = item.id === state.queryId ? ' active' : '';
787 const badges = [ 912 const badges = [
788 item.has_conflict ? '<span class="pill conflict">L1/L2 冲突</span>' : '', 913 item.has_conflict ? '<span class="pill conflict">L1/L2 冲突</span>' : '',
789 item.has_l1_hint ? '<span class="pill l1">L1 命中候选</span>' : '' 914 item.has_l1_hint ? '<span class="pill l1">L1 命中候选</span>' : '',
915 item.submitted ? '<span class="pill new">已提交</span>' : '',
916 item.reviewed && !item.submitted ? '<span class="pill">已审核</span>' : '',
917 item.claimed_by_me ? '<span class="pill new">我正在审核</span>' : '',
918 item.review_claimed_by && !item.claimed_by_me
919 ? `<span class="pill">${esc(item.review_claimed_by)} 正在审核</span>` : ''
790 ].join(' '); 920 ].join(' ');
791 return `<button class="query-item${active}${hitClass}${conflict}" data-query-id="${esc(item.id)}"> 921 return `<button class="query-item${active}${hitClass}${conflict}" data-query-id="${esc(item.id)}">
792 <div class="query-title">${esc(item.query_name || '(无歌名)')}</div> 922 <div class="query-title">${esc(item.query_name || '(无歌名)')}</div>
...@@ -799,8 +929,12 @@ ...@@ -799,8 +929,12 @@
799 btn.addEventListener('click', async () => { 929 btn.addEventListener('click', async () => {
800 state.queryId = btn.dataset.queryId; 930 state.queryId = btn.dataset.queryId;
801 state.candidateId = ''; 931 state.candidateId = '';
932 showSyncNotice('');
933 renderList();
802 await loadQueryRows(); 934 await loadQueryRows();
803 renderAll(); 935 await claimCurrentReview();
936 renderList();
937 renderDetail();
804 }); 938 });
805 } 939 }
806 els.prevPageBtn.disabled = state.page <= 1; 940 els.prevPageBtn.disabled = state.page <= 1;
...@@ -858,21 +992,19 @@ ...@@ -858,21 +992,19 @@
858 </div>`; 992 </div>`;
859 } 993 }
860 994
861 async function updateStagingReview(sourceId, decision, note) { 995 async function updateStagingReview(query, decision, note) {
862 const res = await fetch('/api/staging-review', { 996 return postJSON('/api/staging-review', {
863 method: 'POST', 997 staging_id: String(query.staging_id),
864 headers: { 'Content-Type': 'application/json' }, 998 expected_version: Number(query.review_version || 0),
865 body: JSON.stringify({ source_id: sourceId, decision, note }) 999 client_token: state.clientToken,
1000 reviewer: state.reviewer,
1001 decision,
1002 note
866 }); 1003 });
867 if (!res.ok) {
868 const err = await res.json().catch(() => ({}));
869 throw new Error(err.error || `${res.status}`);
870 }
871 return res.json();
872 } 1004 }
873 1005
874 const _DB_REVIEW_LABELS = { 1006 const _DB_REVIEW_LABELS = {
875 approved_import: '确认不重复（已入库）', 1007 approved_import: '确认不重复（待入库）',
876 rejected_duplicate: '确认重复', 1008 rejected_duplicate: '确认重复',
877 unsure: '待确认', 1009 unsure: '待确认',
878 deleted: '已删除', 1010 deleted: '已删除',
...@@ -943,6 +1075,7 @@ ...@@ -943,6 +1075,7 @@
943 } 1075 }
944 1076
945 async function renderDetail() { 1077 async function renderDetail() {
1078 const generation = ++state.detailGeneration;
946 const rows = selectedQueryRows(); 1079 const rows = selectedQueryRows();
947 if (!rows.length) { 1080 if (!rows.length) {
948 els.stickyInfo.innerHTML = ''; 1081 els.stickyInfo.innerHTML = '';
...@@ -951,9 +1084,12 @@ ...@@ -951,9 +1084,12 @@
951 } 1084 }
952 const query = rows[0]; 1085 const query = rows[0];
953 const candidate = selectedCandidate(rows); 1086 const candidate = selectedCandidate(rows);
954 const queryText = await readText(query.query_lyrics_path);
955 const candidatePath = candidate?.candidate_lyrics_path; 1087 const candidatePath = candidate?.candidate_lyrics_path;
956 const candidateText = await readText(candidatePath); 1088 const [queryText, candidateText] = await Promise.all([
1089 readText(query.query_lyrics_path),
1090 readText(candidatePath)
1091 ]);
1092 if (generation !== state.detailGeneration) return;
957 const hitDecision = candidate?.candidate_decision; 1093 const hitDecision = candidate?.candidate_decision;
958 const candidateName = candidate?.candidate_name; 1094 const candidateName = candidate?.candidate_name;
959 const candidateLyricist = candidate?.candidate_lyricist; 1095 const candidateLyricist = candidate?.candidate_lyricist;
...@@ -964,6 +1100,12 @@ ...@@ -964,6 +1100,12 @@
964 const l1 = candidate ? l1Match(candidate) : false; 1100 const l1 = candidate ? l1Match(candidate) : false;
965 const queryAudioUrl = query.audio_url || ''; 1101 const queryAudioUrl = query.audio_url || '';
966 const candidateAudioUrl = candidate?.candidate_audio_url || ''; 1102 const candidateAudioUrl = candidate?.candidate_audio_url || '';
1103 const needsManualReview = rowDecision(query) === 'review';
1104 const hasFinalReview = Boolean(query.review_status) &&
1105 !['pending', 'not_required', 'unsure'].includes(query.review_status);
1106 const canUndoReview = hasFinalReview && query.staging_status !== 'imported' &&
1107 query.reviewed_by === state.reviewer;
1108 const canDeleteRecord = query.staging_status !== 'imported';
967 1109
968 // Sticky area: 新入库歌词 + 召回候选（各含音频播放器） 1110 // Sticky area: 新入库歌词 + 召回候选（各含音频播放器）
969 els.stickyInfo.innerHTML = ` 1111 els.stickyInfo.innerHTML = `
...@@ -1007,22 +1149,24 @@ ...@@ -1007,22 +1149,24 @@
1007 <div class="section"> 1149 <div class="section">
1008 <div class="section-head"> 1150 <div class="section-head">
1009 <h2 class="section-title">人工标注</h2> 1151 <h2 class="section-title">人工标注</h2>
1010 ${query.review_status && query.review_status !== 'pending' && query.review_status !== 'not_required' 1152 ${hasFinalReview
1011 ? `<span class="pill ${query.review_status === 'approved_import' ? 'new' : query.review_status === 'rejected_duplicate' ? 'duplicate' : query.review_status === 'deleted' ? 'duplicate' : ''}">${query.review_status === 'deleted' ? '已删除' : '已审核'}</span>` 1153 ? `<span class="pill ${query.review_status === 'approved_import' ? 'new' : query.review_status === 'rejected_duplicate' ? 'duplicate' : query.review_status === 'deleted' ? 'duplicate' : ''}">${query.review_status === 'deleted' ? '已删除' : query.staging_status === 'imported' ? '已提交' : '已审核'}</span>`
1012 : ''} 1154 : ''}
1013 </div> 1155 </div>
1014 <div class="section-body"> 1156 <div class="section-body">
1015 ${query.review_status && query.review_status !== 'pending' && query.review_status !== 'not_required' ? ` 1157 ${hasFinalReview ? `
1016 <div class="review-bar" style="align-items:center"> 1158 <div class="review-bar" style="align-items:center">
1017 <span style="flex:1"> 1159 <span style="flex:1">
1018 <b>${esc(_DB_REVIEW_LABELS[query.review_status] || query.review_status)}</b> 1160 <b>${esc(query.review_status === 'approved_import' && query.staging_status === 'imported'
1161 ? '确认不重复（已提交）'
1162 : (_DB_REVIEW_LABELS[query.review_status] || query.review_status))}</b>
1019 ${query.reviewed_by ? ` · 审核人 ${esc(query.reviewed_by)}` : ''} 1163 ${query.reviewed_by ? ` · 审核人 ${esc(query.reviewed_by)}` : ''}
1020 ${query.reviewed_at ? ` · ${esc(query.reviewed_at)}` : ''} 1164 ${query.reviewed_at ? ` · ${esc(query.reviewed_at)}` : ''}
1021 ${query.review_note ? ` · 备注: ${esc(query.review_note)}` : ''} 1165 ${query.review_note ? ` · 备注: ${esc(query.review_note)}` : ''}
1022 </span> 1166 </span>
1023 <button id="undoReviewBtn" class="secondary">撤销审核</button> 1167 ${canUndoReview ? '<button id="undoReviewBtn" class="secondary">撤销审核</button>' : ''}
1024 </div> 1168 </div>
1025 ` : ` 1169 ` : needsManualReview ? `
1026 <div class="review-bar"> 1170 <div class="review-bar">
1027 ${['duplicate', 'not_duplicate', 'unsure'].map(value => ` 1171 ${['duplicate', 'not_duplicate', 'unsure'].map(value => `
1028 <button class="review-choice ${selectedReview.final_decision === value ? 'active' : ''}" data-review="${value}"> 1172 <button class="review-choice ${selectedReview.final_decision === value ? 'active' : ''}" data-review="${value}">
...@@ -1031,6 +1175,11 @@ ...@@ -1031,6 +1175,11 @@
1031 <input id="reviewNote" value="${esc(selectedReview.note || '')}" placeholder="人工备注"> 1175 <input id="reviewNote" value="${esc(selectedReview.note || '')}" placeholder="人工备注">
1032 <button id="deleteReviewBtn" class="secondary" style="color:#c0392b;border-color:#c0392b">删除</button> 1176 <button id="deleteReviewBtn" class="secondary" style="color:#c0392b;border-color:#c0392b">删除</button>
1033 </div> 1177 </div>
1178 ` : `
1179 <div class="review-bar" style="align-items:center">
1180 <span style="flex:1">该记录的自动判定为 <b>${esc(rowDecision(query) || '-')}</b>，当前仅供浏览，无需领取。</span>
1181 ${canDeleteRecord ? '<button id="deleteReviewBtn" class="secondary" style="color:#c0392b;border-color:#c0392b">删除</button>' : ''}
1182 </div>
1034 `} 1183 `}
1035 </div> 1184 </div>
1036 </div> 1185 </div>
...@@ -1069,14 +1218,24 @@ ...@@ -1069,14 +1218,24 @@
1069 setReview(candidate, { final_decision: localDecision, note }); 1218 setReview(candidate, { final_decision: localDecision, note });
1070 btn.disabled = true; 1219 btn.disabled = true;
1071 try { 1220 try {
1072 await updateStagingReview(query.query_source_id, dbDecision, note); 1221 if (!query.claimed_by_me && !await claimCurrentReview()) {
1073 // 重新加载当前样本数据以刷新 DB 状态 1222 renderDetail();
1074 await loadQueryRows(query.query_source_id); 1223 return;
1224 }
1225 const payload = await updateStagingReview(query, dbDecision, note);
1226 for (const row of state.queryRows) applyReviewSnapshot(row, payload.record);
1227 await reloadSummary();
1228 // 重新加载列表并跳转到下一条
1229 await advanceToNext();
1075 } catch (e) { 1230 } catch (e) {
1076 console.error('DB同步失败:', e); 1231 console.error('DB同步失败:', e);
1077 alert(`标注已保存到本地，但数据库同步失败: ${e.message}`); 1232 if (e.status === 409) {
1233 showSyncNotice(`提交冲突：${e.message}。备注草稿已保留，请刷新后确认。`);
1234 } else {
1235 showSyncNotice(`数据库同步失败：${e.message}。备注草稿已保留，可稍后重试。`);
1078 } 1236 }
1079 renderDetail(); 1237 renderDetail();
1238 }
1080 }); 1239 });
1081 } 1240 }
1082 const deleteBtn = document.getElementById('deleteReviewBtn'); 1241 const deleteBtn = document.getElementById('deleteReviewBtn');
...@@ -1085,13 +1244,18 @@ ...@@ -1085,13 +1244,18 @@
1085 if (!confirm('确认删除此样本？删除后将不再导入。')) return; 1244 if (!confirm('确认删除此样本？删除后将不再导入。')) return;
1086 deleteBtn.disabled = true; 1245 deleteBtn.disabled = true;
1087 try { 1246 try {
1088 await updateStagingReview(query.query_source_id, 'deleted', 1247 if (!query.claimed_by_me && !await claimCurrentReview({ allowNonReview: true })) {
1248 renderDetail();
1249 return;
1250 }
1251 await updateStagingReview(query, 'deleted',
1089 document.getElementById('reviewNote')?.value || ''); 1252 document.getElementById('reviewNote')?.value || '');
1090 await loadQueryRows(query.query_source_id); 1253 await reloadSummary();
1254 await advanceToNext();
1091 } catch (e) { 1255 } catch (e) {
1092 alert(`删除失败: ${e.message}`); 1256 alert(`删除失败: ${e.message}`);
1093 }
1094 renderDetail(); 1257 renderDetail();
1258 }
1095 }); 1259 });
1096 } 1260 }
1097 const undoBtn = document.getElementById('undoReviewBtn'); 1261 const undoBtn = document.getElementById('undoReviewBtn');
...@@ -1100,9 +1264,14 @@ ...@@ -1100,9 +1264,14 @@
1100 if (!confirm('确认撤销此样本的人工审核结果？')) return; 1264 if (!confirm('确认撤销此样本的人工审核结果？')) return;
1101 undoBtn.disabled = true; 1265 undoBtn.disabled = true;
1102 try { 1266 try {
1103 await updateStagingReview(query.query_source_id, 'pending', ''); 1267 if (!query.claimed_by_me && !await claimCurrentReview({ allowReviewed: true })) {
1268 renderDetail();
1269 return;
1270 }
1271 await updateStagingReview(query, 'pending', '');
1104 if (candidate) setReview(candidate, { final_decision: undefined, note: '' }); 1272 if (candidate) setReview(candidate, { final_decision: undefined, note: '' });
1105 await loadQueryRows(query.query_source_id); 1273 await loadQueryRows();
1274 await reloadSummary();
1106 } catch (e) { 1275 } catch (e) {
1107 alert(`撤销失败: ${e.message}`); 1276 alert(`撤销失败: ${e.message}`);
1108 } 1277 }
...@@ -1122,6 +1291,12 @@ ...@@ -1122,6 +1291,12 @@
1122 const rows = []; 1291 const rows = [];
1123 for (const row of state.queryRows) { 1292 for (const row of state.queryRows) {
1124 const review = reviews[reviewKey(row)] || {}; 1293 const review = reviews[reviewKey(row)] || {};
1294 const databaseDecision = {
1295 approved_import: 'not_duplicate',
1296 rejected_duplicate: 'duplicate',
1297 unsure: 'unsure',
1298 deleted: 'deleted'
1299 }[row.review_status] || '';
1125 rows.push({ 1300 rows.push({
1126 run_id: state.run?.id || '', 1301 run_id: state.run?.id || '',
1127 top_k: state.topK, 1302 top_k: state.topK,
...@@ -1136,9 +1311,9 @@ ...@@ -1136,9 +1311,9 @@
1136 l2_decision: rowDecision(row), 1311 l2_decision: rowDecision(row),
1137 l1_metadata_match: l1Match(row) ? '1' : '0', 1312 l1_metadata_match: l1Match(row) ? '1' : '0',
1138 l1_l2_conflict: isConflict(row) ? '1' : '0', 1313 l1_l2_conflict: isConflict(row) ? '1' : '0',
1139 final_decision: review.final_decision || '', 1314 final_decision: state.run?.type === 'staging' ? databaseDecision : (review.final_decision || ''),
1140 note: review.note || '', 1315 note: state.run?.type === 'staging' ? (row.review_note || '') : (review.note || ''),
1141 updated_at: review.updated_at || '' 1316 updated_at: state.run?.type === 'staging' ? (row.reviewed_at || '') : (review.updated_at || '')
1142 }); 1317 });
1143 } 1318 }
1144 const fields = Object.keys(rows[0] || { 1319 const fields = Object.keys(rows[0] || {
...@@ -1161,41 +1336,19 @@ ...@@ -1161,41 +1336,19 @@
1161 1336
1162 async function importReviewed() { 1337 async function importReviewed() {
1163 if (!state.run) return; 1338 if (!state.run) return;
1164 const reviews = loadReviews(); 1339 if (state.run.type !== 'staging') {
1165 const prefix = `${state.run.id}::${state.topK || ''}::`; 1340 alert('历史报表仅供查看，请在 staging-db 中执行入库。');
1166 const rows = [];
1167 for (const [key, review] of Object.entries(reviews)) {
1168 if (!key.startsWith(prefix)) continue;
1169 if (review.final_decision !== 'not_duplicate') continue;
1170 const parts = key.split('::');
1171 const sourceId = parts[2] || '';
1172 if (!sourceId) continue;
1173 rows.push({
1174 source_id: sourceId,
1175 review_decision: 'import',
1176 review_note: review.note || ''
1177 });
1178 }
1179 if (!rows.length) {
1180 // 诊断信息：显示当前结果下实际找到的标注情况
1181 const allForRun = Object.entries(reviews).filter(([k]) => k.startsWith(prefix));
1182 const decisionSummary = allForRun.map(([, v]) => v.final_decision || '未标注').join(', ') || '无';
1183 alert(`当前结果中没有标注为“确认不重复”的 review 样本。\n\n` +
1184 `当前结果下找到 ${allForRun.length} 条标注，状态: ${decisionSummary}`);
1185 return; 1341 return;
1186 } 1342 }
1187 if (!confirm(`确认入库 ${rows.length} 条审核通过样本？`)) return; 1343 if (!confirm('确认入库数据库中所有“确认不重复（待入库）”样本？')) return;
1188 els.importReviewedBtn.disabled = true; 1344 els.importReviewedBtn.disabled = true;
1189 els.importReviewedBtn.textContent = '入库中...'; 1345 els.importReviewedBtn.textContent = '入库中...';
1190 try { 1346 try {
1191 const res = await fetch('/api/import-reviewed', { 1347 const payload = await postJSON('/api/import-reviewed', { reviewer: state.reviewer });
1192 method: 'POST', 1348 alert(`入库完成：${payload.inserted_count} 条。结果文件：${payload.review_csv}`);
1193 headers: { 'Content-Type': 'application/json' }, 1349 await loadGroups();
1194 body: JSON.stringify({ run_id: state.run.id, rows }) 1350 await reloadSummary();
1195 }); 1351 renderAll();
1196 const payload = await res.json();
1197 if (!res.ok) throw new Error(payload.error || payload.output || `${res.status} ${res.statusText}`);
1198 alert(`入库完成。结果文件：${payload.review_csv}`);
1199 } catch (err) { 1352 } catch (err) {
1200 alert(`入库失败：${err.message}`); 1353 alert(`入库失败：${err.message}`);
1201 } finally { 1354 } finally {
...@@ -1221,7 +1374,8 @@ ...@@ -1221,7 +1374,8 @@
1221 decision: state.decisionFilter, 1374 decision: state.decisionFilter,
1222 q: els.searchInput.value.trim(), 1375 q: els.searchInput.value.trim(),
1223 page: String(state.page), 1376 page: String(state.page),
1224 page_size: String(state.pageSize) 1377 page_size: String(state.pageSize),
1378 client_token: state.clientToken
1225 }); 1379 });
1226 const data = await getJSON(`/api/groups?${params.toString()}`); 1380 const data = await getJSON(`/api/groups?${params.toString()}`);
1227 state.groups = data.groups || []; 1381 state.groups = data.groups || [];
...@@ -1232,20 +1386,103 @@ ...@@ -1232,20 +1386,103 @@
1232 state.candidateId = ''; 1386 state.candidateId = '';
1233 } 1387 }
1234 await loadQueryRows(); 1388 await loadQueryRows();
1389 // 仅 pending/unsure 的 review 记录会真正发起领取；其他分类只浏览。
1390 await claimCurrentReview();
1235 } 1391 }
1236 1392
1237 async function loadQueryRows() { 1393 async function loadQueryRows() {
1238 state.queryRows = []; 1394 const generation = ++state.queryLoadGeneration;
1395 const requestedQueryId = state.queryId;
1239 if (!state.queryId || !state.run || !state.topK) return; 1396 if (!state.queryId || !state.run || !state.topK) return;
1240 const params = new URLSearchParams({ 1397 const params = new URLSearchParams({
1241 path: dataPath(), 1398 path: dataPath(),
1242 top_k: state.topK, 1399 top_k: state.topK,
1243 query_id: state.queryId 1400 query_id: state.queryId
1244 }); 1401 });
1402 state.queryRows = [];
1245 const data = await getJSON(`/api/query?${params.toString()}`); 1403 const data = await getJSON(`/api/query?${params.toString()}`);
1404 if (generation !== state.queryLoadGeneration || requestedQueryId !== state.queryId) return;
1246 state.queryRows = data.rows || []; 1405 state.queryRows = data.rows || [];
1247 } 1406 }
1248 1407
1408 async function reloadSummary() {
1409 if (!state.run) return;
1410 const summary = await getJSON(`/api/csv?path=${encodeURIComponent(state.run.summary)}`);
1411 const nextSummary = summary.rows || [];
1412 if (JSON.stringify(nextSummary) !== JSON.stringify(state.summary)) {
1413 state.summary = nextSummary;
1414 renderMetrics();
1415 }
1416 }
1417
1418 async function advanceToNext() {
1419 // 重新加载列表（应用当前筛选，已审核的会被过滤掉）
1420 await loadGroups();
1421 renderAll();
1422 }
1423
1424 async function syncVisibleReviewStatuses() {
1425 if (state.syncBusy || state.run?.type !== 'staging' || !state.groups.length) return;
1426 const ids = state.groups.map(group => group.staging_id).filter(Boolean);
1427 if (!ids.length) return;
1428 state.syncBusy = true;
1429 try {
1430 const params = new URLSearchParams({
1431 staging_ids: ids.join(','),
1432 client_token: state.clientToken
1433 });
1434 const payload = await getJSON(`/api/review-statuses?${params.toString()}`);
1435 const statuses = new Map((payload.rows || []).map(row => [String(row.staging_id), row]));
1436 let listChanged = false;
1437 for (const group of state.groups) {
1438 const remote = statuses.get(String(group.staging_id));
1439 if (!remote) continue;
1440 if (Number(remote.review_version || 0) !== Number(group.review_version || 0) ||
1441 Boolean(remote.claimed_by_me) !== Boolean(group.claimed_by_me)) {
1442 listChanged = true;
1443 }
1444 group.review_version = Number(remote.review_version || 0);
1445 group.review_claimed_by = remote.review_claimed_by || '';
1446 group.review_claim_expires_at = remote.review_claim_expires_at || '';
1447 group.claimed_by_me = Boolean(remote.claimed_by_me);
1448 group.reviewed = ['approved_import', 'rejected_duplicate', 'deleted'].includes(remote.biz_review_status);
1449 group.submitted = group.reviewed && remote.staging_status === 'imported';
1450 }
1451 const query = currentQuery();
1452 const selectedRemote = query ? statuses.get(String(query.staging_id)) : null;
1453 if (selectedRemote && Number(selectedRemote.review_version || 0) !== Number(query.review_version || 0)) {
1454 if (!selectedRemote.claimed_by_me) {
1455 showSyncNotice(`该记录已由 ${selectedRemote.reviewed_by || selectedRemote.review_claimed_by || '其他审核人'} 更新；当前输入未被覆盖，请刷新确认。`);
1456 }
1457 }
1458 if (listChanged) renderList();
1459 } catch (error) {
1460 console.error('同步审核状态失败:', error);
1461 } finally {
1462 state.syncBusy = false;
1463 }
1464 }
1465
1466 async function heartbeatCurrentReview() {
1467 const query = currentQuery();
1468 if (state.run?.type !== 'staging' || !query?.staging_id || !query.review_claimed_by) return;
1469 const heartbeatStagingId = String(query.staging_id);
1470 try {
1471 const payload = await postJSON('/api/review-heartbeat', {
1472 staging_id: String(query.staging_id),
1473 client_token: state.clientToken
1474 });
1475 if (String(currentQuery()?.staging_id || '') === heartbeatStagingId) {
1476 for (const row of state.queryRows) applyReviewSnapshot(row, payload.record);
1477 }
1478 } catch (error) {
1479 if (error.status === 409 && String(currentQuery()?.staging_id || '') === heartbeatStagingId) {
1480 showSyncNotice(error.message);
1481 }
1482 else console.error('审核领取续期失败:', error);
1483 }
1484 }
1485
1249 async function loadRun(runId) { 1486 async function loadRun(runId) {
1250 state.run = state.runs.find(run => run.id === runId) || state.runs[0]; 1487 state.run = state.runs.find(run => run.id === runId) || state.runs[0];
1251 if (!state.run) return; 1488 if (!state.run) return;
...@@ -1280,6 +1517,24 @@ ...@@ -1280,6 +1517,24 @@
1280 } 1517 }
1281 1518
1282 els.runSelect.addEventListener('change', () => loadRun(els.runSelect.value)); 1519 els.runSelect.addEventListener('change', () => loadRun(els.runSelect.value));
1520 els.reviewerInput.addEventListener('change', async () => {
1521 const nextReviewer = els.reviewerInput.value.trim();
1522 if (!nextReviewer) {
1523 els.reviewerInput.value = state.reviewer;
1524 return;
1525 }
1526 if (currentQuery()?.review_claimed_by === state.reviewer) {
1527 alert('请先完成当前审核，或等待领取过期后再切换审核人。');
1528 els.reviewerInput.value = state.reviewer;
1529 return;
1530 }
1531 state.reviewer = nextReviewer;
1532 state.clientToken = newClientToken();
1533 localStorage.setItem('l2ReviewerName', state.reviewer);
1534 sessionStorage.setItem('l2ReviewClientToken', state.clientToken);
1535 await loadGroups();
1536 renderAll();
1537 });
1283 els.topKSelect.addEventListener('change', async () => { 1538 els.topKSelect.addEventListener('change', async () => {
1284 state.topK = els.topKSelect.value; 1539 state.topK = els.topKSelect.value;
1285 state.page = 1; 1540 state.page = 1;
...@@ -1288,12 +1543,15 @@ ...@@ -1288,12 +1543,15 @@
1288 await loadGroups(); 1543 await loadGroups();
1289 renderAll(); 1544 renderAll();
1290 }); 1545 });
1291 els.searchInput.addEventListener('input', async () => { 1546 els.searchInput.addEventListener('input', () => {
1547 window.clearTimeout(state.searchTimer);
1548 state.searchTimer = window.setTimeout(async () => {
1292 state.page = 1; 1549 state.page = 1;
1293 state.queryId = ''; 1550 state.queryId = '';
1294 state.candidateId = ''; 1551 state.candidateId = '';
1295 await loadGroups(); 1552 await loadGroups();
1296 renderAll(); 1553 renderAll();
1554 }, 350);
1297 }); 1555 });
1298 els.decisionFilter.addEventListener('change', async () => { 1556 els.decisionFilter.addEventListener('change', async () => {
1299 state.decisionFilter = els.decisionFilter.value; 1557 state.decisionFilter = els.decisionFilter.value;
...@@ -1331,6 +1589,21 @@ ...@@ -1331,6 +1589,21 @@
1331 renderAll(); 1589 renderAll();
1332 }); 1590 });
1333 1591
1592 try {
1593 initReviewerIdentity();
1594 } catch (err) {
1595 els.detail.innerHTML = `<div class="empty">${esc(err.message)}</div>`;
1596 throw err;
1597 }
1598 window.setInterval(() => {
1599 if (!document.hidden) syncVisibleReviewStatuses();
1600 }, 15000);
1601 window.setInterval(() => {
1602 if (!document.hidden) reloadSummary().catch(console.error);
1603 }, 60000);
1604 window.setInterval(() => {
1605 if (!document.hidden) heartbeatCurrentReview();
1606 }, 60000);
1334 1607
1335 loadRuns().catch(err => { 1608 loadRuns().catch(err => {
1336 els.detail.innerHTML = `<div class="empty">${esc(err.message)}</div>`; 1609 els.detail.innerHTML = `<div class="empty">${esc(err.message)}</div>`;
......
...@@ -26,6 +26,7 @@ DASHBOARD = ROOT / "l2_review_dashboard.html" ...@@ -26,6 +26,7 @@ DASHBOARD = ROOT / "l2_review_dashboard.html"
26 REPORT_DIR.mkdir(parents=True, exist_ok=True) 26 REPORT_DIR.mkdir(parents=True, exist_ok=True)
27 GROUP_INDEX_CACHE: dict[tuple[str, int, int, str], list[dict[str, object]]] = {} 27 GROUP_INDEX_CACHE: dict[tuple[str, int, int, str], list[dict[str, object]]] = {}
28 load_dotenv(ROOT / ".env") 28 load_dotenv(ROOT / ".env")
29 REVIEW_CLAIM_TTL_SECONDS = max(60, int(os.getenv("REVIEW_CLAIM_TTL_SECONDS", "600")))
29 30
30 TARGET_DB_CONFIG = { 31 TARGET_DB_CONFIG = {
31 "host": os.getenv("TARGET_DB_HOST"), 32 "host": os.getenv("TARGET_DB_HOST"),
...@@ -52,6 +53,33 @@ TARGET_COLUMNS = [ ...@@ -52,6 +53,33 @@ TARGET_COLUMNS = [
52 "commit_desc", "price", "source_table_name", "source_song_id", 53 "commit_desc", "price", "source_table_name", "source_song_id",
53 "lyric_archive_element_id", "melody_archive_element_id", "audio_fingerprint", 54 "lyric_archive_element_id", "melody_archive_element_id", "audio_fingerprint",
54 ] 55 ]
56 REVIEW_SCHEMA_COLUMNS = {
57 "review_claimed_by": "varchar(64) DEFAULT NULL COMMENT '当前领取审核人'",
58 "review_claim_token": "varchar(64) DEFAULT NULL COMMENT '领取会话凭证'",
59 "review_claimed_at": "datetime DEFAULT NULL COMMENT '领取时间'",
60 "review_claim_expires_at": "datetime DEFAULT NULL COMMENT '领取过期时间'",
61 "review_version": "bigint(20) NOT NULL DEFAULT '0' COMMENT '审核乐观锁版本'",
62 }
63
64
65 class ReviewConflictError(Exception):
66 """The staging row changed or is owned by another review session."""
67
68 def __init__(self, message: str, current_record: dict[str, object] | None = None) -> None:
69 super().__init__(message)
70 self.current_record = current_record or {}
71
72
73 def _conflict_response(handler: BaseHTTPRequestHandler, exc: ReviewConflictError) -> None:
74 _json_response(
75 handler,
76 {
77 "error": str(exc),
78 "code": "review_conflict",
79 "current_record": exc.current_record,
80 },
81 status=409,
82 )
55 83
56 84
57 def _json_response(handler: BaseHTTPRequestHandler, payload: object, status: int = 200) -> None: 85 def _json_response(handler: BaseHTTPRequestHandler, payload: object, status: int = 200) -> None:
...@@ -59,6 +87,7 @@ def _json_response(handler: BaseHTTPRequestHandler, payload: object, status: int ...@@ -59,6 +87,7 @@ def _json_response(handler: BaseHTTPRequestHandler, payload: object, status: int
59 try: 87 try:
60 handler.send_response(status) 88 handler.send_response(status)
61 handler.send_header("Content-Type", "application/json; charset=utf-8") 89 handler.send_header("Content-Type", "application/json; charset=utf-8")
90 handler.send_header("Cache-Control", "no-store")
62 handler.send_header("Content-Length", str(len(body))) 91 handler.send_header("Content-Length", str(len(body)))
63 handler.end_headers() 92 handler.end_headers()
64 handler.wfile.write(body) 93 handler.wfile.write(body)
...@@ -97,6 +126,38 @@ def _target_conn(): ...@@ -97,6 +126,38 @@ def _target_conn():
97 return pymysql.connect(**TARGET_DB_CONFIG) 126 return pymysql.connect(**TARGET_DB_CONFIG)
98 127
99 128
129 def _ensure_review_schema(apply_migration: bool = False) -> None:
130 """Validate or add the columns required by collaborative reviewing."""
131 with _target_conn() as conn, conn.cursor() as cursor:
132 cursor.execute(f"SHOW COLUMNS FROM {TARGET_TABLE_NAME_TMP}")
133 existing_columns = {str(row["Field"]) for row in cursor.fetchall()}
134 missing = [name for name in REVIEW_SCHEMA_COLUMNS if name not in existing_columns]
135 if missing and not apply_migration:
136 names = ", ".join(missing)
137 raise RuntimeError(
138 f"暂存表缺少多人审核字段: {names}。"
139 "请先使用 --migrate-review-schema 启动一次。"
140 )
141 for name in missing:
142 cursor.execute(
143 f"ALTER TABLE {TARGET_TABLE_NAME_TMP} "
144 f"ADD COLUMN `{name}` {REVIEW_SCHEMA_COLUMNS[name]}"
145 )
146 cursor.execute(f"SHOW INDEX FROM {TARGET_TABLE_NAME_TMP} WHERE Key_name = %s", ("idx_staging_review_claim",))
147 if not cursor.fetchone():
148 if not apply_migration:
149 raise RuntimeError(
150 "暂存表缺少索引 idx_staging_review_claim。"
151 "请先使用 --migrate-review-schema 启动一次。"
152 )
153 cursor.execute(
154 f"ALTER TABLE {TARGET_TABLE_NAME_TMP} "
155 "ADD KEY `idx_staging_review_claim` (`biz_review_status`, `review_claim_expires_at`)"
156 )
157 if apply_migration:
158 conn.commit()
159
160
100 def _staging_summary() -> list[dict[str, str]]: 161 def _staging_summary() -> list[dict[str, str]]:
101 sql = f""" 162 sql = f"""
102 SELECT 163 SELECT
...@@ -104,9 +165,15 @@ def _staging_summary() -> list[dict[str, str]]: ...@@ -104,9 +165,15 @@ def _staging_summary() -> list[dict[str, str]]:
104 SUM(CASE WHEN dedup_action = 'new' THEN 1 ELSE 0 END) AS new_count, 165 SUM(CASE WHEN dedup_action = 'new' THEN 1 ELSE 0 END) AS new_count,
105 SUM(CASE WHEN dedup_action = 'merge' THEN 1 ELSE 0 END) AS merge_count, 166 SUM(CASE WHEN dedup_action = 'merge' THEN 1 ELSE 0 END) AS merge_count,
106 SUM(CASE WHEN dedup_action = 'review' THEN 1 ELSE 0 END) AS review_count, 167 SUM(CASE WHEN dedup_action = 'review' THEN 1 ELSE 0 END) AS review_count,
107 SUM(CASE WHEN dedup_action = 'skip' THEN 1 ELSE 0 END) AS skip_count 168 SUM(CASE WHEN dedup_action = 'skip' THEN 1 ELSE 0 END) AS skip_count,
169 SUM(CASE WHEN dedup_action = 'review' AND (biz_review_status IS NULL OR biz_review_status IN ('pending', 'unsure')) THEN 1 ELSE 0 END) AS pending_review_count,
170 SUM(CASE WHEN biz_review_status = 'approved_import' THEN 1 ELSE 0 END) AS approved_count,
171 SUM(CASE WHEN biz_review_status = 'rejected_duplicate' THEN 1 ELSE 0 END) AS rejected_count,
172 SUM(CASE WHEN biz_review_status = 'deleted' THEN 1 ELSE 0 END) AS deleted_count,
173 SUM(CASE WHEN biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted') AND staging_status <> 'imported' THEN 1 ELSE 0 END) AS reviewed_count,
174 SUM(CASE WHEN biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted') AND staging_status = 'imported' THEN 1 ELSE 0 END) AS submitted_count
108 FROM ( 175 FROM (
109 SELECT source_song_id, dedup_action 176 SELECT source_song_id, dedup_action, biz_review_status, staging_status
110 FROM {TARGET_TABLE_NAME_TMP} 177 FROM {TARGET_TABLE_NAME_TMP}
111 WHERE (source_song_id, staging_id) IN ( 178 WHERE (source_song_id, staging_id) IN (
112 SELECT source_song_id, MAX(staging_id) 179 SELECT source_song_id, MAX(staging_id)
...@@ -129,6 +196,12 @@ def _staging_summary() -> list[dict[str, str]]: ...@@ -129,6 +196,12 @@ def _staging_summary() -> list[dict[str, str]]:
129 "skip_count": str(row.get("skip_count") or 0), 196 "skip_count": str(row.get("skip_count") or 0),
130 "new_count": str(row.get("new_count") or 0), 197 "new_count": str(row.get("new_count") or 0),
131 "total_count": str(row.get("total_count") or 0), 198 "total_count": str(row.get("total_count") or 0),
199 "approved_count": str(row.get("approved_count") or 0),
200 "rejected_count": str(row.get("rejected_count") or 0),
201 "deleted_count": str(row.get("deleted_count") or 0),
202 "reviewed_count": str(row.get("reviewed_count") or 0),
203 "submitted_count": str(row.get("submitted_count") or 0),
204 "pending_review_count": str(row.get("pending_review_count") or 0),
132 }] 205 }]
133 206
134 207
...@@ -161,52 +234,83 @@ def _staging_dashboard_row(row: dict[str, object]) -> dict[str, str]: ...@@ -161,52 +234,83 @@ def _staging_dashboard_row(row: dict[str, object]) -> dict[str, str]:
161 "review_note": str(row.get("biz_review_note") or ""), 234 "review_note": str(row.get("biz_review_note") or ""),
162 "reviewed_by": str(row.get("reviewed_by") or ""), 235 "reviewed_by": str(row.get("reviewed_by") or ""),
163 "reviewed_at": str(row.get("reviewed_at") or ""), 236 "reviewed_at": str(row.get("reviewed_at") or ""),
237 "review_version": str(row.get("review_version") or 0),
238 "review_claimed_by": str(row.get("review_claimed_by") or ""),
239 "review_claim_expires_at": str(row.get("review_claim_expires_at") or ""),
240 "staging_status": str(row.get("staging_status") or ""),
164 "l1_metadata_match": "1" if row.get("l1_matched_id") else "0", 241 "l1_metadata_match": "1" if row.get("l1_matched_id") else "0",
165 "l1_l2_conflict": "0", 242 "l1_l2_conflict": "0",
166 "audio_url": str(row.get("audio_url") or ""), 243 "audio_url": str(row.get("audio_url") or ""),
167 } 244 }
168 245
169 246
170 def _staging_groups_response(mode: str, term: str, page: int, page_size: int, decision: str = '') -> dict[str, object]: 247 def _staging_groups_response(
248 mode: str,
249 term: str,
250 page: int,
251 page_size: int,
252 decision: str = "",
253 client_token: str = "",
254 ) -> dict[str, object]:
171 where_clauses = [] 255 where_clauses = []
172 params: list[object] = [] 256 params: list[object] = []
173 if term: 257 if term:
174 where_clauses.append("(CAST(source_song_id AS CHAR) LIKE %s OR name LIKE %s OR lyricist LIKE %s OR composer LIKE %s)") 258 where_clauses.append("(CAST(s.source_song_id AS CHAR) LIKE %s OR s.name LIKE %s OR s.lyricist LIKE %s OR s.composer LIKE %s)")
175 like = f"%{term}%" 259 like = f"%{term}%"
176 params.extend([like, like, like, like]) 260 params.extend([like, like, like, like])
177 # hits 模式下只展示有命中的记录（merge/review） 261 # hits 模式下只展示有命中的记录（merge/review）
178 if mode == "hits": 262 if mode == "hits":
179 where_clauses.append("dedup_action IN ('merge', 'review')") 263 where_clauses.append("s.dedup_action IN ('merge', 'review')")
180 # 决策筛选：l1_hit / l2_hit / conflict / new / skip 264 # 决策筛选：l1_hit / l2_hit / conflict / new / skip
181 if decision == 'l1_hit': 265 if decision == 'l1_hit':
182 where_clauses.append("l1_matched_id IS NOT NULL") 266 where_clauses.append("s.l1_matched_id IS NOT NULL")
183 elif decision == 'l2_duplicate': 267 elif decision == 'l2_duplicate':
184 where_clauses.append("dedup_action = 'merge'") 268 where_clauses.append("s.dedup_action = 'merge'")
185 elif decision == 'l2_review': 269 elif decision == 'l2_review':
186 where_clauses.append("dedup_action = 'review'") 270 where_clauses.append("s.dedup_action = 'review'")
187 elif decision == 'conflict': 271 elif decision == 'conflict':
188 where_clauses.append("(l1_matched_id IS NOT NULL AND dedup_action = 'new') OR (l1_matched_id IS NULL AND dedup_action IN ('merge', 'review'))") 272 where_clauses.append("(s.l1_matched_id IS NOT NULL AND s.dedup_action = 'new') OR (s.l1_matched_id IS NULL AND s.dedup_action IN ('merge', 'review'))")
189 elif decision == 'new': 273 elif decision == 'new':
190 where_clauses.append("dedup_action = 'new' AND l1_matched_id IS NULL") 274 where_clauses.append("s.dedup_action = 'new' AND s.l1_matched_id IS NULL")
191 elif decision == 'skip': 275 elif decision == 'skip':
192 where_clauses.append("dedup_action = 'skip'") 276 where_clauses.append("s.dedup_action = 'skip'")
277 elif decision == 'reviewed':
278 where_clauses.append("s.biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted')")
279 where_clauses.append("s.staging_status <> 'imported'")
280 elif decision == 'submitted':
281 where_clauses.append("s.biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted')")
282 where_clauses.append("s.staging_status = 'imported'")
283 elif decision == 'pending':
284 where_clauses.append("(s.biz_review_status IS NULL OR s.biz_review_status IN ('pending', 'unsure'))")
285 where_clauses.append("s.dedup_action = 'review'")
286 where_clauses.append("s.staging_status <> 'imported'")
287 where_clauses.append(
288 "(s.review_claim_expires_at IS NULL OR s.review_claim_expires_at <= NOW() OR s.review_claim_token = %s)"
289 )
290 params.append(client_token)
193 # 兼容旧的 dedup_action 筛选 291 # 兼容旧的 dedup_action 筛选
194 elif decision and decision in ('merge', 'review'): 292 elif decision and decision in ('merge', 'review'):
195 where_clauses.append("dedup_action = %s") 293 where_clauses.append("s.dedup_action = %s")
196 params.append(decision) 294 params.append(decision)
197 where = "WHERE " + " AND ".join(where_clauses) if where_clauses else "" 295 where = "WHERE " + " AND ".join(where_clauses) if where_clauses else ""
198 # 同一首歌可能因多次导入批次产生多行，按 source_song_id 去重取最新一行 296 # 同一首歌可能因多次导入批次产生多行，按 source_song_id 去重取最新一行
199 count_sql = f"SELECT COUNT(DISTINCT source_song_id) AS n FROM {TARGET_TABLE_NAME_TMP} {where}" 297 latest_join = f"""
200 sql = f"""
201 SELECT s.staging_id, s.source_song_id, s.name, s.lyricist, s.composer, s.dedup_action,
202 s.biz_review_status, s.l1_matched_id
203 FROM {TARGET_TABLE_NAME_TMP} s 298 FROM {TARGET_TABLE_NAME_TMP} s
204 INNER JOIN ( 299 INNER JOIN (
205 SELECT source_song_id, MAX(staging_id) AS max_staging_id 300 SELECT source_song_id, MAX(staging_id) AS max_staging_id
206 FROM {TARGET_TABLE_NAME_TMP} 301 FROM {TARGET_TABLE_NAME_TMP}
207 {where}
208 GROUP BY source_song_id 302 GROUP BY source_song_id
209 ) latest ON s.staging_id = latest.max_staging_id 303 ) latest ON s.staging_id = latest.max_staging_id
304 """
305 count_sql = f"SELECT COUNT(*) AS n {latest_join} {where}"
306 sql = f"""
307 SELECT s.staging_id, s.source_song_id, s.name, s.lyricist, s.composer, s.dedup_action,
308 s.biz_review_status, s.staging_status, s.l1_matched_id, s.review_version,
309 CASE WHEN s.review_claim_expires_at > NOW() THEN s.review_claimed_by ELSE NULL END AS review_claimed_by,
310 s.review_claim_expires_at,
311 CASE WHEN s.review_claim_token = %s AND s.review_claim_expires_at > NOW() THEN 1 ELSE 0 END AS claimed_by_me
312 {latest_join}
313 {where}
210 ORDER BY 314 ORDER BY
211 CASE s.dedup_action WHEN 'review' THEN 0 WHEN 'merge' THEN 1 ELSE 2 END, 315 CASE s.dedup_action WHEN 'review' THEN 0 WHEN 'merge' THEN 1 ELSE 2 END,
212 s.staging_create_time DESC 316 s.staging_create_time DESC
...@@ -215,7 +319,7 @@ def _staging_groups_response(mode: str, term: str, page: int, page_size: int, de ...@@ -215,7 +319,7 @@ def _staging_groups_response(mode: str, term: str, page: int, page_size: int, de
215 with _target_conn() as conn, conn.cursor() as cursor: 319 with _target_conn() as conn, conn.cursor() as cursor:
216 cursor.execute(count_sql, params) 320 cursor.execute(count_sql, params)
217 total = (cursor.fetchone() or {}).get("n", 0) 321 total = (cursor.fetchone() or {}).get("n", 0)
218 cursor.execute(sql, [*params, page_size, max(0, (page - 1) * page_size)]) 322 cursor.execute(sql, [client_token, *params, page_size, max(0, (page - 1) * page_size)])
219 raw_rows = cursor.fetchall() 323 raw_rows = cursor.fetchall()
220 # Python 层面再去重，防止 SQL 层面因数据库版本或数据异常未能完全去重 324 # Python 层面再去重，防止 SQL 层面因数据库版本或数据异常未能完全去重
221 seen_source_ids: set[str] = set() 325 seen_source_ids: set[str] = set()
...@@ -237,6 +341,16 @@ def _staging_groups_response(mode: str, term: str, page: int, page_size: int, de ...@@ -237,6 +341,16 @@ def _staging_groups_response(mode: str, term: str, page: int, page_size: int, de
237 "has_conflict": False, 341 "has_conflict": False,
238 "has_l1_hint": bool(row.get("l1_matched_id")), 342 "has_l1_hint": bool(row.get("l1_matched_id")),
239 "decision": row.get("dedup_action") or "", 343 "decision": row.get("dedup_action") or "",
344 "reviewed": row.get("biz_review_status") in {"approved_import", "rejected_duplicate", "deleted"},
345 "submitted": (
346 row.get("biz_review_status") in {"approved_import", "rejected_duplicate", "deleted"}
347 and row.get("staging_status") == "imported"
348 ),
349 "staging_id": str(row.get("staging_id") or ""),
350 "review_version": int(row.get("review_version") or 0),
351 "review_claimed_by": str(row.get("review_claimed_by") or ""),
352 "review_claim_expires_at": str(row.get("review_claim_expires_at") or ""),
353 "claimed_by_me": bool(row.get("claimed_by_me")),
240 } 354 }
241 for row in rows 355 for row in rows
242 ] 356 ]
...@@ -371,107 +485,241 @@ def _staging_query_rows(query_id: str) -> dict[str, object]: ...@@ -371,107 +485,241 @@ def _staging_query_rows(query_id: str) -> dict[str, object]:
371 return {"rows": rows} 485 return {"rows": rows}
372 486
373 487
374 def _update_staging_review(source_song_id: str, decision: str, note: str, reviewer: str = "dashboard") -> dict[str, object]: 488 def _review_snapshot(cursor, staging_id: int) -> dict[str, object]:
375 """更新暂存表的人工审核状态。 489 cursor.execute(
490 f"""
491 SELECT staging_id, source_song_id, dedup_action, biz_review_status, biz_review_note,
492 reviewed_by, reviewed_at, review_version, review_claimed_by,
493 review_claim_expires_at, staging_status
494 FROM {TARGET_TABLE_NAME_TMP}
495 WHERE staging_id = %s
496 """,
497 (staging_id,),
498 )
499 return cursor.fetchone() or {}
500
376 501
377 decision: 'approved_import' | 'rejected_duplicate' | 'unsure' | 'pending' (撤销) | 'deleted' 502 def _claim_staging_review(staging_id: int, reviewer: str, client_token: str) -> dict[str, object]:
378 """ 503 if not reviewer or not client_token:
379 if decision == "pending": 504 raise ValueError("reviewer and client_token are required")
380 # 撤销审核 505 with _target_conn() as conn, conn.cursor() as cursor:
381 sql = f""" 506 cursor.execute(
507 f"""
382 UPDATE {TARGET_TABLE_NAME_TMP} 508 UPDATE {TARGET_TABLE_NAME_TMP}
383 SET biz_review_status = 'pending', 509 SET review_claimed_by = %s,
384 reviewed_by = NULL, 510 review_claim_token = %s,
385 reviewed_at = NULL, 511 review_claimed_at = NOW(),
386 biz_review_note = NULLIF(%s, '') 512 review_claim_expires_at = DATE_ADD(NOW(), INTERVAL %s SECOND),
387 WHERE source_song_id = %s AND dedup_action = 'review' 513 review_version = review_version + 1
514 WHERE staging_id = %s
515 AND staging_status <> 'imported'
516 AND (
517 biz_review_status IN ('pending', 'unsure')
518 OR biz_review_status = 'not_required'
519 OR (reviewed_by = %s AND biz_review_status IN ('approved_import', 'rejected_duplicate', 'deleted'))
520 )
521 AND (
522 review_claim_expires_at IS NULL
523 OR review_claim_expires_at <= NOW()
524 OR review_claim_token = %s
525 )
526 """,
527 (reviewer, client_token, REVIEW_CLAIM_TTL_SECONDS, staging_id, reviewer, client_token),
528 )
529 if cursor.rowcount != 1:
530 current = _review_snapshot(cursor, staging_id)
531 conn.rollback()
532 if not current:
533 message = f"找不到 staging_id={staging_id}，请刷新页面后重试"
534 elif current.get("staging_status") == "imported":
535 message = "该记录已经入库，只能浏览，不能再领取审核"
536 elif current.get("review_claimed_by"):
537 message = f"该记录已由 {current['review_claimed_by']} 领取"
538 else:
539 message = f"该记录当前状态为 {current.get('biz_review_status') or '-'}，不能领取"
540 raise ReviewConflictError(message, current)
541 current = _review_snapshot(cursor, staging_id)
542 conn.commit()
543 return current
544
545
546 def _heartbeat_staging_review(staging_id: int, client_token: str) -> dict[str, object]:
547 with _target_conn() as conn, conn.cursor() as cursor:
548 cursor.execute(
549 f"""
550 UPDATE {TARGET_TABLE_NAME_TMP}
551 SET review_claim_expires_at = DATE_ADD(NOW(), INTERVAL %s SECOND)
552 WHERE staging_id = %s
553 AND review_claim_token = %s
554 AND review_claim_expires_at > NOW()
555 """,
556 (REVIEW_CLAIM_TTL_SECONDS, staging_id, client_token),
557 )
558 if cursor.rowcount != 1:
559 current = _review_snapshot(cursor, staging_id)
560 conn.rollback()
561 raise ReviewConflictError("审核领取已过期或已转给其他审核人", current)
562 current = _review_snapshot(cursor, staging_id)
563 conn.commit()
564 return current
565
566
567 def _review_statuses(staging_ids: list[int], client_token: str = "") -> list[dict[str, object]]:
568 if not staging_ids:
569 return []
570 placeholders = ",".join(["%s"] * len(staging_ids))
571 with _target_conn() as conn, conn.cursor() as cursor:
572 cursor.execute(
573 f"""
574 SELECT staging_id, source_song_id, biz_review_status, staging_status, reviewed_by, reviewed_at,
575 review_version,
576 CASE WHEN review_claim_expires_at > NOW() THEN review_claimed_by ELSE NULL END AS review_claimed_by,
577 review_claim_expires_at,
578 CASE WHEN review_claim_token = %s AND review_claim_expires_at > NOW() THEN 1 ELSE 0 END AS claimed_by_me
579 FROM {TARGET_TABLE_NAME_TMP}
580 WHERE staging_id IN ({placeholders})
581 """,
582 [client_token, *staging_ids],
583 )
584 return cursor.fetchall()
585
586
587 def _update_staging_review(
588 staging_id: int,
589 decision: str,
590 note: str,
591 reviewer: str,
592 expected_version: int,
593 client_token: str,
594 ) -> dict[str, object]:
595 """Persist one claimed review using optimistic concurrency control."""
596 if decision == "pending":
597 status_fields = """
598 biz_review_status = 'pending', reviewed_by = NULL, reviewed_at = NULL,
599 staging_status = CASE WHEN staging_status = 'deleted' THEN 'staged' ELSE staging_status END
388 """ 600 """
389 params = [note, source_song_id]
390 elif decision == "deleted": 601 elif decision == "deleted":
391 # 删除：不限定 dedup_action，对 review/skip 等各种 action 均可执行 602 status_fields = """
392 sql = f""" 603 biz_review_status = 'deleted', reviewed_by = %s, reviewed_at = NOW(),
393 UPDATE {TARGET_TABLE_NAME_TMP} 604 staging_status = 'deleted'
394 SET staging_status = 'deleted',
395 biz_review_status = 'deleted',
396 reviewed_by = %s,
397 reviewed_at = NOW(),
398 biz_review_note = NULLIF(%s, '')
399 WHERE source_song_id = %s
400 """ 605 """
401 params = [reviewer, note, source_song_id]
402 else: 606 else:
403 sql = f""" 607 status_fields = "biz_review_status = %s, reviewed_by = %s, reviewed_at = NOW()"
404 UPDATE {TARGET_TABLE_NAME_TMP} 608
405 SET biz_review_status = %s, 609 params: list[object]
406 reviewed_by = %s, 610 if decision == "pending":
407 reviewed_at = NOW(), 611 params = [note, staging_id, expected_version, client_token, reviewer]
408 biz_review_note = NULLIF(%s, '') 612 elif decision == "deleted":
409 WHERE source_song_id = %s AND dedup_action = 'review' 613 params = [reviewer, note, staging_id, expected_version, client_token, reviewer]
410 """ 614 else:
411 params = [decision, reviewer, note, source_song_id] 615 params = [decision, reviewer, note, staging_id, expected_version, client_token, reviewer]
412 with _target_conn() as conn, conn.cursor() as cursor: 616 with _target_conn() as conn, conn.cursor() as cursor:
413 cursor.execute(sql, params) 617 cursor.execute(
414 affected = cursor.rowcount 618 f"""
619 UPDATE {TARGET_TABLE_NAME_TMP}
620 SET {status_fields},
621 biz_review_note = NULLIF(%s, ''),
622 review_version = review_version + 1,
623 review_claimed_by = NULL,
624 review_claim_token = NULL,
625 review_claimed_at = NULL,
626 review_claim_expires_at = NULL
627 WHERE staging_id = %s
628 AND review_version = %s
629 AND review_claim_token = %s
630 AND review_claimed_by = %s
631 AND review_claim_expires_at > NOW()
632 """,
633 params,
634 )
635 if cursor.rowcount != 1:
636 current = _review_snapshot(cursor, staging_id)
637 conn.rollback()
638 raise ReviewConflictError("记录已被其他人修改，当前提交未覆盖数据库", current)
639 current = _review_snapshot(cursor, staging_id)
415 conn.commit() 640 conn.commit()
416 return {"source_song_id": source_song_id, "decision": decision, "affected": affected} 641 return current
417 642
418 643
419 def _import_approved_staging(source_ids: list[str], reviewer: str = "dashboard") -> dict[str, object]: 644 def _import_approved_staging(source_ids: list[str] | None = None) -> dict[str, object]:
420 if not source_ids: 645 """Import approved rows while holding staging-row locks to prevent duplicate imports."""
421 raise ValueError("没有可入库的 source_id")
422 placeholders = ",".join(["%s"] * len(source_ids))
423 columns = ", ".join(f"`{column}`" for column in TARGET_COLUMNS) 646 columns = ", ".join(f"`{column}`" for column in TARGET_COLUMNS)
424 select_columns = ", ".join(f"s.`{column}`" for column in TARGET_COLUMNS) 647 select_columns = ", ".join(f"s.`{column}`" for column in TARGET_COLUMNS)
425 update_imported_sql = f""" 648 with _target_conn() as conn, conn.cursor() as cursor:
426 UPDATE {TARGET_TABLE_NAME_TMP} 649 source_filter = ""
427 SET biz_review_status = 'approved_import', 650 lock_params: list[object] = []
428 reviewed_by = %s, 651 if source_ids:
429 reviewed_at = NOW() 652 source_placeholders = ",".join(["%s"] * len(source_ids))
430 WHERE source_song_id IN ({placeholders}) 653 source_filter = f" AND s.source_song_id IN ({source_placeholders})"
431 AND dedup_action = 'review' 654 lock_params.extend(source_ids)
432 AND biz_review_status IN ('pending', 'unsure') 655 cursor.execute(
433 """ 656 f"""
434 insert_sql = f""" 657 SELECT s.staging_id, s.source_song_id
658 FROM {TARGET_TABLE_NAME_TMP} s
659 INNER JOIN (
660 SELECT source_song_id, MAX(staging_id) AS max_staging_id
661 FROM {TARGET_TABLE_NAME_TMP}
662 GROUP BY source_song_id
663 ) latest ON s.staging_id = latest.max_staging_id
664 WHERE s.dedup_action = 'review'
665 AND s.biz_review_status = 'approved_import'
666 AND s.staging_status = 'staged'
667 AND (s.review_claim_expires_at IS NULL OR s.review_claim_expires_at <= NOW())
668 {source_filter}
669 ORDER BY staging_id
670 FOR UPDATE
671 """,
672 lock_params,
673 )
674 approved_rows = cursor.fetchall()
675 if not approved_rows:
676 conn.rollback()
677 raise ValueError("没有可入库的审核通过样本")
678 staging_ids = [row["staging_id"] for row in approved_rows]
679 approved_source_ids = [str(row["source_song_id"]) for row in approved_rows]
680 staging_placeholders = ",".join(["%s"] * len(staging_ids))
681 cursor.execute(
682 f"""
435 INSERT INTO {TARGET_TABLE_NAME} ({columns}) 683 INSERT INTO {TARGET_TABLE_NAME} ({columns})
436 SELECT {select_columns} 684 SELECT {select_columns}
437 FROM {TARGET_TABLE_NAME_TMP} s 685 FROM {TARGET_TABLE_NAME_TMP} s
438 WHERE s.source_song_id IN ({placeholders}) 686 WHERE s.staging_id IN ({staging_placeholders})
439 AND s.dedup_action = 'review' 687 """,
440 AND s.biz_review_status = 'approved_import' 688 staging_ids,
441 AND s.staging_status <> 'imported' 689 )
442 """ 690 inserted_count = cursor.rowcount
443 # 入库后查询目标表实际 ID，正确回填 imported_song_id（不能用暂存表自己的 id） 691 source_placeholders = ",".join(["%s"] * len(approved_source_ids))
444 lookup_target_id_sql = f""" 692 cursor.execute(
693 f"""
445 SELECT source_song_id, id AS target_id 694 SELECT source_song_id, id AS target_id
446 FROM {TARGET_TABLE_NAME} 695 FROM {TARGET_TABLE_NAME}
447 WHERE source_song_id IN ({placeholders}) 696 WHERE source_song_id IN ({source_placeholders})
448 """ 697 """,
449 mark_sql_template = f""" 698 approved_source_ids,
450 UPDATE {TARGET_TABLE_NAME_TMP} 699 )
451 SET staging_status = 'imported',
452 imported_song_id = %s,
453 error_message = NULL
454 WHERE source_song_id = %s
455 AND dedup_action = 'review'
456 AND biz_review_status = 'approved_import'
457 """
458 with _target_conn() as conn, conn.cursor() as cursor:
459 cursor.execute(update_imported_sql, [reviewer, *source_ids])
460 approved_count = cursor.rowcount
461 cursor.execute(insert_sql, source_ids)
462 inserted_count = cursor.rowcount
463 # 从目标表查出实际生成的主键 ID，逐条回填 imported_song_id
464 cursor.execute(lookup_target_id_sql, source_ids)
465 id_map = {str(row["source_song_id"]): row["target_id"] for row in cursor.fetchall()} 700 id_map = {str(row["source_song_id"]): row["target_id"] for row in cursor.fetchall()}
466 imported_updated = 0 701 imported_updated = 0
467 for sid, target_id in id_map.items(): 702 for approved_row in approved_rows:
468 cursor.execute(mark_sql_template, [target_id, sid]) 703 sid = str(approved_row["source_song_id"])
704 target_id = id_map.get(sid)
705 if target_id is None:
706 raise RuntimeError(f"入库后未找到目标记录: source_song_id={sid}")
707 cursor.execute(
708 f"""
709 UPDATE {TARGET_TABLE_NAME_TMP}
710 SET staging_status = 'imported', imported_song_id = %s, error_message = NULL,
711 review_version = review_version + 1
712 WHERE staging_id = %s AND staging_status = 'staged'
713 """,
714 (target_id, approved_row["staging_id"]),
715 )
469 imported_updated += cursor.rowcount 716 imported_updated += cursor.rowcount
470 conn.commit() 717 conn.commit()
471 return { 718 return {
472 "approved_count": approved_count, 719 "approved_count": len(approved_rows),
473 "inserted_count": inserted_count, 720 "inserted_count": inserted_count,
474 "imported_song_ids_updated": imported_updated, 721 "imported_song_ids_updated": imported_updated,
722 "source_ids": approved_source_ids,
475 } 723 }
476 724
477 725
...@@ -671,6 +919,10 @@ def _groups_response(path: Path, top_k: str, mode: str, term: str, page: int, pa ...@@ -671,6 +919,10 @@ def _groups_response(path: Path, top_k: str, mode: str, term: str, page: int, pa
671 groups = [g for g in groups if g.get("decision") == "new" and not g.get("has_l1_hint")] 919 groups = [g for g in groups if g.get("decision") == "new" and not g.get("has_l1_hint")]
672 elif decision == 'skip': 920 elif decision == 'skip':
673 groups = [g for g in groups if g.get("decision") == "skip"] 921 groups = [g for g in groups if g.get("decision") == "skip"]
922 elif decision == 'reviewed':
923 groups = [g for g in groups if g.get("reviewed")]
924 elif decision == 'pending':
925 groups = [g for g in groups if not g.get("reviewed")]
674 elif decision and decision in ('merge', 'review', 'duplicate'): 926 elif decision and decision in ('merge', 'review', 'duplicate'):
675 groups = [g for g in groups if g.get("decision") == decision] 927 groups = [g for g in groups if g.get("decision") == decision]
676 filtered = [group for group in groups if _matches_term(group, term)] 928 filtered = [group for group in groups if _matches_term(group, term)]
...@@ -779,16 +1031,32 @@ class Handler(BaseHTTPRequestHandler): ...@@ -779,16 +1031,32 @@ class Handler(BaseHTTPRequestHandler):
779 mode = params.get("mode", ["retrieval"])[0] 1031 mode = params.get("mode", ["retrieval"])[0]
780 term = _norm(params.get("q", [""])[0]) 1032 term = _norm(params.get("q", [""])[0])
781 decision = _norm(params.get("decision", [""])[0]) 1033 decision = _norm(params.get("decision", [""])[0])
1034 client_token = params.get("client_token", [""])[0].strip()
782 page = max(1, int(params.get("page", ["1"])[0])) 1035 page = max(1, int(params.get("page", ["1"])[0]))
783 page_size = min(200, max(10, int(params.get("page_size", ["50"])[0]))) 1036 page_size = min(200, max(10, int(params.get("page_size", ["50"])[0])))
784 if raw_path == STAGING_DB_PATH: 1037 if raw_path == STAGING_DB_PATH:
785 _json_response(self, _staging_groups_response(mode, term, page, page_size, decision)) 1038 _json_response(
1039 self,
1040 _staging_groups_response(mode, term, page, page_size, decision, client_token),
1041 )
786 return 1042 return
787 path = _safe_path(raw_path) 1043 path = _safe_path(raw_path)
788 _json_response(self, _groups_response(path, top_k, mode, term, page, page_size, decision)) 1044 _json_response(self, _groups_response(path, top_k, mode, term, page, page_size, decision))
789 except Exception as exc: # noqa: BLE001 - local diagnostic API 1045 except Exception as exc: # noqa: BLE001 - local diagnostic API
790 _error(self, str(exc), status=404) 1046 _error(self, str(exc), status=404)
791 return 1047 return
1048 if parsed.path == "/api/review-statuses":
1049 params = parse_qs(parsed.query)
1050 try:
1051 raw_ids = params.get("staging_ids", [""])[0]
1052 staging_ids = [int(value) for value in raw_ids.split(",") if value.strip()]
1053 if len(staging_ids) > 200:
1054 raise ValueError("最多同步 200 条记录")
1055 client_token = params.get("client_token", [""])[0].strip()
1056 _json_response(self, {"rows": _review_statuses(staging_ids, client_token)})
1057 except Exception as exc: # noqa: BLE001
1058 _error(self, str(exc), status=400)
1059 return
792 if parsed.path == "/api/query": 1060 if parsed.path == "/api/query":
793 params = parse_qs(parsed.query) 1061 params = parse_qs(parsed.query)
794 try: 1062 try:
...@@ -825,20 +1093,47 @@ class Handler(BaseHTTPRequestHandler): ...@@ -825,20 +1093,47 @@ class Handler(BaseHTTPRequestHandler):
825 1093
826 def do_POST(self) -> None: 1094 def do_POST(self) -> None:
827 parsed = urlparse(self.path) 1095 parsed = urlparse(self.path)
1096 if parsed.path in {"/api/review-claim", "/api/review-heartbeat"}:
1097 try:
1098 length = int(self.headers.get("Content-Length", "0"))
1099 payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
1100 staging_id = int(payload.get("staging_id") or 0)
1101 client_token = str(payload.get("client_token") or "").strip()
1102 if not staging_id:
1103 raise ValueError("staging_id is required")
1104 if parsed.path == "/api/review-claim":
1105 reviewer = str(payload.get("reviewer") or "").strip()
1106 current = _claim_staging_review(staging_id, reviewer, client_token)
1107 else:
1108 current = _heartbeat_staging_review(staging_id, client_token)
1109 _json_response(self, {"record": current})
1110 except ReviewConflictError as exc:
1111 _conflict_response(self, exc)
1112 except Exception as exc: # noqa: BLE001
1113 _error(self, str(exc), status=400)
1114 return
828 if parsed.path == "/api/staging-review": 1115 if parsed.path == "/api/staging-review":
829 try: 1116 try:
830 length = int(self.headers.get("Content-Length", "0")) 1117 length = int(self.headers.get("Content-Length", "0"))
831 payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}") 1118 payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
832 source_id = str(payload.get("source_id", "")).strip() 1119 staging_id = int(payload.get("staging_id") or 0)
833 decision = str(payload.get("decision", "")).strip() 1120 decision = str(payload.get("decision", "")).strip()
834 note = str(payload.get("note", "")).strip() 1121 note = str(payload.get("note", "")).strip()
835 reviewer = str(payload.get("reviewer", "dashboard")).strip() or "dashboard" 1122 reviewer = str(payload.get("reviewer") or "").strip()
836 if not source_id: 1123 client_token = str(payload.get("client_token") or "").strip()
837 raise ValueError("source_id is required") 1124 expected_version = int(payload.get("expected_version", -1))
1125 if not staging_id:
1126 raise ValueError("staging_id is required")
1127 if not reviewer or not client_token or expected_version < 0:
1128 raise ValueError("reviewer, client_token and expected_version are required")
838 if decision not in ("approved_import", "rejected_duplicate", "unsure", "pending", "deleted"): 1129 if decision not in ("approved_import", "rejected_duplicate", "unsure", "pending", "deleted"):
839 raise ValueError(f"invalid decision: {decision}") 1130 raise ValueError(f"invalid decision: {decision}")
840 result = _update_staging_review(source_id, decision, note, reviewer) 1131 result = _update_staging_review(
841 _json_response(self, result) 1132 staging_id, decision, note, reviewer, expected_version, client_token
1133 )
1134 _json_response(self, {"record": result})
1135 except ReviewConflictError as exc:
1136 _conflict_response(self, exc)
842 except Exception as exc: # noqa: BLE001 1137 except Exception as exc: # noqa: BLE001
843 _error(self, str(exc), status=400) 1138 _error(self, str(exc), status=400)
844 return 1139 return
...@@ -848,23 +1143,19 @@ class Handler(BaseHTTPRequestHandler): ...@@ -848,23 +1143,19 @@ class Handler(BaseHTTPRequestHandler):
848 try: 1143 try:
849 length = int(self.headers.get("Content-Length", "0")) 1144 length = int(self.headers.get("Content-Length", "0"))
850 payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}") 1145 payload = json.loads(self.rfile.read(length).decode("utf-8") or "{}")
851 rows = payload.get("rows") or [] 1146 source_ids = payload.get("source_ids")
852 if not isinstance(rows, list): 1147 if source_ids is not None and not isinstance(source_ids, list):
853 raise ValueError("rows must be a list") 1148 raise ValueError("source_ids must be a list")
1149 result = _import_approved_staging(
1150 [str(value).strip() for value in source_ids if str(value).strip()]
1151 if source_ids
1152 else None
1153 )
854 approved = [ 1154 approved = [
855 { 1155 {"source_id": source_id, "review_decision": "import", "review_note": ""}
856 "source_id": str(row.get("source_id", "")).strip(), 1156 for source_id in result["source_ids"]
857 "review_decision": "import",
858 "review_note": str(row.get("review_note", "")).strip(),
859 }
860 for row in rows
861 if str(row.get("source_id", "")).strip()
862 and str(row.get("review_decision", "")).strip().lower() in {"import", "导入", "not_duplicate"}
863 ] 1157 ]
864 if not approved:
865 raise ValueError("没有可入库的审核通过样本")
866 review_csv = _write_review_import_csv(approved) 1158 review_csv = _write_review_import_csv(approved)
867 result = _import_approved_staging([row["source_id"] for row in approved])
868 _json_response(self, {"review_csv": str(review_csv), **result}) 1159 _json_response(self, {"review_csv": str(review_csv), **result})
869 except Exception as exc: # noqa: BLE001 - local diagnostic API 1160 except Exception as exc: # noqa: BLE001 - local diagnostic API
870 _error(self, str(exc), status=400) 1161 _error(self, str(exc), status=400)
...@@ -877,6 +1168,8 @@ class Handler(BaseHTTPRequestHandler): ...@@ -877,6 +1168,8 @@ class Handler(BaseHTTPRequestHandler):
877 body = path.read_bytes() 1168 body = path.read_bytes()
878 self.send_response(200) 1169 self.send_response(200)
879 self.send_header("Content-Type", content_type) 1170 self.send_header("Content-Type", content_type)
1171 if path.suffix.lower() in {".html", ".js", ".css"}:
1172 self.send_header("Cache-Control", "no-cache")
880 self.send_header("Content-Length", str(len(body))) 1173 self.send_header("Content-Length", str(len(body)))
881 self.end_headers() 1174 self.end_headers()
882 self.wfile.write(body) 1175 self.wfile.write(body)
...@@ -886,8 +1179,14 @@ def main() -> None: ...@@ -886,8 +1179,14 @@ def main() -> None:
886 parser = argparse.ArgumentParser(description="Serve L2 lyric review dashboard") 1179 parser = argparse.ArgumentParser(description="Serve L2 lyric review dashboard")
887 parser.add_argument("--host", default="127.0.0.1") 1180 parser.add_argument("--host", default="127.0.0.1")
888 parser.add_argument("--port", type=int, default=8765) 1181 parser.add_argument("--port", type=int, default=8765)
1182 parser.add_argument(
1183 "--migrate-review-schema",
1184 action="store_true",
1185 help="add the staging-table columns/index required for collaborative reviewing",
1186 )
889 args = parser.parse_args() 1187 args = parser.parse_args()
890 1188
1189 _ensure_review_schema(apply_migration=args.migrate_review_schema)
891 server = ThreadingHTTPServer((args.host, args.port), Handler) 1190 server = ThreadingHTTPServer((args.host, args.port), Handler)
892 print(f"L2 dashboard: http://{args.host}:{args.port}") 1191 print(f"L2 dashboard: http://{args.host}:{args.port}")
893 server.serve_forever() 1192 server.serve_forever()
......
1 from __future__ import annotations
2
3 from pathlib import Path
4 from unittest.mock import patch
5
6 import pytest
7
8 import serve_l2_dashboard as dashboard
9
10
11 class FakeCursor:
12 def __init__(self, update_rowcount: int, snapshot: dict[str, object]) -> None:
13 self.update_rowcount = update_rowcount
14 self.snapshot = snapshot
15 self.rowcount = 0
16 self.executions: list[tuple[str, object]] = []
17
18 def __enter__(self):
19 return self
20
21 def __exit__(self, *_args):
22 return False
23
24 def execute(self, sql, params=None):
25 self.executions.append((sql, params))
26 self.rowcount = self.update_rowcount if sql.lstrip().startswith("UPDATE") else 0
27
28 def fetchone(self):
29 return self.snapshot
30
31
32 class FakeConnection:
33 def __init__(self, cursor: FakeCursor) -> None:
34 self._cursor = cursor
35 self.committed = False
36 self.rolled_back = False
37
38 def __enter__(self):
39 return self
40
41 def __exit__(self, *_args):
42 return False
43
44 def cursor(self):
45 return self._cursor
46
47 def commit(self):
48 self.committed = True
49
50 def rollback(self):
51 self.rolled_back = True
52
53
54 def test_review_update_uses_staging_version_and_claim_token():
55 snapshot = {
56 "staging_id": 42,
57 "source_song_id": "song-1",
58 "biz_review_status": "approved_import",
59 "review_version": 8,
60 }
61 cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
62 conn = FakeConnection(cursor)
63
64 with patch.object(dashboard, "_target_conn", return_value=conn):
65 result = dashboard._update_staging_review(
66 42, "approved_import", "ok", "alice", 7, "browser-token"
67 )
68
69 update_sql, params = cursor.executions[0]
70 assert "WHERE staging_id = %s" in update_sql
71 assert "review_version = %s" in update_sql
72 assert "review_claim_token = %s" in update_sql
73 assert params[-4:] == [42, 7, "browser-token", "alice"]
74 assert result == snapshot
75 assert conn.committed
76
77
78 def test_stale_review_returns_conflict_without_overwrite():
79 snapshot = {
80 "staging_id": 42,
81 "biz_review_status": "rejected_duplicate",
82 "reviewed_by": "bob",
83 "review_version": 9,
84 }
85 cursor = FakeCursor(update_rowcount=0, snapshot=snapshot)
86 conn = FakeConnection(cursor)
87
88 with patch.object(dashboard, "_target_conn", return_value=conn):
89 with pytest.raises(dashboard.ReviewConflictError) as raised:
90 dashboard._update_staging_review(
91 42, "approved_import", "stale", "alice", 7, "browser-token"
92 )
93
94 assert raised.value.current_record == snapshot
95 assert conn.rolled_back
96 assert not conn.committed
97
98
99 def test_claim_conflict_exposes_current_owner():
100 snapshot = {
101 "staging_id": 42,
102 "review_claimed_by": "bob",
103 "review_version": 3,
104 }
105 cursor = FakeCursor(update_rowcount=0, snapshot=snapshot)
106 conn = FakeConnection(cursor)
107
108 with patch.object(dashboard, "_target_conn", return_value=conn):
109 with pytest.raises(dashboard.ReviewConflictError) as raised:
110 dashboard._claim_staging_review(42, "alice", "alice-token")
111
112 assert raised.value.current_record["review_claimed_by"] == "bob"
113 assert conn.rolled_back
114
115
116 def test_heartbeat_does_not_increment_review_version():
117 snapshot = {"staging_id": 42, "review_version": 5}
118 cursor = FakeCursor(update_rowcount=1, snapshot=snapshot)
119 conn = FakeConnection(cursor)
120
121 with patch.object(dashboard, "_target_conn", return_value=conn):
122 dashboard._heartbeat_staging_review(42, "browser-token")
123
124 update_sql, _params = cursor.executions[0]
125 assert "review_version" not in update_sql
126 assert conn.committed
127
128
129 class ImportCursor(FakeCursor):
130 def __init__(self) -> None:
131 super().__init__(update_rowcount=1, snapshot={})
132 self.result_sets = [
133 [{"staging_id": 42, "source_song_id": "song-1"}],
134 [{"source_song_id": "song-1", "target_id": 9001}],
135 ]
136
137 def execute(self, sql, params=None):
138 self.executions.append((sql, params))
139 self.rowcount = 1
140
141 def fetchall(self):
142 return self.result_sets.pop(0)
143
144
145 def test_import_locks_only_unclaimed_staged_rows():
146 cursor = ImportCursor()
147 conn = FakeConnection(cursor)
148
149 with patch.object(dashboard, "_target_conn", return_value=conn):
150 result = dashboard._import_approved_staging()
151
152 lock_sql, _params = cursor.executions[0]
153 assert "FOR UPDATE" in lock_sql
154 assert "staging_status = 'staged'" in lock_sql
155 assert "review_claim_expires_at <= NOW()" in lock_sql
156 assert result["inserted_count"] == 1
157 assert result["source_ids"] == ["song-1"]
158 assert conn.committed
159
160
161 def test_browsing_a_list_item_can_auto_claim_without_reloading_groups():
162 html = Path("l2_review_dashboard.html").read_text(encoding="utf-8")
163 start = html.index("for (const btn of els.queryList.querySelectorAll('.query-item'))")
164 end = html.index("function selectedQueryRows()", start)
165 click_handler = html[start:end]
166
167 assert "loadQueryRows()" in click_handler
168 assert "claimCurrentReview" in click_handler
169 assert "loadGroups()" not in click_handler
170
171
172 def test_staging_bigint_is_never_converted_to_javascript_number():
173 html = Path("l2_review_dashboard.html").read_text(encoding="utf-8")
174
175 assert "Number(query.staging_id)" not in html
176 assert "staging_id: String(query.staging_id)" in html
177
178
179 def test_dashboard_distinguishes_reviewed_from_submitted():
180 html = Path("l2_review_dashboard.html").read_text(encoding="utf-8")
181 row = dashboard._staging_dashboard_row(
182 {
183 "staging_id": 42,
184 "source_song_id": "song-1",
185 "dedup_action": "review",
186 "biz_review_status": "approved_import",
187 "staging_status": "imported",
188 }
189 )
190
191 assert row["staging_status"] == "imported"
192 assert '<option value="reviewed">已审核</option>' in html
193 assert '<option value="submitted">已提交</option>' in html