Lab3 создание реп-я и issue через API Gitea
This commit is contained in:
commit
cbcd878182
12
.gitignore
vendored
Normal file
12
.gitignore
vendored
Normal file
@ -0,0 +1,12 @@
|
||||
pycache/
|
||||
*.pyc
|
||||
*.pyo
|
||||
.env
|
||||
network_env/
|
||||
.DS_Store
|
||||
*.log
|
||||
*.tmp
|
||||
test_*
|
||||
venv/
|
||||
.env.local
|
||||
*.env
|
||||
23
README.md
Normal file
23
README.md
Normal file
@ -0,0 +1,23 @@
|
||||
# Лабораторная работа 3 — Сетевые соединения в Python
|
||||
|
||||
## Подготовительный этап
|
||||
- Создано виртуальное окружение `.venv`
|
||||
- Установлена библиотека `requests`
|
||||
- Инициализирован Git-репозиторий
|
||||
- Добавлены `.gitignore` и `README.md`
|
||||
|
||||
## Состав проекта
|
||||
- `tcp_server.py` — TCP сервер
|
||||
- `tcp_client.py` — TCP клиент
|
||||
- `udp_server.py` — UDP сервер
|
||||
- `udp_client.py` — UDP клиент
|
||||
- `http_client.py` — простой HTTP-запрос через socket
|
||||
- `http_requests.py` — HTTP-запросы через библиотеку requests
|
||||
- `gitea_api.py` — работа с API Gitea
|
||||
|
||||
## Инструменты
|
||||
- Python 3.x
|
||||
- requests
|
||||
- VS Code с расширением Python
|
||||
- Git
|
||||
- Wireshark
|
||||
26
create-repo.py
Normal file
26
create-repo.py
Normal file
@ -0,0 +1,26 @@
|
||||
# create_repo.py
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
import requests
|
||||
|
||||
load_dotenv()
|
||||
TOKEN = os.getenv("GITEA_WRITE_TOKEN")
|
||||
if not TOKEN:
|
||||
raise SystemExit("Нет токена! Установите GITEA_WRITE_TOKEN в .env")
|
||||
|
||||
headers = {
|
||||
"Authorization": f"token {TOKEN}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
|
||||
data = {
|
||||
"name": "lab3-lusnikov",
|
||||
"description": "Репозиторий создан через Gitea API",
|
||||
"private": False
|
||||
}
|
||||
|
||||
url = "https://git.vyatsu.ru/api/v1/user/repos"
|
||||
resp = requests.post(url, headers=headers, json=data)
|
||||
|
||||
print(resp.status_code, resp.json())
|
||||
24
create_issue.py
Normal file
24
create_issue.py
Normal file
@ -0,0 +1,24 @@
|
||||
# create_issue.py
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
import requests
|
||||
|
||||
load_dotenv()
|
||||
TOKEN = os.getenv("GITEA_WRITE_TOKEN")
|
||||
if not TOKEN:
|
||||
raise SystemExit("Нет токена! Установите GITEA_WRITE_TOKEN в .env")
|
||||
|
||||
headers = {"Authorization": f"token {TOKEN}"}
|
||||
|
||||
owner = "stud178862"
|
||||
repo = "lab3-lusnikov"
|
||||
|
||||
data = {
|
||||
"title": "Issue из Python API",
|
||||
"body": "Эта задача создана автоматически через API."
|
||||
}
|
||||
|
||||
url = f"https://git.vyatsu.ru/api/v1/repos/{owner}/{repo}/issues"
|
||||
resp = requests.post(url, headers=headers, json=data)
|
||||
|
||||
print(resp.status_code, resp.json())
|
||||
165
gitea_api.py
Normal file
165
gitea_api.py
Normal file
@ -0,0 +1,165 @@
|
||||
import requests
|
||||
import os
|
||||
from dotenv import load_dotenv
|
||||
|
||||
# Загружаем переменные окружения из файла .env
|
||||
load_dotenv()
|
||||
|
||||
class GiteaAPI:
|
||||
def __init__(self):
|
||||
self.base_url = "https://git.vyatsu.ru/api/v1"
|
||||
self.token = os.getenv('GITEA_TOKEN')
|
||||
|
||||
if not self.token:
|
||||
print("❌ Ошибка: GITEA_TOKEN не найден в переменных окружения")
|
||||
print(" Убедитесь, что файл .env существует и содержит GITEA_TOKEN=your_token")
|
||||
self.headers = None
|
||||
return
|
||||
|
||||
self.headers = {
|
||||
"Authorization": f"token {self.token}",
|
||||
"Content-Type": "application/json"
|
||||
}
|
||||
print("✅ Gitea API клиент инициализирован")
|
||||
|
||||
def get_user_info(self):
|
||||
"""Получение информации о текущем пользователе"""
|
||||
if not self.headers:
|
||||
return None
|
||||
|
||||
print("\n👤 Получаем информацию о пользователе...")
|
||||
try:
|
||||
response = requests.get(
|
||||
f"{self.base_url}/user",
|
||||
headers=self.headers,
|
||||
timeout=10
|
||||
)
|
||||
response.raise_for_status()
|
||||
user_data = response.json()
|
||||
|
||||
print("✅ Информация о пользователе получена:")
|
||||
print(f" • ID: {user_data.get('id')}")
|
||||
print(f" • Логин: {user_data.get('login')}")
|
||||
print(f" • Имя: {user_data.get('full_name', 'Не указано')}")
|
||||
print(f" • Email: {user_data.get('email', 'Не указан')}")
|
||||
print(f" • Админ: {'Да' if user_data.get('is_admin') else 'Нет'}")
|
||||
|
||||
return user_data
|
||||
|
||||
except requests.RequestException as e:
|
||||
print(f"❌ Ошибка при получении информации о пользователе: {e}")
|
||||
return None
|
||||
|
||||
def list_user_repos(self):
|
||||
"""Получение списка репозиториев пользователя"""
|
||||
if not self.headers:
|
||||
return None
|
||||
|
||||
print("\n📚 Получаем список репозиториев...")
|
||||
try:
|
||||
response = requests.get(
|
||||
f"{self.base_url}/user/repos",
|
||||
headers=self.headers,
|
||||
timeout=10
|
||||
)
|
||||
response.raise_for_status()
|
||||
repos = response.json()
|
||||
|
||||
print(f"✅ Найдено репозиториев: {len(repos)}")
|
||||
for i, repo in enumerate(repos[:5], 1): # Показываем первые 5
|
||||
print(f" {i}. {repo['name']}")
|
||||
print(f" Описание: {repo.get('description', 'Нет описания')}")
|
||||
print(f" URL: {repo.get('html_url')}")
|
||||
print(f" Приватный: {'Да' if repo.get('private') else 'Нет'}")
|
||||
print()
|
||||
|
||||
if len(repos) > 5:
|
||||
print(f" ... и еще {len(repos) - 5} репозиториев")
|
||||
|
||||
return repos
|
||||
|
||||
except requests.RequestException as e:
|
||||
print(f"❌ Ошибка при получении списка репозиториев: {e}")
|
||||
return None
|
||||
|
||||
def create_repository(self, repo_name, description=""):
|
||||
"""Создание нового репозитория"""
|
||||
if not self.headers:
|
||||
return None
|
||||
|
||||
print(f"\n🆕 Создаем репозиторий '{repo_name}'...")
|
||||
|
||||
data = {
|
||||
"name": repo_name,
|
||||
"description": description,
|
||||
"auto_init": True, # Создать README автоматически
|
||||
"private": False, # Публичный репозиторий
|
||||
"readme": "Default" # Использовать README по умолчанию
|
||||
}
|
||||
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{self.base_url}/user/repos",
|
||||
headers=self.headers,
|
||||
json=data,
|
||||
timeout=10
|
||||
)
|
||||
|
||||
if response.status_code == 201:
|
||||
repo_data = response.json()
|
||||
print(f"✅ Репозиторий '{repo_name}' успешно создан!")
|
||||
print(f" URL: {repo_data.get('html_url')}")
|
||||
print(f" SSH: {repo_data.get('ssh_url')}")
|
||||
return repo_data
|
||||
else:
|
||||
print(f"❌ Ошибка при создании репозитория: {response.status_code}")
|
||||
error_detail = response.json()
|
||||
print(f" Сообщение: {error_detail.get('message', 'Неизвестная ошибка')}")
|
||||
return None
|
||||
|
||||
except requests.RequestException as e:
|
||||
print(f"❌ Ошибка при создании репозитория: {e}")
|
||||
return None
|
||||
|
||||
def main():
|
||||
# Проверяем наличие токена
|
||||
if not os.getenv('GITEA_TOKEN'):
|
||||
print("❌ Ошибка: GITEA_TOKEN не найден в переменных окружения")
|
||||
print("\n📝 Инструкция по настройке:")
|
||||
print("1. Создайте файл .env в корне проекта")
|
||||
print("2. Добавьте в него строку: GITEA_TOKEN=your_token_here")
|
||||
print("3. Замените your_token_here на реальный токен из Gitea")
|
||||
print("4. Убедитесь, что .env добавлен в .gitignore")
|
||||
return
|
||||
|
||||
# Создаем экземпляр API клиента
|
||||
gitea = GiteaAPI()
|
||||
|
||||
# Получаем информацию о пользователе
|
||||
user_info = gitea.get_user_info()
|
||||
if not user_info:
|
||||
return
|
||||
|
||||
# Получаем список репозиториев
|
||||
repos = gitea.list_user_repos()
|
||||
|
||||
# Создаем репозиторий для лабораторной работы
|
||||
new_repo_name = f"network-programming-lab3-{user_info.get('login')}"
|
||||
new_repo_description = "Третья лабораторная работа по сетевым соединениям в Python"
|
||||
|
||||
print("\n" + "="*50)
|
||||
create_repo = input("Создать новый репозиторий для лабораторной работы? (y/n): ")
|
||||
|
||||
if create_repo.lower() == 'y':
|
||||
new_repo = gitea.create_repository(new_repo_name, new_repo_description)
|
||||
|
||||
if new_repo:
|
||||
print(f"\n🎉 Репозиторий для лабораторной работы создан!")
|
||||
print(f" Вы можете перейти по ссылке: {new_repo.get('html_url')}")
|
||||
else:
|
||||
print("\n⚠️ Не удалось создать репозиторий. Возможно, он уже существует.")
|
||||
else:
|
||||
print("Создание репозитория пропущено.")
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
30
http_requests.py
Normal file
30
http_requests.py
Normal file
@ -0,0 +1,30 @@
|
||||
import requests
|
||||
|
||||
def http_via_requests():
|
||||
try:
|
||||
print("🔄 Отправляем HTTP-запрос к vyatsu.ru...")
|
||||
response = requests.get("http://vyatsu.ru", timeout=10)
|
||||
|
||||
print("✅ Ответ получен!")
|
||||
print("\n" + "="*50)
|
||||
print("ИНФОРМАЦИЯ О ЗАПРОСЕ:")
|
||||
print("="*50)
|
||||
|
||||
print(f"📊 Статус код: {response.status_code}")
|
||||
print(f"🔗 URL: {response.url}")
|
||||
print(f"📏 Размер ответа: {len(response.text)} символов")
|
||||
|
||||
print(f"\n📋 ЗАГОЛОВКИ ОТВЕТА:")
|
||||
for header, value in response.headers.items():
|
||||
print(f" {header}: {value}")
|
||||
|
||||
print(f"\n📄 СОДЕРЖИМОЕ (первые 500 символов):")
|
||||
print(response.text[:500])
|
||||
if len(response.text) > 500:
|
||||
print("... (содержимое обрезано)")
|
||||
|
||||
except requests.RequestException as e:
|
||||
print(f"❌ Ошибка запроса: {e}")
|
||||
|
||||
if __name__ == "__main__":
|
||||
http_via_requests()
|
||||
64
http_socket.py
Normal file
64
http_socket.py
Normal file
@ -0,0 +1,64 @@
|
||||
import socket
|
||||
|
||||
def http_via_socket():
|
||||
# Создаем TCP сокет
|
||||
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
|
||||
try:
|
||||
print("🔄 Устанавливаем соединение с vyatsu.ru...")
|
||||
client.connect(('vyatsu.ru', 80))
|
||||
print("✅ Соединение установлено")
|
||||
|
||||
# Формируем HTTP-запрос
|
||||
request = (
|
||||
"GET / HTTP/1.1\r\n"
|
||||
"Host: vyatsu.ru\r\n"
|
||||
"User-Agent: Python-Socket-Client/1.0\r\n"
|
||||
"Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8\r\n"
|
||||
"Accept-Language: ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7\r\n"
|
||||
"Connection: close\r\n"
|
||||
"\r\n"
|
||||
)
|
||||
|
||||
print("📤 Отправляем HTTP-запрос...")
|
||||
client.sendall(request.encode())
|
||||
|
||||
print("📨 Получаем ответ...")
|
||||
response = b""
|
||||
while True:
|
||||
chunk = client.recv(4096)
|
||||
if not chunk:
|
||||
break
|
||||
response += chunk
|
||||
|
||||
# Декодируем ответ
|
||||
decoded_response = response.decode('utf-8', errors='ignore')
|
||||
|
||||
print("\n" + "="*50)
|
||||
print("ОТВЕТ ОТ СЕРВЕРА:")
|
||||
print("="*50)
|
||||
|
||||
# Отделяем заголовки от тела
|
||||
headers_end = decoded_response.find('\r\n\r\n')
|
||||
if headers_end != -1:
|
||||
headers = decoded_response[:headers_end]
|
||||
body_start = headers_end + 4
|
||||
body_preview = decoded_response[body_start:body_start + 500]
|
||||
|
||||
print("📋 ЗАГОЛОВКИ:")
|
||||
print(headers)
|
||||
print(f"\n📄 ТЕЛО (первые 500 символов):")
|
||||
print(body_preview)
|
||||
if len(decoded_response) > body_start + 500:
|
||||
print("... (ответ обрезан)")
|
||||
else:
|
||||
print(decoded_response[:1000])
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Ошибка: {e}")
|
||||
finally:
|
||||
client.close()
|
||||
print("\n🔒 Соединение закрыто")
|
||||
|
||||
if __name__ == "__main__":
|
||||
http_via_socket()
|
||||
38
tcp_client.py
Normal file
38
tcp_client.py
Normal file
@ -0,0 +1,38 @@
|
||||
import socket
|
||||
|
||||
def run_tcp_client():
|
||||
# Создаем TCP сокет
|
||||
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
|
||||
try:
|
||||
# Подключаемся к серверу
|
||||
client.connect(('127.0.0.1', 10000))
|
||||
print("✅ Подключение к серверу установлено")
|
||||
print("Введите сообщения для отправки на сервер")
|
||||
print("Для выхода введите 'exit'")
|
||||
|
||||
while True:
|
||||
# Получаем сообщение от пользователя
|
||||
message = input("💬 Введите сообщение: ")
|
||||
# Отправляем сообщение серверу
|
||||
client.sendall(message.encode())
|
||||
|
||||
# Если пользователь ввел exit - выходим
|
||||
if message.lower() == 'exit':
|
||||
break
|
||||
|
||||
# Получаем ответ от сервера
|
||||
data = client.recv(1024)
|
||||
response = data.decode()
|
||||
print(f"📨 Ответ от сервера: {response}")
|
||||
|
||||
except ConnectionRefusedError:
|
||||
print("❌ Не удалось подключиться к серверу. Убедитесь, что сервер запущен.")
|
||||
except Exception as e:
|
||||
print(f"❌ Ошибка: {e}")
|
||||
finally:
|
||||
client.close()
|
||||
print("🔒 Соединение закрыто")
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_tcp_client()
|
||||
49
tcp_server.py
Normal file
49
tcp_server.py
Normal file
@ -0,0 +1,49 @@
|
||||
import socket
|
||||
|
||||
def run_tcp_server():
|
||||
# Создаем TCP сокет
|
||||
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
# Разрешаем повторное использование адреса
|
||||
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
# Привязываем сокет к адресу и порту
|
||||
server.bind(("0.0.0.0", 10000))
|
||||
# Начинаем прослушивание (максимум 1 подключение в очереди)
|
||||
server.listen(1)
|
||||
print("🚀 TCP сервер запущен на порту 10000")
|
||||
print("Ожидание подключений...")
|
||||
|
||||
try:
|
||||
while True:
|
||||
# Принимаем подключение
|
||||
conn, addr = server.accept()
|
||||
print(f"✅ Подключение установлено от {addr}")
|
||||
|
||||
with conn:
|
||||
while True:
|
||||
# Получаем данные от клиента
|
||||
data = conn.recv(1024)
|
||||
if not data:
|
||||
break
|
||||
|
||||
message = data.decode().strip()
|
||||
print(f"📨 Получено сообщение: '{message}'")
|
||||
|
||||
# Проверяем команду завершения
|
||||
if message.upper() == 'EXIT':
|
||||
conn.sendall("🛑 Сервер завершает работу...".encode("utf-8"))
|
||||
print("🛑 Получена команда EXIT, завершение работы сервера")
|
||||
return
|
||||
|
||||
# Модифицируем сообщение и отправляем обратно
|
||||
modified_message = f"🔄 ECHO: {message.upper()}"
|
||||
conn.sendall(modified_message.encode())
|
||||
print(f"📤 Отправлен ответ: '{modified_message}'")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n🛑 Сервер остановлен пользователем")
|
||||
finally:
|
||||
server.close()
|
||||
print("🔒 Сокет сервера закрыт")
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_tcp_server()
|
||||
33
udp_client.py
Normal file
33
udp_client.py
Normal file
@ -0,0 +1,33 @@
|
||||
import socket
|
||||
|
||||
def run_udp_client():
|
||||
# Создаем UDP сокет
|
||||
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
|
||||
try:
|
||||
print("UDP клиент запущен")
|
||||
print("Введите сообщения для отправки на сервер")
|
||||
print("Для выхода введите 'exit'")
|
||||
|
||||
while True:
|
||||
# Получаем сообщение от пользователя
|
||||
message = input("💬 Введите сообщение: ")
|
||||
# Отправляем сообщение серверу
|
||||
client.sendto(message.encode(), ('127.0.0.1', 10001))
|
||||
|
||||
# Если пользователь ввел exit - выходим
|
||||
if message.lower() == 'exit':
|
||||
break
|
||||
|
||||
# Получаем ответ от сервера
|
||||
data, _ = client.recvfrom(1024)
|
||||
print(f"📨 Ответ от сервера: {data.decode()}")
|
||||
|
||||
except Exception as e:
|
||||
print(f"❌ Ошибка: {e}")
|
||||
finally:
|
||||
client.close()
|
||||
print("🔒 Клиент закрыт")
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_udp_client()
|
||||
35
udp_server.py
Normal file
35
udp_server.py
Normal file
@ -0,0 +1,35 @@
|
||||
import socket
|
||||
|
||||
def run_udp_server():
|
||||
# Создаем UDP сокет
|
||||
server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
# Привязываем сокет к адресу и порту
|
||||
server.bind(('0.0.0.0', 10001))
|
||||
print("🚀 UDP сервер запущен на порту 10001")
|
||||
print("Ожидание сообщений...")
|
||||
|
||||
try:
|
||||
while True:
|
||||
# Получаем данные и адрес отправителя
|
||||
data, addr = server.recvfrom(1024)
|
||||
message = data.decode().strip()
|
||||
print(f"📨 Сообщение от {addr}: '{message}'")
|
||||
|
||||
# Проверяем команду завершения
|
||||
if message.upper() == 'EXIT':
|
||||
print("🛑 Получена команда EXIT, завершение работы сервера")
|
||||
break
|
||||
|
||||
# Модифицируем сообщение и отправляем обратно
|
||||
modified_message = f"🔄 UDP-ECHO: {message.upper()} [modified]"
|
||||
server.sendto(modified_message.encode(), addr)
|
||||
print(f"📤 Отправлен ответ клиенту: '{modified_message}'")
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\n🛑 Сервер остановлен пользователем")
|
||||
finally:
|
||||
server.close()
|
||||
print("🔒 Сокет сервера закрыт")
|
||||
|
||||
if __name__ == "__main__":
|
||||
run_udp_server()
|
||||
Loading…
Reference in New Issue
Block a user