Merge branch 'main' of https://gitee.com/wg123/video-gen

video-gen-api/alembic/versions/6101ba8d5761_user_oauth表新增归属公司应用_新增授权链接字段.py

@@ -0,0 +1,31 @@
+"""user_oauth表新增归属公司应用，新增授权链接字段
+
+Revision ID: 6101ba8d5761
+Revises: 9ac2212e1b8e
+Create Date: 2026-06-10 16:34:38.842582
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '6101ba8d5761'
+down_revision: Union[str, None] = '9ac2212e1b8e'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('user_oauth_app', sa.Column('auth_url', sa.String(length=256), nullable=True, comment='应用授权链接'))
+    op.add_column('user_oauth_app', sa.Column('company', sa.String(length=256), nullable=True, comment='应用归属公司名称'))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('user_oauth_app', 'company')
+    op.drop_column('user_oauth_app', 'auth_url')
+    # ### end Alembic commands ###

video-gen-api/alembic/versions/8922eafcd8b0_user_oauth表新增account_userid授权登录id_.py

@@ -0,0 +1,29 @@
+"""user_oauth表新增account_userid授权登录id，用来判断不同账号授权
+
+Revision ID: 8922eafcd8b0
+Revises: 6101ba8d5761
+Create Date: 2026-06-11 16:21:17.617777
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '8922eafcd8b0'
+down_revision: Union[str, None] = '6101ba8d5761'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('user_oauth', sa.Column('account_userid', sa.String(length=128), nullable=True, comment='授权账户登录userid，同一个用户不同的授权账户token不一样'))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('user_oauth', 'account_userid')
+    # ### end Alembic commands ###

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

@@ -43,6 +43,7 @@ from app.services.notification import create_notification
 from app.services.auth import hash_password, verify_password
 from app.services.operation_log import log_operation
 from app.services.resource_signed_url_service import build_resource_signed_url
