export to csv/html/xml method

Author: gkowalc

atlassian/confluence.py

@@ -2657,6 +2657,122 @@ def get_page_as_word(self, page_id):
         url = "exportword?pageId={pageId}".format(pageId=page_id)
         return self.get(url, headers=headers, not_json_response=True)
 
+
+    def export_space_pdf(self, url):
+        try:
+            running_task = True
+            headers = self.form_token_headers
+            log.info("Initiate PDF export from Confluence Cloud")
+            response = self.session.post(url, headers=headers)
+            print(response.text)
+            response_string = response.decode(encoding="utf-8", errors="ignore")
+            task_id = response_string.split('name="ajs-taskId" content="')[1].split('">')[0]
+            poll_url = "/services/api/v1/task/{0}/progress".format(task_id)
+            while running_task:
+                log.info("Check if export task has completed.")
+                progress_response = self.get(poll_url)
+                print(progress_response)
+                percentage_complete = int(progress_response.get("progress", 0))
+                task_state = progress_response.get("state")
+                if task_state == "FAILED":
+                    log.error("PDF conversion not successful.")
+                    return None
+                elif percentage_complete == 100:
+                    running_task = False
+                    log.info("Task completed - {task_state}".format(task_state=task_state))
+                    log.debug("Extract task results to download PDF.")
+                    task_result_url = progress_response.get("result")
+                else:
+                    log.info(
+                        "{percentage_complete}% - {task_state}".format(
+                            percentage_complete=percentage_complete, task_state=task_state
+                        )
+                    )
+                    time.sleep(3)
+            log.debug("Task successfully done, querying the task result for the download url")
+            # task result url starts with /wiki, remove it.
+            task_content = self.get(task_result_url[5:], not_json_response=True)
+            download_url = task_content.decode(encoding="utf-8", errors="strict")
+            log.debug("Successfully got the download url")
+            return download_url
+        except IndexError as e:
+            log.error(e)
+            return None
+    def get_space_export(self, space_key: str, export_type: str) -> str:
+        def get_atl_request(url):
+            # this is only applicable to html/csv/xml export
+            response = self.get(url, advanced_mode=True)
+            parsed_html = BeautifulSoup(response.text, "html.parser")
+            atl_token = parsed_html.find("input", {"name": "atl_token"}).get("value")
+            return atl_token
+        try:
+            running_task = True
+            headers = self.form_token_headers
+            print("Initiate PDF export from Confluence Cloud")
+            log.info("Initiate PDF export from Confluence Cloud")
+            form_data = {}
+            url = ''
+            if export_type == "csv":
+                form_data = {
+                    "atl_token": get_atl_request(f"spaces/exportspacecsv.action?key={space_key}"),
+                    "exportType": "TYPE_CSV",
+                    "contentOption": "all",
+                    "includeComments": "true",
+                    "confirm": "Export"
+                }
+            elif export_type == "html":
+                form_data = {
+                    "atl_token": get_atl_request(f"spaces/exportspacehtml.action?key={space_key}"),
+                    "exportType": "TYPE_HTML",
+                    "contentOption": "visibleOnly",
+                    "includeComments": True,
+                    "confirm": "Export"
+                }
+            elif export_type == "xml":
+                form_data = {
+                    "atl_token": get_atl_request(f"spaces/exportspacexml.action?key={space_key}"),
+                    "exportType": "TYPE_XML",
+                    "contentOption": "all",
+                    "includeComments": "true",
+                    "confirm": "Export" }
+            elif export_type == "pdf":
+
+                form_data = {
+                 #   "atl_token": get_atl_request(f"spaces/flyingpdf/flyingpdf.action?key={space_key}"),
+                    "synchronous": "false",
+                    "contentOption": "visibleOnly",
+                    "confirm": "Export"
+                }
+            else:
+                raise ValueError("Invalid export type")
+            url = f"/spaces/exportspace.action?key={space_key}"
+            # bypass self.confluence_client.post method because it serializes form data as JSON which is wrong
+            if export_type == "pdf":
+
+                url = self.url_joiner(url=self.url,
+                                      path=f"spaces/flyingpdf/doflyingpdf.action?key={space_key}")
+            elif export_type == "csv" or export_type == "html" or export_type == "xml":
+                url = self.url_joiner(url=self.url, path=f"spaces/doexportspace.action?key={space_key}")
+            response = self.session.post(url, headers=self.form_token_headers,
+                                                           data=form_data)
+
+            parsed_html = BeautifulSoup(response.text, "html.parser")
+            print(parsed_html)
+            poll_url = parsed_html.find("meta", {"name": "ajs-pollURI"}).get("content")
+            running_task = True
+            while running_task:
+                progress_response = self.get(poll_url)
+                if progress_response['complete']:
+                    parsed_html = BeautifulSoup(progress_response['message'], "html.parser")
+                    download_url = parsed_html.find("a", {"class": "space-export-download-path"}).get("href")
+                    return self.url.replace('/wiki', '') + download_url
+                time.sleep(5)
+            return None
+        except Exception as e:
+            print(e)
+            return None
+
+
     def export_page(self, page_id):
         """
         Alias method for export page as pdf