From 0885a46e5894fbefd2ffa1c3c7fd30fb5d11383a Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 13:53:40 +0800 Subject: [PATCH 01/20] =?UTF-8?q?=E9=87=8D=E6=9E=84=EF=BC=9A=E5=88=86?= =?UTF-8?q?=E7=A6=BB=E5=8F=91=E5=B8=83=E5=B9=B3=E5=8F=B0=E4=B8=8E=E6=89=A7?= =?UTF-8?q?=E8=A1=8C=E6=96=B9=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 3 + app/core/config.py | 3 + app/db/database.py | 134 ++++++++++++++++++++++++++- app/models/task.py | 6 +- app/services/auto_publish_service.py | 18 ++-- app/services/publish_adapters.py | 12 +-- app/services/publish_domain.py | 42 +++++++++ app/services/publish_scheduler.py | 14 +-- app/services/publish_service.py | 57 +++++++----- 9 files changed, 235 insertions(+), 54 deletions(-) create mode 100644 app/services/publish_domain.py diff --git a/.env.example b/.env.example index 047a9e8..34fc951 100644 --- a/.env.example +++ b/.env.example @@ -66,6 +66,9 @@ OPENCLI_HOST_BRIDGE_URL=http://host.docker.internal:8765 # 默认只导出本地发布包,不调用真实平台 API,不保存账号、密码、cookie 或 token。 PUBLISH_SCHEDULER_ENABLED=true PUBLISH_SCHEDULER_INTERVAL_SECONDS=60 +PUBLISH_DEFAULT_MODE=opencli_publish +PUBLISH_JOB_STALE_MINUTES=30 +# 已废弃:仅为旧环境兼容保留,不再覆盖发布任务的目标平台。 PUBLISH_SCHEDULER_DEFAULT_PLATFORM=manual_export PUBLISH_SCHEDULER_MAX_RETRY_COUNT=3 PUBLISH_SCHEDULER_EXPORT_DIR= diff --git a/app/core/config.py b/app/core/config.py index 6d89e08..62872f0 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -163,7 +163,10 @@ class Settings: opencli_host_bridge_url: str = _env("OPENCLI_HOST_BRIDGE_URL", "") publish_scheduler_enabled: bool = _env_bool("PUBLISH_SCHEDULER_ENABLED", True) publish_scheduler_interval_seconds: int = int(_env("PUBLISH_SCHEDULER_INTERVAL_SECONDS", "60")) + publish_default_mode: str = _env("PUBLISH_DEFAULT_MODE", "opencli_publish") + # 已废弃:仅保留读取能力,旧值不再覆盖 publish_jobs.platform。 publish_scheduler_default_platform: str = _env("PUBLISH_SCHEDULER_DEFAULT_PLATFORM", "manual_export") + publish_job_stale_minutes: int = int(_env("PUBLISH_JOB_STALE_MINUTES", "30")) publish_scheduler_max_retry_count: int = int(_env("PUBLISH_SCHEDULER_MAX_RETRY_COUNT", "3")) publish_scheduler_export_dir: Path = _env_path( "PUBLISH_SCHEDULER_EXPORT_DIR", diff --git a/app/db/database.py b/app/db/database.py index 2fb2849..e667a05 100644 --- a/app/db/database.py +++ b/app/db/database.py @@ -1,3 +1,4 @@ +import json import sqlite3 from datetime import datetime from collections.abc import Iterator @@ -315,6 +316,10 @@ def _create_indexes(connection: sqlite3.Connection) -> None: "CREATE INDEX IF NOT EXISTS idx_publish_jobs_status_platform_created ON publish_jobs(status, platform, created_at)", "CREATE INDEX IF NOT EXISTS idx_publish_jobs_task_output ON publish_jobs(task_id, output_clip_id)", "CREATE INDEX IF NOT EXISTS idx_publish_jobs_status_scheduled ON publish_jobs(status, scheduled_at)", + """CREATE UNIQUE INDEX IF NOT EXISTS uq_publish_jobs_active_clip_platform_mode + ON publish_jobs(output_clip_id, platform, publish_mode) + WHERE status NOT IN ('PUBLISHED', 'EXPORTED', 'CANCELLED') + AND output_clip_id IS NOT NULL AND output_clip_id <> ''""", # OAuth state 过期清理 "CREATE INDEX IF NOT EXISTS idx_oauth_states_expires ON oauth_states(expires_at)", ] @@ -710,6 +715,8 @@ def _migrate_publish_jobs_table(connection: sqlite3.Connection) -> None: connection.execute(statement) columns = _get_table_columns(connection, "publish_jobs") + _backup_publish_database_before_data_migration(connection) + _migrate_publish_platform_and_mode_values(connection) if {"clip_id", "output_clip_id"}.issubset(columns): connection.execute("UPDATE publish_jobs SET clip_id = output_clip_id WHERE clip_id IS NULL OR clip_id = ''") if {"video_path", "video_file_path"}.issubset(columns): @@ -738,6 +745,7 @@ def _migrate_publish_jobs_table(connection: sqlite3.Connection) -> None: UPDATE publish_jobs SET status = 'SCHEDULED' WHERE status IN ('ready', 'scheduled'); UPDATE publish_jobs SET status = 'PUBLISHING' WHERE status = 'publishing'; UPDATE publish_jobs SET status = 'PUBLISHED' WHERE status = 'published'; + UPDATE publish_jobs SET status = 'EXPORTED' WHERE status = 'exported'; UPDATE publish_jobs SET status = 'FAILED' WHERE status = 'failed'; UPDATE publish_jobs SET status = 'CANCELLED' WHERE status = 'cancelled'; UPDATE publish_jobs SET status = 'NEED_REVIEW' WHERE status = 'need_review'; @@ -748,10 +756,134 @@ def _migrate_publish_jobs_table(connection: sqlite3.Connection) -> None: SET status = 'SCHEDULED' WHERE status IS NULL OR status = '' OR status NOT IN ( 'DRAFT', 'SCHEDULED', 'WAITING', 'PUBLISHING', - 'PUBLISHED', 'FAILED', 'CANCELLED', 'NEED_REVIEW' + 'PUBLISHED', 'EXPORTED', 'FAILED', 'CANCELLED', 'NEED_REVIEW' ); """ ) + _cancel_duplicate_active_publish_jobs(connection) + + +def _backup_publish_database_before_data_migration(connection: sqlite3.Connection) -> None: + """仅在发现旧值或有效重复任务时创建一次迁移前备份。""" + legacy_count = connection.execute( + """ + SELECT COUNT(*) + FROM publish_jobs + WHERE platform NOT IN ('douyin', 'bilibili') + OR publish_mode NOT IN ('opencli_publish', 'manual_export', 'api_publish', 'local_browser') + """ + ).fetchone()[0] + duplicate_count = connection.execute( + """ + SELECT COUNT(*) FROM ( + SELECT output_clip_id, platform, publish_mode + FROM publish_jobs + WHERE status NOT IN ('PUBLISHED', 'EXPORTED', 'CANCELLED') + GROUP BY output_clip_id, platform, publish_mode + HAVING COUNT(*) > 1 + ) + """ + ).fetchone()[0] + if not legacy_count and not duplicate_count: + return + database_path = settings.database_path + if not database_path.exists(): + return + backup_dir = settings.data_dir / "backups" + backup_dir.mkdir(parents=True, exist_ok=True) + timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") + backup_path = backup_dir / f"workflow-before-publish-migration-{timestamp}.sqlite3" + if backup_path.exists(): + return + with sqlite3.connect(str(backup_path)) as backup_connection: + connection.backup(backup_connection) + + +def _provider_target_platform(raw_value: str | None) -> str: + try: + payload = json.loads(raw_value or "{}") + except (json.JSONDecodeError, TypeError): + return "" + if not isinstance(payload, dict): + return "" + target = str(payload.get("target_platform") or "").strip().lower() + return target if target in {"douyin", "bilibili"} else "" + + +def _migrate_publish_platform_and_mode_values(connection: sqlite3.Connection) -> None: + default_mode = str(settings.publish_default_mode or "opencli_publish").strip().lower() + if default_mode not in {"opencli_publish", "manual_export", "api_publish", "local_browser"}: + default_mode = "opencli_publish" + rows = connection.execute( + """ + SELECT publish_jobs.id, publish_jobs.platform, publish_jobs.publish_mode, + publish_jobs.provider_response, tasks.platform AS task_platform + FROM publish_jobs + LEFT JOIN tasks ON tasks.id = publish_jobs.task_id + WHERE publish_jobs.platform NOT IN ('douyin', 'bilibili') + OR publish_jobs.publish_mode NOT IN ('opencli_publish', 'manual_export', 'api_publish', 'local_browser') + """ + ).fetchall() + now = datetime.now().astimezone().isoformat(timespec="seconds") + for row in rows: + platform = _provider_target_platform(row["provider_response"]) + if not platform: + task_platform = str(row["task_platform"] or "").strip().lower() + platform = task_platform if task_platform in {"douyin", "bilibili"} else "douyin" + old_platform = str(row["platform"] or "").strip().lower() + old_mode = str(row["publish_mode"] or "").strip().lower() + mode = old_mode if old_mode in {"opencli_publish", "manual_export", "api_publish", "local_browser"} else default_mode + if old_platform in {"manual_export", "local_browser"}: + mode = default_mode + connection.execute( + "UPDATE publish_jobs SET platform = ?, publish_mode = ?, updated_at = ? WHERE id = ?", + (platform, mode, now, row["id"]), + ) + + +def _cancel_duplicate_active_publish_jobs(connection: sqlite3.Connection) -> None: + groups = connection.execute( + """ + SELECT output_clip_id, platform, publish_mode + FROM publish_jobs + WHERE status NOT IN ('PUBLISHED', 'EXPORTED', 'CANCELLED') + GROUP BY output_clip_id, platform, publish_mode + HAVING COUNT(*) > 1 + """ + ).fetchall() + now = datetime.now().astimezone().isoformat(timespec="seconds") + for group in groups: + rows = connection.execute( + """ + SELECT id, provider_response + FROM publish_jobs + WHERE output_clip_id = ? AND platform = ? AND publish_mode = ? + AND status NOT IN ('PUBLISHED', 'EXPORTED', 'CANCELLED') + ORDER BY COALESCE(NULLIF(updated_at, ''), created_at) DESC, created_at DESC, id DESC + """, + (group["output_clip_id"], group["platform"], group["publish_mode"]), + ).fetchall() + for duplicate in rows[1:]: + migration_payload = { + "migration_reason": "duplicate_active_publish_job", + "message": "迁移时发现同一切片、平台和执行方式的重复未发布任务,已保留最新一条。", + "previous_provider_response": duplicate["provider_response"] or "", + } + connection.execute( + """ + UPDATE publish_jobs + SET status = 'CANCELLED', error_code = 'migration_duplicate_cancelled', + error_message = ?, last_error = ?, provider_response = ?, updated_at = ? + WHERE id = ? + """, + ( + migration_payload["message"], + migration_payload["message"], + json.dumps(migration_payload, ensure_ascii=False), + now, + duplicate["id"], + ), + ) def _migrate_workflow_jobs_table(connection: sqlite3.Connection) -> None: diff --git a/app/models/task.py b/app/models/task.py index e71bb48..32fd745 100644 --- a/app/models/task.py +++ b/app/models/task.py @@ -144,9 +144,9 @@ class PublishAccountCreate(BaseModel): class PublishJobCreate(BaseModel): task_id: str = Field(..., min_length=1, max_length=80) output_clip_id: str = Field(..., min_length=1, max_length=80) - platform: Literal["douyin", "bilibili", "manual_export", "local_browser"] + platform: Literal["douyin", "bilibili"] account_id: Optional[str] = Field(default="", max_length=80) - publish_mode: Literal["draft", "manual_review", "manual_export", "local_browser", "api_publish", "opencli_publish"] = "manual_review" + publish_mode: Literal["manual_export", "local_browser", "api_publish", "opencli_publish"] = "opencli_publish" video_source: Literal["original", "subtitled"] = "original" title: str = Field(..., min_length=1, max_length=120) description: Optional[str] = Field(default="", max_length=2000) @@ -194,7 +194,7 @@ class PublishBatchJobCreate(BaseModel): output_clip_ids: list[str] = Field(default_factory=list) platform: Literal["douyin", "bilibili"] account_id: Optional[str] = Field(default="", max_length=80) - publish_mode: Literal["draft", "manual_review"] = "manual_review" + publish_mode: Literal["manual_export", "local_browser", "api_publish", "opencli_publish"] = "opencli_publish" video_source: Literal["original", "subtitled"] = "original" title_prefix: Optional[str] = Field(default="", max_length=80) description: Optional[str] = Field(default="", max_length=2000) diff --git a/app/services/auto_publish_service.py b/app/services/auto_publish_service.py index 4085513..a17bb41 100644 --- a/app/services/auto_publish_service.py +++ b/app/services/auto_publish_service.py @@ -8,6 +8,7 @@ from app.core.config import settings from app.db.database import get_connection from app.services.publish_service import DEFAULT_BILIBILI_TID, get_publish_job +from app.services.publish_domain import validate_publish_mode, validate_target_platform from app.services.task_service import _now_iso @@ -31,17 +32,18 @@ def create_auto_publish_jobs(task: dict, scheduled_items: list[dict]) -> dict: for item in scheduled_items: output_clip = item["output_clip"] metadata = item["metadata"] - target_platform = metadata["platform"] - platform = settings.publish_scheduler_default_platform or target_platform + platform = validate_target_platform(metadata["platform"]) + publish_mode = validate_publish_mode(settings.publish_default_mode) existing = connection.execute( """ SELECT id FROM publish_jobs - WHERE output_clip_id = ? AND platform = ? AND publish_mode = 'manual_export' + WHERE output_clip_id = ? AND platform = ? AND publish_mode = ? + AND status NOT IN ('PUBLISHED', 'EXPORTED', 'CANCELLED') ORDER BY created_at DESC LIMIT 1 """, - (output_clip["id"], platform), + (output_clip["id"], platform, publish_mode), ).fetchone() if existing: skipped_ids.append(existing["id"]) @@ -52,12 +54,13 @@ def create_auto_publish_jobs(task: dict, scheduled_items: list[dict]) -> dict: job_id = uuid4().hex[:12] provider_response = { "source": "auto_pipeline", - "target_platform": target_platform, + "target_platform": platform, "metadata_source": metadata.get("source") or "", "metadata_error": metadata.get("error") or "", "cover_text": metadata.get("cover_text") or "", "risk_flags": metadata.get("risk_flags") or [], - "note": "全自动流水线只创建待发送任务,发布时间在发送中心统一设置。", + "publish_mode": publish_mode, + "note": "全自动流水线已直接创建最终发布任务,可在发送中心设置排期。", } connection.execute( """ @@ -70,7 +73,7 @@ def create_auto_publish_jobs(task: dict, scheduled_items: list[dict]) -> dict: status, audit_status, error_message, last_error, provider_response, publish_result, created_at, updated_at ) - VALUES (?, ?, ?, ?, NULL, ?, 'manual_export', 'original', ?, ?, ?, ?, ?, ?, ?, ?, ?, 'public', + VALUES (?, ?, ?, ?, NULL, ?, ?, 'original', ?, ?, ?, ?, ?, ?, ?, ?, ?, 'public', 'auto', 0, 1, ?, 'original', '', '', ?, ?, 'not_submitted', '', '', ?, '', ?, ?) """, ( @@ -79,6 +82,7 @@ def create_auto_publish_jobs(task: dict, scheduled_items: list[dict]) -> dict: output_clip["id"], output_clip["id"], platform, + publish_mode, output_clip.get("output_file_path") or "", output_clip.get("output_file_path") or "", metadata.get("title") or "精彩片段", diff --git a/app/services/publish_adapters.py b/app/services/publish_adapters.py index 6bd49c8..1a69ce9 100644 --- a/app/services/publish_adapters.py +++ b/app/services/publish_adapters.py @@ -173,12 +173,12 @@ def publish(self, job: dict[str, Any]) -> PublishResult: def publisher_for_job(job: dict[str, Any]) -> BasePublisher: - platform = str(job.get("platform") or "").strip().lower() publish_mode = str(job.get("publish_mode") or "").strip().lower() - if platform == "local_browser" or publish_mode == "local_browser": + if publish_mode == "local_browser": return LocalBrowserPublisher() - if platform == "manual_export" or publish_mode == "manual_export": + if publish_mode == "manual_export": return ManualExportPublisher() - if settings.publish_scheduler_default_platform == "manual_export": - return ManualExportPublisher() - return ManualExportPublisher() + raise PublishValidationError( + f"publish_mode={publish_mode or '(empty)'} 不能由本地发布包适配器执行", + "unsupported_publish_mode", + ) diff --git a/app/services/publish_domain.py b/app/services/publish_domain.py new file mode 100644 index 0000000..6c3e38a --- /dev/null +++ b/app/services/publish_domain.py @@ -0,0 +1,42 @@ +"""发布领域常量:目标平台、执行方式与状态。""" + +TARGET_PLATFORMS = { + "douyin": "抖音", + "bilibili": "B站", +} + +PUBLISH_MODES = { + "opencli_publish": "opencli 网页发送", + "manual_export": "本地发布包导出", + "api_publish": "平台 API 发布", + "local_browser": "本地浏览器发布(未实现)", +} + +PUBLISH_STATUSES = { + "DRAFT", + "WAITING", + "SCHEDULED", + "PUBLISHING", + "PUBLISHED", + "EXPORTED", + "FAILED", + "CANCELLED", + "NEED_REVIEW", +} + +TERMINAL_PUBLISH_STATUSES = {"PUBLISHED", "EXPORTED", "CANCELLED"} +ACTIVE_PUBLISH_STATUSES = PUBLISH_STATUSES - TERMINAL_PUBLISH_STATUSES + + +def validate_target_platform(platform: str) -> str: + value = (platform or "").strip().lower() + if value not in TARGET_PLATFORMS: + raise ValueError("目标平台只能是 douyin 或 bilibili") + return value + + +def validate_publish_mode(publish_mode: str) -> str: + value = (publish_mode or "").strip().lower() + if value not in PUBLISH_MODES: + raise ValueError("不支持的发布执行方式") + return value diff --git a/app/services/publish_scheduler.py b/app/services/publish_scheduler.py index c9b8589..d22e54f 100644 --- a/app/services/publish_scheduler.py +++ b/app/services/publish_scheduler.py @@ -9,21 +9,10 @@ from app.core.config import settings from app.db.database import get_connection, init_db from app.services.publish_adapters import PublishValidationError, publisher_for_job +from app.services.publish_domain import PUBLISH_STATUSES from app.services.task_log_service import append_task_log -PUBLISH_STATUSES = { - "DRAFT", - "SCHEDULED", - "WAITING", - "PUBLISHING", - "PUBLISHED", - "FAILED", - "CANCELLED", - "NEED_REVIEW", -} - - def now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") @@ -547,6 +536,7 @@ def queue_snapshot(task_id: str | None = None) -> dict[str, Any]: "pending": by_status["SCHEDULED"] + by_status["WAITING"], "publishing": by_status["PUBLISHING"], "published": by_status["PUBLISHED"], + "exported": by_status["EXPORTED"], "failed": by_status["FAILED"], "need_review": by_status["NEED_REVIEW"], "cancelled": by_status["CANCELLED"], diff --git a/app/services/publish_service.py b/app/services/publish_service.py index e921901..6b9ee14 100644 --- a/app/services/publish_service.py +++ b/app/services/publish_service.py @@ -33,14 +33,12 @@ DouyinPublishProvider, PublishProviderError, ) +from app.services.publish_domain import PUBLISH_MODES, TARGET_PLATFORMS from app.services.storage_service import get_artifact_paths, resolve_video_file_path from app.services.video_cut_service import ensure_ffmpeg_available, sanitize_filename_part, summarize_stderr -PLATFORM_LABELS = { - "douyin": "抖音", - "bilibili": "B站", -} +PLATFORM_LABELS = TARGET_PLATFORMS STATUS_LABELS = { "draft": "草稿", @@ -79,6 +77,7 @@ PUBLISH_STATUS_WAITING = "WAITING" PUBLISH_STATUS_PUBLISHING = "PUBLISHING" PUBLISH_STATUS_PUBLISHED = "PUBLISHED" +PUBLISH_STATUS_EXPORTED = "EXPORTED" PUBLISH_STATUS_FAILED = "FAILED" PUBLISH_STATUS_CANCELLED = "CANCELLED" PUBLISH_STATUS_NEED_REVIEW = "NEED_REVIEW" @@ -94,19 +93,7 @@ "need_review": PUBLISH_STATUS_NEED_REVIEW, } -PLATFORM_LABELS.update( - { - "manual_export": "发布包导出", - "local_browser": "本地浏览器", - } -) - -PUBLISH_MODE_LABELS.update( - { - "manual_export": "手动发布包导出", - "local_browser": "本地浏览器发布", - } -) +PUBLISH_MODE_LABELS.update(PUBLISH_MODES) STATUS_LABELS = { PUBLISH_STATUS_DRAFT: "草稿", @@ -114,6 +101,7 @@ PUBLISH_STATUS_WAITING: "等待处理", PUBLISH_STATUS_PUBLISHING: "发布中", PUBLISH_STATUS_PUBLISHED: "已发布", + PUBLISH_STATUS_EXPORTED: "已导出发布包", PUBLISH_STATUS_FAILED: "发送失败", PUBLISH_STATUS_CANCELLED: "已取消", PUBLISH_STATUS_NEED_REVIEW: "需人工复核", @@ -131,6 +119,7 @@ PUBLISH_STATUS_WAITING: "amber", PUBLISH_STATUS_PUBLISHING: "purple", PUBLISH_STATUS_PUBLISHED: "green", + PUBLISH_STATUS_EXPORTED: "blue", PUBLISH_STATUS_FAILED: "red", PUBLISH_STATUS_CANCELLED: "amber", PUBLISH_STATUS_NEED_REVIEW: "amber", @@ -998,6 +987,7 @@ def _find_opencli_job(output_clip_id: str, platform: str) -> dict | None: """ SELECT * FROM publish_jobs WHERE output_clip_id = ? AND platform = ? AND publish_mode = 'opencli_publish' + AND status NOT IN ('PUBLISHED', 'EXPORTED', 'CANCELLED') ORDER BY created_at DESC LIMIT 1 """, @@ -1006,6 +996,22 @@ def _find_opencli_job(output_clip_id: str, platform: str) -> dict | None: return _normalize_job(row) if row else None +def _find_active_publish_job(output_clip_id: str, platform: str) -> dict | None: + """查找任意执行方式的有效任务,避免刷新队列改变用户已选择的执行方式。""" + with get_connection() as connection: + row = connection.execute( + """ + SELECT * FROM publish_jobs + WHERE output_clip_id = ? AND platform = ? + AND status NOT IN ('PUBLISHED', 'EXPORTED', 'CANCELLED') + ORDER BY COALESCE(NULLIF(updated_at, ''), created_at) DESC + LIMIT 1 + """, + (output_clip_id, platform), + ).fetchone() + return _normalize_job(row) if row else None + + def _batch_find_opencli_jobs(output_clip_ids: list[str]) -> dict[str, dict[str, dict]]: """一次查询获得所有 output_clip 在各平台的 opencli 发布任务。 @@ -1122,10 +1128,10 @@ def ensure_cover_for_item() -> dict: return cover_state["cover"] or {} for platform in PLATFORM_LABELS: - existing_job = _find_opencli_job(item["output_clip_id"], platform) + existing_job = _find_active_publish_job(item["output_clip_id"], platform) if existing_job: skipped += 1 - if not existing_job.get("cover_file_path"): + if existing_job.get("publish_mode") == "opencli_publish" and not existing_job.get("cover_file_path"): cover = ensure_cover_for_item() if cover.get("cover_file_path"): _update_job_cover(existing_job["id"], cover) @@ -1448,7 +1454,10 @@ def create_publish_job(payload: PublishJobCreate) -> dict: cover_file_path = (payload.cover_file_path or "").strip() cover_time_seconds = float(payload.cover_time_seconds or 0) cover_mode = payload.cover_mode - provider_payload = "真实发布任务已创建,等待执行。" if payload.publish_mode == "api_publish" else "本地发布任务已创建,等待人工确认。" + existing = _find_active_publish_job(payload.output_clip_id, payload.platform) + if existing and existing.get("publish_mode") == payload.publish_mode: + return {"status": "exists", "message": "同一切片、平台和执行方式已有有效任务。", "job": existing} + provider_payload = "真实发布任务已创建,等待执行。" if not cover_file_path: try: auto_cover = _generate_default_publish_cover( @@ -1474,11 +1483,11 @@ def create_publish_job(payload: PublishJobCreate) -> dict: job_id = uuid4().hex[:12] now = _now_iso() - status = PUBLISH_STATUS_DRAFT if payload.publish_mode == "draft" else PUBLISH_STATUS_WAITING + status = PUBLISH_STATUS_WAITING if (payload.scheduled_at or "").strip(): status = PUBLISH_STATUS_SCHEDULED - if payload.publish_mode == "api_publish": - status = PUBLISH_STATUS_PUBLISHING + if payload.publish_mode == "api_publish" and not (payload.scheduled_at or "").strip(): + status = PUBLISH_STATUS_WAITING with get_connection() as connection: connection.execute( """ @@ -1521,8 +1530,6 @@ def create_publish_job(payload: PublishJobCreate) -> dict: ) connection.commit() - if payload.publish_mode == "api_publish" and config and account: - return _execute_publish_job(job_id, config=config, account=account, video_path=resolved_video_path) return {"status": "ok", "message": "发布任务已创建。", "job": get_publish_job(job_id)} From 831d63368a60164fd08e1d2c646c81c74936b405 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 14:00:23 +0800 Subject: [PATCH 02/20] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E6=8E=A5?= =?UTF-8?q?=E9=80=9A=E6=8E=92=E6=9C=9F=E8=B0=83=E5=BA=A6=E4=B8=8E=20opencl?= =?UTF-8?q?i=20=E7=9C=9F=E5=AE=9E=E5=8F=91=E9=80=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/db/database.py | 3 + app/routers/publish.py | 15 ++- app/services/publish_executor.py | 57 +++++++++ app/services/publish_scheduler.py | 195 +++++++++++++++++++++++------- app/services/publish_service.py | 48 +++----- 5 files changed, 241 insertions(+), 77 deletions(-) create mode 100644 app/services/publish_executor.py diff --git a/app/db/database.py b/app/db/database.py index e667a05..adc4b52 100644 --- a/app/db/database.py +++ b/app/db/database.py @@ -789,6 +789,9 @@ def _backup_publish_database_before_data_migration(connection: sqlite3.Connectio database_path = settings.database_path if not database_path.exists(): return + # sqlite backup 不能在源连接持有写事务时执行;这里只提交此前的建表/加列操作, + # 真正的数据修复尚未开始,因此备份仍然是数据迁移前快照。 + connection.commit() backup_dir = settings.data_dir / "backups" backup_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d-%H%M%S") diff --git a/app/routers/publish.py b/app/routers/publish.py index 4d07d4c..6e39787 100644 --- a/app/routers/publish.py +++ b/app/routers/publish.py @@ -17,7 +17,7 @@ PublishSendStart, ) from app.services import publish_service -from app.services.publish_scheduler import PublishScheduler, queue_snapshot +from app.services.publish_scheduler import PublishScheduler, queue_snapshot, scheduler_health router = APIRouter(prefix="/api/publish", tags=["publish"]) @@ -93,7 +93,14 @@ async def get_publish_queue_snapshot(task_id: str | None = None) -> dict: @router.post("/scheduler/run-once") async def run_publish_scheduler_once() -> dict: - return PublishScheduler().run_once() + import asyncio + + return await asyncio.to_thread(PublishScheduler().run_once) + + +@router.get("/scheduler/health") +async def get_publish_scheduler_health() -> dict: + return scheduler_health() @router.post("/queue/refresh") @@ -189,8 +196,10 @@ async def retry_publish_job(job_id: str) -> dict: @router.post("/jobs/{job_id}/publish-now") async def publish_job_now(job_id: str) -> dict: + import asyncio + try: - return PublishScheduler().publish_now(job_id) + return await asyncio.to_thread(PublishScheduler().publish_now, job_id) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc diff --git a/app/services/publish_executor.py b/app/services/publish_executor.py new file mode 100644 index 0000000..9bc18de --- /dev/null +++ b/app/services/publish_executor.py @@ -0,0 +1,57 @@ +"""统一发布执行入口。状态抢占由 PublishScheduler 负责。""" + +from __future__ import annotations + +from typing import Any, Callable + +from app.services.publish_adapters import ManualExportPublisher, PublishValidationError + + +def execute_publish_job( + job_id: str, + force: bool = False, + *, + runner: Callable[[list[str]], Any] | None = None, +) -> dict[str, Any]: + """按 publish_mode 执行任务,禁止未知类型静默降级。""" + from app.services import publish_service + from app.services.publish_scheduler import get_publish_job_raw + + job = get_publish_job_raw(job_id) + if not job: + raise PublishValidationError("发布任务不存在", "publish_job_not_found") + publish_mode = str(job.get("publish_mode") or "").strip().lower() + + if publish_mode == "opencli_publish": + result = publish_service.execute_opencli_send_job(job_id, runner=runner) + return { + "status": "published" if result.get("status") == "ok" else "failed", + "job_id": job_id, + "message": result.get("message") or "", + "job": result.get("job"), + } + if publish_mode == "manual_export": + result = ManualExportPublisher().publish(job) + return { + "status": "exported", + "job_id": job_id, + "payload": result.payload, + "remote_video_id": result.remote_video_id, + } + if publish_mode == "api_publish": + result = publish_service.execute_api_publish_job(job_id) + return { + "status": "published" if result.get("status") == "ok" else "failed", + "job_id": job_id, + "message": result.get("message") or "", + "job": result.get("job"), + } + if publish_mode == "local_browser": + raise PublishValidationError( + "local_browser 当前未实现,任务不会降级为发布包导出", + "local_browser_not_implemented", + ) + raise PublishValidationError( + f"不支持的 publish_mode:{publish_mode or '(empty)'}", + "unsupported_publish_mode", + ) diff --git a/app/services/publish_scheduler.py b/app/services/publish_scheduler.py index d22e54f..735e53b 100644 --- a/app/services/publish_scheduler.py +++ b/app/services/publish_scheduler.py @@ -3,13 +3,14 @@ import argparse import asyncio import json -from datetime import datetime, time, timedelta +from datetime import datetime, time, timedelta, timezone from typing import Any from app.core.config import settings from app.db.database import get_connection, init_db -from app.services.publish_adapters import PublishValidationError, publisher_for_job +from app.services.publish_adapters import PublishValidationError from app.services.publish_domain import PUBLISH_STATUSES +from app.services.publish_executor import execute_publish_job from app.services.task_log_service import append_task_log @@ -17,6 +18,14 @@ def now_iso() -> str: return datetime.now().astimezone().isoformat(timespec="seconds") +_SCHEDULER_HEALTH: dict[str, Any] = { + "running": False, + "scanning": False, + "last_scan_at": "", + "next_scan_at": "", +} + + def parse_datetime(value: str | None) -> datetime: text = (value or "").strip() if not text: @@ -112,49 +121,83 @@ def __init__( def run_once(self) -> dict[str, Any]: init_db() - self.recover_interrupted_jobs() - jobs = self.list_due_jobs() - results = [self.execute_job(job["id"]) for job in jobs] - return { - "status": "ok", - "checked_at": now_iso(), - "matched_count": len(jobs), - "published_count": sum(1 for item in results if item.get("status") == "published"), - "failed_count": sum(1 for item in results if item.get("status") == "failed"), - "skipped_count": sum(1 for item in results if item.get("status") == "skipped"), - "results": results, - } + _SCHEDULER_HEALTH["scanning"] = True + try: + self.recover_interrupted_jobs() + jobs = self.list_due_jobs() + results = [self.execute_job(job["id"]) for job in jobs] + checked_at = datetime.now().astimezone() + _SCHEDULER_HEALTH["last_scan_at"] = checked_at.isoformat(timespec="seconds") + _SCHEDULER_HEALTH["next_scan_at"] = ( + checked_at + timedelta(seconds=self.interval_seconds) + ).isoformat(timespec="seconds") + return { + "status": "ok", + "checked_at": _SCHEDULER_HEALTH["last_scan_at"], + "matched_count": len(jobs), + "published_count": sum(1 for item in results if item.get("status") == "published"), + "exported_count": sum(1 for item in results if item.get("status") == "exported"), + "failed_count": sum(1 for item in results if item.get("status") == "failed"), + "skipped_count": sum(1 for item in results if item.get("status") == "skipped"), + "results": results, + } + finally: + _SCHEDULER_HEALTH["scanning"] = False async def run_forever(self) -> None: init_db() self._stop_event = asyncio.Event() - while not self._stop_event.is_set(): - self.run_once() - try: - await asyncio.wait_for(self._stop_event.wait(), timeout=self.interval_seconds) - except TimeoutError: - continue + _SCHEDULER_HEALTH["running"] = True + try: + while not self._stop_event.is_set(): + await asyncio.to_thread(self.run_once) + try: + await asyncio.wait_for(self._stop_event.wait(), timeout=self.interval_seconds) + except TimeoutError: + continue + finally: + _SCHEDULER_HEALTH["running"] = False def stop(self) -> None: if self._stop_event: self._stop_event.set() def recover_interrupted_jobs(self) -> int: - now = now_iso() + now = datetime.now(timezone.utc) + stale_before = now - timedelta(minutes=max(1, int(settings.publish_job_stale_minutes))) + recovered = 0 with get_connection() as connection: - cursor = connection.execute( + rows = connection.execute( """ - UPDATE publish_jobs - SET status = 'SCHEDULED', - last_error = COALESCE(NULLIF(last_error, ''), 'Recovered from interrupted PUBLISHING state'), - error_message = COALESCE(NULLIF(error_message, ''), 'Recovered from interrupted PUBLISHING state'), - updated_at = ? - WHERE status = 'PUBLISHING' AND published_at IS NULL - """, - (now,), - ) + SELECT id, updated_at, provider_response, publish_result + FROM publish_jobs + WHERE status = 'PUBLISHING' AND (published_at IS NULL OR published_at = '') + AND (platform_item_id IS NULL OR platform_item_id = '') + AND (remote_video_id IS NULL OR remote_video_id = '') + """ + ).fetchall() + for row in rows: + try: + updated_at = parse_datetime(row["updated_at"]).astimezone(timezone.utc) + except ValueError: + continue + success_text = f"{row['provider_response'] or ''} {row['publish_result'] or ''}".lower() + if updated_at > stale_before or any(marker in success_text for marker in ('"completed"', '"success"', 'published')): + continue + cursor = connection.execute( + """ + UPDATE publish_jobs + SET status = 'SCHEDULED', + last_error = '陈旧 PUBLISHING 任务已恢复,等待重新执行', + error_message = '陈旧 PUBLISHING 任务已恢复,等待重新执行', + updated_at = ? + WHERE id = ? AND status = 'PUBLISHING' AND updated_at = ? + """, + (now.isoformat(timespec="seconds"), row["id"], row["updated_at"]), + ) + recovered += int(cursor.rowcount or 0) connection.commit() - return int(cursor.rowcount or 0) + return recovered def list_due_jobs(self) -> list[dict[str, Any]]: current = datetime.now().astimezone() @@ -179,7 +222,14 @@ def list_due_jobs(self) -> list[dict[str, Any]]: due.append(job) return due - def execute_job(self, job_id: str, *, force: bool = False, allow_republish: bool = False) -> dict[str, Any]: + def execute_job( + self, + job_id: str, + *, + force: bool = False, + allow_republish: bool = False, + runner=None, + ) -> dict[str, Any]: job = get_publish_job_raw(job_id) if not job: return {"status": "failed", "job_id": job_id, "message": "publish job not found"} @@ -189,7 +239,7 @@ def execute_job(self, job_id: str, *, force: bool = False, allow_republish: bool return {"status": "skipped", "job_id": job_id, "message": "already published"} if status in {"CANCELLED", "NEED_REVIEW"}: return {"status": "skipped", "job_id": job_id, "message": f"status is {status}"} - if status not in {"SCHEDULED", "FAILED", "PUBLISHING"} and not force: + if status != "SCHEDULED": return {"status": "skipped", "job_id": job_id, "message": f"status is {status}"} if _risk_flags(job) and not settings.publish_scheduler_allow_publish_without_review: @@ -207,20 +257,27 @@ def execute_job(self, job_id: str, *, force: bool = False, allow_republish: bool if not force and attempts >= self.max_retry_count: return self._mark_failed(job_id, "max_retry_exceeded", "max retry count exceeded") - self._mark_publishing(job_id) - job = get_publish_job_raw(job_id) or job + if not self._claim_scheduled_job(job_id): + return {"status": "skipped", "job_id": job_id, "message": "job was claimed by another scheduler"} try: - result = publisher_for_job(job).publish(job) + result = execute_publish_job(job_id, force=force, runner=runner) except PublishValidationError as exc: return self._mark_failed(job_id, exc.error_code, exc.message) except Exception as exc: return self._mark_failed(job_id, "publish_failed", str(exc) or exc.__class__.__name__) - return self._mark_published(job_id, result.payload, result.remote_video_id) + if result.get("status") == "exported": + return self._mark_exported(job_id, result.get("payload") or {}, result.get("remote_video_id") or "") + return result - def publish_now(self, job_id: str, *, allow_republish: bool = False) -> dict[str, Any]: + def publish_now(self, job_id: str, *, allow_republish: bool = False, runner=None) -> dict[str, Any]: + job = get_publish_job_raw(job_id) + if not job: + raise ValueError("publish job not found") + if str(job.get("status") or "").upper() not in {"WAITING", "SCHEDULED", "FAILED"}: + return {"status": "skipped", "job_id": job_id, "message": f"status is {job.get('status')}"} self._set_schedule_to_now(job_id) - return self.execute_job(job_id, force=True, allow_republish=allow_republish) + return self.execute_job(job_id, force=True, allow_republish=allow_republish, runner=runner) def retry_failed(self, job_id: str) -> dict[str, Any]: job = get_publish_job_raw(job_id) @@ -266,14 +323,15 @@ def update_schedule(self, job_id: str, scheduled_at: str) -> dict[str, Any]: if str(job.get("status") or "").upper() == "PUBLISHED": raise ValueError("published jobs cannot be rescheduled") now = now_iso() + next_status = "NEED_REVIEW" if str(job.get("status") or "").upper() == "NEED_REVIEW" else "SCHEDULED" with get_connection() as connection: connection.execute( """ UPDATE publish_jobs - SET scheduled_at = ?, status = 'SCHEDULED', updated_at = ? + SET scheduled_at = ?, status = ?, updated_at = ? WHERE id = ? """, - (parsed.isoformat(timespec="seconds"), now, job_id), + (parsed.isoformat(timespec="seconds"), next_status, now, job_id), ) connection.commit() return {"status": "ok", "job": get_publish_job_raw(job_id)} @@ -380,11 +438,11 @@ def _set_schedule_to_now(self, job_id: str) -> None: ) connection.commit() - def _mark_publishing(self, job_id: str) -> None: + def _claim_scheduled_job(self, job_id: str) -> bool: now = now_iso() payload = json.dumps({"publisher": "started", "started_at": now}, ensure_ascii=False) with get_connection() as connection: - connection.execute( + cursor = connection.execute( """ UPDATE publish_jobs SET status = 'PUBLISHING', @@ -395,11 +453,12 @@ def _mark_publishing(self, job_id: str) -> None: publish_result = ?, provider_response = ?, updated_at = ? - WHERE id = ? + WHERE id = ? AND status = 'SCHEDULED' """, (payload, payload, now, job_id), ) connection.commit() + return int(cursor.rowcount or 0) == 1 def _mark_published(self, job_id: str, payload: dict[str, Any], remote_video_id: str) -> dict[str, Any]: now = now_iso() @@ -428,6 +487,25 @@ def _mark_published(self, job_id: str, payload: dict[str, Any], remote_video_id: self._append_log(job.get("task_id") or "", f"Publish job {job_id} completed by manual_export") return {"status": "published", "job_id": job_id, "publish_result": payload} + def _mark_exported(self, job_id: str, payload: dict[str, Any], remote_video_id: str) -> dict[str, Any]: + now = now_iso() + publish_result = json.dumps(payload, ensure_ascii=False) + with get_connection() as connection: + connection.execute( + """ + UPDATE publish_jobs + SET status = 'EXPORTED', publish_result = ?, provider_response = ?, + remote_video_id = ?, published_at = NULL, audit_status = 'not_submitted', + last_error = '', error_message = '', error_code = '', updated_at = ? + WHERE id = ? AND status = 'PUBLISHING' + """, + (publish_result, publish_result, remote_video_id, now, job_id), + ) + connection.commit() + job = get_publish_job_raw(job_id) or {"task_id": ""} + self._append_log(job.get("task_id") or "", f"Publish job {job_id} exported a local package") + return {"status": "exported", "job_id": job_id, "publish_result": payload} + def _mark_failed(self, job_id: str, error_code: str, message: str) -> dict[str, Any]: now = now_iso() payload = json.dumps( @@ -545,6 +623,33 @@ def queue_snapshot(task_id: str | None = None) -> dict[str, Any]: } +def scheduler_health() -> dict[str, Any]: + from app.services.publish_service import _opencli_status + + with get_connection() as connection: + counts = connection.execute( + """ + SELECT + SUM(CASE WHEN status = 'SCHEDULED' THEN 1 ELSE 0 END) AS scheduled_count, + SUM(CASE WHEN status = 'PUBLISHING' THEN 1 ELSE 0 END) AS publishing_count + FROM publish_jobs + """ + ).fetchone() + opencli = _opencli_status() + return { + "enabled": bool(settings.publish_scheduler_enabled), + "running": bool(_SCHEDULER_HEALTH["running"]), + "scanning": bool(_SCHEDULER_HEALTH["scanning"]), + "last_scan_at": _SCHEDULER_HEALTH["last_scan_at"], + "next_scan_at": _SCHEDULER_HEALTH["next_scan_at"], + "interval_seconds": int(settings.publish_scheduler_interval_seconds), + "scheduled_count": int(counts["scheduled_count"] or 0), + "publishing_count": int(counts["publishing_count"] or 0), + "opencli_available": bool(opencli["available"]), + "opencli_message": opencli["message"], + } + + async def start_scheduler_background() -> PublishScheduler | None: if not settings.publish_scheduler_enabled: return None diff --git a/app/services/publish_service.py b/app/services/publish_service.py index 6b9ee14..4c551a9 100644 --- a/app/services/publish_service.py +++ b/app/services/publish_service.py @@ -2563,10 +2563,11 @@ def execute_opencli_send_job(job_id: str, runner: CommandRunner | None = None) - """ UPDATE publish_jobs SET status = 'PUBLISHED', audit_status = 'submitted', - error_code = '', error_message = '', provider_response = ?, updated_at = ? + error_code = '', error_message = '', provider_response = ?, + published_at = ?, updated_at = ? WHERE id = ? """, - (json.dumps(response, ensure_ascii=False), now, job_id), + (json.dumps(response, ensure_ascii=False), now, now, job_id), ) connection.commit() return {"status": "ok", "message": "opencli 发送流程已执行完成。", "job": get_publish_job(job_id)} @@ -2574,7 +2575,7 @@ def execute_opencli_send_job(job_id: str, runner: CommandRunner | None = None) - def _ready_opencli_job_ids(job_ids: list[str] | None = None) -> list[str]: params: list[str] = [] - where = "publish_mode = 'opencli_publish' AND status IN ('WAITING', 'FAILED', 'ready', 'failed')" + where = "publish_mode = 'opencli_publish' AND status IN ('WAITING', 'SCHEDULED', 'FAILED', 'ready', 'scheduled', 'failed')" if job_ids: placeholders = ",".join("?" for _ in job_ids) where += f" AND id IN ({placeholders})" @@ -2592,7 +2593,10 @@ def run_opencli_send_batch(job_ids: list[str] | None = None, runner: CommandRunn return {"status": "busy", "message": "发送队列正在运行,请等待当前批次结束。", "jobs": list_publish_jobs(limit=100)} try: ids = _ready_opencli_job_ids(job_ids) - results = [execute_opencli_send_job(job_id, runner=runner) for job_id in ids] + from app.services.publish_scheduler import PublishScheduler + + scheduler = PublishScheduler() + results = [scheduler.publish_now(job_id, runner=runner) for job_id in ids] return {"status": "ok", "message": f"发送批次已处理 {len(results)} 条任务。", "results": results, **get_publish_center_context()} finally: _SEND_LOCK.release() @@ -2624,21 +2628,17 @@ def retry_publish_job(job_id: str) -> dict: job = get_publish_job(job_id) if not job: raise ValueError("发布任务不存在。") - if ( - job.get("platform") == "manual_export" - or job.get("publish_mode") == "manual_export" - or ( - job.get("status") == PUBLISH_STATUS_FAILED - and settings.publish_scheduler_default_platform == "manual_export" - ) - ): - from app.services.publish_scheduler import PublishScheduler + from app.services.publish_scheduler import PublishScheduler - return PublishScheduler().retry_failed(job_id) - if job.get("publish_mode") == "opencli_publish": - return execute_opencli_send_job(job_id) + return PublishScheduler().retry_failed(job_id) + + +def execute_api_publish_job(job_id: str) -> dict: + job = get_publish_job(job_id) + if not job: + raise ValueError("发布任务不存在。") if job.get("publish_mode") != "api_publish": - raise ValueError("只有真实接口发布任务可以重试。") + raise ValueError("只能执行 api_publish 任务。") output_clip = _get_output_clip_for_publish(job["task_id"], job["output_clip_id"]) if not output_clip: raise ValueError("切片记录不存在。") @@ -2647,17 +2647,6 @@ def retry_publish_job(job_id: str) -> dict: account = get_account(job.get("account_id") or "") if not config or not account: raise ValueError("平台配置或账号不存在。") - with get_connection() as connection: - connection.execute( - """ - UPDATE publish_jobs - SET status = 'PUBLISHING', retry_count = retry_count + 1, - error_code = '', error_message = '', updated_at = ? - WHERE id = ? - """, - (_now_iso(), job_id), - ) - connection.commit() return _execute_publish_job(job_id, config=config, account=account, video_path=video_path) @@ -2695,7 +2684,7 @@ def _execute_publish_job(job_id: str, config: dict, account: dict, video_path: P UPDATE publish_jobs SET status = 'PUBLISHED', audit_status = ?, platform_item_id = ?, platform_upload_id = ?, error_code = '', error_message = '', - provider_response = ?, updated_at = ? + provider_response = ?, published_at = ?, updated_at = ? WHERE id = ? """, ( @@ -2704,6 +2693,7 @@ def _execute_publish_job(job_id: str, config: dict, account: dict, video_path: P result.upload_id, json.dumps(result.response or {}, ensure_ascii=False), now, + now, job_id, ), ) From a69dbab3ad68cbc5e16b8c43c921dd2a35b89599 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 14:03:41 +0800 Subject: [PATCH 03/20] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E7=BB=9F?= =?UTF-8?q?=E4=B8=80=E6=8E=92=E6=9C=9F=E6=97=B6=E5=8C=BA=E4=B8=8E=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E6=B5=81=E8=BD=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/db/database.py | 1 + app/models/task.py | 5 +- app/routers/publish.py | 22 ++++- app/services/publish_scheduler.py | 132 +++++++++++++++++++++++++----- app/services/publish_service.py | 30 ++++++- requirements.txt | 3 + 6 files changed, 165 insertions(+), 28 deletions(-) diff --git a/app/db/database.py b/app/db/database.py index adc4b52..c9ad75e 100644 --- a/app/db/database.py +++ b/app/db/database.py @@ -694,6 +694,7 @@ def _migrate_publish_jobs_table(connection: sqlite3.Connection) -> None: "bilibili_source": "ALTER TABLE publish_jobs ADD COLUMN bilibili_source TEXT", "cover_file_path": "ALTER TABLE publish_jobs ADD COLUMN cover_file_path TEXT", "scheduled_at": "ALTER TABLE publish_jobs ADD COLUMN scheduled_at TEXT", + "schedule_timezone": "ALTER TABLE publish_jobs ADD COLUMN schedule_timezone TEXT NOT NULL DEFAULT 'Asia/Shanghai'", "status": "ALTER TABLE publish_jobs ADD COLUMN status TEXT NOT NULL DEFAULT 'ready'", "audit_status": "ALTER TABLE publish_jobs ADD COLUMN audit_status TEXT NOT NULL DEFAULT 'not_submitted'", "platform_item_id": "ALTER TABLE publish_jobs ADD COLUMN platform_item_id TEXT", diff --git a/app/models/task.py b/app/models/task.py index 32fd745..9a40eca 100644 --- a/app/models/task.py +++ b/app/models/task.py @@ -169,8 +169,9 @@ class PublishJobScheduleUpdate(BaseModel): class PublishBatchScheduleUpdate(BaseModel): job_ids: list[str] = Field(default_factory=list) action: Literal["apply", "clear"] = "apply" - start_at: Optional[str] = Field(default="", max_length=80) - interval_hours: int = Field(default=3, ge=1, le=168) + start_at_local: Optional[str] = Field(default="", max_length=80) + timezone: str = Field(default="Asia/Shanghai", min_length=1, max_length=80) + interval_minutes: int = Field(default=180, ge=1, le=10080) daily_start_time: str = Field(default="09:00", min_length=5, max_length=5) daily_end_time: str = Field(default="21:00", min_length=5, max_length=5) diff --git a/app/routers/publish.py b/app/routers/publish.py index 6e39787..cd4dbb4 100644 --- a/app/routers/publish.py +++ b/app/routers/publish.py @@ -234,8 +234,26 @@ async def update_publish_jobs_schedule_batch(payload: PublishBatchScheduleUpdate return PublishScheduler().update_batch_schedule( payload.job_ids, action=payload.action, - start_at=payload.start_at or "", - interval_hours=payload.interval_hours, + start_at_local=payload.start_at_local or "", + timezone_name=payload.timezone, + interval_minutes=payload.interval_minutes, + daily_start_time=payload.daily_start_time, + daily_end_time=payload.daily_end_time, + ) + except ValueError as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + + +@router.post("/schedules/preview") +async def preview_publish_jobs_schedule(payload: PublishBatchScheduleUpdate) -> dict: + if payload.action != "apply": + raise HTTPException(status_code=400, detail="排期预览只支持 apply") + try: + return PublishScheduler().preview_batch_schedule( + payload.job_ids, + start_at_local=payload.start_at_local or "", + timezone_name=payload.timezone, + interval_minutes=payload.interval_minutes, daily_start_time=payload.daily_start_time, daily_end_time=payload.daily_end_time, ) diff --git a/app/services/publish_scheduler.py b/app/services/publish_scheduler.py index 735e53b..9fe650d 100644 --- a/app/services/publish_scheduler.py +++ b/app/services/publish_scheduler.py @@ -5,6 +5,7 @@ import json from datetime import datetime, time, timedelta, timezone from typing import Any +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from app.core.config import settings from app.db.database import get_connection, init_db @@ -47,16 +48,28 @@ def parse_clock(value: str, field_label: str) -> time: def build_batch_schedule_times( count: int, *, - start_at: str, - interval_hours: int, + start_at_local: str, + timezone_name: str, + interval_minutes: int, daily_start_time: str, daily_end_time: str, ) -> list[str]: if count <= 0: return [] - cursor = parse_datetime(start_at) - interval = timedelta(hours=max(1, int(interval_hours))) + try: + schedule_zone = ZoneInfo((timezone_name or "").strip()) + except ZoneInfoNotFoundError as exc: + raise ValueError("时区无效,请使用例如 Asia/Shanghai 的 IANA 时区") from exc + start_text = (start_at_local or "").strip() + if not start_text: + raise ValueError("起始时间不能为空") + try: + cursor = datetime.fromisoformat(start_text) + except ValueError as exc: + raise ValueError("起始时间格式无效") from exc + cursor = cursor.replace(tzinfo=schedule_zone) if cursor.tzinfo is None else cursor.astimezone(schedule_zone) + interval = timedelta(minutes=max(1, int(interval_minutes))) window_start = parse_clock(daily_start_time, "每日开始时间") window_end = parse_clock(daily_end_time, "每日结束时间") if window_end <= window_start: @@ -73,11 +86,46 @@ def build_batch_schedule_times( tzinfo=cursor.tzinfo ) continue - scheduled.append(cursor.isoformat(timespec="seconds")) + scheduled.append( + cursor.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") + ) cursor += interval return scheduled +def build_batch_schedule_preview( + job_ids: list[str], + *, + start_at_local: str, + timezone_name: str, + interval_minutes: int, + daily_start_time: str, + daily_end_time: str, +) -> list[dict[str, str]]: + utc_times = build_batch_schedule_times( + len(job_ids), + start_at_local=start_at_local, + timezone_name=timezone_name, + interval_minutes=interval_minutes, + daily_start_time=daily_start_time, + daily_end_time=daily_end_time, + ) + schedule_zone = ZoneInfo(timezone_name) + preview = [] + for job_id, utc_value in zip(job_ids, utc_times, strict=True): + local_value = parse_datetime(utc_value).astimezone(schedule_zone) + preview.append( + { + "job_id": job_id, + "scheduled_at_utc": utc_value, + "scheduled_at_local": local_value.isoformat(timespec="seconds"), + "scheduled_at_local_display": local_value.strftime("%Y-%m-%d %H:%M"), + "timezone": timezone_name, + } + ) + return preview + + def _row_to_dict(row) -> dict[str, Any] | None: return dict(row) if row else None @@ -320,10 +368,11 @@ def update_schedule(self, job_id: str, scheduled_at: str) -> dict[str, Any]: job = get_publish_job_raw(job_id) if not job: raise ValueError("publish job not found") - if str(job.get("status") or "").upper() == "PUBLISHED": - raise ValueError("published jobs cannot be rescheduled") + if str(job.get("status") or "").upper() in {"PUBLISHED", "EXPORTED", "CANCELLED"}: + raise ValueError("已完成或已取消的任务不能重新排期") now = now_iso() next_status = "NEED_REVIEW" if str(job.get("status") or "").upper() == "NEED_REVIEW" else "SCHEDULED" + scheduled_at_utc = parsed.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") with get_connection() as connection: connection.execute( """ @@ -331,7 +380,7 @@ def update_schedule(self, job_id: str, scheduled_at: str) -> dict[str, Any]: SET scheduled_at = ?, status = ?, updated_at = ? WHERE id = ? """, - (parsed.isoformat(timespec="seconds"), next_status, now, job_id), + (scheduled_at_utc, next_status, now, job_id), ) connection.commit() return {"status": "ok", "job": get_publish_job_raw(job_id)} @@ -341,8 +390,9 @@ def update_batch_schedule( job_ids: list[str], *, action: str, - start_at: str = "", - interval_hours: int = 3, + start_at_local: str = "", + timezone_name: str = "Asia/Shanghai", + interval_minutes: int = 180, daily_start_time: str = "09:00", daily_end_time: str = "21:00", ) -> dict[str, Any]: @@ -365,26 +415,37 @@ def update_batch_schedule( blocked = [ job_id for job_id in normalized_ids - if str(jobs_by_id[job_id].get("status") or "").upper() in {"PUBLISHED", "CANCELLED"} + if str(jobs_by_id[job_id].get("status") or "").upper() in {"PUBLISHED", "EXPORTED", "CANCELLED"} ] if blocked: raise ValueError("已发布或已取消的任务不能修改排期") schedule_times = ( - build_batch_schedule_times( - len(normalized_ids), - start_at=start_at, - interval_hours=interval_hours, + build_batch_schedule_preview( + normalized_ids, + start_at_local=start_at_local, + timezone_name=timezone_name, + interval_minutes=interval_minutes, daily_start_time=daily_start_time, daily_end_time=daily_end_time, ) if action == "apply" - else [""] * len(normalized_ids) + else [ + { + "job_id": job_id, + "scheduled_at_utc": "", + "scheduled_at_local": "", + "scheduled_at_local_display": "未排期", + "timezone": timezone_name, + } + for job_id in normalized_ids + ] ) now = now_iso() with get_connection() as connection: - for job_id, scheduled_at in zip(normalized_ids, schedule_times, strict=True): + for job_id, schedule_item in zip(normalized_ids, schedule_times, strict=True): + scheduled_at = schedule_item["scheduled_at_utc"] current_status = str(jobs_by_id[job_id].get("status") or "").upper() if current_status == "NEED_REVIEW": next_status = "NEED_REVIEW" @@ -397,13 +458,13 @@ def update_batch_schedule( connection.execute( """ UPDATE publish_jobs - SET scheduled_at = ?, status = ?, updated_at = ?, + SET scheduled_at = ?, schedule_timezone = ?, status = ?, updated_at = ?, error_code = CASE WHEN ? = 'apply' THEN '' ELSE error_code END, error_message = CASE WHEN ? = 'apply' THEN '' ELSE error_message END, last_error = CASE WHEN ? = 'apply' THEN '' ELSE last_error END WHERE id = ? """, - (scheduled_at, next_status, now, action, action, action, job_id), + (scheduled_at, timezone_name, next_status, now, action, action, action, job_id), ) connection.commit() @@ -417,10 +478,41 @@ def update_batch_schedule( else f"已清除 {len(normalized_ids)} 条任务的发布时间。" ), "jobs": [get_publish_job_raw(job_id) for job_id in normalized_ids], + "schedule": schedule_times, } + def preview_batch_schedule( + self, + job_ids: list[str], + *, + start_at_local: str, + timezone_name: str, + interval_minutes: int, + daily_start_time: str, + daily_end_time: str, + ) -> dict[str, Any]: + normalized_ids = list(dict.fromkeys(str(job_id).strip() for job_id in job_ids if str(job_id).strip())) + if not normalized_ids: + raise ValueError("至少选择一条发布任务") + with get_connection() as connection: + found = connection.execute( + f"SELECT COUNT(*) FROM publish_jobs WHERE id IN ({','.join('?' for _ in normalized_ids)})", + normalized_ids, + ).fetchone()[0] + if int(found) != len(normalized_ids): + raise ValueError("部分发布任务不存在") + schedule = build_batch_schedule_preview( + normalized_ids, + start_at_local=start_at_local, + timezone_name=timezone_name, + interval_minutes=interval_minutes, + daily_start_time=daily_start_time, + daily_end_time=daily_end_time, + ) + return {"status": "ok", "timezone": timezone_name, "schedule": schedule} + def _set_schedule_to_now(self, job_id: str) -> None: - now = now_iso() + now = datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") with get_connection() as connection: connection.execute( """ diff --git a/app/services/publish_service.py b/app/services/publish_service.py index 4c551a9..3302579 100644 --- a/app/services/publish_service.py +++ b/app/services/publish_service.py @@ -5,12 +5,13 @@ import subprocess import urllib.error import urllib.request -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone from pathlib import Path from threading import Lock from typing import Any, Callable from urllib.parse import urlsplit from uuid import uuid4 +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError from app.core.config import settings from app.db.database import get_connection @@ -487,12 +488,19 @@ def _normalize_publish_status(status: str | None) -> str: return LEGACY_STATUS_MAP.get(raw.lower(), raw) -def _format_publish_schedule(value: str | None) -> str: +def _format_publish_schedule(value: str | None, timezone_name: str = "Asia/Shanghai") -> str: text = (value or "").strip() if not text: return "未排期" try: - return datetime.fromisoformat(text.replace("Z", "+00:00")).strftime("%Y-%m-%d %H:%M") + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + try: + display_zone = ZoneInfo(timezone_name or "Asia/Shanghai") + except ZoneInfoNotFoundError: + display_zone = ZoneInfo("Asia/Shanghai") + return parsed.astimezone(display_zone).strftime("%Y-%m-%d %H:%M") except ValueError: return text @@ -507,6 +515,17 @@ def _normalize_job(row) -> dict: clip_id = job.get("clip_id") or job.get("output_clip_id") or "" video_path = job.get("video_path") or job.get("video_file_path") or "" error_message = job.get("error_message") or job.get("last_error") or "" + schedule_timezone = job.get("schedule_timezone") or "Asia/Shanghai" + scheduled_at_utc = job.get("scheduled_at") or "" + scheduled_at_local = "" + if scheduled_at_utc: + try: + parse_datetime_value = datetime.fromisoformat(scheduled_at_utc.replace("Z", "+00:00")) + if parse_datetime_value.tzinfo is None: + parse_datetime_value = parse_datetime_value.replace(tzinfo=timezone.utc) + scheduled_at_local = parse_datetime_value.astimezone(ZoneInfo(schedule_timezone)).isoformat(timespec="seconds") + except (ValueError, ZoneInfoNotFoundError): + scheduled_at_local = scheduled_at_utc job.update( { "status": status, @@ -526,7 +545,10 @@ def _normalize_job(row) -> dict: "status_tone": STATUS_TONES.get(status, "blue"), "video_source_label": VIDEO_SOURCE_LABELS.get(job.get("video_source"), job.get("video_source")), "publish_mode_label": PUBLISH_MODE_LABELS.get(job.get("publish_mode"), job.get("publish_mode")), - "scheduled_at_display": _format_publish_schedule(job.get("scheduled_at")), + "schedule_timezone": schedule_timezone, + "scheduled_at_utc": scheduled_at_utc, + "scheduled_at_local": scheduled_at_local, + "scheduled_at_display": _format_publish_schedule(scheduled_at_utc, schedule_timezone), "account_name": job.get("account_name") or "未选择账号", "cover_media_url": _cover_media_url(job.get("task_id") or "", job.get("cover_file_path")), "video_media_url": _video_media_url( diff --git a/requirements.txt b/requirements.txt index ebb24ec..0d5d54a 100644 --- a/requirements.txt +++ b/requirements.txt @@ -14,6 +14,9 @@ python-multipart>=0.0.28 # 数据校验 pydantic>=2.13.0 +# Windows 的 zoneinfo IANA 时区数据库(排期需要 Asia/Shanghai 等名称) +tzdata + # 异步文件操作 aiofiles>=25.1.0 From 41382de193ebe3994ce9f8d782adbfee3c217910 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 11 Jul 2026 14:09:17 +0800 Subject: [PATCH 04/20] =?UTF-8?q?=E4=BC=98=E5=8C=96=EF=BC=9A=E9=87=8D?= =?UTF-8?q?=E6=9E=84=E5=8F=91=E9=80=81=E4=B8=AD=E5=BF=83=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E6=8E=92=E6=9C=9F=E4=BA=A4=E4=BA=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/publish_service.py | 18 ++ app/static/css/styles.css | 117 ++++++++++ app/static/js/app.js | 26 +++ app/static/js/publish-center.js | 316 +++++++++++++++++++++++++ app/templates/base.html | 3 +- app/templates/publish.html | 402 +++++++++----------------------- 6 files changed, 590 insertions(+), 292 deletions(-) create mode 100644 app/static/js/publish-center.js diff --git a/app/services/publish_service.py b/app/services/publish_service.py index 3302579..ffe48e0 100644 --- a/app/services/publish_service.py +++ b/app/services/publish_service.py @@ -2798,6 +2798,18 @@ def get_publish_center_context() -> dict: ) jobs = list_publish_jobs(limit=200) + pending_jobs = [ + job for job in jobs + if job.get("status") in {PUBLISH_STATUS_DRAFT, PUBLISH_STATUS_WAITING, PUBLISH_STATUS_FAILED, PUBLISH_STATUS_NEED_REVIEW} + ] + scheduled_jobs = sorted( + [job for job in jobs if job.get("status") in {PUBLISH_STATUS_SCHEDULED, PUBLISH_STATUS_PUBLISHING}], + key=lambda job: (job.get("scheduled_at") or "", job.get("created_at") or ""), + ) + history_jobs = [ + job for job in jobs + if job.get("status") in {PUBLISH_STATUS_PUBLISHED, PUBLISH_STATUS_EXPORTED, PUBLISH_STATUS_FAILED, PUBLISH_STATUS_CANCELLED} + ] jobs_by_platform = { platform: [job for job in jobs if job["platform"] == platform] for platform in PLATFORM_LABELS @@ -2808,14 +2820,20 @@ def get_publish_center_context() -> dict: failed_count = sum(1 for job in jobs if job.get("status") == PUBLISH_STATUS_FAILED) need_review_count = sum(1 for job in jobs if job.get("status") == PUBLISH_STATUS_NEED_REVIEW) opencli_status = _opencli_status() + from app.services.publish_scheduler import scheduler_health + return { "publish_items": publish_items, "send_queue_items": queue_items, "publish_jobs": jobs, + "pending_jobs": pending_jobs, + "scheduled_jobs": scheduled_jobs, + "history_jobs": history_jobs, "jobs_by_platform": jobs_by_platform, "platforms": [{"id": platform, "label": label} for platform, label in PLATFORM_LABELS.items()], "opencli_available": opencli_status["available"], "opencli_status": opencli_status, + "scheduler_health": scheduler_health(), "stats": [ {"label": "需复核", "value": need_review_count, "tone": "amber"}, {"label": "可入队切片", "value": len(publish_items), "tone": "green"}, diff --git a/app/static/css/styles.css b/app/static/css/styles.css index 8c2c0c4..d806305 100644 --- a/app/static/css/styles.css +++ b/app/static/css/styles.css @@ -4873,3 +4873,120 @@ td a, min-width: 38px; } } +/* 发送中心:紧凑列表、批量栏与排期抽屉 */ +.scheduler-health-card { + display: flex; + align-items: center; + gap: 18px; + flex-wrap: wrap; + margin-bottom: 18px; + padding: 14px 18px; + border: 1px solid rgba(124, 151, 190, 0.2); + border-radius: 16px; + background: rgba(255, 255, 255, 0.74); +} + +.scheduler-health-card div { display: flex; align-items: center; gap: 8px; } +.scheduler-health-card small { color: var(--muted); flex: 1 1 280px; } +.health-dot { width: 9px; height: 9px; border-radius: 50%; background: #f59e0b; } +.health-dot.is-ok { background: #22c55e; box-shadow: 0 0 0 5px rgba(34, 197, 94, 0.12); } + +.publish-center-tabs { + display: flex; + gap: 8px; + margin-bottom: 14px; + padding: 5px; + width: fit-content; + border-radius: 14px; + background: rgba(227, 235, 247, 0.72); +} + +.publish-center-tabs button { + border: 0; + border-radius: 10px; + padding: 10px 16px; + color: var(--muted); + background: transparent; + cursor: pointer; +} + +.publish-center-tabs button.active { color: var(--blue); background: #fff; box-shadow: 0 6px 18px rgba(60, 90, 130, 0.12); } +.publish-center-tabs span { margin-left: 6px; opacity: 0.7; } +.publish-center-panel[hidden] { display: none; } +.publish-compact-list { display: grid; gap: 10px; } + +.publish-compact-row { + display: grid; + grid-template-columns: 28px 104px minmax(220px, 1fr) 170px auto; + align-items: center; + gap: 14px; + padding: 12px; + border: 1px solid rgba(126, 151, 187, 0.18); + border-radius: 16px; + background: rgba(255, 255, 255, 0.82); +} + +.publish-row-thumb { width: 104px; height: 64px; border-radius: 10px; object-fit: cover; background: #111827; } +.publish-row-main { display: grid; gap: 4px; min-width: 0; } +.publish-row-main > strong, .publish-row-main > span { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } +.publish-row-main > span, .publish-row-time small { color: var(--muted); font-size: 0.82rem; } +.publish-row-time { display: grid; gap: 4px; } +.publish-row-actions { display: flex; justify-content: flex-end; gap: 8px; flex-wrap: wrap; } + +.publish-inline-editor { + grid-column: 2 / -1; + display: grid; + grid-template-columns: 1fr 1fr; + gap: 12px; + padding: 16px; + border-radius: 14px; + background: rgba(237, 244, 255, 0.7); +} + +.publish-inline-editor[hidden] { display: none; } +.publish-inline-editor label { display: grid; gap: 6px; } +.publish-inline-editor .span-2 { grid-column: 1 / -1; } +.publish-inline-editor textarea { min-height: 92px; } +.publish-advanced-grid { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr)); gap: 12px; margin-top: 12px; } + +.publish-selection-bar { + position: fixed; + z-index: 80; + left: calc(50% + 110px); + bottom: 24px; + transform: translateX(-50%); + display: flex; + align-items: center; + gap: 10px; + padding: 12px 16px; + border: 1px solid rgba(90, 130, 185, 0.25); + border-radius: 18px; + background: rgba(255, 255, 255, 0.94); + box-shadow: 0 18px 50px rgba(30, 55, 90, 0.2); + backdrop-filter: blur(18px); +} + +.publish-selection-bar[hidden], .schedule-drawer[hidden], .schedule-drawer-backdrop[hidden] { display: none; } +.schedule-drawer-backdrop { position: fixed; z-index: 89; inset: 0; background: rgba(15, 23, 42, 0.22); } +.schedule-drawer { position: fixed; z-index: 90; top: 0; right: 0; width: min(440px, 94vw); height: 100vh; padding: 24px; overflow-y: auto; background: #f9fbff; box-shadow: -20px 0 50px rgba(30, 55, 90, 0.2); } +.schedule-drawer-header { display: flex; align-items: flex-start; justify-content: space-between; margin-bottom: 22px; } +.schedule-drawer form { display: grid; gap: 16px; } +.schedule-drawer form > label, .schedule-window-grid label { display: grid; gap: 7px; } +.schedule-window-grid { display: grid; grid-template-columns: 1fr 1fr; gap: 12px; } +.timezone-note { padding: 10px 12px; border-radius: 10px; background: #eaf2ff; } +.schedule-preview-list { display: grid; gap: 8px; max-height: 260px; overflow-y: auto; } +.schedule-preview-list > div { display: flex; justify-content: space-between; gap: 12px; padding: 10px; border-radius: 10px; background: #fff; } +.schedule-preview-list strong { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } +.schedule-preview-list time { flex: none; color: var(--blue); } +.publish-history-table { display: grid; gap: 8px; } +.publish-history-row { display: grid; grid-template-columns: 150px 80px minmax(200px, 1fr) 150px 110px auto; gap: 12px; align-items: center; padding: 12px; border-bottom: 1px solid rgba(126, 151, 187, 0.18); } +.publish-history-row small { grid-column: 3 / -1; } + +@media (max-width: 960px) { + .publish-compact-row { grid-template-columns: 28px 86px 1fr; } + .publish-row-thumb { width: 86px; } + .publish-row-time, .publish-row-actions { grid-column: 3; } + .publish-inline-editor { grid-column: 1 / -1; } + .publish-selection-bar { left: 50%; width: calc(100% - 28px); overflow-x: auto; } + .publish-history-row { grid-template-columns: 1fr 1fr; } +} diff --git a/app/static/js/app.js b/app/static/js/app.js index 543b51b..0f09426 100644 --- a/app/static/js/app.js +++ b/app/static/js/app.js @@ -1,3 +1,29 @@ +async function apiFetch(url, options = {}) { + const requestOptions = { ...options }; + const headers = new Headers(options.headers || {}); + const token = document.querySelector('meta[name="local-admin-token"]')?.content || ""; + if (options.body && !(options.body instanceof FormData) && !headers.has("Content-Type")) { + headers.set("Content-Type", "application/json"); + } + if (token && !headers.has("Authorization")) { + headers.set("Authorization", `Bearer ${token}`); + } + requestOptions.headers = headers; + const response = await fetch(url, requestOptions); + let data = {}; + try { + data = await response.json(); + } catch (_error) { + data = {}; + } + if (!response.ok) { + throw new Error(data.detail || data.message || `请求失败(HTTP ${response.status})`); + } + return data; +} + +window.apiFetch = apiFetch; + const newTaskForm = document.querySelector("#new-task-form"); const newTaskAutoMode = newTaskForm?.querySelector("input[name='auto_mode']"); const newTaskSubmitButton = document.querySelector("#new-task-submit-button"); diff --git a/app/static/js/publish-center.js b/app/static/js/publish-center.js new file mode 100644 index 0000000..e6da281 --- /dev/null +++ b/app/static/js/publish-center.js @@ -0,0 +1,316 @@ +const publishCenterRoot = document.querySelector("[data-center-panel]"); + +if (publishCenterRoot) { + const selectedJobIds = new Set(); + const messageNode = document.querySelector("#send-center-message"); + const selectionBar = document.querySelector("[data-selection-bar]"); + const selectedCountNode = document.querySelector("[data-selected-count]"); + const drawer = document.querySelector("[data-schedule-drawer]"); + const drawerBackdrop = document.querySelector("[data-schedule-backdrop]"); + const drawerCount = document.querySelector("[data-drawer-count]"); + const scheduleForm = document.querySelector("[data-schedule-form]"); + const previewList = document.querySelector("[data-schedule-preview]"); + const confirmScheduleButton = document.querySelector("[data-confirm-schedule]"); + const timezoneName = Intl.DateTimeFormat().resolvedOptions().timeZone || "Asia/Shanghai"; + let latestPreviewSignature = ""; + + function showMessage(message, tone = "info") { + if (!messageNode) return; + messageNode.hidden = false; + messageNode.textContent = message; + messageNode.classList.toggle("tone-red", tone === "error"); + messageNode.classList.toggle("tone-blue", tone !== "error"); + } + + function localDatetimeValue(date) { + const local = new Date(date.getTime() - date.getTimezoneOffset() * 60000); + return local.toISOString().slice(0, 16); + } + + function updateSelectionUi() { + document.querySelectorAll("[data-publish-select]").forEach((checkbox) => { + checkbox.checked = selectedJobIds.has(checkbox.value); + }); + const count = selectedJobIds.size; + if (selectedCountNode) selectedCountNode.textContent = String(count); + if (drawerCount) drawerCount.textContent = String(count); + if (selectionBar) selectionBar.hidden = count === 0; + } + + function selectedRows() { + return Array.from(selectedJobIds) + .map((jobId) => document.querySelector(`[data-publish-row][data-job-id="${CSS.escape(jobId)}"]`)) + .filter(Boolean); + } + + function statusLabel(status) { + return { + WAITING: "等待安排", + SCHEDULED: "已排期", + PUBLISHING: "发布中", + PUBLISHED: "已发布", + EXPORTED: "已导出发布包", + FAILED: "发送失败", + CANCELLED: "已取消", + NEED_REVIEW: "需人工复核", + }[status] || status; + } + + function updateRowFromJob(job) { + if (!job?.id) return; + document.querySelectorAll(`[data-publish-row][data-job-id="${CSS.escape(job.id)}"]`).forEach((row) => { + const status = String(job.status || row.dataset.status || "").toUpperCase(); + row.dataset.status = status; + const statusNode = row.querySelector("[data-row-status]"); + if (statusNode) statusNode.textContent = job.status_label || statusLabel(status); + const titleNode = row.querySelector("[data-row-title]"); + if (titleNode && job.title) titleNode.textContent = job.title; + const timeNode = row.querySelector("[data-row-schedule]"); + if (timeNode) { + const utcValue = job.scheduled_at_utc || job.scheduled_at || ""; + timeNode.dataset.utc = utcValue; + timeNode.textContent = job.scheduled_at_display || (utcValue ? new Date(utcValue).toLocaleString() : "未排期"); + } + }); + } + + function schedulePayload(action = "apply") { + const preset = String(scheduleForm?.elements.interval_preset?.value || "180"); + const intervalMinutes = preset === "custom" + ? Number(scheduleForm?.elements.interval_minutes?.value || 180) + : Number(preset); + return { + job_ids: Array.from(selectedJobIds), + action, + start_at_local: String(scheduleForm?.elements.start_at_local?.value || ""), + timezone: timezoneName, + interval_minutes: intervalMinutes, + daily_start_time: String(scheduleForm?.elements.daily_start_time?.value || "09:00"), + daily_end_time: String(scheduleForm?.elements.daily_end_time?.value || "21:00"), + }; + } + + function previewSignature(payload) { + return JSON.stringify(payload); + } + + function invalidatePreview() { + latestPreviewSignature = ""; + if (confirmScheduleButton) confirmScheduleButton.disabled = true; + } + + function openDrawer() { + if (!selectedJobIds.size) return; + drawer.hidden = false; + drawerBackdrop.hidden = false; + document.body.classList.add("has-schedule-drawer"); + updateSelectionUi(); + } + + function closeDrawer() { + drawer.hidden = true; + drawerBackdrop.hidden = true; + document.body.classList.remove("has-schedule-drawer"); + } + + document.querySelectorAll("[data-center-tab]").forEach((button) => { + button.addEventListener("click", () => { + const tab = button.dataset.centerTab; + document.querySelectorAll("[data-center-tab]").forEach((item) => item.classList.toggle("active", item === button)); + document.querySelectorAll("[data-center-panel]").forEach((panel) => { + const active = panel.dataset.centerPanel === tab; + panel.hidden = !active; + panel.classList.toggle("active", active); + }); + }); + }); + + document.addEventListener("change", (event) => { + const checkbox = event.target.closest("[data-publish-select]"); + if (checkbox) { + if (checkbox.checked) selectedJobIds.add(checkbox.value); + else selectedJobIds.delete(checkbox.value); + updateSelectionUi(); + } + if (event.target.closest("[data-schedule-form]")) invalidatePreview(); + }); + + document.addEventListener("click", async (event) => { + const toggleEditor = event.target.closest("[data-toggle-publish-editor]"); + if (toggleEditor) { + const editor = toggleEditor.closest("[data-publish-row]")?.querySelector("[data-publish-editor]"); + if (editor) { + editor.hidden = !editor.hidden; + toggleEditor.textContent = editor.hidden ? "展开编辑" : "收起编辑"; + } + return; + } + + const publishNowButton = event.target.closest("[data-publish-now]"); + if (publishNowButton) { + const row = publishNowButton.closest("[data-publish-row]"); + const jobId = row?.dataset.jobId; + if (!jobId) return; + publishNowButton.disabled = true; + publishNowButton.textContent = "发送中…"; + try { + const result = await window.apiFetch(`/api/publish/jobs/${jobId}/publish-now`, { method: "POST" }); + updateRowFromJob(result.job || { id: jobId, status: result.status?.toUpperCase() }); + selectedJobIds.delete(jobId); + updateSelectionUi(); + showMessage(result.message || "单条任务已执行。", result.status === "failed" ? "error" : "success"); + } catch (error) { + showMessage(`立即发送失败:${error.message}`, "error"); + } finally { + publishNowButton.disabled = false; + publishNowButton.textContent = "立即发送"; + } + return; + } + + const clearScheduleButton = event.target.closest("[data-clear-schedule]"); + if (clearScheduleButton) { + const row = clearScheduleButton.closest("[data-publish-row]"); + const jobId = row?.dataset.jobId; + if (!jobId) return; + try { + const data = await window.apiFetch("/api/publish/jobs/schedule-batch", { + method: "PATCH", + body: JSON.stringify({ job_ids: [jobId], action: "clear", timezone: timezoneName }), + }); + updateRowFromJob(data.jobs?.[0]); + row.querySelector("[data-row-schedule]").textContent = "未排期"; + showMessage("已取消排期,任务回到等待安排状态。", "success"); + } catch (error) { + showMessage(`取消排期失败:${error.message}`, "error"); + } + } + }); + + document.querySelectorAll("[data-publish-editor]").forEach((form) => { + form.addEventListener("submit", async (event) => { + event.preventDefault(); + const row = form.closest("[data-publish-row]"); + const jobId = row?.dataset.jobId; + const resultNode = form.querySelector("[data-editor-result]"); + const payload = { + title: String(form.elements.title.value || "").trim(), + description: String(form.elements.description.value || "").trim(), + tags: String(form.elements.tags.value || "").trim(), + visibility: String(form.elements.visibility.value || "public"), + cover_file_path: String(form.elements.cover_file_path.value || ""), + cover_time_seconds: Number(form.elements.cover_time_seconds.value || 0), + allow_download: Boolean(form.elements.allow_download.checked), + bilibili_tid: String(form.elements.bilibili_tid.value || "娱乐"), + bilibili_copyright: String(form.elements.bilibili_copyright.value || "original"), + bilibili_source: String(form.elements.bilibili_source.value || ""), + }; + try { + const data = await window.apiFetch(`/api/publish/jobs/${jobId}/send-content`, { method: "PATCH", body: JSON.stringify(payload) }); + updateRowFromJob(data.job); + if (resultNode) resultNode.textContent = "已保存"; + } catch (error) { + if (resultNode) resultNode.textContent = `保存失败:${error.message}`; + } + }); + }); + + document.querySelector("[data-open-schedule-drawer]")?.addEventListener("click", openDrawer); + document.querySelector("[data-close-schedule-drawer]")?.addEventListener("click", closeDrawer); + drawerBackdrop?.addEventListener("click", closeDrawer); + document.querySelector("[data-clear-selection]")?.addEventListener("click", () => { + selectedJobIds.clear(); + updateSelectionUi(); + }); + document.querySelector("[data-expand-selected]")?.addEventListener("click", () => { + selectedRows().forEach((row) => { + const editor = row.querySelector("[data-publish-editor]"); + if (editor) editor.hidden = false; + }); + }); + + document.querySelector("[data-send-selected]")?.addEventListener("click", async () => { + const ids = Array.from(selectedJobIds); + for (const jobId of ids) { + try { + const result = await window.apiFetch(`/api/publish/jobs/${jobId}/publish-now`, { method: "POST" }); + updateRowFromJob(result.job || { id: jobId, status: result.status?.toUpperCase() }); + } catch (error) { + showMessage(`任务 ${jobId} 发送失败:${error.message}`, "error"); + break; + } + } + selectedJobIds.clear(); + updateSelectionUi(); + }); + + scheduleForm?.elements.interval_preset?.addEventListener("change", () => { + const custom = scheduleForm.elements.interval_preset.value === "custom"; + document.querySelector("[data-custom-interval]").hidden = !custom; + }); + + document.querySelector("[data-preview-schedule]")?.addEventListener("click", async () => { + const payload = schedulePayload("apply"); + if (!payload.start_at_local) { + showMessage("请先选择起始时间。", "error"); + return; + } + try { + const data = await window.apiFetch("/api/publish/schedules/preview", { method: "POST", body: JSON.stringify(payload) }); + previewList.innerHTML = ""; + (data.schedule || []).forEach((item) => { + const row = document.querySelector(`[data-publish-row][data-job-id="${CSS.escape(item.job_id)}"]`); + const line = document.createElement("div"); + line.innerHTML = `${row?.querySelector("[data-row-title]")?.textContent || item.job_id}`; + previewList.appendChild(line); + }); + latestPreviewSignature = previewSignature(payload); + confirmScheduleButton.disabled = false; + } catch (error) { + showMessage(`排期预览失败:${error.message}`, "error"); + } + }); + + scheduleForm?.addEventListener("submit", async (event) => { + event.preventDefault(); + const payload = schedulePayload("apply"); + if (latestPreviewSignature !== previewSignature(payload)) { + showMessage("排期参数已变化,请重新预览后再确认。", "error"); + return; + } + try { + const data = await window.apiFetch("/api/publish/jobs/schedule-batch", { method: "PATCH", body: JSON.stringify(payload) }); + (data.jobs || []).forEach(updateRowFromJob); + (data.schedule || []).forEach((item) => { + const row = document.querySelector(`[data-publish-row][data-job-id="${CSS.escape(item.job_id)}"]`); + const timeNode = row?.querySelector("[data-row-schedule]"); + if (timeNode) timeNode.textContent = item.scheduled_at_local_display; + }); + showMessage(data.message || "排期已保存。", "success"); + selectedJobIds.clear(); + updateSelectionUi(); + closeDrawer(); + } catch (error) { + showMessage(`排期保存失败:${error.message}`, "error"); + } + }); + + document.querySelector("[data-supplement-publish-jobs]")?.addEventListener("click", async (event) => { + const button = event.currentTarget; + button.disabled = true; + try { + const data = await window.apiFetch("/api/publish/queue/refresh?use_ai=false", { method: "POST" }); + showMessage(data.message || "缺失任务已补充。", "success"); + } catch (error) { + showMessage(`补充任务失败:${error.message}`, "error"); + } finally { + button.disabled = false; + } + }); + + if (scheduleForm?.elements.start_at_local) { + scheduleForm.elements.start_at_local.value = localDatetimeValue(new Date(Date.now() + 10 * 60 * 1000)); + } + document.querySelector("[data-current-timezone]").textContent = timezoneName; + updateSelectionUi(); +} diff --git a/app/templates/base.html b/app/templates/base.html index a905f14..1de9ae4 100644 --- a/app/templates/base.html +++ b/app/templates/base.html @@ -8,6 +8,7 @@ + {% block extra_head %}{% endblock %} @@ -60,7 +61,7 @@ - {% block extra_scripts %}{% endblock %} + {% block extra_scripts %}{% endblock %} diff --git a/app/templates/publish.html b/app/templates/publish.html index 62e028e..64ed5e1 100644 --- a/app/templates/publish.html +++ b/app/templates/publish.html @@ -2,314 +2,134 @@ {% block title %}发送中心 - {{ settings.app_name }}{% endblock %} +{% macro compact_job(job, section) %} +
+ + +
+

