from django.contrib import messages from django.contrib.auth.decorators import login_required from django.http import FileResponse, Http404, HttpResponseForbidden, JsonResponse from django.shortcuts import get_object_or_404, redirect, render from django.urls import reverse from accounts.models import User from reviews.models import ( Article, DataExtraction, DataExtractionField, DataExtractionLookup, Hypothesis, Invite, Keyword, QualityAnswer, QualityAssessment, QualityQuestion, Question, Review, SearchSession, SelectionCriteria, Source, ) from rsl.decorators import author_required, main_author_required, phase_required from collections import Counter from reviews.models import ( Article, DataExtractionField, QualityAssessment, SelectionCriteria, ) from rsl.services.articles import delete_articles_by_source from rsl.services.bibtex import import_bibtex_to_review from rsl.services.duplicates import find_duplicate_groups, resolve_all_duplicates from rsl.services.exports import export_articles_xlsx from rsl.decorators import author_required, is_ajax_request, phase_required from rsl.services.search_sessions import ( copy_base_to_sources, generate_and_save_base, get_search_overview, save_base_string, save_source_string, ) from rsl.services.workflow import get_conducting_steps, screening_funnel, steps_for_template @login_required @author_required def index(request, review): return render( request, "conducting/index.html", { "review": review, "steps": steps_for_template(get_conducting_steps(review), review.id), }, ) @login_required @author_required def search(request, review): """Redireciona para o protocolo unificado (bases + strings na mesma tela).""" review.ensure_search_sessions() protocol_url = reverse("planning:protocol", kwargs={"review_id": review.id}) + "#strings" if request.method == "GET": return redirect(protocol_url) if request.method == "POST": action = request.POST.get("action") if action == "save_base": save_base_string(review, request.POST.get("base_search_string", "")) messages.success(request, "String de busca geral salva.") elif action == "generate_base": generate_and_save_base(review) messages.success(request, "String geral gerada a partir das palavras-chave.") elif action == "copy_base_all": n = copy_base_to_sources(review) if n: messages.success(request, f"String geral copiada para {n} base(s) de dados.") else: messages.warning(request, "Defina a string geral antes de copiar.") elif action == "copy_base_one": try: sid = int(request.POST.get("source_id")) except (TypeError, ValueError): sid = None if sid and copy_base_to_sources(review, source_ids=[sid]): messages.success(request, "String geral copiada para esta base.") elif action == "save_sources": for source in review.sources.all(): save_source_string( review, source.id, request.POST.get(f"search_{source.id}", "") ) messages.success(request, "Strings por base de dados salvas.") elif action == "save_source": try: sid = int(request.POST.get("source_id")) except (TypeError, ValueError): sid = None if sid and save_source_string(review, sid, request.POST.get("search_string", "")): messages.success(request, "String da base salva.") return redirect(protocol_url) return redirect(protocol_url) @login_required @author_required @phase_required("import") def import_studies(request, review): sources = list(review.sources.order_by("name")) overview = get_search_overview(review) if request.method == "POST": action = request.POST.get("action") if action == "import_bibtex": source_id = request.POST.get("source_id") source = next((s for s in sources if str(s.id) == source_id), None) content = request.POST.get("bibtex_content", "") if not source: messages.warning(request, "Selecione a base de dados de origem.") elif not content.strip(): messages.warning(request, "Cole o conteúdo BibTeX ou envie um arquivo.") else: imported, skipped = import_bibtex_to_review(review, source, content, request.user) msg = f"{imported} artigo(s) importado(s) de «{source.name}»." if skipped: msg += f" {skipped} duplicado(s) ignorado(s)." messages.success(request, msg) elif action == "import_file": source_id = request.POST.get("source_id") source = next((s for s in sources if str(s.id) == source_id), None) file = request.FILES.get("bibtex_file") if source and file and file.filename: content = file.read().decode("utf-8", errors="ignore") imported, skipped = import_bibtex_to_review(review, source, content, request.user) msg = f"{imported} artigo(s) importado(s) de «{source.name}» (arquivo)." if skipped: msg += f" {skipped} duplicado(s) ignorado(s)." messages.success(request, msg) else: messages.warning(request, "Selecione a base de dados e um arquivo .bib.") elif action == "add_manual": source_id = request.POST.get("source_id") source = next((s for s in sources if str(s.id) == source_id), None) title = (request.POST.get("title") or "").strip() if not source: messages.warning(request, "Selecione a base de dados de origem.") elif not title: messages.warning(request, "Informe o título do artigo.") else: article = Article( review_id=review.id, source_id=source.id, title=title, authors=request.POST.get("authors", ""), year=request.POST.get("year", ""), journal=request.POST.get("journal", ""), doi=request.POST.get("doi", ""), abstract=request.POST.get("abstract", ""), created_by_id=request.user.id, updated_by_id=request.user.id, ) article.save() messages.success(request, f"Artigo adicionado à base «{source.name}».") elif action == "delete_by_source": try: source_id = int(request.POST.get("source_id")) except (TypeError, ValueError): source_id = None source = next((s for s in sources if s.id == source_id), None) if not source: messages.warning(request, "Base de dados inválida.") else: removed = delete_articles_by_source(review, source.id) if removed: messages.success( request, f"{removed} artigo(s) da base «{source.name}» foram excluídos.", ) else: messages.info(request, f"Nenhum artigo importado em «{source.name}».") return redirect(reverse("conducting:import_studies", kwargs={"review_id": review.id})) return render( request, "conducting/import.html", { "review": review, "sources": sources, "overview": overview, }, ) def _selection_redirect(review_id: int, status_filter: str, source_filter: str): kwargs: dict = {"review_id": review_id} if status_filter: kwargs["status"] = status_filter if source_filter: kwargs["source_id"] = source_filter return redirect(reverse("conducting:selection")) @login_required @author_required @phase_required("selection") def selection(request, review): criteria = SelectionCriteria.objects.filter(review_id=review.id).all() status_filter = request.GET.get("status", "") source_filter = request.GET.get("source_id", "") linked_source_ids = set(review.sources.values_list("id", flat=True)) query = Article.objects.filter(review_id=review.id) if status_filter: query = query.filter(status=status_filter) if source_filter: if source_filter == "none": query = query.filter(source_id__isnull=True) else: try: sid = int(source_filter) if sid in linked_source_ids: query = query.filter(source_id=sid) else: source_filter = "" except (TypeError, ValueError): source_filter = "" articles = query.order_by("title").all() duplicate_groups = find_duplicate_groups(review) source_stats = get_search_overview(review)["sources"] sources_filter = [] for item in source_stats: q = Article.objects.filter(review_id=review.id, source_id=item["source"].id) if status_filter: q = q.filter(status=status_filter) sources_filter.append( { "source": item["source"], "count": q.count(), } ) no_source_count = Article.objects.filter( review_id=review.id, source_id__isnull=True ) if status_filter: no_source_count = no_source_count.filter(status=status_filter) no_source_count = no_source_count.count() if request.method == "POST": action = request.POST.get("action") if action == "update_status": article = Article.objects.filter( id=request.POST.get("article_id"), review_id=review.id ).first() if article: article.status = request.POST.get("status", "U") criteria_id = request.POST.get("selection_criteria_id") article.selection_criteria_id = int(criteria_id) if criteria_id else None article.comments = request.POST.get("comments", "") article.updated_by_id = request.user.id messages.success(request, "Artigo atualizado.") elif action == "resolve_duplicates": count = resolve_all_duplicates(review) messages.success(request, f"Marcados {count} como duplicados.") elif action == "bulk_status": ids = request.POST.getlist("article_ids") new_status = request.POST.get("bulk_status", "U") for aid in ids: article = Article.objects.filter(id=aid, review_id=review.id).first() if article: article.status = new_status article.updated_by_id = request.user.id messages.success(request, f"Atualizados {len(ids)} artigos.") return _selection_redirect(review.id, status_filter, source_filter) status_choices = [ ("", "Todos"), ("U", "Não classificados"), ("A", "Aceitos"), ("R", "Rejeitados"), ("D", "Duplicados"), ] status_labels = {k: v for k, v in status_choices if k} return render( request, "conducting/selection.html", { "review": review, "articles": articles, "criteria": criteria, "status_filter": status_filter, "source_filter": source_filter, "sources_filter": sources_filter, "no_source_count": no_source_count, "duplicate_groups": duplicate_groups, "source_stats": source_stats, "status_choices": status_choices, "status_labels": status_labels, }, ) def _save_article_quality(article, questions, form) -> bool: """Persiste respostas do checklist para um artigo. Retorna False se faltar resposta.""" for question in questions: answer_id = form.get(f"q_{question.id}") if not answer_id: return False assessment = QualityAssessment.objects.filter( article_id=article.id, question_id=question.id ).first() if not assessment: assessment = QualityAssessment( article_id=article.id, question_id=question.id, user_id=request.user.id ) assessment.save() assessment.answer_id = int(answer_id) assessment.user_id = request.user.id return True @login_required @author_required @phase_required("qa") def quality_assessment(request, review): from rsl.services.quality_score import article_passes_cutoff, article_quality_complete articles = Article.objects.filter(review_id=review.id, status="A").order_by("title").all() questions = list(review.quality_questions.all()) answers = list(review.quality_answers.all()) n_questions = len(questions) max_weight = max((a.weight for a in answers), default=1.0) max_score = round(n_questions * max_weight, 1) if n_questions else 0.0 cutoff = review.quality_assessment_cutoff_score or 0.0 if request.method == "POST": article_id = request.POST.get("article_id") article = Article.objects.filter(id=article_id, review_id=review.id).first() if article: if _save_article_quality(article, questions, request.POST): score = article.quality_score() if cutoff > 0 and score >= cutoff: messages.success( request, f"Avaliação salva — pontuação {score:.1f} (aprovado, corte {cutoff:.1f}).", ) elif cutoff > 0: messages.warning( request, f"Avaliação salva — pontuação {score:.1f} (abaixo do corte {cutoff:.1f}; " "não entra na síntese final).", ) else: messages.success(request, "Avaliação de qualidade salva.") if request.POST.get("next") == "1": for candidate in articles: if not article_quality_complete(candidate, review): return redirect( reverse( "conducting:quality_assessment", kwargs={"review_id": review.id}, ) + f"#art-{candidate.id}" ) else: messages.error(request, "Responda todas as perguntas antes de salvar.") next_filter = request.GET.get("filter", "all") return redirect( reverse("conducting:quality_assessment", kwargs={"review_id": review.id}) + (f"?filter={next_filter}" if next_filter != "all" else "") ) filter_mode = request.GET.get("filter", "pending") if filter_mode not in ("all", "pending", "approved", "below"): filter_mode = "pending" matrix = {} article_rows = [] n_complete = 0 n_approved = 0 n_below = 0 for article in articles: matrix[article.id] = { a.question_id: a.answer_id for a in article.quality_assessments.all() } complete = article_quality_complete(article, review) score = article.quality_score() if complete else None passes = article_passes_cutoff(article, review) if complete else None if complete: n_complete += 1 if passes: n_approved += 1 elif cutoff > 0: n_below += 1 row = { "article": article, "complete": complete, "score": score, "passes": passes, } if filter_mode == "pending" and complete: continue if filter_mode == "approved" and not (complete and passes): continue if filter_mode == "below" and not (complete and passes is False): continue article_rows.append(row) # Pendentes primeiro na lista article_rows.sort(key=lambda r: (r["complete"], r["article"].title.lower())) stats = { "total": len(articles), "complete": n_complete, "pending": len(articles) - n_complete, "approved": n_approved, "below": n_below, "max_score": max_score, "cutoff": cutoff, } return render( request, "conducting/quality.html", { "review": review, "article_rows": article_rows, "articles": articles, "questions": questions, "answers": answers, "matrix": matrix, "stats": stats, "filter_mode": filter_mode, "n_questions": n_questions, }, ) def _extraction_autosave_response(review, fields): """Salva um artigo via POST AJAX e retorna JSON.""" from rsl.services.extraction_save import save_article_extraction article_id = request.POST.get("article_id") article = Article.objects.filter( id=article_id, review_id=review.id, status="A" ).first() if not article: return JsonResponse(ok=False, error="Artigo não encontrado."), 404 ok, err, meta = save_article_extraction( article, fields, request.user.id, request.POST ) if not ok: return JsonResponse(ok=False, error=err), 400 return JsonResponse(ok=True, message="Salvo automaticamente", **meta) @login_required @author_required @phase_required("extraction") def extraction_autosave(request, review): fields = DataExtractionField.objects.filter(review_id=review.id).order_by("order").all() return _extraction_autosave_response(review, fields) @login_required @author_required @phase_required("extraction") def data_extraction(request, review): articles = ( Article.objects.filter(review_id=review.id, status="A") .order_by("title") .all() ) from rsl.services.extraction_form import ensure_extraction_fields added_fields = ensure_extraction_fields(review) if added_fields: messages.info( request, f"Formulário atualizado: {added_fields} campo(s) adicionado(s) (incl. Descrição).", ) fields = DataExtractionField.objects.filter(review_id=review.id).order_by("order").all() if request.method == "POST": action = request.POST.get("action") if action == "prefill_metadata": from rsl.services.extraction_prefill import prefill_review_metadata pending_only = request.POST.get("pending_only") == "1" stats = prefill_review_metadata( review, request.user.id, only_if_empty=True, pending_only=pending_only, ) messages.success( request, f"Metadados preenchidos em {stats['articles_updated']} artigo(s) " f"({stats['fields_updated']} campo(s): título, autores, ano, base).", ) return redirect(reverse("conducting:data_extraction", kwargs={"review_id": review.id})) if action == "prefill_article": from rsl.services.extraction_prefill import prefill_article_metadata article = Article.objects.filter( id=request.POST.get("article_id"), review_id=review.id ).first() if article: n = prefill_article_metadata( article, request.user.id, only_if_empty=True, pending_only=False ) messages.success(request, f"{n} campo(s) preenchido(s) a partir do artigo.") return redirect( reverse("conducting:data_extraction", kwargs={"review_id": review.id}) + f"#article-{article.id if article else ''}" ) if action == "add_lookup": from rsl.services.extraction_form import add_extraction_lookup is_ajax = request.headers.get("X-Requested-With") == "XMLHttpRequest" try: field_id = int(request.POST.get("field_id") or 0) except (TypeError, ValueError): field_id = 0 lookup, err = add_extraction_lookup( review, field_id, request.POST.get("lookup_value", ""), ) if is_ajax: if err: return JsonResponse(ok=False, error=err), 400 return JsonResponse( ok=True, message=f'Opção "{lookup.value}" adicionada.', lookup={"id": lookup.id, "value": lookup.value}, field_id=lookup.field_id, field_description=lookup.field.description, ) if err: messages.error(request, err) else: messages.success( request, f'Opção "{lookup.value}" adicionada em "{lookup.field.description}". ' "Disponível em todos os artigos.", ) anchor = request.POST.get("article_id", "").strip() url = reverse("conducting:data_extraction", kwargs={"review_id": review.id}) if anchor: url += f"#article-{anchor}" return redirect(url) if action == "add_field": from rsl.services.extraction_form import ( create_extraction_field, parse_lookup_options_text, ) from rsl.services.extraction_api import ( field_to_dict, render_field_blocks_html, ) is_ajax = request.headers.get("X-Requested-With") == "XMLHttpRequest" field, err = create_extraction_field( review, request.POST.get("description", ""), request.POST.get("field_type", "O"), lookup_options=parse_lookup_options_text( request.POST.get("lookup_options", "") ), ) if is_ajax: if err: return JsonResponse(ok=False, error=err), 400 field = DataExtractionField.objects.filter( id=field.id, review_id=review.id ).first() return JsonResponse( ok=True, message=f'Campo "{field.description}" adicionado.', field=field_to_dict(field), html_by_article=render_field_blocks_html(review, articles, field), ) if err: messages.error(request, err) else: messages.success( request, f'Campo "{field.description}" adicionado ao formulário. ' "Ele já aparece em todos os artigos abaixo.", ) anchor = request.POST.get("article_id", "").strip() url = reverse("conducting:data_extraction", kwargs={"review_id": review.id}) if anchor: url += f"#article-{anchor}" return redirect(url) if action == "import_xls": from pathlib import Path from app.config import BASE_DIR from rsl.services.import_data_extraction import import_data_extraction_xls xls_path = BASE_DIR / "arquivos" / "data_extraction.xls" try: stats = import_data_extraction_xls(review, request.user, xls_path) messages.success( request, f"Importação: {stats['matched']} linha(s) de dados, " f"{stats.get('descriptions_saved', 0)} descrição(ões) em português, " f"{stats['extractions_saved']} valores no total, " f"{stats['marked_finished']} concluídos. " f"Não encontrados: {stats['not_found']}.", ) except FileNotFoundError as e: messages.error(request, str(e)) except Exception as e: messages.error(request, f"Erro na importação: {e}") return redirect(reverse("conducting:data_extraction", kwargs={"review_id": review.id})) if action in (None, "", "save", "autosave"): is_ajax = is_ajax_request() or action == "autosave" if is_ajax: return _extraction_autosave_response(review, fields) article_id = request.POST.get("article_id") article = Article.objects.filter( id=article_id, review_id=review.id, status="A" ).first() if article: from rsl.services.extraction_save import save_article_extraction ok, err, meta = save_article_extraction( article, fields, request.user.id, request.POST ) if not ok: messages.error(request, err) else: messages.success(request, "Extração de dados salva.") url = reverse("conducting:data_extraction", kwargs={"review_id": review.id}) url += f"#article-{article.id}" return redirect(url) return redirect(reverse("conducting:data_extraction", kwargs={"review_id": review.id})) sources_in_articles = sorted( {a.source.name for a in articles if a.source}, key=str.lower, ) stats = { "total": len(articles), "finished": sum(1 for a in articles if a.finished_data_extraction), "pending": sum(1 for a in articles if not a.finished_data_extraction), "highlighted": sum(1 for a in articles if a.thesis_highlight), } from reviews.models import DataExtractionField as ExtractionFieldModel from rsl.services.extraction_form import FIELD_TYPE_LABELS config = { "url": reverse("conducting:data_extraction", kwargs={"review_id": review.id}), "autosaveUrl": reverse( "conducting:extraction_autosave", kwargs={"review_id": review.id} ), "reviewId": review.id, "csrfToken": "", } return render( request, "conducting/extraction.html", { "review": review, "articles": articles, "fields": fields, "sources_in_articles": sources_in_articles, "extraction_stats": stats, "field_type_labels": FIELD_TYPE_LABELS, "extraction_field_types": list(ExtractionFieldModel.FIELD_TYPES.items()), "config": config, }, ) @login_required @author_required def analysis(request, review): by_source = Counter() accepted_by_source = Counter() by_year = Counter() for article in review.articles.all(): source_name = article.source.name if article.source else "Sem base" by_source[source_name] += 1 if article.status == "A": accepted_by_source[source_name] += 1 for article in review.get_final_selection_articles(): if article.year: by_year[article.year] += 1 funnel = screening_funnel(review) return render( request, "conducting/analysis.html", { "review": review, "by_source": dict(by_source), "accepted_by_source": dict(accepted_by_source), "by_year": dict(sorted(by_year.items())), "funnel": funnel, }, ) @login_required @author_required def export_articles(request, review): output = export_articles_xlsx(review) filename = f"{review.slug}-artigos.xlsx" return FileResponse(output, as_attachment=True, download_name=filename)