Merge branch 'main' of https://gitlab.minzhong.cn/mz/video-gen

video-gen-api/alembic/versions/20260814_bank_transactions.py

@@ -0,0 +1,51 @@
+"""银行交易流水表
+
+Revision ID: 20260814_bank_tx
+Revises: 20260813_bank_scheduled
+Create Date: 2026-08-14 10:00:00.000000
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = '20260814_bank_tx'
+down_revision = '20260813_bank_scheduled'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        'bank_transactions',
+        sa.Column('id', sa.String(32), primary_key=True),
+        sa.Column('account_id', sa.String(32), nullable=False, comment='银行账户ID'),
+        sa.Column('account_no', sa.String(64), nullable=False, comment='银行账号'),
+        sa.Column('transaction_no', sa.String(128), unique=True, nullable=True, comment='交易流水号(唯一)'),
+        sa.Column('transaction_time', sa.DateTime(timezone=True), nullable=True, comment='交易时间'),
+        sa.Column('transaction_amount', sa.String(32), nullable=False, comment='交易金额'),
+        sa.Column('balance_direction', sa.String(8), nullable=True, comment='借贷方向: DR/CR'),
+        sa.Column('balance_after', sa.String(32), nullable=True, comment='交易后余额'),
+        sa.Column('counterparty_name', sa.String(128), nullable=True, comment='对方户名'),
+        sa.Column('counterparty_account', sa.String(64), nullable=True, comment='对方账号'),
+        sa.Column('counterparty_bank', sa.String(128), nullable=True, comment='对方开户行'),
+        sa.Column('remark', sa.Text, nullable=True, comment='摘要/备注'),
+        sa.Column('digest_code', sa.String(64), nullable=True, comment='摘要码'),
+        sa.Column('purpose', sa.String(256), nullable=True, comment='用途'),
+        sa.Column('raw_data', sa.Text, nullable=True, comment='接口返回原始JSON'),
+        sa.Column('sync_batch', sa.String(32), nullable=True, comment='同步批次号'),
+        sa.Column('is_synced', sa.Boolean, default=True, comment='是否同步成功'),
+    )
+    op.create_index('ix_bank_transactions_account_id', 'bank_transactions', ['account_id'])
+    op.create_index('ix_bank_transactions_account_no', 'bank_transactions', ['account_no'])
+    op.create_index('ix_bank_transactions_transaction_time', 'bank_transactions', ['transaction_time'])
+    op.create_index('ix_bank_transactions_sync_batch', 'bank_transactions', ['sync_batch'])
+    op.create_index('ix_bank_transactions_account_time', 'bank_transactions', ['account_no', 'transaction_time'])
+
+
+def downgrade():
+    op.drop_index('ix_bank_transactions_account_time', table_name='bank_transactions')
+    op.drop_index('ix_bank_transactions_sync_batch', table_name='bank_transactions')
+    op.drop_index('ix_bank_transactions_transaction_time', table_name='bank_transactions')
+    op.drop_index('ix_bank_transactions_account_no', table_name='bank_transactions')
+    op.drop_index('ix_bank_transactions_account_id', table_name='bank_transactions')
+    op.drop_table('bank_transactions')

video-gen-api/app/admin_api/scheduled_tasks/routes.py

@@ -50,12 +50,12 @@ async def create_task(
 ):
     """创建定时任务。"""
     name = (body.get("name") or "").strip()
-    task_type = (body.get("task_type") or "").strip()
+    task_type = (body.get("task_type") or "internal_method").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")
+    if not name or not schedule:
+        raise HTTPException(status_code=400, detail="任务名称、调度表达式不能为空")
+    if task_type != "internal_method":
+        raise HTTPException(status_code=400, detail="任务类型仅支持 internal_method")
 
     config = body.get("config")
     if isinstance(config, dict):
@@ -98,8 +98,8 @@ async def update_task(
         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")
+        if t != "internal_method":
+            raise HTTPException(status_code=400, detail="任务类型仅支持 internal_method")
         task.task_type = t
     if "schedule" in body:
         task.schedule = str(body["schedule"]).strip()

video-gen-api/app/api/v1/__init__.py

@@ -1,6 +1,7 @@
 from fastapi import APIRouter
 
 from app.api.v1.auth import router as auth_router
+from app.api.v1.bank import router as bank_router
 from app.api.v1.projects import router as projects_router
 from app.api.v1.generation import router as generation_router
 from app.api.v1.credits import router as credits_router
@@ -41,6 +42,7 @@ from app.api.v1.invoice_headers import router as invoice_headers_router
 
 api_router = APIRouter()
 api_router.include_router(auth_router)
+api_router.include_router(bank_router)
 api_router.include_router(projects_router)
 api_router.include_router(generation_router)
 api_router.include_router(credits_router)

video-gen-api/app/api/v1/bank.py

@@ -0,0 +1,83 @@
+"""银行账户前端查询接口。"""
+
+from fastapi import APIRouter, Depends
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.dependencies import get_current_user, get_db
+from app.models.bank_account import BankAccount
+from app.models.user import User
+
+router = APIRouter(prefix="/bank", tags=["bank"])
+
+
+@router.get("/default-account")
+async def get_default_account(
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """获取默认银行账户信息（前端展示用）。
+
+    优先返回标记为默认的启用账户；如果没有默认账户，返回第一个启用账户。
+    """
+    # 先查默认账户
+    result = await db.execute(
+        select(BankAccount).where(
+            BankAccount.is_active.is_(True),
+            BankAccount.is_default.is_(True),
+        )
+    )
+    account = result.scalar_one_or_none()
+
+    # 没有默认账户则取第一个启用账户
+    if account is None:
+        result = await db.execute(
+            select(BankAccount).where(
+                BankAccount.is_active.is_(True),
+            ).order_by(BankAccount.created_at.asc())
+        )
+        account = result.scalar_one_or_none()
+
+    if account is None:
+        return {
+            "has_account": False,
+            "account": None,
+        }
+
+    return {
+        "has_account": True,
+        "account": {
+            "id": account.id,
+            "accountName": account.account_name,
+            "bankName": account.bank_name,
+            "accountNo": account.account_no,
+            "description": account.description,
+        },
+    }
+
+
+@router.get("/accounts")
+async def list_active_accounts(
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """获取所有启用的银行账户列表（前端展示用）。"""
+    result = await db.execute(
+        select(BankAccount).where(
+            BankAccount.is_active.is_(True),
+        ).order_by(BankAccount.is_default.desc(), BankAccount.created_at.asc())
+    )
+    accounts = result.scalars().all()
+    return {
+        "items": [
+            {
+                "id": a.id,
+                "accountName": a.account_name,
+                "bankName": a.bank_name,
+                "accountNo": a.account_no,
+                "isDefault": a.is_default,
+                "description": a.description,
+            }
+            for a in accounts
+        ]
+    }

video-gen-api/app/models/bank_transaction.py

@@ -0,0 +1,45 @@
+"""银行交易流水模型。"""
+
+from sqlalchemy import Boolean, DateTime, Index, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.models.base import Base
+
+
+class BankTransaction(Base):
+    __tablename__ = "bank_transactions"
+
+    id: Mapped[str] = mapped_column(String(32), primary_key=True)
+
+    # 账户信息
+    account_id: Mapped[str] = mapped_column(String(32), index=True, nullable=False, comment="银行账户ID")
+    account_no: Mapped[str] = mapped_column(String(64), index=True, nullable=False, comment="银行账号")
+
+    # 交易信息
+    transaction_no: Mapped[str | None] = mapped_column(String(128), unique=True, nullable=True, comment="交易流水号(唯一)")
+    transaction_time: Mapped[DateTime | None] = mapped_column(DateTime(timezone=True), index=True, nullable=True, comment="交易时间")
+    transaction_amount: Mapped[str] = mapped_column(String(32), nullable=False, comment="交易金额(字符串保持精度)")
+    balance_direction: Mapped[str | None] = mapped_column(String(8), nullable=True, comment="借贷方向: DR-借/出金, CR-贷/入金")
+    balance_after: Mapped[str | None] = mapped_column(String(32), nullable=True, comment="交易后余额")
+
+    # 对方信息
+    counterparty_name: Mapped[str | None] = mapped_column(String(128), nullable=True, comment="对方户名")
+    counterparty_account: Mapped[str | None] = mapped_column(String(64), nullable=True, comment="对方账号")
+    counterparty_bank: Mapped[str | None] = mapped_column(String(128), nullable=True, comment="对方开户行")
+
+    # 交易摘要
+    remark: Mapped[str | None] = mapped_column(Text, nullable=True, comment="摘要/备注")
+    digest_code: Mapped[str | None] = mapped_column(String(64), nullable=True, comment="摘要码")
+    purpose: Mapped[str | None] = mapped_column(String(256), nullable=True, comment="用途")
+
+    # 原始数据
+    raw_data: Mapped[str | None] = mapped_column(Text, nullable=True, comment="接口返回原始JSON")
+
+    # 同步信息
+    sync_batch: Mapped[str | None] = mapped_column(String(32), index=True, nullable=True, comment="同步批次号")
+    is_synced: Mapped[bool] = mapped_column(Boolean, default=True, comment="是否同步成功")
+
+    __table_args__ = (
+        Index("ix_bank_transactions_account_time", "account_no", "transaction_time"),
+        Index("ix_bank_transactions_sync_batch", "sync_batch"),
+    )

video-gen-api/app/services/bank/sync_service.py

@@ -0,0 +1,470 @@
+"""银行交易流水同步服务。
+
+提供内部方法供定时任务调用，从外部银行接口拉取交易流水并存储到数据库。
+
+定时任务配置 JSON 格式:
+{
+    "url": "http://ceshi.web.minzhong.cn/api/api/v1/internal/get-blank-transfer-acct-time",
+    "api_key": "jixekCxm8piLFi0AlfA24bDtCKJ82bfu",
+    "acct_no": "110972289710001",
+    "start_date": "2026-08-01",
+    "end_date": "2026-08-14",
+    "dc_flag": null
+}
+
+服务会自动翻页查询所有数据，直到拉完全部流水。
+"""
+
+import json
+import logging
+from datetime import datetime, timedelta, timezone
+
+import httpx
+from sqlalchemy import select
+
+from app.models.base import async_session
+from app.models.bank_account import BankAccount
+from app.models.bank_transaction import BankTransaction
+from app.services.bank.file_logger import log_bank_api_request
+from app.utils.id_gen import generate_id
+
+logger = logging.getLogger("video_gen")
+
+# 每页条数（固定）
+_PAGE_SIZE = 100
+
+
+def sync_bank_transactions(
+    url: str = "",
+    api_key: str = "",
+    acct_no: str = "",
+    start_date: str = "",
+    end_date: str = "",
+    dc_flag: int | None = None,
+) -> dict:
+    """同步银行交易流水到数据库（内部方法，供定时任务调用）。
+
+    从配置中读取请求参数，自动翻页调用外部银行接口拉取全部交易流水并存储。
+
+    :param url: 接口地址
+    :param api_key: 接口密钥
+    :param acct_no: 银行账号
+    :param start_date: 开始日期 (YYYY-MM-DD)
+    :param end_date: 结束日期 (YYYY-MM-DD)
+    :param dc_flag: 借贷方向 (0=出金, 1=入金, None=全部)
+    :return: 同步结果统计
+    """
+    url = url.strip()
+    api_key = api_key.strip()
+    acct_no = acct_no.strip()
+    start_date = start_date.strip()
+    end_date = end_date.strip()
+
+    if not url:
+        raise ValueError("配置中缺少 url（接口地址）")
+    if not acct_no:
+        raise ValueError("配置中缺少 acct_no（银行账号）")
+
+    # 默认日期范围
+    today = datetime.now(timezone.utc).date()
+    if not end_date:
+        end_date = today.isoformat()
+    if not start_date:
+        start_date = today.isoformat()
+
+    sync_batch = generate_id()
+    stats = {
+        "sync_batch": sync_batch,
+        "acct_no": acct_no,
+        "start_date": start_date,
+        "end_date": end_date,
+        "pages_fetched": 0,
+        "transactions_fetched": 0,
+        "transactions_new": 0,
+        "transactions_dup": 0,
+        "errors": [],
+    }
+
+    import asyncio
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    try:
+        loop.run_until_complete(
+            _do_sync(
+                url=url,
+                api_key=api_key,
+                acct_no=acct_no,
+                start_date=start_date,
+                end_date=end_date,
+                dc_flag=dc_flag,
+                sync_batch=sync_batch,
+                stats=stats,
+            )
+        )
+    finally:
+        loop.close()
+
+    logger.info(
+        "银行流水同步完成: batch=%s, acct=%s, 翻页=%d, 获取=%d, 新增=%d, 重复=%d",
+        sync_batch,
+        acct_no,
+        stats["pages_fetched"],
+        stats["transactions_fetched"],
+        stats["transactions_new"],
+        stats["transactions_dup"],
+    )
+    return stats
+
+
+async def _do_sync(
+    url: str,
+    api_key: str,
+    acct_no: str,
+    start_date: str,
+    end_date: str,
+    dc_flag: int | None,
+    sync_batch: str,
+    stats: dict,
+) -> None:
+    """执行同步逻辑（自动翻页）。"""
+    async with async_session() as db:
+        # 查找对应的银行账户
+        result = await db.execute(
+            select(BankAccount).where(BankAccount.account_no == acct_no)
+        )
+        account = result.scalar_one_or_none()
+        account_id = account.id if account else ""
+
+        # 自动翻页拉取全部数据
+        all_transactions: list[dict] = []
+        page = 1
+        total_count = None
+
+        while True:
+            data = await _fetch_transactions(
+                url=url,
+                api_key=api_key,
+                acct_no=acct_no,
+                start_date=start_date,
+                end_date=end_date,
+                dc_flag=dc_flag,
+                page=page,
+            )
+            stats["pages_fetched"] += 1
+
+            # 解析交易记录
+            transactions = _parse_transactions(data)
+            all_transactions.extend(transactions)
+
+            # 获取总数（仅第一页）
+            if total_count is None:
+                total_count = _extract_total(data)
+
+            logger.info(
+                "银行流水同步翻页: page=%d, 本页=%d, 累计=%d, 总数=%s",
+                page, len(transactions), len(all_transactions), total_count,
+            )
+
+            # 判断是否还有下一页
+            if len(transactions) < _PAGE_SIZE:
+                break
+            if total_count is not None and len(all_transactions) >= total_count:
+                break
+
+            page += 1
+
+            # 安全限制：最多翻 100 页
+            if page > 100:
+                logger.warning("银行流水同步达到翻页上限: page=100")
+                break
+
+        stats["transactions_fetched"] = len(all_transactions)
+
+        # 逐条写入数据库（按 transaction_no 去重）
+        for tx in all_transactions:
+            tx["account_id"] = account_id
+            tx["account_no"] = acct_no
+            tx["sync_batch"] = sync_batch
+
+            tx_no = tx.get("transaction_no")
+            if tx_no:
+                existing = await db.execute(
+                    select(BankTransaction).where(
+                        BankTransaction.transaction_no == tx_no
+                    )
+                )
+                if existing.scalar_one_or_none():
+                    stats["transactions_dup"] += 1
+                    continue
+
+            transaction = BankTransaction(**tx)
+            db.add(transaction)
+            stats["transactions_new"] += 1
+
+        await db.commit()
+
+
+def _extract_total(data: dict | None) -> int | None:
+    """从接口返回中提取总条数。"""
+    if not data or not isinstance(data, dict):
+        return None
+    inner = data.get("data") or {}
+    if isinstance(inner, dict):
+        return inner.get("total")
+    return None
+
+
+async def _fetch_transactions(
+    url: str,
+    api_key: str,
+    acct_no: str,
+    start_date: str,
+    end_date: str,
+    dc_flag: int | None,
+    page: int,
+) -> dict:
+    """调用外部银行接口获取交易流水（单页）。"""
+    import time
+    headers = {
+        "Content-Type": "application/json",
+    }
+    if api_key:
+        headers["api-key"] = api_key
+
+    payload: dict = {
+        "acct_no": acct_no,
+        "start_date": start_date,
+        "end_date": end_date,
+        "page": page,
+        "page_size": _PAGE_SIZE,
+    }
+    if dc_flag is not None:
+        payload["dc_flag"] = dc_flag
+
+    request_params = {k: v for k, v in payload.items()}
+    start_time = time.monotonic()
+    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()
+            data = response.json()
+            duration_ms = int((time.monotonic() - start_time) * 1000)
+            log_bank_api_request(
+                acct_no=acct_no,
+                request_params=request_params,
+                response_data=data,
+                status_code=response.status_code,
+                is_success=True,
+                admin_id="scheduled_task",
+                duration_ms=duration_ms,
+            )
+            return data
+    except httpx.HTTPStatusError 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.response.status_code,
+            is_success=False,
+            error_msg=f"HTTP {e.response.status_code}",
+            admin_id="scheduled_task",
+            duration_ms=duration_ms,
+        )
+        raise ValueError(f"银行接口返回错误: HTTP {e.response.status_code}") from e
+    except httpx.TimeoutException 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=None,
+            is_success=False,
+            error_msg="请求超时",
+            admin_id="scheduled_task",
+            duration_ms=duration_ms,
+        )
+        raise ValueError("银行接口请求超时") from e
+    except httpx.RequestError 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=None,
+            is_success=False,
+            error_msg=str(e),
+            admin_id="scheduled_task",
+            duration_ms=duration_ms,
+        )
+        raise ValueError(f"银行接口请求失败: {e}") from e
+
+
+def _parse_transactions(data: dict | None) -> list[dict]:
+    """解析接口返回的数据为交易记录列表。
+
+    接口返回格式:
+    {
+        "code": 200,
+        "message": "成功",
+        "data": {
+            "list": [...],
+            "total": 1,
+            "page": 1,
+            "page_size": 20
+        }
+    }
+    """
+    if not data:
+        return []
+
+    # 提取列表
+    records = []
+    if isinstance(data, dict):
+        inner = data.get("data") or {}
+        if isinstance(inner, dict):
+            records = inner.get("list") or []
+        elif isinstance(inner, list):
+            records = inner
+    elif isinstance(data, list):
+        records = data
+
+    transactions = []
+    for item in records:
+        if not isinstance(item, dict):
+            continue
+        tx = _map_record(item)
+        transactions.append(tx)
+    return transactions
+
+
+def _map_record(item: dict) -> dict:
+    """将接口返回的单条记录映射为 BankTransaction 字段。"""
+    # 流水号：优先 tellerSeqno，其次 transferId
+    tx_no = (
+        item.get("tellerSeqno")
+        or item.get("transferId")
+        or item.get("transactionNo")
+        or item.get("transaction_no")
+        or ""
+    )
+
+    # 交易时间
+    tx_time_raw = (
+        item.get("transTimeStr")
+        or item.get("trans_time")
+        or item.get("transactionTime")
+        or item.get("transaction_time")
+        or ""
+    )
+    tx_time = _parse_datetime(tx_time_raw)
+
+    # 交易金额
+    amount = (
+        item.get("transAmt")
+        or item.get("amount")
+        or item.get("transactionAmount")
+        or item.get("transaction_amount")
+        or "0"
+    )
+
+    # 借贷方向
+    direction = _resolve_direction(item.get("dcFlag"), item.get("dcFlagLabel"))
+
+    # 余额
+    balance = (
+        item.get("balance")
+        or item.get("balanceAfter")
+        or item.get("balance_after")
+    )
+
+    # 对方信息
+    counterparty_name = (
+        item.get("cnterName")
+        or item.get("counterpartyName")
+        or item.get("counterparty_name")
+    )
+
+    counterparty_account = (
+        item.get("counterAcctNo")
+        or item.get("counterpartyAccount")
+        or item.get("counterparty_account")
+    )
+
+    counterparty_bank = (
+        item.get("cnterBankName")
+        or item.get("counterpartyBank")
+        or item.get("counterparty_bank")
+    )
+
+    # 摘要/备注
+    remark = (
+        item.get("remark")
+        or item.get("memo")
+    )
+
+    digest = item.get("digestCode") or item.get("digest_code")
+
+    # 我方账号
+    acct_no = (
+        item.get("acctNo")
+        or item.get("acct_no")
+        or ""
+    )
+
+    return {
+        "id": generate_id(),
+        "transaction_no": tx_no or None,
+        "transaction_time": tx_time,
+        "transaction_amount": str(amount) if amount else "0",
+        "balance_direction": direction,
+        "balance_after": str(balance) if balance else None,
+        "counterparty_name": counterparty_name,
+        "counterparty_account": counterparty_account,
+        "counterparty_bank": counterparty_bank,
+        "remark": remark,
+        "digest_code": digest,
+        "raw_data": json.dumps(item, ensure_ascii=False, default=str),
+        "is_synced": True,
+        "account_no": acct_no,
+    }
+
+
+def _resolve_direction(dc_flag, dc_flag_label: str | None = None) -> str | None:
+    """解析借贷方向。"""
+    if dc_flag_label:
+        label = str(dc_flag_label)
+        if label.startswith("借") or "出金" in label:
+            return "DR"
+        if label.startswith("贷") or "入金" in label:
+            return "CR"
+
+    if dc_flag is not None:
+        if dc_flag in (0, "0"):
+            return "DR"
+        if dc_flag in (1, "1"):
+            return "CR"
+
+    return None
+
+
+def _parse_datetime(value) -> datetime | None:
+    """解析日期时间字符串为 datetime 对象。"""
+    if not value or not isinstance(value, str):
+        return None
+    formats = [
+        "%Y-%m-%d %H:%M:%S",
+        "%Y-%m-%dT%H:%M:%S",
+        "%Y-%m-%dT%H:%M:%S%z",
+        "%Y-%m-%dT%H:%M:%S.%f",
+        "%Y-%m-%dT%H:%M:%S.%f%z",
+        "%Y-%m-%d %H:%M",
+        "%Y-%m-%d",
+    ]
+    for fmt in formats:
+        try:
+            return datetime.strptime(value, fmt)
+        except ValueError:
+            continue
+    return None

video-gen-api/app/tasks/celery_app.py

@@ -57,8 +57,103 @@ def _derive_redis_db(url: str, db_no: int) -> str:
     return url.rstrip("/") + f"/{db_no}"
 
 
+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
+
+
+# 同步引擎（用于 Beat 启动时加载数据库定时任务）
+_sync_engine = None
+
+
+def _get_sync_engine():
+    global _sync_engine
+    if _sync_engine is None:
+        from sqlalchemy import create_engine
+        from app.config import settings
+
+        db_url = settings.DATABASE_URL
+        # 异步 URL → 同步 URL（asyncpg → psycopg2 / aiosqlite → sqlite3）
+        if db_url.startswith("postgresql+asyncpg"):
+            db_url = db_url.replace("postgresql+asyncpg", "postgresql+psycopg2", 1)
+        elif db_url.startswith("sqlite+aiosqlite"):
+            db_url = db_url.replace("sqlite+aiosqlite", "sqlite", 1)
+        elif db_url.startswith("mysql+aiomysql"):
+            db_url = db_url.replace("mysql+aiomysql", "mysql+pymysql", 1)
+
+        _sync_engine = create_engine(db_url, pool_pre_ping=True)
+    return _sync_engine
+
+
+def _load_dynamic_beat_tasks() -> dict:
+    """从数据库加载活跃定时任务，返回 beat_schedule 格式的字典。
+
+    在 Celery 配置阶段同步调用，确保 Beat 启动时能读取到动态任务。
+    """
+    from sqlalchemy import select
+    from sqlalchemy.orm import Session
+
+    from app.models.scheduled_task import ScheduledTask
+
+    dynamic_schedule = {}
+    try:
+        engine = _get_sync_engine()
+        with Session(engine) as session:
+            result = session.execute(
+                select(ScheduledTask).where(ScheduledTask.is_active.is_(True))
+            )
+            tasks = result.scalars().all()
+
+            for task in 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}"
+                dynamic_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,
+                )
+    except Exception:
+        logger.exception("加载动态定时任务失败，跳过")
+    return dynamic_schedule
+
+
 def _beat_schedule() -> dict:
-    schedule: dict = {}
+    schedule: dict = _load_dynamic_beat_tasks()
     if bool(getattr(settings, "POLL_DUE_DISPATCH_ENABLED", True)):
         schedule["dispatch-due-poll-tasks-every-minute"] = {
             "task": CeleryTaskName.DISPATCH_DUE_POLL.value,
@@ -293,85 +388,6 @@ 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

video-gen-api/app/tasks/scheduled_tasks.py

@@ -1,10 +1,19 @@
 """定时任务执行器。
 
