22072536647

Committed 16 Feb 2026 05:41PM UTC coverage: 42.093%. First build

Build # 22072536647

Build Type

Pull #59

github

Pull Request Pull Request #59: apps/summerization: Integrate Document Summary into Workflow

Coverage Stats

51 of 314 new or added lines in 7 files covered. (16.24%)

3564 of 8467 relevant lines covered (42.09%)

0.42 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

32.81

/apps/summarization/services.py

"""Service for text summarization using AI providers."""

import json
import logging
from datetime import timedelta

from django.conf import settings
from django.utils import timezone
from pydantic import BaseModel
from sentry_sdk import capture_exception

from .models import ProjectSummary
from .providers import AIProvider
from .providers import AIRequest
from .providers import ProviderConfig
from .pydantic_models import DocumentInputItem
from .pydantic_models import DocumentSummaryItem
from .pydantic_models import DocumentSummaryResponse
from .pydantic_models import ProjectSummaryResponse
from .pydantic_models import SummaryItem
from .sentry_tags import set_sentry_project_tags
from .utils import extract_text_from_document

logger = logging.getLogger(__name__)


# Rate limits (can be moved to settings)
PROJECT_SUMMARY_RATE_LIMIT_MINUTES = getattr(
    settings, "PROJECT_SUMMARY_RATE_LIMIT_MINUTES", 5
)
SUMMARY_GLOBAL_LIMIT_PER_HOUR = getattr(settings, "SUMMARY_GLOBAL_LIMIT_PER_HOUR", 100)


