91 lines
3.8 KiB
Python
91 lines
3.8 KiB
Python
import json
|
|
import mimetypes
|
|
import uuid
|
|
from pathlib import Path
|
|
from urllib import error, request
|
|
|
|
|
|
class WebhookService:
|
|
def _multipart_parts_for_fields(self, boundary: str, fields: dict[str, str]) -> list[bytes]:
|
|
parts: list[bytes] = []
|
|
for key, value in fields.items():
|
|
parts.extend([
|
|
f"--{boundary}\r\n".encode(),
|
|
f'Content-Disposition: form-data; name="{key}"\r\n\r\n'.encode(),
|
|
str(value).encode(),
|
|
b"\r\n",
|
|
])
|
|
return parts
|
|
|
|
def _multipart_parts_for_file(self, boundary: str, file_field: str, file_path: str) -> list[bytes]:
|
|
path = Path(file_path)
|
|
mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
|
|
return [
|
|
f"--{boundary}\r\n".encode(),
|
|
f'Content-Disposition: form-data; name="{file_field}"; filename="{path.name}"\r\n'.encode(),
|
|
f"Content-Type: {mime}\r\n\r\n".encode(),
|
|
path.read_bytes(),
|
|
b"\r\n",
|
|
]
|
|
|
|
def _multipart_body(self, fields: dict[str, str], file_field: str, file_path: str) -> tuple[bytes, str]:
|
|
boundary = f"----DorisSchoolhouse{uuid.uuid4().hex}"
|
|
parts = self._multipart_parts_for_fields(boundary, fields)
|
|
parts.extend(self._multipart_parts_for_file(boundary, file_field, file_path))
|
|
parts.append(f"--{boundary}--\r\n".encode())
|
|
return b"".join(parts), boundary
|
|
|
|
def _multipart_body_many(self, fields: dict[str, str], file_field: str, file_paths: list[str]) -> tuple[bytes, str]:
|
|
boundary = f"----DorisSchoolhouse{uuid.uuid4().hex}"
|
|
parts = self._multipart_parts_for_fields(boundary, fields)
|
|
for file_path in file_paths:
|
|
parts.extend(self._multipart_parts_for_file(boundary, file_field, file_path))
|
|
parts.append(f"--{boundary}--\r\n".encode())
|
|
return b"".join(parts), boundary
|
|
|
|
def post_multipart(self, url: str, fields: dict[str, str], file_field: str, file_path: str) -> dict:
|
|
body, boundary = self._multipart_body(fields, file_field, file_path)
|
|
req = request.Request(
|
|
url,
|
|
data=body,
|
|
headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
|
|
method="POST",
|
|
)
|
|
|
|
def parse_raw(raw: str):
|
|
try:
|
|
return json.loads(raw)
|
|
except Exception:
|
|
return {"raw": raw}
|
|
|
|
try:
|
|
with request.urlopen(req, timeout=120) as response:
|
|
raw = response.read().decode("utf-8", errors="replace")
|
|
return {"ok": True, "status_code": response.status, "body": parse_raw(raw)}
|
|
except error.HTTPError as exc:
|
|
raw = exc.read().decode("utf-8", errors="replace")
|
|
return {"ok": False, "status_code": exc.code, "body": parse_raw(raw), "error": str(exc)}
|
|
|
|
def post_multipart_many(self, url: str, fields: dict[str, str], file_field: str, file_paths: list[str]) -> dict:
|
|
body, boundary = self._multipart_body_many(fields, file_field, file_paths)
|
|
req = request.Request(
|
|
url,
|
|
data=body,
|
|
headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
|
|
method="POST",
|
|
)
|
|
|
|
def parse_raw(raw: str):
|
|
try:
|
|
return json.loads(raw)
|
|
except Exception:
|
|
return {"raw": raw}
|
|
|
|
try:
|
|
with request.urlopen(req, timeout=120) as response:
|
|
raw = response.read().decode("utf-8", errors="replace")
|
|
return {"ok": True, "status_code": response.status, "body": parse_raw(raw)}
|
|
except error.HTTPError as exc:
|
|
raw = exc.read().decode("utf-8", errors="replace")
|
|
return {"ok": False, "status_code": exc.code, "body": parse_raw(raw), "error": str(exc)}
|