feat: add delete avatar

src/infrastructure/storage/s3_service.py

@@ -24,6 +24,52 @@ class S3Service:
         self._endpoint_url = endpoint_url.strip().rstrip('/') if endpoint_url and endpoint_url.strip() else None
         self._use_reg_ru_website_public_host = use_reg_ru_website_public_host
 
+    @staticmethod
+    def _url_prefix_variants(prefix: str) -> list[str]:
+        p = prefix.rstrip('/') + '/'
+        out = [p]
+        if p.startswith('https://'):
+            out.append('http://' + p[8:])
+        elif p.startswith('http://'):
+            out.append('https://' + p[7:])
+        return out
+
+    def _public_url_prefixes(self) -> list[str]:
+        acc: list[str] = []
+        pb = self._public_base_url
+        if pb:
+            acc.extend(self._url_prefix_variants(pb))
+        ep = self._endpoint_url
+        if ep:
+            base = f'{ep.rstrip("/")}/{self._bucket}'
+            acc.extend(self._url_prefix_variants(base))
+        if ep and self._use_reg_ru_website_public_host and 's3.regru.cloud' in ep.lower():
+            wh = f'https://{self._bucket}.website.regru.cloud'
+            acc.extend(self._url_prefix_variants(wh))
+        if not ep:
+            if self._region == 'us-east-1':
+                h = f'https://{self._bucket}.s3.amazonaws.com'
+            else:
+                h = f'https://{self._bucket}.s3.{self._region}.amazonaws.com'
+            acc.extend(self._url_prefix_variants(h))
+        seen: set[str] = set()
+        uniq: list[str] = []
+        for x in sorted(acc, key=len, reverse=True):
+            if x not in seen:
+                seen.add(x)
+                uniq.append(x)
+        return uniq
+
+    def object_key_from_public_url(self, url: str) -> str | None:
+        u = (url or '').strip()
+        if not u:
+            return None
+        for p in self._public_url_prefixes():
+            if u.startswith(p):
+                k = u[len(p):].split('?', 1)[0].split('#', 1)[0]
+                return k if k else None
+        return None
+
     def _object_url(self, key: str) -> str:
         if self._public_base_url:
             return f'{self._public_base_url}/{key}'
@@ -62,3 +108,18 @@ class S3Service:
                 ContentType=content_type,
             )
         return self._object_url(key)
+
+    async def delete_object(self, *, key: str) -> None:
+        session = get_session()
+        kw: dict[str, object] = {'region_name': self._region}
+        aid = self._access_key_id
+        sk = self._secret_access_key
+        ep = self._endpoint_url
+        if aid:
+            kw['aws_access_key_id'] = aid
+        if sk:
+            kw['aws_secret_access_key'] = sk
+        if ep:
+            kw['endpoint_url'] = ep
+        async with session.create_client('s3', **kw) as client:
+            await client.delete_object(Bucket=self._bucket, Key=key)