1、后台菜单增加定时任务和银行交易

2、增加后台配置银行账户和请求网址
2026-08-13 15:12:16 +08:00
parent 33a5cc4f94
commit 7dcedf974d
22 changed files with 1741 additions and 3 deletions
@@ -0,0 +1,3 @@
+from app.admin_api.bank.routes import router
+
+__all__ = ["router"]
@@ -0,0 +1,173 @@
+"""银行账户管理与交易查询后台路由。"""
+
+from fastapi import APIRouter, Depends, HTTPException, Path, Query
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.dependencies import get_admin_user, get_db
+from app.models.bank_account import BankAccount
+from app.models.user import User
+from app.services.bank.service import query_transactions_with_log
+from app.utils.id_gen import generate_id
+
+router = APIRouter(prefix="/admin/bank", tags=["admin-bank"])
+
+
+# ============================================================
+# 银行账户 CRUD
+# ============================================================
+
+
+@router.get("/accounts")
+async def list_accounts(
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """列出所有银行账户。"""
+    result = await db.execute(select(BankAccount).order_by(BankAccount.is_default.desc(), BankAccount.created_at.desc()))
+    accounts = result.scalars().all()
+    return {"items": [
+        {
+            "id": a.id,
+            "account_name": a.account_name,
+            "bank_name": a.bank_name,
+            "account_no": a.account_no,
+            "is_active": a.is_active,
+            "is_default": a.is_default,
+            "description": a.description,
+            "created_at": a.created_at.isoformat() if a.created_at else None,
+            "updated_at": a.updated_at.isoformat() if a.updated_at else None,
+        }
+        for a in accounts
+    ]}
+
+
+@router.post("/accounts")
+async def create_account(
+    body: dict,
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """新增银行账户。"""
+    account_name = (body.get("account_name") or "").strip()
+    bank_name = (body.get("bank_name") or "").strip()
+    account_no = (body.get("account_no") or "").strip()
+    if not account_name or not bank_name or not account_no:
+        raise HTTPException(status_code=400, detail="账户名称、开户银行、银行账号不能为空")
+
+    # 检查账号唯一性
+    existing = await db.execute(select(BankAccount).where(BankAccount.account_no == account_no))
+    if existing.scalar_one_or_none():
+        raise HTTPException(status_code=409, detail="该银行账号已存在")
+
+    is_default = bool(body.get("is_default", False))
+    # 如果设为默认，取消其他默认
+    if is_default:
+        await db.execute(
+            BankAccount.__table__.update().where(BankAccount.is_default.is_(True)).values(is_default=False)
+        )
+
+    account = BankAccount(
+        id=generate_id(),
+        account_name=account_name,
+        bank_name=bank_name,
+        account_no=account_no,
+        is_active=bool(body.get("is_active", True)),
+        is_default=is_default,
+        description=body.get("description"),
+    )
+    db.add(account)
+    await db.commit()
+    return {"id": account.id, "message": "创建成功"}
+
+
+@router.put("/accounts/{account_id}")
+async def update_account(
+    account_id: str = Path(..., description="账户 ID"),
+    body: dict = ...,
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """编辑银行账户。"""
+    result = await db.execute(select(BankAccount).where(BankAccount.id == account_id))
+    account = result.scalar_one_or_none()
+    if account is None:
+        raise HTTPException(status_code=404, detail="账户不存在")
+
+    if "account_name" in body:
+        account.account_name = str(body["account_name"]).strip()
+    if "bank_name" in body:
+        account.bank_name = str(body["bank_name"]).strip()
+    if "account_no" in body:
+        new_no = str(body["account_no"]).strip()
+        if new_no != account.account_no:
+            existing = await db.execute(select(BankAccount).where(BankAccount.account_no == new_no))
+            if existing.scalar_one_or_none():
+                raise HTTPException(status_code=409, detail="该银行账号已存在")
+        account.account_no = new_no
+    if "is_active" in body:
+        account.is_active = bool(body["is_active"])
+    if "description" in body:
+        account.description = body.get("description")
+
+    if body.get("is_default"):
+        await db.execute(
+            BankAccount.__table__.update()
+            .where(BankAccount.is_default.is_(True))
+            .where(BankAccount.id != account_id)
+            .values(is_default=False)
+        )
+        account.is_default = True
+
+    await db.commit()
+    return {"message": "更新成功"}
+
+
+@router.delete("/accounts/{account_id}")
+async def delete_account(
+    account_id: str = Path(..., description="账户 ID"),
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """删除银行账户。"""
+    result = await db.execute(select(BankAccount).where(BankAccount.id == account_id))
+    account = result.scalar_one_or_none()
+    if account is None:
+        raise HTTPException(status_code=404, detail="账户不存在")
+    await db.delete(account)
+    await db.commit()
+    return {"message": "删除成功"}
+
+
+# ============================================================
+# 银行交易查询
+# ============================================================
+
+
+@router.get("/transactions")
+async def list_transactions(
+    account_id: str = Query(..., description="银行账户 ID"),
+    start_date: str = Query(..., description="开始日期 (YYYY-MM-DD)"),
+    end_date: str = Query(..., description="结束日期 (YYYY-MM-DD)"),
+    dc_flag: int | None = Query(None, description="借贷方向: 0-借/出金, 1-贷/入金, 不传返回全部"),
+    page: int = Query(1, ge=1, description="页码"),
+    page_size: int = Query(20, ge=1, le=100, description="每页数量"),
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """查询银行交易流水（带接口请求记录到文件日志）。"""
+    result = await db.execute(select(BankAccount).where(BankAccount.id == account_id))
+    account = result.scalar_one_or_none()
+    if account is None:
+        raise HTTPException(status_code=404, detail="银行账户不存在")
+
+    data = await query_transactions_with_log(
+        admin.id,
+        acct_no=account.account_no,
+        start_date=start_date,
+        end_date=end_date,
+        dc_flag=dc_flag,
+        page=page,
+        page_size=page_size,
+    )
+    return data
@@ -0,0 +1,3 @@
+from app.admin_api.scheduled_tasks.routes import router
+
+__all__ = ["router"]
@@ -0,0 +1,170 @@
+"""定时任务管理后台路由。"""
+
+import json
+
+from fastapi import APIRouter, Depends, HTTPException, Path
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.dependencies import get_admin_user, get_db
+from app.models.scheduled_task import ScheduledTask
+from app.models.user import User
+from app.utils.id_gen import generate_id
+
+router = APIRouter(prefix="/admin/scheduled-tasks", tags=["admin-scheduled-tasks"])
+
+
+def _task_to_dict(task: ScheduledTask) -> dict:
+    return {
+        "id": task.id,
+        "name": task.name,
+        "task_type": task.task_type,
+        "schedule": task.schedule,
+        "config": task.config,
+        "is_active": task.is_active,
+        "last_run_at": task.last_run_at,
+        "last_status": task.last_status,
+        "last_error": task.last_error,
+        "created_by": task.created_by,
+        "created_at": task.created_at.isoformat() if task.created_at else None,
+        "updated_at": task.updated_at.isoformat() if task.updated_at else None,
+    }
+
+
+@router.get("")
+async def list_tasks(
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """列出所有定时任务。"""
+    result = await db.execute(select(ScheduledTask).order_by(ScheduledTask.created_at.desc()))
+    tasks = result.scalars().all()
+    return {"items": [_task_to_dict(t) for t in tasks]}
+
+
+@router.post("")
+async def create_task(
+    body: dict,
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """创建定时任务。"""
+    name = (body.get("name") or "").strip()
+    task_type = (body.get("task_type") or "").strip()
+    schedule = (body.get("schedule") or "").strip()
+    if not name or not task_type or not schedule:
+        raise HTTPException(status_code=400, detail="任务名称、类型、调度表达式不能为空")
+    if task_type not in ("external_api", "internal_method"):
+        raise HTTPException(status_code=400, detail="任务类型必须为 external_api 或 internal_method")
+
+    config = body.get("config")
+    if isinstance(config, dict):
+        config = json.dumps(config, ensure_ascii=False)
+    elif isinstance(config, str):
+        # 验证 JSON 合法性
+        try:
+            json.loads(config)
+        except json.JSONDecodeError:
+            raise HTTPException(status_code=400, detail="config 不是合法的 JSON")
+
+    task = ScheduledTask(
+        id=generate_id(),
+        name=name,
+        task_type=task_type,
+        schedule=schedule,
+        config=config,
+        is_active=bool(body.get("is_active", True)),
+        created_by=admin.id,
+    )
+    db.add(task)
+    await db.commit()
+    return {"id": task.id, "message": "创建成功"}
+
+
+@router.put("/{task_id}")
+async def update_task(
+    task_id: str = Path(..., description="任务 ID"),
+    body: dict = ...,
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """更新定时任务。"""
+    result = await db.execute(select(ScheduledTask).where(ScheduledTask.id == task_id))
+    task = result.scalar_one_or_none()
+    if task is None:
+        raise HTTPException(status_code=404, detail="任务不存在")
+
+    if "name" in body:
+        task.name = str(body["name"]).strip()
+    if "task_type" in body:
+        t = body["task_type"]
+        if t not in ("external_api", "internal_method"):
+            raise HTTPException(status_code=400, detail="任务类型必须为 external_api 或 internal_method")
+        task.task_type = t
+    if "schedule" in body:
+        task.schedule = str(body["schedule"]).strip()
+    if "is_active" in body:
+        task.is_active = bool(body["is_active"])
+    if "config" in body:
+        config = body["config"]
+        if isinstance(config, dict):
+            config = json.dumps(config, ensure_ascii=False)
+        elif isinstance(config, str):
+            try:
+                json.loads(config)
+            except json.JSONDecodeError:
+                raise HTTPException(status_code=400, detail="config 不是合法的 JSON")
+        task.config = config
+
+    await db.commit()
+    return {"message": "更新成功"}
+
+
+@router.delete("/{task_id}")
+async def delete_task(
+    task_id: str = Path(..., description="任务 ID"),
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """删除定时任务。"""
+    result = await db.execute(select(ScheduledTask).where(ScheduledTask.id == task_id))
+    task = result.scalar_one_or_none()
+    if task is None:
+        raise HTTPException(status_code=404, detail="任务不存在")
+    await db.delete(task)
+    await db.commit()
+    return {"message": "删除成功"}
+
+
+@router.post("/{task_id}/run")
+async def run_task(
+    task_id: str = Path(..., description="任务 ID"),
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """手动执行一次定时任务。"""
+    result = await db.execute(select(ScheduledTask).where(ScheduledTask.id == task_id))
+    task = result.scalar_one_or_none()
+    if task is None:
+        raise HTTPException(status_code=404, detail="任务不存在")
+
+    from app.tasks.scheduled_tasks import execute_scheduled_task
+
+    execute_scheduled_task.apply_async(args=[task_id])
+    return {"message": "任务已提交执行"}
+
+
+@router.post("/{task_id}/toggle")
+async def toggle_task(
+    task_id: str = Path(..., description="任务 ID"),
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """启用/禁用定时任务。"""
+    result = await db.execute(select(ScheduledTask).where(ScheduledTask.id == task_id))
+    task = result.scalar_one_or_none()
+    if task is None:
+        raise HTTPException(status_code=404, detail="任务不存在")
+    task.is_active = not task.is_active
+    await db.commit()
+    return {"is_active": task.is_active, "message": "已启用" if task.is_active else "已禁用"}
@@ -15,6 +15,8 @@ from app.api.admin.contact import router as admin_contact_router
 from app.admin_api.api_keys import router as api_keys_admin_router
 from app.admin_api.api_model_pricings import router as api_model_pricings_admin_router
 from app.admin_api.vp_v3_quota import router as vp_v3_quota_admin_router
+from app.admin_api.bank import router as bank_admin_router
+from app.admin_api.scheduled_tasks import router as scheduled_tasks_admin_router

 router = APIRouter()
 router.include_router(video_prompt_schema_config_router)
@@ -32,3 +34,5 @@ router.include_router(admin_contact_router)
 router.include_router(api_keys_admin_router)
 router.include_router(api_model_pricings_admin_router)
 router.include_router(vp_v3_quota_admin_router)
+router.include_router(bank_admin_router)
+router.include_router(scheduled_tasks_admin_router)
@@ -131,6 +131,10 @@ class Settings(BaseSettings):

    CAPTCHA_ENABLED: bool = True

+    # 银行交易查询接口配置
+    BANK_API_BASE: str = ""
+    BANK_API_KEY: str = ""
+
    BASE_URL: str = ""

    CORS_ORIGINS: list[str] = ["*"]
@@ -43,6 +43,8 @@ from app.models.home_material import HomeMaterialAsset, HomeMaterialCategory, Ho
 from app.models.contact_request import ContactRequest
 from app.models.invoice import Invoice, InvoiceOrder
 from app.models.invoice_header import InvoiceHeader
+from app.models.bank_account import BankAccount
+from app.models.scheduled_task import ScheduledTask
 from app.models.private_portrait import PrivatePortraitProject, PrivatePortraitValidateSession, PrivatePortraitAssetGroup, PrivatePortraitAsset
 from app.models.api import ApiKey, ApiGenerationTask, ApiUsageLog, ApiKeyUpscaleConfig, ApiUpscaleLink

@@ -68,4 +70,5 @@ __all__ = [
    "ApiKey", "ApiGenerationTask", "ApiUsageLog", "ApiKeyUpscaleConfig", "ApiUpscaleLink",
    "ApiModelPricing",
    "Invoice", "InvoiceOrder", "InvoiceHeader",
+    "BankAccount", "ScheduledTask",
 ]
@@ -0,0 +1,18 @@
+"""银行账户信息模型。"""
+
+from sqlalchemy import Boolean, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import Base, TimestampMixin
+
+
+class BankAccount(Base, TimestampMixin):
+    __tablename__ = "bank_accounts"
+
+    id: Mapped[str] = mapped_column(String(32), primary_key=True)
+    account_name: Mapped[str] = mapped_column(String(128), nullable=False, comment="账户名称")
+    bank_name: Mapped[str] = mapped_column(String(128), nullable=False, comment="开户银行")
+    account_no: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, comment="银行账号")
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True, comment="是否启用")
+    is_default: Mapped[bool] = mapped_column(Boolean, default=False, comment="是否默认账户")
+    description: Mapped[str | None] = mapped_column(String(256), nullable=True, comment="备注")
@@ -0,0 +1,23 @@
+"""定时任务配置模型。"""
+
+from sqlalchemy import Boolean, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import Base, TimestampMixin
+
+
+class ScheduledTask(Base, TimestampMixin):
+    __tablename__ = "scheduled_tasks"
+
+    id: Mapped[str] = mapped_column(String(32), primary_key=True)
+    name: Mapped[str] = mapped_column(String(128), nullable=False, comment="任务名称")
+    task_type: Mapped[str] = mapped_column(String(32), nullable=False, comment="类型: external_api / internal_method")
+    schedule: Mapped[str] = mapped_column(String(128), nullable=False, comment="Cron 表达式或间隔秒数")
+    config: Mapped[str | None] = mapped_column(Text, nullable=True, comment="任务配置 JSON")
+    # external_api config: {url, method, headers, payload}
+    # internal_method config: {module, function, args}
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True, comment="是否启用")
+    last_run_at: Mapped[str | None] = mapped_column(String(64), nullable=True, comment="最后执行时间 ISO")
+    last_status: Mapped[str | None] = mapped_column(String(16), nullable=True, comment="最后执行状态: success / error")
+    last_error: Mapped[str | None] = mapped_column(Text, nullable=True, comment="最后执行错误信息")
+    created_by: Mapped[str | None] = mapped_column(String(32), nullable=True, comment="创建者管理员 ID")
@@ -0,0 +1,4 @@
+from app.services.bank.client import query_bank_transactions, BankApiError
+from app.services.bank.service import query_transactions_with_log
+
+__all__ = ["query_bank_transactions", "BankApiError", "query_transactions_with_log"]
@@ -0,0 +1,76 @@
+"""银行交易查询 API 客户端。"""
+
+import httpx
+
+from app.config import settings
+
+
+class BankApiError(RuntimeError):
+    """银行接口错误。"""
+
+    def __init__(
+        self,
+        message: str,
+        retryable: bool = False,
+        http_status: int | None = None,
+    ):
+        super().__init__(message)
+        self.retryable = retryable
+        self.http_status = http_status
+
+
+async def query_bank_transactions(
+    acct_no: str,
+    start_date: str,
+    end_date: str,
+    dc_flag: int | None = None,
+    page: int = 1,
+    page_size: int = 20,
+) -> dict:
+    """查询银行交易流水。
+
+    调用外部银行接口，按账号和日期范围查询交易记录。
+
+    :param acct_no: 我方银行账号
+    :param start_date: 开始日期 (YYYY-MM-DD)
+    :param end_date: 结束日期 (YYYY-MM-DD)
+    :param dc_flag: 借贷方向 (0-借/出金, 1-贷/入金, None-全部)
+    :param page: 页码
+    :param page_size: 每页数量 (最大 100)
+    :return: 接口返回的原始数据
+    """
+    base = (settings.BANK_API_BASE or "").rstrip("/")
+    if not base:
+        raise BankApiError("银行接口地址未配置 (BANK_API_BASE)", retryable=False)
+
+    url = f"{base}/api/v1/internal/get-blank-transfer-acct-time"
+    headers = {
+        "api-key": settings.BANK_API_KEY or "",
+        "Content-Type": "application/json",
+    }
+    payload: dict = {
+        "acct_no": acct_no,
+        "start_date": start_date,
+        "end_date": end_date,
+        "page": page,
+        "page_size": min(page_size, 100),
+    }
+    if dc_flag is not None:
+        payload["dc_flag"] = dc_flag
+
+    timeout = httpx.Timeout(30.0)
+    try:
+        async with httpx.AsyncClient(timeout=timeout, follow_redirects=True) as client:
+            response = await client.post(url, headers=headers, json=payload)
+            response.raise_for_status()
+            return response.json()
+    except httpx.HTTPStatusError as e:
+        raise BankApiError(
+            f"银行接口返回错误: HTTP {e.response.status_code}",
+            retryable=e.response.status_code >= 500,
+            http_status=e.response.status_code,
+        ) from e
+    except httpx.TimeoutException as e:
+        raise BankApiError("银行接口请求超时", retryable=True) from e
+    except httpx.RequestError as e:
+        raise BankApiError(f"银行接口请求失败: {e}", retryable=True) from e
@@ -0,0 +1,50 @@
+"""银行接口请求文件日志记录器。
+
+使用 get_logger 模式，日志输出到 log/bank_api/bank_api-YYYY-MM-DD.log。
+"""
+
+import json
+from datetime import datetime, timezone
+
+from app.utils.logger import get_logger
+
+bank_api_logger = get_logger("bank_api", "bank_api")
+
+
+def log_bank_api_request(
+    acct_no: str,
+    request_params: dict | None,
+    response_data: dict | None,
+    status_code: int | None,
+    is_success: bool,
+    error_msg: str | None = None,
+    admin_id: str | None = None,
+    duration_ms: int | None = None,
+) -> None:
+    """记录银行接口请求到日志文件。
+
+    :param acct_no: 查询的银行账号
+    :param request_params: 请求参数
+    :param response_data: 响应数据
+    :param status_code: HTTP 状态码
+    :param is_success: 是否成功
+    :param error_msg: 错误信息
+    :param admin_id: 操作管理员 ID
+    :param duration_ms: 请求耗时（毫秒）
+    """
+    log_data = {
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "acct_no": acct_no,
+        "request_params": request_params,
+        "response_data": response_data,
+        "status_code": status_code,
+        "is_success": is_success,
+        "error_msg": error_msg,
+        "admin_id": admin_id,
+        "duration_ms": duration_ms,
+    }
+    line = json.dumps(log_data, ensure_ascii=False, default=str)
+    if is_success:
+        bank_api_logger.info(line)
+    else:
+        bank_api_logger.error(line)
@@ -0,0 +1,47 @@
+"""银行交易查询服务 — 接口请求日志写入文件。"""
+
+import time
+
+from app.services.bank.client import query_bank_transactions, BankApiError
+from app.services.bank.file_logger import log_bank_api_request
+
+
+async def query_transactions_with_log(
+    admin_id: str,
+    **kwargs,
+) -> dict:
+    """查询银行流水并记录接口日志到文件。
+
+    :param admin_id: 操作管理员 ID
+    :param kwargs: 透传给 query_bank_transactions 的参数
+    :return: 接口返回的原始数据
+    """
+    acct_no = kwargs.get("acct_no", "")
+    request_params = {k: v for k, v in kwargs.items()}
+    start_time = time.monotonic()
+    try:
+        result = await query_bank_transactions(**kwargs)
+        duration_ms = int((time.monotonic() - start_time) * 1000)
+        log_bank_api_request(
+            acct_no=acct_no,
+            request_params=request_params,
+            response_data=result,
+            status_code=200,
+            is_success=True,
+            admin_id=admin_id,
+            duration_ms=duration_ms,
+        )
+        return result
+    except BankApiError as e:
+        duration_ms = int((time.monotonic() - start_time) * 1000)
+        log_bank_api_request(
+            acct_no=acct_no,
+            request_params=request_params,
+            response_data=None,
+            status_code=e.http_status,
+            is_success=False,
+            error_msg=str(e),
+            admin_id=admin_id,
+            duration_ms=duration_ms,
+        )
+        raise
@@ -40,6 +40,7 @@ CELERY_TASK_IMPORTS = (
    "app.tasks.api_generation_tasks",
    "app.tasks.api_recovery_tasks",
    "app.tasks.api_upscale_tasks",
+    "app.tasks.scheduled_tasks",
 )


@@ -292,6 +293,85 @@ else:
    celery_app = None


+def _parse_schedule_to_celery(schedule_str: str):
+    """将 schedule 字符串解析为 Celery 可识别的调度值。
+
+    - 纯数字：视为间隔秒数（返回 int）
+    - cron 表达式 (5 字段空格分隔)：返回 crontab 对象
+    """
+    from celery.schedules import crontab
+
+    s = (schedule_str or "").strip()
+    if not s:
+        return None
+    # 纯数字 → 间隔秒数
+    if s.isdigit():
+        return int(s)
+    # cron 表达式 (分 时 日 月 周)
+    parts = s.split()
+    if len(parts) == 5:
+        try:
+            return crontab(
+                minute=parts[0],
+                hour=parts[1],
+                day_of_month=parts[2],
+                month_of_year=parts[3],
+                day_of_week=parts[4],
+            )
+        except Exception:
+            logger.exception("解析 cron 表达式失败: %s", s)
+            return None
+    logger.warning("无法解析 schedule 表达式: %s", s)
+    return None
+
+
+@celery_app.on_after_configure.connect  # type: ignore
+def _setup_dynamic_beat_tasks(sender, **kwargs):
+    """从数据库加载活跃定时任务并注册到 Beat 调度。
+
+    通过 @celery_app.on_after_configure.connect 在 Celery 配置完成后执行，
+    适用于 Worker 和 Beat 启动场景。
+    """
+    if celery_app is None:
+        return
+
+    async def _load():
+        from sqlalchemy import select
+
+        from app.models.base import async_session
+        from app.models.scheduled_task import ScheduledTask
+
+        async with async_session() as db:
+            result = await db.execute(
+                select(ScheduledTask).where(ScheduledTask.is_active.is_(True))
+            )
+            tasks = result.scalars().all()
+            return tasks
+
+    try:
+        active_tasks = run_async(_load())
+    except Exception:
+        logger.exception("加载定时任务失败，跳过动态 Beat 注册")
+        return
+
+    for task in active_tasks:
+        schedule_val = _parse_schedule_to_celery(task.schedule)
+        if schedule_val is None:
+            logger.warning("定时任务 %s schedule 无效，跳过注册: %s", task.id, task.schedule)
+            continue
+        beat_key = f"dynamic-scheduled-task-{task.id}"
+        sender.conf.beat_schedule[beat_key] = {
+            "task": "execute_scheduled_task",
+            "schedule": schedule_val,
+            "args": (task.id,),
+            "options": {"queue": RECOVERY_QUEUE},
+        }
+        logger.info(
+            "动态注册定时任务到 Beat: %s (%s) schedule=%s",
+            task.name, task.id, task.schedule,
+        )
+
+
 async def _try_acquire_startup_recovery_lock() -> bool:
    """任意 worker 启动时都可尝试抢恢复投递锁，避免依赖 hostname 命名。"""
    from app.services.redis_registry_service import redis_acquire_lock
@@ -0,0 +1,138 @@
+"""定时任务执行器。
+
+支持两种任务类型：
+- external_api: 调用外部 HTTP 接口
+- internal_method: 动态导入并执行内部函数
+
+任务在 Celery Worker 中执行，通过 run_async 桥接异步操作。
+"""
+
+import json
+import logging
+import time
+from datetime import datetime, timezone
+
+import httpx
+from sqlalchemy import select
+
+from app.models.base import async_session
+from app.models.scheduled_task import ScheduledTask
+from app.tasks.async_runner import run_async
+from app.tasks.celery_app import celery_app
+
+logger = logging.getLogger("video_gen")
+
+
+def _update_task_status(task_id: str, status: str, error_msg: str | None = None) -> None:
+    """更新任务最后执行状态。"""
+
+    async def _do():
+        async with async_session() as db:
+            result = await db.execute(select(ScheduledTask).where(ScheduledTask.id == task_id))
+            task = result.scalar_one_or_none()
+            if task is None:
+                return
+            task.last_run_at = datetime.now(timezone.utc).isoformat()
+            task.last_status = status
+            task.last_error = error_msg
+            await db.commit()
+
+    run_async(_do())
+
+
+def _execute_external_api(config: str | None) -> dict:
+    """执行外部 API 调用。"""
+    cfg = json.loads(config or "{}")
+    url = cfg.get("url", "").strip()
+    method = cfg.get("method", "GET").upper()
+    headers = cfg.get("headers") or {}
+    payload = cfg.get("payload")
+    timeout_val = float(cfg.get("timeout", 30))
+
+    if not url:
+        raise ValueError("外部接口地址 (url) 未配置")
+
+    start = time.monotonic()
+    try:
+        with httpx.Client(timeout=httpx.Timeout(timeout_val, connect=10.0)) as client:
+            if method == "GET":
+                resp = client.get(url, headers=headers, params=payload)
+            elif method == "DELETE":
+                resp = client.delete(url, headers=headers, params=payload)
+            else:
+                resp = client.request(method, url, headers=headers, json=payload)
+            duration_ms = int((time.monotonic() - start) * 1000)
+            resp.raise_for_status()
+            return {
+                "status_code": resp.status_code,
+                "duration_ms": duration_ms,
+                "body": resp.text[:2000],
+            }
+    except httpx.HTTPError as e:
+        duration_ms = int((time.monotonic() - start) * 1000)
+        raise RuntimeError(f"外部接口请求失败: {e} (耗时 {duration_ms}ms)") from e
+
+
+def _execute_internal_method(config: str | None) -> dict:
+    """执行内部方法调用。"""
+    cfg = json.loads(config or "{}")
+    module_path = cfg.get("module", "").strip()
+    function_name = cfg.get("function", "").strip()
+    args = cfg.get("args") or []
+
+    if not module_path or not function_name:
+        raise ValueError("内部方法需要指定 module 和 function")
+
+    import importlib
+
+    module = importlib.import_module(module_path)
+    func = getattr(module, function_name, None)
+    if func is None or not callable(func):
+        raise ValueError(f"模块 {module_path} 中不存在可调用函数 {function_name}")
+
+    start = time.monotonic()
+    result = func(*args)
+    duration_ms = int((time.monotonic() - start) * 1000)
+    return {
+        "duration_ms": duration_ms,
+        "result": str(result)[:1000] if result is not None else None,
+    }
+
+
+@celery_app.task(name="execute_scheduled_task", bind=True, ignore_result=True)  # type: ignore[call-arg]
+def execute_scheduled_task(self, task_id: str):
+    """执行定时任务（Celery 任务入口）。
+
+    通过 run_async 桥接到异步上下文读取任务配置并执行。
+    """
+
+    async def _run():
+        async with async_session() as db:
+            result = await db.execute(select(ScheduledTask).where(ScheduledTask.id == task_id))
+            task = result.scalar_one_or_none()
+            if task is None:
+                logger.warning("定时任务不存在: %s", task_id)
+                return
+            if not task.is_active:
+                logger.info("定时任务已禁用，跳过执行: %s", task_id)
+                return
+
+            task_config = task.config
+            task_type = task.task_type
+
+        try:
+            if task_type == "external_api":
+                exec_result = _execute_external_api(task_config)
+            elif task_type == "internal_method":
+                exec_result = _execute_internal_method(task_config)
+            else:
+                raise ValueError(f"未知的任务类型: {task_type}")
+
+            _update_task_status(task_id, "success")
+            logger.info("定时任务执行成功: %s (%s) -> %s", task_id, task_type, exec_result)
+        except Exception as e:
+            error_msg = str(e)
+            _update_task_status(task_id, "error", error_msg)
+            logger.exception("定时任务执行失败: %s (%s)", task_id, task_type)
+
+    run_async(_run())