mfinder/db/analytics_sql.py

import datetime
from sqlalchemy import Column, Integer, TEXT, DateTime, inspect, func
from mfinder.db.settings_sql import BASE, SESSION
from mfinder.db.files_sql import Files, clean_movie_title, INSERTION_LOCK, reconnect_session, OperationalError

class SearchAnalytics(BASE):
    __tablename__ = "search_analytics"
    query = Column(TEXT, primary_key=True)
    search_count = Column(Integer, default=1)
    last_searched = Column(DateTime, default=datetime.datetime.utcnow)

    def __init__(self, query):
        self.query = query.strip().lower()
        self.search_count = 1
        self.last_searched = datetime.datetime.utcnow()

# Ensure table exists in DB automatically on module load
try:
    session = SESSION()
    bind = session.get_bind()
    inspector = inspect(bind)
    if not inspector.has_table("search_analytics"):
        SearchAnalytics.__table__.create(bind=bind, checkfirst=True)
except Exception as _e:
    pass

async def log_search_query(query_text: str):
    """Logs or increments search query count in analytics database."""
    if not query_text or len(query_text.strip()) < 2:
        return
    
    clean_q = query_text.strip().lower()
    session = SESSION()
    try:
        record = session.query(SearchAnalytics).filter_by(query=clean_q).first()
        if record:
            record.search_count += 1
            record.last_searched = datetime.datetime.utcnow()
        else:
            record = SearchAnalytics(query=clean_q)
            session.add(record)
        session.commit()
    except Exception as e:
        session.rollback()
    finally:
        try:
            SESSION.close()
        except Exception:
            pass

async def cleanup_old_search_analytics():
    """Deletes search records older than 7 days to maintain a clean weekly trending cycle."""
    session = SESSION()
    try:
        one_week_ago = datetime.datetime.utcnow() - datetime.timedelta(days=7)
        session.query(SearchAnalytics).filter(SearchAnalytics.last_searched < one_week_ago).delete()
        session.commit()
    except Exception:
        session.rollback()
    finally:
        try:
            SESSION.close()
        except Exception:
            pass

def extract_season(text: str) -> str:
    if not text:
        return None
    import re
    # Match S02, s2, S-2, s_2, S(2), Season 2, Season-02, s02e03, etc.
    match = re.search(r'\b[sS](?:eason)?[\s_.-]*\(?(\d+)\)?(?:\b|[eE])', text)
    if match:
        season_num = int(match.group(1))
        return f"S{season_num}"
    return None

async def get_latest_uploaded_movies(limit: int = 25) -> list:
    """
    Retrieves the latest distinct uploaded movies from the database files table,
    sorted by newest created_at timestamp first.
    Returns list of tuples: (cleaned_title, file_count, year, season_str)
    """
    retries = 3
    while retries > 0:
        try:
            with INSERTION_LOCK:
                files = (
                    SESSION.query(Files.file_name, Files.caption)
                    .order_by(Files.created_at.desc(), Files.file_name.asc())
                    .limit(300)
                    .all()
                )

                movie_counts = {}
                movie_years = {}
                movie_seasons = {}
                for f_name, caption in files:
                    title = clean_movie_title(f_name)
                    if title and len(title) >= 2:
                        movie_counts[title] = movie_counts.get(title, 0) + 1
                        if title not in movie_years:
                            import re
                            year_match = re.search(r'\b(19\d{2}|20[0-2]\d|2030)\b', f_name)
                            if year_match:
                                movie_years[title] = year_match.group(1)
                                
                        # Extract season
                        season = extract_season(f_name)
                        if not season and caption:
                            season = extract_season(caption)
                        if season:
                            if title not in movie_seasons:
                                movie_seasons[title] = set()
                            movie_seasons[title].add(season)

                latest_list = []
                for title, count in movie_counts.items():
                    year = movie_years.get(title)
                    seasons = movie_seasons.get(title)
                    season_str = None
                    if seasons:
                        sorted_seasons = sorted(list(seasons), key=lambda s: int(s[1:]))
                        if len(sorted_seasons) > 1:
                            season_str = " & ".join(sorted_seasons)
                        else:
                            season_str = sorted_seasons[0]
                            
                    latest_list.append((title, count, year, season_str))
                    if len(latest_list) >= limit:
                        break

                return latest_list
        except OperationalError:
            reconnect_session()
            retries -= 1
        except Exception as e:
            retries -= 1
        finally:
            try:
                SESSION.close()
            except Exception:
                pass
    return []

