Spaces:
Running
Running
File size: 4,720 Bytes
90fe073 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | 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")
|