Refactor update changelog

pr_agent/git_providers/bitbucket_provider.py

@@ -342,6 +342,41 @@ class BitbucketProvider(GitProvider):
     def _get_pr(self):
         return self._get_repo().pullrequests.get(self.pr_num)
 
+    def get_pr_file_content(self, file_path: str, branch: str) -> str:
+        try:
+            if branch == self.pr.source_branch:
+                branch = self.pr.data["source"]["commit"]["hash"]
+            elif branch == self.pr.destination_branch:
+                branch = self.pr.data["destination"]["commit"]["hash"]
+            url = (f"https://api.bitbucket.org/2.0/repositories/{self.workspace_slug}/{self.repo_slug}/src/"
+                   f"{branch}/{file_path}")
+            response = requests.request("GET", url, headers=self.headers)
+            if response.status_code == 404:  # not found
+                return ""
+            contents = response.text
+            return contents
+        except Exception:
+            return ""
+
+
+    def create_or_update_pr_file(self, file_path: str, branch: str, contents="", message="") -> None:
+        url = (f"https://api.bitbucket.org/2.0/repositories/{self.workspace_slug}/{self.repo_slug}/src/")
+        if not message:
+            if contents:
+                message = f"Update {file_path}"
+            else:
+                message = f"Create {file_path}"
+        files={file_path: contents}
+        data={
+            "message": message,
+            "branch": branch
+        }
+        headers = {'Authorization':self.headers['Authorization']} if 'Authorization' in self.headers else {}
+        try:
+            requests.request("POST", url, headers=headers, data=data, files=files)
+        except Exception:
+            get_logger().exception(f"Failed to create empty file {file_path} in branch {branch}")
+
     def _get_pr_file_content(self, remote_link: str):
         return ""
 

pr_agent/git_providers/github_provider.py

@@ -603,13 +603,36 @@ class GithubProvider(GitProvider):
     def _get_pr(self):
         return self._get_repo().get_pull(self.pr_num)
 
-    def _get_pr_file_content(self, file: FilePatchInfo, sha: str) -> str:
+    def get_pr_file_content(self, file_path: str, branch: str) -> str:
         try:
-            file_content_str = str(self._get_repo().get_contents(file.filename, ref=sha).decoded_content.decode())
+            file_content_str = str(
+                self._get_repo()
+                .get_contents(file_path, ref=branch)
+                .decoded_content.decode()
+            )
         except Exception:
             file_content_str = ""
         return file_content_str
 
+    def create_or_update_pr_file(
+        self, file_path: str, branch: str, contents="", message=""
+    ) -> None:
+        try:
+            file_obj = self._get_repo().get_contents(file_path, ref=branch)
+            sha1=file_obj.sha
+        except Exception:
+            sha1=""
+        self.repo_obj.update_file(
+            path=file_path,
+            message=message,
+            content=contents,
+            sha=sha1,
+            branch=branch,
+        )
+
+    def _get_pr_file_content(self, file: FilePatchInfo, sha: str) -> str:
+        return self.get_pr_file_content(file.filename, sha)
+
     def publish_labels(self, pr_types):
         try:
             label_color_map = {"Bug fix": "1d76db", "Tests": "e99695", "Bug fix with tests": "c5def5",

pr_agent/git_providers/gitlab_provider.py

@@ -63,7 +63,7 @@ class GitLabProvider(GitProvider):
             raise DiffNotFoundError(f"Could not get diff for merge request {self.id_mr}") from e
 
 
-    def _get_pr_file_content(self, file_path: str, branch: str) -> str:
+    def get_pr_file_content(self, file_path: str, branch: str) -> str:
         try:
             return self.gl.projects.get(self.id_project).files.get(file_path, branch).decode()
         except GitlabGetError:
@@ -88,10 +88,8 @@ class GitLabProvider(GitProvider):
         diff_files = []
         for diff in diffs:
             if is_valid_file(diff['new_path']):
-                # original_file_content_str = self._get_pr_file_content(diff['old_path'], self.mr.target_branch)
-                # new_file_content_str = self._get_pr_file_content(diff['new_path'], self.mr.source_branch)
-                original_file_content_str = self._get_pr_file_content(diff['old_path'], self.mr.diff_refs['base_sha'])
-                new_file_content_str = self._get_pr_file_content(diff['new_path'], self.mr.diff_refs['head_sha'])
+                original_file_content_str = self.get_pr_file_content(diff['old_path'], self.mr.diff_refs['base_sha'])
+                new_file_content_str = self.get_pr_file_content(diff['new_path'], self.mr.diff_refs['head_sha'])
 
                 try:
                     if isinstance(original_file_content_str, bytes):