douyin/db/user_info.py

import datetime
from typing import Optional
import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(__file__)))
from db.engine import engine

from sqlmodel import Field, SQLModel,Relationship,create_engine,Session,select,func,Column
import psycopg2
from config import DB_URL,logger
# from db.common import engine
from sqlalchemy import UniqueConstraint, Index, asc  
from sqlalchemy.dialects.postgresql import insert
from db.base import DouyinBaseRepository

'''
https://developer.open-douyin.com/docs/resource/zh-CN/dop/develop/openapi/account-management/get-account-open-info
理论上无需将用户信息存入数据库，因为抖音已经存储了这些信息，我本地再存储一份无疑造成数据冗余。

直接从抖音获取：
- 可以获得很好的实时性。比如用户更新了头像，在我的网站应用中也能实时更新
- 避免数据冗余，不增加无意义的代码量
- 没有请求次数限制

存储在本地：
- 我有大量的本地数据分析的需求
- 避免延迟和其他网络问题，但通常抖音服务器不会有网络问题

综上，我认为不需要将用户信息存储在本地数据库。因为抖音并没有限制该接口请求次数。

'''

# 避免循环导入问题，参考 SQLmodel 官方： https://sqlmodel.tiangolo.com/tutorial/code-structure/
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
    from .user import User


class UserInfoLink(SQLModel, table=True):
    user_id: Optional[int] = Field(
        default=None, foreign_key="user.id", primary_key=True
    )
    info_id: Optional[int] = Field(
        default=None, foreign_key="userinfo.id", primary_key=True
    )

class UserInfo(SQLModel, table=True):  
    id: Optional[int] = Field(default=None, primary_key=True)  
    avatar: Optional[str] = Field(default=None)  
    avatar_larger: Optional[str] = Field(default=None)  
    client_key:  Optional[str] = Field(default=None)  
    e_account_role:  Optional[str] = Field(default=None)    
    nickname:  Optional[str] = Field(default=None)   
    # 外键约束有助于：级联操作、避免冗余、数据完整性
    open_id: Optional[str] = Field(index=True, unique=True)
    union_id:  Optional[str] = Field(default=None)  
    update_time: datetime.datetime = Field(default_factory=datetime.datetime.now)  # 添加时间戳字段 
    user:"User" = Relationship(back_populates="info", link_model=UserInfoLink, sa_relationship_kwargs={"cascade": "all, delete-orphan", "single_parent":True})


class UserInfoRepository(DouyinBaseRepository):  
    def __init__(self, engine=engine):  
        super().__init__(UserInfo, engine)  
        self.model:UserInfo
    
async def main():
    from douyin.user_info import get_user_info
    from db.user_oauth import UserOAuthRepository
    db_oauth = UserOAuthRepository()
    oauth_model = db_oauth.get_by_open_id()
    data = await get_user_info(oauth_model.open_id, oauth_model.access_token)
    if data.get("error_code") != 0:
        raise HTTPException(status_code=400, detail=data)  
    db_user = UserInfoRepository()
    user_info_model = db_user.dict_to_model(data)
    db_user.add_or_update(user_info_model)
    
    # 生成并返回 token，包含过期时间  
    payload = {  
        "sub": data["open_id"],
        "exp": exp  # 添加过期时间戳（北京时间）到 payload  
    }  
    account_token = jwt.encode(payload, SECRET_KEY, algorithm="HS256")  
    logger.info(f"login success, expires_time:{datetime.datetime.fromtimestamp(exp).strftime('%Y-%m-%d %H:%M:%S') }, token:{account_token}")


if __name__ == "__main__":
    main()