"""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()