class AIService:
    """Service for summarizing text using configured AI provider."""

    def __init__(
        self, provider_handle: str = None, document_provider_handle: str = None
    ):
        """Initialize AI service with providers."""
        self.provider = self._init_provider(provider_handle, "AI_PROVIDER")
        self.document_provider = self._init_provider(
            document_provider_handle, "AI_DOCUMENT_PROVIDER"
        )

    def _init_provider(self, handle: str | None, settings_key: str) -> AIProvider:
        """Initialize a provider from handle or settings."""
        if not handle:
            handle = getattr(settings, settings_key, None)
            if not handle:
                raise ValueError(
                    f"No provider configured. Pass {settings_key.lower()} or set {settings_key} in settings."
                )
        return AIProvider(ProviderConfig.from_handle(handle))

    def summarize(
        self,
        text: str,
        prompt: str | None = None,
        result_type: type[BaseModel] = SummaryItem,
    ) -> BaseModel:
        """Summarize text."""
        request = SummaryRequest(text=text, prompt=prompt)
        return self.provider.request(request, result_type=result_type)

    def project_summarize(
        self,
        project,
        text: str,
        prompt: str | None = None,
        result_type: type[BaseModel] = ProjectSummaryResponse,
        is_rate_limit: bool = True,
        allow_regeneration: bool = True,
    ) -> BaseModel:
        """Summarize project data with caching.

        - Exact hash match: reuse cached summary and update last_checked_at.
        - Rate limits (if enabled): optionally reuse latest summary without touching last_checked_at.
        - If allow_regeneration is False and a summary exists, always return the latest summary
          without generating a new one, even when the hash changed.
        """
        set_sentry_project_tags(project)
        request = SummaryRequest(text=text, prompt=prompt)
        latest = self._get_latest_summary(project)
        text_hash = ProjectSummary.compute_hash(text)

        cached = self._get_cached_response(
            project=project,
            text_hash=text_hash,
            latest=latest,
            is_rate_limit=is_rate_limit,
        )
        if cached:
            return cached

        # No cache hit:
        # - If regeneration is not allowed and we already have a summary,
        #   return the latest one unchanged (used by the button endpoint).
        if not allow_regeneration and latest:
            logger.debug(
                "Regeneration disabled and no cache hit; returning latest summary "
                f"for project {project.id}"
            )
            return ProjectSummaryResponse(**latest.response_data)

        # If there is no existing summary, we must generate one at least once.
        logger.info(f"Generating summary for project {project.id} ({project.slug})")

        try:
            response = self.provider.request(request, result_type=result_type)
            self._save_to_cache(project, request.prompt_text, text_hash, response)
            return response
        except Exception as e:
            logger.error(f"Summary generation failed: {e}", exc_info=True)
            capture_exception(e)

            raise

    def _get_latest_summary(self, project):
        """Get most recent summary for project."""
        return (
            ProjectSummary.objects.filter(project=project)
            .order_by("-created_at")
            .first()
        )

    def _get_cached_response(
        self,
        project,
        text_hash: str,
        latest,
        is_rate_limit: bool,
    ) -> ProjectSummaryResponse | None:
        """Return cached response if valid.

        - Exact hash match: always used, updates last_checked_at.
        - Rate limits: optionally reuse latest summary without touching last_checked_at.
        """
        if not latest:
            return None

        # Exact match of input hash: content is up to date.
        if latest.input_text_hash == text_hash:
            logger.debug(f"Cache hit (exact match) for project {project.id}")
            latest.last_checked_at = timezone.now()
            latest.save(update_fields=["last_checked_at"])
            return ProjectSummaryResponse(**latest.response_data)

        if not is_rate_limit:
            return None

        # Rate limit checks (reuse latest summary without updating last_checked_at)
        age = timezone.now() - latest.created_at

        if age < timedelta(minutes=PROJECT_SUMMARY_RATE_LIMIT_MINUTES):
            logger.debug(f"Using rate-limited cache for project {project.id}")
            return ProjectSummaryResponse(**latest.response_data)

        if age < timedelta(hours=1):
            recent = ProjectSummary.objects.filter(
                created_at__gte=timezone.now() - timedelta(hours=1)
            ).count()
            if recent >= SUMMARY_GLOBAL_LIMIT_PER_HOUR:
                logger.debug(
                    f"Global rate limit reached, using cache for project {project.id}"
                )
                return ProjectSummaryResponse(**latest.response_data)

        return None

    def _save_to_cache(
        self,
        project,
        prompt: str,
        text_hash: str,
        response: BaseModel,
    ):
        """Save successful response to cache."""
        if isinstance(response, ProjectSummaryResponse):
            ProjectSummary.objects.create(
                project=project,
                prompt=prompt,
                input_text_hash=text_hash,
                response_data=json.loads(response.model_dump_json()),
                last_checked_at=timezone.now(),
            )
            logger.info(f"Cached summary for project {project.id}")

    def request_vision_dict(
        self,
        documents_dict: dict[str, str],
        prompt: str | None = None,
        project=None,
    ) -> DocumentSummaryResponse:
        """Process documents from dictionary format."""
        items = [DocumentInputItem(handle=h, url=u) for h, u in documents_dict.items()]
        return self.request_vision(items, prompt, project=project)

    def request_vision(
        self,
        documents: list[DocumentInputItem],
        prompt: str | None = None,
        project=None,
    ) -> DocumentSummaryResponse:
        """Process documents and images, return combined summaries."""
        if project is not None:
            set_sentry_project_tags(project)
        docs, images = self._split_documents(documents)

        results = []
        if docs[0]:
            results.extend(self._process_documents(docs))
        if images[0]:
            results.extend(self._process_images(images, prompt))

        return DocumentSummaryResponse(documents=results)

    def _split_documents(self, documents):
        """Split into regular docs and images."""
        docs_urls, docs_handles = [], []
        img_urls, img_handles = [], []

        for doc in documents:
            if (
                not self.document_provider.config.supports_documents
                and doc.is_document()
            ):
                docs_urls.append(doc.url)
                docs_handles.append(doc.handle)
            else:
                img_urls.append(doc.url)
                img_handles.append(doc.handle)

        return (docs_urls, docs_handles), (img_urls, img_handles)

    def _process_documents(self, docs_data):
        """Extract text from PDFs/DOCX files."""
        urls, handles = docs_data
        results = []

        for url, handle in zip(urls, handles):
            try:
                text = extract_text_from_document(url)
                results.append(DocumentSummaryItem(handle=handle, summary=text))
            except Exception as e:
                logger.error(
                    f"Failed to extract text from {handle}: {e}", exc_info=True
                )
                capture_exception(e)

        return results

    def _process_images(self, images_data, prompt):
        """Process images with vision API."""
        urls, handles = images_data

        if not prompt:
            prompt = (
                f"Summarize each image separately. Handles in order: {handles}. "
                f"Return list of summaries with handles."
            )

        request = MultimodalSummaryRequest(image_urls=urls, prompt=prompt)
        response = self.document_provider.request(request, DocumentSummaryResponse)
        return response.documents