+ {{ job.platform_label }} + {{ job.status_label }} + {{ job.publish_mode_label }} +

+ {{ job.title or job.output_file_name or job.id }} + {{ job.task_name or "未命名任务" }} · {{ job.output_file_name or job.output_clip_id }} +
+
+ 计划时间 + +
+
+ {% if section == "pending" %}{% endif %} + {% if job.status in ["WAITING", "SCHEDULED", "FAILED"] %}{% endif %} + {% if section == "scheduled" and job.status == "SCHEDULED" %}{% endif %} +
+ {% if section == "pending" %} + + {% endif %} +
+{% endmacro %} + {% block content %}
-

Send Center 2.0

-

抖音 + B站发送中心

-

切好的视频、自动封面帧、AI 标题和平台 #话题在这里排队;确认后由 opencli 使用 Chrome 登录态逐条发送。

-
-
- - - +

Send Center

+

发送中心

+

发布任务由全自动流程直接生成;在这里复核、排期,并通过已登录的 Windows Chrome 发送到目标平台。

+
-{% if publish_message %} -
{{ publish_message }}
-{% endif %} - -{% if not opencli_available %} -
- {{ opencli_status.message }} -

请继续使用 Docker 主页面 http://127.0.0.1:8001,不要切到第二个网页。

-

推荐处理:在项目目录运行 {{ opencli_status.restart_command }},脚本会启动 Windows opencli 辅助服务、刷新 Docker,并打开 {{ opencli_status.publish_url }}

-

页面打开后按 Ctrl + F5 强制刷新,再回到发送中心测试自动发送。

-
-{% endif %} - -