(self, to, inner_path, content_base64, ignore_bad_files=False)
| 551 | |
| 552 | # Write a file to disk |
| 553 | def actionFileWrite(self, to, inner_path, content_base64, ignore_bad_files=False): |
| 554 | valid_signers = self.site.content_manager.getValidSigners(inner_path) |
| 555 | auth_address = self.user.getAuthAddress(self.site.address) |
| 556 | if not self.hasFilePermission(inner_path): |
| 557 | self.log.error("FileWrite forbidden %s not in valid_signers %s" % (auth_address, valid_signers)) |
| 558 | return self.response(to, {"error": "Forbidden, you can only modify your own files"}) |
| 559 | |
| 560 | # Try not to overwrite files currently in sync |
| 561 | content_inner_path = re.sub("^(.*)/.*?$", "\\1/content.json", inner_path) # Also check the content.json from same directory |
| 562 | if (self.site.bad_files.get(inner_path) or self.site.bad_files.get(content_inner_path)) and not ignore_bad_files: |
| 563 | found = self.site.needFile(inner_path, update=True, priority=10) |
| 564 | if not found: |
| 565 | self.cmd( |
| 566 | "confirm", |
| 567 | [_["This file still in sync, if you write it now, then the previous content may be lost."], _["Write content anyway"]], |
| 568 | lambda res: self.actionFileWrite(to, inner_path, content_base64, ignore_bad_files=True) |
| 569 | ) |
| 570 | return False |
| 571 | |
| 572 | try: |
| 573 | import base64 |
| 574 | content = base64.b64decode(content_base64) |
| 575 | # Save old file to generate patch later |
| 576 | if ( |
| 577 | inner_path.endswith(".json") and not inner_path.endswith("content.json") and |
| 578 | self.site.storage.isFile(inner_path) and not self.site.storage.isFile(inner_path + "-old") |
| 579 | ): |
| 580 | try: |
| 581 | self.site.storage.rename(inner_path, inner_path + "-old") |
| 582 | except Exception: |
| 583 | # Rename failed, fall back to standard file write |
| 584 | f_old = self.site.storage.open(inner_path, "rb") |
| 585 | f_new = self.site.storage.open(inner_path + "-old", "wb") |
| 586 | shutil.copyfileobj(f_old, f_new) |
| 587 | |
| 588 | self.site.storage.write(inner_path, content) |
| 589 | except Exception as err: |
| 590 | self.log.error("File write error: %s" % Debug.formatException(err)) |
| 591 | return self.response(to, {"error": "Write error: %s" % Debug.formatException(err)}) |
| 592 | |
| 593 | if inner_path.endswith("content.json"): |
| 594 | self.site.content_manager.loadContent(inner_path, add_bad_files=False, force=True) |
| 595 | |
| 596 | self.response(to, "ok") |
| 597 | |
| 598 | # Send sitechanged to other local users |
| 599 | for ws in self.site.websockets: |
| 600 | if ws != self: |
| 601 | ws.event("siteChanged", self.site, {"event": ["file_done", inner_path]}) |
| 602 | |
| 603 | def actionFileDelete(self, to, inner_path): |
| 604 | if not self.hasFilePermission(inner_path): |
nothing calls this directly
no test coverage detected