Spaces:
Sleeping
Sleeping
| """数据库模型""" | |
| from __future__ import annotations | |
| import os | |
| from datetime import datetime | |
| from sqlalchemy import Column, DateTime, Float, Integer, String, Text, create_engine | |
| from sqlalchemy.orm import DeclarativeBase, Session, sessionmaker | |
| DB_PATH = os.getenv("DB_PATH", "speechbridge.db") | |
| engine = create_engine(f"sqlite:///{DB_PATH}", echo=False) | |
| SessionLocal = sessionmaker(bind=engine) | |
| class Base(DeclarativeBase): | |
| pass | |
| class User(Base): | |
| """用户""" | |
| __tablename__ = "users" | |
| id = Column(Integer, primary_key=True, autoincrement=True) | |
| username = Column(String(50), unique=True, nullable=False) | |
| email = Column(String(120), unique=True, nullable=False) | |
| password_hash = Column(String(256), nullable=False) | |
| # 个性化设置 | |
| default_scene = Column(String(50), default="general") | |
| default_accessibility = Column(Integer, default=0) | |
| language = Column(String(10), default="zh") | |
| created_at = Column(DateTime, default=datetime.utcnow) | |
| class Task(Base): | |
| """处理任务""" | |
| __tablename__ = "tasks" | |
| id = Column(Integer, primary_key=True, autoincrement=True) | |
| user_id = Column(Integer, nullable=True) # 关联用户 (可选) | |
| filename = Column(String(255), nullable=False) | |
| scene = Column(String(50), default="general") | |
| accessibility = Column(Integer, default=0) | |
| # ASR 结果 | |
| asr_text = Column(Text, default="") | |
| asr_confidence = Column(Float, default=0.0) | |
| asr_language = Column(String(10), default="en") | |
| asr_duration = Column(Float, default=0.0) | |
| # 语义恢复结果 | |
| corrected_text = Column(Text, default="") | |
| intent = Column(Text, default="") | |
| entities = Column(Text, default="[]") | |
| actions = Column(Text, default="[]") | |
| corrections = Column(Text, default="[]") | |
| recovery_confidence = Column(Float, default=0.0) | |
| # 元信息 | |
| processing_time = Column(Float, default=0.0) | |
| status = Column(String(20), default="pending") | |
| error_message = Column(Text, default="") | |
| created_at = Column(DateTime, default=datetime.utcnow) | |
| updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) | |
| def init_db(): | |
| """初始化数据库""" | |
| Base.metadata.create_all(engine) | |
| def get_db(): | |
| """获取数据库会话""" | |
| db = SessionLocal() | |
| try: | |
| yield db | |
| finally: | |
| db.close() | |