class SummaryRequest(AIRequest):
    """Request model for text summarization."""

    DEFAULT_PROMPT = """
You are a JSON generator. Return ONLY valid JSON.

Schema:
{
  "title": "Summary of participation",
  "general_info": {"summary": "string", "goals": ["string"]},
  "phases": {
    "past": {"modules": [{"module_id": "number", "module_name": "string", "status": "past", "final": {"summary": "string", "bullets": ["string"]}, "debug": {...}}]},
    "current": {"modules": [{"module_id": "number", "module_name": "string", "status": "current", "final": {"summary": "string", "bullets": ["string"]}, "debug": {...}}]},
    "upcoming": {"modules": [{"module_id": "number", "module_name": "string", "status": "upcoming", "final": {"summary": "string", "bullets": ["string"]}, "debug": {...}}]}
  }
}

NOTE: module_id in the output should match the given module_id of the input for each module

Each module MUST include a 'debug' object with:
- module_type: string
- signals_snapshot: list of strings
- draft_before_qa: string
- claims: list of {claim_text, evidence_type(from_votes|from_ratings|from_open_answers|from_comments|from_base_text|uncertain), action(keep|soften|replace|remove), fix_hint}
- quantifier_fixes: list of {original_phrase, replacement, reason}
- anchors: list of strings
- coverage_gaps: list of strings
- coverage_patch: optional string
- patches: list of {patch_type(REPLACE|REMOVE|ADD_SENTENCE), target, replacement}
- after_qa: string
- diff_summary: optional string
- qa_status: PASS|FAIL

Extract real data from the project export.
Respond with ONLY the JSON object.
"""

    def __init__(self, text: str, prompt: str | None = None):
        super().__init__()
        self.text = text
        self.prompt_text = prompt or self.DEFAULT_PROMPT

    def prompt(self) -> str:
        return f"{self.prompt_text}\n\n{self.text}"


class MultimodalSummaryRequest(AIRequest):
    """Request model for multimodal document summarization."""

    vision_support = True
    PROMPT = "Summarize this image/document. Return JSON with summary field."

    def __init__(
        self, image_urls: list[str], text: str | None = None, prompt: str | None = None
    ):
        super().__init__()
        self.image_urls = image_urls
        self.prompt_text = prompt or self.PROMPT
        self.text = text

    def prompt(self) -> str:
        base = self.prompt_text
        return f"{base}\n\nText:\n{self.text}" if self.text else base


class DocumentRequest(AIRequest):
    """Request model for document summarization."""

    vision_support = True
    PROMPT = "Summarize this document. Return JSON with summary field."

    def __init__(self, url: str, prompt: str | None = None):
        super().__init__()
        self.image_urls = [url]
        self.prompt_text = prompt or self.PROMPT

    def prompt(self) -> str:
        return self.prompt_text

