fix(card): 修复bgs卡片auto_score字段值Not Available导致响应出错

alembic/versions/96a29dfd123f_add_user_fields.py

@@ -0,0 +1,36 @@
+"""add user fields
+
+Revision ID: 96a29dfd123f
+Revises: 43b2d2efe0f7
+Create Date: 2025-10-15 19:57:02.236128
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '96a29dfd123f'
+down_revision: Union[str, None] = '43b2d2efe0f7'
+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('system_users', sa.Column('description', sa.String(length=200), nullable=False, comment='用户简介'))
+    op.add_column('system_users', sa.Column('fans_count', sa.Integer(), nullable=False, comment='粉丝数'))
+    op.add_column('system_users', sa.Column('followers_count', sa.Integer(), nullable=False, comment='关注数'))
+    op.add_column('system_users', sa.Column('last_active_at', sa.DateTime(), nullable=True, comment='最后活跃时间'))
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_column('system_users', 'last_active_at')
+    op.drop_column('system_users', 'followers_count')
+    op.drop_column('system_users', 'fans_count')
+    op.drop_column('system_users', 'description')
+    # ### end Alembic commands ###

associated_script.py

@@ -1,6 +0,0 @@
-# -*- coding = utf-8 -*-
-# @Time : 2025/3/8 下午11:23
-# @File : associated_script.py
-# @Software : PyCharm
-# @Author : xingc
-# @Desc : 协同清理脚本

core/base/schema.py

@@ -5,6 +5,7 @@
 # @Author : xingc
 # @Desc :
 import enum
+from datetime import date, datetime
 from typing import Optional, Generic, TypeVar
 
 from pydantic import BaseModel
@@ -12,6 +13,14 @@ from pydantic import BaseModel
 T = TypeVar('T')
 
 
+class BaseMixin(BaseModel):
+    class Config:
+        json_encoders = {
+            datetime: lambda dt: dt.strftime('%Y-%m-%d %H:%M:%S'),
+            date: lambda dt: dt.strftime('%Y-%m-%d')
+        }
+
+
 class BaseResponse(BaseModel, Generic[T]):
     status: int = 200
     message: str = 'success'

core/db/models.py

@@ -11,9 +11,9 @@ from typing import List
 
 import ujson
 from sqlalchemy import (
-    Column, Integer, String, Text, Date,
+    Integer, String, Text, Date,
     ForeignKey, DECIMAL, CheckConstraint,
-    UniqueConstraint, Index, Sequence, JSON, Enum, Boolean, UUID, Table
+    UniqueConstraint, Index, Sequence, JSON, Enum, Boolean
 )
 from sqlalchemy.orm import (
     DeclarativeBase, Mapped, mapped_column, relationship
@@ -80,6 +80,10 @@ class User(BaseModel, DateModel):
     phone: Mapped[str] = mapped_column(String(11), nullable=True, comment='手机号码')
     vip_expire_date: Mapped[date] = mapped_column(Date, default=set_default_vip_expire_date, comment='会员到期时间')
     points: Mapped[int] = mapped_column(Integer, default=0, comment='会员积分')
+    description: Mapped[str] = mapped_column(String(200), default='', comment='用户简介')
+    fans_count: Mapped[int] = mapped_column(Integer, default=0, comment='粉丝数')
+    followers_count: Mapped[int] = mapped_column(Integer, default=0, comment='关注数')
+    last_active_at: Mapped[datetime] = mapped_column(nullable=True, comment='最后活跃时间')
 
     collections: Mapped[list['UserCollection']] = relationship(
         back_populates='user',

core/routers/auth/routes.py

@@ -5,6 +5,7 @@
 # @Author : xingc
 # @Desc :
 import logging
+from datetime import datetime
 from typing import Annotated
 
 import ujson
@@ -21,7 +22,7 @@ from core.db.crud import CRUD
 from core.db.engine import get_async_session, get_cache
 from core.db.models import User
 from core.utils.helper import generate_keyname
-from core.wechat.api import wechat_api, decrypt_sensitive_data
+from core.wechat.api import wechat_api
 
 from . import schema
 from .services import create_access_token
@@ -96,10 +97,16 @@ async def wechat_login(
                 'wechat_union_id': union_id,
                 'phone': payload.phone,
                 'avatar_url': payload.avatar_url,
+                'last_active_at': datetime.now(),
             },
             filters={'wechat_openid': openid}
         )
         await redis_conn.delete(keyname)
+    else:
+        await db_crud.update(
+            data={'last_active_at': datetime.now()},
+            filters={'wechat_openid': openid}
+        )
 
     result = create_access_token({
         'sub': str(user.user_id)

core/routers/card/scraper.py

@@ -22,6 +22,17 @@ from core.utils.helper import async_retry, decrypt_aes_cbc, generate_keyname
 logger = logging.getLogger(__name__)
 
 
+async def get_cf_clearance(platform: str) -> dict:
+    redis_conn = await get_cache()
+    async with redis_conn:
+        cache_session_str = await redis_conn.get(generate_keyname(f'session:{platform}'))
+    if isinstance(cache_session_str, str):
+        headers = ujson.decode(cache_session_str)
+    else:
+        headers = {}
+    return headers
+
+
 class SiteApi:
     def __init__(self):
         self.user_agent = UserAgent(
@@ -99,9 +110,12 @@ class SiteApi:
         headers = {
             # 'X-Requested-With': 'XMLHttpRequest',
             'Origin': 'https://www.psacard.com',
-            'Referer': f'https://www.psacard.com/cert/{cert_number}',
+            'Referer': f'https://www.psacard.com/cert/{cert_number}/psa',
         }
-        r = await self.get(url, headers=headers, impersonate="chrome110")
+
+        cf_clearance_headers = await get_cf_clearance(platform='psa')
+        headers.update(cf_clearance_headers)
+        r = await self.get(url, headers=headers, use_abroad_proxy=True, use_random_ua=False, http_version=2, impersonate="chrome110")
         return r
 
     async def psa_images(self, cert_number: str | int, images_id: str | int):
@@ -130,14 +144,7 @@ class SiteApi:
     async def cgc(self, cert_number: str | int):
         """CGC"""
         url = f'https://www.cgccards.com/certlookup/{cert_number}/'
-        redis_conn = await get_cache()
-        async with redis_conn:
-            cache_session_str = await redis_conn.get(generate_keyname('session:cgc'))
-        if isinstance(cache_session_str, str):
-            headers = ujson.decode(cache_session_str)
-        else:
-            headers = {}
-
+        headers = await get_cf_clearance(platform='cgc')
         return await self.get(url, headers=headers, use_random_ua=False, use_abroad_proxy=True)
 
     async def gbtc(self, cert_number: str | int):
@@ -229,7 +236,7 @@ class CardScraper:
         if html.xpath('//div[contains(@class, "wrapper")]//h1/text()').get() is not None:
             return
 
-        info_elements = html.xpath('//div[@hidden and @id]/div[contains(@class, "py-3")]')
+        info_elements = html.xpath('//dl[@class="text-body1"]/div[contains(@class, "py-3")]')
         info = dict()
         for info_element in info_elements:
             name = info_element.xpath('./dt[1]').xpath('string()').get() or info_element.xpath(
@@ -267,8 +274,8 @@ class CardScraper:
         item['card_type'] = info.get('category')
         # 卡片正、反图片（远程）
         item['card_spare_picture'] = {
-            'front_picture': html.xpath('//div[contains(@class, "flex")]//span/noscript/img[@alt="left"]/@src').get(),
-            'reverse_picture': html.xpath('//div[contains(@class, "flex")]//span/noscript/img[@alt="right"]/@src').get()
+            'front_picture': html.xpath('//div[contains(@class, "flex")]//span//img[@alt="left"]/@src').get(),
+            'reverse_picture': html.xpath('//div[contains(@class, "flex")]//span//img[@alt="right"]/@src').get()
         }
 
         # 来源
@@ -297,7 +304,7 @@ class CardScraper:
         item['card_score'] = result.get('final_grade')
         # 签字评分
         auto_score = result.get('autograph_grade')
-        item['auto_score'] = auto_score if auto_score != '0.0' else None
+        item['auto_score'] = auto_score if auto_score != '0.0' and auto_score != 'Not Available' else None
         # 卡片名称
         item['card_name'] = result.get('player_name')
         # 卡片编号 无

core/routers/chat/routes.py

@@ -270,6 +270,42 @@ async def delete_single_chat(
     return {'message': '聊天已删除'}
 
 
+@chat_router.post(
+    '/messages/make-all-read',
+    summary='聊天信息标记为已读',
+    description='支持批量操作多个聊天', response_model=base_schema.BaseResponse
+)
+async def make_messages_read(
+        payload: schema.MakeMessageReadIn,
+        login_user: LoginUser = Depends(get_current_user_form_http),
+        db_session: AsyncSession = Depends(get_async_session),
+):
+    db_curd = CRUD(db_session, Chat)
+    chat_mapping = {str(chat.chat_guid): chat.last_read_message_id for chat in payload.chats}
+    chats = await db_curd.get_all(filters={'guid__in': chat_mapping.keys()})
+
+    for chat in chats:
+        last_read_message_id = chat_mapping.get(chat.guid)
+        if last_read_message_id and await db_curd.exists(
+                model=Message,
+                filters={
+                    'id': last_read_message_id,
+                    'chat_id': chat.id,
+                }
+        ):
+            await db_curd.update(
+                model=ChatParticipant,
+                data={'last_read_message_id': last_read_message_id},
+                filters={
+                    'chat_id': chat.id,
+                    'user_id': login_user.user_id,
+                    'last_read_message_id__lt': last_read_message_id,
+                },
+            )
+
+    return {'message': 'ok'}
+
+
 @chat_router.websocket('/ws')
 async def get_chat_websocket(
         websocket: WebSocket,

core/routers/chat/schema.py

@@ -6,7 +6,7 @@
 # @Desc :
 from typing import Literal, Optional, List
 
-from pydantic import BaseModel, conint, UUID4
+from pydantic import BaseModel, UUID4
 
 from core.base.schema import MessageType, Paginated
 from core.routers.user.schema import LiteUser
@@ -35,6 +35,15 @@ class ChatPaginated(Paginated):
     unread_messages_total: int
 
 
+class MakeMessageRead(BaseModel):
+    chat_guid: UUID4
+    last_read_message_id: int
+
+
+class MakeMessageReadIn(BaseModel):
+    chats: List[MakeMessageRead]
+
+
 class ChatUnreadOut(BaseModel):
     chat_guid: UUID4
     created_at: str

core/routers/user/actions.py

@@ -0,0 +1,62 @@
+# -*- coding: utf-8 -*-
+# @File: actions.py
+# @Date: 2025/10/15 18:58
+# @Software: PyCharm
+# @Author: xingc
+# @Desc:
+import logging
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.db.crud import CRUD
+from core.db.engine import async_db_session
+from core.db.models import User, UserFollows
+
+logger = logging.getLogger(__name__)
+
+async def update_user_follower_total(
+        follower_id: int,
+        followed_id: int,
+        db_session: AsyncSession
+):
+    """
+    更新用户计数
+    :param follower_id: 关注者id
+    :param followed_id: 被关注者id
+    :param db_session:
+    :return:
+    """
+    db_crud = CRUD(db_session, UserFollows)
+
+    # 更新关注数
+    followers_count = await  db_crud.total(
+        filters={
+            'follower_id': follower_id,
+            'followed_id': followed_id,
+        }
+    )
+    await db_crud.update(
+        model=User,
+        data={
+            'followers_count': followers_count,
+        },
+        filters={
+            'user_id': follower_id,
+        }
+    )
+
+    # 更新粉丝数
+    fans_count = await  db_crud.total(
+        filters={
+            'followed_id': followed_id,
+        }
+    )
+    await db_crud.update(
+        model=User,
+        data={
+            'fans_count': fans_count,
+        },
+        filters={
+            'user_id': followed_id,
+        }
+    )

core/routers/user/routes.py

@@ -18,6 +18,7 @@ from core.routers.auth.schema import LoginUser
 from core.routers.auth.services import get_current_user_form_http
 
 from . import schema
+from .actions import update_user_follower_total
 
 user_router = APIRouter(prefix='/user', tags=['用户模块'])
 
@@ -158,10 +159,15 @@ async def follow_user(
     }
     if action == 'follower':
         await db_crud.create(data=record, filters=record)
-
-    if action == 'unfollower':
+    elif action == 'unfollower':
         await db_crud.delete(filters=record)
 
+    await update_user_follower_total(
+        follower_id=login_user.user_id,
+        followed_id=follower_user.user_id,
+        db_session=db_session
+    )
+
     return {'message': 'ok'}
 
 

core/routers/user/schema.py

@@ -4,14 +4,15 @@
 # @Software : PyCharm
 # @Author : xingc
 # @Desc :
-from datetime import date
+from datetime import date, datetime
 from typing import Optional
 
-from fastapi.openapi.models import Schema
-from pydantic import BaseModel, constr
+from pydantic import constr
+
+from core.base.schema import BaseMixin
 
 
-class User(BaseModel):
+class User(BaseMixin):
     nickname: str
     user_id: int
     gender: Optional[int] = None
@@ -22,9 +23,13 @@ class User(BaseModel):
     vip_expire_date: Optional[date] = None
     points: Optional[int] = 0
     displayed_card_total: Optional[int] = 0
+    description: Optional[str | None] = None
+    last_active_at: Optional[datetime | None] = None
+    fans_count: Optional[int] = 0
+    followers_count: Optional[int] = 0
 
 
-class LiteUser(BaseModel):
+class LiteUser(BaseMixin):
     nickname: str
     user_id: int
     gender: Optional[int] = None
@@ -35,16 +40,20 @@ class LiteUser(BaseModel):
 class VisitUser(LiteUser):
     is_follow: Optional[bool] = False
     displayed_card_total: Optional[int] = 0
+    description: Optional[str | None] = None
+    last_active_at: Optional[datetime | None] = None
+    fans_count: Optional[int] = 0
 
 
-class UserFollows(BaseModel):
+class UserFollows(BaseMixin):
     nickname: str
     user_id: int
     gender: Optional[int] = None
     avatar_url: Optional[int] = None
 
 
-class UpdateUserIn(BaseModel):
+class UpdateUserIn(BaseMixin):
     nickname: Optional[constr(max_length=20)] = None
     gender: Optional[int] = None
     avatar_url: Optional[constr(pattern='^http[a-zA-Z0-9:/\.]+')] = None
+    description: Optional[constr(max_length=200) | None] = None

tasks.py

@@ -0,0 +1,161 @@
+# -*- coding: utf-8 -*-
+# @File: tasks.py
+# @Date: 2025/8/14 02:47
+# @Software: PyCharm
+# @Author: xingc
+# @Desc: 辅助定时任务
+import asyncio
+import logging
+from datetime import datetime, timedelta
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.cron import CronTrigger
+from sqlalchemy import func, desc, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from core.db.crud import CRUD
+from core.db.engine import get_cache, async_db_session
+from core.db.models import User, SearchCardCache, PointsTransaction
+from core.utils.helper import generate_keyname
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s | %(name)s | %(levelname)s - %(message)s",  # 日志格式
+)
+logger = logging.getLogger(__name__)
+
+
+async def check_cache_key_expire():
+    """检测需要定义过期的键"""
+    redis_conn = await get_cache()
+    current_date = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
+    second = ((current_date + timedelta(hours=1)) - current_date).total_seconds()
+    expire_keynames = set()
+    async with redis_conn:
+        keynames = await redis_conn.keys(generate_keyname('*', use_today=True))
+        logger.info(keynames)
+        for keyname in keynames:
+            if await redis_conn.ttl(keyname) == -1:
+                expire_keynames.add(keyname)
+                await redis_conn.expire(keyname, int(second))
+    logger.info('【任务】：检测需要定义过期的键\t'
+                f'键名：{expire_keynames}\t'
+                f'数量：{len(expire_keynames)}')
+
+
+@async_db_session()
+async def daily_member_points(db_session: AsyncSession, batch_step: int = 50):
+    """会员每日积分赠送"""
+    member_total_smt = select(
+        func.count(User.user_id),
+    ).where(
+        User.vip_expire_date > User.created_at
+    )
+    member_total_result = await db_session.execute(member_total_smt)
+    member_total = member_total_result.scalar()
+    if member_total < 1:
+        logger.info("【任务】：会员每日积分赠送\t没有需要更新积分的会员用户")
+        return
+
+    page = 0
+    success_total = 0
+    while True:
+        stmt = (select(User).filter(
+            User.vip_expire_date > User.created_at
+        ).with_for_update().offset(page * batch_step)
+                .limit(batch_step)
+                .order_by(User.created_at.asc()))
+        result = await db_session.execute(stmt)
+        users = result.scalars().all()
+        if not users:
+            break
+
+        user_ids = {user.user_id for user in users}
+        try:
+            update_stmt = update(
+                User
+            ).values(
+                points=User.points + 1
+            ).where(
+                User.user_id.in_(user_ids),
+            )
+            await db_session.execute(update_stmt)
+
+            current_date = datetime.now()
+            points_records = [
+                PointsTransaction(
+                    user_id=user.user_id,
+                    transaction_type=3,
+                    points=1,
+                    balance=user.points + 1,
+                    description="会员每日积分赠送",
+                    created_at=current_date
+                )
+                for user in users
+            ]
+            db_session.add_all(points_records)
+            await db_session.commit()
+            success_total += len(users)
+
+        except Exception as e:
+            logger.info(f"【任务】：会员每日积分赠送\tuser_ids -> {user_ids}\terror -> {e}")
+            await db_session.rollback()
+            continue
+
+        finally:
+            page += 1
+
+    logger.info(f"【任务】：会员每日积分赠送\t成功为 {success_total} 个会员用户赠送积分")
+
+
+@async_db_session()
+async def clear_expire_cache_card(db_session: AsyncSession):
+    """清除过期卡片缓存信息"""
+    expire_date = datetime.now() + timedelta(days=30)
+    db_crud = CRUD(db_session, SearchCardCache)
+    result = await db_crud.deletes(
+        filters={
+            'created_at__lt': expire_date
+        }
+    )
+    logger.info('【任务】：清除过期卡片缓存信息\t'
+                f'过期时间：{expire_date}\t'
+                f'结果：{result}')
+
+
+async def main():
+    scheduler = AsyncIOScheduler()
+    # crontab
+    scheduler.add_job(
+        check_cache_key_expire,
+        trigger=CronTrigger(hour='*/1'),
+        id='check_cache_key_expire'
+    )
+
+    scheduler.add_job(
+        daily_member_points,
+        trigger=CronTrigger(day='*/1', hour='8'),
+        id='daily_member_points'
+    )
+
+    scheduler.add_job(
+        clear_expire_cache_card,
+        trigger=CronTrigger(day='*/1', hour='3'),
+        id='clear_expire_cache_card'
+    )
+
+    scheduler.start()
+
+    try:
+        while True:
+            await asyncio.sleep(1)
+    except (KeyboardInterrupt, SystemExit):
+        scheduler.shutdown()
+        # await engine.dispose()
+
+
+if __name__ == '__main__':
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        logger.info("程序已停止")