diff --git a/apps/desktop-client/src/main/adapters/baijiahao.ts b/apps/desktop-client/src/main/adapters/baijiahao.ts index 7fb2da2..0daf286 100644 --- a/apps/desktop-client/src/main/adapters/baijiahao.ts +++ b/apps/desktop-client/src/main/adapters/baijiahao.ts @@ -343,10 +343,14 @@ async function processContentImages( html: string, appId: string, ): Promise { - const processed = await uploadHtmlImages(html, async (sourceUrl) => { - await waitForHumanPace(context.signal) - return await uploadImage(context, sourceUrl, appId, false) - }, { signal: context.signal }) + const processed = await uploadHtmlImages( + html, + async (sourceUrl) => { + await waitForHumanPace(context.signal) + return await uploadImage(context, sourceUrl, appId, false) + }, + { signal: context.signal }, + ) return processed.html } @@ -562,7 +566,7 @@ export const baijiahaoAdapter: PublishAdapter = { throw new Error('baijiahao_cover_upload_failed') } - context.reportProgress('baijiahao.submit') + await context.reportProgress('baijiahao.submit') await waitForHumanPace(context.signal) const response: BaijiahaoPublishResponse = await sessionFetchJson( context.session, diff --git a/apps/desktop-client/src/main/adapters/base.ts b/apps/desktop-client/src/main/adapters/base.ts index 8507391..14526ce 100644 --- a/apps/desktop-client/src/main/adapters/base.ts +++ b/apps/desktop-client/src/main/adapters/base.ts @@ -17,7 +17,7 @@ export interface AdapterContext { } | null signal: AbortSignal phase: AdapterTaskPhase - reportProgress(stage: string): void + reportProgress(stage: string): void | Promise } export interface PublishAdapterContext extends AdapterContext { diff --git a/apps/desktop-client/src/main/adapters/bilibili.ts b/apps/desktop-client/src/main/adapters/bilibili.ts index 89edc83..c6354f9 100644 --- a/apps/desktop-client/src/main/adapters/bilibili.ts +++ b/apps/desktop-client/src/main/adapters/bilibili.ts @@ -72,9 +72,7 @@ type BilibiliEditorResult = { articleId?: string } -type BilibiliSubmitOutcome = - | { ok: true; articleId: string } - | { ok: false; message: string } +type BilibiliSubmitOutcome = { ok: true; articleId: string } | { ok: false; message: string } const mixinKeyEncTab = [ 46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49, 33, 9, 42, 19, 29, 28, @@ -319,10 +317,7 @@ async function fillAndSubmitEditor( await page.waitForLoadState('networkidle', { timeout: 5_000 }).catch(() => undefined) const fillResult = await page.evaluate( - async ({ - title: nextTitle, - html: nextHtml, - }): Promise => { + async ({ title: nextTitle, html: nextHtml }): Promise => { const sleep = (ms: number) => new Promise((resolve) => { setTimeout(resolve, ms) @@ -740,8 +735,7 @@ export const bilibiliAdapter: PublishAdapter = { const coverAssetUrl = context.article.cover_asset_url?.trim() || '' if (coverAssetUrl) { - const hasExistingCover = - extractImageSources(html)[0] === coverAssetUrl + const hasExistingCover = extractImageSources(html)[0] === coverAssetUrl if (!hasExistingCover) { html = `${html}` } @@ -752,7 +746,7 @@ export const bilibiliAdapter: PublishAdapter = { const title = context.article.title?.trim() || '未命名文章' const startedAt = Date.now() - context.reportProgress('bilibili.submit_editor') + await context.reportProgress('bilibili.submit_editor') const editorResult = await fillAndSubmitEditor(context, title, processedHtml) if (!editorResult.success) { throw new Error(editorResult.message || 'bilibili_publish_failed') diff --git a/apps/desktop-client/src/main/adapters/dongchedi.ts b/apps/desktop-client/src/main/adapters/dongchedi.ts index d515624..9fc08a6 100644 --- a/apps/desktop-client/src/main/adapters/dongchedi.ts +++ b/apps/desktop-client/src/main/adapters/dongchedi.ts @@ -291,26 +291,26 @@ async function dongchediPageFetchJson( if (page && !page.isClosed()) { const envelope = await raceWithAbort( page.evaluate( - async ({ fetchInput, fetchInit }) => { - const controller = new AbortController() - const timeout = setTimeout(() => controller.abort(), 30_000) - try { - const response = await fetch(fetchInput, { - ...fetchInit, - signal: controller.signal, - }) - const text = await response.text().catch(() => '') - return { ok: response.ok, status: response.status, body: text } - } catch (error) { - return { ok: false, error: String((error && (error as Error).message) || error) } - } finally { - clearTimeout(timeout) - } - }, - { - fetchInput: input, - fetchInit: effectiveInit, - }, + async ({ fetchInput, fetchInit }) => { + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), 30_000) + try { + const response = await fetch(fetchInput, { + ...fetchInit, + signal: controller.signal, + }) + const text = await response.text().catch(() => '') + return { ok: response.ok, status: response.status, body: text } + } catch (error) { + return { ok: false, error: String((error && (error as Error).message) || error) } + } finally { + clearTimeout(timeout) + } + }, + { + fetchInput: input, + fetchInit: effectiveInit, + }, ), { signal: context.signal, timeoutMs: 35_000 }, ) @@ -337,10 +337,13 @@ async function dongchediPageFetchJson( } })()` - const serialized = await raceWithAbort(webContents.executeJavaScript(script, true) as Promise, { - signal: context.signal, - timeoutMs: 35_000, - }) + const serialized = await raceWithAbort( + webContents.executeJavaScript(script, true) as Promise, + { + signal: context.signal, + timeoutMs: 35_000, + }, + ) if (typeof serialized !== 'string' || !serialized) { throw new Error('dongchedi_page_fetch_failed') } @@ -406,29 +409,32 @@ async function fetchDongchediCsrfToken(context: PublishAdapterContext): Promise< const page = context.playwright?.page if (page && !page.isClosed()) { - const token = await raceWithAbort(page.evaluate(async (url) => { - const controller = new AbortController() - const timeout = setTimeout(() => controller.abort(), 15_000) - try { - const response = await fetch(url, { - method: 'HEAD', - credentials: 'include', - signal: controller.signal, - headers: { - 'x-secsdk-csrf-request': '1', - 'x-secsdk-csrf-version': '1.2.10', - }, - }) - return response.headers.get('x-ware-csrf-token') || '' - } catch { - return '' - } finally { - clearTimeout(timeout) - } - }, `${DONGCHEDI_ORIGIN}/motor/mp_index/api/user/info?`), { - signal: context.signal, - timeoutMs: 18_000, - }) + const token = await raceWithAbort( + page.evaluate(async (url) => { + const controller = new AbortController() + const timeout = setTimeout(() => controller.abort(), 15_000) + try { + const response = await fetch(url, { + method: 'HEAD', + credentials: 'include', + signal: controller.signal, + headers: { + 'x-secsdk-csrf-request': '1', + 'x-secsdk-csrf-version': '1.2.10', + }, + }) + return response.headers.get('x-ware-csrf-token') || '' + } catch { + return '' + } finally { + clearTimeout(timeout) + } + }, `${DONGCHEDI_ORIGIN}/motor/mp_index/api/user/info?`), + { + signal: context.signal, + timeoutMs: 18_000, + }, + ) const parts = token.split(',') return parts.length >= 2 && parts[1] ? parts[1] : null } @@ -1126,7 +1132,7 @@ export const dongchediAdapter: PublishAdapter = { const contentImageSources = extractImageSources(html) const content = await processDongchediContentImages(context, html, contentImageSources) - context.reportProgress('dongchedi.save_draft') + await context.reportProgress('dongchedi.save_draft') const draftBody = buildPublishBody(title, content, verticalCover, feedCover, { save: 0, }) @@ -1163,7 +1169,7 @@ export const dongchediAdapter: PublishAdapter = { ) } - context.reportProgress('dongchedi.submit') + await context.reportProgress('dongchedi.submit') const publishBody = buildPublishBody(title, content, verticalCover, feedCover, { save: 1, pgcId: draftArticleId, diff --git a/apps/desktop-client/src/main/adapters/jianshu.ts b/apps/desktop-client/src/main/adapters/jianshu.ts index 132ec5d..ae0c7d3 100644 --- a/apps/desktop-client/src/main/adapters/jianshu.ts +++ b/apps/desktop-client/src/main/adapters/jianshu.ts @@ -86,8 +86,8 @@ export const jianshuAdapter: PublishAdapter = { fetchJson: (input, init) => sessionFetchJson(context.session, input, init, { signal: context.signal }), fetchImageBlob: (sourceUrl) => fetchAssetBlob(sourceUrl, context.signal), - reportProgress(stage) { - context.reportProgress(`jianshu.${stage}`) + async reportProgress(stage) { + await context.reportProgress(`jianshu.${stage}`) }, }, ) diff --git a/apps/desktop-client/src/main/adapters/juejin.ts b/apps/desktop-client/src/main/adapters/juejin.ts index 28b029e..520d1cd 100644 --- a/apps/desktop-client/src/main/adapters/juejin.ts +++ b/apps/desktop-client/src/main/adapters/juejin.ts @@ -1060,7 +1060,7 @@ export const juejinAdapter: PublishAdapter = { context.signal, ) - context.reportProgress('juejin.create_draft') + await context.reportProgress('juejin.create_draft') const draftId = await createDraft(context, payload, markdown, uuid) const draftUrl = `${JUEJIN_ORIGIN}/editor/drafts/${encodeURIComponent(draftId)}` const publishType = normalizePublishType(payload) @@ -1073,7 +1073,7 @@ export const juejinAdapter: PublishAdapter = { } } - context.reportProgress('juejin.submit_editor') + await context.reportProgress('juejin.submit_editor') const editorResult = await publishDraft(context, draftId) if (!editorResult.success) { throw new Error(editorResult.message || 'juejin_publish_failed') diff --git a/apps/desktop-client/src/main/adapters/qiehao.ts b/apps/desktop-client/src/main/adapters/qiehao.ts index 2c43fc6..abb59d8 100644 --- a/apps/desktop-client/src/main/adapters/qiehao.ts +++ b/apps/desktop-client/src/main/adapters/qiehao.ts @@ -534,7 +534,7 @@ export const qiehaoAdapter: PublishAdapter = { ? `${QIEHAO_ORIGIN}/marticlepublish/omPublish` : `${QIEHAO_ORIGIN}/marticlepublish/omSave` - context.reportProgress('qiehao.submit') + await context.reportProgress('qiehao.submit') const response = await sessionFetchJson( context.session, endpoint, diff --git a/apps/desktop-client/src/main/adapters/smzdm.ts b/apps/desktop-client/src/main/adapters/smzdm.ts index 9ca7126..f396087 100644 --- a/apps/desktop-client/src/main/adapters/smzdm.ts +++ b/apps/desktop-client/src/main/adapters/smzdm.ts @@ -13,7 +13,6 @@ import { normalizeArticleHtml, raceWithAbort, sessionCookieHeader, - sessionCookieValue, sessionFetchJson, } from './common' @@ -624,8 +623,7 @@ async function calculateEditorSignature( return editor.commands.getTextCount() }, html), { signal: context.signal, timeoutMs: 20_000 }, - ) - .catch(() => null) + ).catch(() => null) const normalizedCount = typeof textCount === 'number' && Number.isFinite(textCount) ? Math.round(textCount) : null @@ -874,7 +872,7 @@ export const smzdmAdapter: PublishAdapter = { throw new Error('smzdm_not_logged_in') } - context.reportProgress('smzdm.create_article_id') + await context.reportProgress('smzdm.create_article_id') const articleId = await getArticleId(context) let html = normalizeArticleHtml(context.article) @@ -905,7 +903,7 @@ export const smzdmAdapter: PublishAdapter = { account.uid, ) - context.reportProgress('smzdm.submit') + await context.reportProgress('smzdm.submit') await waitForHumanPace(context.signal) const response = await sessionFetchJson( context.session, diff --git a/apps/desktop-client/src/main/adapters/sohuhao.ts b/apps/desktop-client/src/main/adapters/sohuhao.ts index 9f22a4e..447d6d4 100644 --- a/apps/desktop-client/src/main/adapters/sohuhao.ts +++ b/apps/desktop-client/src/main/adapters/sohuhao.ts @@ -374,7 +374,7 @@ export const sohuhaoAdapter: PublishAdapter = { ) const publishType = resolvePublishType(payload) - context.reportProgress('sohuhao.submit') + await context.reportProgress('sohuhao.submit') const articleId = await submitArticle(context, account, processed.html, publishType) return { diff --git a/apps/desktop-client/src/main/adapters/toutiao.ts b/apps/desktop-client/src/main/adapters/toutiao.ts index 3980e41..fa950df 100644 --- a/apps/desktop-client/src/main/adapters/toutiao.ts +++ b/apps/desktop-client/src/main/adapters/toutiao.ts @@ -48,8 +48,8 @@ export const toutiaoAdapter: PublishAdapter = { fetchImageBlob: (sourceUrl) => fetchToutiaoImageBlob(sourceUrl, context.signal), uploadHtmlImages: (sourceHtml, uploader) => uploadHtmlImages(sourceHtml, uploader, { signal: context.signal }), - reportProgress(stage) { - context.reportProgress(`toutiao.${stage}`) + async reportProgress(stage) { + await context.reportProgress(`toutiao.${stage}`) }, }, ) diff --git a/apps/desktop-client/src/main/adapters/wangyihao.ts b/apps/desktop-client/src/main/adapters/wangyihao.ts index bd953ca..60735fd 100644 --- a/apps/desktop-client/src/main/adapters/wangyihao.ts +++ b/apps/desktop-client/src/main/adapters/wangyihao.ts @@ -1030,10 +1030,10 @@ export const wangyihaoAdapter: PublishAdapter = { { signal: context.signal }, ) - context.reportProgress('wangyihao.save_draft') + await context.reportProgress('wangyihao.save_draft') const draftId = await saveDraft(context, account, detail, processed.html, token) - context.reportProgress('wangyihao.publish_draft') + await context.reportProgress('wangyihao.publish_draft') const published = await publishDraft(context, draftId) if (!published) { throw new Error('wangyihao_publish_click_failed') diff --git a/apps/desktop-client/src/main/adapters/weixin-gzh.ts b/apps/desktop-client/src/main/adapters/weixin-gzh.ts index e2b869a..e2f54b8 100644 --- a/apps/desktop-client/src/main/adapters/weixin-gzh.ts +++ b/apps/desktop-client/src/main/adapters/weixin-gzh.ts @@ -1609,7 +1609,7 @@ async function completeWeixinPublishConfirmation( return response } - context.reportProgress('weixin_gzh.disable_mass_notification') + await context.reportProgress('weixin_gzh.disable_mass_notification') await waitForWeixinDelay(context.signal, 300) const notification = await disableWeixinMassNotification(page) if (notification.found && notification.reason === 'switch_missing') { @@ -1657,7 +1657,7 @@ async function completeWeixinPublishConfirmation( signal: context.signal, }) if (continueConfirm.clicked) { - context.reportProgress('weixin_gzh.continue_publish_without_notification') + await context.reportProgress('weixin_gzh.continue_publish_without_notification') } const response = await waitForWeixinPublishResponseOrAuthorization( @@ -2125,8 +2125,9 @@ export const weixinGzhAdapter: PublishAdapter = { } context.reportProgress('weixin_gzh.upload_content_images') - const processed = await uploadHtmlImages(html, async (sourceUrl) => - uploadImage(context, meta, sourceUrl), + const processed = await uploadHtmlImages( + html, + async (sourceUrl) => uploadImage(context, meta, sourceUrl), { signal: context.signal }, ) const content = wrapWeixinContent(processed.html) @@ -2136,7 +2137,7 @@ export const weixinGzhAdapter: PublishAdapter = { ? await uploadCover(context, meta, coverAssetUrl).catch(() => null) : null - context.reportProgress('weixin_gzh.save_draft') + await context.reportProgress('weixin_gzh.save_draft') const form = buildOperateForm(context, meta, content, cover) const response = await sessionFetchJson( context.session, @@ -2169,7 +2170,7 @@ export const weixinGzhAdapter: PublishAdapter = { throw new Error(`weixin_gzh_save_failed:${message}`) } - context.reportProgress('weixin_gzh.publish') + await context.reportProgress('weixin_gzh.publish') const articleTitle = context.article.title?.trim() || '未命名文章' let publishResponse: WeixinOperateResponse | null try { diff --git a/apps/desktop-client/src/main/adapters/zhihu.ts b/apps/desktop-client/src/main/adapters/zhihu.ts index 136c5d6..a8edbe5 100644 --- a/apps/desktop-client/src/main/adapters/zhihu.ts +++ b/apps/desktop-client/src/main/adapters/zhihu.ts @@ -54,10 +54,15 @@ async function zhihuFetch( input: string, init?: RequestInit, ): Promise { - return await sessionFetchJson(context.session, input, { - ...init, - headers: buildHeaders(init?.headers), - }, { signal: context.signal }) + return await sessionFetchJson( + context.session, + input, + { + ...init, + headers: buildHeaders(init?.headers), + }, + { signal: context.signal }, + ) } function markdownToHtml(markdown: string): string { @@ -245,7 +250,7 @@ async function createDraft( context.reportProgress('zhihu.upload_content_images') const content = await processContentImages(context, transformedContent) - context.reportProgress('zhihu.create_draft') + await context.reportProgress('zhihu.create_draft') const created = await zhihuFetch( context, 'https://zhuanlan.zhihu.com/api/articles/drafts', @@ -276,7 +281,7 @@ async function publishDraft( ): Promise { const xsrfToken = await sessionCookieValue(context.session, 'zhihu.com', '_xsrf') - context.reportProgress('zhihu.publish_draft') + await context.reportProgress('zhihu.publish_draft') const result = await zhihuFetch( context, 'https://www.zhihu.com/api/v4/content/publish', diff --git a/apps/desktop-client/src/main/adapters/zol.ts b/apps/desktop-client/src/main/adapters/zol.ts index 180f062..b2612c8 100644 --- a/apps/desktop-client/src/main/adapters/zol.ts +++ b/apps/desktop-client/src/main/adapters/zol.ts @@ -570,7 +570,7 @@ export const zolAdapter: PublishAdapter = { const form = new FormData() appendBaseFormFields(form, title, account.uid) - context.reportProgress('zol.create_draft') + await context.reportProgress('zol.create_draft') const draftId = await createDraftId(context, form) const coverAssetUrl = context.article.cover_asset_url?.trim() || '' @@ -598,7 +598,7 @@ export const zolAdapter: PublishAdapter = { form.append('taskType', '1') form.append('taskIds', '[]') - context.reportProgress('zol.submit') + await context.reportProgress('zol.submit') const response = await sessionFetchJson( context.session, `${ZOL_API_ORIGIN}/api/v1/creator.content.save.orther`, diff --git a/apps/desktop-client/src/main/runtime-controller.ts b/apps/desktop-client/src/main/runtime-controller.ts index d7610f8..f1a8682 100644 --- a/apps/desktop-client/src/main/runtime-controller.ts +++ b/apps/desktop-client/src/main/runtime-controller.ts @@ -1677,7 +1677,11 @@ function armPublishTaskDeadline( function resolvePublishTaskTimeoutMs(platform: string): number { const envKey = `GEO_DESKTOP_PUBLISH_TIMEOUT_${platform.toUpperCase().replace(/[^A-Z0-9]/g, '_')}_MS` - return resolvePositiveEnvInteger(envKey) ?? resolvePositiveEnvInteger('GEO_DESKTOP_PUBLISH_TIMEOUT_MS') ?? defaultPublishTaskTimeoutMs + return ( + resolvePositiveEnvInteger(envKey) ?? + resolvePositiveEnvInteger('GEO_DESKTOP_PUBLISH_TIMEOUT_MS') ?? + defaultPublishTaskTimeoutMs + ) } function resolvePositiveEnvInteger(key: string): number | null { @@ -1695,16 +1699,30 @@ function resolvePositiveEnvInteger(key: string): number | null { function isIrreversiblePublishStage(stage: string): boolean { const normalized = stage.trim().toLowerCase() return ( + normalized.endsWith('.create_note') || + normalized.endsWith('.create_article_id') || + normalized.endsWith('.create_draft') || + normalized.endsWith('.publish') || + normalized.endsWith('.publish_draft') || + normalized.endsWith('.save_content') || normalized.endsWith('.submit') || + normalized.endsWith('.submit_editor') || normalized.endsWith('.save_draft') || + normalized === 'create_note' || + normalized === 'create_article_id' || + normalized === 'create_draft' || + normalized === 'publish' || + normalized === 'publish_draft' || + normalized === 'save_content' || normalized === 'submit' || + normalized === 'submit_editor' || normalized === 'save_draft' ) } -// Persist a durable "submit started" marker (server-side, before/at the irreversible POST) and a -// local flag the abort path reads. Fires at most once per execution and never blocks publishing. -function markPublishSubmitStartedIfNeeded(taskId: string, stage: string): void { +// Persist a durable "submit started" marker before the irreversible platform write. The local flag +// is set only after the server confirms the marker, so recovery can trust publish_submit_started_at. +async function markPublishSubmitStartedIfNeeded(taskId: string, stage: string): Promise { if (!isIrreversiblePublishStage(stage)) { return } @@ -1712,19 +1730,31 @@ function markPublishSubmitStartedIfNeeded(taskId: string, stage: string): void { if (!active || active.kind !== 'publish' || active.submitStartedAt !== null) { return } - active.submitStartedAt = Date.now() - state.activeExecutions.set(taskId, active) - const leaseToken = state.tasks.get(taskId)?.leaseToken if (!leaseToken) { - return + throw new Error('publish_submit_marker_lease_missing') + } + await markDesktopTaskPublishSubmitStarted(taskId, leaseToken) + + active.submitStartedAt = Date.now() + state.activeExecutions.set(taskId, active) +} + +function asSubmitUncertainExecution(result: AdapterExecutionResult): AdapterExecutionResult { + if (result.status !== 'failed') { + return result + } + return { + ...result, + status: 'unknown', + summary: '发布可能已提交到平台,结果待人工确认,已避免自动重复发布。', + error: { + ...(result.error ?? {}), + publish_submit_uncertain: true, + original_status: 'failed', + original_summary: result.summary, + }, } - void markDesktopTaskPublishSubmitStarted(taskId, leaseToken).catch((error) => { - console.warn('[desktop-runtime] mark publish submit started failed', { - taskId, - message: error instanceof Error ? error.message : String(error), - }) - }) } function pumpExecutionLoop(): void { @@ -2824,11 +2854,7 @@ async function extendActiveLease(taskId: string, leaseToken: string): Promise } export type JianshuPublishResult = @@ -394,7 +394,7 @@ export async function publishJianshuArticle( input: JianshuPublishArticleInput, transport: JianshuPublishTransport, ): Promise { - transport.reportProgress?.('media_info') + await transport.reportProgress?.('media_info') const media = await fetchJianshuMediaSnapshot(transport.fetchJson) if (!media?.platformUid) { return { @@ -415,7 +415,7 @@ export async function publishJianshuArticle( } } - transport.reportProgress?.('fetch_notebook') + await transport.reportProgress?.('fetch_notebook') const notebookId = await fetchPrimaryNotebookId(transport.fetchJson) if (!notebookId) { return { @@ -426,7 +426,7 @@ export async function publishJianshuArticle( } } - transport.reportProgress?.('create_note') + await transport.reportProgress?.('create_note') const note = await createNote(transport.fetchJson, notebookId, input.title) if (!note?.id) { return { @@ -442,10 +442,10 @@ export async function publishJianshuArticle( let publishOk = false try { - transport.reportProgress?.('upload_content_images') + await transport.reportProgress?.('upload_content_images') const processedContent = await processContentImages(transport, content) - transport.reportProgress?.('save_content') + await transport.reportProgress?.('save_content') await transport.fetchJson(`${JIANSHU_ORIGIN}/author/notes/${noteId}`, { method: 'PUT', headers: jianshuJsonHeaders(), @@ -458,11 +458,11 @@ export async function publishJianshuArticle( }) if (input.coverAssetUrl?.trim()) { - transport.reportProgress?.('upload_cover') + await transport.reportProgress?.('upload_cover') await attachCoverImage(transport, noteId, input.coverAssetUrl.trim()) } - transport.reportProgress?.('publish') + await transport.reportProgress?.('publish') const publicize = await transport .fetchJson(`${JIANSHU_ORIGIN}/author/notes/${noteId}/publicize`, { method: 'POST', diff --git a/packages/publisher-platforms/src/toutiao.ts b/packages/publisher-platforms/src/toutiao.ts index 3e84847..638535a 100644 --- a/packages/publisher-platforms/src/toutiao.ts +++ b/packages/publisher-platforms/src/toutiao.ts @@ -61,7 +61,9 @@ export interface ToutiaoPublishTransport { html: string, uploader: (sourceUrl: string) => Promise, ): Promise<{ html: string; uploaded?: Map }> - reportProgress?(stage: 'media_info' | 'upload_content_images' | 'upload_cover' | 'submit'): void + reportProgress?( + stage: 'media_info' | 'upload_content_images' | 'upload_cover' | 'submit', + ): void | Promise } export type ToutiaoPublishResult = @@ -236,7 +238,7 @@ export async function publishToutiaoArticle( input: ToutiaoPublishArticleInput, transport: ToutiaoPublishTransport, ): Promise { - transport.reportProgress?.('media_info') + await transport.reportProgress?.('media_info') const media = await fetchToutiaoMediaSnapshot(transport.fetchJson) if (!media?.platformUid || !media.screenName) { return { @@ -257,7 +259,7 @@ export async function publishToutiaoArticle( } } - transport.reportProgress?.('upload_content_images') + await transport.reportProgress?.('upload_content_images') const processed = await transport.uploadHtmlImages(normalizedHtml, async (sourceUrl) => uploadContentImage(transport, sourceUrl), ) @@ -275,7 +277,7 @@ export async function publishToutiaoArticle( } } - transport.reportProgress?.('upload_cover') + await transport.reportProgress?.('upload_cover') const cover = input.coverAssetUrl?.trim() ? await uploadCover(transport, input.coverAssetUrl.trim(), input.publishType) : null @@ -306,7 +308,7 @@ export async function publishToutiaoArticle( }), }) - transport.reportProgress?.('submit') + await transport.reportProgress?.('submit') const response: ToutiaoPublishResponse = await transport .fetchJson( 'https://mp.toutiao.com/mp/agw/article/publish?source=mp&type=article&aid=1231', diff --git a/server/internal/shared/swagger/descriptions.go b/server/internal/shared/swagger/descriptions.go index b769291..9121927 100644 --- a/server/internal/shared/swagger/descriptions.go +++ b/server/internal/shared/swagger/descriptions.go @@ -48,16 +48,22 @@ var routeDocs = map[string]routeDoc{ "POST /api/ops/admin-users/:id/reset-login-lock": {"解除登录锁定", "解除租户管理员账号的登录失败锁定。"}, // --- Ops:业务运营 --- - "GET /api/ops/kol/packages": {"KOL 套餐列表(Ops)", "Ops 查看 KOL 市场套餐。"}, - "GET /api/ops/kol/subscriptions": {"KOL 订阅列表(Ops)", "Ops 查看租户 KOL 订阅记录。"}, - "POST /api/ops/kol/subscriptions/manual-bind": {"手动绑定 KOL 订阅", "Ops 为租户手动绑定 KOL 订阅。"}, - "POST /api/ops/kol/subscriptions/:id/approve": {"审批 KOL 订阅", "Ops 审批指定 KOL 订阅。"}, - "POST /api/ops/kol/subscriptions/:id/revoke": {"撤销 KOL 订阅", "Ops 撤销指定 KOL 订阅。"}, - "GET /api/ops/audits": {"审计日志列表", "分页查询 Ops 审计日志。"}, - "GET /api/ops/jobs": {"任务列表(Ops)", "分页查询后台任务。"}, - "GET /api/ops/jobs/:source/:id": {"任务详情(Ops)", "读取指定来源与 ID 的任务详情。"}, - "POST /api/ops/jobs/:source/:id/retry": {"重试任务(Ops)", "对失败任务发起重试。"}, - "POST /api/ops/jobs/:source/:id/cancel": {"取消任务(Ops)", "取消指定后台任务。"}, + "GET /api/ops/kol/packages": {"KOL 套餐列表(Ops)", "Ops 查看 KOL 市场套餐。"}, + "GET /api/ops/kol/subscriptions": {"KOL 订阅列表(Ops)", "Ops 查看租户 KOL 订阅记录。"}, + "POST /api/ops/kol/subscriptions/manual-bind": {"手动绑定 KOL 订阅", "Ops 为租户手动绑定 KOL 订阅。"}, + "POST /api/ops/kol/subscriptions/:id/approve": {"审批 KOL 订阅", "Ops 审批指定 KOL 订阅。"}, + "POST /api/ops/kol/subscriptions/:id/revoke": {"撤销 KOL 订阅", "Ops 撤销指定 KOL 订阅。"}, + "GET /api/ops/audits": {"审计日志列表", "分页查询 Ops 审计日志。"}, + "GET /api/ops/jobs": {"任务列表(Ops)", "分页查询后台任务。"}, + "GET /api/ops/jobs/:source/:id": {"任务详情(Ops)", "读取指定来源与 ID 的任务详情。"}, + "POST /api/ops/jobs/:source/:id/retry": {"重试任务(Ops)", "对失败任务发起重试。"}, + "POST /api/ops/jobs/:source/:id/cancel": {"取消任务(Ops)", "取消指定后台任务。"}, + "GET /api/ops/media-supply/resources": {"媒介资源列表(Ops)", "Ops 分页查看媒介供应资源。"}, + "PUT /api/ops/media-supply/resources/:id/price": {"调整媒介资源价格(Ops)", "Ops 调整指定媒介资源的销售价格。"}, + "PUT /api/ops/media-supply/resources/:id/visibility": {"切换媒介资源可见性", "Ops 启用或隐藏指定媒介供应资源。"}, + "POST /api/ops/media-supply/sync-jobs": {"同步媒介资源(Ops)", "Ops 创建媒介供应资源同步任务。"}, + "GET /api/ops/media-supply/wallets": {"媒介钱包列表(Ops)", "Ops 查询租户媒介供应钱包余额。"}, + "POST /api/ops/media-supply/wallets/adjustments": {"调整媒介钱包(Ops)", "Ops 为租户创建媒介钱包余额调整。"}, // --- Ops:调度器 --- "GET /api/ops/scheduler/jobs": {"调度任务列表", "分页查询调度器任务。"}, @@ -159,12 +165,13 @@ var routeDocs = map[string]routeDoc{ "DELETE /api/desktop/accounts/:id": {"删除媒体账号(桌面)", "桌面客户端解绑账号,仅清理本地侧记录。"}, // --- Desktop:发布任务执行 --- - "POST /api/desktop/tasks/lease": {"领取发布任务", "桌面客户端从队列领取一批可执行任务,原子加锁。"}, - "POST /api/desktop/tasks/:id/lease": {"续锁/重领任务", "对指定任务再次加锁,常用于异常重试场景。"}, - "POST /api/desktop/tasks/:id/extend": {"延长任务锁", "执行较慢时延长锁定时间,避免任务被其他客户端抢走。"}, - "POST /api/desktop/tasks/:id/cancel": {"取消任务(桌面)", "桌面客户端在执行失败/用户中止时通知服务端取消。"}, - "POST /api/desktop/tasks/:id/result": {"上报任务结果", "桌面客户端将发布结果(成功/失败/截图/链接)回传服务端。"}, - "POST /api/desktop/publish-tasks/:id/retry": {"重试发布任务", "对失败的发布任务发起重试。"}, + "POST /api/desktop/tasks/lease": {"领取发布任务", "桌面客户端从队列领取一批可执行任务,原子加锁。"}, + "POST /api/desktop/tasks/:id/lease": {"续锁/重领任务", "对指定任务再次加锁,常用于异常重试场景。"}, + "POST /api/desktop/tasks/:id/extend": {"延长任务锁", "执行较慢时延长锁定时间,避免任务被其他客户端抢走。"}, + "POST /api/desktop/tasks/:id/publish-submit-marker": {"标记发布已进入提交阶段", "桌面客户端在执行不可逆平台提交前写入幂等保护标记,避免恢复流程自动重复发文。"}, + "POST /api/desktop/tasks/:id/cancel": {"取消任务(桌面)", "桌面客户端在执行失败/用户中止时通知服务端取消。"}, + "POST /api/desktop/tasks/:id/result": {"上报任务结果", "桌面客户端将发布结果(成功/失败/截图/链接)回传服务端。"}, + "POST /api/desktop/publish-tasks/:id/retry": {"重试发布任务", "对失败的发布任务发起重试。"}, // --- Desktop:监控采集任务 --- "POST /api/desktop/monitoring/tasks/lease": {"领取监控任务", "桌面客户端领取 GEO 监控的爬取任务(AI 平台问答抓取)。"}, @@ -185,6 +192,23 @@ var routeDocs = map[string]routeDoc{ // --- Tenant:媒体 --- "GET /api/tenant/media/platforms": {"媒体平台列表", "返回当前租户可用的媒体平台元数据(图标、能力开关等)。"}, + // --- Tenant:媒介供应 --- + "GET /api/tenant/media-supply/resources": {"媒介资源列表", "查询当前 Workspace 可购买的媒介供应资源。"}, + "GET /api/tenant/media-supply/resources/by-ids": {"按 ID 查询媒介资源", "按资源 ID 集合查询媒介供应资源详情。"}, + "GET /api/tenant/media-supply/resources/search-options": {"媒介资源筛选项", "返回媒介供应资源列表的筛选条件与可选项。"}, + "PUT /api/tenant/media-supply/resources/:id/price": {"调整媒介资源售价", "租户侧调整指定媒介资源的销售价格。"}, + "POST /api/tenant/media-supply/sync-jobs": {"同步媒介资源", "创建媒介供应资源同步任务。"}, + "POST /api/tenant/media-supply/orders": {"创建媒介投放订单", "为选定媒介资源创建投放订单。"}, + "GET /api/tenant/media-supply/orders": {"媒介订单列表", "分页查询当前 Workspace 的媒介投放订单。"}, + "GET /api/tenant/media-supply/orders/:id": {"媒介订单详情", "读取指定媒介投放订单详情。"}, + "POST /api/tenant/media-supply/orders/sync-backlinks": {"同步媒介订单链接", "从供应商侧同步媒介订单的发布链接。"}, + "GET /api/tenant/media-supply/wallet": {"媒介钱包状态", "读取当前 Workspace 的媒介供应钱包余额与状态。"}, + "GET /api/tenant/media-supply/wallet/ledgers": {"媒介钱包流水", "分页查询媒介供应钱包收支流水。"}, + "POST /api/tenant/media-supply/wallet/adjustments": {"调整媒介钱包", "创建媒介钱包余额调整记录。"}, + "GET /api/tenant/media-supply/supplier-session": {"供应商会话状态", "查询媒介供应商登录会话状态。"}, + "PUT /api/tenant/media-supply/supplier-session": {"导入供应商会话", "导入或更新媒介供应商登录会话。"}, + "POST /api/tenant/media-supply/supplier-session/login": {"登录供应商账号", "使用已配置账号登录媒介供应商并刷新会话。"}, + // --- Tenant:内容合规 --- "GET /api/tenant/compliance/runtime-status": {"合规运行状态", "返回当前租户内容合规模块是否启用、执行模式、词库版本与检测预算。"}, "POST /api/tenant/compliance/check": {"纯文本合规检测", "对标题/正文纯文本执行一次合规检测,返回聚合 check_record 与违规项。"}, diff --git a/server/internal/tenant/app/desktop_task_service.go b/server/internal/tenant/app/desktop_task_service.go index 840a245..1a14d03 100644 --- a/server/internal/tenant/app/desktop_task_service.go +++ b/server/internal/tenant/app/desktop_task_service.go @@ -722,8 +722,9 @@ func (s *DesktopTaskService) Extend(ctx context.Context, client *repository.Desk // MarkPublishSubmitStarted records a durable intent marker right before the client performs the // irreversible platform submit POST. Lease-recovery uses it to avoid silently re-posting a // non-idempotent article: once submit may have started, recovery routes the task to 'unknown' -// (manual reconcile) instead of auto-requeueing it. Best-effort — a stale/lost lease is a no-op -// and must never block the in-flight publish. +// (manual reconcile) instead of auto-requeueing it. The client must receive a +// positive marker write before it performs the platform submit; a stale/lost +// lease is rejected so the client can stop before the irreversible request. func (s *DesktopTaskService) MarkPublishSubmitStarted(ctx context.Context, client *repository.DesktopClient, taskID uuid.UUID, leaseToken string) error { if s == nil || s.pool == nil || client == nil { return nil @@ -732,7 +733,7 @@ func (s *DesktopTaskService) MarkPublishSubmitStarted(ctx context.Context, clien if leaseToken == "" { return response.ErrBadRequest(40001, "invalid_params", "lease_token is required") } - if _, err := s.pool.Exec(ctx, ` + tag, err := s.pool.Exec(ctx, ` UPDATE desktop_tasks SET publish_submit_started_at = COALESCE(publish_submit_started_at, NOW()), updated_at = NOW() @@ -741,9 +742,13 @@ func (s *DesktopTaskService) MarkPublishSubmitStarted(ctx context.Context, clien AND kind = 'publish' AND status = 'in_progress' AND lease_token_hash = $3 - `, taskID, client.ID, HashDesktopClientToken(leaseToken)); err != nil { + `, taskID, client.ID, HashDesktopClientToken(leaseToken)) + if err != nil { return response.ErrInternal(50200, "desktop_task_submit_marker_failed", "failed to mark publish submit started") } + if tag.RowsAffected() == 0 { + return response.ErrConflict(40983, "desktop_task_lease_invalid", "desktop task lease token is invalid or expired") + } return nil } diff --git a/server/internal/tenant/app/media_supply_test.go b/server/internal/tenant/app/media_supply_test.go index b163231..9b8fd3d 100644 --- a/server/internal/tenant/app/media_supply_test.go +++ b/server/internal/tenant/app/media_supply_test.go @@ -558,17 +558,17 @@ func TestMeijiequanUpstreamThrottleSerializesRequests(t *testing.T) { redis := goredis.NewClient(&goredis.Options{Addr: server.Addr()}) defer redis.Close() - client := NewMeijiequanClient(redis, config.MeijiequanConfig{UpstreamMinInterval: 25 * time.Millisecond}) + client := NewMeijiequanClient(redis, config.MeijiequanConfig{UpstreamMinInterval: 80 * time.Millisecond}) ctx := context.Background() - if err := client.waitForUpstreamTurn(ctx, 25*time.Millisecond); err != nil { + if err := client.waitForUpstreamTurn(ctx, 80*time.Millisecond); err != nil { t.Fatalf("first wait failed: %v", err) } start := time.Now() - if err := client.waitForUpstreamTurn(ctx, 25*time.Millisecond); err != nil { + if err := client.waitForUpstreamTurn(ctx, 80*time.Millisecond); err != nil { t.Fatalf("second wait failed: %v", err) } - if elapsed := time.Since(start); elapsed < 20*time.Millisecond { + if elapsed := time.Since(start); elapsed < 50*time.Millisecond { t.Fatalf("expected serialized upstream wait, elapsed %s", elapsed) } }