-支持两种任务类型：
-- external_api: 调用外部 HTTP 接口
-- internal_method: 动态导入并执行内部函数
+仅支持内部方法执行类型，供 Celery Beat 定时调用。
 
-任务在 Celery Worker 中执行，通过 run_async 桥接异步操作。
+配置 JSON 格式（示例）:
+{
+    "module": "app.services.bank.sync_service",
+    "function": "sync_bank_transactions",
+    "url": "http://...",
+    "api_key": "...",
+    "acct_no": "...",
+    "start_date": "2026-08-01",
+    "end_date": "2026-08-14"
+}
+
+其中 module 和 function 指定要调用的函数，其余字段作为 kwargs 传给该函数。
 """
 
 import json
@@ -12,7 +21,6 @@ import logging
 import time
 from datetime import datetime, timezone
 
-import httpx
 from sqlalchemy import select
 
 from app.models.base import async_session
@@ -40,45 +48,14 @@ def _update_task_status(task_id: str, status: str, error_msg: str | None = None)
     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:
-    """执行内部方法调用。"""
+    """执行内部方法调用。
+
+    配置 JSON 中 module 和 function 指定要调用的函数，其余字段作为 kwargs 传入。
+    """
     cfg = json.loads(config or "{}")
-    module_path = cfg.get("module", "").strip()
-    function_name = cfg.get("function", "").strip()
-    args = cfg.get("args") or []
+    module_path = cfg.pop("module", "").strip()
+    function_name = cfg.pop("function", "").strip()
 
     if not module_path or not function_name:
         raise ValueError("内部方法需要指定 module 和 function")
@@ -90,8 +67,9 @@ def _execute_internal_method(config: str | None) -> dict:
     if func is None or not callable(func):
         raise ValueError(f"模块 {module_path} 中不存在可调用函数 {function_name}")
 
+    # 剩余字段作为 kwargs 传给函数
     start = time.monotonic()
-    result = func(*args)
+    result = func(**cfg)
     duration_ms = int((time.monotonic() - start) * 1000)
     return {
         "duration_ms": duration_ms,
@@ -101,10 +79,7 @@ def _execute_internal_method(config: str | None) -> dict:
 
 @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 桥接到异步上下文读取任务配置并执行。
-    """
+    """执行定时任务（Celery 任务入口）。"""
 
     async def _run():
         async with async_session() as db:
@@ -118,21 +93,14 @@ def execute_scheduled_task(self, task_id: str):
                 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}")
-
+            exec_result = _execute_internal_method(task_config)
             _update_task_status(task_id, "success")
-            logger.info("定时任务执行成功: %s (%s) -> %s", task_id, task_type, exec_result)
+            logger.info("定时任务执行成功: %s -> %s", task_id, 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)
+            logger.exception("定时任务执行失败: %s", task_id)
 
     run_async(_run())