ai-yunying/testm/models.py

import datetime
import pydantic
from typing import List, Dict, Any, Optional
import sqlalchemy
import os
import sys
sys.path.append(os.path.dirname(os.path.dirname(__file__)))

from database.config import chat_task_table,DB_URL,ai_yunying_db
# from sqlmodel import Field, SQLModel,Relationship,Column,Session,select,func,UniqueConstraint
from typing import Optional,Dict,Tuple
from sqlalchemy.dialects.postgresql import ARRAY  

'''
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)  # 添加时间戳字段 

'''
'''
class UnReadUserData(SQLModel, table=True):
    id:Optional[int] = Field(default=None, primary_key=True)  
    name: Optional[str]  = Field(default=None)  
    avator: Optional[str]  = Field(default=None)  
    msg: Optional[str]  = Field(default=None)  
    unread_msg_count: int = Field(default=None)  
    msg_time: int | None = Field(default=None)  
    # chat_history: list = Field(sa_column=Column(ARRAY(String)),default=[])
    detail: int = Field(default=None)  
    create_time:datetime.datetime = Field(default_factory=datetime.datetime.now)
    is_done:bool = Field(default=False)
'''
'''
class UnReadUserData(pydantic.BaseModel):
    id:Optional[int] 
    name: Optional[str]
    avatar: Optional[str] 
    msg: Optional[str]
    unread_msg_count: Optional[int]
    msg_time: Optional[int|str] 
    chat_history: Optional[list|dict|str]
    detail: Optional[int]
    create_time: Optional[datetime.datetime]
    is_done: Optional[bool]

    @property
    def chat_history(self):
        return self.chat_history
'''
'''
class UnReadUserData(Schema):
    id:Optional[int] 
    name = fields.Str()
    avatar = fields.Str()
    msg = fields.Str()
    unread_msg_count = fields.Int()
    msg_time = fields.Str()
    chat_history = fields.List(fields.Dict())
    detail = fields.Str()
    create_time = fields.Time()
    is_done = fields.Bool()
'''
'''
# https://github.com/ponyorm/pony
# https://docs.ponyorm.org/
from pony.orm import *
from pony import orm
db = Database()
db.bind(provider='postgres', user='pg', password='pg', host='sv-v',port=5432, database='ai_yunying')

class Person(db.Entity):
    name = Required(str)
    age = Required(int)
    cars = Set('Car')

class Car(db.Entity):
    make = Required(str)
    model = Required(str)
    owner = Required(Person)
db.generate_mapping(create_tables=True)
    p1 = Person(name='John', age=20)
    p2 = Person(name='Mary', age=22)
    p3 = Person(name='Bob', age=30)
    c1 = Car(make='Toyota', model='Prius', owner=p2)
    c2 = Car(make='Ford', model='Explorer', owner=p3)
    commit()
'''

'''
from sqlalchemy import Column, Integer, String, Boolean,DateTime
from sqlalchemy.dialects.postgresql import JSON
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()
from sqlalchemy import create_engine  
from sqlalchemy.orm import sessionmaker  
from dataset import Table
from sqlalchemy.orm import as_declarative 

class UnReadUserData(Base):
    __tablename__ = 'unread_user_data'

    id = Column(Integer, primary_key=True, )
    name = Column(String)
    avatar = Column(String)
    msg = Column(DateTime)
    unread_msg_count = Column(Integer)
    msg_time = Column(String)  # 或者其他适合的类型
    chat_history = Column(JSON)
    detail = Column(JSON)
    create_time = Column(DateTime)
    is_done = Column(Boolean)
    
    def to_dict(self, ignore=['id']): 
        ret = {}
        for c in self.__table__.columns:
             if c.name not in ignore:
                ret[c.name] = getattr(self, c.name)
        return ret
# 创建引擎  
engine = create_engine(DB_URL, echo=False)  
  
# 创建表（如果表不存在）  
Base.metadata.create_all(engine)  
  
# 创建会话制造工厂  
def add_uu():
    Session = sessionmaker(bind=engine)
    session = Session()  
    session.add(uu)  
    session.commit()  
    session.close()

def dataset_get():
    print(uu)
    print(uu.chat_history)
    print(type(uu))
    unread_user_data_table = ai_yunying_db['unread_user_data']
    # id = chat_task_table.insert(d)
    res = unread_user_data_table.find_one(id=1)
    chat_history = res.get('chat_history')
    print(res.get('chat_history'))
    print(type(chat_history))

'''

def main():
    import json
    import time
    d = {'name': '程序员马工', 'avatar': 'https://p3.huoshanimg.com/aweme/100x100/aweme-avatar/tos-cn-i-0813_66c4e34ae8834399bbf967c3d3c919db.jpeg?from=3782654143', 'msg': '你好', 'msg_time': '刚刚', 'chat_history': [{1:"555"}, {'你好':555}, {1:555}], 'detail': {'1':123}, 'unread_msg_count': 2}
    # uu  = UnReadUserData(name='sdfasd', avatar='12312')
    d['msg_time'] = time.time()
    uu  = UnReadUserData(**d)
    # unread_user_data_table:Table = ai_yunying_db['unread_user_data']
    unread_user_data_table = chat_task_table
    id = unread_user_data_table.insert(uu.to_dict(), types={"chat_history": JSON, "detail": JSON})
    res = unread_user_data_table.find_one(id=id)
    chat_history = res.get('chat_history')
    print("chat_history " , chat_history)
    print(type(chat_history))
    detail = res.get('detail')
    print(detail)
    print(type(detail))
    msg_time = res.get('msg_time')
    print("msg_time ", msg_time)
    print(type(msg_time))
    # chat_history = json.loads(chat_history)
    return
    create_all_table()
    with Session(engine) as session:
        session.add(user_data)
        session.commit()
    # data = user_data.dict()
    # data.pop('id')
    # chat_task_table.insert(data)

if __name__ == "__main__":
    main()