Spaces:
Sleeping
Sleeping
| import os | |
| from sqlalchemy import create_engine, Column, Integer, String, DateTime, Float, Text, ForeignKey, Boolean | |
| from sqlalchemy.ext.declarative import declarative_base | |
| from sqlalchemy.orm import sessionmaker, relationship | |
| from datetime import datetime | |
| Base = declarative_base() | |
| class BrandAnalysis(Base): | |
| """Store overall analysis results for a brand search""" | |
| __tablename__ = 'brand_analyses' | |
| id = Column(Integer, primary_key=True) | |
| search_query = Column(String(500), nullable=False) | |
| brand_name = Column(String(200), nullable=False) | |
| search_engine = Column(String(50), nullable=False) # google, bing, duckduckgo | |
| total_articles = Column(Integer, default=0) | |
| articles_with_mentions = Column(Integer, default=0) | |
| total_mentions = Column(Integer, default=0) | |
| positive_count = Column(Integer, default=0) | |
| negative_count = Column(Integer, default=0) | |
| neutral_count = Column(Integer, default=0) | |
| created_at = Column(DateTime, default=datetime.utcnow) | |
| # Relationships | |
| mentions = relationship("BrandMention", back_populates="analysis", cascade="all, delete-orphan") | |
| articles = relationship("Article", back_populates="analysis", cascade="all, delete-orphan") | |
| class Article(Base): | |
| """Store article information""" | |
| __tablename__ = 'articles' | |
| id = Column(Integer, primary_key=True) | |
| analysis_id = Column(Integer, ForeignKey('brand_analyses.id'), nullable=False) | |
| url = Column(Text, nullable=False) | |
| title = Column(Text) | |
| content = Column(Text) | |
| overall_sentiment = Column(String(20)) | |
| summary = Column(Text) | |
| created_at = Column(DateTime, default=datetime.utcnow) | |
| # Relationships | |
| analysis = relationship("BrandAnalysis", back_populates="articles") | |
| mentions = relationship("BrandMention", back_populates="article", cascade="all, delete-orphan") | |
| class BrandMention(Base): | |
| """Store individual brand mentions""" | |
| __tablename__ = 'brand_mentions' | |
| id = Column(Integer, primary_key=True) | |
| analysis_id = Column(Integer, ForeignKey('brand_analyses.id'), nullable=False) | |
| article_id = Column(Integer, ForeignKey('articles.id'), nullable=False) | |
| brand_name = Column(String(200), nullable=False) | |
| mention_type = Column(String(20)) # explicit, indirect | |
| mention_text = Column(Text) | |
| context = Column(Text) | |
| sentiment = Column(String(20)) # positive, negative, neutral | |
| confidence = Column(Float, default=0.0) # Confidence score for sentiment | |
| explanation = Column(Text) | |
| created_at = Column(DateTime, default=datetime.utcnow) | |
| # Relationships | |
| analysis = relationship("BrandAnalysis", back_populates="mentions") | |
| article = relationship("Article", back_populates="mentions") | |
| class ScheduledMonitoring(Base): | |
| """Store scheduled monitoring jobs""" | |
| __tablename__ = 'scheduled_monitoring' | |
| id = Column(Integer, primary_key=True) | |
| search_query = Column(String(500), nullable=False) | |
| brand_names = Column(Text, nullable=False) # Comma-separated brand names | |
| search_engines = Column(Text, nullable=False) # Comma-separated search engines | |
| schedule_type = Column(String(20), default='weekly') # daily, weekly, monthly | |
| is_active = Column(Boolean, default=True) | |
| last_run = Column(DateTime) | |
| next_run = Column(DateTime) | |
| created_at = Column(DateTime, default=datetime.utcnow) | |
| updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow) | |
| class CoMention(Base): | |
| """Store co-mention relationships between brands""" | |
| __tablename__ = 'co_mentions' | |
| id = Column(Integer, primary_key=True) | |
| brand1 = Column(String(200), nullable=False) | |
| brand2 = Column(String(200), nullable=False) | |
| article_id = Column(Integer, ForeignKey('articles.id'), nullable=False) | |
| co_occurrence_count = Column(Integer, default=1) | |
| created_at = Column(DateTime, default=datetime.utcnow) | |
| # Database connection and session management | |
| def get_database_engine(): | |
| """Create and return database engine""" | |
| database_url = os.getenv('DATABASE_URL') | |
| if not database_url: | |
| raise ValueError("DATABASE_URL environment variable not set") | |
| return create_engine(database_url) | |
| def get_session(): | |
| """Create and return database session""" | |
| engine = get_database_engine() | |
| Session = sessionmaker(bind=engine) | |
| return Session() | |
| def init_database(): | |
| """Initialize database tables""" | |
| engine = get_database_engine() | |
| Base.metadata.create_all(engine) | |
| print("Database tables created successfully") | |
| def drop_all_tables(): | |
| """Drop all tables (use with caution!)""" | |
| engine = get_database_engine() | |
| Base.metadata.drop_all(engine) | |
| print("All tables dropped") | |