hello-agents/Co-creation-projects/DeLunnLi-PaperGraph/backend/app/services/daily/user_behavior_analytics.py

"""用户行为分析 —— 基于浏览/保存/搜索行为更新用户偏好画像."""

from __future__ import annotations

import contextlib
import re
import sqlite3
from collections import Counter
from dataclasses import dataclass

from ...utils import build_in_clause

@dataclass
class UserInterestProfile:

    top_keywords: list[tuple[str, float]]

    top_subdomains: list[tuple[str, float]]

    high_interest_paper_ids: list[int]

    recent_active_topics: list[str]

    preferred_years: list[int]

    preferred_sources: list[str]

class UserBehaviorAnalytics:

    def __init__(self, db_path: str) -> None:
        self.db_path = db_path

    def _get_connection(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn

    @contextlib.contextmanager
    def _cursor(self):
        conn = self._get_connection()
        try:
            yield conn.cursor()
        finally:
            conn.close()

    def extract_keywords_from_text(self, text: str) -> list[str]:

        if not text:
            return []
        t = text.lower()
        t = re.sub(r"[^a-z0-9\u4e00-\u9fff]+", " ", t)
        tokens = [x.strip() for x in t.split() if x.strip()]
        stop = {
            "the", "a", "an", "and", "or", "of", "to", "in", "for", "with", "on",
            "we", "our", "is", "are", "be", "via", "from", "this", "that",
            "using", "use", "based", "towards", "paper", "propose", "method",
            "learning", "network", "model", "deep", "neural",
        }
        return [x for x in tokens if x not in stop and len(x) >= 4][:200]

    def get_papers_by_reading_time(
        self, *, days: int = 30, min_duration: int = 30, top_n: int = 50,
    ) -> list[tuple[int, float]]:
        with self._cursor() as cur:
            cur.execute(
                "SELECT paper_id,SUM(duration_sec) AS total_duration FROM paper_reading_sessions WHERE day_key>=date('now',?) AND duration_sec>=? GROUP BY paper_id ORDER BY total_duration DESC LIMIT ?",
                (f"-{days} days", min_duration, top_n),
            )
            return [(int(r["paper_id"]), float(r["total_duration"])) for r in cur.fetchall()]

    def get_papers_by_reading_frequency(
        self, *, days: int = 30, min_sessions: int = 2, top_n: int = 30,
    ) -> list[tuple[int, int]]:
        with self._cursor() as cur:
            cur.execute(
                "SELECT paper_id,COUNT(*) AS session_count FROM paper_reading_sessions WHERE day_key>=date('now',?) GROUP BY paper_id HAVING COUNT(*)>=? ORDER BY session_count DESC LIMIT ?",
                (f"-{days} days", min_sessions, top_n),
            )
            return [(int(r["paper_id"]), int(r["session_count"])) for r in cur.fetchall()]

    def get_recently_saved_papers(
        self, *, days: int = 30, top_n: int = 50,
    ) -> list[tuple[int, str]]:
        with self._cursor() as cur:
            cur.execute(
                "SELECT id,category FROM papers WHERE created_at>=strftime('%s','now',?) ORDER BY created_at DESC LIMIT ?",
                (f"-{days} days", top_n),
            )
            return [(int(r["id"]), str(r["category"] or "")) for r in cur.fetchall()]

    def extract_keywords_from_high_interest_papers(
        self, paper_ids: list[int],
    ) -> Counter[str]:
        if not paper_ids:
            return Counter()
        with self._cursor() as cur:
            in_clause, params = build_in_clause("id", paper_ids)
            cur.execute(f"SELECT title,abstract,keywords FROM papers WHERE {in_clause}", params)
            all_kw: list[str] = []
            for row in cur.fetchall():
                all_kw.extend(self.extract_keywords_from_text(str(row["title"] or "")))
                all_kw.extend(self.extract_keywords_from_text(str(row["abstract"] or "")))
                for kw in (str(row["keywords"] or "")).split(","):
                    k = kw.strip().lower()
                    if k and len(k) >= 3:
                        all_kw.append(k)
        return Counter(all_kw)

    def get_interest_subdomains_from_papers(self, paper_ids: list[int]) -> Counter[str]:
        if not paper_ids:
            return Counter()
        with self._cursor() as cur:
            in_clause, params = build_in_clause("id", paper_ids)
            cur.execute(f"SELECT title,category,journal FROM papers WHERE {in_clause}", params)
            subdomains = self._extract_subdomains_from_rows(cur.fetchall())
        return subdomains

    def _extract_subdomains_from_rows(self, rows) -> Counter[str]:
        """Count topics from paper metadata. LLM daily pipeline handles semantic classification."""
        subdomains: Counter[str] = Counter()
        for row in rows:
            cat = (row.get("category") or "").strip()
            if cat:
                subdomains[cat.lower()] += 1
        return subdomains

    def get_feedback_enhanced_keywords(
        self,
        *,
        days: int = 21,
    ) -> Counter[str]:
        from .daily_recommend_feedback import get_high_value_keywords_from_feedback

        try:
            keywords_set = get_high_value_keywords_from_feedback(self.db_path, days=days, top_n=30)
            return Counter({kw: 2.0 for kw in keywords_set})
        except Exception:
            return Counter()

    def get_user_interest_profile(
        self,
        *,
        reading_days: int = 30,
        saved_days: int = 60,
        feedback_days: int = 21,
    ) -> UserInterestProfile:
        high_duration_papers = self.get_papers_by_reading_time(days=reading_days, top_n=50)
        high_duration_ids = [pid for pid, _ in high_duration_papers]

        freq_papers = self.get_papers_by_reading_frequency(days=reading_days, top_n=30)
        freq_ids = [pid for pid, _ in freq_papers]

        saved_papers = self.get_recently_saved_papers(days=saved_days, top_n=50)
        saved_ids = [pid for pid, _ in saved_papers]

        all_interest_ids = list(set(high_duration_ids + freq_ids + saved_ids))

        reading_keywords = self.extract_keywords_from_high_interest_papers(all_interest_ids)

        feedback_keywords = self.get_feedback_enhanced_keywords(days=feedback_days)

        combined_keywords: Counter[str] = Counter()
        for kw, count in reading_keywords.items():
            combined_keywords[kw] += count * 1.0
        for kw, weight in feedback_keywords.items():
            combined_keywords[kw] += weight

        if saved_ids:
            saved_paper_ids = [pid for pid, _ in saved_papers]
            if saved_paper_ids:
                saved_keywords = self.extract_keywords_from_high_interest_papers(saved_paper_ids)
                for kw, count in saved_keywords.items():

                    if kw in reading_keywords:
                        combined_keywords[kw] += count * 0.5
                    else:
                        combined_keywords[kw] += count * 1.5

        subdomains = self.get_interest_subdomains_from_papers(all_interest_ids)

        preferred_years = self._extract_preferred_years(all_interest_ids)

        recent_topics = self._extract_recent_active_topics(reading_days=14)

        top_keywords = combined_keywords.most_common(40)
        top_subdomains = subdomains.most_common(10)

        return UserInterestProfile(
            top_keywords=top_keywords,
            top_subdomains=top_subdomains,
            high_interest_paper_ids=all_interest_ids[:100],
            recent_active_topics=recent_topics,
            preferred_years=preferred_years,
            preferred_sources=[],
        )

    def _extract_preferred_years(self, paper_ids: list[int]) -> list[int]:
        if not paper_ids:
            return []
        with self._cursor() as cur:
            in_clause, params = build_in_clause("id", paper_ids)
            cur.execute(f"SELECT year,COUNT(*) AS cnt FROM papers WHERE {in_clause} AND year IS NOT NULL GROUP BY year ORDER BY cnt DESC LIMIT 5", params)
            return [int(r["year"]) for r in cur.fetchall() if r["year"]]

    def _extract_recent_active_topics(self, reading_days: int = 14) -> list[str]:
        with self._cursor() as cur:
            cur.execute(
                "SELECT DISTINCT p.title,p.abstract,p.category FROM papers p INNER JOIN paper_reading_sessions prs ON p.id=prs.paper_id WHERE prs.day_key>=date('now',?) LIMIT 30",
                (f"-{reading_days} days",),
            )
            topics = self._extract_subdomains_from_rows(cur.fetchall())
        return [t for t, _ in topics.most_common(5)]

def get_user_interest_profile_for_daily_recommend(db_path: str) -> UserInterestProfile:
    analytics = UserBehaviorAnalytics(db_path)
    return analytics.get_user_interest_profile(
        reading_days=30,
        saved_days=60,
        feedback_days=21,
    )