增加银行流水和定时任务程序

2026-08-14 14:35:25 +08:00
parent 148b89c3ca
commit 7c3ef21762
9 changed files with 616 additions and 94 deletions
@@ -28,7 +28,7 @@
        }
      })();
    </script>
-    <script type="module" crossorigin src="/assets/index-b8paHjMP.js"></script>
+    <script type="module" crossorigin src="/assets/index-DIVmQY5H.js"></script>
    <link rel="stylesheet" crossorigin href="/assets/index-D3fwIbOp.css">
  </head>
  <body>
@@ -18,12 +18,10 @@ import type { ScheduledTask } from '../types';
 const { TextArea } = Input;

 const SCHEDULE_TYPE_OPTIONS = [
-  { value: 'external_api', label: '外部接口调用' },
  { value: 'internal_method', label: '内部方法执行' },
 ];

 const TASK_TYPE_LABELS: Record<string, string> = {
-  external_api: '外部接口',
  internal_method: '内部方法',
 };

@@ -78,7 +76,7 @@ const AdminScheduledTasks: React.FC = () => {
    } else {
      setEditing(null);
      form.resetFields();
-      form.setFieldsValue({ is_active: true, task_type: 'external_api' });
+      form.setFieldsValue({ is_active: true, task_type: 'internal_method' });
    }
    setActiveTab('basic');
    setModal(true);
@@ -142,15 +140,13 @@ const AdminScheduledTasks: React.FC = () => {
    }
  };

-  const taskType = Form.useWatch('task_type', form);
-
  const columns = [
    { title: '任务名称', dataIndex: 'name', width: 160, ellipsis: true },
    {
      title: '类型',
      dataIndex: 'taskType',
      width: 100,
-      render: (v: string) => <Tag color={v === 'external_api' ? 'blue' : 'purple'}>{TASK_TYPE_LABELS[v] || v}</Tag>,
+      render: (v: string) => <Tag color="purple">{TASK_TYPE_LABELS[v] || v}</Tag>,
    },
    { title: '调度表达式', dataIndex: 'schedule', width: 140, render: (v: string) => <code style={{ background: '#f1f5f9', padding: '2px 6px', borderRadius: 4 }}>{v}</code> },
    {
@@ -194,9 +190,9 @@ const AdminScheduledTasks: React.FC = () => {

  const scheduleHelp = (
    <div style={{ fontSize: 12, color: '#64748b', marginTop: 4 }}>
-      <div>• 纯数字：间隔秒数（如 60 = 每 60 秒执行一次）</div>
+      <div>• 纯数字：间隔秒数（如 3600 = 每小时执行一次）</div>
      <div>• Cron 表达式（5 字段）：分 时 日 月 周</div>
-      <div>• 示例：<code>* * * * *</code> = 每分钟 | <code>0 * * * *</code> = 每小时</div>
+      <div>• 示例：<code>0 * * * *</code> = 每小时 | <code>0 2 * * *</code> = 每天凌晨2点</div>
    </div>
  );

@@ -283,18 +279,11 @@ const AdminScheduledTasks: React.FC = () => {
                  <Form.Item
                    name="config"
                    label="配置 JSON"
-                    extra={
-                      taskType === 'external_api'
-                        ? '字段：url（地址）、method（GET/POST/PUT/DELETE）、headers（对象）、payload（对象/参数）、timeout（秒，默认 30）'
-                        : '字段：module（模块路径，如 app.services.xxx）、function（函数名）、args（参数数组）'
-                    }
+                    extra='module 和 function 指定调用的函数，其余字段作为参数传入。服务会自动翻页拉取全部流水：'
                  >
                    <TextArea
-                      rows={8}
-                      placeholder={taskType === 'external_api'
-                        ? '{\n  "url": "https://api.example.com/data",\n  "method": "POST",\n  "headers": {},\n  "payload": {}\n}'
-                        : '{\n  "module": "app.services.report",\n  "function": "generate_daily_report",\n  "args": []\n}'
-                      }
+                      rows={10}
+                      placeholder={'{\n  "module": "app.services.bank.sync_service",\n  "function": "sync_bank_transactions",\n  "url": "http://ceshi.web.minzhong.cn/api/api/v1/internal/get-blank-transfer-acct-time",\n  "api_key": "jixekCxm8piLFi0AlfA24bDtCKJ82bfu",\n  "acct_no": "110972289710001",\n  "start_date": "2026-08-01",\n  "end_date": "2026-08-14"\n}'}
                      style={{ fontFamily: 'monospace', fontSize: 13 }}
                    />
                  </Form.Item>
@@ -688,7 +688,7 @@ const AdminSettings: React.FC = () => {
  ];

  return (
-    <div style={{ maxWidth: 720 }}>
+    <div style={{ maxWidth: 1080 }}>
      <Card variant="outlined" style={{ borderRadius: 12, border: '1px solid #f0f0f5', marginBottom: 16 }}>
        <div style={{ display: 'flex', alignItems: 'center', gap: 12, marginBottom: 24 }}>
          <div style={{
@@ -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')
@@ -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()
@@ -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"),
+    )
@@ -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
@@ -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}")
-
            _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())