yu-zhi-ran/platform/backend/app/models.py

from sqlalchemy import Column, String, Integer, Float, Date, DateTime, Text, Boolean, JSON
from sqlalchemy.sql import func
from .database import Base
from datetime import datetime


class AuditLog(Base):
    __tablename__ = "audit_logs"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    user_id = Column(Integer, nullable=True, index=True)  # 操作用户ID（未登录/匿名可为空）
    username = Column(String, nullable=False)  # 操作用户名（冗余存储）
    action = Column(String, nullable=False, index=True)  # 操作类型: login/logout/create_user/update_user/delete_user/publish/etc.
    resource_type = Column(String, nullable=True, index=True)  # 资源类型: user/topic/publish_record/etc.
    resource_id = Column(String, nullable=True)  # 资源ID
    details = Column(JSON, default=dict, nullable=True)  # 操作详情（变更前后、额外信息等）
    ip_address = Column(String, nullable=True)  # IP 地址
    user_agent = Column(String, nullable=True)  # User-Agent
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    def to_dict(self):
        return {
            "id": self.id,
            "user_id": self.user_id,
            "username": self.username,
            "action": self.action,
            "resource_type": self.resource_type,
            "resource_id": self.resource_id,
            "details": self.details or {},
            "ip_address": self.ip_address,
            "user_agent": self.user_agent,
            "created_at": self.created_at.isoformat() if self.created_at else None
        }


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    username = Column(String, unique=True, nullable=False, index=True)
    password_hash = Column(String, nullable=False)  # bcrypt 哈希
    role = Column(String, default="user", nullable=False)  # admin/user
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    def to_dict(self):
        return {
            "id": self.id,
            "username": self.username,
            "role": self.role,
            "created_at": self.created_at.isoformat() if self.created_at else None
        }


class Topic(Base):
    __tablename__ = "topics"

    id = Column(String, primary_key=True, index=True)
    title = Column(String, nullable=False)
    field = Column(String, nullable=False)
    format = Column(String)
    core_concept = Column(Text)
    audience_pain = Column(Text)
    unique_angle = Column(Text)
    priority = Column(String)  # 高/中
    priority_score = Column(Integer, default=0)
    total_score = Column(Float)
    status = Column(String, default="pending")  # pending/draft/ready/published
    cases = Column(JSON, default=list)
    source_file = Column(String)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())
    generated_at = Column(DateTime(timezone=True), nullable=True)  # 选题创作完成时间
    ready_at = Column(Date)
    published_at = Column(Date)
    compliance_score = Column(Integer)
    platform_urls = Column(JSON, default=dict)  # {"zhihu": "...", "wechat": "...", "xiaohongshu": "..."}


class Article(Base):
    __tablename__ = "articles"

    id = Column(String, primary_key=True)  # e.g., A01_zhihu
    topic_id = Column(String, nullable=False)
    platform = Column(String, nullable=False)
    file_path = Column(String, nullable=False)
    status = Column(String, default="draft")  # draft/optimized/published
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    compliance_score = Column(Integer)
    html_content = Column(Text)  # 可缓存HTML内容以便预览


class PublishRecord(Base):
    __tablename__ = "publish_records"

    id = Column(Integer, primary_key=True, index=True)
    topic_id = Column(String, nullable=False)
    platform = Column(String, nullable=False)  # 发布平台：zhihu/wechat/xiaohongshu
    action = Column(String, nullable=False)  # 操作：publish/update/delete/invalid
    status = Column(String, nullable=False)  # 状态：success/failed/partial/cancelled
    operator = Column(String, nullable=True)  # 操作人
    description = Column(Text, nullable=True)  # 发布说明
    suggestion = Column(Text, nullable=True)  # 建议内容
    url = Column(String, nullable=True)  # 发布后链接
    error_msg = Column(Text, nullable=True)  # 错误信息
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    # 关联查询
    # 可以添加外键关联到 Topic, 但这里保持简单


class Case(Base):
    __tablename__ = "cases"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    title = Column(String, nullable=False)
    field = Column(String, nullable=False)  # 对应四大支柱或其子领域
    summary = Column(Text)
    key_metrics = Column(Text, nullable=True)  # 关键数据
    date = Column(String, nullable=True)  # 日期或年份字符串
    source = Column(String, nullable=True)
    credibility_rating = Column(String, nullable=True)  # 如 "⭐⭐⭐"
    china_applicability = Column(String, nullable=True)  # 如 "⭐⭐⭐⭐"
    source_url = Column(String, nullable=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    def to_dict(self):
        return {
            "id": self.id,
            "title": self.title,
            "field": self.field,
            "summary": self.summary,
            "key_metrics": self.key_metrics,
            "date": self.date,
            "source": self.source,
            "credibility_rating": self.credibility_rating,
            "china_applicability": self.china_applicability,
            "source_url": self.source_url,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
        }


class TaskLog(Base):
    __tablename__ = "task_logs"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    task_name = Column(String, nullable=False)  # collector, creator, research, outline, writer, optimizer
    topic_id = Column(String, nullable=True)  # 关联选题ID
    status = Column(String, nullable=False)  # started, completed, failed
    message = Column(Text, nullable=True)
    started_at = Column(DateTime(timezone=True), server_default=func.now())
    finished_at = Column(DateTime(timezone=True), nullable=True)
    duration = Column(Integer, nullable=True)  # 秒数

    def to_dict(self):
        return {
            "id": self.id,
            "task_name": self.task_name,
            "topic_id": self.topic_id,
            "status": self.status,
            "message": self.message,
            "started_at": self.started_at.isoformat() if self.started_at else None,
            "finished_at": self.finished_at.isoformat() if self.finished_at else None,
            "duration": self.duration,
        }


class LLMConfig(Base):
    __tablename__ = "llm_configs"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    name = Column(String, unique=True, nullable=False)  # 如 default_expand
    system_prompt = Column(Text, nullable=True)
    user_prompt_template = Column(Text, nullable=False)  # 含占位符 {topic.get('title')} 等
    temperature = Column(Float, default=0.7)
    max_tokens = Column(Integer, default=2000)
    model = Column(String, nullable=True)  # 后端模型名
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "system_prompt": self.system_prompt,
            "user_prompt_template": self.user_prompt_template,
            "temperature": self.temperature,
            "max_tokens": self.max_tokens,
            "model": self.model,
            "is_active": self.is_active,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
        }


class SystemConfig(Base):
    __tablename__ = "system_configs"

    id = Column(Integer, primary_key=True, index=True, autoincrement=True)
    key = Column(String, unique=True, nullable=False)
    value = Column(Text, nullable=True)  # 可存储 JSON 字符串
    description = Column(String, nullable=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())
    updated_at = Column(DateTime(timezone=True), onupdate=func.now())

    def to_dict(self):
        import json
        return {
            "id": self.id,
            "key": self.key,
            "value": json.loads(self.value) if self.value else None,
            "description": self.description,
            "created_at": self.created_at.isoformat() if self.created_at else None,
            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
        }