1	"""Service for text summarization using AI providers."""
2
3	import json	1✔
4	import logging	1✔
5	from datetime import timedelta	1✔
6
7	from django.conf import settings	1✔
8	from django.utils import timezone	1✔
9	from pydantic import BaseModel	1✔
10	from sentry_sdk import capture_exception	1✔
11
12	from .models import ProjectSummary	1✔
13	from .providers import AIProvider	1✔
14	from .providers import AIRequest	1✔
15	from .providers import ProviderConfig	1✔
16	from .pydantic_models import DocumentInputItem	1✔
17	from .pydantic_models import DocumentSummaryItem	1✔
18	from .pydantic_models import DocumentSummaryResponse	1✔
19	from .pydantic_models import ProjectSummaryResponse	1✔
20	from .pydantic_models import SummaryItem	1✔
21	from .sentry_tags import set_sentry_project_tags	1✔
22	from .utils import extract_text_from_document
23		1✔
24	logger = logging.getLogger(__name__)
25
26		1✔
27	# Rate limits (can be moved to settings)
28	PROJECT_SUMMARY_RATE_LIMIT_MINUTES = getattr(
29	settings, "PROJECT_SUMMARY_RATE_LIMIT_MINUTES", 5	1✔
30	)
31	SUMMARY_GLOBAL_LIMIT_PER_HOUR = getattr(settings, "SUMMARY_GLOBAL_LIMIT_PER_HOUR", 100)
32		1✔
33
34	class AIService:
35	"""Service for summarizing text using configured AI provider."""	1✔
36
37	def __init__(
38	self, provider_handle: str = None, document_provider_handle: str = None
39	):	×
40	"""Initialize AI service with providers."""
41	self.provider = self._init_provider(provider_handle, "AI_PROVIDER")
42	self.document_provider = self._init_provider(
43	document_provider_handle, "AI_DOCUMENT_PROVIDER"	×
44	)	×
45
NEW 46	def _init_provider(self, handle: str \| None, settings_key: str) -> AIProvider:	×
47	"""Initialize a provider from handle or settings."""
48	if not handle:
NEW 49	handle = getattr(settings, settings_key, None)	×
NEW 50	if not handle:	×
NEW 51	raise ValueError(	×
52	f"No provider configured. Pass {settings_key.lower()} or set {settings_key} in settings."
NEW 53	)	×
54	return AIProvider(ProviderConfig.from_handle(handle))
55		1✔
56	def summarize(
57	self,
58	text: str,
59	prompt: str \| None = None,
60	result_type: type[BaseModel] = SummaryItem,
61	) -> BaseModel:
62	"""Summarize text."""	×
63	request = SummaryRequest(text=text, prompt=prompt)	×
64	return self.provider.request(request, result_type=result_type)	×
65
66	def project_summarize(	1✔
67	self,
68	project,
69	text: str,
70	prompt: str \| None = None,
71	result_type: type[BaseModel] = ProjectSummaryResponse,
72	is_rate_limit: bool = True,
73	allow_regeneration: bool = True,
74	) -> BaseModel:
75	"""Summarize project data with caching.
76
77	- Exact hash match: reuse cached summary and update last_checked_at.	×
78	- Rate limits (if enabled): optionally reuse latest summary without touching last_checked_at.
79	- If allow_regeneration is False and a summary exists, always return the latest summary
80	without generating a new one, even when the hash changed.	×
81	"""
82	set_sentry_project_tags(project)
83	request = SummaryRequest(text=text, prompt=prompt)
84	latest = self._get_latest_summary(project)
85	text_hash = ProjectSummary.compute_hash(text)
86
87	cached = self._get_cached_response(	×
88	project=project,
89	text_hash=text_hash,	×
90	latest=latest,	×
91	is_rate_limit=is_rate_limit,	×
92	)
93	if cached:
94	return cached	×
95
96	# No cache hit:
97	# - If regeneration is not allowed and we already have a summary,	×
98	# return the latest one unchanged (used by the button endpoint).	×
99	if not allow_regeneration and latest:	×
100	logger.debug(
101	"Regeneration disabled and no cache hit; returning latest summary "
102	f"for project {project.id}"	×
103	)
104	return ProjectSummaryResponse(**latest.response_data)
105		×
106	# If there is no existing summary, we must generate one at least once.	×
107	logger.info(f"Generating summary for project {project.id} ({project.slug})")	×
108
109	try:
110	response = self.provider.request(request, result_type=result_type)
111	self._save_to_cache(project, request.prompt_text, text_hash, response)	×
112	return response	×
113	except Exception as e:
114	logger.error(f"Summary generation failed: {e}", exc_info=True)
115	capture_exception(e)	×
116
117	raise
118		×
119	def _get_latest_summary(self, project):
120	"""Get most recent summary for project."""
121	return (
122	ProjectSummary.objects.filter(project=project)
123	.order_by("-created_at")	×
124	.first()	×
125	)	×
126
127	def _get_cached_response(
128	self,	×
129	project,	×
130	text_hash: str,	×
131	latest,
132	is_rate_limit: bool,
133	) -> ProjectSummaryResponse \| None:
134	"""Return cached response if valid.
135
136	- Exact hash match: always used, updates last_checked_at.	×
137	- Rate limits: optionally reuse latest summary without touching last_checked_at.
138	"""	1✔
139	if not latest:
140	return None
141
142	# Exact match of input hash: content is up to date.
143	if latest.input_text_hash == text_hash:
NEW 144	logger.debug(f"Cache hit (exact match) for project {project.id}")	×
NEW 145	latest.last_checked_at = timezone.now()	×
NEW 146	latest.save(update_fields=["last_checked_at"])	×
NEW 147	return ProjectSummaryResponse(**latest.response_data)	×
148
NEW 149	if not is_rate_limit:	×
NEW 150	return None	×
151
152	# Rate limit checks (reuse latest summary without updating last_checked_at)
153	age = timezone.now() - latest.created_at
NEW 154		×
NEW 155	if age < timedelta(minutes=PROJECT_SUMMARY_RATE_LIMIT_MINUTES):	×
156	logger.debug(f"Using rate-limited cache for project {project.id}")
NEW 157	return ProjectSummaryResponse(**latest.response_data)	×
NEW 158		×
159	if age < timedelta(hours=1):
NEW 160	recent = ProjectSummary.objects.filter(	×
NEW 161	created_at__gte=timezone.now() - timedelta(hours=1)	×
NEW 162	).count()	×
NEW 163	if recent >= SUMMARY_GLOBAL_LIMIT_PER_HOUR:	×
164	logger.debug(
NEW 165	f"Global rate limit reached, using cache for project {project.id}"	×
NEW 166	)	×
NEW 167	return ProjectSummaryResponse(**latest.response_data)	×
NEW 168		×
169	return None
NEW 170		×
171	def _save_to_cache(
172	self,	1✔
173	project,
174	prompt: str,
175	text_hash: str,
176	response: BaseModel,
177	):
178	"""Save successful response to cache."""
179	if isinstance(response, ProjectSummaryResponse):
180	ProjectSummary.objects.create(
181	project=project,
182	prompt=prompt,
183	input_text_hash=text_hash,
184	response_data=json.loads(response.model_dump_json()),
185	last_checked_at=timezone.now(),
186	)
NEW 187	logger.info(f"Cached summary for project {project.id}")	×
188
189	def request_vision_dict(
190	self,
NEW 191	documents_dict: dict[str, str],	×
192	prompt: str \| None = None,
193	project=None,	1✔
194	) -> DocumentSummaryResponse:
195	"""Process documents from dictionary format."""
196	items = [DocumentInputItem(handle=h, url=u) for h, u in documents_dict.items()]
197	return self.request_vision(items, prompt, project=project)
198
NEW 199	def request_vision(	×
NEW 200	self,	×
201	documents: list[DocumentInputItem],
NEW 202	prompt: str \| None = None,	×
203	project=None,
204	) -> DocumentSummaryResponse:
205	"""Process documents and images, return combined summaries."""
206	if project is not None:
207	set_sentry_project_tags(project)
208	docs, images = self._split_documents(documents)
NEW 209		×
210	results = []
211	if docs[0]:
212	results.extend(self._process_documents(docs))
NEW 213	if images[0]:	×
214	results.extend(self._process_images(images, prompt))
215
216	return DocumentSummaryResponse(documents=results)
217		1✔
218	def _split_documents(self, documents):
219	"""Split into regular docs and images."""
220	docs_urls, docs_handles = [], []
221	img_urls, img_handles = [], []
222
NEW 223	for doc in documents:	×
224	if (
NEW 225	not self.document_provider.config.supports_documents	×
NEW 226	and doc.is_document()	×
NEW 227	):	×
NEW 228	docs_urls.append(doc.url)	×
229	docs_handles.append(doc.handle)
230	else:
NEW 231	img_urls.append(doc.url)	×
NEW 232	img_handles.append(doc.handle)	×
233
234	return (docs_urls, docs_handles), (img_urls, img_handles)
235
NEW 236	def _process_documents(self, docs_data):	×
237	"""Extract text from PDFs/DOCX files."""
NEW 238	urls, handles = docs_data	×
239	results = []
240
241	for url, handle in zip(urls, handles):
242	try:
243	text = extract_text_from_document(url)
244	results.append(DocumentSummaryItem(handle=handle, summary=text))	1✔
245	except Exception as e:
246	logger.error(
247	f"Failed to extract text from {handle}: {e}", exc_info=True	1✔
248	)
249	capture_exception(e)
250
251	return results
252
253	def _process_images(self, images_data, prompt):
254	"""Process images with vision API."""
255	urls, handles = images_data
256
257	if not prompt:
258	prompt = (
259	f"Summarize each image separately. Handles in order: {handles}. "
260	f"Return list of summaries with handles."
261	)
262
263	request = MultimodalSummaryRequest(image_urls=urls, prompt=prompt)
264	response = self.document_provider.request(request, DocumentSummaryResponse)
265	return response.documents
266
267
268	class SummaryRequest(AIRequest):
269	"""Request model for text summarization."""
270
271	DEFAULT_PROMPT = """
272	You are a JSON generator. Return ONLY valid JSON.
273
274	Schema:
275	{
276	"title": "Summary of participation",
277	"general_info": {"summary": "string", "goals": ["string"]},
278	"phases": {
279	"past": {"modules": [{"module_id": "number", "module_name": "string", "status": "past", "final": {"summary": "string", "bullets": ["string"]}, "debug": {...}}]},
280	"current": {"modules": [{"module_id": "number", "module_name": "string", "status": "current", "final": {"summary": "string", "bullets": ["string"]}, "debug": {...}}]},
281	"upcoming": {"modules": [{"module_id": "number", "module_name": "string", "status": "upcoming", "final": {"summary": "string", "bullets": ["string"]}, "debug": {...}}]}
282	}
283	}
284
285	NOTE: module_id in the output should match the given module_id of the input for each module
286
287	Each module MUST include a 'debug' object with:
288	- module_type: string	1✔
289	- signals_snapshot: list of strings	×
290	- draft_before_qa: string	×
291	- claims: list of {claim_text, evidence_type(from_votes\|from_ratings\|from_open_answers\|from_comments\|from_base_text\|uncertain), action(keep\|soften\|replace\|remove), fix_hint}	×
292	- quantifier_fixes: list of {original_phrase, replacement, reason}
293	- anchors: list of strings	1✔
294	- coverage_gaps: list of strings	×
295	- coverage_patch: optional string
296	- patches: list of {patch_type(REPLACE\|REMOVE\|ADD_SENTENCE), target, replacement}
297	- after_qa: string
298	- diff_summary: optional string
299	- qa_status: PASS\|FAIL
300		1✔
301	Extract real data from the project export.
302	Respond with ONLY the JSON object.
303	"""	1✔
304
305	def __init__(self, text: str, prompt: str \| None = None):	1✔
306	super().__init__()
307	self.text = text
308	self.prompt_text = prompt or self.DEFAULT_PROMPT
309
310	def prompt(self) -> str:
311	return f"{self.prompt_text}\n\n{self.text}"	1✔
312
313
314	class MultimodalSummaryRequest(AIRequest):
315	"""Request model for multimodal document summarization."""
316
NEW 317	vision_support = True	×
NEW 318	PROMPT = "Summarize this image/document. Return JSON with summary field."	×
NEW 319		×
320	def __init__(	×
321	self, image_urls: list[str], text: str \| None = None, prompt: str \| None = None
322	):	1✔
323	super().__init__()	×
324	self.image_urls = image_urls	×
325	self.prompt_text = prompt or self.PROMPT	×
326	self.text = text
327
328	def prompt(self) -> str:	1✔
329	base = self.prompt_text
330	return f"{base}\n\nText:\n{self.text}" if self.text else base
331		1✔
332
333	class DocumentRequest(AIRequest):	1✔
334	"""Request model for document summarization."""
335
336	vision_support = True
337	PROMPT = "Summarize this document. Return JSON with summary field."
338
339	def __init__(self, url: str, prompt: str \| None = None):	1✔
340	super().__init__()
341	self.image_urls = [url]
342	self.prompt_text = prompt or self.PROMPT
343
NEW 344	def prompt(self) -> str:	×
NEW 345	return self.prompt_text	×

liqd / roots / 22072536647

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous