feat: add avatar payments and finance management

2026-09-08 18:47:29 +08:00
parent 62eb9578fd
commit 8585d101d5
36 changed files with 10228 additions and 43 deletions
@@ -1,6 +1,6 @@
 """API路由汇总"""
 from fastapi import APIRouter
-from app.api.endpoints import users, interactions, ai_models, dashboard, system, logs, avatars
+from app.api.endpoints import users, interactions, ai_models, dashboard, system, logs, avatars, finance

 router = APIRouter()

@@ -11,3 +11,4 @@ router.include_router(dashboard.router, prefix="/dashboard", tags=["数据看板
 router.include_router(system.router, prefix="/system", tags=["系统设置"])
 router.include_router(logs.router, prefix="/logs", tags=["日志管理"])
 router.include_router(avatars.router, prefix="/avatars", tags=["数字分身管理"])
+router.include_router(finance.router, prefix="/finance", tags=["财务管理"])
@@ -0,0 +1,130 @@
+"""Admin finance API for avatar Token orders, refunds and invoices."""
+
+from fastapi import APIRouter, Body, HTTPException, Query
+
+from app.schemas import ApiResponse
+from app.services.avatar_service import get_session, is_available
+from app.services.finance_service import FinanceServiceError, finance_service
+
+
+router = APIRouter()
+
+
+def _session():
+    if not is_available():
+        raise HTTPException(status_code=503, detail="数字分身数据库尚未初始化")
+    return get_session()
+
+
+def _raise(exc: FinanceServiceError):
+    raise HTTPException(status_code=exc.status_code, detail=str(exc))
+
+
+@router.get("/summary")
+def summary():
+    db = _session()
+    try:
+        return ApiResponse(data=finance_service.summary(db))
+    finally:
+        db.close()
+
+
+@router.get("/orders")
+def orders(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    keyword: str = Query(""),
+    status: str = Query(""),
+    provider: str = Query(""),
+):
+    db = _session()
+    try:
+        total, items = finance_service.list_orders(
+            db, page=page, page_size=page_size, keyword=keyword.strip(), status=status, provider=provider
+        )
+        return ApiResponse(data={"total": total, "page": page, "page_size": page_size, "items": items})
+    finally:
+        db.close()
+
+
+@router.patch("/orders/{order_no}/status")
+def update_order_status(order_no: str, body: dict = Body(...)):
+    db = _session()
+    try:
+        finance_service.close_order(
+            db, order_no, status=str(body.get("status") or ""), reason=str(body.get("reason") or "")
+        )
+        return ApiResponse(message="订单状态已更新")
+    except FinanceServiceError as exc:
+        _raise(exc)
+    finally:
+        db.close()
+
+
+@router.post("/orders/{order_no}/refund")
+def request_refund(order_no: str, body: dict = Body(...)):
+    try:
+        data = finance_service.request_refund(
+            order_no,
+            reason=str(body.get("reason") or "").strip(),
+            operator=str(body.get("operator") or "后台管理员").strip(),
+        )
+        return ApiResponse(data=data, message="退款申请已提交")
+    except FinanceServiceError as exc:
+        _raise(exc)
+
+
+@router.get("/refunds")
+def refunds(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    status: str = Query(""),
+):
+    db = _session()
+    try:
+        total, items = finance_service.list_refunds(db, page=page, page_size=page_size, status=status)
+        return ApiResponse(data={"total": total, "page": page, "page_size": page_size, "items": items})
+    finally:
+        db.close()
+
+
+@router.post("/refunds/{refund_no}/confirm")
+def confirm_refund(refund_no: str, body: dict = Body(...)):
+    try:
+        data = finance_service.confirm_refund(refund_no, body)
+        return ApiResponse(data=data, message="退款结果已登记")
+    except FinanceServiceError as exc:
+        _raise(exc)
+
+
+@router.get("/invoices")
+def invoices(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    status: str = Query(""),
+):
+    db = _session()
+    try:
+        total, items = finance_service.list_invoices(db, page=page, page_size=page_size, status=status)
+        return ApiResponse(data={"total": total, "page": page, "page_size": page_size, "items": items})
+    finally:
+        db.close()
+
+
+@router.patch("/invoices/{invoice_id}")
+def update_invoice(invoice_id: str, body: dict = Body(...)):
+    db = _session()
+    try:
+        finance_service.update_invoice(
+            db,
+            invoice_id,
+            status=str(body.get("status") or ""),
+            invoice_no=str(body.get("invoiceNo") or ""),
+            invoice_url=str(body.get("invoiceUrl") or ""),
+            remark=str(body.get("remark") or ""),
+        )
+        return ApiResponse(message="发票申请已处理")
+    except FinanceServiceError as exc:
+        _raise(exc)
+    finally:
+        db.close()
@@ -35,7 +35,7 @@ def _get_engine_and_session():
            return None, None
        _engine = create_engine(
            f"sqlite:///{db_path}",
-            connect_args={"check_same_thread": False},
+            connect_args={"check_same_thread": False, "timeout": 30},
        )
        _SessionLocal = sessionmaker(bind=_engine, autoflush=False, expire_on_commit=False)
    return _engine, _SessionLocal()
@@ -0,0 +1,254 @@
+"""Finance operations for digital-avatar Token purchases."""
+
+import os
+from datetime import datetime
+
+import httpx
+from sqlalchemy import text
+
+from app.core.config import settings
+
+
+class FinanceServiceError(RuntimeError):
+    def __init__(self, message: str, status_code: int = 400):
+        super().__init__(message)
+        self.status_code = status_code
+
+
+def _mapping(row):
+    return dict(row._mapping) if row is not None else None
+
+
+def _iso(value):
+    return value.isoformat() if hasattr(value, "isoformat") else value
+
+
+def _money(cents):
+    return round(int(cents or 0) / 100, 2)
+
+
+class FinanceService:
+    @staticmethod
+    def _tables_ready(db) -> bool:
+        names = {
+            row[0]
+            for row in db.execute(text(
+                "SELECT name FROM sqlite_master WHERE type='table' AND "
+                "name IN ('token_payment_orders','payment_refunds','invoice_applications')"
+            )).fetchall()
+        }
+        return len(names) == 3
+
+    @classmethod
+    def summary(cls, db) -> dict:
+        if not cls._tables_ready(db):
+            return {
+                "paid_revenue": 0,
+                "paid_orders": 0,
+                "pending_orders": 0,
+                "processing_refunds": 0,
+                "pending_invoices": 0,
+            }
+        row = db.execute(text("""
+            SELECT
+                COALESCE(SUM(CASE WHEN status='paid' THEN price_cents ELSE 0 END), 0) paid_revenue,
+                SUM(CASE WHEN status='paid' THEN 1 ELSE 0 END) paid_orders,
+                SUM(CASE WHEN status='pending' THEN 1 ELSE 0 END) pending_orders
+            FROM token_payment_orders
+        """)).fetchone()
+        processing_refunds = db.execute(text(
+            "SELECT COUNT(*) FROM payment_refunds WHERE status IN ('pending','processing')"
+        )).scalar() or 0
+        pending_invoices = db.execute(text(
+            "SELECT COUNT(*) FROM invoice_applications WHERE status='pending'"
+        )).scalar() or 0
+        return {
+            "paid_revenue": _money(row.paid_revenue),
+            "paid_orders": int(row.paid_orders or 0),
+            "pending_orders": int(row.pending_orders or 0),
+            "processing_refunds": int(processing_refunds),
+            "pending_invoices": int(pending_invoices),
+        }
+
+    @classmethod
+    def list_orders(cls, db, *, page=1, page_size=20, keyword="", status="", provider=""):
+        if not cls._tables_ready(db):
+            return 0, []
+        clauses = ["1=1"]
+        params = {}
+        if keyword:
+            clauses.append("(o.order_no LIKE :keyword OR u.phone LIKE :keyword OR u.nickname LIKE :keyword)")
+            params["keyword"] = f"%{keyword}%"
+        if status:
+            clauses.append("o.status=:status")
+            params["status"] = status
+        if provider:
+            clauses.append("o.provider=:provider")
+            params["provider"] = provider
+        where = " AND ".join(clauses)
+        total = db.execute(text(f"""
+            SELECT COUNT(*) FROM token_payment_orders o
+            LEFT JOIN users u ON u.id=o.user_id WHERE {where}
+        """), params).scalar() or 0
+        params.update({"limit": page_size, "offset": (page - 1) * page_size})
+        rows = db.execute(text(f"""
+            SELECT o.*, u.nickname user_nickname, u.phone user_phone,
+                   i.status invoice_status, i.id invoice_id
+            FROM token_payment_orders o
+            LEFT JOIN users u ON u.id=o.user_id
+            LEFT JOIN invoice_applications i ON i.order_no=o.order_no
+            WHERE {where}
+            ORDER BY o.created_at DESC LIMIT :limit OFFSET :offset
+        """), params).fetchall()
+        items = []
+        for row in rows:
+            item = _mapping(row)
+            item["price"] = _money(item.pop("price_cents"))
+            for key in ("created_at", "updated_at", "paid_at", "refunded_at"):
+                item[key] = _iso(item.get(key))
+            item.pop("pay_message", None)
+            items.append(item)
+        return int(total), items
+
+    @classmethod
+    def list_refunds(cls, db, *, page=1, page_size=20, status=""):
+        if not cls._tables_ready(db):
+            return 0, []
+        where = "WHERE r.status=:status" if status else ""
+        params = {"status": status} if status else {}
+        total = db.execute(text(f"SELECT COUNT(*) FROM payment_refunds r {where}"), params).scalar() or 0
+        params.update({"limit": page_size, "offset": (page - 1) * page_size})
+        rows = db.execute(text(f"""
+            SELECT r.*, o.provider, o.payment_method, u.nickname user_nickname, u.phone user_phone
+            FROM payment_refunds r
+            JOIN token_payment_orders o ON o.order_no=r.order_no
+            LEFT JOIN users u ON u.id=o.user_id
+            {where}
+            ORDER BY r.created_at DESC LIMIT :limit OFFSET :offset
+        """), params).fetchall()
+        items = []
+        for row in rows:
+            item = _mapping(row)
+            item["amount"] = _money(item.pop("amount_cents"))
+            for key in ("created_at", "updated_at", "completed_at"):
+                item[key] = _iso(item.get(key))
+            items.append(item)
+        return int(total), items
+
+    @classmethod
+    def list_invoices(cls, db, *, page=1, page_size=20, status=""):
+        if not cls._tables_ready(db):
+            return 0, []
+        where = "WHERE i.status=:status" if status else ""
+        params = {"status": status} if status else {}
+        total = db.execute(text(f"SELECT COUNT(*) FROM invoice_applications i {where}"), params).scalar() or 0
+        params.update({"limit": page_size, "offset": (page - 1) * page_size})
+        rows = db.execute(text(f"""
+            SELECT i.*, u.nickname user_nickname, u.phone user_phone
+            FROM invoice_applications i
+            LEFT JOIN users u ON u.id=i.user_id
+            {where}
+            ORDER BY i.created_at DESC LIMIT :limit OFFSET :offset
+        """), params).fetchall()
+        items = []
+        for row in rows:
+            item = _mapping(row)
+            item["amount"] = _money(item.pop("amount_cents"))
+            for key in ("created_at", "updated_at", "issued_at"):
+                item[key] = _iso(item.get(key))
+            items.append(item)
+        return int(total), items
+
+    @staticmethod
+    def close_order(db, order_no: str, *, status: str, reason: str):
+        if status not in {"closed", "failed"}:
+            raise FinanceServiceError("后台只能将待支付订单关闭或标记失败")
+        order = db.execute(text(
+            "SELECT status FROM token_payment_orders WHERE order_no=:order_no"
+        ), {"order_no": order_no}).fetchone()
+        if not order:
+            raise FinanceServiceError("订单不存在", 404)
+        if order.status != "pending":
+            raise FinanceServiceError("只有待支付订单可以修改状态", 409)
+        db.execute(text("""
+            UPDATE token_payment_orders
+            SET status=:status, failure_reason=:reason, updated_at=:updated_at
+            WHERE order_no=:order_no
+        """), {
+            "status": status,
+            "reason": (reason or "后台关闭订单")[:500],
+            "updated_at": datetime.utcnow(),
+            "order_no": order_no,
+        })
+        db.commit()
+
+    @staticmethod
+    def _avatar_admin_call(path: str, payload: dict):
+        base_url = (settings.AVATAR_BACKEND_URL or os.getenv("AVATAR_BACKEND_URL", "")).rstrip("/")
+        secret = os.getenv("AVATAR_FINANCE_ADMIN_SECRET", "").strip()
+        if not base_url or len(secret) < 16:
+            raise FinanceServiceError("数字分身财务服务尚未完成配置", 503)
+        try:
+            response = httpx.post(
+                f"{base_url}/api{path}",
+                json=payload,
+                headers={"X-Avatar-Finance-Key": secret},
+                timeout=35,
+            )
+            data = response.json()
+        except (httpx.HTTPError, ValueError) as exc:
+            raise FinanceServiceError("数字分身财务服务暂时不可用", 502) from exc
+        if response.status_code >= 400 or data.get("code") not in (0, 200, "0", "200"):
+            raise FinanceServiceError(data.get("message") or data.get("detail") or "财务操作失败", response.status_code)
+        return data.get("data")
+
+    @classmethod
+    def request_refund(cls, order_no: str, *, reason: str, operator: str):
+        return cls._avatar_admin_call(
+            f"/token/admin/orders/{order_no}/refund",
+            {"reason": reason, "operator": operator},
+        )
+
+    @classmethod
+    def confirm_refund(cls, refund_no: str, payload: dict):
+        return cls._avatar_admin_call(f"/token/admin/refunds/{refund_no}/confirm", payload)
+
+    @staticmethod
+    def update_invoice(db, invoice_id: str, *, status: str, invoice_no="", invoice_url="", remark=""):
+        row = db.execute(text(
+            "SELECT * FROM invoice_applications WHERE id=:invoice_id"
+        ), {"invoice_id": invoice_id}).fetchone()
+        if not row:
+            raise FinanceServiceError("发票申请不存在", 404)
+        if row.status != "pending":
+            raise FinanceServiceError("该发票申请已处理", 409)
+        if status == "issued":
+            if not invoice_no.strip():
+                raise FinanceServiceError("请填写发票号码")
+            if invoice_url.strip() and not invoice_url.strip().lower().startswith(("https://", "http://")):
+                raise FinanceServiceError("电子发票地址必须是 HTTP 或 HTTPS 链接")
+            issued_at = datetime.utcnow()
+        elif status == "rejected":
+            if not remark.strip():
+                raise FinanceServiceError("请填写驳回原因")
+            issued_at = None
+        else:
+            raise FinanceServiceError("发票状态只能是已开具或已驳回")
+        db.execute(text("""
+            UPDATE invoice_applications
+            SET status=:status, invoice_no=:invoice_no, invoice_url=:invoice_url,
+                remark=:remark, issued_at=:issued_at, updated_at=:updated_at
+            WHERE id=:invoice_id
+        """), {
+            "status": status,
+            "invoice_no": invoice_no.strip()[:120],
+            "invoice_url": invoice_url.strip()[:500],
+            "remark": remark.strip()[:500],
+            "issued_at": issued_at,
+            "updated_at": datetime.utcnow(),
+            "invoice_id": invoice_id,
+        })
+        db.commit()
+
+
+finance_service = FinanceService()
@@ -0,0 +1,91 @@
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import Session
+
+from app.services.finance_service import FinanceService
+
+
+def _db():
+    engine = create_engine("sqlite:///:memory:")
+    db = Session(engine)
+    db.execute(text("""
+        CREATE TABLE users (id TEXT PRIMARY KEY, nickname TEXT, phone TEXT)
+    """))
+    db.execute(text("""
+        CREATE TABLE token_payment_orders (
+            id TEXT, order_no TEXT PRIMARY KEY, user_id TEXT, plan_id TEXT,
+            payment_method TEXT, pay_type TEXT, pay_way TEXT, points_amount INTEGER,
+            price_cents INTEGER, status TEXT, provider TEXT, provider_order_id TEXT,
+            provider_order_no TEXT, provider_status TEXT, pay_message TEXT,
+            failure_reason TEXT, refund_status TEXT, created_at TEXT, updated_at TEXT,
+            paid_at TEXT, refunded_at TEXT
+        )
+    """))
+    db.execute(text("""
+        CREATE TABLE payment_refunds (
+            id TEXT, refund_no TEXT, order_no TEXT, amount_cents INTEGER,
+            points_amount INTEGER, reason TEXT, status TEXT, provider_refund_no TEXT,
+            requested_by TEXT, failure_reason TEXT, created_at TEXT, updated_at TEXT,
+            completed_at TEXT
+        )
+    """))
+    db.execute(text("""
+        CREATE TABLE invoice_applications (
+            id TEXT, order_no TEXT, user_id TEXT, amount_cents INTEGER, title TEXT,
+            invoice_type TEXT, tax_number TEXT, email TEXT, status TEXT,
+            invoice_no TEXT, invoice_url TEXT, remark TEXT, created_at TEXT,
+            updated_at TEXT, issued_at TEXT
+        )
+    """))
+    db.execute(text("INSERT INTO users VALUES ('u1','测试用户','13800000000')"))
+    db.execute(text("""
+        INSERT INTO token_payment_orders VALUES (
+            'o1','AV1','u1','1','wechat','WECHAT','APP',2000000,1000,'paid','huihui',
+            '','','SUCCESS','secret-payment-message','','none','2026-09-08 12:00:00',
+            '2026-09-08 12:01:00','2026-09-08 12:01:00',NULL
+        )
+    """))
+    db.execute(text("""
+        INSERT INTO invoice_applications VALUES (
+            'i1','AV1','u1',1000,'测试用户','personal','','u@example.com','pending',
+            '','','','2026-09-08 12:02:00','2026-09-08 12:02:00',NULL
+        )
+    """))
+    db.commit()
+    return db
+
+
+def test_finance_summary_and_orders_hide_provider_payment_payload():
+    db = _db()
+    try:
+        summary = FinanceService.summary(db)
+        assert summary == {
+            "paid_revenue": 10.0,
+            "paid_orders": 1,
+            "pending_orders": 0,
+            "processing_refunds": 0,
+            "pending_invoices": 1,
+        }
+        total, orders = FinanceService.list_orders(db, keyword="测试用户")
+        assert total == 1
+        assert orders[0]["price"] == 10.0
+        assert orders[0]["invoice_status"] == "pending"
+        assert "pay_message" not in orders[0]
+    finally:
+        db.close()
+
+
+def test_invoice_can_be_issued_and_pending_order_can_be_closed():
+    db = _db()
+    try:
+        FinanceService.update_invoice(db, "i1", status="issued", invoice_no="FP-001", invoice_url="", remark="")
+        assert db.execute(text("SELECT status, invoice_no FROM invoice_applications WHERE id='i1'" )).fetchone() == ("issued", "FP-001")
+        db.execute(text("""
+            INSERT INTO token_payment_orders
+            (id,order_no,user_id,plan_id,payment_method,pay_type,pay_way,points_amount,price_cents,status,provider,refund_status)
+            VALUES ('o2','AV2','u1','1','alipay','ALIPAY','H5',1,100,'pending','huihui','none')
+        """))
+        db.commit()
+        FinanceService.close_order(db, "AV2", status="closed", reason="超时")
+        assert db.execute(text("SELECT status FROM token_payment_orders WHERE order_no='AV2'" )).scalar() == "closed"
+    finally:
+        db.close()