From 70e0e5b034861053da420e15eef72fd10c345e4e Mon Sep 17 00:00:00 2001 From: xingc Date: Thu, 3 Apr 2025 17:21:29 +0800 Subject: [PATCH] =?UTF-8?q?feat(all):=20=E5=90=8C=E6=AD=A5=E6=9C=80?= =?UTF-8?q?=E6=96=B0=E8=BF=9B=E5=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.dev => .env.example | 5 +- associated_script.py | 6 + config.py | 5 + core/app.py | 52 ++++- core/base/exceptions.py | 5 + core/base/schema.py | 19 ++ core/db/crud.py | 64 +++--- core/db/models.py | 192 ++++++++++++------ core/routers/__init__.py | 9 +- core/routers/auth/routes.py | 3 +- core/routers/card/routes.py | 1 - core/routers/card/schema.py | 10 +- core/routers/{im => chat}/__init__.py | 0 core/routers/chat/routes.py | 276 ++++++++++++++++++++++++++ core/routers/file/routes.py | 4 +- core/routers/im/routes.py | 33 --- core/routers/pay/__init__.py | 6 + core/routers/pay/routes.py | 109 ++++++++++ core/routers/pay/schema.py | 28 +++ core/routers/user/schema.py | 2 +- core/utils/helper.py | 16 ++ core/wechat/api.py | 135 +++++++++---- main.py | 4 +- requirements.txt | 6 +- 24 files changed, 806 insertions(+), 184 deletions(-) rename .env.dev => .env.example (85%) create mode 100644 associated_script.py rename core/routers/{im => chat}/__init__.py (100%) create mode 100644 core/routers/chat/routes.py delete mode 100644 core/routers/im/routes.py create mode 100644 core/routers/pay/__init__.py create mode 100644 core/routers/pay/routes.py create mode 100644 core/routers/pay/schema.py diff --git a/.env.dev b/.env.example similarity index 85% rename from .env.dev rename to .env.example index 1060866..c814d6c 100644 --- a/.env.dev +++ b/.env.example @@ -24,4 +24,7 @@ OVERSEAS_PROXY = http://127.0.0.1:7890 # 微信 APPID = wx3d30771f3987cc92 -APP_SECRET = 8e57b1b5344d1dbd058cd45f3b2c4d3c \ No newline at end of file +APP_SECRET = 8e57b1b5344d1dbd058cd45f3b2c4d3c +PAY_MCH_ID = +PAY_API_KEY = +PAY_NOTIFY_URL = \ No newline at end of file diff --git a/associated_script.py b/associated_script.py new file mode 100644 index 0000000..fb811b5 --- /dev/null +++ b/associated_script.py @@ -0,0 +1,6 @@ +# -*- coding = utf-8 -*- +# @Time : 2025/3/8 下午11:23 +# @File : associated_script.py +# @Software : PyCharm +# @Author : xingc +# @Desc : 协同清理脚本 diff --git a/config.py b/config.py index 40715d2..657c184 100644 --- a/config.py +++ b/config.py @@ -62,6 +62,9 @@ class Settings(BaseSettings): # 微信 APPID: str = '' APP_SECRET: str = '' + PAY_MCH_ID: str = '' + PAY_API_KEY: str = '' + PAY_NOTIFY_URL: str = '' model_config = SettingsConfigDict(env_file=f'.env.{ENVIRONMENT}', env_file_encoding='utf-8') @@ -75,6 +78,8 @@ class ProductionSettings(Settings): ACCESS_SECRET_KEY: str = '1f790219840ccd10b5d8b33d7f595b7fe0eed262403689f6fa1cab80149b00f2' ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 1 # 1days + model_config = SettingsConfigDict(env_file=f'.env.{ENVIRONMENT}', env_file_encoding='utf-8') + class DevelopmentSettings(Settings): # 日志配置 diff --git a/core/app.py b/core/app.py index 102304e..5605239 100644 --- a/core/app.py +++ b/core/app.py @@ -6,10 +6,16 @@ # @Desc : import logging -from fastapi import FastAPI, Request, Response +import redis.asyncio as aioredis +from fastapi import FastAPI, Request, Response, status +from fastapi.responses import UJSONResponse +from fastapi.exceptions import HTTPException, RequestValidationError +from fastapi_limiter import FastAPILimiter from config import settings -from core.routers import routers +from core.db.engine import get_cache +from core.routers.route import routers +from core.routers.chat.handlers import socket_manager logger = logging.getLogger(__name__) @@ -29,7 +35,7 @@ async def exception_log(request: Request, exc): def create_app() -> FastAPI: app = FastAPI( debug=settings.DEBUG, - openapi_url='/openapi.json' if settings.DEBUG else None, + openapi_url='/openapi.json' if settings.DEBUG else None ) for router in routers: @@ -41,15 +47,49 @@ def create_app() -> FastAPI: @app.on_event('startup') async def startup_event(): - logger.info(f'Application <<{settings.NAME}>> is started') + redis_conn = await get_cache() + await FastAPILimiter.init(redis_conn) + logger.info(f'Application ``{settings.NAME}`` is started') @app.on_event('shutdown') async def shutdown_event(): - logger.info(f'Application <<{settings.NAME}>> is closed') + logger.info(f'Application ``{settings.NAME}`` is closed') + await socket_manager.pubsub_client.disconnect() + + @app.exception_handler(HTTPException) + async def http_exception_handler(request: Request, exc: HTTPException): + await exception_log(request, exc) + return UJSONResponse( + status_code=exc.status_code, + content={ + 'status': exc.status_code, + 'data': None, + 'message': exc.detail + } + ) + + @app.exception_handler(RequestValidationError) + async def http_exception_handler(request: Request, exc: RequestValidationError): + await exception_log(request, exc) + return UJSONResponse( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + content={ + 'status': status.HTTP_422_UNPROCESSABLE_ENTITY, + 'data': exc.errors(), + 'message': 'Validation error' + } + ) # @app.exception_handler(Exception) # async def http_exception_handler(request: Request, exc: Exception): # await exception_log(request, exc) - # return base_response.error(status_code=500, message='Server error') + # return UJSONResponse( + # status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + # content={ + # 'status': status.HTTP_500_INTERNAL_SERVER_ERROR, + # 'data': None, + # 'message': 'Internal server error' + # } + # ) return app diff --git a/core/base/exceptions.py b/core/base/exceptions.py index 26bf9c2..8033ee3 100644 --- a/core/base/exceptions.py +++ b/core/base/exceptions.py @@ -27,3 +27,8 @@ class SearchCardRetry(ScraperError): class WechatApiError(CardBookException): """微信开放服务调用api错误""" pass + + +class WebsocketTooManyRequests(CardBookException): + """websocket请求太多""" + pass diff --git a/core/base/schema.py b/core/base/schema.py index 92265cf..a16e8ef 100644 --- a/core/base/schema.py +++ b/core/base/schema.py @@ -47,9 +47,28 @@ class PlatformSourceType(enum.Enum): class MessageType(enum.Enum): TEXT = 'text' IMAGE = 'image' + READ = 'read' class CardCategoryType(enum.Enum): SPORTS = 'sports' ANIME = 'anime' OTHER = 'other' + + +class ProductCategory(enum.Enum): + VIP = 'vip' + POINTS = 'points' + + +class ProductAction(enum.Enum): + VIP_RENEWAL = 'vip_renewal' + RECHARGE_POINTS = 'recharge_points' + + +class OrderStatusType(enum.Enum): + INIT = 'init' + FAIL = 'fail' + OK = 'ok' + WAIT_PAYMENT = 'wait_payment' + CANCEL = 'cancel' diff --git a/core/db/crud.py b/core/db/crud.py index 72e9782..003a219 100644 --- a/core/db/crud.py +++ b/core/db/crud.py @@ -21,20 +21,24 @@ class CRUD: self.session = session self.model = model - async def get(self, filters: dict) -> Optional[Model]: + async def get(self, filters: dict, model: Model = None) -> Optional[Model]: + model = model or self.model async with self.session.begin(): - result = await self.session.execute(select(self.model).filter( - *self.build_filter_conditions(filters)) + result = await self.session.execute(select(model).filter( + *self.build_filter_conditions(filters, model)) ) record = result.scalar_one_or_none() return record - async def get_paginated(self, filters: dict, page: int = 1, page_size: int = 10) -> Tuple[List[Model], int]: + async def get_paginated( + self, filters: dict, page: int = 1, page_size: int = 10, model: Model = None + ) -> Tuple[List[Model], int]: + model = model or self.model async with self.session.begin(): filter_conditions = self.build_filter_conditions(filters) if filters else [] - query = select(self.model).filter(*filter_conditions).order_by(desc(self.model.created_at)) + query = select(model).filter(*filter_conditions).order_by(desc(model.created_at)) - total_query = select(func.count()).select_from(self.model).filter(*filter_conditions) + total_query = select(func.count()).select_from(model).filter(*filter_conditions) total_result = await self.session.execute(total_query) total = total_result.scalar() @@ -44,9 +48,12 @@ class CRUD: return records, total - async def get_grouped_count(self, group_by_field: str, filters: Dict[str, Any] = None) -> Dict[Any, int]: + async def get_grouped_count( + self, group_by_field: str, filters: Dict[str, Any] = None, model: Model = None + ) -> Dict[Any, int]: + model = model or self.model filter_conditions = self.build_filter_conditions(filters) if filters else [] - group_by = getattr(self.model, group_by_field) + group_by = getattr(model, group_by_field) query = select(group_by, func.count()).filter(*filter_conditions).group_by(group_by) async with self.session.begin(): @@ -55,32 +62,35 @@ class CRUD: return {record[0]: record[1] for record in records} - async def create(self, data: dict, filters: dict) -> Union[Model, bool]: + async def create(self, data: dict, filters: dict, model: Model = None) -> Union[Model, bool]: + model = model or self.model async with self.session.begin(): - result = await self.session.execute(select(self.model).filter( - *self.build_filter_conditions(filters)) + result = await self.session.execute(select(model).filter( + *self.build_filter_conditions(filters, model)) ) if result.scalars().first() is not None: return False - record = self.model(**data) + record = model(**data) self.session.add(record) await self.session.commit() return record - async def get_all(self, filters: dict = None) -> List[Model]: + async def get_all(self, filters: dict = None, model: Model = None) -> List[Model]: + model = model or self.model async with self.session.begin(): result = await self.session.execute( - select(self.model) if filters is None else select(self.model).filter( - *self.build_filter_conditions(filters)).order_by(desc(self.model.insert_time)) + select(model) if filters is None else select(model).filter( + *self.build_filter_conditions(filters, model)).order_by(desc(model.insert_time)) ) records = result.scalars().all() return records - async def update(self, data: dict, filters: dict) -> Optional[Model]: + async def update(self, data: dict, filters: dict, model: Model = None) -> Optional[Model]: + model = model or self.model async with self.session.begin(): - result = await self.session.execute(select(self.model).filter( - *self.build_filter_conditions(filters)) + result = await self.session.execute(select(model).filter( + *self.build_filter_conditions(filters, model)) ) record = result.scalars().first() @@ -101,10 +111,11 @@ class CRUD: await self.session.commit() return record - async def updates(self, data: dict, filters: dict) -> Optional[Model]: + async def updates(self, data: dict, filters: dict, model: Model = None) -> Optional[Model]: + model = model or self.model async with self.session.begin(): - result = await self.session.execute(select(self.model).filter( - *self.build_filter_conditions(filters)) + result = await self.session.execute(select(model).filter( + *self.build_filter_conditions(filters, model)) ) records = result.scalars().all() @@ -125,10 +136,11 @@ class CRUD: await self.session.commit() return records - async def delete(self, filters: dict) -> bool: + async def delete(self, filters: dict, model: Model = None) -> bool: + model = model or self.model async with self.session.begin(): - result = await self.session.execute(select(self.model).filter( - *self.build_filter_conditions(filters)) + result = await self.session.execute(select(model).filter( + *self.build_filter_conditions(filters, model)) ) record = result.scalars().first() @@ -139,7 +151,7 @@ class CRUD: await self.session.commit() return True - def build_filter_conditions(self, filters: Dict[str, Any]) -> tuple: + def build_filter_conditions(self, filters: Dict[str, Any], model: Model) -> tuple: conditions = [] for key, value in filters.items(): @@ -158,7 +170,7 @@ class CRUD: else: field_name = key - field = getattr(self.model, field_name) + field = getattr(model, field_name) if operator == 'like': # 模糊匹配 conditions.append(field.like(f'%{value}%')) diff --git a/core/db/models.py b/core/db/models.py index 5b29de9..b98e106 100644 --- a/core/db/models.py +++ b/core/db/models.py @@ -5,15 +5,15 @@ # @Author : xingc # @Desc : import enum -import random import uuid from datetime import datetime, date +from typing import List import ujson from sqlalchemy import ( Column, Integer, String, Text, Date, ForeignKey, DECIMAL, CheckConstraint, - UniqueConstraint, Index, Sequence, JSON, UUID, Enum, Boolean + UniqueConstraint, Index, Sequence, JSON, Enum, Boolean, UUID, Table ) from sqlalchemy.orm import ( DeclarativeBase, Mapped, mapped_column, relationship @@ -57,10 +57,11 @@ class DateModel: class User(BaseModel, DateModel): __tablename__ = 'system_users' __table_args__ = ( - Index('wechat_openid', 'wechat_openid'), + Index('idx_wechat_openid', 'wechat_openid'), UniqueConstraint('user_id', name='uniq_user_id'), {'comment': '用户表'} ) + id = None user_id: Mapped[int] = mapped_column( Integer, Sequence('non_primary_id_seq', start=100000, increment=1), primary_key=True, autoincrement=True, @@ -75,22 +76,28 @@ class User(BaseModel, DateModel): vip_expire_date: Mapped[date] = mapped_column(Date, default=datetime.now().date, comment='会员到期时间') points: Mapped[int] = mapped_column(Integer, default=0, comment='会员积分') - collections: Mapped[list["UserCollection"]] = relationship( - back_populates="user", - cascade="all, delete-orphan" + collections: Mapped[list['UserCollection']] = relationship( + back_populates='user', + cascade='all, delete-orphan' ) # 关注者 - following: Mapped[list["UserFollows"]] = relationship( + following: Mapped[list['UserFollows']] = relationship( 'UserFollows', foreign_keys='UserFollows.follower_id', - cascade="all, delete-orphan" + cascade='all, delete-orphan' ) # 定义被关注关系 - followers: Mapped[list["UserFollows"]] = relationship( + followers: Mapped[list['UserFollows']] = relationship( 'UserFollows', foreign_keys='UserFollows.followed_id', - cascade="all, delete-orphan" + cascade='all, delete-orphan' ) + chats: Mapped[List['Chat']] = relationship(secondary='chat_participant', back_populates='users') + messages: Mapped[List['Message']] = relationship(back_populates='user') + read_statuses: Mapped[List['ReadStatus']] = relationship(back_populates='user') + + def __str__(self): + return f'{self.user_id}' class UserFollows(BaseModel, DateModel): @@ -100,24 +107,23 @@ class UserFollows(BaseModel, DateModel): ) follower_id: Mapped[int] = mapped_column( - Integer, ForeignKey('system_users.user_id', ondelete="CASCADE"), comment='关注者的用户id' + Integer, ForeignKey('system_users.user_id', ondelete='CASCADE'), comment='关注者的用户id' ) followed_id: Mapped[int] = mapped_column( - Integer, ForeignKey('system_users.user_id', ondelete="CASCADE"), comment='被关注者的用户id' + Integer, ForeignKey('system_users.user_id', ondelete='CASCADE'), comment='被关注者的用户id' ) - updated_at = None follower = relationship('User', foreign_keys=[follower_id], back_populates='following') followed = relationship('User', foreign_keys=[followed_id], back_populates='followers') def __repr__(self): - return f"" + return f'' class Card(BaseModel, DateModel): __tablename__ = 'system_cards' __table_args__ = ( - Index('cert_number_category_source', 'cert_number', 'category', 'source'), + Index('idx_cert_number_category_source', 'cert_number', 'category', 'source'), {'comment': '卡片表'} ) @@ -129,28 +135,28 @@ class Card(BaseModel, DateModel): card_name: Mapped[str] = mapped_column(String(225), nullable=True, comment='卡片名称') card_number: Mapped[str] = mapped_column(String(225), nullable=True, comment='卡片编号') card_type: Mapped[str] = mapped_column(String(20), nullable=True, comment='卡片类型') - card_picture: Mapped[dict] = mapped_column(JSON, default=dict, comment='卡片正、反图片（本地）') - card_spare_picture: Mapped[dict] = mapped_column(JSON, default=dict, comment='卡片正、反图片（远程）') + card_picture: Mapped[dict] = mapped_column(JSON, default=lambda: dict(), comment='卡片正、反图片（本地）') + card_spare_picture: Mapped[dict] = mapped_column(JSON, default=lambda: dict(), comment='卡片正、反图片（远程）') create_user_id: Mapped[int] = mapped_column(Integer, nullable=True, comment='创建者') perfect_user_id: Mapped[int] = mapped_column(Integer, nullable=True, comment='完善者') category: Mapped[str] = mapped_column( Enum(base_schema.CardCategoryType, inherit_schema=True), - nullable=True, default=base_schema.CardCategoryType.OTHER, comment='卡片分类' + nullable=True, server_default=base_schema.CardCategoryType.OTHER.value, comment='卡片分类' ) source: Mapped[str] = mapped_column( Enum(base_schema.PlatformSourceType, inherit_schema=True), comment='来源，PSA、BGS、CGC、GBTC、CCG、BCTC ：采集 、OTHER：自定义' ) - collections: Mapped[list["UserCollection"]] = relationship( - back_populates="card", - cascade="all, delete-orphan" + collections: Mapped[list['UserCollection']] = relationship( + back_populates='card', + cascade='all, delete-orphan' ) class UserCollection(BaseModel, DateModel): - __tablename__ = "system_user_collections" + __tablename__ = 'system_user_collections' __table_args__ = ( UniqueConstraint('user_id', 'card_id', name='uniq_user_card'), CheckConstraint('purchase_price >= 0', name='check_purchase_price'), @@ -175,16 +181,15 @@ class UserCollection(BaseModel, DateModel): displayed: Mapped[bool] = mapped_column( Boolean, default=True ) - updated_at = None - user: Mapped["User"] = relationship(back_populates='collections') - card: Mapped["Card"] = relationship(back_populates='collections') + user: Mapped['User'] = relationship(back_populates='collections') + card: Mapped['Card'] = relationship(back_populates='collections') class SearchCardCache(BaseModel, DateModel): __tablename__ = 'system_cache_cards' __table_args__ = ( - Index('cert_and_card_number_source', 'cert_number', 'source'), + Index('idx_cert_and_card_number_source', 'cert_number', 'source'), {'comment': '搜索卡片信息缓存表'} ) @@ -196,38 +201,105 @@ class SearchCardCache(BaseModel, DateModel): comment='数据来源：PSA、BGS、CGC、GBTC、CCG、BCTC' ) -# class Product(BaseModel, DateModel): -# __tablename__ = 'system_products' -# __table_args__ = {'comment': '产品信息表'} -# -# name: Mapped[str] = mapped_column(String(50), comment='产品名') -# price: Mapped[float] = mapped_column(DECIMAL(10, 2), comment='价格') -# action: Mapped[str] = mapped_column(String(10), comment='付款后的动作，会员续期：vip_renewal 积分：recharge_points') -# number: Mapped[int] = mapped_column(Integer, comment='执行动作的增量值') -# description: Mapped[str] = mapped_column(Text, comment='商品描述') -# -# -# class Order(BaseModel, DateModel): -# __tablename__ = 'system_orders' -# __table_args__ = {'comment': '产品信息表'} -# -# order_id: Mapped[int] = mapped_column(String(20), comment='订单id') -# user_id: Mapped[int] = mapped_column(Integer, comment='用户id') -# product_id: Mapped[int] = mapped_column(Integer, comment='产品id') -# price: Mapped[float] = mapped_column(DECIMAL(10, 2), comment='价格') -# pay_url: Mapped[str] = mapped_column(String(255), comment='支付地址') -# status: Mapped[int] = mapped_column(Integer, comment='订单状态，-1：初始化 0：失败 1：完成 2：待支付') -# -# class Message(BaseModel, DateModel): -# __tablename__ = 'system_messages' -# __table_args__ = {'comment': '聊天消息表'} -# -# guid: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), default=uuid.uuid4) -# message_type: Mapped[str] = mapped_column( -# Enum(base_schema.MessageType, inherit_schema=True), default=base_schema.MessageType.TEXT -# ) -# content: Mapped[str] = mapped_column(String(5000), comment='文本消息') -# send_timestamp: Mapped[int] = mapped_column(Integer, comment='发送时间') -# -# user: Mapped["User"] = relationship(back_populates="system_messages") +class Product(BaseModel, DateModel): + __tablename__ = 'system_products' + __table_args__ = {'comment': '产品信息表'} + + name: Mapped[str] = mapped_column(String(20), comment='产品名') + category: Mapped[str] = mapped_column( + Enum(base_schema.ProductCategory, inherit_schema=True), comment='商品分类, vip-会员服务 points-积分服务' + ) + price: Mapped[int] = mapped_column(Integer, comment='价格，不带小数点 单位为分') + action: Mapped[str] = mapped_column(String(10), comment='付款后的动作，会员续期：vip_renewal 积分：recharge_points') + number: Mapped[int] = mapped_column(Integer, comment='执行动作的增量值') + description: Mapped[str] = mapped_column(Text, nullable=True, comment='商品描述') + + +class Order(BaseModel, DateModel): + __tablename__ = 'system_orders' + __table_args__ = ( + Index('idx_order_on_order_id', 'order_id'), + {'comment': '产品订单表'} + ) + + order_id: Mapped[str] = mapped_column(String(32), comment='订单id') + transaction_id: Mapped[str] = mapped_column(String(32), nullable=True, comment='微信订单id') + user_id: Mapped[int] = mapped_column(Integer, comment='用户id') + product_id: Mapped[int] = mapped_column(Integer, comment='产品id') + total_fee: Mapped[int] = mapped_column(Integer, nullable=True, comment='发起的总金额，单位为分') + notify_total_fee: Mapped[int] = mapped_column(Integer, nullable=True, comment='回调成功支付的总金额，单位为分') + pay_url: Mapped[str] = mapped_column(String(255), nullable=True, comment='支付地址') + status: Mapped[str] = mapped_column( + Enum(base_schema.OrderStatusType, inherit_schema=True), + server_default=base_schema.OrderStatusType.INIT.value, + comment='订单状态，init：初始化 fail：失败 ok：完成 wait payment：待支付 cancel. 取消' + ) + + +class Chat(BaseModel, DateModel): + __tablename__ = 'system_chats' + __table_args__ = ( + Index('idx_chat_on_guid', 'guid'), + {'comment': '聊天表'} + ) + + guid: Mapped[uuid.UUID] = mapped_column(String(36), nullable=True, default=lambda: str(uuid.uuid4())) + + users: Mapped[List['User']] = relationship(secondary='chat_participant', back_populates='chats') + messages: Mapped[List['Message']] = relationship(back_populates='chat', cascade='all,delete') + is_deleted: Mapped[bool] = mapped_column(Boolean, default=False) + read_statuses: Mapped[List['ReadStatus']] = relationship(back_populates='chat', cascade='all,delete') + + +class Message(BaseModel, DateModel): + __tablename__ = 'system_messages' + __table_args__ = ( + Index('idx_message_on_user_id', 'user_id'), + Index('idx_message_on_chat_id', 'chat_id'), + Index('idx_message_on_user_id_chat_id', 'chat_id', 'user_id'), + {'comment': '聊天消息表'} + ) + + guid: Mapped[uuid.UUID] = mapped_column(String(36), nullable=True, default=lambda: str(uuid.uuid4())) + user_id: Mapped[int] = mapped_column(ForeignKey('system_users.user_id')) + chat_id: Mapped[int] = mapped_column(ForeignKey('system_chats.id')) + + message_type: Mapped[str] = mapped_column( + Enum(base_schema.MessageType, inherit_schema=True), default=base_schema.MessageType.TEXT + ) + content: Mapped[str] = mapped_column(String(5000), nullable=True, comment='文本消息') + + chat: Mapped['Chat'] = relationship(back_populates='messages') + user: Mapped['User'] = relationship(back_populates='messages') + + def __str__(self): + return f'{self.content}' + + +class ReadStatus(BaseModel, DateModel): + __tablename__ = 'system_read_status' + __table_args__ = ( + Index('idx_read_status_on_chat_id', 'chat_id'), + Index('idx_read_status_on_user_id', 'user_id'), + {'comment': '聊天消息读取状态表'} + ) + + last_read_message_id: Mapped[int] = mapped_column(nullable=True) + user_id: Mapped[int] = mapped_column(ForeignKey('system_users.user_id')) + chat_id: Mapped[int] = mapped_column(ForeignKey('system_chats.id')) + + chat: Mapped['Chat'] = relationship(back_populates='read_statuses') + user: Mapped['User'] = relationship(back_populates='read_statuses') + + def __str__(self): + return f'User: {self.user_id}, Message: {self.last_read_message_id}' + + +chat_participant = Table( + 'chat_participant', + BaseModel.metadata, + Column('user_id', Integer, ForeignKey('system_users.user_id'), primary_key=True), + Column('chat_id', Integer, ForeignKey('system_chats.id'), primary_key=True), + comment='聊天参与者关联表' +) \ No newline at end of file diff --git a/core/routers/__init__.py b/core/routers/__init__.py index dc52581..0519c84 100644 --- a/core/routers/__init__.py +++ b/core/routers/__init__.py @@ -3,11 +3,4 @@ # @File : __init__.py.py # @Software : PyCharm # @Author : xingc -# @Desc : -from .auth.routes import auth_router, test_auth_router -from .card.routes import card_router -from .im.routes import im_router -from .file.routes import file_router -from .user.routes import user_router - -routers = [auth_router, test_auth_router, card_router, im_router, file_router, user_router] +# @Desc : \ No newline at end of file diff --git a/core/routers/auth/routes.py b/core/routers/auth/routes.py index 42cc141..274e821 100644 --- a/core/routers/auth/routes.py +++ b/core/routers/auth/routes.py @@ -91,8 +91,9 @@ async def wechat_login( session_key = wechat_session.get('session_key') union_id = wechat_session.get('union_id') try: - profile = await decrypt_sensitive_data(session_key, payload.encrypted_data, payload.iv) + profile = decrypt_sensitive_data(session_key, payload.encrypted_data, payload.iv) except (ValueError, TypeError) as e: + logger.exception(f"[wechat_login] Exception, User information decryption failed, detail: {e}") raise HTTPException(status.HTTP_400_BAD_REQUEST, detail='请重试登录操作') logger.info(f"decrypt plaintext: {wechat_session}") diff --git a/core/routers/card/routes.py b/core/routers/card/routes.py index 4ebd939..cedf939 100644 --- a/core/routers/card/routes.py +++ b/core/routers/card/routes.py @@ -4,7 +4,6 @@ # @Software : PyCharm # @Author : xingc # @Desc : -import re import logging from typing import List, Literal diff --git a/core/routers/card/schema.py b/core/routers/card/schema.py index f99460c..31dea82 100644 --- a/core/routers/card/schema.py +++ b/core/routers/card/schema.py @@ -12,7 +12,7 @@ from pydantic import BaseModel, constr, condecimal, field_validator from core.base.schema import SupportPlatformType, PlatformSourceType, CardCategoryType -class CardPrice(BaseModel): +class CardPicture(BaseModel): front_picture: Optional[str] = None reverse_picture: Optional[str] = None @@ -29,13 +29,13 @@ class Card(BaseModel): cert_number: str card_year: Optional[constr(pattern=r'^\d{4}([-~]\d{1,4})?$')] card_brand: Optional[str] = None - card_score: Optional[float] = None - auto_score: Optional[float] = None + card_score: Optional[condecimal(max_digits=10, decimal_places=1)] = None + auto_score: Optional[condecimal(max_digits=10, decimal_places=1)] = None card_name: str card_number: Optional[str] = None card_type: Optional[str] = None - card_picture: Optional[CardPrice] = {} - card_spare_picture: Optional[dict] = {} + card_picture: Optional[CardPicture] = CardPicture() + card_spare_picture: Optional[dict] = CardPicture() source: SupportPlatformType diff --git a/core/routers/im/__init__.py b/core/routers/chat/__init__.py similarity index 100% rename from core/routers/im/__init__.py rename to core/routers/chat/__init__.py diff --git a/core/routers/chat/routes.py b/core/routers/chat/routes.py new file mode 100644 index 0000000..3cf7a18 --- /dev/null +++ b/core/routers/chat/routes.py @@ -0,0 +1,276 @@ +# -*- coding = utf-8 -*- +# @Time : 2025/2/13 下午4:53 +# @File : routes.py +# @Software : PyCharm +# @Author : xingc +# @Desc : +import logging +from typing import List, Dict +from json.decoder import JSONDecodeError + +from fastapi import APIRouter, Query, Depends, WebSocket, WebSocketDisconnect, HTTPException, status +from fastapi_limiter.depends import WebSocketRateLimiter +import redis.asyncio as aioredis +from sqlalchemy import select, func, and_, exists +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import selectinload + +from core.base import schema as base_schema +from core.db.crud import CRUD +from core.db.engine import get_async_session, get_cache +from core.db.models import Message, Chat, User, ReadStatus, chat_participant +from core.routers.auth.schema import LoginUser +from core.routers.auth.services import get_current_user +from core.base.exceptions import WebsocketTooManyRequests + +from . import schema +from .handlers import socket_manager +from .services import websocket_callback, get_user_active_direct_chats, get_unread_messages_per_chat + +logger = logging.getLogger(__name__) + +chat_router = APIRouter(prefix='/chat', tags=['聊天模块']) + + +@chat_router.get( + '/all/', summary='获取所有聊天', + response_model=base_schema.BaseResponse[ + base_schema.Paginated[ + List[schema.ChatUnreadOut] + ] + ] +) +async def get_all_chats( + login_user: LoginUser = Depends(get_current_user), + db_session: AsyncSession = Depends(get_async_session), + page: int = Query(default=1, gt=0), + page_size: int = Query(default=15, gt=0, le=200), +): + filter_conditions = ( + exists().where( + (chat_participant.c.chat_id == Chat.id) & + (chat_participant.c.user_id == login_user.user_id) + ) & + Chat.is_deleted.is_(False) + ) + query = ( + select(Chat) + .where(filter_conditions).order_by(Chat.updated_at.desc()) + ) + + total_query = select(func.count()).select_from(Chat).filter(filter_conditions) + total_result = await db_session.execute(total_query) + total = total_result.scalar() + + query = query.limit(page_size).offset((page - 1) * page_size) + result = await db_session.execute(query) + chats = result.scalars().all() + + unread_messages_chats_count = await get_unread_messages_per_chat(db_session, chats, login_user) + + return { + 'data': { + 'list': unread_messages_chats_count, + 'total': total, + 'page': page, + 'page_size': page_size, + } + } + + +@chat_router.get( + '/{chat_guid}/messages/', summary='获取单个聊天的历史消息', + response_model=base_schema.BaseResponse[ + base_schema.Paginated[ + List[Dict] + ] + ] +) +async def get_single_chat_messages( + chat_guid: str, + login_user: LoginUser = Depends(get_current_user), + db_session: AsyncSession = Depends(get_async_session), + page: int = Query(default=1, gt=0), + page_size: int = Query(default=15, gt=0, le=200), + last_read_message_id: int = Query(default=0, ge=0), +): + db_crud = CRUD(db_session, Chat) + chat: Chat | None = await db_crud.get(filters={'guid': chat_guid, 'is_deleted': False}) + if not chat: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='您访问的聊天不存在') + + chat_user_ids = [user.user_id for user in await chat.awaitable_attrs.users] + if login_user.user_id not in chat_user_ids: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail='您无权访问此聊天') + + # 组装查询条件 + conditions = and_( + Message.chat_id == chat.id, + Message.id > last_read_message_id + ) + + # 记录总数查询 + total_stmt = ( + select(func.count()).where( + conditions + ) + ) + total = await db_session.execute(total_stmt) + # 分页记录查询 + stmt = ( + select(Message) + .where(conditions) + .order_by(Message.created_at.desc()) + .limit(page_size).offset((page - 1) * page_size) + .options(selectinload(Message.user), selectinload(Message.chat)) + ) + records = await db_session.execute(stmt) + + return { + 'data': { + 'list': [ + { + **card.dict(), + **user_collection.dict() + } for card, user_collection in records + ], + 'total': total.scalar(), + 'page': page, + 'page_size': page_size, + }, + } + + +@chat_router.post('/create', summary='创建新聊天', response_model=base_schema.BaseResponse[schema.Chat]) +async def create_new_chat( + payload: schema.ChatCreateIn, + login_user: LoginUser = Depends(get_current_user), + db_session: AsyncSession = Depends(get_async_session), +): + if payload.recipient_user_id == login_user.user_id: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail='操作禁止') + + recipient_user: User | None = await CRUD(db_session, User).get(filters={'user_id': payload.recipient_user_id}) + if recipient_user is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='收件人不存在') + + initiator_user: User = await CRUD(db_session, User).get(filters={'user_id': login_user.user_id}) + # 查询是否已存在 对应且未关闭的聊天 + query = ( + select(Chat) + .where( + exists().where( + (chat_participant.c.chat_id == Chat.id) & + (chat_participant.c.user_id == login_user.user_id) + ) & + exists().where( + (chat_participant.c.chat_id == Chat.id) & + (chat_participant.c.user_id == recipient_user.user_id) + ) & + Chat.is_deleted == False + ).order_by(Chat.created_at.desc()) + ) + result = await db_session.execute(query) + chat = result.scalar_one_or_none() + + if chat is None: + try: + chat = Chat() + chat.users.append(initiator_user) + chat.users.append(recipient_user) + db_session.add(chat) + await db_session.flush() + + initiator_read_status = ReadStatus(chat_id=chat.id, user_id=initiator_user.user_id, last_read_message_id=0) + recipient_read_status = ReadStatus(chat_id=chat.id, user_id=recipient_user.user_id, last_read_message_id=0) + db_session.add_all([initiator_read_status, recipient_read_status]) + await db_session.commit() + + except Exception as exc_info: + await db_session.rollback() + raise exc_info + + return { + 'data': { + 'chat_guid': chat.guid, + 'friend_user': recipient_user, + } + } + + +@chat_router.delete('/{chat_guid}/', summary='删除聊天', response_model=base_schema.BaseResponse) +async def delete_single_chat( + chat_guid: str, + login_user: LoginUser = Depends(get_current_user), + db_session: AsyncSession = Depends(get_async_session), +): + chat = await CRUD(db_session, Chat).update( + data={'is_deleted': True}, + filters={'chat_guid': chat_guid, 'user_id': login_user.user_id} + ) + if chat is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail='聊天不存在') + + return {'message': '聊天已删除'} + + +@chat_router.websocket('/ws') +async def get_chat_websocket( + websocket: WebSocket, + login_user: LoginUser = Depends(get_current_user), + db_session: AsyncSession = Depends(get_async_session), + redis_conn: aioredis.Redis = Depends(get_cache) +): + current_user = await CRUD(db_session, User).get(filters={'user_id': login_user.user_id}) + await socket_manager.connect_socket(websocket=websocket) + logger.info('Websocket connection is established') + ratelimit = WebSocketRateLimiter(times=50, seconds=10, callback=websocket_callback) + user_id = str(login_user.user_id) + await socket_manager.add_user_socket_connection(user_id, websocket) + + if chats := await get_user_active_direct_chats(db_session, current_user=current_user): + for chat_guid in chats: + await socket_manager.add_user_to_chat(chat_guid, websocket) + else: + chats = dict() + + try: + while True: + try: + incoming_message = await websocket.receive_json() + await ratelimit(websocket) + + try: + message = schema.Message(**incoming_message) + except ValueError: + await socket_manager.send_error(websocket, 'The message type or format you sent is invalid') + continue + + handler = socket_manager.handlers.get(message.type) + if not handler: + logger.error(f'No handler [{message.type}] exists') + await socket_manager.send_error(websocket, f'Type: {message.type} was not found') + continue + + await handler( + websocket=websocket, + db_session=db_session, + cache=redis_conn, + current_user=current_user, + chats=chats, + message=message, + ) + + except (JSONDecodeError, AttributeError) as exc: + logger.exception(f'Websocket error, detail: {exc}') + await websocket.send_json({'status': 'error', 'message': 'Wrong message format'}) + + except ValueError as exc: + logger.exception(f'Websocket error, detail: {exc}') + await websocket.send_json({'status': 'error', 'message': 'Could not validate incoming message'}) + + except WebsocketTooManyRequests: + logger.exception(f'User: {login_user} sent too many ws requests') + await websocket.send_json({'status': 'error', 'message': 'You have sent too many requests'}) + except WebSocketDisconnect: + logging.info('Websocket is disconnected') diff --git a/core/routers/file/routes.py b/core/routers/file/routes.py index b00ad53..684730d 100644 --- a/core/routers/file/routes.py +++ b/core/routers/file/routes.py @@ -70,9 +70,9 @@ async def upload_file( return {'data': file_info} except Exception as e: - logger.error(f'Upload failed: {str(e)}') + logger.error(f'Upload failed: {e}') logger.exception(e) raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, - detail=f'File upload failed' + detail='Internal server error' ) diff --git a/core/routers/im/routes.py b/core/routers/im/routes.py deleted file mode 100644 index 9023cda..0000000 --- a/core/routers/im/routes.py +++ /dev/null @@ -1,33 +0,0 @@ -# -*- coding = utf-8 -*- -# @Time : 2025/2/13 下午4:53 -# @File : routes.py -# @Software : PyCharm -# @Author : xingc -# @Desc : -from typing import Annotated - -from fastapi import APIRouter, Query - -im_router = APIRouter(prefix='/im', tags=['聊天模块']) - - -# @im_router.get('/chat/{user_id}/messages/') -# def get_single_chat_messages( -# user_id: int, -# limit: Annotated[int, Query(ge=1, le=200)] = 20, -# cursor: int = -1, -# ): -# pass -# -# -# @im_router.get('/chats/messages/') -# def get_chats_messages( -# limit: Annotated[int, Query(ge=1, le=200)] = 100, -# cursor: int = -1, -# ): -# pass -# -# -# @im_router.websocket('/chat') -# def get_chat(): -# pass diff --git a/core/routers/pay/__init__.py b/core/routers/pay/__init__.py new file mode 100644 index 0000000..d56e1ba --- /dev/null +++ b/core/routers/pay/__init__.py @@ -0,0 +1,6 @@ +# -*- coding = utf-8 -*- +# @Time : 2025/2/24 上午9:03 +# @File : __init__.py.py +# @Software : PyCharm +# @Author : xingc +# @Desc : diff --git a/core/routers/pay/routes.py b/core/routers/pay/routes.py new file mode 100644 index 0000000..3e7cfe9 --- /dev/null +++ b/core/routers/pay/routes.py @@ -0,0 +1,109 @@ +# -*- coding = utf-8 -*- +# @Time : 2025/2/24 上午9:03 +# @File : routes.py +# @Software : PyCharm +# @Author : xingc +# @Desc : +import logging + +import xmltodict +from fastapi import APIRouter, Request, HTTPException, status, Depends, BackgroundTasks +from sqlalchemy.ext.asyncio import AsyncSession + +from core.base import schema as base_schema +from core.db.crud import CRUD +from core.db.engine import get_async_session +from core.db.models import Order +from core.routers.auth.schema import LoginUser +from core.routers.auth.services import get_current_user +from core.utils.helper import generate_keyname, generate_order_number +from core.wechat.api import wechat_pay_api + +from . import schema +from .actions import actions_after_payment + +logger = logging.getLogger(__name__) +pay_router = APIRouter(prefix='/pay', tags=['支付模块']) + + +@pay_router.post( + '/order/create', + summary='订单创建', + response_model=base_schema.BaseResponse[schema.CreateOrderOut] +) +async def create_order( + request: Request, + payload: schema.CreateOrderIn, + login_user: LoginUser = Depends(get_current_user), + db_session: AsyncSession = Depends(get_async_session), +): + client_ip = ( + request.headers.get('X-Real-IP') or + request.headers.get('x-real-ip') or + request.client.host + ) + out_trade_no = generate_order_number() + prepay_id = wechat_pay_api.create_unified_order( + **payload.dict(exclude_none=True), + out_trade_no=out_trade_no, + spbill_create_ip=client_ip + ) + payment_params = wechat_pay_api.generate_payment_params(prepay_id) + db_crud = CRUD(db_session, Order) + order: Order = await db_crud.create( + data={ + 'order_id': out_trade_no, + 'user_id': login_user.user_id, + 'product_id': payload.product_id, + 'total_fee': payload.total_fee, + 'status': base_schema.OrderStatusType.WAIT_PAYMENT + }, + filters={ + 'out_trade_no': out_trade_no, + } + ) + return {'data': {'payment_params': payment_params, 'order_id': order.order_id}} + + +@pay_router.post('/wechat/notify', summary='处理支付结果通知', response_model=base_schema.BaseResponse) +async def wechat_notify( + request: Request, + background_tasks: BackgroundTasks, + db_session: AsyncSession = Depends(get_async_session) +): + body = await request.body() + logger.debug(body.decode()) + result = xmltodict.parse(body.decode()) + + # 验证签名 + params = result['xml'] + sign = params.pop('sign') + local_sign = wechat_pay_api.generate_sign(params) + if sign != local_sign: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail='签名验证失败' + ) + + # 处理支付结果 + db_crud = CRUD(db_session, Order) + record = { + 'out_trade_no': params['out_trade_no'], + 'transaction_id': params['transaction_id'] + } + if params['return_code'] == 'SUCCESS' and params['result_code'] == 'SUCCESS': + record.update({ + 'notify_total_fee': params['total_fee'], + 'status': base_schema.OrderStatusType.OK + }) + result = {'code': 200, 'message': 'OK'} + background_tasks.add_task(actions_after_payment, notify_params=params) + else: + result = {'code': 400, 'message': '支付失败'} + record.update({ + 'status': base_schema.OrderStatusType.FAIL, + }) + + await db_crud.update(data=record, filters={'order_id': record['out_trade_no']}) + + return result diff --git a/core/routers/pay/schema.py b/core/routers/pay/schema.py new file mode 100644 index 0000000..27473f0 --- /dev/null +++ b/core/routers/pay/schema.py @@ -0,0 +1,28 @@ +# -*- coding = utf-8 -*- +# @Time : 2025/2/24 下午9:51 +# @File : schema.py +# @Software : PyCharm +# @Author : xingc +# @Desc : +from pydantic import BaseModel + + +class CreateOrderIn(BaseModel): + product_id: int + openid: str + total_fee: int + body: str + + +class PaymentParamsOut(BaseModel): + appId: str + timeStamp: str + nonceStr: str + package: str + package: str + signType: str + + +class CreateOrderOut(BaseModel): + payment_params: PaymentParamsOut + order_id: str diff --git a/core/routers/user/schema.py b/core/routers/user/schema.py index 98a14ab..a7a9e1c 100644 --- a/core/routers/user/schema.py +++ b/core/routers/user/schema.py @@ -33,7 +33,7 @@ class LiteUser(BaseModel): displayed_card_total: Optional[int] = 0 -class UserFollows(Schema): +class UserFollows(BaseModel): nickname: str user_id: int gender: Optional[int] = None diff --git a/core/utils/helper.py b/core/utils/helper.py index bc326e5..33b2356 100644 --- a/core/utils/helper.py +++ b/core/utils/helper.py @@ -6,6 +6,8 @@ # @Desc : import asyncio import functools +import random +import string from datetime import datetime from typing import Callable @@ -57,3 +59,17 @@ def check_picture_for_empty(picture: dict, or_: bool = False) -> bool: picture.get('reverse_picture') is None ] return any(results) if or_ else all(results) + + +def generate_order_number(): + random_part = ''.join(random.choices(string.digits, k=18)) + order_number = f"{datetime.now().strftime('%Y%m%d%H%M%S')}{random_part}" + + return order_number + + +def format_date(date: datetime | None) -> str | None: + if date is None: + return date + else: + return date.strftime('%Y-%m-%d %H:%M:%S') diff --git a/core/wechat/api.py b/core/wechat/api.py index c641efd..12c0f7b 100644 --- a/core/wechat/api.py +++ b/core/wechat/api.py @@ -11,8 +11,11 @@ import time import httpx import ujson +import xmltodict from Crypto.Cipher import AES from Crypto.Util.Padding import unpad +from fastapi import HTTPException +from simple_spider_tool import md5, jsonpath from config import settings from core.base.exceptions import WechatApiError @@ -22,22 +25,10 @@ from core.db.engine import get_cache logger = logging.getLogger(__name__) -class WechatApi: - def __init__(self, appid, secret): - self.appid = appid - self.secret = secret - self._access_token = None - self._access_token_expire_time = -1 +class BaseClient: + def __init__(self): self.session = httpx.AsyncClient() - async def access_token(self): - if any([ - self._access_token is None, - self._access_token_expire_time < math.floor(time.time()), - ]): - await self.get_access_token() - return self._access_token - async def request(self, method, url, headers=None, params=None, data=None, json=None, **kwargs): r = await self.session.request( method, url, headers=headers, params=params, data=data, json=json, **kwargs @@ -62,6 +53,23 @@ class WechatApi: 'POST', url, headers=headers, params=params, data=data, json=json, **kwargs ) + +class WechatApi(BaseClient): + def __init__(self, appid, secret): + self.appid = appid + self.secret = secret + self._access_token = None + self._access_token_expire_time = -1 + super().__init__() + + async def access_token(self): + if any([ + self._access_token is None, + self._access_token_expire_time < math.floor(time.time()), + ]): + await self.get_access_token() + return self._access_token + async def get_access_token(self): redis_conn = await get_cache() result_str = await redis_conn.get('access_token') @@ -113,39 +121,96 @@ class WechatApi: wechat_api = WechatApi(appid=settings.APPID, secret=settings.APP_SECRET) -async def decrypt_sensitive_data(session_key: str, encrypted_data: str, iv: str) -> dict: +def decrypt_sensitive_data(session_key: str, encrypted_data: str, iv: str) -> dict: """解密敏感数据""" session_key_bytes = base64.b64decode(session_key) iv_bytes = base64.b64decode(iv) encrypted_data_bytes = base64.b64decode(encrypted_data) cipher = AES.new(session_key_bytes, AES.MODE_CBC, iv_bytes) - # decrypted = cipher.decrypt(encrypted_data_bytes) - # pad = decrypted[-1] - # decrypted = decrypted[:-pad] decrypted = unpad(cipher.decrypt(encrypted_data_bytes), AES.block_size) return ujson.decode(decrypted.decode('utf-8')) -def decrypt_phone_number(encrypted_data: str, session_key: str, iv: str) -> dict: - try: - # Base64 解码 - session_key_bytes = base64.b64decode(session_key) - encrypted_bytes = base64.b64decode(encrypted_data) - iv_bytes = base64.b64decode(iv) +class WechatPayApi(BaseClient): + def __init__(self, appid, mch_id, app_key, notify_url, debug: bool = False): + """ + :param appid: 小程序id + :param mch_id: 商户id + :param app_key: 接口密钥 + :param debug: + """ + self.appid = appid + self.mch_id = mch_id + self.appi_key = app_key + self.notify_url = notify_url + self.base_url = 'https://api.mch.weixin.qq.com' + super().__init__() - # AES 解密 - cipher = AES.new(session_key_bytes, AES.MODE_CBC, iv_bytes) - decrypted = unpad(cipher.decrypt(encrypted_bytes), AES.block_size) - decrypted_data = ujson.loads(decrypted.decode('utf-8')) + @staticmethod + def generate_nonce_str() -> str: + """生成随机字符串""" + return md5(f'{time.time()}') - # 验证数据 - if decrypted_data.get("watermark", {}).get("appid") != "your-appid": - raise ValueError("Invalid appid in decrypted data") - return decrypted_data - except Exception as e: - logger.info(e) - # raise HTTPException(status_code=400, detail=f"解密失败: {str(e)}") + def generate_sign(self, params: dict) -> str: + """生成签名""" + plaintext = '&'.join([f'{k}={params[k]}' for k in sorted(params.keys())]) + return md5(f'{plaintext}&key={self.appi_key}').upper() + def generate_payment_params(self, prepay_id) -> dict: + """生成小程序支付参数""" + params = { + "appId": self.appid, + "timeStamp": f'{int(time.time())}', + "nonceStr": self.generate_nonce_str(), + "package": f"prepay_id={prepay_id}", + "signType": "MD5", + } + params["paySign"] = self.generate_sign(params) + return params + + async def create_unified_order( + self, + openid: str, + body: str, + out_trade_no: str, + total_fee: int, + create_ip: str, + ): + url = self.base_url + '/pay/unifiedorder' + params = { + 'appid': self.appid, + 'mch_id': self.mch_id, + 'nonce_str': self.generate_nonce_str(), + 'body': body, + 'out_trade_no': out_trade_no, + 'total_fee': total_fee, + 'spbill_create_ip': create_ip, + 'notify_url': self.notify_url, + 'trade_type': 'JSAPI', + 'openid': openid, + } + params['sign'] = self.generate_sign(params) + headers = { + 'Content-Type': 'application/xml', + } + data = xmltodict.unparse({'xml': params}, full_document=False) + r = await self.post(url, headers=headers, data=data) + if r.status_code != 200: + raise HTTPException(status_code=500, detail='微信支付接口请求失败') + + # 解析返回的 XML + result = xmltodict.parse(r.text) + return_code = jsonpath(result, '$.xml.return_code', first=True) + if return_code != 'SUCCESS': + return_msg = jsonpath(result, '$.xml.return_msg', first=True) + raise HTTPException(status_code=500, detail=return_msg) + + return result['xml']['prepay_id'] + + +wechat_pay_api = WechatPayApi( + appid=settings.APPID, mch_id=settings.PAY_MCH_ID, app_key=settings.PAY_API_KEY, notify_url=settings.PAY_NOTIFY_URL +) if __name__ == '__main__': import asyncio diff --git a/main.py b/main.py index 238c4ba..3ba1e97 100644 --- a/main.py +++ b/main.py @@ -4,12 +4,12 @@ # @Software : PyCharm # @Author : xingc # @Desc : -import logging +import logging.config from config import LOGGING_CONFIG from core.app import create_app -# logging.config.dictConfig(LOGGING_CONFIG) +logging.config.dictConfig(LOGGING_CONFIG) app = create_app() diff --git a/requirements.txt b/requirements.txt index 125406d..f7ae2de 100644 --- a/requirements.txt +++ b/requirements.txt @@ -20,9 +20,9 @@ SQLAlchemy[asyncio] SQLAlchemy[aiomysql] ujson uvicorn[standard] - - - +pycryptodome +xmltodict +fastapi-limiter