async def get_top_trending_movies(limit: int = 10) -> list:
    """
    Retrieves top searched queries from the 7-day weekly cycle in search_analytics.
    Returns list of tuples: (movie_title, file_count)
    """
    # Trigger weekly cleanup of old search records
    await cleanup_old_search_analytics()

    retries = 3
    while retries > 0:
        try:
            with INSERTION_LOCK:
                one_week_ago = datetime.datetime.utcnow() - datetime.timedelta(days=7)
                top_records = (
                    SESSION.query(SearchAnalytics)
                    .filter(SearchAnalytics.last_searched >= one_week_ago)
                    .order_by(SearchAnalytics.search_count.desc(), SearchAnalytics.last_searched.desc())
                    .limit(50)
                    .all()
                )

                if not top_records:
                    return []

                trending_movies = []
                seen_titles = set()

                for rec in top_records:
                    q = rec.query
                    files = (
                        SESSION.query(Files.file_name, Files.caption)
                        .filter(Files.file_name.ilike(f"%{q}%"))
                        .limit(50)
                        .all()
                    )

                    if files:
                        for f_name, caption in files:
                            title = clean_movie_title(f_name)
                            if title and len(title) >= 2:
                                t_key = title.lower()
                                if t_key not in seen_titles:
                                    seen_titles.add(t_key)
                                    count = SESSION.query(Files).filter(Files.file_name.ilike(f"%{title}%")).count()
                                    
                                    # Extract year and seasons from matched files
                                    import re
                                    year = None
                                    seasons = set()
                                    for fn, cap in files:
                                        if clean_movie_title(fn) == title:
                                            year_match = re.search(r'\b(19\d{2}|20[0-2]\d|2030)\b', fn)
                                            if year_match:
                                                year = year_match.group(1)
                                            season = extract_season(fn)
                                            if not season and cap:
                                                season = extract_season(cap)
                                            if season:
                                                seasons.add(season)
                                            
                                    season_str = None
                                    if seasons:
                                        sorted_seasons = sorted(list(seasons), key=lambda s: int(s[1:]))
                                        if len(sorted_seasons) > 1:
                                            season_str = " & ".join(sorted_seasons)
                                        else:
                                            season_str = sorted_seasons[0]
                                            
                                    trending_movies.append((title, count if count > 0 else len(files), year, season_str))
                                    if len(trending_movies) >= limit:
                                        break
                    if len(trending_movies) >= limit:
                        break

                return trending_movies
        except OperationalError:
            reconnect_session()
            retries -= 1
        except Exception:
            retries -= 1
        finally:
            try:
                SESSION.close()
            except Exception:
                pass
    return []

from sqlalchemy import BigInteger

class UserInteraction(BASE):
    __tablename__ = "user_interactions"
    user_id = Column(BigInteger, primary_key=True)
    last_active = Column(DateTime, default=datetime.datetime.utcnow)

    def __init__(self, user_id):
        self.user_id = user_id
        self.last_active = datetime.datetime.utcnow()

# Ensure table exists in DB automatically on module load
try:
    session = SESSION()
    bind = session.get_bind()
    inspector = inspect(bind)
    if not inspector.has_table("user_interactions"):
        UserInteraction.__table__.create(bind=bind, checkfirst=True)
except Exception as _e:
    pass

async def log_user_interaction(user_id: int):
    if not user_id:
        return
    session = SESSION()
    try:
        record = session.query(UserInteraction).filter_by(user_id=user_id).first()
        if record:
            record.last_active = datetime.datetime.utcnow()
        else:
            record = UserInteraction(user_id=user_id)
            session.add(record)
        session.commit()
    except Exception as e:
        session.rollback()
    finally:
        try:
            SESSION.close()
        except Exception:
            pass

async def get_today_active_users_count() -> int:
    session = SESSION()
    try:
        now = datetime.datetime.utcnow()
        local_now = now + datetime.timedelta(hours=5, minutes=30)
        local_today_start = datetime.datetime(local_now.year, local_now.month, local_now.day)
        utc_today_start = local_today_start - datetime.timedelta(hours=5, minutes=30)
        count = session.query(UserInteraction).filter(UserInteraction.last_active >= utc_today_start).count()
        return count
    except Exception:
        return 0
    finally:
        try:
            SESSION.close()
        except Exception:
            pass

def get_movie_quality_sync(title: str) -> str:
    from mfinder.db.files_sql import Files
    from mfinder.utils.helpers import detect_file_quality
    session = SESSION()
    try:
        files = (
            session.query(Files.file_name)
            .filter(Files.file_name.ilike(f"%{title}%"))
            .order_by(Files.created_at.desc())
            .limit(10)
            .all()
        )
        if not files:
            return "HD"
        for (f_name,) in files:
            if detect_file_quality(f_name) == "Theatre Print [Clear Audio]":
                return "Theatre Print [Clear Audio]"
        return "HD"
    except Exception:
        return "HD"
    finally:
        try:
            session.close()
        except Exception:
            pass