Coverage for postrfp/buyer/api/endpoints/reports/msword.py: 100%

1import re 

2import logging 

3from pathlib import Path 

4from itertools import tee 

5 

6from docx import Document 

7from sqlalchemy.orm import subqueryload, Session 

8 

9from postrfp import conf 

10from postrfp.authorisation import perms 

11from postrfp.model.questionnaire.answering import Answer 

12from postrfp.shared import fetch 

13from postrfp.shared.decorators import http 

14from postrfp.buyer.api import authorise 

15from postrfp.model import QuestionInstance, QElement, Project, User, Issue, Section 

16from postrfp.shared.response import XAccelTempResponse, XAccelResponse 

17from postrfp.shared.constants import MimeTypes 

18 

19 

20TEMPLATE_DOCX = "assets/q-template.docx" 

21log = logging.getLogger(__name__) 

22 

23 

24def is_comments_label(label_text): 

25 if label_text is None: 

26 return False 

27 patt = r"(comments|qualifications)" 

28 match = re.search(patt, label_text, re.IGNORECASE) is not None 

29 return match and len(label_text) < 50 

30 

31 

32def render_paragraph_question(doc, question, answers): 

33 elements, following = tee(question.question_def.elements) 

34 next(following, None) 

35 

36 for element in elements: 

37 next_el = next(following, None) 

38 if is_comments_label(element.label): 

39 if next_el.id not in answers: 

40 # Don't print comments / quals label or answer 

41 # if Not Answered 

42 next(elements, None) 

43 next(following, None) 

44 continue 

45 if answers and element.is_answerable: 

46 if element.id in answers: 

47 answer = answers[element.id].answer 

48 else: 

49 answer = "Not Answered" 

50 doc.add_paragraph(answer, style="Answer") 

51 else: 

52 doc.add_paragraph(element.label) 

53 

54 

55def render_table_question(doc, question, answers): 

56 """ 

57 TODO - doesn't handle multi column tables 

58 """ 

59 col_count = question.question_def.column_count 

60 table = doc.add_table(rows=1, cols=col_count) 

61 current_row = -1 

62 row_cells = None 

63 for el in question.question_def.elements: 

64 if el.row > current_row: 

65 row_cells = table.add_row().cells 

66 cell = row_cells[el.col - 1] 

67 if el.colspan > 1: 

68 try: 

69 # fix broken colspans 

70 right_col_index = (el.col - 1) + el.colspan 

71 if right_col_index > (col_count - 1): 

72 right_col_index = col_count - 1 

73 other_cell = row_cells[right_col_index] 

74 cell.merge(other_cell) 

75 except IndexError: # pragma: no cover 

76 log.warning( 

77 "Fix colspan failed for question instance ID %s", question.id 

78 ) 

79 if answers and el.is_answerable: 

80 if el.id in answers: 

81 answer = answers[el.id].answer 

82 else: 

83 answer = "- Not Answered -" 

84 cell.add_paragraph(answer, style="Answer") 

85 else: 

86 cell.add_paragraph(el.label) 

87 current_row = el.row 

88 

89 

90def _el_answer_lookup(project: Project, issue: Issue, user: User, session): 

91 authorise.check( 

92 user, 

93 perms.ISSUE_VIEW_ANSWERS, 

94 project=project, 

95 issue=issue, 

96 section_id=project.section_id, 

97 ) 

98 

99 answer_query = ( 

100 session.query(Answer.element_id, Answer.answer) 

101 .join(QElement) 

102 .join(QuestionInstance) 

103 .filter(Answer.issue == issue, QuestionInstance.project == project) 

104 ) 

105 

106 return {a.element_id: a for a in answer_query} 

107 

108 

109@http 

110def get_project_report_msword( 

111 session: Session, user: User, project_id: int, q_issue_id: int | None = None 

112) -> XAccelResponse: 

113 """ 

114 Generate an MS Word report for the quesionnaire associated with the given Project ID. 

115 If an issue ID is provided the answers are populated for that Issue. 

116 """ 

117 project = fetch.project(session, project_id) 

118 authorise.check(user, perms.PROJECT_ACCESS, project=project, deny_restricted=True) 

119 

120 sections = ( 

121 fetch.sections(project, user).options(subqueryload(Section.questions)).all() 

122 ) 

123 

124 if q_issue_id is not None: 

125 issue = project.get_issue(q_issue_id) 

126 answers = _el_answer_lookup(project, issue, user, session) 

127 fname = "{0}-{1}.docx".format(project.title[:20], issue.respondent.name[:15]) 

128 else: 

129 answers = {} 

130 fname = "{0}.docx".format(project.title[:25]) 

131 

132 doc_path = Path(__file__).parent.joinpath(TEMPLATE_DOCX) 

133 doc = Document(docx=str(doc_path)) 

134 

135 for section in sections: 

136 if section.is_top_level: 

137 doc.add_heading(sections[0].title, 0) 

138 else: 

139 heading_text = "{} {}".format(section.number, section.title) 

140 level = section.number.count(".") + 1 

141 doc.add_heading(heading_text, level) 

142 for question in section.questions: 

143 q_title = "{} {}".format(question.number, question.title) 

144 doc.add_paragraph(q_title, style="QuestionTitle") 

145 if question.question_def.is_tabular(): 

146 render_table_question(doc, question, answers) 

147 else: 

148 render_paragraph_question(doc, question, answers) 

149 

150 cache_file_name, cache_file_path = conf.CONF.random_cache_file_path() 

151 

152 cache_dir = Path(cache_file_path) 

153 cache_dir.parent.mkdir(parents=True, exist_ok=True) 

154 with open(cache_file_path, "wb+") as fp: 

155 doc.save(fp) 

156 

157 return XAccelTempResponse(cache_file_name, fname, content_type=MimeTypes.DOCX)