File size: 10,526 Bytes
7c6ffa6
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
"""PYQ Pattern Analysis Service.

Analyses extracted PYQ questions for patterns:
- Repeated topics across years
- Marks distribution
- Answer type distribution
- Chapter-wise and topic-wise question counts
- Years seen

Rules:
- Every pattern must be backed by actual question IDs.
- If fewer than 2 years exist, say "not enough data for pattern analysis."
- Do not use words like "sure shot" or fake high-yield.
- Use "seen in uploaded papers" instead.
"""
from __future__ import annotations

from collections import Counter
from dataclasses import dataclass, field
from typing import Any

from sqlalchemy import select
from sqlalchemy.orm import Session

from app.models.previous_question import PreviousQuestion
from app.models.previous_paper import PreviousPaper


@dataclass
class PatternEvidence:
    question_id: str
    question_text: str
    year: int | None
    marks: int | None
    chapter: str
    topic: str
    paper_id: str
    paper_title: str
    source_url: str | None


@dataclass
class TopicPattern:
    topic: str
    chapter: str
    count: int
    years_seen: list[int]
    question_ids: list[str]
    evidence: list[PatternEvidence]


@dataclass
class MarksDistribution:
    marks: int
    count: int
    percentage: float


@dataclass
class AnswerTypeDistribution:
    answer_type: str
    count: int
    percentage: float


@dataclass
class ChapterWiseCount:
    chapter: str
    total_questions: int
    total_marks: int
    topics: list[str]


@dataclass
class PatternAnalysisResult:
    repeated_topics: list[TopicPattern] = field(default_factory=list)
    marks_distribution: list[MarksDistribution] = field(default_factory=list)
    answer_type_distribution: list[AnswerTypeDistribution] = field(default_factory=list)
    chapter_wise_count: list[ChapterWiseCount] = field(default_factory=list)
    topic_wise_count: list[TopicPattern] = field(default_factory=list)
    years_seen: list[int] = field(default_factory=list)
    total_questions: int = 0
    total_papers: int = 0
    evidence_list: list[PatternEvidence] = field(default_factory=list)
    analysis_note: str = ""
    confidence: float = 0.0


_MIN_YEARS_FOR_PATTERN = 2


def analyze_pyq_patterns(
    db: Session,
    user_id: str,
    *,
    subject: str = "",
    board: str = "",
    class_level: str = "",
    chapter: str = "",
    topic: str = "",
    year_range: tuple[int | None, int | None] | None = None,
    paper_ids: list[str] | None = None,
) -> PatternAnalysisResult:
    """Analyze PYQ patterns from uploaded questions in the database.

    All patterns are backed by actual question IDs from the database.
    """
    result = PatternAnalysisResult()

    paper_query = (
        select(PreviousPaper)
        .where(PreviousPaper.user_id == user_id)
        .where(PreviousPaper.verification_status == "verified")
    )
    if paper_ids:
        paper_query = paper_query.where(PreviousPaper.id.in_(paper_ids))
    if subject:
        paper_query = paper_query.where(PreviousPaper.subject == subject)
    if board:
        paper_query = paper_query.where(PreviousPaper.board == board)
    if class_level:
        paper_query = paper_query.where(PreviousPaper.class_level == class_level)

    papers = list(db.scalars(paper_query).all())

    if year_range:
        min_year, max_year = year_range
        papers = [
            paper for paper in papers
            if (not min_year or not paper.year or paper.year >= min_year)
            and (not max_year or not paper.year or paper.year <= max_year)
        ]

    result.total_papers = len(papers)

    questions: list[PreviousQuestion] = []
    for paper in papers:
        paper_questions = list(paper.questions)
        for q in paper_questions:
            # Apply filters
            if chapter and q.chapter and chapter.lower() not in q.chapter.lower():
                continue
            if topic and q.topic and topic.lower() not in q.topic.lower():
                continue
            if year_range:
                min_year, max_year = year_range
                if min_year and q.year and q.year < min_year:
                    continue
                if max_year and q.year and q.year > max_year:
                    continue
            questions.append(q)

    result.total_questions = len(questions)

    if not questions:
        result.analysis_note = (
            "No uploaded PYQ questions found for the given filters. "
            "Upload previous year question papers to unlock pattern analysis."
        )
        return result

    # Build evidence list
    evidence_map: dict[str, PatternEvidence] = {}
    for q in questions:
        evidence = PatternEvidence(
            question_id=q.id,
            question_text=q.question_text[:200],
            year=q.year,
            marks=q.marks,
            chapter=q.chapter or "",
            topic=q.topic or "",
            paper_id=q.previous_paper.id,
            paper_title=q.previous_paper.title,
            source_url=q.previous_paper.source_url,
        )
        evidence_map[q.id] = evidence
        result.evidence_list.append(evidence)

    # Years seen
    all_years = sorted({q.year for q in questions if q.year is not None})
    result.years_seen = all_years

    # Check if enough data for pattern analysis
    if len(all_years) < _MIN_YEARS_FOR_PATTERN:
        result.analysis_note = (
            f"Only {len(all_years)} year(s) of data available. "
            "Upload papers from at least 2 different years for meaningful pattern analysis."
        )
        result.confidence = 0.3
        # Still provide basic stats
        _compute_basic_stats(questions, evidence_map, result)
        return result

    # Sufficient data — compute full patterns
    result.confidence = min(0.5 + len(all_years) * 0.05, 1.0)

    # Repeated topics
    topic_year_map: dict[str, dict[int, list[str]]] = {}
    for q in questions:
        # Unknown-year evidence cannot prove repetition across exam years.
        if q.year is None:
            continue
        topic_key = q.topic or q.chapter or "Unknown"
        if topic_key not in topic_year_map:
            topic_year_map[topic_key] = {}
        year = q.year
        if year not in topic_year_map[topic_key]:
            topic_year_map[topic_key][year] = []
        topic_year_map[topic_key][year].append(q.id)

    for topic_key, year_map in topic_year_map.items():
        if len(year_map) >= 2:
            years = sorted([y for y in year_map if y > 0])
            q_ids = []
            evidence_items = []
            for y, ids in year_map.items():
                q_ids.extend(ids)
                evidence_items.extend([evidence_map[qid] for qid in ids if qid in evidence_map])

            # Find chapter for this topic
            chapter_val = ""
            for q in questions:
                if (q.topic or q.chapter or "") == topic_key:
                    chapter_val = q.chapter or ""
                    break

            result.repeated_topics.append(TopicPattern(
                topic=topic_key,
                chapter=chapter_val,
                count=len(q_ids),
                years_seen=years,
                question_ids=q_ids,
                evidence=evidence_items[:5],
            ))

    # Sort repeated topics by frequency
    result.repeated_topics.sort(key=lambda t: t.count, reverse=True)

    _compute_basic_stats(questions, evidence_map, result)

    if not result.analysis_note:
        result.analysis_note = (
            f"Pattern analysis based on {result.total_questions} questions "
            f"across {len(all_years)} years ({', '.join(str(y) for y in all_years)}). "
            "All patterns are backed by uploaded question paper evidence."
        )

    return result


