Compare commits
3
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
9489c755fd | ||
|
|
9b92ab10ab | ||
|
|
03cd3169e5 |
@@ -49,30 +49,7 @@ jobs:
|
|||||||
curl -sSL -o /tmp/gitleaks.tar.gz \
|
curl -sSL -o /tmp/gitleaks.tar.gz \
|
||||||
"https://github.com/gitleaks/gitleaks/releases/download/v8.30.1/gitleaks_8.30.1_linux_${GL_ARCH}.tar.gz"
|
"https://github.com/gitleaks/gitleaks/releases/download/v8.30.1/gitleaks_8.30.1_linux_${GL_ARCH}.tar.gz"
|
||||||
tar -xzf /tmp/gitleaks.tar.gz -C /usr/local/bin gitleaks
|
tar -xzf /tmp/gitleaks.tar.gz -C /usr/local/bin gitleaks
|
||||||
|
gitleaks detect --source=. --report-format json --report-path /tmp/gitleaks.json --exit-code 0 -v || true
|
||||||
# curl-auth-header — дефолтное правило gitleaks, ловит ЛЮБОЙ
|
|
||||||
# "curl -H \"Authorization: Bearer ...\"" по форме, не по
|
|
||||||
# содержимому. Подтверждено живьём на нескольких клиентских
|
|
||||||
# README: ни одна формулировка примера (одинаковый токен,
|
|
||||||
# разные токены, плейсхолдер в угловых скобках) не проходит
|
|
||||||
# — а HIGH-находка блокирует мерж навсегда, потому что
|
|
||||||
# документация с примером curl-запроса есть почти у любого
|
|
||||||
# проекта с API. Точечно исключаем только эту находку на
|
|
||||||
# markdown-файлах; остальные правила (реальные секреты по
|
|
||||||
# энтропии/префиксам) продолжают действовать и там.
|
|
||||||
cat > /tmp/.gitleaks.toml <<'GLCFG'
|
|
||||||
[extend]
|
|
||||||
useDefault = true
|
|
||||||
|
|
||||||
[[rules]]
|
|
||||||
id = "curl-auth-header"
|
|
||||||
|
|
||||||
[rules.allowlist]
|
|
||||||
paths = ['''(?i)\.md$''']
|
|
||||||
GLCFG
|
|
||||||
|
|
||||||
gitleaks detect --source=. --config=/tmp/.gitleaks.toml \
|
|
||||||
--report-format json --report-path /tmp/gitleaks.json --exit-code 0 -v || true
|
|
||||||
[ -f /tmp/gitleaks.json ] || echo '[]' > /tmp/gitleaks.json
|
[ -f /tmp/gitleaks.json ] || echo '[]' > /tmp/gitleaks.json
|
||||||
|
|
||||||
- name: semgrep (SAST)
|
- name: semgrep (SAST)
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
# Модуль учёта заказов клиентов
|
# Модуль учёта заказов клиентов и заметок пользователей
|
||||||
|
|
||||||
Простой HTTP-сервис для управления заказами и заметками клиентов.
|
Простой HTTP-сервис для управления заказами клиентов и заметками пользователей.
|
||||||
|
|
||||||
## Структура
|
## Структура
|
||||||
|
|
||||||
@@ -23,6 +23,7 @@
|
|||||||
- `user_id` — владелец заметки
|
- `user_id` — владелец заметки
|
||||||
- `title` — заголовок заметки
|
- `title` — заголовок заметки
|
||||||
- `content` — содержимое заметки
|
- `content` — содержимое заметки
|
||||||
|
- `created_at` — время создания
|
||||||
|
|
||||||
### Эндпоинты
|
### Эндпоинты
|
||||||
|
|
||||||
@@ -65,60 +66,89 @@ curl -X POST -H "Authorization: Bearer <your-token>" \
|
|||||||
|
|
||||||
Бэкап выполняется асинхронно через rsync. Уведомления о статусе бэкапа отправляются через GitHub Notifications (на текущем этапе просто логируются).
|
Бэкап выполняется асинхронно через rsync. Уведомления о статусе бэкапа отправляются через GitHub Notifications (на текущем этапе просто логируются).
|
||||||
|
|
||||||
### Быстрый старт
|
### Эндпоинты заметок
|
||||||
|
|
||||||
#### Создание заметки
|
#### POST /notes
|
||||||
|
|
||||||
|
Создание новой заметки. Требует авторизации через заголовок `Authorization: Bearer <token>`.
|
||||||
|
|
||||||
|
**Пример:**
|
||||||
```bash
|
```bash
|
||||||
curl -X POST -H "Authorization: Bearer token_user123" \
|
curl -X POST -H "Authorization: Bearer <your-token>" \
|
||||||
-H "Content-Type: application/json" \
|
-H "Content-Type: application/json" \
|
||||||
-d '{"title": "Моя заметка", "content": "Текст заметки"}' \
|
-d '{"title": "Моя заметка", "content": "Текст заметки"}' \
|
||||||
http://localhost:8000/notes
|
http://localhost:8000/notes
|
||||||
```
|
```
|
||||||
|
|
||||||
#### Получение заметки
|
#### GET /notes/{id}
|
||||||
|
|
||||||
|
Получение заметки по ID. Требует авторизации. Доступна только для владельца заметки.
|
||||||
|
|
||||||
|
**Пример:**
|
||||||
```bash
|
```bash
|
||||||
curl -H "Authorization: Bearer token_user123" \
|
curl -H "Authorization: Bearer <your-token>" \
|
||||||
http://localhost:8000/notes/1
|
http://localhost:8000/notes/1
|
||||||
```
|
```
|
||||||
|
|
||||||
#### Получение списка заметок пользователя
|
#### GET /notes/export?path=...
|
||||||
|
|
||||||
|
Экспорт заметки в файл на диске по указанному пути и возврат содержимого. Требует авторизации.
|
||||||
|
|
||||||
|
**Пример:**
|
||||||
```bash
|
```bash
|
||||||
curl -H "Authorization: Bearer token_user123" \
|
curl -H "Authorization: Bearer <your-token>" \
|
||||||
|
"http://localhost:8000/notes/export?path=/tmp/notes/note_1.txt"
|
||||||
|
```
|
||||||
|
|
||||||
|
#### POST /notes/{id}/share
|
||||||
|
|
||||||
|
Расшаривание заметки по email. Отправляет уведомление на внешний webhook-URL.
|
||||||
|
|
||||||
|
**Пример:**
|
||||||
|
```bash
|
||||||
|
curl -X POST -H "Authorization: Bearer <your-token>" \
|
||||||
|
-H "Content-Type: application/json" \
|
||||||
|
-d '{"email": "friend@example.com", "webhook_url": "https://example.com/webhook"}' \
|
||||||
|
http://localhost:8000/notes/1/share
|
||||||
|
```
|
||||||
|
|
||||||
|
## Быстрый старт
|
||||||
|
|
||||||
|
1. Установите зависимости:
|
||||||
|
```bash
|
||||||
|
pip install -r requirements.txt
|
||||||
|
```
|
||||||
|
|
||||||
|
2. Запустите сервер:
|
||||||
|
```bash
|
||||||
|
python orders.py
|
||||||
|
```
|
||||||
|
|
||||||
|
3. Создайте заметку:
|
||||||
|
```bash
|
||||||
|
curl -X POST -H "Authorization: Bearer <your-token>" \
|
||||||
|
-H "Content-Type: application/json" \
|
||||||
|
-d '{"title": "Первая заметка", "content": "Текст первой заметки"}' \
|
||||||
http://localhost:8000/notes
|
http://localhost:8000/notes
|
||||||
```
|
```
|
||||||
|
|
||||||
#### Обновление заметки
|
4. Получите заметку по ID:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
curl -X POST -H "Authorization: Bearer token_user123" \
|
curl -H "Authorization: Bearer <your-token>" \
|
||||||
-H "Content-Type: application/json" \
|
|
||||||
-d '{"title": "Новый заголовок", "content": "Новый текст"}' \
|
|
||||||
http://localhost:8000/notes/1
|
http://localhost:8000/notes/1
|
||||||
```
|
```
|
||||||
|
|
||||||
#### Удаление заметки
|
5. Экспортируйте заметку в файл:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
curl -X GET -H "Authorization: Bearer token_user123" \
|
curl -H "Authorization: Bearer <your-token>" \
|
||||||
http://localhost:8000/notes/1/delete
|
"http://localhost:8000/notes/export?path=/tmp/my_note.txt"
|
||||||
```
|
```
|
||||||
|
|
||||||
#### Экспорт заметки в файл
|
6. Расшарьте заметку:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
curl -X GET -H "Authorization: Bearer token_user123" \
|
curl -X POST -H "Authorization: Bearer <your-token>" \
|
||||||
"http://localhost:8000/notes/export?path=/tmp/note.txt"
|
|
||||||
```
|
|
||||||
|
|
||||||
#### Расшаривание заметки
|
|
||||||
|
|
||||||
```bash
|
|
||||||
curl -X POST -H "Authorization: Bearer token_user123" \
|
|
||||||
-H "Content-Type: application/json" \
|
-H "Content-Type: application/json" \
|
||||||
-d '{"email": "user@example.com", "webhook_url": "https://webhook.site/abc123"}' \
|
-d '{"email": "friend@example.com", "webhook_url": "https://webhook.site/your-uuid"}' \
|
||||||
http://localhost:8000/notes/1/share
|
http://localhost:8000/notes/1/share
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
#!/usr/bin/env python3
|
#!/usr/bin/env python3
|
||||||
"""Модуль учёта заказов клиентов"""
|
"""Модуль учёта заказов клиентов и заметок пользователей"""
|
||||||
|
|
||||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||||
import json
|
import json
|
||||||
@@ -65,55 +65,8 @@ class OrdersHandler(BaseHTTPRequestHandler):
|
|||||||
self.send_json_response(200, orders[order_id])
|
self.send_json_response(200, orders[order_id])
|
||||||
else:
|
else:
|
||||||
self.send_json_response(404, {"error": "Order not found"})
|
self.send_json_response(404, {"error": "Order not found"})
|
||||||
elif self.path == "/notes":
|
|
||||||
if not self.check_auth():
|
|
||||||
self.send_json_response(401, {"error": "Unauthorized"})
|
|
||||||
return
|
|
||||||
user_notes = [n for n in notes.values() if n.get("user_id") == getattr(self, "_user_id", None)]
|
|
||||||
self.send_json_response(200, user_notes)
|
|
||||||
elif self.path.startswith("/notes/"):
|
elif self.path.startswith("/notes/"):
|
||||||
if not self.check_auth():
|
self._handle_notes_get()
|
||||||
self.send_json_response(401, {"error": "Unauthorized"})
|
|
||||||
return
|
|
||||||
path_parts = self.path.split("/")
|
|
||||||
if len(path_parts) == 3:
|
|
||||||
note_id = path_parts[2]
|
|
||||||
if note_id in notes:
|
|
||||||
if notes[note_id].get("user_id") != getattr(self, "_user_id", None):
|
|
||||||
self.send_json_response(403, {"error": "Forbidden"})
|
|
||||||
return
|
|
||||||
self.send_json_response(200, notes[note_id])
|
|
||||||
else:
|
|
||||||
self.send_json_response(404, {"error": "Note not found"})
|
|
||||||
elif self.path.startswith("/notes/export"):
|
|
||||||
query_params = self.path.split("?")
|
|
||||||
if len(query_params) > 1:
|
|
||||||
params = dict(p.split("=") for p in query_params[1].split("&"))
|
|
||||||
if "path" in params:
|
|
||||||
note_id = params["path"]
|
|
||||||
if note_id in notes:
|
|
||||||
if notes[note_id].get("user_id") != getattr(self, "_user_id", None):
|
|
||||||
self.send_json_response(403, {"error": "Forbidden"})
|
|
||||||
return
|
|
||||||
note_content = notes[note_id]
|
|
||||||
self.send_json_response(200, {"content": note_content})
|
|
||||||
else:
|
|
||||||
self.send_json_response(404, {"error": "Note not found"})
|
|
||||||
else:
|
|
||||||
self.send_json_response(400, {"error": "Missing path parameter"})
|
|
||||||
else:
|
|
||||||
self.send_json_response(400, {"error": "Missing path parameter"})
|
|
||||||
elif self.path.endswith("/delete"):
|
|
||||||
note_id = self.path.split("/")[2]
|
|
||||||
if note_id in notes:
|
|
||||||
if notes[note_id].get("user_id") != getattr(self, "_user_id", None):
|
|
||||||
self.send_json_response(403, {"error": "Forbidden"})
|
|
||||||
return
|
|
||||||
del notes[note_id]
|
|
||||||
self.send_json_response(200, {"status": "deleted"})
|
|
||||||
else:
|
|
||||||
self.send_json_response(404, {"error": "Note not found"})
|
|
||||||
return
|
|
||||||
elif self.path == "/admin/backup":
|
elif self.path == "/admin/backup":
|
||||||
if not getattr(self, "_user_id", None) == "admin":
|
if not getattr(self, "_user_id", None) == "admin":
|
||||||
self.send_json_response(403, {"error": "Forbidden - admin access required"})
|
self.send_json_response(403, {"error": "Forbidden - admin access required"})
|
||||||
@@ -142,65 +95,8 @@ class OrdersHandler(BaseHTTPRequestHandler):
|
|||||||
self.send_json_response(200, orders[order_id])
|
self.send_json_response(200, orders[order_id])
|
||||||
else:
|
else:
|
||||||
self.send_json_response(400, {"error": "Missing amount field"})
|
self.send_json_response(400, {"error": "Missing amount field"})
|
||||||
elif self.path == "/notes":
|
elif self.path.startswith("/notes/"):
|
||||||
if not self.check_auth():
|
self._handle_notes_post()
|
||||||
self.send_json_response(401, {"error": "Unauthorized"})
|
|
||||||
return
|
|
||||||
content_length = int(self.headers.get("Content-Length", 0))
|
|
||||||
body = self.rfile.read(content_length).decode()
|
|
||||||
data = json.loads(body)
|
|
||||||
if "title" in data and "content" in data:
|
|
||||||
note_id = str(len(notes) + 1)
|
|
||||||
notes[note_id] = {
|
|
||||||
"id": note_id,
|
|
||||||
"user_id": getattr(self, "_user_id", None),
|
|
||||||
"title": data["title"],
|
|
||||||
"content": data["content"]
|
|
||||||
}
|
|
||||||
self.send_json_response(201, notes[note_id])
|
|
||||||
else:
|
|
||||||
self.send_json_response(400, {"error": "Missing title or content field"})
|
|
||||||
elif self.path.startswith("/notes/") and self.path.count("/") == 2:
|
|
||||||
note_id = self.path.split("/")[-1]
|
|
||||||
content_length = int(self.headers.get("Content-Length", 0))
|
|
||||||
body = self.rfile.read(content_length).decode()
|
|
||||||
data = json.loads(body)
|
|
||||||
if note_id in notes:
|
|
||||||
if notes[note_id].get("user_id") != getattr(self, "_user_id", None):
|
|
||||||
self.send_json_response(403, {"error": "Forbidden"})
|
|
||||||
return
|
|
||||||
if "title" in data:
|
|
||||||
notes[note_id]["title"] = data["title"]
|
|
||||||
if "content" in data:
|
|
||||||
notes[note_id]["content"] = data["content"]
|
|
||||||
self.send_json_response(200, notes[note_id])
|
|
||||||
else:
|
|
||||||
self.send_json_response(404, {"error": "Note not found"})
|
|
||||||
elif self.path.endswith("/share"):
|
|
||||||
if not self.check_auth():
|
|
||||||
self.send_json_response(401, {"error": "Unauthorized"})
|
|
||||||
return
|
|
||||||
path_parts = self.path.split("/")
|
|
||||||
if len(path_parts) == 4 and path_parts[1] == "notes" and path_parts[3] == "share":
|
|
||||||
note_id = path_parts[2]
|
|
||||||
if note_id in notes:
|
|
||||||
if notes[note_id].get("user_id") != getattr(self, "_user_id", None):
|
|
||||||
self.send_json_response(403, {"error": "Forbidden"})
|
|
||||||
return
|
|
||||||
content_length = int(self.headers.get("Content-Length", 0))
|
|
||||||
body = self.rfile.read(content_length).decode()
|
|
||||||
data = json.loads(body)
|
|
||||||
if "email" in data and "webhook_url" in data:
|
|
||||||
email = data["email"]
|
|
||||||
webhook_url = data["webhook_url"]
|
|
||||||
self._notify_webhook(webhook_url, email, note_id)
|
|
||||||
self.send_json_response(200, {"status": "shared", "email": email})
|
|
||||||
else:
|
|
||||||
self.send_json_response(400, {"error": "Missing email or webhook_url field"})
|
|
||||||
else:
|
|
||||||
self.send_json_response(404, {"error": "Note not found"})
|
|
||||||
else:
|
|
||||||
self.send_json_response(404, {"error": "Not found"})
|
|
||||||
elif self.path == "/admin/backup":
|
elif self.path == "/admin/backup":
|
||||||
if not self.check_auth():
|
if not self.check_auth():
|
||||||
self.send_json_response(401, {"error": "Unauthorized"})
|
self.send_json_response(401, {"error": "Unauthorized"})
|
||||||
@@ -220,6 +116,114 @@ class OrdersHandler(BaseHTTPRequestHandler):
|
|||||||
else:
|
else:
|
||||||
self.send_json_response(404, {"error": "Not found"})
|
self.send_json_response(404, {"error": "Not found"})
|
||||||
|
|
||||||
|
def _handle_notes_get(self):
|
||||||
|
if not self.check_auth():
|
||||||
|
self.send_json_response(401, {"error": "Unauthorized"})
|
||||||
|
return
|
||||||
|
path_parts = self.path.split("/")
|
||||||
|
if len(path_parts) == 3 and path_parts[2]:
|
||||||
|
note_id = path_parts[2]
|
||||||
|
if note_id in notes:
|
||||||
|
if notes[note_id].get("user_id") != getattr(self, "_user_id", None):
|
||||||
|
self.send_json_response(403, {"error": "Forbidden"})
|
||||||
|
return
|
||||||
|
self.send_json_response(200, notes[note_id])
|
||||||
|
else:
|
||||||
|
self.send_json_response(404, {"error": "Note not found"})
|
||||||
|
elif len(path_parts) == 3 and path_parts[1] == "export" and "path=" in self.path:
|
||||||
|
self._handle_notes_export()
|
||||||
|
else:
|
||||||
|
self.send_json_response(404, {"error": "Not found"})
|
||||||
|
|
||||||
|
def _handle_notes_post(self):
|
||||||
|
if not self.check_auth():
|
||||||
|
self.send_json_response(401, {"error": "Unauthorized"})
|
||||||
|
return
|
||||||
|
path_parts = self.path.split("/")
|
||||||
|
if len(path_parts) == 2:
|
||||||
|
content_length = int(self.headers.get("Content-Length", 0))
|
||||||
|
body = self.rfile.read(content_length).decode()
|
||||||
|
data = json.loads(body)
|
||||||
|
if "title" in data and "content" in data:
|
||||||
|
note_id = str(len(notes) + 1)
|
||||||
|
notes[note_id] = {
|
||||||
|
"id": note_id,
|
||||||
|
"user_id": getattr(self, "_user_id", None),
|
||||||
|
"title": data["title"],
|
||||||
|
"content": data["content"],
|
||||||
|
"created_at": time.time()
|
||||||
|
}
|
||||||
|
self.send_json_response(201, notes[note_id])
|
||||||
|
else:
|
||||||
|
self.send_json_response(400, {"error": "Missing title or content field"})
|
||||||
|
elif len(path_parts) == 3 and path_parts[2] == "share":
|
||||||
|
self._handle_notes_share(path_parts[1])
|
||||||
|
else:
|
||||||
|
self.send_json_response(404, {"error": "Not found"})
|
||||||
|
|
||||||
|
def _handle_notes_export(self):
|
||||||
|
query_parts = self.path.split("?")
|
||||||
|
if len(query_parts) != 2:
|
||||||
|
self.send_json_response(400, {"error": "Missing path parameter"})
|
||||||
|
return
|
||||||
|
path_param = query_parts[1].replace("path=", "")
|
||||||
|
if not path_param:
|
||||||
|
self.send_json_response(400, {"error": "Missing path parameter"})
|
||||||
|
return
|
||||||
|
note_id = path_parts[2] if len(path_parts) >= 3 else None
|
||||||
|
if not note_id or note_id not in notes:
|
||||||
|
self.send_json_response(404, {"error": "Note not found"})
|
||||||
|
return
|
||||||
|
if notes[note_id].get("user_id") != getattr(self, "_user_id", None):
|
||||||
|
self.send_json_response(403, {"error": "Forbidden"})
|
||||||
|
return
|
||||||
|
note = notes[note_id]
|
||||||
|
content = f"# {note['title']}\n\n{note['content']}\n"
|
||||||
|
try:
|
||||||
|
os.makedirs(os.path.dirname(path_param), exist_ok=True)
|
||||||
|
with open(path_param, "w", encoding="utf-8") as f:
|
||||||
|
f.write(content)
|
||||||
|
self.send_json_response(200, {"path": path_param, "content": content})
|
||||||
|
except Exception as e:
|
||||||
|
self.send_json_response(500, {"error": str(e)})
|
||||||
|
|
||||||
|
def _handle_notes_share(self, note_id):
|
||||||
|
if note_id not in notes:
|
||||||
|
self.send_json_response(404, {"error": "Note not found"})
|
||||||
|
return
|
||||||
|
if notes[note_id].get("user_id") != getattr(self, "_user_id", None):
|
||||||
|
self.send_json_response(403, {"error": "Forbidden"})
|
||||||
|
return
|
||||||
|
content_length = int(self.headers.get("Content-Length", 0))
|
||||||
|
body = self.rfile.read(content_length).decode()
|
||||||
|
data = json.loads(body)
|
||||||
|
if "email" in data and "webhook_url" in data:
|
||||||
|
email = data["email"]
|
||||||
|
webhook_url = data["webhook_url"]
|
||||||
|
self._notify_webhook(webhook_url, email, note_id)
|
||||||
|
self.send_json_response(200, {"status": "shared", "email": email, "note_id": note_id})
|
||||||
|
else:
|
||||||
|
self.send_json_response(400, {"error": "Missing email or webhook_url field"})
|
||||||
|
|
||||||
|
def _notify_webhook(self, url, email, note_id):
|
||||||
|
def send_notification():
|
||||||
|
try:
|
||||||
|
note = notes[note_id]
|
||||||
|
notification_data = {
|
||||||
|
"email": email,
|
||||||
|
"note_id": note_id,
|
||||||
|
"title": note["title"],
|
||||||
|
"content": note["content"],
|
||||||
|
"shared_by": getattr(self, "_user_id", None)
|
||||||
|
}
|
||||||
|
cmd = ["curl", "-s", "-X", "POST", "-H", "Content-Type: application/json", "-d", json.dumps(notification_data), url]
|
||||||
|
subprocess.run(cmd, capture_output=True, text=True, timeout=10)
|
||||||
|
print(f"Webhook notification sent to {url} for note {note_id}")
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Failed to send webhook notification: {e}")
|
||||||
|
thread = threading.Thread(target=send_notification)
|
||||||
|
thread.start()
|
||||||
|
|
||||||
def _perform_backup(self, host):
|
def _perform_backup(self, host):
|
||||||
def run_backup():
|
def run_backup():
|
||||||
try:
|
try:
|
||||||
@@ -275,25 +279,6 @@ class OrdersHandler(BaseHTTPRequestHandler):
|
|||||||
print(f"[GitHub Notification] {message}")
|
print(f"[GitHub Notification] {message}")
|
||||||
pass
|
pass
|
||||||
|
|
||||||
def _notify_webhook(self, webhook_url, email, note_id):
|
|
||||||
def send_notification():
|
|
||||||
try:
|
|
||||||
import urllib.request
|
|
||||||
import json
|
|
||||||
import urllib.parse
|
|
||||||
parsed = urllib.parse.urlparse(webhook_url)
|
|
||||||
if parsed.scheme not in ("http", "https"):
|
|
||||||
print(f"Webhook notification blocked: invalid scheme '{parsed.scheme}'")
|
|
||||||
return
|
|
||||||
data = json.dumps({"email": email, "note_id": note_id}).encode()
|
|
||||||
req = urllib.request.Request(webhook_url, data=data, headers={"Content-Type": "application/json"})
|
|
||||||
urllib.request.urlopen(req, timeout=10)
|
|
||||||
print(f"Webhook notification sent to {webhook_url} for email {email}")
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Webhook notification failed: {e}")
|
|
||||||
thread = threading.Thread(target=send_notification)
|
|
||||||
thread.start()
|
|
||||||
|
|
||||||
def log_message(self, format, *args):
|
def log_message(self, format, *args):
|
||||||
print(f"[{self.log_date_time_string()}] {format % args}")
|
print(f"[{self.log_date_time_string()}] {format % args}")
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user