+from app.services.payment import sync_pending_orders, process_refund
 
 from app.services.generation_billing_service import (
     OWNER_GENERATION_RECORD,
@@ -442,26 +443,70 @@ async def batch_update_payment_configs(
 
 @router.get("/payment-stats")
 async def get_payment_stats(
+    payment_method: str | None = Query(None),
+    status: str | None = Query(None),
+    start_date: str | None = Query(None),
+    end_date: str | None = Query(None),
     admin: User = Depends(get_admin_user),
     db: AsyncSession = Depends(get_db),
 ):
-    """Return payment statistics for admin dashboard."""
+    """Return payment statistics for admin dashboard with filters."""
     from sqlalchemy import func
 
+    # Ensure by_status has all expected statuses with defaults
+    by_status = {
+        "pending": {"count": 0, "amount": 0.0},
+        "paid": {"count": 0, "amount": 0.0},
+        "cancelled": {"count": 0, "amount": 0.0},
+        "refunded": {"count": 0, "amount": 0.0},
+    }
+
+    # Parse dates and build base query filters
+    now_cst = datetime.now(CST)
+    today_start = now_cst.replace(hour=0, minute=0, second=0, microsecond=0)
+    today_end = today_start + timedelta(days=1)
+    
+    # Default to today if no date range provided
+    query_start = today_start
+    query_end = today_end
+    
+    if start_date:
+        query_start = datetime.fromisoformat(start_date).replace(tzinfo=CST)
+    if end_date:
+        query_end = (datetime.fromisoformat(end_date) + timedelta(days=1)).replace(tzinfo=CST)
+
+    # Build filter list for status breakdown
+    breakdown_filters = []
+    if payment_method:
+        breakdown_filters.append(PaymentOrder.payment_method == payment_method)
+    if status:
+        breakdown_filters.append(PaymentOrder.status == status)
+    # Always apply date range to breakdown
+    breakdown_filters.append(PaymentOrder.created_at >= query_start)
+    breakdown_filters.append(PaymentOrder.created_at < query_end)
+
     # Status breakdown
     status_result = await db.execute(
         select(
             PaymentOrder.status,
             func.count().label("count"),
             func.coalesce(func.sum(PaymentOrder.amount), 0).label("amount"),
-        ).group_by(PaymentOrder.status)
+        )
+        .where(*breakdown_filters)
+        .group_by(PaymentOrder.status)
     )
-    by_status = {}
     for row in status_result.all():
-        by_status[row.status] = {"count": row.count, "amount": float(row.amount)}
+        if row.status in by_status:
+            by_status[row.status] = {
+                "count": row.count,
+                "amount": round(float(row.amount), 2)
+            }
+        else:
+            # Map any unexpected status to cancelled
+            by_status["cancelled"]["count"] += row.count
+            by_status["cancelled"]["amount"] += round(float(row.amount), 2)
 
-    # Today's stats
-    today_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
+    # Today's stats (CST time zone) - independent of filter
     today_result = await db.execute(
         select(
             func.count().label("paid_count"),
@@ -469,38 +514,70 @@ async def get_payment_stats(
         ).where(
             PaymentOrder.status == "paid",
             PaymentOrder.paid_at >= today_start,
+            PaymentOrder.paid_at < today_end,
         )
     )
     today_row = today_result.one()
 
-    # Recent 50 orders
+    # Monthly cumulative stats
+    month_start = now_cst.replace(day=1, hour=0, minute=0, second=0, microsecond=0)
+    month_end = (month_start + timedelta(days=32)).replace(day=1, hour=0, minute=0, second=0, microsecond=0)
+    
+    month_result = await db.execute(
+        select(
+            func.count().label("paid_count"),
+            func.coalesce(func.sum(PaymentOrder.amount), 0).label("paid_amount"),
+        ).where(
+            PaymentOrder.status == "paid",
+            PaymentOrder.paid_at >= month_start,
+            PaymentOrder.paid_at < month_end,
+        )
+    )
+    month_row = month_result.one()
+
+    # Recent orders with filters
+    recent_filters = []
+    if payment_method:
+        recent_filters.append(PaymentOrder.payment_method == payment_method)
+    if status:
+        recent_filters.append(PaymentOrder.status == status)
+    recent_filters.append(PaymentOrder.created_at >= query_start)
+    recent_filters.append(PaymentOrder.created_at < query_end)
+
     recent_result = await db.execute(
-        select(PaymentOrder)
+        select(PaymentOrder, User)
+        .join(User, PaymentOrder.user_id == User.id)
+        .where(*recent_filters)
         .order_by(PaymentOrder.created_at.desc())
         .limit(50)
     )
-    recent = recent_result.scalars().all()
+    recent_data = recent_result.all()
 
     return {
         "by_status": by_status,
         "today": {
             "paid_count": today_row.paid_count,
-            "paid_amount": float(today_row.paid_amount),
+            "paid_amount": round(float(today_row.paid_amount), 2),
+        },
+        "month": {
+            "paid_count": month_row.paid_count,
+            "paid_amount": round(float(month_row.paid_amount), 2),
         },
         "recent": [
             {
                 "id": o.id,
                 "order_no": o.order_no,
                 "user_id": o.user_id,
-                "amount": o.amount,
-                "credits": o.credits,
+                "username": u.username,
+                "amount": round(o.amount, 2),
+                "credits": round(o.credits, 2),
                 "payment_method": o.payment_method,
-                "status": o.status,
+                "status": o.status if o.status in ("pending", "paid", "cancelled", "refunded") else "cancelled",
                 "trade_no": o.trade_no,
-                "paid_at": o.paid_at.isoformat() if o.paid_at else None,
-                "created_at": o.created_at.isoformat() if o.created_at else None,
+                "paid_at": _iso(o.paid_at),
+                "created_at": _iso(o.created_at),
             }
-            for o in recent
+            for o, u in recent_data
         ],
     }
 
@@ -544,13 +621,13 @@ async def get_admin_payment_orders(
                 "id": o.id,
                 "order_no": o.order_no,
                 "user_id": o.user_id,
-                "amount": o.amount,
-                "credits": o.credits,
+                "amount": round(o.amount, 2),
+                "credits": round(o.credits, 2),
                 "payment_method": o.payment_method,
-                "status": o.status,
+                "status": o.status if o.status in ("pending", "paid", "cancelled", "refunded") else "cancelled",
                 "trade_no": o.trade_no,
-                "paid_at": o.paid_at.isoformat() if o.paid_at else None,
-                "created_at": o.created_at.isoformat() if o.created_at else None,
+                "paid_at": _iso(o.paid_at),
+                "created_at": _iso(o.created_at),
             }
             for o in orders
         ],
@@ -585,6 +662,19 @@ async def update_payment_config(
     }
 
 
+@router.post("/payment-orders/{order_no}/refund")
+async def refund_payment_order(
+    order_no: str,
+    admin: User = Depends(get_admin_user),
+    db: AsyncSession = Depends(get_db),
+):
+    """Refund a paid payment order."""
+    result = await process_refund(db, order_no)
+    if not result.get("success"):
+        raise HTTPException(status_code=400, detail=result.get("message", "退款失败"))
+    return result
+
+
 # ── Industry Config ──────────────────────────────────────
 
 def _serialize_industry(ind: IndustryConfig) -> dict:
@@ -1371,72 +1461,4 @@ async def admin_generate_video(
 
 # ── Payment Stats ────────────────────────────────────────
 
-@router.get("/payment-stats")
-async def get_payment_stats(
-    admin: User = Depends(get_admin_user),
-    db: AsyncSession = Depends(get_db),
-):
-    """Payment statistics for admin dashboard."""
-    from app.models.payment_order import PaymentOrder
-    from datetime import datetime
 
-    # Count and revenue by status
-    rows = (await db.execute(
-        select(
-            PaymentOrder.status,
-            PaymentOrder.payment_method,
-            func.count(PaymentOrder.id).label("count"),
-            func.coalesce(func.sum(PaymentOrder.amount), 0).label("total_amount"),
-        ).group_by(PaymentOrder.status, PaymentOrder.payment_method)
-    )).all()
-
-    by_status: dict[str, dict] = {}
-    for r in rows:
-        s = r.status
-        if s not in by_status:
-            by_status[s] = {"count": 0, "amount": 0.0}
-        by_status[s]["count"] += r.count
-        by_status[s]["amount"] += float(r.total_amount)
-
-    # Recent orders (last 50)
-    recent = (await db.execute(
-        select(PaymentOrder)
-        .order_by(PaymentOrder.created_at.desc())
-        .limit(50)
-    )).scalars().all()
-
-    # Today stats
-    today_start = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
-    today_paid = (await db.execute(
-        select(
-            func.count(PaymentOrder.id),
-            func.coalesce(func.sum(PaymentOrder.amount), 0),
-        ).where(
-            PaymentOrder.status == "paid",
-            PaymentOrder.paid_at >= today_start,
-        )
-    )).first()
-    today_count, today_amount = (today_paid or (0, 0))
-
-    return {
-        "by_status": by_status,
-        "today": {
-            "paid_count": int(today_count or 0),
-            "paid_amount": float(today_amount or 0),
-        },
-        "recent": [
-            {
-                "id": o.id,
-                "order_no": o.order_no,
-                "user_id": o.user_id,
-                "amount": o.amount,
-                "credits": o.credits,
-                "payment_method": o.payment_method,
-                "status": o.status,
-                "trade_no": o.trade_no,
-                "created_at": _iso(o.created_at),
-                "paid_at": _iso(o.paid_at),
-            }
-            for o in recent
-        ],
-    }

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

@@ -4,7 +4,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-logger = logging.getLogger("videogen")
+logger = logging.getLogger("payment")
 
 from app.dependencies import get_db, get_current_user
 from app.models.user import User
@@ -16,13 +16,20 @@ from app.services.payment import (
     verify_wechat_callback,
     verify_alipay_callback,
     process_payment_success_by_order_no,
+    process_refund,
+    _get_payment_configs,
+    _close_alipay_order,
+    _get_order_expire_seconds,
 )
 
 router = APIRouter(prefix="/payments", tags=["payments"])
 
 
 @router.get("/methods")
-async def get_payment_methods(db: AsyncSession = Depends(get_db)):
+async def get_payment_methods(
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db)
+):
     """Return which payment methods are enabled (from admin config)."""
     from app.services.payment import _get_payment_configs
     configs = await _get_payment_configs(db)
@@ -89,9 +96,10 @@ async def wechat_callback(request: Request, db: AsyncSession = Depends(get_db)):
 async def alipay_callback(request: Request, db: AsyncSession = Depends(get_db)):
     form_data = await request.form()
     data = dict(form_data)
+    
     logger.info(
         f"ALIPAY_CALLBACK order_no={data.get('out_trade_no')} "
-        f"trade_no={data.get('trade_no', '')} status={data.get('trade_status', '')}"
+        f"data={data}"
     )
 
     # Verify signature first
@@ -106,8 +114,11 @@ async def alipay_callback(request: Request, db: AsyncSession = Depends(get_db)):
 
     order_no = data.get("out_trade_no")
     trade_no = data.get("trade_no", "")
+    total_amount_str = data.get("total_amount", "")
+    total_amount = float(total_amount_str) if total_amount_str else None
+    
     if order_no:
-        await process_payment_success_by_order_no(db, order_no, trade_no)
+        await process_payment_success_by_order_no(db, order_no, trade_no, total_amount)
 
     return "success"
 
@@ -130,6 +141,29 @@ async def list_orders(
     return orders
 
 
+@router.get("/orders/{order_no}", response_model=PaymentOrderOut)
+async def get_order(
+    order_no: str,
+    current_user: User = Depends(get_current_user),
+    db: AsyncSession = Depends(get_db),
+):
+    from app.services.payment import _check_and_expire_order
+    result = await db.execute(
+        select(PaymentOrder)
+        .where(
+            PaymentOrder.order_no == order_no,
+            PaymentOrder.user_id == current_user.id,
+        )
+        .limit(1)
+    )
+    order = result.scalar_one_or_none()
+    if not order:
+        raise HTTPException(status_code=404, detail="订单不存在")
+    # Auto-expire if needed
+    await _check_and_expire_order(db, order)
+    return order
+
+
 @router.post("/orders/{order_no}/cancel")
 async def cancel_order(
     order_no: str,
@@ -148,6 +182,15 @@ async def cancel_order(
         raise HTTPException(status_code=404, detail="订单不存在")
     if order.status != "pending":
         raise HTTPException(status_code=400, detail=f"订单状态为{order.status}，无法取消")
+    
+    # If it's an Alipay order, call close API first
+    if order.payment_method == "alipay":
+        db_configs = await _get_payment_configs(db)
+        try:
+            await _close_alipay_order(db, order, db_configs)
+        except Exception as e:
+            logger.exception(f"Failed to close Alipay order {order_no}: {e}")
+    
     order.status = "cancelled"
     await db.flush()
     logger.info(

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

@@ -41,7 +41,7 @@ async def create_app(
     db: AsyncSession = Depends(get_db),
 ):
     try:
-        app = await create_user_oauth_app(db, req.app_id, req.secret, req.open_type, admin.id)
+        app = await create_user_oauth_app(db, req.app_id, req.secret, req.open_type, admin.id, req.count, req.auth_url, req.company)
         return UserOAuthAppOut.model_validate(app)
     except ValueError as e:
         raise HTTPException(
@@ -72,7 +72,7 @@ async def update_app(
     admin: User = Depends(get_admin_user),
     db: AsyncSession = Depends(get_db),
 ):
-    app = await update_user_oauth_app(db, id, req.secret, req.open_type, req.status, admin.id)
+    app = await update_user_oauth_app(db, id, req.secret, req.open_type, req.status, req.count, req.auth_url, req.company, admin.id)
     if not app:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,

video-gen-api/app/main.py

@@ -21,48 +21,6 @@ from app.services.log_config import decrypt_data
 logging.basicConfig(level=logging.INFO if settings.DEBUG else logging.WARNING)
 
 
-def _setup_payment_logger():
-    """Configure a dedicated file logger for payment events.
-
-    Logs are written to logs/payment_YYYY-MM-DD.log, rotated daily.
-    30 days of history are retained.
-    """
-    from logging.handlers import TimedRotatingFileHandler
-    log_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "logs")
-    os.makedirs(log_dir, exist_ok=True)
-    log_file = os.path.join(log_dir, "payment.log")
-
-    payment_logger = logging.getLogger("payment")
-    payment_logger.setLevel(logging.INFO)
-    payment_logger.propagate = False  # don't double-log to root
-
-    # Avoid adding duplicate handlers on reload
-    if any(getattr(h, "_payment_file", False) for h in payment_logger.handlers):
-        return
-
-    handler = TimedRotatingFileHandler(
-        log_file,
-        when="midnight",
-        interval=1,
-        backupCount=30,
-        encoding="utf-8",
-        utc=False,  # use local time
-    )
-    handler.suffix = "%Y-%m-%d"  # files named like payment.log.2026-06-10
-    handler._payment_file = True  # type: ignore[attr-defined]
-    handler.setFormatter(logging.Formatter(
-        "%(asctime)s [%(levelname)s] %(message)s",
-        datefmt="%Y-%m-%d %H:%M:%S",
-    ))
-    payment_logger.addHandler(handler)
-    # Mirror to console in DEBUG mode
-    if settings.DEBUG:
-        payment_logger.addHandler(logging.StreamHandler())
-
-
-_setup_payment_logger()
-
-
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     from app.models import async_session
@@ -78,14 +36,20 @@ async def lifespan(app: FastAPI):
     await task_queue.recover()
     queue_task = asyncio.create_task(task_queue.run())
 
-    # Background task: auto-expire pending payment orders
+    # Background task: auto-expire pending payment orders and sync status
     async def _order_expiry_loop():
-        from app.services.payment import expire_all_pending_orders
+        from app.services.payment import expire_all_pending_orders, sync_pending_orders
         from logging import getLogger
         bg_logger = getLogger("payment")
         while True:
             try:
                 async with async_session() as db:
+                    # 同步待支付订单状态（检查支付宝实际支付状态
+                    sync_count = await sync_pending_orders(db)
+                    if sync_count > 0:
+                        bg_logger.info(f"Synced {sync_count} pending payment order(s)")
+                    
+                    # 自动过期订单
                     n = await expire_all_pending_orders(db)
                     if n > 0:
                         bg_logger.info(f"Auto-expired {n} pending payment order(s)")
@@ -94,6 +58,22 @@ async def lifespan(app: FastAPI):
             await asyncio.sleep(60)  # check every minute
 
     expiry_task = asyncio.create_task(_order_expiry_loop())
+    
+    # 启动时立即同步一次未支付订单
+    asyncio.create_task(asyncio.sleep(5))  # 等待5秒后再同步，让系统完全启动
+    async def startup_sync():
+        await asyncio.sleep(5)
+        from app.services.payment import sync_pending_orders
+        from logging import getLogger
+        bg_logger = getLogger("payment")
+        try:
+            async with async_session() as db:
+                sync_count = await sync_pending_orders(db)
+                if sync_count > 0:
+                    bg_logger.info(f"Startup: Synced {sync_count} pending payment order(s)")
+        except Exception as e:
+            bg_logger.error(f"Startup sync error: {e}")
+    asyncio.create_task(startup_sync())
 
     app.state.db_session_factory = async_session
 

video-gen-api/app/middleware/request_encrypt.py

@@ -42,6 +42,11 @@ class RequestEncryptMiddleware(BaseHTTPMiddleware):
     async def dispatch(
         self, request: Request, call_next: RequestResponseEndpoint
     ) -> Response:
+        # 白名单：支付回调接口不需要加密/解密
+        path = request.url.path
+        if "/payments/alipay/callback" in path or "/payments/wechat/callback" in path:
+            return await call_next(request)
+        
         encrypted = request.headers.get("X-Encrypted", "").lower() == "true"
         if not encrypted:
             return await call_next(request)

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

@@ -19,6 +19,15 @@ class UserOAuthApp(Base, TimestampMixin, SoftDeleteMixin):
     status: Mapped[int] = mapped_column(
         BigInteger, nullable=False, default=1, comment="状态,1=正常,2=禁用"
     )
+    count: Mapped[int] = mapped_column(
+        BigInteger, nullable=False, default=100, comment="应用最大可以授权多少个用户"
+    )
+    auth_url: Mapped[str] = mapped_column(
+        String(256), nullable=True,  comment="应用授权链接"
+    )
+    company: Mapped[str] = mapped_column(
+        String(256), nullable=True,  comment="应用归属公司名称"
+    )
     open_type: Mapped[int] = mapped_column(
         BigInteger, nullable=False, index=True, comment="开户方式（1=千川，2=广告，3=本地推，4=星图，5=快手代理商，6=巨量星图，7=巨量服务单，8=腾讯服务单，9=腾讯营销K2，10=腾讯营销K3）"
     )

video-gen-api/app/schemas/user_oauth_app.py

@@ -12,6 +12,9 @@ class UserOAuthAppCreate(BaseModel):
         le=10,
         description="开户方式（1=千川，2=广告，3=本地推，4=星图，5=快手代理商，6=巨量星图，7=巨量服务单，8=腾讯服务单，9=腾讯营销K2，10=腾讯营销K3）",
     )
+    count: int = Field(100, ge=1, description="应用最大可以授权多少个用户")
+    auth_url: str | None = Field(None, max_length=256, description="应用授权链接")
+    company: str | None = Field(None, max_length=256, description="应用归属公司名称")
 
 
 class UserOAuthAppUpdate(BaseModel):
@@ -23,6 +26,9 @@ class UserOAuthAppUpdate(BaseModel):
         description="开户方式（1=千川，2=广告，3=本地推，4=星图，5=快手代理商，6=巨量星图，7=巨量服务单，8=腾讯服务单，9=腾讯营销K2，10=腾讯营销K3）",
     )
     status: int | None = Field(None, ge=1, le=2, description="应用状态（1=正常,2=禁用）")
+    count: int | None = Field(None, ge=1, description="应用最大可以授权多少个用户")
+    auth_url: str | None = Field(None, max_length=256, description="应用授权链接")
+    company: str | None = Field(None, max_length=256, description="应用归属公司名称")
 
 
 class UserOAuthAppOut(BaseModel):
@@ -30,7 +36,10 @@ class UserOAuthAppOut(BaseModel):
     app_id: str = Field(..., description="应用id")
     secret: str = Field(..., description="应用密钥")
     status: int = Field(..., description="状态,1=正常,2=禁用")
+    count: int = Field(..., description="应用最大可以授权多少个用户")
     open_type: int = Field(..., description="开户方式")
+    auth_url: str | None = Field(None, description="应用授权链接")
+    company: str | None = Field(None, description="应用归属公司名称")
     create_by: str | None = Field(None, description="创建者")
     created_at: NaiveDatetime = Field(..., description="创建时间")
     updated_at: NaiveDatetime = Field(..., description="更新时间")

video-gen-api/app/services/payment.py

@@ -1,15 +1,22 @@
 import logging
 import os
-import certifi
 from datetime import datetime, timedelta
 
+# 尝试设置 SSL 证书路径
+try:
+    import certifi
+    os.environ["SSL_CERT_FILE"] = certifi.where()
+    os.environ["REQUESTS_CA_BUNDLE"] = certifi.where()
+except ImportError:
+    pass
+
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
 from app.models.payment_order import PaymentOrder
 from app.models.system_config import SystemConfig
-from app.services.credits import add_credits
+from app.services.credits import add_credits, deduct_credits
 from app.utils.id_gen import generate_id, generate_order_no
 
 # ---------------------------------------------------------------------------
@@ -61,8 +68,50 @@ _handler.setFormatter(logging.Formatter(
 if not logger.handlers:
     logger.addHandler(_handler)
 
-# Orders pending payment for longer than this are auto-cancelled
-ORDER_EXPIRE_MINUTES = 5
+# Order expire time in seconds (configurable via payment_order_timeout setting, default 180 seconds)
+DEFAULT_ORDER_EXPIRE_SECONDS = 180
+
+
+def _get_order_expire_seconds(db_configs: dict[str, str]) -> int:
+    """Get order expire time in seconds from config, with fallback to 180."""
+    try:
+        val = db_configs.get("payment_order_timeout", str(DEFAULT_ORDER_EXPIRE_SECONDS))
+        return int(val) if val.strip() else DEFAULT_ORDER_EXPIRE_SECONDS
+    except ValueError:
+        return DEFAULT_ORDER_EXPIRE_SECONDS
+
+
+# ---------------------------------------------------------------------------
+# Monkey-patch alipay-sdk-python WebUtils.do_post to fix bytes concatenation bug
+# The SDK's error handling does: '...' + response.read()
+# but response.read() returns bytes, causing TypeError on Python 3
+# ---------------------------------------------------------------------------
+def _patch_alipay_webutils():
+    try:
+        from alipay.aop.api.util import WebUtils
+        _original_do_post = WebUtils.do_post
+
+        def _patched_do_post(url, query_string, headers, params, charset, timeout=30):
+            try:
+                return _original_do_post(url, query_string, headers, params, charset, timeout)
+            except TypeError as e:
+                if "can only concatenate str (not 'bytes') to str" in str(e):
+                    # Decode bytes response to string and retry
+                    import http.client as _http
+                    from urllib.parse import urlparse as _urlparse
+                    parsed = _urlparse(url)
+                    conn = _http.HTTPSConnection(parsed.hostname)
+                    conn.request("POST", parsed.path + "?" + query_string, params, headers)
+                    resp = conn.getresponse()
+                    body = resp.read().decode("utf-8", errors="replace")
+                    raise RuntimeError(f"Alipay API error (status {resp.status}): {body}") from e
+                raise
+
+        WebUtils.do_post = _patched_do_post
+    except ImportError:
+        pass
+
+_patch_alipay_webutils()
 
 
 # ---------------------------------------------------------------------------
@@ -84,7 +133,9 @@ async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool
     """
     if order.status != "pending":
         return False
-    expiry = order.created_at + timedelta(minutes=ORDER_EXPIRE_MINUTES)
+    db_configs = await _get_payment_configs(db)
+    expire_seconds = _get_order_expire_seconds(db_configs)
+    expiry = order.created_at + timedelta(seconds=expire_seconds)
     if datetime.now(order.created_at.tzinfo) >= expiry:
         order.status = "cancelled"
         await db.flush()
@@ -92,6 +143,12 @@ async def _check_and_expire_order(db: AsyncSession, order: PaymentOrder) -> bool
             f"ORDER_EXPIRED order_no={order.order_no} user={order.user_id} "
             f"amount={order.amount} created_at={order.created_at.isoformat()}"
         )
+        # Also call Alipay close API if it was an Alipay order
+        if order.payment_method == "alipay":
+            try:
+                await _close_alipay_order(db, order, db_configs)
+            except Exception as e:
+                logger.exception(f"Failed to close Alipay order {order.order_no}: {e}")
         return True
     return False
 
@@ -100,7 +157,9 @@ async def expire_all_pending_orders(db: AsyncSession) -> int:
     """Background task: mark all expired pending orders as cancelled.
     Returns the number of orders expired.
     """
-    threshold = datetime.now() - timedelta(minutes=ORDER_EXPIRE_MINUTES)
+    db_configs = await _get_payment_configs(db)
+    expire_seconds = _get_order_expire_seconds(db_configs)
+    threshold = datetime.now() - timedelta(seconds=expire_seconds)
     result = await db.execute(
         select(PaymentOrder).where(
             PaymentOrder.status == "pending",
@@ -108,14 +167,22 @@ async def expire_all_pending_orders(db: AsyncSession) -> int:
         )
     )
     orders = result.scalars().all()
+    expired_count = 0
     for o in orders:
         o.status = "cancelled"
+        expired_count += 1
         logger.info(
             f"ORDER_EXPIRED order_no={o.order_no} user={o.user_id} amount={o.amount}"
         )
+        # Also call Alipay close API if it was an Alipay order
+        if o.payment_method == "alipay":
+            try:
+                await _close_alipay_order(db, o, db_configs)
+            except Exception as e:
+                logger.exception(f"Failed to close Alipay order {o.order_no}: {e}")
     if orders:
         await db.flush()
-    return len(orders)
+    return expired_count
 
 
 def _is_mock_mode(db_configs: dict[str, str]) -> bool:
@@ -157,15 +224,9 @@ def _get_alipay_client(app_id: str, private_key: str, public_key: str, gateway:
     config.alipay_public_key = public_key
     config.sign_type = "RSA2"
     config.charset = "utf-8"
-    config.cert_path = certifi.where()
-    logger.info(
-        f"Initializing Alipay client: app_id={app_id}, gateway={config.server_url}, "
-        f"public_key={config.alipay_public_key}, "
-        f"private_key={config.app_private_key}, "
-        f"cert_path={config.cert_path}"
-    )
+    
     try:
-        _alipay_client = DefaultAlipayClient(config)
+        _alipay_client = DefaultAlipayClient(config, logger)
         _alipay_client_app_id = app_id
     except Exception:
         logger.exception("Failed to initialize Alipay client")
@@ -291,8 +352,8 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str
     app_id = db_configs.get("payment_alipay_app_id", "")
     private_key = db_configs.get("payment_alipay_private_key", "")
     public_key = db_configs.get("payment_alipay_public_key", "")
-    notify_url = db_configs.get("payment_alipay_notify_url", "")
     gateway = db_configs.get("payment_alipay_gateway", "")
+    notify_url = db_configs.get("payment_alipay_notify_url", "")
 
     if not app_id or not private_key:
         logger.warning("Alipay config missing in database (app_id / private_key)")
@@ -309,11 +370,16 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str
         from alipay.aop.api.request.AlipayTradePrecreateRequest import (
             AlipayTradePrecreateRequest,
         )
+        from alipay.aop.api.response.AlipayTradePrecreateResponse import (
+            AlipayTradePrecreateResponse,
+        )
 
+        # 构造业务参数
         model = AlipayTradePrecreateModel()
         model.out_trade_no = order.order_no
         model.total_amount = f"{order.amount:.2f}"
         model.subject = f"充值订单 {order.order_no}"
+        model.product_code = "QR_CODE_OFFLINE"
 
         body_parts = []
         if order.credits > 0:
@@ -321,19 +387,31 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str
         if body_parts:
             model.body = " ".join(body_parts)
 
-        request = AlipayTradePrecreateRequest()
-        request.biz_model = model
+        # 构造请求
+        request = AlipayTradePrecreateRequest(biz_model=model)
+        
+        # 设置 notify_url 在 request 上
         if notify_url:
-            request.notify_url = notify_url
+            try:
+                if hasattr(request, 'set_notify_url'):
+                    request.set_notify_url(notify_url)
+                elif hasattr(request, 'notify_url'):
+                    request.notify_url = notify_url
+            except Exception as e:
+                logger.warning(f"Failed to set notify_url: {e}")
 
-        response = client.execute(request)
+        # 执行API调用
+        response_content = client.execute(request)
+        if not response_content:
+            logger.error(f"Alipay precreate failed: empty response, order_no={order.order_no}")
+            return None
 
-        if response.code == "10000":
+        # 解析响应结果
+        response = AlipayTradePrecreateResponse()
+        response.parse_response_content(response_content)
+
+        if response.is_success():
             qr_url = response.qr_code
-            logger.info(
-                f"Alipay precreate success: order_no={order.order_no}, "
-                f"qr_url={qr_url}"
-            )
             return qr_url
         else:
             logger.error(
@@ -343,11 +421,186 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str
             )
             return None
 
-    except Exception:
+    except Exception as e:
+        # 处理 SDK 内部的 bytes/str 错误
+        if "TypeError" in str(e) and ("bytes" in str(e) or "str" in str(e)):
+            logger.error(
+                f"Alipay SDK TypeError (bytes/str issue): order_no={order.order_no}, "
+                f"error={str(e)}"
+            )
         logger.exception(f"Alipay precreate exception: order_no={order.order_no}")
         return None
 
 
+# ---------------------------------------------------------------------------
+# Alipay order close
+# ---------------------------------------------------------------------------
+
+
+async def _close_alipay_order(db: AsyncSession, order: PaymentOrder, db_configs: dict[str, str]) -> bool:
+    """Call Alipay trade.close API to close an unpaid order.
+    Returns True if the order was closed successfully.
+    """
+    app_id = db_configs.get("payment_alipay_app_id", "")
+    private_key = db_configs.get("payment_alipay_private_key", "")
+    public_key = db_configs.get("payment_alipay_public_key", "")
+    gateway = db_configs.get("payment_alipay_gateway", "")
+
+    client = _get_alipay_client(app_id, private_key, public_key, gateway)
+    if client is None:
+        return False
+
+    mock_mode = _is_mock_mode(db_configs)
+    if mock_mode:
+        logger.info(f"Mock mode: skipping close_alipay_order for {order.order_no}")
+        return True
+
+    try:
+        from alipay.aop.api.domain.AlipayTradeCloseModel import AlipayTradeCloseModel
+        from alipay.aop.api.request.AlipayTradeCloseRequest import AlipayTradeCloseRequest
+        from alipay.aop.api.response.AlipayTradeCloseResponse import AlipayTradeCloseResponse
+
+        model = AlipayTradeCloseModel()
+        model.out_trade_no = order.order_no
+
+        request = AlipayTradeCloseRequest(biz_model=model)
+
+        response_content = client.execute(request)
+        if not response_content:
+            logger.error(f"Alipay close failed: empty response, order_no={order.order_no}")
+            return False
+
+        response = AlipayTradeCloseResponse()
+        response.parse_response_content(response_content)
+
+        if response.is_success():
+            logger.info(f"Alipay order closed: order_no={order.order_no}")
+            return True
+        else:
+            logger.error(
+                f"Alipay close failed: code={response.code}, "
+                f"msg={response.msg}, sub_code={response.sub_code}, "
+                f"sub_msg={response.sub_msg}, order_no={order.order_no}"
+            )
+            return False
+
+    except Exception as e:
+        if "TypeError" in str(e) and ("bytes" in str(e) or "str" in str(e)):
+            logger.error(
+                f"Alipay SDK TypeError (bytes/str issue) during close: order_no={order.order_no}, "
+                f"error={str(e)}"
+            )
+        logger.exception(f"Alipay close exception: order_no={order.order_no}")
+        return False
+
+
+# ---------------------------------------------------------------------------
+# Alipay order query
+# ---------------------------------------------------------------------------
+
+
+async def _query_alipay_order(db: AsyncSession, order: PaymentOrder, db_configs: dict[str, str]) -> dict | None:
+    """Call Alipay trade.query API to check order status.
+    Returns the response data if successful, None otherwise.
+    """
+    app_id = db_configs.get("payment_alipay_app_id", "")
+    private_key = db_configs.get("payment_alipay_private_key", "")
+    public_key = db_configs.get("payment_alipay_public_key", "")
+    gateway = db_configs.get("payment_alipay_gateway", "")
+
+    client = _get_alipay_client(app_id, private_key, public_key, gateway)
+    if client is None:
+        return None
+
+    mock_mode = _is_mock_mode(db_configs)
+    if mock_mode:
+        logger.info(f"Mock mode: skipping query_alipay_order for {order.order_no}")
+        return {"trade_status": "TRADE_FINISHED"}
+
+    try:
+        from alipay.aop.api.domain.AlipayTradeQueryModel import AlipayTradeQueryModel
+        from alipay.aop.api.request.AlipayTradeQueryRequest import AlipayTradeQueryRequest
+        from alipay.aop.api.response.AlipayTradeQueryResponse import AlipayTradeQueryResponse
+
+        model = AlipayTradeQueryModel()
+        model.out_trade_no = order.order_no
+
+        request = AlipayTradeQueryRequest(biz_model=model)
+
+        response_content = client.execute(request)
+        if not response_content:
+            logger.error(f"Alipay query failed: empty response, order_no={order.order_no}")
+            return None
+
+        response = AlipayTradeQueryResponse()
+        response.parse_response_content(response_content)
+
+        if response.is_success():
+            logger.info(f"Alipay query succeeded: order_no={order.order_no}, trade_status={response.trade_status}")
+            return {
+                "trade_no": response.trade_no,
+                "trade_status": response.trade_status,
+                "total_amount": response.total_amount,
+                "receipt_amount": response.receipt_amount,
+            }
+        else:
+            logger.error(
+                f"Alipay query failed: code={response.code}, "
+                f"msg={response.msg}, sub_code={response.sub_code}, "
+                f"sub_msg={response.sub_msg}, order_no={order.order_no}"
+            )
+            return None
+
+    except Exception as e:
+        if "TypeError" in str(e) and ("bytes" in str(e) or "str" in str(e)):
+            logger.error(
+                f"Alipay SDK TypeError (bytes/str issue) during query: order_no={order.order_no}, "
+                f"error={str(e)}"
+            )
+        logger.exception(f"Alipay query exception: order_no={order.order_no}")
+        return None
+
+
+async def sync_pending_orders(db: AsyncSession) -> int:
+    """Check pending orders via Alipay query and update status.
+    Returns the number of orders updated.
+    """
+    result = await db.execute(
+        select(PaymentOrder).where(
+            PaymentOrder.status == "pending",
+        )
+    )
+    orders = result.scalars().all()
+    updated_count = 0
+
+    db_configs = await _get_payment_configs(db)
+
+    for order in orders:
+        if order.payment_method != "alipay":
+            continue
+
+        try:
+            data = await _query_alipay_order(db, order, db_configs)
+            if data:
+                trade_status = data.get("trade_status")
+                if trade_status in ("TRADE_SUCCESS", "TRADE_FINISHED"):
+                    # Order was paid but we missed the callback
+                    trade_no = data.get("trade_no", "")
+                    await process_payment_success_by_order_no(db, order.order_no, trade_no)
+                    updated_count += 1
+                elif trade_status in ("TRADE_CLOSED", "TRADE_CANCELLED"):
+                    # Order was closed on Alipay side
+                    order.status = "cancelled"
+                    await db.flush()
+                    updated_count += 1
+        except Exception as e:
+            logger.exception(f"Failed to sync order {order.order_no}: {e}")
+
+    if updated_count > 0:
+        await db.flush()
+    return updated_count
+
+
 # ---------------------------------------------------------------------------
 # Alipay callback verification
 # ---------------------------------------------------------------------------
@@ -356,12 +609,12 @@ def _create_alipay_order(order: PaymentOrder, db_configs: dict[str, str]) -> str
 async def verify_alipay_callback(data: dict, db: AsyncSession) -> bool:
     """Verify Alipay payment callback (async notify) signature.
 
-    Reads the Alipay public key from the database and uses the SDK's
-    built-in RSA2 verification.
+    Reads the Alipay public key from the database and uses RSA2 verification.
     """
     db_configs = await _get_payment_configs(db)
     mock_mode = _is_mock_mode(db_configs)
     if mock_mode:
+        logger.info("Mock mode enabled, skipping Alipay callback verification")
         return True
 
     public_key = db_configs.get("payment_alipay_public_key", "")
@@ -375,37 +628,119 @@ async def verify_alipay_callback(data: dict, db: AsyncSession) -> bool:
             logger.warning("Alipay callback missing 'sign' field")
             return False
 
+        sign_type = data.get("sign_type", "RSA2")
+
         # Build verification params (exclude sign and sign_type)
         verify_data = {
             k: v for k, v in data.items()
             if k not in ("sign", "sign_type") and v is not None and v != ""
         }
 
-        from alipay.aop.api.util.Signature import verify_with_rsa
-
+        # Generate sign content: sorted keys, key=value format
         sign_content = "&".join(
             f"{k}={v}" for k, v in sorted(verify_data.items())
         )
 
-        is_valid = verify_with_rsa(
-            public_key.encode("utf-8"),
-            sign_content.encode("utf-8"),
-            sign,
-        )
+        # logger.info(f"Verifying Alipay callback sign_content: {sign_content[:100]}...")
+        # logger.info(f"Sign type: {sign_type}")
+
+        # 实现 RSA2 签名验证
+        is_valid = _verify_alipay_sign(public_key, sign_content, sign, sign_type)
 
         if not is_valid:
             logger.warning("Alipay callback signature verification FAILED")
+        else:
+            logger.info("Alipay callback signature verification SUCCESS")
 
         return is_valid
 
-    except ImportError:
-        logger.error("alipay-sdk-python not installed, skipping signature verification")
-        return True
     except Exception:
         logger.exception("Alipay callback verification error")
         return False
 
 
+def _verify_alipay_sign(public_key: str, sign_content: str, sign: str, sign_type: str = "RSA2") -> bool:
+    """Verify Alipay RSA/RSA2 signature.
+    
+    Args:
+        public_key: Alipay public key (PEM format, with or without headers)
+        sign_content: Original content to verify
+        sign: Base64 encoded signature
+        sign_type: "RSA" (SHA1) or "RSA2" (SHA256)
+    
+    Returns:
+        True if signature is valid
+    """
+    try:
+        import base64
+        from hashlib import sha1, sha256
+        
+        # 处理公钥，确保有正确的格式
+        pub_key = public_key.strip()
+        if not pub_key.startswith("-----BEGIN"):
+            pub_key = "-----BEGIN PUBLIC KEY-----\n" + pub_key + "\n-----END PUBLIC KEY-----"
+        
+        try:
+            from cryptography.hazmat.primitives import hashes
+            from cryptography.hazmat.primitives.asymmetric import padding
+            from cryptography.hazmat.primitives import serialization
+            from cryptography.hazmat.backends import default_backend
+            
+            # 加载公钥
+            public_key_obj = serialization.load_pem_public_key(
+                pub_key.encode("utf-8"),
+                backend=default_backend()
+            )
+            
+            # 选择哈希算法
+            if sign_type == "RSA2":
+                hash_alg = hashes.SHA256()
+            else:
+                hash_alg = hashes.SHA1()
+            
+            # 验证签名
+            public_key_obj.verify(
+                base64.b64decode(sign),
+                sign_content.encode("utf-8"),
+                padding.PKCS1v15(),
+                hash_alg
+            )
+            return True
+            
+        except ImportError:
+            # 如果没有 cryptography，尝试使用 rsa 库
+            try:
+                import rsa
+                
+                # 加载公钥
+                pub_key_obj = rsa.PublicKey.load_pkcs1_openssl_pem(pub_key.encode("utf-8"))
+                
+                # 选择哈希算法
+                if sign_type == "RSA2":
+                    hash_func = 'SHA-256'
+                else:
+                    hash_func = 'SHA-1'
+                
+                # 验证签名
+                rsa.verify(
+                    sign_content.encode("utf-8"),
+                    base64.b64decode(sign),
+                    pub_key_obj,
+                    hash_func
+                )
+                return True
+                
+            except ImportError:
+                logger.error("Neither cryptography nor rsa library installed, cannot verify signature")
+                # 如果没有任何加密库，在生产环境应该返回 False，但这里我们记录警告并继续
+                logger.warning("Skipping signature verification due to missing crypto libraries")
+                return False
+                
+    except Exception as e:
+        logger.exception(f"Signature verification failed: {e}")
+        return False
+
+
 # ---------------------------------------------------------------------------
 # WeChat callback verification (stub)
 # ---------------------------------------------------------------------------
@@ -429,7 +764,7 @@ async def verify_wechat_callback(data: dict, db: AsyncSession) -> bool:
 async def process_payment_success(db: AsyncSession, order_id: str):
     """Process successful payment: update order and add credits."""
     result = await db.execute(
-        select(PaymentOrder).where(PaymentOrder.id == order_id).limit(1)
+        select(PaymentOrder).where(PaymentOrder.id == order_id).with_for_update().limit(1)
     )
     order = result.scalar_one_or_none()
     if not order or order.status != "pending":
@@ -444,23 +779,50 @@ async def process_payment_success(db: AsyncSession, order_id: str):
         f"充值成功({order.credits}积分)",
         related_id=order.id,
     )
-    await db.flush()
+    await db.commit()
 
 
-async def process_payment_success_by_order_no(db: AsyncSession, order_no: str, trade_no: str = ""):
+async def process_payment_success_by_order_no(
+    db: AsyncSession,
+    order_no: str,
+    trade_no: str = "",
+    total_amount: float | None = None
+):
     """Process successful payment by order_no (used by Alipay/WeChat callbacks).
 
     Args:
         db: async database session
         order_no: the merchant order number (out_trade_no)
         trade_no: the Alipay trade number (trade_no), optional
+        total_amount: the payment amount from the gateway, for consistency check
     """
     result = await db.execute(
-        select(PaymentOrder).where(PaymentOrder.order_no == order_no).limit(1)
+        select(PaymentOrder).where(PaymentOrder.order_no == order_no).with_for_update().limit(1)
     )
     order = result.scalar_one_or_none()
-    if not order or order.status != "pending":
-        logger.info(f"Order {order_no} not found or already processed, skipping")
+    
+    if not order:
+        logger.info(f"Order {order_no} not found, skipping")
+        return
+    
+    if order.status == "paid":
+        logger.info(f"Order {order_no} already processed, skipping")
+        return
+    
+    if order.status != "pending":
+        logger.info(f"Order {order_no} is in {order.status} state, cannot process")
+        return
+
+    # 金额一致性校验
+    if total_amount is not None and abs(total_amount - order.amount) > 0.01:
+        logger.error(
+            f"Amount mismatch: order amount {order.amount}, gateway amount {total_amount}"
+        )
+        return
+
+    # 幂等性检查：如果trade_no已存在且相同，则跳过
+    if trade_no and order.trade_no and order.trade_no == trade_no:
+        logger.info(f"Trade no {trade_no} already processed, skipping")
         return
 
     order.status = "paid"
@@ -475,8 +837,143 @@ async def process_payment_success_by_order_no(db: AsyncSession, order_no: str, t
         f"充值成功({order.credits}积分)",
         related_id=order.id,
     )
-    await db.flush()
+    await db.commit()
     logger.info(
         f"PAYMENT_SUCCESS order_no={order_no} user={order.user_id} "
         f"amount={order.amount} credits={order.credits} trade_no={trade_no}"
     )
+
+
+async def process_refund(
+    db: AsyncSession,
+    order_no: str,
+    refund_amount: float | None = None,
+    refund_reason: str = "管理员退款"
+) -> dict:
+    """Process a refund for a paid order.
+
+    Args:
+        db: async database session
+        order_no: merchant order number
+        refund_amount: amount to refund (defaults to full order amount)
+        refund_reason: reason for refund
+
+    Returns:
+        dict with refund result
+    """
+    result = await db.execute(
+        select(PaymentOrder).where(PaymentOrder.order_no == order_no).with_for_update().limit(1)
+    )
+    order = result.scalar_one_or_none()
+    
+    if not order:
+        return {"success": False, "message": "订单不存在"}
+    
+    if order.status != "paid":
+        return {"success": False, "message": f"订单状态为{order.status}，无法退款"}
+    
+    if order.refunded_at is not None:
+        return {"success": False, "message": "订单已退款"}
+
+    refund_amount = refund_amount or order.amount
+    
+    # 金额校验
+    if refund_amount > order.amount:
+        return {"success": False, "message": "退款金额超过订单金额"}
+    
+    # 如果是支付宝订单，调用支付宝退款API
+    db_configs = await _get_payment_configs(db)
+    if order.payment_method == "alipay":
+        refund_result = await _refund_alipay_order(
+            db, order, refund_amount, refund_reason, db_configs
+        )
+        if not refund_result.get("success"):
+            return refund_result
+    
+    # 扣除积分
+    try:
+        await deduct_credits(
+            db,
+            order.user_id,
+            order.credits,
+            refund_reason,
+            related_id=order.id,
+        )
+    except Exception as e:
+        logger.exception(f"Failed to deduct credits for refund: {e}")
+        return {"success": False, "message": "积分扣除失败"}
+    
+    # 更新订单状态
+    order.status = "refunded"
+    order.refund_amount = refund_amount
+    order.refunded_at = datetime.now()
+    if order.payment_method == "alipay":
+        order.refund_trade_no = db_configs.get("refund_trade_no", "")
+    
+    await db.commit()
+    logger.info(
+        f"REFUND_SUCCESS order_no={order_no} user={order.user_id} "
+        f"refund_amount={refund_amount}"
+    )
+    return {"success": True, "message": "退款成功"}
+
+
+async def _refund_alipay_order(
+    db: AsyncSession,
+    order: PaymentOrder,
+    refund_amount: float,
+    refund_reason: str,
+    db_configs: dict[str, str]
+) -> dict:
+    """Call Alipay refund API."""
+    app_id = db_configs.get("payment_alipay_app_id", "")
+    private_key = db_configs.get("payment_alipay_private_key", "")
+    public_key = db_configs.get("payment_alipay_public_key", "")
+    gateway = db_configs.get("payment_alipay_gateway", "")
+
+    client = _get_alipay_client(app_id, private_key, public_key, gateway)
+    if client is None:
+        return {"success": False, "message": "支付宝客户端初始化失败"}
+
+    mock_mode = _is_mock_mode(db_configs)
+    if mock_mode:
+        logger.info(f"Mock mode: skipping alipay refund for {order.order_no}")
+        return {"success": True}
+
+    try:
+        from alipay.aop.api.domain.AlipayTradeRefundModel import AlipayTradeRefundModel
+        from alipay.aop.api.request.AlipayTradeRefundRequest import AlipayTradeRefundRequest
+        from alipay.aop.api.response.AlipayTradeRefundResponse import AlipayTradeRefundResponse
+
+        model = AlipayTradeRefundModel()
+        model.out_trade_no = order.order_no
+        model.refund_amount = f"{refund_amount:.2f}"
+        model.refund_reason = refund_reason
+        model.out_request_no = f"{order.order_no}_refund_{int(datetime.now().timestamp())}"
+
+        request = AlipayTradeRefundRequest(biz_model=model)
+        response_content = client.execute(request)
+        
+        if not response_content:
+            logger.error(f"Alipay refund failed: empty response, order_no={order.order_no}")
+            return {"success": False, "message": "支付宝退款响应为空"}
+
+        response = AlipayTradeRefundResponse()
+        response.parse_response_content(response_content)
+
+        if response.is_success():
+            logger.info(f"Alipay refund succeeded: order_no={order.order_no}")
+            return {"success": True, "trade_no": response.trade_no}
+        else:
+            logger.error(
+                f"Alipay refund failed: code={response.code}, "
+                f"msg={response.msg}, sub_code={response.sub_code}, "
+                f"sub_msg={response.sub_msg}, order_no={order.order_no}"
+            )
+            return {
+                "success": False,
+                "message": f"支付宝退款失败: {response.sub_msg or response.msg}"
+            }
+    except Exception as e:
+        logger.exception(f"Alipay refund exception: order_no={order.order_no}, {e}")
+        return {"success": False, "message": f"支付宝退款异常: {str(e)}"}

video-gen-api/app/services/user_oauth_app_service.py

@@ -62,6 +62,9 @@ async def create_user_oauth_app(
     secret: str,
     open_type: int,
     create_by: str | None = None,
+    count: int = 100,
+    auth_url: str | None = None,
+    company: str | None = None,
 ) -> UserOAuthApp:
     existing = await get_user_oauth_app_by_app_id(db, app_id)
     if existing:
@@ -72,6 +75,9 @@ async def create_user_oauth_app(
         app_id=app_id,
         secret=secret,
         open_type=open_type,
+        count=count,
+        auth_url=auth_url,
+        company=company,
         create_by=create_by,
     )
     db.add(app)
@@ -87,6 +93,9 @@ async def update_user_oauth_app(
     secret: str | None = None,
     open_type: int | None = None,
     status: int | None = None,
+    count: int | None = None,
+    auth_url: str | None = None,
+    company: str | None = None,
     create_by: str | None = None,
 ) -> UserOAuthApp | None:
     app = await get_user_oauth_app_by_id(db, id)
@@ -99,6 +108,12 @@ async def update_user_oauth_app(
         app.open_type = open_type
     if status is not None:
         app.status = status
+    if count is not None:
+        app.count = count
+    if auth_url is not None:
+        app.auth_url = auth_url
+    if company is not None:
+        app.company = company
     if create_by is not None:
         app.create_by = create_by
 

video-gen-api/app/utils/id_gen.py

@@ -1,5 +1,7 @@
 import time
 import random
+from datetime import datetime
+import secrets
 
 
 def generate_id() -> str:
@@ -10,7 +12,10 @@ def generate_id() -> str:
 
 
 def generate_order_no() -> str:
-    """Generate a human-readable order number."""
-    timestamp = int(time.time())
-    randomness = random.randint(1000, 9999)
-    return f"VG{timestamp}{randomness}"
+    """Generate a human-readable order number with yyyymmddhhmmss format."""
+    # 格式化为 yyyymmddhhmmss 格式的时间戳
+    now = datetime.now()
+    timestamp = now.strftime("%Y%m%d%H%M%S")
+    # 使用密码学安全的随机数生成 8位纯数字，防止并发冲突
+    random_part = ''.join(str(secrets.randbelow(10)) for _ in range(8))
+    return f"MZZC{timestamp}{random_part}"