database.py 2.9 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101
  1. # 导入:
  2. import uuid
  3. from datetime import datetime
  4. from urllib import parse
  5. import yaml
  6. from sqlalchemy import Column, Integer, String, DateTime, Text, and_
  7. from sqlalchemy.ext.declarative import declarative_base
  8. from sqlalchemy.orm import sessionmaker
  9. from common_func import get_db_engine
  10. # 创建对象的基类:
  11. Base = declarative_base()
  12. passowrd = parse.quote_plus('hcst@2020')
  13. # 定义User对象:
  14. class Task(Base):
  15. # 表的名字:
  16. __tablename__ = 'tb_asr_result'
  17. # 表的结构:
  18. signature = Column(String(100))
  19. task_id = Column(Integer, primary_key=True)
  20. task_status = Column(Integer)
  21. task_result = Column(Text)
  22. word_text = Column(Text)
  23. word_split = Column(Text)
  24. word_text_md5 = Column(Text)
  25. create_time = Column(DateTime, nullable=True, default=datetime.now)
  26. update_time = Column(DateTime)
  27. # 初始化数据库连接:
  28. with open('/data/pythonProject/video_to_word/config/config.yaml', mode='r', encoding='utf-8') as f:
  29. config = yaml.load(f.read(), Loader=yaml.FullLoader)
  30. # 1-1 数据库连接引擎,依据开发环境/生产环境 进行切换
  31. mac = uuid.UUID(int=uuid.getnode()).hex[-12:]
  32. if mac in ['5254003fa716', '52540003f5dd']:
  33. engine = get_db_engine(config['ai_word_dev_db'])
  34. else:
  35. engine = get_db_engine(config['ai_word_product_db'])
  36. # 创建DBSession类型:
  37. DBSession = sessionmaker(bind=engine)
  38. def query(signature, task_status, task_id):
  39. condition = (1 == 1)
  40. if signature != None:
  41. condition = and_(condition, Task.signature == signature)
  42. if task_status != None:
  43. condition = and_(condition, Task.task_status == task_status)
  44. if task_id != None:
  45. condition = and_(condition, Task.task_id == task_id)
  46. # 创建Session:
  47. session = DBSession()
  48. # 创建Query查询,filter是where条件,最后调用one()返回唯一行,如果调用all()则返回所有行:
  49. task = session.query(Task).filter(condition).order_by(Task.create_time.desc()).all()
  50. # 关闭Session:
  51. session.close()
  52. return task
  53. def insert(task: Task):
  54. # 创建session对象:
  55. session = DBSession()
  56. # 添加到session:
  57. session.add(task)
  58. session.flush()
  59. # 提交即保存到数据库:
  60. session.commit()
  61. # 关闭session:
  62. session.close()
  63. return task
  64. def update(task):
  65. # 创建session对象:
  66. session = DBSession()
  67. taskdb = session.query(Task).filter(Task.task_id == task.task_id).one()
  68. taskdb.task_status = task.task_status
  69. taskdb.word_text = task.word_text
  70. taskdb.word_split = task.word_split
  71. taskdb.word_text_md5 = task.word_text_md5
  72. taskdb.task_result = task.task_result
  73. # 提交即保存到数据库:
  74. session.commit()
  75. # 关闭session:
  76. session.close()
  77. if __name__ == '__main__':
  78. print('test query')
  79. # task = Task(task_id = '3',task_status=1,signature='dx')
  80. # task = insert(task)
  81. task = query(None, None, 1)[0]
  82. print(task.signature)
  83. # update(task)