def _compute_basic_stats(
    questions: list[PreviousQuestion],
    evidence_map: dict[str, PatternEvidence],
    result: PatternAnalysisResult,
) -> None:
    """Compute marks distribution, answer type distribution, and chapter-wise counts."""
    # Marks distribution
    marks_counter: Counter[int] = Counter()
    for q in questions:
        if q.marks is not None:
            marks_counter[q.marks] += 1

    total = len(questions) or 1
    result.marks_distribution = [
        MarksDistribution(
            marks=marks,
            count=count,
            percentage=round(count / total * 100, 1),
        )
        for marks, count in sorted(marks_counter.items())
    ]

    # Answer type distribution
    type_counter: Counter[str] = Counter()
    for q in questions:
        # Standard extraction fills question_type; answer_type is optional.
        at = q.answer_type or q.question_type or "unknown"
        type_counter[at] += 1

    result.answer_type_distribution = [
        AnswerTypeDistribution(
            answer_type=at,
            count=count,
            percentage=round(count / total * 100, 1),
        )
        for at, count in type_counter.most_common()
    ]

    # Chapter-wise count
    chapter_map: dict[str, dict[str, Any]] = {}
    for q in questions:
        ch = q.chapter or "Unknown"
        if ch not in chapter_map:
            chapter_map[ch] = {"count": 0, "marks": 0, "topics": set()}
        chapter_map[ch]["count"] += 1
        chapter_map[ch]["marks"] += q.marks or 0
        if q.topic:
            chapter_map[ch]["topics"].add(q.topic)

    result.chapter_wise_count = [
        ChapterWiseCount(
            chapter=ch,
            total_questions=data["count"],
            total_marks=data["marks"],
            topics=sorted(data["topics"]),
        )
        for ch, data in sorted(chapter_map.items(), key=lambda x: x[1]["count"], reverse=True)
    ]

    # Topic-wise count (all topics, not just repeated)
    topic_map: dict[str, dict[str, Any]] = {}
    for q in questions:
        t = q.topic or q.chapter or "Unknown"
        if t not in topic_map:
            topic_map[t] = {"count": 0, "years": set(), "ids": [], "chapter": q.chapter or ""}
        topic_map[t]["count"] += 1
        if q.year:
            topic_map[t]["years"].add(q.year)
        topic_map[t]["ids"].append(q.id)

    result.topic_wise_count = [
        TopicPattern(
            topic=t,
            chapter=data["chapter"],
            count=data["count"],
            years_seen=sorted(data["years"]),
            question_ids=data["ids"],
            evidence=[evidence_map[qid] for qid in data["ids"][:5] if qid in evidence_map],
        )
        for t, data in sorted(topic_map.items(), key=lambda x: x[1]["count"], reverse=True)
    ]