"""SQLAlchemy 数据模型""" from datetime import datetime from sqlalchemy import Column, Integer, String, Text, Boolean, DateTime, ForeignKey from sqlalchemy.orm import relationship from database import Base class Feed(Base): """RSS 源""" __tablename__ = "feeds" id = Column(Integer, primary_key=True, index=True) url = Column(String(2048), unique=True, nullable=False, index=True) title = Column(String(512), default="") description = Column(Text, default="") category = Column(String(128), default="") is_active = Column(Boolean, default=True, index=True) fetch_interval_minutes = Column(Integer, default=60) # 抓取统计 last_fetch_at = Column(DateTime, nullable=True) last_fetch_status = Column(String(20), default="") last_error = Column(Text, default="") success_count = Column(Integer, default=0) fail_count = Column(Integer, default=0) article_count = Column(Integer, default=0) created_at = Column(DateTime, default=datetime.utcnow) # 关联 articles = relationship("Article", back_populates="feed", cascade="all, delete-orphan") fetch_logs = relationship("FetchLog", back_populates="feed", cascade="all, delete-orphan") def health_status(self): """计算健康度 🟢 健康: 成功率 >= 90%, 最近7天有更新 🟡 警告: 成功率 50%-90%, 或超过3天未更新 🔴 异常: 成功率 < 50%, 或超过7天未更新 """ total = self.success_count + self.fail_count if total == 0: return "unknown" success_rate = self.success_count / total days_since_last_fetch = None if self.last_fetch_at: days_since_last_fetch = (datetime.utcnow() - self.last_fetch_at).days if success_rate >= 0.9 and (days_since_last_fetch is None or days_since_last_fetch <= 7): return "healthy" elif success_rate >= 0.5 and (days_since_last_fetch is None or days_since_last_fetch <= 7): return "warning" else: return "unhealthy" class Article(Base): """RSS 文章""" __tablename__ = "articles" id = Column(Integer, primary_key=True, index=True) feed_id = Column(Integer, ForeignKey("feeds.id", ondelete="CASCADE"), nullable=False, index=True) title = Column(String(1024), default="", index=True) link = Column(String(2048), unique=True, nullable=False, index=True) author = Column(String(256), default="") published_at = Column(DateTime, nullable=True, index=True) content = Column(Text, default="") summary = Column(Text, default="") is_read = Column(Boolean, default=False) created_at = Column(DateTime, default=datetime.utcnow, index=True) # 关联 feed = relationship("Feed", back_populates="articles") class FetchLog(Base): """抓取日志""" __tablename__ = "fetch_logs" id = Column(Integer, primary_key=True, index=True) feed_id = Column(Integer, ForeignKey("feeds.id", ondelete="CASCADE"), nullable=False, index=True) status = Column(String(20), nullable=False) # success / fail articles_fetched = Column(Integer, default=0) error_message = Column(Text, default="") response_time_ms = Column(Integer, nullable=True) created_at = Column(DateTime, default=datetime.utcnow, index=True) # 关联 feed = relationship("Feed", back_populates="fetch_logs")